Apply code review fixes

Author: KyleAMathews

packages/typescript-client/src/client.ts

@@ -1571,6 +1571,13 @@ export class ShapeStream<T extends Row<unknown> = Row>
     // Filter messages using snapshot tracker
     const messagesToProcess = batch.filter((message) => {
       if (isChangeMessage(message)) {
+        const changeLsn = message.headers.lsn
+        if (typeof changeLsn === `string` && changeLsn) {
+          // A quiet SSE stream may deliver its first later change before the
+          // next up-to-date boundary. Retire snapshots against that change's
+          // WAL position before resolving its wrapped transaction ID.
+          this.#snapshotTracker.lastSeenUpdate(BigInt(changeLsn))
+        }
         return !this.#snapshotTracker.shouldRejectMessage(message)
       }
 

packages/typescript-client/test/pbt-micro.test.ts

@@ -546,12 +546,16 @@ describe(`SnapshotTracker database LSN retirement`, () => {
             {
               key: `k1`,
               value: { id: 1, version: `duplicate` },
-              headers: { operation: `update`, txids: [679_865_407] },
+              headers: {
+                operation: `update`,
+                txids: [679_865_407],
+                lsn: `122`,
+              },
             },
             {
               headers: {
                 control: `up-to-date`,
-                global_last_seen_lsn: `123`,
+                global_last_seen_lsn: `100`,
               },
             },
           ]),
@@ -565,7 +569,11 @@ describe(`SnapshotTracker database LSN retirement`, () => {
             {
               key: `k1`,
               value: { id: 1, version: `future` },
-              headers: { operation: `update`, txids: [2_827_350_156] },
+              headers: {
+                operation: `update`,
+                txids: [2_827_350_156],
+                lsn: `124`,
+              },
             },
             {
               headers: {

packages/typescript-client/test/snapshot-tracker.test.ts

@@ -116,7 +116,7 @@ describe(`SnapshotTracker`, () => {
       }
       tracker.addSnapshot(metadata, new Set([`user:1`]))
 
-      const message: ChangeMessage<Row<unknown>> = {
+      const newerMessage: ChangeMessage<Row<unknown>> = {
         key: `user:1`,
         value: { id: 1, name: `Alice` },
         headers: {
@@ -125,9 +125,11 @@ describe(`SnapshotTracker`, () => {
         },
       }
 
-      expect(tracker.shouldRejectMessage(message)).toBe(false)
+      expect(tracker.shouldRejectMessage(newerMessage)).toBe(false)
 
-      const olderMessage: ChangeMessage<Row<unknown>> = {
+      // This would be rejected if the newer message had not retired the
+      // snapshot after resolving its wrapped xid against xmax.
+      const messageAfterRetirement: ChangeMessage<Row<unknown>> = {
         key: `user:1`,
         value: { id: 1, name: `Alice` },
         headers: {
@@ -136,7 +138,7 @@ describe(`SnapshotTracker`, () => {
         },
       }
 
-      expect(tracker.shouldRejectMessage(olderMessage)).toBe(false)
+      expect(tracker.shouldRejectMessage(messageAfterRetirement)).toBe(false)
     })
   })