fix(zero-cache): handle subscribers ahead of the change-streamer change-log

packages/zero-cache/src/services/change-streamer/change-streamer-service.pg-test.ts

@@ -16,6 +16,7 @@ import {Queue} from '../../../../shared/src/queue.ts';
 import {Database} from '../../../../zqlite/src/db.ts';
 import {StatementRunner} from '../../db/statements.ts';
 import {testDBs} from '../../test/db.ts';
+import {stringify} from '../../types/bigint-json.ts';
 import type {PostgresDB} from '../../types/pg.ts';
 import type {Source} from '../../types/streams.ts';
 import {Subscription} from '../../types/subscription.ts';
@@ -102,7 +103,7 @@ describe('change-streamer/service', () => {
 
   async function nextChange(sub: Queue<Downstream>) {
     const down = await sub.dequeue();
-    assert(down[0] !== 'error');
+    assert(down[0] !== 'error', `Unexpected error ${stringify(down)}`);
     return down[1];
   }
 
@@ -302,6 +303,57 @@ describe('change-streamer/service', () => {
     ]);
   });
 
+  test('subscriber ahead of change log', async () => {
+    // Process some changes upstream.
+    changes.push(['begin', messages.begin(), {commitWatermark: '09'}]);
+    changes.push(['data', messages.insert('foo', {id: 'hello'})]);
+    changes.push(['data', messages.insert('foo', {id: 'world'})]);
+    changes.push([
+      'commit',
+      messages.commit({extra: 'stuff'}),
+      {watermark: '09'},
+    ]);
+
+    // Subscribe to a watermark from "the future".
+    const sub = await streamer.subscribe({
+      id: 'myid',
+      mode: 'serving',
+      watermark: '0b',
+      replicaVersion: REPLICA_VERSION,
+      initial: true,
+    });
+
+    // Process more upstream changes.
+    changes.push(['begin', messages.begin(), {commitWatermark: '0b'}]);
+    changes.push(['data', messages.delete('foo', {id: 'world'})]);
+    changes.push([
+      'commit',
+      messages.commit({more: 'stuff'}),
+      {watermark: '0b'},
+    ]);
+
+    // Finally something the subscriber hasn't seen.
+    changes.push(['begin', messages.begin(), {commitWatermark: '0c'}]);
+    changes.push(['data', messages.insert('foo', {id: 'voila'})]);
+    changes.push([
+      'commit',
+      messages.commit({something: 'new'}),
+      {watermark: '0c'},
+    ]);
+
+    // The subscriber should only see what's new to it.
+    const downstream = drainToQueue(sub);
+    expect(await nextChange(downstream)).toMatchObject({tag: 'begin'});
+    expect(await nextChange(downstream)).toMatchObject({
+      tag: 'insert',
+      new: {id: 'voila'},
+    });
+    expect(await nextChange(downstream)).toMatchObject({
+      tag: 'commit',
+      something: 'new',
+    });
+  });
+
   test('data types (forwarded and catchup)', async () => {
     const sub = await streamer.subscribe({
       id: 'myid',

packages/zero-cache/src/services/change-streamer/error-type-enum.ts

@@ -3,9 +3,7 @@
 export const Unknown = 0;
 export const WrongReplicaVersion = 1;
 export const WatermarkTooOld = 2;
-export const WatermarkNotFound = 3;
 
 export type Unknown = typeof Unknown;
 export type WrongReplicaVersion = typeof WrongReplicaVersion;
 export type WatermarkTooOld = typeof WatermarkTooOld;
-export type WatermarkNotFound = typeof WatermarkNotFound;

packages/zero-cache/src/services/change-streamer/storer.pg-test.ts

@@ -215,22 +215,6 @@ describe('change-streamer/storer', () => {
     ]);
   });
 
-  test('watermark not found', async () => {
-    // '123' is some watermark from the future.
-    const [sub, _, stream] = createSubscriber('123');
-    storer.catchup(sub);
-
-    expect(await drain(stream)).toEqual([
-      [
-        'error',
-        {
-          type: ErrorType.WatermarkNotFound,
-          message: 'cannot catch up from requested watermark 123',
-        },
-      ],
-    ]);
-  });
-
   test('queued if transaction in progress', async () => {
     const [sub1, _0, stream1] = createSubscriber('03');
     const [sub2, _1, stream2] = createSubscriber('06');

packages/zero-cache/src/services/change-streamer/storer.ts

@@ -239,10 +239,6 @@ export class Storer implements Service {
           }
         }
         if (watermarkFound) {
-          // Flushes the backlog of messages buffered during catchup and
-          // allows the subscription to forward subsequent messages immediately.
-          sub.setCaughtUp();
-
           this.#lc.info?.(
             `caught up ${sub.id} with ${count} changes (${
               Date.now() - start
@@ -251,13 +247,12 @@ export class Storer implements Service {
         } else {
           const lastStoredWatermark = await this.getLastStoredWatermark();
           this.#lc.warn?.(
-            `rejecting subscriber at watermark ${sub.watermark} (latest watermark: ${lastStoredWatermark})`,
-          );
-          sub.close(
-            ErrorType.WatermarkNotFound,
-            `cannot catch up from requested watermark ${sub.watermark}`,
+            `subscriber at watermark ${sub.watermark} is ahead of latest watermark: ${lastStoredWatermark}`,
           );
         }
+        // Flushes the backlog of messages buffered during catchup and
+        // allows the subscription to forward subsequent messages immediately.
+        sub.setCaughtUp();
       });
     } catch (err) {
       sub.fail(err);