(MSSQL) Capture instance can change without triggering a re-snapshot after restart

[stevensJourney]

AI Disclosure: The following issue was generated with the help of Codex GPT 5.6 Sol. I found the issue while reviewing code, asked Codex to create an integration test to reproduce and then constructed this issue with context.

Context

For MSSQL, a schema change on a source database table is not automatically reflected in its
existing CDC change table. To start capturing the changed schema, the customer creates a new CDC
capture instance for the same source database table.

While a PowerSync replication process is running, it keeps the selected capture instance in memory.
The schema poller periodically queries the available capture instances and compares the latest one
with the instance currently attached to the in-memory source table. The
comparison uses the CDC change-table object ID.
If it detects a different ID, it treats that as a new capture instance,
drops the existing PowerSync source table data, and re-snapshots the source database table
using the new instance.

The capture instance itself is not persisted as part of the PowerSync source table identity. We
persist the SQL Server object ID of the physical source database table, but that ID is shared by all
CDC capture instances for the table. After a restart, PowerSync therefore cannot restore the
previously selected instance. It queries the currently available instances and
selects instances[0] as the capture instance.

Normally, starting with a new instance whose minimum LSN is ahead of the persisted replication LSN
causes a limited re-snapshot. The
retention check only marks a table when minLSN > checkpointLSN,
and checkSnapshotStatus
turns those marked tables into a limited re-snapshot.
This makes the restart appear safe in many cases because the new instance clearly does not contain
all changes after PowerSync's checkpoint. There is, however, a timing window where this LSN check
does not protect us.

Failure mode

PowerSync can be actively replicating from the original capture instance when the replacement
instance is created. The
production default checks for schema changes every 60 seconds,
while the MSSQL test context normally checks every 500 milliseconds. During that production polling
window, the original instance can continue processing changes and advance the persisted replication
LSN beyond the replacement instance's minimum LSN. When polling a capture table, PowerSync also
moves the table's lower polling bound up to its minimum LSN,
which is safe only when the selected capture instance is the one the stream is expected to use.

Consider a table whose original capture instance contains id and description columns. A new column named
new_column is added to the source database table, and a replacement capture instance is created
which also captures new_column. Before the next schema check, rows containing values for
new_column are inserted. PowerSync is still reading the original instance, so it replicates those
rows without new_column and commits an LSN after the replacement instance was created.

If the process keeps running, the next schema check detects the different capture instance and
explicitly re-snapshots the table. The problem occurs if PowerSync restarts before that check.

On restart, PowerSync no longer knows that the replication stream was using the original instance.
It resolves the physical source database table, selects the latest capture instance, and resumes
from the persisted LSN. Because replication had already advanced beyond the replacement instance's
minimum LSN, the retention check succeeds. PowerSync considers initial replication complete and does
not re-snapshot the table.

The replacement instance contains the earlier rows with their new_column values, but those changes
are before the persisted resume LSN and are skipped. The existing PowerSync data for those rows came
from the original instance and therefore has no new_column. Future rows are read from the
replacement instance and do contain new_column.

The result is that the stream silently switches capture schemas without reprocessing the existing
table data. Rows replicated shortly before the restart can remain present but permanently miss
values from newly captured columns, while rows replicated after the restart use the new schema.

Reproduction

The failure can be reproduced by using the production-length schema-check interval in the MSSQL
schema-change test. Replication is started with the original capture instance, a replacement
instance is created, and rows containing values for the newly added column are inserted. The test
waits for the original instance to replicate those rows and persist a checkpoint beyond the
replacement instance's minimum LSN, then restarts the replication context before schema polling
detects the replacement.

The following test reproduces the failure:

/**
 * Let the original capture instance replicate rows beyond a replacement instance's minimum LSN.
 * If a restart switches to the replacement without re-snapshotting, the rows remain present in
 * sync data but values from columns added to the replacement instance are missing.
 */
test('Restart after delayed capture-instance detection preserves newly captured column data', async () => {
  let rowsWithNewColumn: any[];

  // Establish the original schema, then delay schema detection while its capture instance
  // continues advancing checkpoints after a replacement instance has been created.
  {
    await using context = await CDCStreamTestContext.open(factory, {
      cdcStreamOptions: { schemaCheckIntervalMs: 60_000 }
    });
    await context.updateSyncRules(WILDCARD_SYNC_RULES);
    const { connectionManager } = context;

    await createTestTableWithBasicId(connectionManager, 'test_data');
    await insertBasicIdTestData(connectionManager, 'test_data');
    await context.replicateSnapshot();
    await context.startStreaming();
    await context.getFinalBucketState('global[]');

    await connectionManager.query(
      `ALTER TABLE ${toQualifiedTableName(connectionManager.schema, 'test_data')} ADD new_column INT`
    );
    await enableCDCForTable({
      connectionManager,
      table: 'test_data',
      captureInstance: 'capture_instance_new',
      // Unlike the original instance, the replacement also captures new_column.
      capturedColumns: ['id', 'description', 'new_column']
    });

    const { recordset } = await connectionManager.query(
      `
      INSERT INTO ${toQualifiedTableName(connectionManager.schema, 'test_data')} (description, new_column)
      OUTPUT INSERTED.id, INSERTED.description, INSERTED.new_column
      VALUES
        (@description1, @newColumn1),
        (@description2, @newColumn2)
      `,
      [
        { name: 'description1', type: sql.NVarChar(sql.MAX), value: 'first_new_column_row' },
        { name: 'newColumn1', type: sql.Int, value: 101 },
        { name: 'description2', type: sql.NVarChar(sql.MAX), value: 'second_new_column_row' },
        { name: 'newColumn2', type: sql.Int, value: 102 }
      ]
    );
    rowsWithNewColumn = recordset;

    // The old capture instance replicates these transactions and a later checkpoint, but cannot
    // include new_column because it was not part of that instance's captured schema.
    const stateBeforeRestart = await context.getFinalBucketState('global[]');
    for (const row of rowsWithNewColumn) {
      const replicated = stateBeforeRestart.find((operation) => operation.object_id === String(row.id));
      expect(replicated, `Row ${row.id} should be replicated before restart`).toBeDefined();
      expect(
        JSON.parse(replicated!.data!),
        `Row ${row.id} should initially use the original capture-instance schema`
      ).toEqual({ id: row.id, description: row.description });
    }
  }

  // Restarting selects the replacement instance, but its minimum LSN is behind the persisted
  // checkpoint. The existing implementation therefore skips both its history and a resnapshot.
  {
    await using replicationContext = await CDCStreamTestContext.open(factory, { doNotClear: true });
    await replicationContext.loadActiveSyncRules();
    await replicationContext.replicateSnapshot();
    await replicationContext.startStreaming();

    const finalState = await replicationContext.getFinalBucketState('global[]');
    for (const row of rowsWithNewColumn) {
      const replicated = finalState.find((operation) => operation.object_id === String(row.id));
      expect(replicated, `Row ${row.id} should remain present after restart`).toBeDefined();

      // This is the assertion that fails on main. The row is present, but new_column is missing.
      expect(
        JSON.parse(replicated!.data!),
        `Row ${row.id} should include data from the newly captured column after switching capture instances`
      ).toEqual({
        id: row.id,
        description: row.description,
        new_column: row.new_column
      });
    }
  }
});

The restart logs Initial replication already done, confirming that no limited re-snapshot was
scheduled. The affected rows are still present after restart, but their synchronized data contains
only id and description; the expected new_column values are missing. The marked assertion is
the one that fails on main, with an actual value such as
{ id: 2, description: "first_new_column_row" } instead of
{ id: 2, description: "first_new_column_row", new_column: 101 }. Changes written after the restart
are captured through the replacement instance and include new_column, confirming that the
capture-instance switch did occur.

This was difficult to expose with the existing tests because their 500 millisecond schema-check
interval usually detects and handles the new instance before the original instance can advance the
persisted LSN through the relevant window.

[stevensJourney]

This should be fixed by #727