Class SyncRPCServer

packages/rpc/src/server.ts:89–228

}

class SyncRPCServer extends RpcTarget implements SyncRPC {
  constructor(
    private readonly db: Database,
    private readonly options: Required<Pick<ServerOptions, "ignore">> &
      Pick<ServerOptions, "afterApply" | "beforeFetch">,
  ) {
    super();
    trackStub(this);
  }

  [Symbol.dispose](): void {
    untrackStub(this);
  }

  async push(input: {
    senderRev: number;
    changes: ReadableStream<ChangeEntry>;
  }): Promise<{ rev: number; appliedPushCursor: ChangeCursor }> {
    const entries: ChangeEntry[] = [];
    const reader = input.changes.getReader();
    try {
      while (true) {
        const { value, done } = await reader.read();
        if (done) break;
        entries.push(value);
      }
    } finally {
      reader.releaseLock();
    }
    // senderRev > 0 — the caller is a sync peer with its
    // own rev space; advance the fetch cursor to that point so
    // subsequent pulls and the cross-side invariant check see the
    // right appliedPushCursor. The apply path's alreadyApplied()
    // check is what stops the entries from ping-ponging back
    // through the sender's own coalesce + apply loop on the next
    // round trip.
    //
    // senderRev === 0 — the caller is an external writer
    // (an orchestrator using the wire as a transport, the
    // soak script, a manual curl). Treat the entries as
    // local writes: bump rev through the normal apply path,
    // leave pushRev untouched so the outbound sync loop
    // ships them upstream on the next tick.
    const isPeer = input.senderRev > 0;
    // Wrap the whole batch in a single transactionSync so a
    // mid-stream failure (e.g. a missing chunk in applyChangesSync's
    // assembly step) rolls back every prior entry. Without this
    // wrapper the receiver could be left with a subset of the
    // pushed entries committed.
    this.db.transactionSync(() => {
      applyChangesSync(this.db, entries, new Map(), {
        source: isPeer ? "upstream" : "local",
      });
      if (isPeer) {
        const nextCursor = { rev: input.senderRev, path: null };
        if (compareChangeCursors(nextCursor, readFetchCursor(this.db)) > 0) {
          writeFetchCursor(this.db, nextCursor);
        }