feat: align ws sync behavior with librocco

packages/ws-client/src/streams/OutboundStream.ts

@@ -10,6 +10,7 @@ export default class OutboundStream {
   #localOnly: boolean = false;
   #timeoutHandle: number | null = null;
   #bufferFullBackoff = 50;
+  #sentProbe = false;
   readonly #disposer;
 
   constructor(db: DB, transport: Transport) {
@@ -22,6 +23,7 @@ export default class OutboundStream {
     this.#lastSent = msg.since;
     this.#excludeSites = msg.excludeSites;
     this.#localOnly = msg.localOnly;
+    this.#sentProbe = false;
     // initial kickoff so we don't wait for a db change event
     this.#dbChanged();
   };
@@ -56,6 +58,15 @@ export default class OutboundStream {
     }
 
     if (changes.length == 0) {
+      if (!this.#sentProbe) {
+        this.#sentProbe = true;
+        this.#transport.sendChanges({
+          _tag: tags.Changes,
+          changes: [],
+          sender: this.#db.siteid,
+          since: lastSent,
+        });
+      }
       return;
     }
     const lastChange = changes[changes.length - 1];
@@ -74,6 +85,7 @@ export default class OutboundStream {
       switch (didSend) {
         case "sent":
           this.#bufferFullBackoff = 50;
+          this.#sentProbe = true;
           break;
         case "buffer-full":
           this.#lastSent = lastSent;

packages/ws-client/src/transport/Transport.ts

@@ -3,6 +3,7 @@ import {
   Changes,
   RejectChanges,
   StartStreaming,
+  SyncStatus,
 } from "@vlcn.io/ws-common";
 
 export type TransporOptions = {
@@ -32,6 +33,8 @@ export interface Transport {
 
   onResetStream: ((msg: StartStreaming) => Promise<void>) | null;
 
+  onSyncStatus?: ((msg: SyncStatus) => void) | null;
+
   close(): void;
 
   // Connection lifecycle callbacks

packages/ws-client/src/transport/WebSocketTransport.ts

@@ -7,6 +7,7 @@ import {
   decode,
   encode,
   tags,
+  type SyncStatus,
 } from "@vlcn.io/ws-common";
 
 export default class WebSocketTransport implements Transport {
@@ -118,6 +119,7 @@ export default class WebSocketTransport implements Transport {
   // Connection event callbacks
   onConnOpen: (() => void) | null = null;
   onConnClose: (() => void) | null = null;
+  onSyncStatus: ((msg: SyncStatus) => void) | null = null;
 
   #processEvent = (data: Uint8Array) => {
     const msg = decode(data);
@@ -143,6 +145,9 @@ export default class WebSocketTransport implements Transport {
           this.onStartStreaming && this.onStartStreaming(msg);
         }
         return;
+      case tags.SyncStatus:
+        this.onSyncStatus && this.onSyncStatus(msg);
+        return;
       case tags.Pong:
         // Right now pong is just a sign of life - handled above
         // TODO: we might implement additional data with the pong / heartbeat in the future

packages/ws-common/src/decode.ts

@@ -12,6 +12,7 @@ import {
   Pong,
   RejectChanges,
   StartStreaming,
+  SyncStatus,
   TagValues,
   tags,
   CreateDbOnPrimaryResponse,
@@ -103,6 +104,62 @@ export function decode(msg: Uint8Array): Msg {
         _reqid: decoding.readVarInt(decoder),
         err: decoding.readVarString(decoder),
       } satisfies Err;
+    case tags.SyncStatus: {
+      const ok = decoding.readUint8(decoder) === 1;
+
+      let siteId: Uint8Array | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        siteId = decoding.readVarUint8Array(decoder);
+      }
+
+      let schemaName: string | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        schemaName = decoding.readVarString(decoder);
+      }
+
+      let schemaVersion: bigint | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        schemaVersion = decoding.readBigInt64(decoder);
+      }
+
+      let schemaHash: string | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        schemaHash = decoding.readVarString(decoder);
+      }
+
+      let stage: SyncStatus["stage"];
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        stage = decoding.readVarString(decoder) as SyncStatus["stage"];
+      }
+
+      let ackDbVersion: bigint | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        ackDbVersion = decoding.readBigInt64(decoder);
+      }
+
+      let reason: string | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        reason = decoding.readVarString(decoder);
+      }
+
+      let message: string | undefined;
+      if (decoding.hasContent(decoder) && decoding.readUint8(decoder) === 1) {
+        message = decoding.readVarString(decoder);
+      }
+
+      return {
+        _tag: tags.SyncStatus,
+        ok,
+        siteId,
+        schemaName,
+        schemaVersion,
+        schemaHash,
+        stage,
+        ackDbVersion,
+        reason,
+        message,
+      } satisfies SyncStatus;
+    }
     default:
       tag as never;
   }

packages/ws-common/src/encode.ts

@@ -58,6 +58,50 @@ export function encode(msg: Msg): Uint8Array {
     case tags.Err:
       encoding.writeVarInt(encoder, msg._reqid);
       encoding.writeVarString(encoder, msg.err);
+      return encoding.toUint8Array(encoder);
+    case tags.SyncStatus:
+      encoding.writeUint8(encoder, msg.ok ? 1 : 0);
+
+      encoding.writeUint8(encoder, msg.siteId ? 1 : 0);
+      if (msg.siteId) {
+        encoding.writeVarUint8Array(encoder, msg.siteId);
+      }
+
+      encoding.writeUint8(encoder, msg.schemaName ? 1 : 0);
+      if (msg.schemaName) {
+        encoding.writeVarString(encoder, msg.schemaName);
+      }
+
+      encoding.writeUint8(encoder, msg.schemaVersion != null ? 1 : 0);
+      if (msg.schemaVersion != null) {
+        encoding.writeBigInt64(encoder, msg.schemaVersion);
+      }
+
+      encoding.writeUint8(encoder, msg.schemaHash ? 1 : 0);
+      if (msg.schemaHash) {
+        encoding.writeVarString(encoder, msg.schemaHash);
+      }
+
+      encoding.writeUint8(encoder, msg.stage ? 1 : 0);
+      if (msg.stage) {
+        encoding.writeVarString(encoder, msg.stage);
+      }
+
+      encoding.writeUint8(encoder, msg.ackDbVersion != null ? 1 : 0);
+      if (msg.ackDbVersion != null) {
+        encoding.writeBigInt64(encoder, msg.ackDbVersion);
+      }
+
+      encoding.writeUint8(encoder, msg.reason ? 1 : 0);
+      if (msg.reason) {
+        encoding.writeVarString(encoder, msg.reason);
+      }
+
+      encoding.writeUint8(encoder, msg.message ? 1 : 0);
+      if (msg.message) {
+        encoding.writeVarString(encoder, msg.message);
+      }
+
       return encoding.toUint8Array(encoder);
   }
 }

packages/ws-common/src/msgTypes.ts

@@ -3,6 +3,7 @@ export type Msg =
   | Changes
   | RejectChanges
   | StartStreaming
+  | SyncStatus
   | CreateDbOnPrimary
   | ApplyChangesOnPrimary
   | Ping
@@ -23,6 +24,7 @@ export const tags = {
   CreateDbOnPrimaryResponse: 9,
   Err: 10,
   ApplyChangesOnPrimaryResponse: 11,
+  SyncStatus: 12,
 } as const;
 
 export type Tags = typeof tags;
@@ -76,6 +78,21 @@ export type StartStreaming = Readonly<{
   localOnly: boolean;
 }>;
 
+export type SyncStatusStage = "handshake" | "steady" | "apply_ack";
+
+export type SyncStatus = Readonly<{
+  _tag: Tags["SyncStatus"];
+  ok: boolean;
+  siteId?: Uint8Array;
+  schemaName?: string;
+  schemaVersion?: bigint;
+  schemaHash?: string;
+  stage?: SyncStatusStage;
+  ackDbVersion?: bigint;
+  reason?: string;
+  message?: string;
+}>;
+
 export type CreateDbOnPrimary = Readonly<{
   _tag: Tags["CreateDbOnPrimary"];
   _reqid: number;