diff --git a/.changeset/open-worms-unite.md b/.changeset/open-worms-unite.md new file mode 100644 index 000000000..beb8d8b98 --- /dev/null +++ b/.changeset/open-worms-unite.md @@ -0,0 +1,9 @@ +--- +'@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 +- CDC polling query now streams results diff --git a/modules/module-mssql/src/replication/CDCPoller.ts b/modules/module-mssql/src/replication/CDCPoller.ts index 897943f98..5a83aa369 100644 --- a/modules/module-mssql/src/replication/CDCPoller.ts +++ b/modules/module-mssql/src/replication/CDCPoller.ts @@ -17,13 +17,28 @@ import { CaptureInstanceMissingError } from './CaptureReconciler.js'; import { MSSQLConnectionManager } from './MSSQLConnectionManager.js'; import { SchemaChange, SchemaChangeType } from './SchemaChange.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 interface CDCEventHandler { onInsert: (row: any, table: MSSQLSourceTable, columns: sql.IColumnMetadata) => Promise; onUpdate: (rowAfter: any, rowBefore: any, table: MSSQLSourceTable, columns: sql.IColumnMetadata) => Promise; @@ -32,8 +47,6 @@ export interface CDCEventHandler { onSchemaChange: (change: SchemaChange) => Promise; } -export const DEFAULT_SCHEMA_CHECK_INTERVAL_MS = 60_000; - export interface CDCPollerOptions { connectionManager: MSSQLConnectionManager; eventHandler: CDCEventHandler; @@ -50,25 +63,31 @@ export interface CDCPollerOptions { schemaCheckIntervalMs?: number; } +/** + * 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(); } @@ -144,7 +163,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(); } @@ -152,9 +171,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...`); @@ -188,20 +207,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()}` @@ -218,7 +238,14 @@ export class CDCPoller { } } - private async pollTable(table: MSSQLSourceTable, bounds: { startLSN: LSN; endLSN: LSN }): Promise { + /** + * 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(); + // CDC cleanup can advance minLSN while the capture-table identity remains unchanged, so use // the latest metadata loaded by the schema check rather than the instance bound at startup. const availableInstances = this.captureInstances.get(table.objectId)?.instances ?? []; @@ -234,62 +261,50 @@ export class CDCPoller { } const minLSN = boundInstance.minLSN; if (minLSN > bounds.endLSN) { - return 0; + return transactionLSNs; } else if (minLSN >= bounds.startLSN) { bounds.startLSN = minLSN; } try { - const { recordset: results } = await this.connectionManager.query( - ` - SELECT * FROM ${table.allChangesFunction}(@from_lsn, @to_lsn, 'all update old') ORDER BY __$start_lsn, __$seqval - `, - [ - { name: 'from_lsn', type: sql.VarBinary, value: bounds.startLSN.toBinary() }, - { name: 'to_lsn', type: sql.VarBinary, value: bounds.endLSN.toBinary() } - ] - ); - - 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}`); + const request = await this.connectionManager.createRequest(); + request.input('from_lsn', sql.VarBinary, bounds.startLSN.toBinary()); + request.input('to_lsn', sql.VarBinary, bounds.endLSN.toBinary()); + + let columns: sql.IColumnMetadata | null = null; + request.on('recordset', (recordsetColumns) => { + columns = recordsetColumns; + }); + const stream = request.toReadableStream(); + request.query(` + SELECT * FROM ${table.allChangesFunction}(@from_lsn, @to_lsn, 'all update old') ORDER BY __$start_lsn, __$seqval, __$operation + `); + + for await (const { transactionLSN, type, rows } of groupLogicalChanges(stream, table)) { + 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); 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, 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, 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. Unlike the check above, this cannot // tell the two apart, so it stays recoverable: the forced schema check classifies it as a @@ -383,3 +398,92 @@ export class CDCPoller { return schemaChanges; } } + +/** + * 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` + */ +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; + } + current ??= { + transactionLSN, + sequence, + rows: [] + }; + current.rows.push(row); + } + + if (current) { + yield toLogicalChange(current.rows, current.transactionLSN, table); + } +} + +function toLogicalChange(rows: any[], startLSN: Buffer, table: MSSQLSourceTable): LogicalChange { + const transactionLSN = LSN.fromBinary(startLSN); + return { + transactionLSN, + type: resolveLogicalChangeType(rows, transactionLSN, table), + rows: rows + }; +} + +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; + + 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..51dde5dc5 --- /dev/null +++ b/modules/module-mssql/test/src/CDCPoller.test.ts @@ -0,0 +1,382 @@ +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 { SourceTable } from '@powersync/service-core'; +import sql from 'mssql'; +import timers from 'timers/promises'; +import { afterEach, beforeEach, describe, expect, test } from 'vitest'; +import { + clearTestDb, + createTestTableWithBasicId, + enableCDCForTable, + insertBasicIdTestData, + 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 createTestTableWithBasicId(connectionManager, tableName); + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + + const beforeInsertLSN = await getLatestLSN(connectionManager); + const initial = await insertBasicIdTestData(connectionManager, tableName); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + const startLSN = await getLatestReplicatedLSN(connectionManager); + + const beforeUpdateLSN = await getLatestLSN(connectionManager); + const replacementDescription = 'replacement'; + const { recordset } = await connectionManager.query( + ` + BEGIN TRAN; + 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); + const { operations } = await collectChanges({ + connectionManager, + tables: [table], + startLSN, + expectedOperationCount: 2 + }); + + expect(operations).toMatchObject([ + { 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 createTestTableWithBasicId(connectionManager, tableName); + const qualifiedName = toQualifiedTableName(connectionManager.schema, tableName); + + const beforeInsertLSN = await getLatestLSN(connectionManager); + const initial = await insertBasicIdTestData(connectionManager, tableName); + await waitForPendingCDCChanges(beforeInsertLSN, connectionManager); + + const startLSN = await getLatestReplicatedLSN(connectionManager); + + const beforeUpdateLSN = await getLatestLSN(connectionManager); + 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); + const { operations } = await collectChanges({ + connectionManager, + tables: [table], + startLSN, + expectedOperationCount: 1 + }); + + 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 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. + 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 + // 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: 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 }]); + }); +}); + +/** + * 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, + 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, including the persisted capture-instance binding + * that would normally be populated by source-table reconciliation. + */ +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 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; +} diff --git a/modules/module-mssql/test/src/CaptureReconciler.test.ts b/modules/module-mssql/test/src/CaptureReconciler.test.ts index f18d05dae..9ee1b20af 100644 --- a/modules/module-mssql/test/src/CaptureReconciler.test.ts +++ b/modules/module-mssql/test/src/CaptureReconciler.test.ts @@ -1,82 +1,10 @@ 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, - 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, 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'; +import { createCaptureInstance, createSourceDescriptor, createSourceTableCandidate } from './util.js'; function reconcile(context: MSSQLTableReconciliationContext, candidates: SourceTable[]) { return createCaptureReconciler(context)({ source: context.source, candidates }); @@ -99,7 +27,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 +38,7 @@ describe('createCaptureReconciler', () => { reconcile( unavailableContext( MSSQLTableReconciliationState.TABLE_MISSING, - source({ objectId: undefined, replicaIdColumns: [] }) + createSourceDescriptor({ objectId: undefined, replicaIdColumns: [] }) ), [] ) @@ -122,9 +50,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 +64,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 +79,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 +104,44 @@ 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 })]) - ).toThrow( - /Table \[dbo\]\.\[users\] no longer matches the source table binding.*already-replicated data is retained/ - ); + 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 }] }) ]); @@ -214,64 +152,20 @@ 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); - - table.setCaptureInstance([instance(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 })]); - - legacy.setCaptureInstance([instance(40)]); - pinned.setCaptureInstance([instance(40)]); - pinned.setCaptureInstance([instance(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]); - - 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] - } - ] - ]); - - const bounds = { - startLSN: LSN.fromString('00000000:00000001:0000'), - endLSN: LSN.fromString('00000000:00000003:0000') - }; - await (poller as any).pollTable(table, bounds); +function readyContext( + captureInstances: CaptureInstance[], + sourceDescriptor: SourceEntityDescriptor = createSourceDescriptor() +): MSSQLTableReconciliationContext { + return { + state: MSSQLTableReconciliationState.READY, + source: sourceDescriptor, + captureInstances + }; +} - expect(bounds.startLSN).toBe(refreshed.minLSN); - expect(table.captureInstance).toBe(refreshed); - expect(table.pinnedCaptureObjectId).toBe(40); - }); -}); +function unavailableContext( + state: MSSQLTableReconciliationState.TABLE_MISSING | MSSQLTableReconciliationState.CDC_DISABLED, + sourceDescriptor: SourceEntityDescriptor = createSourceDescriptor() +): MSSQLTableReconciliationContext { + return { state, source: sourceDescriptor }; +} 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 + }; +} 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. */