Fix some post-merge conflicts.

Author: rkistner

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -146,10 +146,6 @@ export class MongoBucketBatch
     return this.last_checkpoint_lsn;
   }
 
-  get noCheckpointBeforeLsn() {
-    return this.no_checkpoint_before_lsn;
-  }
-
   async flush(options?: storage.BatchBucketFlushOptions): Promise<storage.FlushedResult | null> {
     let result: storage.FlushedResult | null = null;
     // One flush may be split over multiple transactions.

modules/module-mssql/src/replication/CDCStream.ts

@@ -304,7 +304,7 @@ export class CDCStream {
       const postSnapshotLSN = await getLatestLSN(this.connections);
       // Side note: A ROLLBACK would probably also be fine here, since we only read in this transaction.
       await transaction.commit();
-      const [updatedSourceTable] = await batch.markSnapshotDone([table.sourceTable], postSnapshotLSN.toString());
+      const [updatedSourceTable] = await batch.markTableSnapshotDone([table.sourceTable], postSnapshotLSN.toString());
       this.tableCache.updateSourceTable(updatedSourceTable);
     } catch (e) {
       await transaction.rollback();
@@ -500,11 +500,11 @@ export class CDCStream {
 
         // This will not create a consistent checkpoint yet, but will persist the op.
         // Actual checkpoint will be created when streaming replication caught up.
+        const postSnapshotLSN = await getLatestLSN(this.connections);
+        await batch.markAllSnapshotDone(postSnapshotLSN.toString());
         await batch.commit(snapshotLSN);
 
-        this.logger.info(
-          `Snapshot done. Need to replicate from ${snapshotLSN} to ${batch.noCheckpointBeforeLsn} to be consistent`
-        );
+        this.logger.info(`Snapshot done. Need to replicate from ${snapshotLSN} to ${postSnapshotLSN} to be consistent`);
       }
     );
   }

modules/module-mssql/test/src/CDCStream.test.ts

@@ -18,7 +18,9 @@ describe('CDCStream tests', () => {
   describeWithStorage({ timeout: 20_000 }, defineCDCStreamTests);
 });
 
-function defineCDCStreamTests(factory: storage.TestStorageFactory) {
+function defineCDCStreamTests(config: storage.TestStorageConfig) {
+  const { factory } = config;
+
   test('Initial snapshot sync', async () => {
     await using context = await CDCStreamTestContext.open(factory);
     const { connectionManager } = context;

modules/module-mssql/test/src/CDCStream_resumable_snapshot.test.ts

@@ -1,7 +1,7 @@
 import { describe, expect, test } from 'vitest';
 import { env } from './env.js';
 import { createTestTableWithBasicId, describeWithStorage, waitForPendingCDCChanges } from './util.js';
-import { TestStorageFactory } from '@powersync/service-core';
+import { TestStorageConfig, TestStorageFactory } from '@powersync/service-core';
 import { METRICS_HELPER } from '@powersync/service-core-tests';
 import { ReplicationMetric } from '@powersync/service-types';
 import * as timers from 'node:timers/promises';
@@ -10,19 +10,19 @@ import { CDCStreamTestContext } from './CDCStreamTestContext.js';
 import { getLatestLSN } from '@module/utils/mssql.js';
 
 describe.skipIf(!(env.CI || env.SLOW_TESTS))('batch replication', function () {
-  describeWithStorage({ timeout: 240_000 }, function (factory) {
+  describeWithStorage({ timeout: 240_000 }, function (config) {
     test('resuming initial replication (1)', async () => {
       // Stop early - likely to not include deleted row in first replication attempt.
-      await testResumingReplication(factory, 2000);
+      await testResumingReplication(config, 2000);
     });
     test('resuming initial replication (2)', async () => {
       // Stop late - likely to include deleted row in first replication attempt.
-      await testResumingReplication(factory, 8000);
+      await testResumingReplication(config, 8000);
     });
   });
 });
 
-async function testResumingReplication(factory: TestStorageFactory, stopAfter: number) {
+async function testResumingReplication(config: TestStorageConfig, stopAfter: number) {
   // This tests interrupting and then resuming initial replication.
   // We interrupt replication after test_data1 has fully replicated, and
   // test_data2 has partially replicated.
@@ -34,7 +34,9 @@ async function testResumingReplication(factory: TestStorageFactory, stopAfter: n
   //    have been / have not been replicated at that point is not deterministic.
   //    We do allow for some variation in the test results to account for this.
 
-  await using context = await CDCStreamTestContext.open(factory, { cdcStreamOptions: { snapshotBatchSize: 1000 } });
+  await using context = await CDCStreamTestContext.open(config.factory, {
+    cdcStreamOptions: { snapshotBatchSize: 1000 }
+  });
 
   await context.updateSyncRules(`bucket_definitions:
   global:
@@ -80,7 +82,7 @@ async function testResumingReplication(factory: TestStorageFactory, stopAfter: n
   }
 
   // Bypass the usual "clear db on factory open" step.
-  await using context2 = await CDCStreamTestContext.open(factory, {
+  await using context2 = await CDCStreamTestContext.open(config.factory, {
     doNotClear: true,
     cdcStreamOptions: { snapshotBatchSize: 1000 }
   });

modules/module-mssql/test/src/util.ts

@@ -1,6 +1,12 @@
 import * as types from '@module/types/types.js';
 import { logger } from '@powersync/lib-services-framework';
-import { BucketStorageFactory, InternalOpId, ReplicationCheckpoint, TestStorageFactory } from '@powersync/service-core';
+import {
+  BucketStorageFactory,
+  InternalOpId,
+  ReplicationCheckpoint,
+  TestStorageConfig,
+  TestStorageFactory
+} from '@powersync/service-core';
 
 import * as mongo_storage from '@powersync/service-module-mongodb-storage';
 import * as postgres_storage from '@powersync/service-module-postgres-storage';
@@ -20,11 +26,11 @@ export const INITIALIZED_MONGO_STORAGE_FACTORY = mongo_storage.test_utils.mongoT
   isCI: env.CI
 });
 
-export const INITIALIZED_POSTGRES_STORAGE_FACTORY = postgres_storage.test_utils.postgresTestStorageFactoryGenerator({
+export const INITIALIZED_POSTGRES_STORAGE_FACTORY = postgres_storage.test_utils.postgresTestSetup({
   url: env.PG_STORAGE_TEST_URL
 });
 
-export function describeWithStorage(options: TestOptions, fn: (factory: TestStorageFactory) => void) {
+export function describeWithStorage(options: TestOptions, fn: (config: TestStorageConfig) => void) {
   describe.skipIf(!env.TEST_MONGO_STORAGE)(`mongodb storage`, options, function () {
     fn(INITIALIZED_MONGO_STORAGE_FACTORY);
   });

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -111,10 +111,6 @@ export class PostgresBucketBatch
     return this.last_checkpoint_lsn;
   }
 
-  get noCheckpointBeforeLsn() {
-    return this.no_checkpoint_before_lsn;
-  }
-
   async [Symbol.asyncDispose]() {
     super.clearListeners();
   }

modules/module-postgres-storage/src/utils/test-utils.ts

@@ -94,7 +94,3 @@ export function postgresTestSetup(factoryOptions: PostgresTestStorageOptions) {
     tableIdStrings: true
   };
 }
-
-export function postgresTestStorageFactoryGenerator(factoryOptions: PostgresTestStorageOptions) {
-  return postgresTestSetup(factoryOptions).factory;
-}

modules/module-postgres/test/src/storage_combination.test.ts

@@ -7,9 +7,9 @@ describe.skipIf(!env.TEST_POSTGRES_STORAGE)('replication storage combination - p
   test('should allow the same Postgres cluster to be used for data and storage', async () => {
     // Use the same cluster for the storage as the data source
     await using context = await WalStreamTestContext.open(
-      postgres_storage.test_utils.postgresTestStorageFactoryGenerator({
+      postgres_storage.test_utils.postgresTestSetup({
         url: env.PG_TEST_URL
-      }),
+      }).factory,
       { doNotClear: false }
     );
 

packages/service-core/src/storage/BucketStorageBatch.ts

@@ -87,8 +87,6 @@ export interface BucketStorageBatch extends ObserverClient<BucketBatchStorageLis
   markTableSnapshotRequired(table: SourceTable): Promise<void>;
   markAllSnapshotDone(no_checkpoint_before_lsn: string): Promise<void>;
 
-  noCheckpointBeforeLsn: string;
-
   updateTableProgress(table: SourceTable, progress: Partial<TableSnapshotStatus>): Promise<SourceTable>;
 
   /**