fix(schema-diff): fan out metadata by sqlserver capacity

This commit is contained in:
jearhe authored and GitHub committed 2026-09-20 01:28:49 +08:00
1 parent 8b757b97bc
commit dc6c9b5c7c
2 files changed
+477 -19

No files matched your search

@@ -1,5 +1,15 @@
import { describe, expect, it } from "vitest";
import { createConcurrencyLimiter, loadSchemaDetails, mapWithConcurrency, schemaDiffMetadataConcurrency, schemaDiffMetadataLoadPlan, shouldFetchSchemaDiffDdl, type SchemaDiffMetadataApi, type SchemaDiffMetadataProgress } from "../../schema/schemaDiffMetadataLoad";
import {
createConcurrencyLimiter,
loadSchemaDetails,
mapWithConcurrency,
schemaDiffMetadataConcurrency,
schemaDiffMetadataLaneCountForTests,
schemaDiffMetadataLoadPlan,
shouldFetchSchemaDiffDdl,
type SchemaDiffMetadataApi,
type SchemaDiffMetadataProgress,
} from "../../schema/schemaDiffMetadataLoad";
import { getSchemaDiffNextProgressStep, shouldLoadSchemaDiffExtraObjectPhase, shouldLoadSchemaDiffExtraObjects, shouldLoadSchemaDiffRoutines } from "../../schema/schemaDiffProgress";
import { DEFAULT_MYSQL_OPTIONS, DEFAULT_POSTGRES_OPTIONS } from "../../../types/schemaDiff";
import type { TableInfo } from "../../../types/database";
@@ -8,6 +18,80 @@ function wait(ms: number) {
return new Promise<void>((resolve) => setTimeout(resolve, ms));
}
const METADATA_POOL_BUSY = "DBX metadata pool is busy; please retry";
/**
* Mirrors the backend metadata gate (`crates/dbx-core/src/connection/mod.rs`):
* at most `capacity` in-flight metadata requests per gate key, and any request
* that cannot start is what `METADATA_POOL_ACQUIRE_TIMEOUT` (5s) reports as
* `DBX metadata pool is busy; please retry`.
*
* The fake rejects waiting requests instead of modelling the 5s wait, which is
* the conservative side of the real timeout: a test that passes here proves the
* caller never queues on the gate at all, not merely that it waits less than
* five seconds.
*/
function createMetadataGate(capacity: number) {
const active = new Map<string, number>();
let peak = 0;
return {
peak: () => peak,
async run<T>(key: string, task: () => Promise<T>): Promise<T> {
const current = active.get(key) ?? 0;
if (current >= capacity) throw new Error(METADATA_POOL_BUSY);
active.set(key, current + 1);
peak = Math.max(peak, current + 1);
try {
return await task();
} finally {
active.set(key, (active.get(key) ?? 1) - 1);
}
},
};
}
/**
* Wraps a metadata API in the backend gate above. The key matches
* `metadata_gate_key`: connection id plus database (SQL Server ignores the
* client session id there).
*/
function gateMetadataApi(gate: ReturnType<typeof createMetadataGate>, delays: { ddl: number; metadata: number }) {
let inFlight = 0;
let inFlightPeak = 0;
let calls = 0;
const key = (connectionId: string, database: string) => `${connectionId}\u0000${database}`;
const run = <T>(connectionId: string, database: string, delay: number, task: () => T) =>
gate.run(key(connectionId, database), async () => {
inFlight += 1;
inFlightPeak = Math.max(inFlightPeak, inFlight);
calls += 1;
try {
// `0` means "no artificial latency": the fake gate decides admission
// synchronously, so the fan-out model under test does not depend on a
// timer, and a large-scale case can skip hundreds of sequential timers.
if (delay > 0) await wait(delay);
return task();
} finally {
inFlight -= 1;
}
});
const api: SchemaDiffMetadataApi = {
getTableDdl: (connectionId, database, _schema, table) => run(connectionId, database, delays.ddl, () => `ddl:${table}`),
getColumns: (connectionId, database) => run(connectionId, database, delays.metadata, () => []),
listIndexes: (connectionId, database) => run(connectionId, database, delays.metadata, () => []),
listForeignKeys: (connectionId, database) => run(connectionId, database, delays.metadata, () => []),
listTriggers: (connectionId, database) => run(connectionId, database, delays.metadata, () => []),
};
return { api, inFlightPeak: () => inFlightPeak, calls: () => calls };
}
function sqlserverTables(...names: string[]): TableInfo[] {
return names.map((name) => ({ name, table_type: "TABLE" }) as TableInfo);
}
describe("schemaDiffMetadataLoad", () => {
it("uses adaptive metadata concurrency for MySQL-compatible databases", () => {
expect(schemaDiffMetadataConcurrency("mysql")).toBe(2);
@@ -20,6 +104,277 @@ describe("schemaDiffMetadataLoad", () => {
expect(schemaDiffMetadataConcurrency(undefined)).toBe(6);
});
it("uses one serialized metadata request for SQL Server", () => {
// The backend gate admits `METADATA_POOL_SQLSERVER_LIMIT = 1` in-flight
// request per (connection, database) and both SQL Server pools hold one
// serialized session, so any extra fan-out only queues until it fails.
// Native and `sqlserver-legacy` share one dbType and one limit, so there is
// no driver-profile branch to make here.
expect(schemaDiffMetadataConcurrency("sqlserver")).toBe(1);
expect(schemaDiffMetadataConcurrency("SQLServer", 12)).toBe(1);
expect(schemaDiffMetadataConcurrency("sqlserver", 500)).toBe(1);
});
it("keeps SQL Server metadata fan-out inside the backend gate", async () => {
const gate = createMetadataGate(1);
const { api, inFlightPeak } = gateMetadataApi(gate, { ddl: 4, metadata: 1 });
const details = await loadSchemaDetails(
sqlserverTables("orders", "customers", "events", "invoices", "items", "audit"),
{
connectionId: "9596-fanout",
database: "app",
schema: "dbo",
dbType: "sqlserver",
// SQL Server falls back to the PostgreSQL option tree today
// (`getSchemaDiffOptionsForDbType`), which fetches DDL for every table.
options: { ...DEFAULT_POSTGRES_OPTIONS },
},
api,
);
expect(details.map((detail) => detail.name)).toEqual(["orders", "customers", "events", "invoices", "items", "audit"]);
expect(details.map((detail) => detail.ddl)).toEqual(["ddl:orders", "ddl:customers", "ddl:events", "ddl:invoices", "ddl:items", "ddl:audit"]);
expect(gate.peak()).toBe(1);
expect(inFlightPeak()).toBe(1);
});
it("keeps a large SQL Server schema inside the metadata gate", async () => {
// Regression for #9596 at the reporter's scale: compares over a handful of
// tables succeed there, while "more than 100 tables" reproduces
// `DBX metadata pool is busy` every time. 100 is a field observation, not
// a code threshold -- the failure comes from the 6-wide fan-out queueing on
// a capacity-1 resource, and a bigger schema adds metadata volume, tail
// latency and time windows where other metadata work shares the same
// permit. So this case scales the workload instead of special-casing a
// table count: the capacity fix has to hold for a realistically large
// schema, not only for the six tables above.
const tableCount = 120;
const tables = sqlserverTables(...Array.from({ length: tableCount }, (_, index) => `table_${String(index).padStart(3, "0")}`));
const gate = createMetadataGate(1);
// No artificial latency: the fake gate decides admission synchronously, so
// a 6-wide fan-out still trips it on the first burst, while 600 sequential
// timers would only add wall clock (`setTimeout(1)` costs ~15ms on
// Windows).
const { api, inFlightPeak, calls } = gateMetadataApi(gate, { ddl: 0, metadata: 0 });
const details = await loadSchemaDetails(
tables,
{
connectionId: "9596-large-schema",
database: "app",
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
},
api,
);
expect(details).toHaveLength(tableCount);
expect(details.every((detail) => detail.ddl === `ddl:${detail.name}`)).toBe(true);
// 5 metadata requests per table (DDL + columns + indexes + foreign keys +
// triggers), every one of them admitted by a single capacity-1 permit.
expect(calls()).toBe(tableCount * 5);
expect(gate.peak()).toBe(1);
expect(inFlightPeak()).toBe(1);
});
it("would trip the SQL Server metadata gate at the fan-out used before the fix", async () => {
const gate = createMetadataGate(1);
const runFanOut = createConcurrencyLimiter(6);
await expect(Promise.all(Array.from({ length: 6 }, () => runFanOut(() => gate.run("9596-fanout\u0000app", () => wait(1)))))).rejects.toThrow(METADATA_POOL_BUSY);
});
it("keeps SQL Server schema compare source and target databases inside the gate", async () => {
const gate = createMetadataGate(1);
const { api } = gateMetadataApi(gate, { ddl: 3, metadata: 1 });
const context = (database: string) => ({
connectionId: "9596-same-connection",
database,
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
});
const tables = sqlserverTables("orders", "customers", "events");
const source = await loadSchemaDetails(tables, context("app_source"), api);
const target = await loadSchemaDetails(tables, context("app_target"), api);
expect(source).toHaveLength(3);
expect(target).toHaveLength(3);
expect(gate.peak()).toBe(1);
});
it("queues concurrent SQL Server compares on one metadata resource instead of failing on the gate", async () => {
const gate = createMetadataGate(1);
const { api } = gateMetadataApi(gate, { ddl: 4, metadata: 1 });
const context = {
connectionId: "9596-background",
database: "app",
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
};
const tables = sqlserverTables("orders", "customers", "events", "invoices");
const [foreground, background] = await Promise.all([loadSchemaDetails(tables, context, api), loadSchemaDetails(tables, context, api)]);
expect(foreground).toHaveLength(4);
expect(background).toHaveLength(4);
expect(gate.peak()).toBe(1);
});
it("keeps one shared lane for a compare starting after another compare's first table", async () => {
// Regression: the lane used to be evicted by the limiter's idle callback,
// which at SQL Server's concurrency of 1 fires at every table boundary --
// mid-compare. A compare starting there built a second 1-wide lane, and the
// two fan-outs together overran the backend's capacity-1 gate with
// `DBX metadata pool is busy`. The lane must instead live as long as the
// compares actively using it hold it.
const gate = createMetadataGate(1);
const { api } = gateMetadataApi(gate, { ddl: 4, metadata: 1 });
const context = {
connectionId: "9596-mid-flight",
database: "app",
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
};
const tables = sqlserverTables("orders", "customers", "events", "invoices");
let resolvePastFirstTable!: () => void;
const pastFirstTable = new Promise<void>((resolve) => {
resolvePastFirstTable = resolve;
});
const first = loadSchemaDetails(
tables,
{
...context,
onProgress: (value) => {
if (value.current >= 2) resolvePastFirstTable();
},
},
api,
);
await pastFirstTable;
// The first compare is past a table boundary, exactly where the old
// idle-based eviction had already removed its lane mid-compare.
expect(schemaDiffMetadataLaneCountForTests()).toBe(1);
const second = loadSchemaDetails(tables, context, api);
const [firstDetails, secondDetails] = await Promise.all([first, second]);
expect(firstDetails).toHaveLength(4);
expect(secondDetails).toHaveLength(4);
expect(gate.peak()).toBe(1);
expect(schemaDiffMetadataLaneCountForTests()).toBe(0);
});
it("does not pin the metadata lane when the table list is empty", async () => {
// An empty compare (a filter that matches nothing) never runs a metadata
// request, so limiter activity alone can never free the lane. The call's
// own hold must release it, or the lane stays pinned in the map forever.
const gate = createMetadataGate(1);
const { api, calls } = gateMetadataApi(gate, { ddl: 0, metadata: 0 });
const details = await loadSchemaDetails(
[],
{
connectionId: "9596-empty-tables",
database: "app",
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
},
api,
);
expect(details).toEqual([]);
expect(calls()).toBe(0);
expect(schemaDiffMetadataLaneCountForTests()).toBe(0);
});
it("keeps the lane alive until a failed compare's queued requests drain", async () => {
// A failed compare stops awaiting requests it already queued: its
// `Promise.all` rejects while the remaining requests keep running on the
// lane. Those orphans must keep the lane in the map so an immediate retry
// shares it instead of racing the orphans on the gate with a second lane.
let active = 0;
let peak = 0;
const track = async <T>(task: () => Promise<T>) => {
active += 1;
peak = Math.max(peak, active);
try {
return await task();
} finally {
active -= 1;
}
};
const api: SchemaDiffMetadataApi = {
getTableDdl: async (_connectionId, _database, _schema, table) =>
track(async () => {
await wait(6);
return `ddl:${table}`;
}),
getColumns: async () => {
throw new Error("columns failed");
},
listIndexes: async () =>
track(async () => {
await wait(6);
return [];
}),
listForeignKeys: async () =>
track(async () => {
await wait(6);
return [];
}),
listTriggers: async () =>
track(async () => {
await wait(6);
return [];
}),
};
const context = {
connectionId: "9596-orphan-drain",
database: "app",
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
};
const tables = sqlserverTables("orders", "customers");
await expect(loadSchemaDetails(tables, context, api)).rejects.toThrow("columns failed");
// The compare has failed, but its queued index/key/trigger requests are
// still draining behind it: the lane must still exist for the retry.
expect(schemaDiffMetadataLaneCountForTests()).toBe(1);
await expect(loadSchemaDetails(tables, context, api)).rejects.toThrow("columns failed");
for (let attempt = 0; attempt < 100 && schemaDiffMetadataLaneCountForTests() > 0; attempt += 1) await wait(1);
expect(schemaDiffMetadataLaneCountForTests()).toBe(0);
expect(peak).toBe(1);
});
it("keeps SQL Server metadata lanes independent per database", async () => {
const gate = createMetadataGate(1);
const { api, inFlightPeak } = gateMetadataApi(gate, { ddl: 4, metadata: 1 });
const context = (database: string) => ({
connectionId: "9596-lanes",
database,
schema: "dbo",
dbType: "sqlserver",
options: { ...DEFAULT_POSTGRES_OPTIONS },
});
const tables = sqlserverTables("orders", "customers");
await Promise.all([loadSchemaDetails(tables, context("app_a"), api), loadSchemaDetails(tables, context("app_b"), api)]);
expect(gate.peak()).toBe(1);
expect(inFlightPeak()).toBe(2);
});
it("skips view DDL when views are disabled while preserving table DDL options", () => {
expect(shouldFetchSchemaDiffDdl(true, { tables: true, views: false })).toBe(false);
expect(shouldFetchSchemaDiffDdl(true, { tables: false, views: true })).toBe(true);
@@ -8,6 +8,25 @@ const MYSQL_SMALL_SCHEMA_DIFF_METADATA_CONCURRENCY = 4;
const MYSQL_SMALL_SCHEMA_TABLE_LIMIT = 30;
const DEFAULT_SCHEMA_DIFF_METADATA_CONCURRENCY = 6;
/**
* SQL Server metadata is served by a single serialized session per
* (connection, database): the native pool holds one `Mutex<SqlServerClient>`
* and the `sqlserver-legacy` profile holds one Agent session, while the
* backend metadata gate admits `METADATA_POOL_SQLSERVER_LIMIT = 1` in-flight
* request per (connection, database) and turns anything that still has to wait
* after `METADATA_POOL_ACQUIRE_TIMEOUT` (5s) into
* `DBX metadata pool is busy; please retry`
* (crates/dbx-core/src/connection/mod.rs). Fanning out further cannot add
* throughput because the backend serializes those same requests anyway; it
* only builds a queue that is guaranteed to time out.
*
* Both drivers share the same limit, so this is a per-database-type metadata
* capacity rather than a driver-capability check: `metadata_concurrency_limit`
* returns `METADATA_POOL_SQLSERVER_LIMIT` whenever the config is `sqlserver`,
* regardless of driver profile.
*/
const SQLSERVER_SCHEMA_DIFF_METADATA_CONCURRENCY = 1;
function normalizeConcurrencyLimit(limit: number): number {
return Number.isFinite(limit) ? Math.max(1, Math.floor(limit)) : 1;
}
@@ -20,6 +39,9 @@ export function schemaDiffMetadataConcurrency(dbType: string | null | undefined,
}
return MYSQL_LARGE_SCHEMA_DIFF_METADATA_CONCURRENCY;
}
if (normalizedDbType === "sqlserver") {
return SQLSERVER_SCHEMA_DIFF_METADATA_CONCURRENCY;
}
return DEFAULT_SCHEMA_DIFF_METADATA_CONCURRENCY;
}
@@ -84,7 +106,7 @@ export async function mapWithConcurrency<T, R>(items: readonly T[], limit: numbe
return results;
}
export function createConcurrencyLimiter(limit: number) {
export function createConcurrencyLimiter(limit: number, onIdle?: () => void) {
const maxActive = normalizeConcurrencyLimit(limit);
let active = 0;
const queue: Array<() => void> = [];
@@ -106,7 +128,11 @@ export function createConcurrencyLimiter(limit: number) {
function release() {
active = Math.max(0, active - 1);
const next = queue.shift();
if (next) next();
if (next) {
next();
return;
}
if (active === 0) onIdle?.();
}
return async function runLimited<T>(task: () => Promise<T>): Promise<T> {
@@ -119,6 +145,70 @@ export function createConcurrencyLimiter(limit: number) {
};
}
/**
* Schema compare metadata lanes are keyed exactly like the backend metadata
* gate -- `(connection id, database)` -- so the fan-out limit belongs to the
* resource instead of to one compare session. Two compares can hammer the same
* resource at once (closing a compare dialog keeps the compare running in the
* background) and adding their fan-out together would occupy the backend gate
* and fail whoever loses the race with `DBX metadata pool is busy; please
* retry`. Waiting here instead keeps the queue off the backend gate, where
* `METADATA_POOL_ACQUIRE_TIMEOUT` (5s) would turn it into that error.
*
* A lane is held by every `loadSchemaDetails` call running on it: each call
* registers itself as a holder before its first await and releases that hold
* in a `finally`, and the lane leaves the map only once the last holder has
* gone and every request it accepted has settled. The limiter's idle callback
* cannot own this lifecycle: at SQL Server's concurrency of 1 the lane goes
* idle at every table boundary, which used to evict it mid-compare so the next
* compare built a second lane and the two 1-wide fan-outs together overran the
* capacity-1 gate. The request count matters for the same reason on the way
* out: a failed compare stops awaiting requests it already queued, and those
* orphans must keep the lane alive so a retry still shares it instead of
* racing it on the gate. An empty table list never runs a request at all, so
* only the holder count can free its lane.
*/
interface SchemaDiffMetadataLane {
key: string;
run: ReturnType<typeof createConcurrencyLimiter>;
holders: number;
requests: number;
}
const schemaDiffMetadataLanes = new Map<string, SchemaDiffMetadataLane>();
function schemaDiffMetadataLaneKey(connectionId: string, database: string): string {
return `${connectionId}\u0000${database}`;
}
function evictSchemaDiffMetadataLaneIfUnused(lane: SchemaDiffMetadataLane) {
if (lane.holders <= 0 && lane.requests <= 0 && schemaDiffMetadataLanes.get(lane.key) === lane) {
schemaDiffMetadataLanes.delete(lane.key);
}
}
function acquireSchemaDiffMetadataLane(connectionId: string, database: string, limit: number): SchemaDiffMetadataLane {
const key = schemaDiffMetadataLaneKey(connectionId, database);
const existing = schemaDiffMetadataLanes.get(key);
if (existing) {
existing.holders += 1;
return existing;
}
const lane: SchemaDiffMetadataLane = { key, run: createConcurrencyLimiter(limit), holders: 1, requests: 0 };
schemaDiffMetadataLanes.set(key, lane);
return lane;
}
function releaseSchemaDiffMetadataLane(lane: SchemaDiffMetadataLane) {
lane.holders -= 1;
evictSchemaDiffMetadataLaneIfUnused(lane);
}
export function schemaDiffMetadataLaneCountForTests(): number {
return schemaDiffMetadataLanes.size;
}
export interface SchemaDiffMetadataApi {
getTableDdl(connectionId: string, database: string, schema: string, table: string, objectType?: ObjectSourceKind): Promise<string>;
getColumns(connectionId: string, database: string, schema: string, table: string): Promise<ColumnInfo[]>;
@@ -155,23 +245,36 @@ function isViewOrMaterializedView(tableType: string): ObjectSourceKind | undefin
export async function loadSchemaDetails(tables: TableInfo[], context: SchemaDetailLoadContext, api: SchemaDiffMetadataApi): Promise<TableSchemaDetail[]> {
const concurrency = schemaDiffMetadataConcurrency(context.dbType, tables.length);
const runMetadataQuery = createConcurrencyLimiter(concurrency);
// Hold the lane from before the first await so a compare starting while this
// one is between tables shares this lane instead of building a second one.
const lane = acquireSchemaDiffMetadataLane(context.connectionId, context.database, concurrency);
const runMetadataQuery = <T>(task: () => Promise<T>): Promise<T> => {
lane.requests += 1;
return lane.run(task).finally(() => {
lane.requests -= 1;
evictSchemaDiffMetadataLaneIfUnused(lane);
});
};
let completed = 0;
return mapWithConcurrency(tables, concurrency, async (table) => {
const objectType = isViewOrMaterializedView(table.table_type);
const loadPlan = schemaDiffMetadataLoadPlan(isSchemaDiffView(table), context.options);
const ddlPromise = loadPlan.ddl ? runMetadataQuery(() => api.getTableDdl(context.connectionId, context.database, context.schema, table.name, objectType)) : Promise.resolve("");
const [columns, indexes, foreignKeys, triggers, ddl] = await Promise.all([
loadPlan.columns ? runMetadataQuery(() => api.getColumns(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.indexes ? runMetadataQuery(() => api.listIndexes(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.foreignKeys ? runMetadataQuery(() => api.listForeignKeys(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.triggers ? runMetadataQuery(() => api.listTriggers(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
ddlPromise,
]);
try {
return await mapWithConcurrency(tables, concurrency, async (table) => {
const objectType = isViewOrMaterializedView(table.table_type);
const loadPlan = schemaDiffMetadataLoadPlan(isSchemaDiffView(table), context.options);
const ddlPromise = loadPlan.ddl ? runMetadataQuery(() => api.getTableDdl(context.connectionId, context.database, context.schema, table.name, objectType)) : Promise.resolve("");
const [columns, indexes, foreignKeys, triggers, ddl] = await Promise.all([
loadPlan.columns ? runMetadataQuery(() => api.getColumns(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.indexes ? runMetadataQuery(() => api.listIndexes(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.foreignKeys ? runMetadataQuery(() => api.listForeignKeys(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
loadPlan.triggers ? runMetadataQuery(() => api.listTriggers(context.connectionId, context.database, context.schema, table.name)) : Promise.resolve([]),
ddlPromise,
]);
const detail = { name: table.name, columns, indexes, foreignKeys, triggers, ddl };
context.onProgress?.({ current: ++completed, total: tables.length, objectName: table.name });
return detail;
});
const detail = { name: table.name, columns, indexes, foreignKeys, triggers, ddl };
context.onProgress?.({ current: ++completed, total: tables.length, objectName: table.name });
return detail;
});
} finally {
releaseSchemaDiffMetadataLane(lane);
}
}