From 2015bde057ad8670fd49b25baafa6abad27cb2a8 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 31 Jul 2026 15:08:43 +0200 Subject: [PATCH 01/12] MSSQL CDCPoller improvements and fixes: - Ensure correct ordering of CDC results - Handle deferred updates - Correctly count processed transactions in each polling cycle - Improved logging - Added tests --- .../module-mssql/src/replication/CDCPoller.ts | 181 +++++++--- .../module-mssql/test/src/CDCPoller.test.ts | 317 ++++++++++++++++++ 2 files changed, 455 insertions(+), 43 deletions(-) create mode 100644 modules/module-mssql/test/src/CDCPoller.test.ts diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index 047d01cca..44ee9b35e 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -218,20 +218,21 @@ export class CDCPoller { this.logger.info(`Polling bounds are ${startLSN} -> ${endLSN} spanning ${results.length} transaction(s).`); - let transactionCount = 0; + // We poll for batch size transactions, but these include transactions not applicable to our Source Tables. + // A single transaction can also span several Source Tables, so collect the distinct transaction LSNs + // that produced changes rather than counting per table, which would either double count the + // transactions spanning tables or miss the transactions applicable to only one of them. + let transactionLSNs = new Set(); this.logger.debug( `Currently replicating tables: ${this.replicatedTables.map((table) => table.toQualifiedName()).join(', ')}` ); for (const table of this.replicatedTables) { if (table.enabledForCDC()) { - const tableTransactionCount = await this.pollTable(table, { startLSN, endLSN }); - // We poll for batch size transactions, but these include transactions not applicable to our Source Tables. - // Each Source Table may or may not have transactions that are applicable to it, so just keep track of the highest number of transactions processed for any Source Table. - if (tableTransactionCount > transactionCount) { - transactionCount = tableTransactionCount; - } + const transactions = await this.pollTable(table, { startLSN, endLSN }); + transactions.forEach((t) => transactionLSNs.add(t)); } } + const transactionCount = transactionLSNs.size; this.logger.info( `Processed ${results.length} transaction(s), including ${transactionCount} Source Table transaction(s). Commited LSN: ${endLSN.toString()}` @@ -248,11 +249,18 @@ export class CDCPoller { } } - private async pollTable(table: MSSQLSourceTable, bounds: { startLSN: LSN; endLSN: LSN }): Promise { + /** + * Emits the changes this table has within the given bounds, and returns the LSNs of the + * transactions those changes belong to. The LSNs are returned in their string form so that the + * caller can deduplicate them across tables by value. + */ + private async pollTable(table: MSSQLSourceTable, bounds: { startLSN: LSN; endLSN: LSN }): Promise> { + const transactionLSNs = new Set(); + // Ensure that the startLSN is not before the minimum LSN for the table const minLSN = this.captureInstances.get(table.objectId)!.instances[0].minLSN; if (minLSN > bounds.endLSN) { - return 0; + return transactionLSNs; } else if (minLSN >= bounds.startLSN) { bounds.startLSN = minLSN; } @@ -260,7 +268,7 @@ export class CDCPoller { try { const { recordset: results } = await this.connectionManager.query( ` - SELECT * FROM ${table.allChangesFunction}(@from_lsn, @to_lsn, 'all update old') ORDER BY __$start_lsn, __$seqval + SELECT * FROM ${table.allChangesFunction}(@from_lsn, @to_lsn, 'all update old') ORDER BY __$start_lsn, __$seqval, __$operation `, [ { name: 'from_lsn', type: sql.VarBinary, value: bounds.startLSN.toBinary() }, @@ -268,46 +276,26 @@ export class CDCPoller { ] ); - let transactionCount = 0; - let updateBefore: any = null; - let lastTransactionLSN: LSN | null = null; - for (const row of results) { - const transactionLSN = LSN.fromBinary(row.__$start_lsn); - switch (row.__$operation) { - case Operation.DELETE: - await this.eventHandler.onDelete(row, table, results.columns); - this.logger.info(`Processed DELETE row LSN: ${transactionLSN}`); - break; - case Operation.INSERT: - await this.eventHandler.onInsert(row, table, results.columns); - this.logger.info(`Processed INSERT row LSN: ${transactionLSN}`); + for (const { transactionLSN, type, rows } of groupLogicalChanges(results, table)) { + switch (type) { + case LogicalChangeType.DELETE: + await this.eventHandler.onDelete(rows[0], table, results.columns); break; - case Operation.UPDATE_BEFORE: - updateBefore = row; - this.logger.debug(`Processed UPDATE, before row LSN: ${transactionLSN}`); + case LogicalChangeType.INSERT: + await this.eventHandler.onInsert(rows[0], table, results.columns); break; - case Operation.UPDATE_AFTER: - if (updateBefore === null) { - throw new ReplicationAssertionError('Missing before image for update event.'); - } - await this.eventHandler.onUpdate(row, updateBefore, table, results.columns); - updateBefore = null; - this.logger.info(`Processed UPDATE row LSN: ${transactionLSN}`); + case LogicalChangeType.UPDATE: + case LogicalChangeType.DEFERRED_UPDATE: + const [rowBefore, rowAfter] = rows; + await this.eventHandler.onUpdate(rowAfter, rowBefore, table, results.columns); break; - default: - this.logger.warn(`Unknown operation type [${row.__$operation}] encountered in CDC changes.`); } + this.logger.info(`Processed ${type}. Transaction LSN: ${transactionLSN}`); - // Increment transaction count when we encounter a new transaction LSN (except for UPDATE_BEFORE rows) - if (transactionLSN != lastTransactionLSN) { - lastTransactionLSN = transactionLSN; - if (row.__$operation !== Operation.UPDATE_BEFORE) { - transactionCount++; - } - } + transactionLSNs.add(transactionLSN.toString()); } - return transactionCount; + return transactionLSNs; } catch (error) { // This Covers both deleted tables and capture instances if (error.message.includes(`Invalid object name`)) { @@ -443,3 +431,110 @@ export class CDCPoller { ); } } + +enum LogicalChangeType { + INSERT = 'INSERT', + DELETE = 'DELETE', + UPDATE = 'UPDATE', + DEFERRED_UPDATE = 'DEFERRED UPDATE' +} + +/** + * One logical change to a single row, made up of the one or two CDC rows that describe it. + */ +interface LogicalChange { + transactionLSN: LSN; + type: LogicalChangeType; + /** + * Inserts and Deletes resolve to 1 row. + * Updates resolve to 2 rows, the row values before and after the update: [rowBefore, rowAfter]. + */ + rows: any[]; +} + +/** + * Groups CDC change rows into the logical row changes they describe. + * + * SQL Server records a logical change as either one row (a plain insert or delete) or two rows that share + * a `__$seqval`. `__$seqval` represents the ordering of the changes to a row within a transaction. + * CDC operations that can share a `__$seqval` are: + * - The before and after operations of an in-place update + * - The delete and insert operations of a deferred update. + * + * This method groups and emits rows in the same transaction based on their `__$seqval` + * + * The source table is only used to identify the table in the errors raised for change rows that + * do not describe a valid logical change. + */ +function* groupLogicalChanges(rows: any[], table: MSSQLSourceTable): Generator { + let currentRows: any[] = []; + let currentTransactionLSN: Buffer | null = null; + let currentSequence: Buffer | null = null; + + for (const row of rows) { + const nextTransactionLSN: Buffer = row.__$start_lsn; + const nextSequence: Buffer = row.__$seqval; + + if ( + currentRows.length > 0 && + !(nextTransactionLSN.equals(currentTransactionLSN!) && nextSequence!.equals(currentSequence!)) + ) { + yield toLogicalChange(currentRows, currentTransactionLSN!, table); + currentRows = []; + } + currentTransactionLSN = nextTransactionLSN; + currentSequence = nextSequence; + currentRows.push(row); + } + + if (currentRows.length > 0) { + yield toLogicalChange(currentRows, currentTransactionLSN!, table); + } +} + +function toLogicalChange(rows: any[], startLSN: Buffer, table: MSSQLSourceTable): LogicalChange { + // The rows are ordered by operation in the query, but this is an extra safeguard + const orderedRows = [...rows].sort((a, b) => a.__$operation - b.__$operation); + const transactionLSN = LSN.fromBinary(startLSN); + return { + transactionLSN, + type: resolveLogicalChangeType(orderedRows, transactionLSN, table), + rows: orderedRows + }; +} + +function resolveLogicalChangeType(orderedRows: any[], transactionLSN: LSN, table: MSSQLSourceTable): LogicalChangeType { + if (orderedRows.length === 1) { + const operation = orderedRows[0].__$operation; + if (operation === Operation.UPDATE_BEFORE || operation === Operation.UPDATE_AFTER) { + throw new ReplicationAssertionError( + `Incomplete update for table ${table.toQualifiedName()} in transaction LSN ${transactionLSN}: an update must have both a before and an after image.` + ); + } + + if (operation === Operation.INSERT) { + return LogicalChangeType.INSERT; + } else if (operation === Operation.DELETE) { + return LogicalChangeType.DELETE; + } else { + throw new ReplicationAssertionError( + `Unrecognized operation: ${operation} for table ${table.toQualifiedName()} in transaction LSN ${transactionLSN}.` + ); + } + } else if (orderedRows.length === 2) { + const [first, second] = orderedRows; + if (first.__$operation === Operation.UPDATE_BEFORE && second.__$operation === Operation.UPDATE_AFTER) { + return LogicalChangeType.UPDATE; + } else if (first.__$operation === Operation.DELETE && second.__$operation === Operation.INSERT) { + return LogicalChangeType.DEFERRED_UPDATE; + } + + throw new ReplicationAssertionError( + `Unexpected CDC operations [${first.__$operation}, ${second.__$operation}] for a single logical change on table ${table.toQualifiedName()} in transaction LSN ${transactionLSN}.` + ); + } + + throw new ReplicationAssertionError( + `Unexpected number of CDC operations [${orderedRows.length}] for a single logical change on table ${table.toQualifiedName()} in transaction LSN ${transactionLSN}.` + ); +} diff --git a/modules/module-mssql/test/src/CDCPoller.test.ts b/modules/module-mssql/test/src/CDCPoller.test.ts new file mode 100644 index 000000000..fb352f8f9 --- /dev/null +++ b/modules/module-mssql/test/src/CDCPoller.test.ts @@ -0,0 +1,317 @@ +import { LSN } from '@module/common/LSN.js'; +import { MSSQLSourceTable } from '@module/common/MSSQLSourceTable.js'; +import { CDCEventHandler, CDCPoller } from '@module/replication/CDCPoller.js'; +import { MSSQLConnectionManager } from '@module/replication/MSSQLConnectionManager.js'; +import { + createCheckpoint, + escapeIdentifier, + getCaptureInstances, + getLatestLSN, + getLatestReplicatedLSN, + toQualifiedTableName +} from '@module/utils/mssql.js'; +import { getReplicationIdentityColumns } from '@module/utils/schema.js'; +import timers from 'timers/promises'; +import { afterEach, beforeEach, describe, expect, test } from 'vitest'; +import { clearTestDb, enableCDCForTable, TEST_CONNECTION_OPTIONS, waitForPendingCDCChanges } from './util.js'; + +describe('CDCPoller tests', { timeout: 60_000 }, () => { + let connectionManager: MSSQLConnectionManager; + + beforeEach(async () => { + connectionManager = new MSSQLConnectionManager(TEST_CONNECTION_OPTIONS, {}); + await clearTestDb(connectionManager); + }); + + afterEach(async () => { + await connectionManager.end(); + }); + + test('Deferred updates are collapsed into a single UPDATE', async () => { + const tableName = 'test_deferred_update'; + await createDeferredUpdateTestTable(connectionManager, tableName); + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + + const beforeInsertLSN = await getLatestLSN(connectionManager); + await connectionManager.query(`INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1'), (2, 'V2')`); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + // Only replicate changes made after the initial insert. + const startLSN = await getLatestReplicatedLSN(connectionManager); + + const beforeUpdateLSN = await getLatestLSN(connectionManager); + // Updating a column with a unique index, using the actual column value creates deferred updates + await connectionManager.query(`UPDATE ${qualifiedName} SET code = code + 'A'`); + await waitForPendingCDCChanges(beforeUpdateLSN, connectionManager); + + const table = await resolveSourceTable(connectionManager, tableName); + const { operations } = await collectChanges({ + connectionManager, + tables: [table], + startLSN, + expectedOperationCount: 2 + }); + + expect(operations).toMatchObject([ + { operation: 'update', rowBefore: { id: 1, code: 'V1' }, rowAfter: { id: 1, code: 'V1A' } }, + { operation: 'update', rowBefore: { id: 2, code: 'V2' }, rowAfter: { id: 2, code: 'V2A' } } + ]); + }); + + test('Normal deletes and inserts in the same transaction are not collapsed into updates', async () => { + const tableName = 'test_mixed_transaction'; + await createDeferredUpdateTestTable(connectionManager, tableName); + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + + const beforeInsertLSN = await getLatestLSN(connectionManager); + await connectionManager.query(` + INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1'), (2, 'V2') + `); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + const startLSN = await getLatestReplicatedLSN(connectionManager); + + const beforeUpdateLSN = await getLatestLSN(connectionManager); + await connectionManager.query(` + BEGIN TRAN; + DELETE FROM ${qualifiedName} WHERE id = 1; + INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1again'); + COMMIT; + `); + await waitForPendingCDCChanges(beforeUpdateLSN, connectionManager); + + const table = await resolveSourceTable(connectionManager, tableName); + const { operations } = await collectChanges({ + connectionManager, + tables: [table], + startLSN, + expectedOperationCount: 2 + }); + + expect(operations).toMatchObject([ + { operation: 'delete', row: { id: 1, code: 'V1' } }, + { operation: 'insert', row: { id: 1, code: 'V1again' } } + ]); + }); + + test('In place updates emit a single UPDATE with the before and after rows', async () => { + const tableName = 'test_in_place_update'; + await createDeferredUpdateTestTable(connectionManager, tableName); + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + + const beforeInsertLSN = await getLatestLSN(connectionManager); + await connectionManager.query(`INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1')`); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + const startLSN = await getLatestReplicatedLSN(connectionManager); + + const beforeUpdateLSN = await getLatestLSN(connectionManager); + // The new value does not derive from the current one, so the row stays put. + await connectionManager.query(`UPDATE ${qualifiedName} SET code = 'V9' WHERE id = 1`); + await waitForPendingCDCChanges(beforeUpdateLSN, connectionManager); + + const table = await resolveSourceTable(connectionManager, tableName); + const { operations } = await collectChanges({ + connectionManager, + tables: [table], + startLSN, + expectedOperationCount: 1 + }); + + expect(operations).toMatchObject([ + { operation: 'update', rowBefore: { id: 1, code: 'V1' }, rowAfter: { id: 1, code: 'V9' } } + ]); + }); + + test('Committed transaction count covers transactions across all replicated tables', async () => { + const firstTableName = 'test_transaction_count_first'; + const secondTableName = 'test_transaction_count_second'; + await createDeferredUpdateTestTable(connectionManager, firstTableName); + await createDeferredUpdateTestTable(connectionManager, secondTableName); + const firstQualifiedName = toQualifiedTableName(connectionManager.schema, firstTableName); + const secondQualifiedName = toQualifiedTableName(connectionManager.schema, secondTableName); + + const startLSN = await getLatestReplicatedLSN(connectionManager); + + // Two separate transactions, each applying to only one of the tables. Previously this would have been undercounted as 1 transaction. + await connectionManager.query(`INSERT INTO ${firstQualifiedName} (id, code) VALUES (1, 'A1')`); + await connectionManager.query(`INSERT INTO ${secondQualifiedName} (id, code) VALUES (1, 'B1')`); + + // CDC captures transactions in commit order, so waiting for a checkpoint written after both + // inserts guarantees both have been captured before polling starts. That in turn guarantees + // the poller sees both within a single polling cycle, which is what is being asserted here. + const beforeCheckpointLSN = await getLatestLSN(connectionManager); + await createCheckpoint(connectionManager); + await waitForPendingCDCChanges(beforeCheckpointLSN, connectionManager); + + const { operations, commits } = await collectChanges({ + connectionManager, + tables: [ + await resolveSourceTable(connectionManager, firstTableName), + await resolveSourceTable(connectionManager, secondTableName) + ], + startLSN, + expectedOperationCount: 2 + }); + + expect(operations).toMatchObject([ + { operation: 'insert', row: { id: 1, code: 'A1' } }, + { operation: 'insert', row: { id: 1, code: 'B1' } } + ]); + + // The checkpoint table is not replicated here, so only the two inserts are counted. + const transactionCount = commits.reduce((total, commit) => total + commit.transactionCount, 0); + expect(transactionCount).toEqual(2); + }); +}); + +/** + * A change as it was handed to the CDCEventHandler. The CDC rows are kept as received, so that + * assertions can match on whichever source columns they care about. + */ +type RecordedOperation = + | { operation: 'insert'; row: any } + | { operation: 'delete'; row: any } + | { operation: 'update'; rowBefore: any; rowAfter: any }; + +/** + * A commit as it was reported to the CDCEventHandler at the end of a polling cycle. + */ +interface RecordedCommit { + lsn: string; + transactionCount: number; +} + +/** + * Records the operations emitted by the CDCPoller so that their relative order can be asserted. + */ +class RecordingCDCEventHandler implements CDCEventHandler { + readonly operations: RecordedOperation[] = []; + readonly commits: RecordedCommit[] = []; + + async onInsert(row: any): Promise { + this.operations.push({ operation: 'insert', row }); + } + + async onUpdate(rowAfter: any, rowBefore: any): Promise { + this.operations.push({ operation: 'update', rowBefore, rowAfter }); + } + + async onDelete(row: any): Promise { + this.operations.push({ operation: 'delete', row }); + } + + async onCommit(lsn: string, transactionCount: number): Promise { + this.commits.push({ lsn, transactionCount }); + } + + async onSchemaChange(): Promise {} +} + +/** + * How long to wait for the expected operations before returning whatever arrived. + */ +const COLLECT_CHANGES_TIMEOUT_MS = 20_000; + +interface CollectChangesOptions { + connectionManager: MSSQLConnectionManager; + tables: MSSQLSourceTable[]; + startLSN: LSN; + expectedOperationCount: number; +} + +/** + * Runs a CDCPoller from startLSN until the expected number of operations has been emitted, and + * returns the handler holding the operations and the commits in the order they were reported. + */ +async function collectChanges(options: CollectChangesOptions): Promise { + const { connectionManager, tables, startLSN, expectedOperationCount } = options; + const eventHandler = new RecordingCDCEventHandler(); + + const poller = new CDCPoller({ + connectionManager, + eventHandler, + getReplicatedTables: () => tables, + // No sync config patterns, so no unrelated table is ever treated as a new table to replicate. + sourceTables: [], + startLSN, + additionalConfig: { + pollingBatchSize: 100, + pollingIntervalMs: 50, + trustServerCertificate: true + }, + schemaCheckIntervalMs: 60_000 + }); + + const pollerPromise = poller.replicateUntilStopped(); + try { + const deadline = Date.now() + COLLECT_CHANGES_TIMEOUT_MS; + while (eventHandler.operations.length < expectedOperationCount && Date.now() < deadline) { + // Rethrows if the poller fails instead of waiting out the full timeout. + await Promise.race([timers.setTimeout(50), pollerPromise]); + } + } finally { + await poller.stop(); + await pollerPromise; + } + + return eventHandler; +} + +/** + * Creates a table with a clustered primary key and a unique index over a separate column. + * A unique index is one of the prerequisites for triggering deferred updates in SQL Server. + */ +async function createDeferredUpdateTestTable( + connectionManager: MSSQLConnectionManager, + tableName: string +): Promise { + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + await connectionManager.query(` + CREATE TABLE ${qualifiedName} ( + id INT NOT NULL PRIMARY KEY, + code VARCHAR(100) NOT NULL + ) + `); + await connectionManager.query( + `CREATE UNIQUE NONCLUSTERED INDEX ${escapeIdentifier(`UX_${tableName}_code`)} ON ${qualifiedName}(code)` + ); + await enableCDCForTable({ connectionManager, table: tableName }); +} + +/** + * Builds the MSSQLSourceTable that the CDCPoller needs, straight from the table's CDC capture instance. + * The poller only uses the capture instance and object id, so no SourceTables are required here. + */ +async function resolveSourceTable( + connectionManager: MSSQLConnectionManager, + tableName: string +): Promise { + const captureInstances = await getCaptureInstances({ + connectionManager, + table: { schema: connectionManager.schema, name: tableName } + }); + const details = [...captureInstances.values()][0]; + if (details == null) { + throw new Error(`No CDC capture instance found for table ${tableName}`); + } + + const replicaIdColumnsResult = await getReplicationIdentityColumns({ + connectionManager, + tableName, + schema: connectionManager.schema + }); + + const table = new MSSQLSourceTable( + { + connectionTag: connectionManager.connectionTag, + objectId: details.sourceTable.objectId, + schema: details.sourceTable.schema, + name: details.sourceTable.name, + replicaIdColumns: replicaIdColumnsResult.columns + }, + [] + ); + table.setCaptureInstance(details.instances[0]); + return table; +} From 9101c4694c4d1e6e91e011b10b3a149b385603ab Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 31 Jul 2026 15:14:15 +0200 Subject: [PATCH 02/12] Changeset --- .changeset/open-worms-unite.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changeset/open-worms-unite.md diff --git a/.changeset/open-worms-unite.md b/.changeset/open-worms-unite.md new file mode 100644 index 000000000..85703b486 --- /dev/null +++ b/.changeset/open-worms-unite.md @@ -0,0 +1,7 @@ +--- +'@powersync/service-module-mssql': minor +--- + +MSSQL CDCPoller improvements and fixes: +- Ensure correct ordering of CDC results which previously could cause inconsistencies when handling deferred updates +- Correctly count processed transactions in each polling cycle From 6531f15fdc205857ac8bc0c0f54f14a0079a46c8 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 31 Jul 2026 15:23:39 +0200 Subject: [PATCH 03/12] Linting fix on changeset file --- .changeset/open-worms-unite.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.changeset/open-worms-unite.md b/.changeset/open-worms-unite.md index 85703b486..0f0477a43 100644 --- a/.changeset/open-worms-unite.md +++ b/.changeset/open-worms-unite.md @@ -3,5 +3,6 @@ --- MSSQL CDCPoller improvements and fixes: + - Ensure correct ordering of CDC results which previously could cause inconsistencies when handling deferred updates - Correctly count processed transactions in each polling cycle From 5179aa0c0a42aba49b95a0e371c1320d553ccac2 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 5 Aug 2026 22:20:24 +0200 Subject: [PATCH 04/12] Changed CDCPoller update retrieval query to use streaming. Cleaned up cdc operation grouping function --- .../module-mssql/src/replication/CDCPoller.ts | 83 ++++++++++--------- 1 file changed, 42 insertions(+), 41 deletions(-) diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index 44ee9b35e..d4634d7f6 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -266,28 +266,32 @@ export class CDCPoller { } try { - const { recordset: results } = await this.connectionManager.query( - ` + const request = await this.connectionManager.createRequest(); + request.input('from_lsn', sql.VarBinary, bounds.startLSN.toBinary()); + request.input('to_lsn', sql.VarBinary, bounds.endLSN.toBinary()); + + const columnsPromise = new Promise((resolve, reject) => { + request.on('recordset', resolve); + request.on('error', reject); + }); + const stream = request.toReadableStream(); + request.query(` SELECT * FROM ${table.allChangesFunction}(@from_lsn, @to_lsn, 'all update old') ORDER BY __$start_lsn, __$seqval, __$operation - `, - [ - { name: 'from_lsn', type: sql.VarBinary, value: bounds.startLSN.toBinary() }, - { name: 'to_lsn', type: sql.VarBinary, value: bounds.endLSN.toBinary() } - ] - ); + `); - for (const { transactionLSN, type, rows } of groupLogicalChanges(results, table)) { + for await (const { transactionLSN, type, rows } of groupLogicalChanges(stream, table)) { + const columns = await columnsPromise; switch (type) { case LogicalChangeType.DELETE: - await this.eventHandler.onDelete(rows[0], table, results.columns); + await this.eventHandler.onDelete(rows[0], table, columns); break; case LogicalChangeType.INSERT: - await this.eventHandler.onInsert(rows[0], table, results.columns); + await this.eventHandler.onInsert(rows[0], table, columns); break; case LogicalChangeType.UPDATE: case LogicalChangeType.DEFERRED_UPDATE: const [rowBefore, rowAfter] = rows; - await this.eventHandler.onUpdate(rowAfter, rowBefore, table, results.columns); + await this.eventHandler.onUpdate(rowAfter, rowBefore, table, columns); break; } this.logger.info(`Processed ${type}. Transaction LSN: ${transactionLSN}`); @@ -466,51 +470,48 @@ interface LogicalChange { * The source table is only used to identify the table in the errors raised for change rows that * do not describe a valid logical change. */ -function* groupLogicalChanges(rows: any[], table: MSSQLSourceTable): Generator { - let currentRows: any[] = []; - let currentTransactionLSN: Buffer | null = null; - let currentSequence: Buffer | null = null; - - for (const row of rows) { - const nextTransactionLSN: Buffer = row.__$start_lsn; - const nextSequence: Buffer = row.__$seqval; - - if ( - currentRows.length > 0 && - !(nextTransactionLSN.equals(currentTransactionLSN!) && nextSequence!.equals(currentSequence!)) - ) { - yield toLogicalChange(currentRows, currentTransactionLSN!, table); - currentRows = []; +async function* groupLogicalChanges(rows: AsyncIterable, table: MSSQLSourceTable): AsyncGenerator { + interface PendingGroup { + transactionLSN: Buffer; + sequence: Buffer; + rows: any[]; + } + + let current: PendingGroup | null = null; + + for await (const row of rows) { + const transactionLSN: Buffer = row.__$start_lsn; + const sequence: Buffer = row.__$seqval; + + if (current && !(transactionLSN.equals(current.transactionLSN) && sequence.equals(current.sequence))) { + yield toLogicalChange(current.rows, current.transactionLSN, table); + current = null; } - currentTransactionLSN = nextTransactionLSN; - currentSequence = nextSequence; - currentRows.push(row); + current ??= { + transactionLSN, + sequence, + rows: [] + }; + current.rows.push(row); } - if (currentRows.length > 0) { - yield toLogicalChange(currentRows, currentTransactionLSN!, table); + if (current) { + yield toLogicalChange(current.rows, current.transactionLSN, table); } } function toLogicalChange(rows: any[], startLSN: Buffer, table: MSSQLSourceTable): LogicalChange { - // The rows are ordered by operation in the query, but this is an extra safeguard - const orderedRows = [...rows].sort((a, b) => a.__$operation - b.__$operation); const transactionLSN = LSN.fromBinary(startLSN); return { transactionLSN, - type: resolveLogicalChangeType(orderedRows, transactionLSN, table), - rows: orderedRows + type: resolveLogicalChangeType(rows, transactionLSN, table), + rows: rows }; } function resolveLogicalChangeType(orderedRows: any[], transactionLSN: LSN, table: MSSQLSourceTable): LogicalChangeType { if (orderedRows.length === 1) { const operation = orderedRows[0].__$operation; - if (operation === Operation.UPDATE_BEFORE || operation === Operation.UPDATE_AFTER) { - throw new ReplicationAssertionError( - `Incomplete update for table ${table.toQualifiedName()} in transaction LSN ${transactionLSN}: an update must have both a before and an after image.` - ); - } if (operation === Operation.INSERT) { return LogicalChangeType.INSERT; From 915328740c56294bf52ae6ef397e225607c074f3 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 5 Aug 2026 22:22:26 +0200 Subject: [PATCH 05/12] Updated changeset --- .changeset/open-worms-unite.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.changeset/open-worms-unite.md b/.changeset/open-worms-unite.md index 0f0477a43..beb8d8b98 100644 --- a/.changeset/open-worms-unite.md +++ b/.changeset/open-worms-unite.md @@ -6,3 +6,4 @@ MSSQL CDCPoller improvements and fixes: - Ensure correct ordering of CDC results which previously could cause inconsistencies when handling deferred updates - Correctly count processed transactions in each polling cycle +- CDC polling query now streams results From 96704cb5ba95fb753f39a10d06cafd46a06f9a91 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 5 Aug 2026 23:49:31 +0200 Subject: [PATCH 06/12] Removed unnecessary promise wrapping for metadata retrieval in CDCPoller. --- modules/module-mssql/src/replication/CDCPoller.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index d4634d7f6..12458c215 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -270,9 +270,9 @@ export class CDCPoller { request.input('from_lsn', sql.VarBinary, bounds.startLSN.toBinary()); request.input('to_lsn', sql.VarBinary, bounds.endLSN.toBinary()); - const columnsPromise = new Promise((resolve, reject) => { - request.on('recordset', resolve); - request.on('error', reject); + let columns: sql.IColumnMetadata | null = null; + request.on('recordset', (recordsetColumns) => { + columns = recordsetColumns; }); const stream = request.toReadableStream(); request.query(` @@ -280,7 +280,11 @@ export class CDCPoller { `); for await (const { transactionLSN, type, rows } of groupLogicalChanges(stream, table)) { - const columns = await columnsPromise; + if (columns == null) { + throw new ReplicationAssertionError( + `Missing CDC column metadata while polling for updates for table ${table.toQualifiedName()}.` + ); + } switch (type) { case LogicalChangeType.DELETE: await this.eventHandler.onDelete(rows[0], table, columns); From d34145f06275ba84755874d1487509879940cd4c Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Thu, 6 Aug 2026 08:26:17 +0200 Subject: [PATCH 07/12] Minor CDCPoller cleanup --- .../module-mssql/src/replication/CDCPoller.ts | 77 ++++++++++--------- 1 file changed, 39 insertions(+), 38 deletions(-) diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index 12458c215..777ec6d46 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -17,13 +17,28 @@ import { CaptureInstanceDetails, getCaptureInstances, incrementLSN, toQualifiedT import { SourceTableChangeRef, tableExists } from '../utils/schema.js'; import { MSSQLConnectionManager } from './MSSQLConnectionManager.js'; -enum Operation { - DELETE = 1, - INSERT = 2, - UPDATE_BEFORE = 3, - UPDATE_AFTER = 4 +enum LogicalChangeType { + INSERT = 'INSERT', + DELETE = 'DELETE', + UPDATE = 'UPDATE', + DEFERRED_UPDATE = 'DEFERRED UPDATE' } +/** + * One logical change to a single row, made up of the one or two CDC rows that describe it. + */ +interface LogicalChange { + transactionLSN: LSN; + type: LogicalChangeType; + /** + * Inserts and Deletes resolve to 1 row. + * Updates resolve to 2 rows, the row values before and after the update: [rowBefore, rowAfter]. + */ + rows: any[]; +} + +export const DEFAULT_SCHEMA_CHECK_INTERVAL_MS = 60_000; + export enum SchemaChangeType { TABLE_RENAME = 'table_rename', TABLE_DROP = 'table_drop', @@ -55,8 +70,6 @@ export interface CDCEventHandler { onSchemaChange: (change: SchemaChange) => Promise; } -export const DEFAULT_SCHEMA_CHECK_INTERVAL_MS = 60_000; - export interface CDCPollerOptions { connectionManager: MSSQLConnectionManager; eventHandler: CDCEventHandler; @@ -78,27 +91,30 @@ export interface CDCPollerOptions { } /** + * Polls SQL Server CDC change tables for changes at a configurable interval. * + * Processes changes in commit order, groups CDC rows into logical insert, + * update, and delete operations. It only commits once all the operations recorded in a polling cycle have been processed. + * Periodically runs checks to detect schema and capture-instance changes. */ export class CDCPoller { private connectionManager: MSSQLConnectionManager; private eventHandler: CDCEventHandler; private currentLSN: LSN; private logger: Logger; - private listenerError: Error | null; private captureInstances: Map; + private pollingError: Error | null = null; private isStopped: boolean = false; private isStopping: boolean = false; private isPolling: boolean = false; private lastSchemaCheckTime: number = 0; - constructor(public options: CDCPollerOptions) { + constructor(private options: CDCPollerOptions) { this.logger = options.logger ?? defaultLogger; this.connectionManager = options.connectionManager; this.eventHandler = options.eventHandler; this.currentLSN = options.startLSN; - this.listenerError = null; this.captureInstances = new Map(); } @@ -174,7 +190,7 @@ export class CDCPoller { } // Non-recoverable errors - this.listenerError = error as Error; + this.pollingError = error as Error; this.logger.error('Error during CDC polling:', error); this.stop(); } @@ -182,9 +198,9 @@ export class CDCPoller { } } - if (this.listenerError) { - this.logger.error('CDC polling was stopped due to an error:', this.listenerError); - throw this.listenerError; + if (this.pollingError) { + this.logger.error('CDC polling was stopped due to an error:', this.pollingError); + throw this.pollingError; } this.logger.info(`CDC polling stopped...`); @@ -250,7 +266,7 @@ export class CDCPoller { } /** - * Emits the changes this table has within the given bounds, and returns the LSNs of the + * Processes the changes this table has within the given bounds, and returns the LSNs of the * transactions those changes belong to. The LSNs are returned in their string form so that the * caller can deduplicate them across tables by value. */ @@ -440,26 +456,6 @@ export class CDCPoller { } } -enum LogicalChangeType { - INSERT = 'INSERT', - DELETE = 'DELETE', - UPDATE = 'UPDATE', - DEFERRED_UPDATE = 'DEFERRED UPDATE' -} - -/** - * One logical change to a single row, made up of the one or two CDC rows that describe it. - */ -interface LogicalChange { - transactionLSN: LSN; - type: LogicalChangeType; - /** - * Inserts and Deletes resolve to 1 row. - * Updates resolve to 2 rows, the row values before and after the update: [rowBefore, rowAfter]. - */ - rows: any[]; -} - /** * Groups CDC change rows into the logical row changes they describe. * @@ -470,9 +466,6 @@ interface LogicalChange { * - The delete and insert operations of a deferred update. * * This method groups and emits rows in the same transaction based on their `__$seqval` - * - * The source table is only used to identify the table in the errors raised for change rows that - * do not describe a valid logical change. */ async function* groupLogicalChanges(rows: AsyncIterable, table: MSSQLSourceTable): AsyncGenerator { interface PendingGroup { @@ -514,6 +507,14 @@ function toLogicalChange(rows: any[], startLSN: Buffer, table: MSSQLSourceTable) } function resolveLogicalChangeType(orderedRows: any[], transactionLSN: LSN, table: MSSQLSourceTable): LogicalChangeType { + // This matches the actual CDC operation codes:https://learn.microsoft.com/en-us/sql/relational-databases/system-functions/cdc-fn-cdc-get-all-changes-capture-instance-transact-sql?view=sql-server-ver17#table-returned + enum Operation { + DELETE = 1, + INSERT = 2, + UPDATE_BEFORE = 3, + UPDATE_AFTER = 4 + } + if (orderedRows.length === 1) { const operation = orderedRows[0].__$operation; From 3dd2ddbfeff06068f12dfb99798ad1ef35a9b127 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Tue, 11 Aug 2026 19:05:44 +0200 Subject: [PATCH 08/12] Post merge conflict fixes for CDCPoller test --- modules/module-mssql/test/src/CDCPoller.test.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/modules/module-mssql/test/src/CDCPoller.test.ts b/modules/module-mssql/test/src/CDCPoller.test.ts index fb352f8f9..db775c13d 100644 --- a/modules/module-mssql/test/src/CDCPoller.test.ts +++ b/modules/module-mssql/test/src/CDCPoller.test.ts @@ -232,8 +232,6 @@ async function collectChanges(options: CollectChangesOptions): Promise tables, - // No sync config patterns, so no unrelated table is ever treated as a new table to replicate. - sourceTables: [], startLSN, additionalConfig: { pollingBatchSize: 100, @@ -312,6 +310,6 @@ async function resolveSourceTable( }, [] ); - table.setCaptureInstance(details.instances[0]); + table.setCaptureInstance([details.instances[0]]); return table; } From 12af024c3dfc6c5a9c51819dab4b48e9e61e62ad Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Tue, 11 Aug 2026 23:00:04 +0200 Subject: [PATCH 09/12] Fixed CDCPoller.test to include SourceTable metadata --- .../module-mssql/src/replication/CDCPoller.ts | 2 +- .../module-mssql/test/src/CDCPoller.test.ts | 35 ++++++++++++------- 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index ccd32d923..5a83aa369 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -242,7 +242,7 @@ export class CDCPoller { * Processes the changes this table has within the given bounds, and returns the LSNs of the * transactions those changes belong to. The LSNs are returned in their string form so that the * caller can deduplicate them across tables by value. - */ + */ private async pollTable(table: MSSQLSourceTable, bounds: { startLSN: LSN; endLSN: LSN }): Promise> { const transactionLSNs = new Set(); diff --git a/modules/module-mssql/test/src/CDCPoller.test.ts b/modules/module-mssql/test/src/CDCPoller.test.ts index db775c13d..4dd7b87f8 100644 --- a/modules/module-mssql/test/src/CDCPoller.test.ts +++ b/modules/module-mssql/test/src/CDCPoller.test.ts @@ -11,6 +11,7 @@ import { toQualifiedTableName } from '@module/utils/mssql.js'; import { getReplicationIdentityColumns } from '@module/utils/schema.js'; +import { SourceTable } from '@powersync/service-core'; import timers from 'timers/promises'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import { clearTestDb, enableCDCForTable, TEST_CONNECTION_OPTIONS, waitForPendingCDCChanges } from './util.js'; @@ -278,8 +279,8 @@ async function createDeferredUpdateTestTable( } /** - * Builds the MSSQLSourceTable that the CDCPoller needs, straight from the table's CDC capture instance. - * The poller only uses the capture instance and object id, so no SourceTables are required here. + * Builds the MSSQLSourceTable that the CDCPoller needs, including the persisted capture-instance binding + * that would normally be populated by source-table reconciliation. */ async function resolveSourceTable( connectionManager: MSSQLConnectionManager, @@ -300,16 +301,24 @@ async function resolveSourceTable( schema: connectionManager.schema }); - const table = new MSSQLSourceTable( - { - connectionTag: connectionManager.connectionTag, - objectId: details.sourceTable.objectId, - schema: details.sourceTable.schema, - name: details.sourceTable.name, - replicaIdColumns: replicaIdColumnsResult.columns - }, - [] - ); - table.setCaptureInstance([details.instances[0]]); + const ref = { + connectionTag: connectionManager.connectionTag, + objectId: details.sourceTable.objectId, + schema: details.sourceTable.schema, + name: details.sourceTable.name, + replicaIdColumns: replicaIdColumnsResult.columns + }; + const sourceTable = new SourceTable({ + id: `${details.sourceTable.objectId}`, + ref, + objectId: details.sourceTable.objectId, + replicaIdColumns: replicaIdColumnsResult.columns, + snapshotComplete: true, + bucketDataSources: [], + parameterLookupSources: [], + sourceMetadata: { captureTableObjectId: details.instances[0].objectId } + }); + const table = new MSSQLSourceTable(ref, [sourceTable]); + table.setCaptureInstance(details.instances); return table; } From 0604c9f49737d7b2a38e21dba88a526865ed9335 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 12 Aug 2026 10:19:56 +0200 Subject: [PATCH 10/12] Moved CDCPoller test case from CaptureReconciler.test.ts to CDCPoller.test.ts --- .../module-mssql/test/src/CDCPoller.test.ts | 110 ++++++-- .../test/src/CaptureReconciler.test.ts | 246 ++++++++---------- 2 files changed, 195 insertions(+), 161 deletions(-) diff --git a/modules/module-mssql/test/src/CDCPoller.test.ts b/modules/module-mssql/test/src/CDCPoller.test.ts index 4dd7b87f8..51dde5dc5 100644 --- a/modules/module-mssql/test/src/CDCPoller.test.ts +++ b/modules/module-mssql/test/src/CDCPoller.test.ts @@ -12,9 +12,17 @@ import { } from '@module/utils/mssql.js'; import { getReplicationIdentityColumns } from '@module/utils/schema.js'; import { SourceTable } from '@powersync/service-core'; +import sql from 'mssql'; import timers from 'timers/promises'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { clearTestDb, enableCDCForTable, TEST_CONNECTION_OPTIONS, waitForPendingCDCChanges } from './util.js'; +import { + clearTestDb, + createTestTableWithBasicId, + enableCDCForTable, + insertBasicIdTestData, + TEST_CONNECTION_OPTIONS, + waitForPendingCDCChanges +} from './util.js'; describe('CDCPoller tests', { timeout: 60_000 }, () => { let connectionManager: MSSQLConnectionManager; @@ -61,24 +69,32 @@ describe('CDCPoller tests', { timeout: 60_000 }, () => { test('Normal deletes and inserts in the same transaction are not collapsed into updates', async () => { const tableName = 'test_mixed_transaction'; - await createDeferredUpdateTestTable(connectionManager, tableName); + await createTestTableWithBasicId(connectionManager, tableName); const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); const beforeInsertLSN = await getLatestLSN(connectionManager); - await connectionManager.query(` - INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1'), (2, 'V2') - `); + const initial = await insertBasicIdTestData(connectionManager, tableName); await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); const startLSN = await getLatestReplicatedLSN(connectionManager); const beforeUpdateLSN = await getLatestLSN(connectionManager); - await connectionManager.query(` + const replacementDescription = 'replacement'; + const { recordset } = await connectionManager.query( + ` BEGIN TRAN; - DELETE FROM ${qualifiedName} WHERE id = 1; - INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1again'); + DELETE FROM ${qualifiedName} WHERE id = @id; + INSERT INTO ${qualifiedName} (description) + OUTPUT INSERTED.id, INSERTED.description + VALUES (@description); COMMIT; - `); + `, + [ + { name: 'id', type: sql.Int, value: initial.id }, + { name: 'description', type: sql.NVarChar(sql.MAX), value: replacementDescription } + ] + ); + const replacement = recordset[0]; await waitForPendingCDCChanges(beforeUpdateLSN, connectionManager); const table = await resolveSourceTable(connectionManager, tableName); @@ -90,25 +106,28 @@ describe('CDCPoller tests', { timeout: 60_000 }, () => { }); expect(operations).toMatchObject([ - { operation: 'delete', row: { id: 1, code: 'V1' } }, - { operation: 'insert', row: { id: 1, code: 'V1again' } } + { operation: 'delete', row: initial }, + { operation: 'insert', row: replacement } ]); }); test('In place updates emit a single UPDATE with the before and after rows', async () => { const tableName = 'test_in_place_update'; - await createDeferredUpdateTestTable(connectionManager, tableName); + await createTestTableWithBasicId(connectionManager, tableName); const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); const beforeInsertLSN = await getLatestLSN(connectionManager); - await connectionManager.query(`INSERT INTO ${qualifiedName} (id, code) VALUES (1, 'V1')`); + const initial = await insertBasicIdTestData(connectionManager, tableName); await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); const startLSN = await getLatestReplicatedLSN(connectionManager); const beforeUpdateLSN = await getLatestLSN(connectionManager); - // The new value does not derive from the current one, so the row stays put. - await connectionManager.query(`UPDATE ${qualifiedName} SET code = 'V9' WHERE id = 1`); + const updated = { ...initial, description: 'updated' }; + await connectionManager.query(`UPDATE ${qualifiedName} SET description = @description WHERE id = @id`, [ + { name: 'description', type: sql.NVarChar(sql.MAX), value: updated.description }, + { name: 'id', type: sql.Int, value: updated.id } + ]); await waitForPendingCDCChanges(beforeUpdateLSN, connectionManager); const table = await resolveSourceTable(connectionManager, tableName); @@ -119,24 +138,20 @@ describe('CDCPoller tests', { timeout: 60_000 }, () => { expectedOperationCount: 1 }); - expect(operations).toMatchObject([ - { operation: 'update', rowBefore: { id: 1, code: 'V1' }, rowAfter: { id: 1, code: 'V9' } } - ]); + expect(operations).toMatchObject([{ operation: 'update', rowBefore: initial, rowAfter: updated }]); }); test('Committed transaction count covers transactions across all replicated tables', async () => { const firstTableName = 'test_transaction_count_first'; const secondTableName = 'test_transaction_count_second'; - await createDeferredUpdateTestTable(connectionManager, firstTableName); - await createDeferredUpdateTestTable(connectionManager, secondTableName); - const firstQualifiedName = toQualifiedTableName(connectionManager.schema, firstTableName); - const secondQualifiedName = toQualifiedTableName(connectionManager.schema, secondTableName); + await createTestTableWithBasicId(connectionManager, firstTableName); + await createTestTableWithBasicId(connectionManager, secondTableName); const startLSN = await getLatestReplicatedLSN(connectionManager); // Two separate transactions, each applying to only one of the tables. Previously this would have been undercounted as 1 transaction. - await connectionManager.query(`INSERT INTO ${firstQualifiedName} (id, code) VALUES (1, 'A1')`); - await connectionManager.query(`INSERT INTO ${secondQualifiedName} (id, code) VALUES (1, 'B1')`); + const first = await insertBasicIdTestData(connectionManager, firstTableName); + const second = await insertBasicIdTestData(connectionManager, secondTableName); // CDC captures transactions in commit order, so waiting for a checkpoint written after both // inserts guarantees both have been captured before polling starts. That in turn guarantees @@ -156,14 +171,57 @@ describe('CDCPoller tests', { timeout: 60_000 }, () => { }); expect(operations).toMatchObject([ - { operation: 'insert', row: { id: 1, code: 'A1' } }, - { operation: 'insert', row: { id: 1, code: 'B1' } } + { operation: 'insert', row: first }, + { operation: 'insert', row: second } ]); // The checkpoint table is not replicated here, so only the two inserts are counted. const transactionCount = commits.reduce((total, commit) => total + commit.transactionCount, 0); expect(transactionCount).toEqual(2); }); + + test('Refreshes the bound capture instance to use its latest metadata', async () => { + const tableName = 'test_capture_metadata'; + await createTestTableWithBasicId(connectionManager, tableName); + const table = await resolveSourceTable(connectionManager, tableName); + const startupInstance = table.captureInstance; + console.log(startupInstance); + const beforeInsertLSN = await getLatestLSN(connectionManager); + const inserted = await insertBasicIdTestData(connectionManager, tableName); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + const captureInstances = await getCaptureInstances({ + connectionManager, + table: { schema: connectionManager.schema, name: tableName } + }); + console.log(captureInstances.values()); + const refreshed = captureInstances.get(table.objectId)!.instances[0]; + const eventHandler = new RecordingCDCEventHandler(); + const poller = new CDCPoller({ + connectionManager, + eventHandler, + getReplicatedTables: () => [table], + startLSN: LSN.fromString(LSN.ZERO), + additionalConfig: { + pollingBatchSize: 1, + pollingIntervalMs: 1, + trustServerCertificate: true + } + }); + (poller as any).captureInstances = captureInstances; + + const bounds = { + startLSN: LSN.fromString(LSN.ZERO), + endLSN: await getLatestReplicatedLSN(connectionManager) + }; + await (poller as any).pollTable(table, bounds); + + expect(startupInstance).not.toBe(refreshed); + expect(bounds.startLSN).toBe(refreshed.minLSN); + expect(table.captureInstance).toBe(refreshed); + expect(table.pinnedCaptureObjectId).toBe(refreshed.objectId); + expect(eventHandler.operations).toMatchObject([{ operation: 'insert', row: inserted }]); + }); }); /** diff --git a/modules/module-mssql/test/src/CaptureReconciler.test.ts b/modules/module-mssql/test/src/CaptureReconciler.test.ts index f18d05dae..11c3ebaca 100644 --- a/modules/module-mssql/test/src/CaptureReconciler.test.ts +++ b/modules/module-mssql/test/src/CaptureReconciler.test.ts @@ -1,7 +1,6 @@ import { CaptureInstance } from '@module/common/CaptureInstance.js'; import { LSN } from '@module/common/LSN.js'; import { MSSQLSourceTable } from '@module/common/MSSQLSourceTable.js'; -import { CDCPoller } from '@module/replication/CDCPoller.js'; import { createCaptureReconciler, MSSQLSourceMetadata, @@ -10,73 +9,7 @@ import { import type { MSSQLTableReconciliationContext } from '@module/replication/MSSQLTableReconciliationContext.js'; import { MSSQLTableReconciliationState } from '@module/replication/MSSQLTableReconciliationContext.js'; import { SourceEntityDescriptor, SourceTable } from '@powersync/service-core'; -import { describe, expect, it, vi } from 'vitest'; - -/** - * Build a source descriptor with optional identity overrides. - */ -function source(overrides: Partial = {}): SourceEntityDescriptor { - return { - connectionTag: 'default', - schema: 'dbo', - name: 'users', - objectId: 100, - replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], - ...overrides - }; -} - -/** - * Build a persisted source-table candidate with optional capture metadata. - */ -function candidate( - id: string, - metadata?: MSSQLSourceMetadata, - overrides: Partial[0]> = {} -): SourceTable { - return new SourceTable({ - id, - ref: { connectionTag: 'default', schema: 'dbo', name: 'users' }, - objectId: 100, - replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], - snapshotComplete: true, - bucketDataSources: [], - parameterLookupSources: [], - sourceMetadata: metadata, - ...overrides - }); -} - -/** - * Build a capture instance for the given capture-table object id. - */ -function instance(objectId: number): CaptureInstance { - return { - name: `dbo_users_${objectId}`, - objectId, - minLSN: LSN.fromString(LSN.ZERO), - createDate: new Date(), - pendingSchemaChanges: [] - }; -} - -function readyContext( - captureInstances: CaptureInstance[], - sourceDescriptor: SourceEntityDescriptor = source() -): MSSQLTableReconciliationContext { - return { - state: MSSQLTableReconciliationState.READY, - source: sourceDescriptor, - captureInstances - }; -} - -function unavailableContext( - state: MSSQLTableReconciliationState.TABLE_MISSING | MSSQLTableReconciliationState.CDC_DISABLED, - sourceDescriptor: SourceEntityDescriptor = source() -): MSSQLTableReconciliationContext { - return { state, source: sourceDescriptor }; -} +import { describe, expect, it } from 'vitest'; function reconcile(context: MSSQLTableReconciliationContext, candidates: SourceTable[]) { return createCaptureReconciler(context)({ source: context.source, candidates }); @@ -99,7 +32,7 @@ describe('readCaptureMetadata', () => { describe('createCaptureReconciler', () => { it('pins a new binding to the newest available capture instance', () => { - const resolution = reconcile(readyContext([instance(50), instance(40)]), []); + const resolution = reconcile(readyContext([createCaptureInstance(50), createCaptureInstance(40)]), []); expect(resolution.compatibleTables).toEqual([]); expect(resolution.incompatibleTables).toEqual([]); expect(resolution.newTableValues).toEqual({ sourceMetadata: { captureTableObjectId: 50 } }); @@ -110,7 +43,7 @@ describe('createCaptureReconciler', () => { reconcile( unavailableContext( MSSQLTableReconciliationState.TABLE_MISSING, - source({ objectId: undefined, replicaIdColumns: [] }) + createSourceDescriptor({ objectId: undefined, replicaIdColumns: [] }) ), [] ) @@ -122,9 +55,9 @@ describe('createCaptureReconciler', () => { reconcile( unavailableContext( MSSQLTableReconciliationState.TABLE_MISSING, - source({ objectId: undefined, replicaIdColumns: [] }) + createSourceDescriptor({ objectId: undefined, replicaIdColumns: [] }) ), - [candidate('old', { captureTableObjectId: 40 })] + [createSourceTableCandidate('old', { captureTableObjectId: 40 })] ) ).toThrow(/no longer matches the source table binding/); }); @@ -136,7 +69,10 @@ describe('createCaptureReconciler', () => { }); it('updates legacy metadata-free candidates to the newest capture instance', () => { - const resolution = reconcile(readyContext([instance(50)]), [candidate('a'), candidate('b')]); + const resolution = reconcile(readyContext([createCaptureInstance(50)]), [ + createSourceTableCandidate('a'), + createSourceTableCandidate('b') + ]); expect( resolution.compatibleTables.map((table) => ({ id: table.id, sourceMetadata: table.sourceMetadata })) ).toEqual([ @@ -148,21 +84,23 @@ describe('createCaptureReconciler', () => { }); it('fails an existing binding when CDC is disabled', () => { - expect(() => reconcile(unavailableContext(MSSQLTableReconciliationState.CDC_DISABLED), [candidate('a')])).toThrow( - /CDC is no longer enabled/ - ); + expect(() => + reconcile(unavailableContext(MSSQLTableReconciliationState.CDC_DISABLED), [createSourceTableCandidate('a')]) + ).toThrow(/CDC is no longer enabled/); }); it('reports a changed source identity as unavailable when CDC is disabled', () => { - const changedSource = source({ objectId: 200 }); + const changedSource = createSourceDescriptor({ objectId: 200 }); expect(() => - reconcile(unavailableContext(MSSQLTableReconciliationState.CDC_DISABLED, changedSource), [candidate('old')]) + reconcile(unavailableContext(MSSQLTableReconciliationState.CDC_DISABLED, changedSource), [ + createSourceTableCandidate('old') + ]) ).toThrow(/no longer matches the source table binding/); }); it('preserves a pinned capture identity that is still available', () => { - const resolution = reconcile(readyContext([instance(50), instance(40)]), [ - candidate('a', { captureTableObjectId: 40 }) + const resolution = reconcile(readyContext([createCaptureInstance(50), createCaptureInstance(40)]), [ + createSourceTableCandidate('a', { captureTableObjectId: 40 }) ]); expect(resolution.compatibleTables.map((table) => table.id)).toEqual(['a']); expect(resolution.compatibleTables[0].sourceMetadata).toEqual({ captureTableObjectId: 40 }); @@ -171,39 +109,46 @@ describe('createCaptureReconciler', () => { }); it('fails when the pinned capture instance was dropped, even with a replacement available', () => { - expect(() => reconcile(readyContext([instance(50)]), [candidate('a', { captureTableObjectId: 40 })])).toThrow( - /no longer available/ - ); + expect(() => + reconcile(readyContext([createCaptureInstance(50)]), [ + createSourceTableCandidate('a', { captureTableObjectId: 40 }) + ]) + ).toThrow(/no longer available/); }); it('fails on a mixture of metadata-free and pinned candidates', () => { expect(() => - reconcile(readyContext([instance(40)]), [candidate('a'), candidate('b', { captureTableObjectId: 40 })]) + reconcile(readyContext([createCaptureInstance(40)]), [ + createSourceTableCandidate('a'), + createSourceTableCandidate('b', { captureTableObjectId: 40 }) + ]) ).toThrow(/mixture/); }); it('fails on multiple distinct pinned identities', () => { expect(() => - reconcile(readyContext([instance(40), instance(41)]), [ - candidate('a', { captureTableObjectId: 40 }), - candidate('b', { captureTableObjectId: 41 }) + reconcile(readyContext([createCaptureInstance(40), createCaptureInstance(41)]), [ + createSourceTableCandidate('a', { captureTableObjectId: 40 }), + createSourceTableCandidate('b', { captureTableObjectId: 41 }) ]) ).toThrow(/multiple persisted capture identities/); }); it('does not replace an existing binding when the source identity changed', () => { - const changedSource = source({ objectId: 200 }); + const changedSource = createSourceDescriptor({ objectId: 200 }); expect(() => - reconcile(readyContext([instance(50)], changedSource), [candidate('old', { captureTableObjectId: 40 })]) + reconcile(readyContext([createCaptureInstance(50)], changedSource), [ + createSourceTableCandidate('old', { captureTableObjectId: 40 }) + ]) ).toThrow( /Table \[dbo\]\.\[users\] no longer matches the source table binding.*already-replicated data is retained/ ); }); it('drops stale incompatible candidates when a compatible candidate anchors the binding', () => { - const resolution = reconcile(readyContext([instance(50)]), [ - candidate('a'), - candidate('mismatch', undefined, { + const resolution = reconcile(readyContext([createCaptureInstance(50)]), [ + createSourceTableCandidate('a'), + createSourceTableCandidate('mismatch', undefined, { replicaIdColumns: [{ name: 'id', type: 'bigint', typeId: 127 }] }) ]); @@ -216,62 +161,93 @@ describe('createCaptureReconciler', () => { describe('MSSQLSourceTable.setCaptureInstance', () => { it('sets the instance matching the persisted capture-table object id', () => { - const table = new MSSQLSourceTable(source(), [candidate('a', { captureTableObjectId: 40 })]); - const expected = instance(40); + const table = new MSSQLSourceTable(createSourceDescriptor(), [ + createSourceTableCandidate('a', { captureTableObjectId: 40 }) + ]); + const expected = createCaptureInstance(40); - table.setCaptureInstance([instance(50), expected]); + table.setCaptureInstance([createCaptureInstance(50), expected]); expect(table.captureInstance).toBe(expected); }); it('sets null when the binding is legacy or the pinned instance is unavailable', () => { - const legacy = new MSSQLSourceTable(source(), [candidate('legacy')]); - const pinned = new MSSQLSourceTable(source(), [candidate('pinned', { captureTableObjectId: 40 })]); + const legacy = new MSSQLSourceTable(createSourceDescriptor(), [createSourceTableCandidate('legacy')]); + const pinned = new MSSQLSourceTable(createSourceDescriptor(), [ + createSourceTableCandidate('pinned', { captureTableObjectId: 40 }) + ]); - legacy.setCaptureInstance([instance(40)]); - pinned.setCaptureInstance([instance(40)]); - pinned.setCaptureInstance([instance(50)]); + legacy.setCaptureInstance([createCaptureInstance(40)]); + pinned.setCaptureInstance([createCaptureInstance(40)]); + pinned.setCaptureInstance([createCaptureInstance(50)]); expect(legacy.captureInstance).toBeNull(); expect(pinned.captureInstance).toBeNull(); }); }); -describe('CDCPoller capture-instance metadata', () => { - it('refreshes the bound instance to use its latest metadata', async () => { - const persisted = candidate('a', { captureTableObjectId: 40 }); - const table = new MSSQLSourceTable(source(), [persisted]); - const startupInstance = instance(40); - table.setCaptureInstance([startupInstance]); +/** + * Create a capture instance with the given capture-table object id. + */ +function createCaptureInstance(objectId: number): CaptureInstance { + return { + name: `dbo_users_${objectId}`, + objectId, + minLSN: LSN.fromString(LSN.ZERO), + createDate: new Date(), + pendingSchemaChanges: [] + }; +} - const refreshed = instance(40); - refreshed.minLSN = LSN.fromString('00000000:00000002:0000'); - const query = vi.fn().mockResolvedValue({ recordset: [] }); - const poller = new CDCPoller({ - connectionManager: { query } as any, - eventHandler: {} as any, - getReplicatedTables: () => [table], - startLSN: LSN.fromString(LSN.ZERO), - additionalConfig: { pollingIntervalMs: 1, pollingBatchSize: 1, trustServerCertificate: false } - }); - (poller as any).captureInstances = new Map([ - [ - 100, - { - sourceTable: { schema: 'dbo', name: 'users', objectId: 100 }, - instances: [refreshed] - } - ] - ]); +/** + * Build a persisted source-table candidate with optional capture metadata. + */ +function createSourceTableCandidate( + id: string, + metadata?: MSSQLSourceMetadata, + overrides: Partial[0]> = {} +): SourceTable { + return new SourceTable({ + id, + ref: { connectionTag: 'default', schema: 'dbo', name: 'users' }, + objectId: 100, + replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], + snapshotComplete: true, + bucketDataSources: [], + parameterLookupSources: [], + sourceMetadata: metadata, + ...overrides + }); +} - const bounds = { - startLSN: LSN.fromString('00000000:00000001:0000'), - endLSN: LSN.fromString('00000000:00000003:0000') - }; - await (poller as any).pollTable(table, bounds); +/** + * Create a source descriptor with optional identity overrides. + */ +function createSourceDescriptor(overrides: Partial = {}): SourceEntityDescriptor { + return { + connectionTag: 'default', + schema: 'dbo', + name: 'users', + objectId: 100, + replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], + ...overrides + }; +} - expect(bounds.startLSN).toBe(refreshed.minLSN); - expect(table.captureInstance).toBe(refreshed); - expect(table.pinnedCaptureObjectId).toBe(40); - }); -}); +function readyContext( + captureInstances: CaptureInstance[], + sourceDescriptor: SourceEntityDescriptor = createSourceDescriptor() +): MSSQLTableReconciliationContext { + return { + state: MSSQLTableReconciliationState.READY, + source: sourceDescriptor, + captureInstances + }; +} + +function unavailableContext( + state: MSSQLTableReconciliationState.TABLE_MISSING | MSSQLTableReconciliationState.CDC_DISABLED, + sourceDescriptor: SourceEntityDescriptor = createSourceDescriptor() +): MSSQLTableReconciliationContext { + return { state, source: sourceDescriptor }; +} From 678f2d9ce601fcb5d3d09f5ebd87cac7c25d3b37 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 12 Aug 2026 10:26:35 +0200 Subject: [PATCH 11/12] Removed redeclared fields in SourceEntityDescriptor --- packages/service-core/src/storage/SourceEntity.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/service-core/src/storage/SourceEntity.ts b/packages/service-core/src/storage/SourceEntity.ts index 9309fbf86..75984eab5 100644 --- a/packages/service-core/src/storage/SourceEntity.ts +++ b/packages/service-core/src/storage/SourceEntity.ts @@ -25,8 +25,6 @@ export interface SourceEntityDescriptor extends SourceTableRef { * If specified, this is specifically used to detect renames. */ objectId: number | string | undefined; - schema: string; - name: string; /** * The columns that are used to uniquely identify a record in the source entity. */ From 69bcb8e226be43a7b039132beaaba4f31fd0f0dc Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Wed, 12 Aug 2026 10:51:03 +0200 Subject: [PATCH 12/12] Split MSSQLSourceTable test cases to own test file and cleaned up CaptureReconciler tests --- .../test/src/CaptureReconciler.test.ts | 88 +------------------ .../test/src/MSSQLSourceTable.test.ts | 30 +++++++ modules/module-mssql/test/src/util.ts | 56 +++++++++++- 3 files changed, 87 insertions(+), 87 deletions(-) create mode 100644 modules/module-mssql/test/src/MSSQLSourceTable.test.ts diff --git a/modules/module-mssql/test/src/CaptureReconciler.test.ts b/modules/module-mssql/test/src/CaptureReconciler.test.ts index 11c3ebaca..9ee1b20af 100644 --- a/modules/module-mssql/test/src/CaptureReconciler.test.ts +++ b/modules/module-mssql/test/src/CaptureReconciler.test.ts @@ -1,15 +1,10 @@ import { CaptureInstance } from '@module/common/CaptureInstance.js'; -import { LSN } from '@module/common/LSN.js'; -import { MSSQLSourceTable } from '@module/common/MSSQLSourceTable.js'; -import { - createCaptureReconciler, - MSSQLSourceMetadata, - readCaptureMetadata -} from '@module/replication/CaptureReconciler.js'; +import { createCaptureReconciler, readCaptureMetadata } from '@module/replication/CaptureReconciler.js'; import type { MSSQLTableReconciliationContext } from '@module/replication/MSSQLTableReconciliationContext.js'; import { MSSQLTableReconciliationState } from '@module/replication/MSSQLTableReconciliationContext.js'; import { SourceEntityDescriptor, SourceTable } from '@powersync/service-core'; import { describe, expect, it } from 'vitest'; +import { createCaptureInstance, createSourceDescriptor, createSourceTableCandidate } from './util.js'; function reconcile(context: MSSQLTableReconciliationContext, candidates: SourceTable[]) { return createCaptureReconciler(context)({ source: context.source, candidates }); @@ -140,9 +135,7 @@ describe('createCaptureReconciler', () => { reconcile(readyContext([createCaptureInstance(50)], changedSource), [ createSourceTableCandidate('old', { captureTableObjectId: 40 }) ]) - ).toThrow( - /Table \[dbo\]\.\[users\] no longer matches the source table binding.*already-replicated data is retained/ - ); + ).toThrow(/Table \[dbo]\.\[users] no longer matches the source table binding.*already-replicated data is retained/); }); it('drops stale incompatible candidates when a compatible candidate anchors the binding', () => { @@ -159,81 +152,6 @@ describe('createCaptureReconciler', () => { }); }); -describe('MSSQLSourceTable.setCaptureInstance', () => { - it('sets the instance matching the persisted capture-table object id', () => { - const table = new MSSQLSourceTable(createSourceDescriptor(), [ - createSourceTableCandidate('a', { captureTableObjectId: 40 }) - ]); - const expected = createCaptureInstance(40); - - table.setCaptureInstance([createCaptureInstance(50), expected]); - - expect(table.captureInstance).toBe(expected); - }); - - it('sets null when the binding is legacy or the pinned instance is unavailable', () => { - const legacy = new MSSQLSourceTable(createSourceDescriptor(), [createSourceTableCandidate('legacy')]); - const pinned = new MSSQLSourceTable(createSourceDescriptor(), [ - createSourceTableCandidate('pinned', { captureTableObjectId: 40 }) - ]); - - legacy.setCaptureInstance([createCaptureInstance(40)]); - pinned.setCaptureInstance([createCaptureInstance(40)]); - pinned.setCaptureInstance([createCaptureInstance(50)]); - - expect(legacy.captureInstance).toBeNull(); - expect(pinned.captureInstance).toBeNull(); - }); -}); - -/** - * Create a capture instance with the given capture-table object id. - */ -function createCaptureInstance(objectId: number): CaptureInstance { - return { - name: `dbo_users_${objectId}`, - objectId, - minLSN: LSN.fromString(LSN.ZERO), - createDate: new Date(), - pendingSchemaChanges: [] - }; -} - -/** - * Build a persisted source-table candidate with optional capture metadata. - */ -function createSourceTableCandidate( - id: string, - metadata?: MSSQLSourceMetadata, - overrides: Partial[0]> = {} -): SourceTable { - return new SourceTable({ - id, - ref: { connectionTag: 'default', schema: 'dbo', name: 'users' }, - objectId: 100, - replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], - snapshotComplete: true, - bucketDataSources: [], - parameterLookupSources: [], - sourceMetadata: metadata, - ...overrides - }); -} - -/** - * Create a source descriptor with optional identity overrides. - */ -function createSourceDescriptor(overrides: Partial = {}): SourceEntityDescriptor { - return { - connectionTag: 'default', - schema: 'dbo', - name: 'users', - objectId: 100, - replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], - ...overrides - }; -} - function readyContext( captureInstances: CaptureInstance[], sourceDescriptor: SourceEntityDescriptor = createSourceDescriptor() diff --git a/modules/module-mssql/test/src/MSSQLSourceTable.test.ts b/modules/module-mssql/test/src/MSSQLSourceTable.test.ts new file mode 100644 index 000000000..c0850dbd5 --- /dev/null +++ b/modules/module-mssql/test/src/MSSQLSourceTable.test.ts @@ -0,0 +1,30 @@ +import { MSSQLSourceTable } from '@module/common/MSSQLSourceTable.js'; +import { describe, expect, it } from 'vitest'; +import { createCaptureInstance, createSourceDescriptor, createSourceTableCandidate } from './util.js'; + +describe('MSSQLSourceTable.setCaptureInstance', () => { + it('sets the instance matching the persisted capture-table object id', () => { + const table = new MSSQLSourceTable(createSourceDescriptor(), [ + createSourceTableCandidate('a', { captureTableObjectId: 40 }) + ]); + const expected = createCaptureInstance(40); + + table.setCaptureInstance([createCaptureInstance(50), expected]); + + expect(table.captureInstance).toBe(expected); + }); + + it('sets null when the binding is legacy or the pinned instance is unavailable', () => { + const legacy = new MSSQLSourceTable(createSourceDescriptor(), [createSourceTableCandidate('legacy')]); + const pinned = new MSSQLSourceTable(createSourceDescriptor(), [ + createSourceTableCandidate('pinned', { captureTableObjectId: 40 }) + ]); + + legacy.setCaptureInstance([createCaptureInstance(40)]); + pinned.setCaptureInstance([createCaptureInstance(40)]); + pinned.setCaptureInstance([createCaptureInstance(50)]); + + expect(legacy.captureInstance).toBeNull(); + expect(pinned.captureInstance).toBeNull(); + }); +}); diff --git a/modules/module-mssql/test/src/util.ts b/modules/module-mssql/test/src/util.ts index 7dd9f3984..7ff9ffe23 100644 --- a/modules/module-mssql/test/src/util.ts +++ b/modules/module-mssql/test/src/util.ts @@ -1,11 +1,12 @@ import * as types from '@module/types/types.js'; import { logger } from '@powersync/lib-services-framework'; -import { BucketStorageFactory, ReplicationCheckpoint, TestStorageConfig } from '@powersync/service-core'; - +import { BucketStorageFactory, ReplicationCheckpoint, storage, TestStorageConfig } from '@powersync/service-core'; import * as mongo_storage from '@powersync/service-module-mongodb-storage'; import * as postgres_storage from '@powersync/service-module-postgres-storage'; +import { CaptureInstance } from '@module/common/CaptureInstance.js'; import { LSN } from '@module/common/LSN.js'; +import { MSSQLSourceMetadata } from '@module/replication/CaptureReconciler.js'; import { MSSQLConnectionManager } from '@module/replication/MSSQLConnectionManager.js'; import { createCheckpoint, escapeIdentifier, getLatestLSN, toQualifiedTableName } from '@module/utils/mssql.js'; import sql from 'mssql'; @@ -275,3 +276,54 @@ export async function disableCDCForTable( { name: 'capture_instance', value: captureInstance } ]); } + +/** + * Create a capture instance with the given capture-table object id. + */ +export function createCaptureInstance(objectId: number): CaptureInstance { + return { + name: `dbo_users_${objectId}`, + objectId, + minLSN: LSN.fromString(LSN.ZERO), + createDate: new Date(), + pendingSchemaChanges: [] + }; +} +/** + * Build a persisted source-table candidate with optional capture metadata. + */ +export function createSourceTableCandidate( + id: string, + metadata?: MSSQLSourceMetadata, + overrides: Partial[0]> = {} +): storage.SourceTable { + const descriptor = createSourceDescriptor(); + + return new storage.SourceTable({ + id, + ref: descriptor, + objectId: descriptor.objectId, + replicaIdColumns: descriptor.replicaIdColumns, + snapshotComplete: true, + bucketDataSources: [], + parameterLookupSources: [], + sourceMetadata: metadata, + ...overrides + }); +} + +/** + * Create a source descriptor with optional identity overrides. + */ +export function createSourceDescriptor( + overrides: Partial = {} +): storage.SourceEntityDescriptor { + return { + connectionTag: 'default', + schema: 'dbo', + name: 'users', + objectId: 100, + replicaIdColumns: [{ name: 'id', type: 'int', typeId: 56 }], + ...overrides + }; +}