Fix snapshot_lsn handling.

Author: rkistner

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

@@ -736,6 +736,10 @@ export class MongoBucketBatch
                 },
                 '$last_checkpoint'
               ]
+            },
+            // Unset snapshot_lsn on checkpoint
+            snapshot_lsn: {
+              $cond: ['$_can_checkpoint', { $literal: null }, '$snapshot_lsn']
             }
           }
         },
@@ -1025,9 +1029,6 @@ export class MongoBucketBatch
         },
         $max: {
           no_checkpoint_before: no_checkpoint_before_lsn
-        },
-        $unset: {
-          snapshot_lsn: 1
         }
       },
       { session: this.session }

modules/module-mongodb/test/src/change_stream_utils.ts

@@ -22,7 +22,7 @@ import { clearTestDb, TEST_CONNECTION_OPTIONS } from './util.js';
 export class ChangeStreamTestContext {
   private _walStream?: ChangeStream;
   private abortController = new AbortController();
-  private streamPromise?: Promise<void>;
+  private streamPromise?: Promise<any>;
   public storage?: SyncRulesBucketStorage;
 
   /**
@@ -103,7 +103,7 @@ export class ChangeStreamTestContext {
     return this.storage!;
   }
 
-  get walStream() {
+  get streamer() {
     if (this.storage == null) {
       throw new Error('updateSyncRules() first');
     }
@@ -125,7 +125,7 @@ export class ChangeStreamTestContext {
   }
 
   async replicateSnapshot() {
-    await this.walStream.initReplication();
+    await this.streamer.initReplication();
   }
 
   /**
@@ -143,13 +143,17 @@ export class ChangeStreamTestContext {
   }
 
   startStreaming() {
-    return (this.streamPromise = this.walStream.streamChanges());
+    this.streamPromise = this.streamer.streamChanges().catch((e) => e);
   }
 
   async getCheckpoint(options?: { timeout?: number }) {
     let checkpoint = await Promise.race([
       getClientCheckpoint(this.client, this.db, this.factory, { timeout: options?.timeout ?? 15_000 }),
-      this.streamPromise
+      this.streamPromise?.then((e) => {
+        if (e != null) {
+          throw e;
+        }
+      })
     ]);
     if (checkpoint == null) {
       // This indicates an issue with the test setup - streamingPromise completed instead

packages/service-core-tests/src/tests/register-data-storage-data-tests.ts

@@ -1085,6 +1085,7 @@ bucket_definitions:
     const r = await f.configureSyncRules({ content: 'bucket_definitions: {}', validate: false });
     const storage = f.getInstance(r.persisted_sync_rules!);
     await storage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+      await batch.markAllSnapshotDone('1/0');
       await batch.keepalive('1/0');
     });