API index and search · Build metadata
Source snapshot
packages/remote/src/optimistic-source.ts
1// RemoteOptimisticSource: an `OptimisticSource` over a network transport — the ws sibling of2// the in-process native source (OPTIMISTIC-WRITES-DESIGN.md §8). The optimistic protocol is3// the normalized subscription stream with `cv`-stamped frames, plus two extras:4//5// - upstream: `init` (the connection's stable clientID, sent once) and `pushMutation`6// (one named-mutator envelope, §8.1) — confirmation rides the progress frames, so7// `pushMutation` resolves on send;8// - downstream: connection-level `progress` frames `{cvMin}` (§8.6; mutation confirmation is DATA — the lmid system query),9// relayed verbatim to the `OptimisticBackend` (which buffers data frames by `cv` and10// releases all `cv ≤ cvMin` as one coherent step, §8.5).11//12// Per-query validation is the ordinary `NormalizedSubscriber` (epoch/fp/seq); on a gap it13// re-subscribes, and the server re-registers under a NEW epoch and replies with a fresh14// `cv`-stamped snapshot + a progress frame that releases it.1516import { LMID_QUERY_NAME } from "@rindle/client";17import type {18 MutationEnvelope,19 MutationOutcomeFrame,20 NormalizedEvent,21 NormalizedTableSchema,22 OptimisticSource,23 ProgressFrame,24 QueryId,25 RemoteQuery,26} from "@rindle/client";2728import type { AffinityTicketStore } from "./affinity.ts";29import { NormalizedSubscriber } from "./normalized.ts";30import { retryDelayMs } from "./query-error.ts";31import type { NormalizedBatch, NormalizedHello } from "./normalized.ts";32import { ProtocolError } from "./protocol.ts";33import type { ServerMsg } from "./protocol.ts";34import {35 defaultSubscribeTarget,36 isThenable,37 subscribeMessage,38 type MutationEnvelopeSender,39 type SubscribeResolver,40 type SubscribeTarget,41} from "./subscribe.ts";42import { WsTransport } from "./transport.ts";43import type { Transport } from "./transport.ts";4445interface QState {46 remote: RemoteQuery;47 subscriber: NormalizedSubscriber | null;48 /** The epoch of the current subscription (0 before the first hello). */49 epoch: number;50 /** True between sending a re-subscribe and receiving its hello (so a second gap is ignored). */51 resubscribing: boolean;52 /** Monotonic token that cancels stale async lease resolutions. */53 subscribeTicket: number;54 /** Pending retryable-error re-subscribe timer (FOLLOWER-LAG-SHED §6.3), if any. */55 retryTimer: ReturnType<typeof setTimeout> | undefined;56 /** Consecutive retryable errors without a successful hello — the backoff exponent. */57 retryAttempt: number;58 /** Whether the LAST subscribe sent for this query presented a `leaseToken` — i.e. its hello59 * proves the connection is (re-)AUTHENTICATED (a room shell sets the socket's subject from the60 * first verified token; `pushMutation` requires it). After a reconnect on a lease-auth session,61 * queued pushes flush at the first such hello, never earlier — an envelope racing ahead of the62 * token subscribe would be refused by the shell's subject gate (H-v §7.5 rule 3). */63 authed: boolean;64}6566/** Build a transport to a follower's public ws endpoint (READ-ROUTER-DESIGN.md §2.3). Default67 * `(endpoint) => new WsTransport(endpoint)`. */68export type TransportFactory = (endpoint: string) => Transport;6970/** How the source obtains its transport:71 * - a pre-built {@link Transport} (or `{ transport }`) — FIXED: no endpoint migration, `wsEndpoint`72 * on leases is ignored (in-process / tests / a single static daemon);73 * - `{ factory, endpoint? }` — REPLACEABLE: transports are built on demand. An initial `endpoint`74 * (a static `wsUrl` or an SSR-injected bootstrap) opens eagerly; otherwise the first lease's75 * `wsEndpoint` opens it lazily, and a later lease naming a DIFFERENT endpoint migrates the whole76 * session there. */77export type RemoteOptimisticConnection =78 | Transport79 | { transport: Transport }80 | { factory: TransportFactory; endpoint?: string };8182export interface RemoteOptimisticSourceOptions {83 /** Resolve the upstream subscribe target. Defaults to embedded-server `{name,args}`. */84 resolveSubscribe?: SubscribeResolver;85 /** Override named-mutator delivery, e.g. POST envelopes to the app API server. */86 pushMutation?: MutationEnvelopeSender;87 /** Follower-affinity mode (FOLLOWER-AFFINITY-DESIGN.md §3): the shared ticket store. When set, the88 * source records the follower's minted ticket from the `{t:"affinity"}` frame and CLEARS it on a89 * sustained outage so the next (ticketless) reconnect anycasts to a live follower and re-pins90 * (§8). Absent ⇒ affinity off (today's behavior). */91 affinity?: AffinityTicketStore | (() => AffinityTicketStore | undefined);92}9394export class RemoteOptimisticSource implements OptimisticSource {95 /** The CURRENT transport (undefined in pure-lazy mode until the first lease opens one). */96 private transport: Transport | undefined;97 /** The ws endpoint the current transport points at (undefined for a fixed transport). */98 private currentEndpoint: string | undefined;99 /** Builds a transport for an endpoint; undefined ⇒ fixed transport (no migration). */100 private readonly transportFactory: TransportFactory | undefined;101 private readonly clientID: string;102 private readonly resolveSubscribe: SubscribeResolver;103 private readonly pushMutationSender?: MutationEnvelopeSender;104 /** Resolve the current affinity ticket store. A thunk lets the one-call client turn affinity on105 * after its first pure-lazy lease returns a placement ticket, before it opens the socket. */106 private readonly affinityStore: () => AffinityTicketStore | undefined;107 private handler: (qid: QueryId, ev: NormalizedEvent) => void = () => {};108 private progressHandler: (frame: ProgressFrame) => void = () => {};109 private restartHandler: () => void = () => {};110 private outcomeHandler: (frame: MutationOutcomeFrame) => void = () => {};111 private resyncHandler: () => void = () => {};112 /** Set by {@link resync} when this is a LEASE-AUTH session (some sub presented a `leaseToken`):113 * transport pushes queue in {@link pendingPushes} until the first authenticated re-subscribe's114 * hello re-establishes the socket's subject, then flush (see QState.authed). Never set on a115 * token-less (embedded/rindled) session — its pushes need no subject and go straight out. */116 private awaitingAuthedHello = false;117 /** Envelopes held while {@link awaitingAuthedHello} (H-v §7.5 rule 3). Every entry corresponds118 * to a still-pending backend mutation (the re-send reconstructs from pending entries; app119 * invokes in the window are pending by definition), so a superseding resync may CLEAR this —120 * its own re-send regenerates whatever still matters. */121 private pendingPushes: MutationEnvelope[] = [];122 private readonly subs = new Map<QueryId, QState>();123 /** Queries whose subscribe is waiting for a transport to exist — endpoint-less subscribes issued124 * in pure-lazy mode before any lease opens a transport (the lmid system query is registered by125 * the backend at construction). Flushed when a transport comes up. */126 private readonly deferred = new Set<QueryId>();127 /** One warning per source when an APP lease resolves without a `wsEndpoint` in pure-lazy mode —128 * nothing will ever open the transport, which is otherwise silent (views just stay empty). */129 private warnedEndpointlessLease = false;130 /** The daemon's boot id (from each `nhello`); a change means it restarted. */131 private lastBootId: string | undefined;132 /** The client's own typed per-table schemas, for hello validation (CRIT#4); set by the backend. */133 private clientTables: NormalizedTableSchema[] | undefined;134 /** Once true (set by {@link close}), in-flight lease resolutions are inert — they must not open a135 * new transport or send after teardown. */136 private closed = false;137 /** True once any lease has carried a routed `wsEndpoint`. Gates onDown re-leasing so a single138 * UNROUTED daemon keeps its pre-router behavior (recover via reconnect→resync only), not an extra139 * lease POST during an outage. */140 private sawRoutedEndpoint = false;141 /** Bumped at the start of every re-subscribe-all pass. A migrate triggered mid-pass starts a new142 * pass (higher generation); the outer pass then aborts instead of re-subscribing queries twice. */143 private resubscribeGen = 0;144145 constructor(connection: RemoteOptimisticConnection, clientID: string, opts: RemoteOptimisticSourceOptions = {}) {146 this.clientID = clientID;147 this.resolveSubscribe = opts.resolveSubscribe ?? defaultSubscribeTarget;148 this.pushMutationSender = opts.pushMutation;149 const affinity = opts.affinity;150 this.affinityStore = typeof affinity === "function" ? affinity : () => affinity;151 if (isTransport(connection)) {152 this.transportFactory = undefined;153 this.bringUp(connection, undefined);154 } else if ("transport" in connection) {155 this.transportFactory = undefined;156 this.bringUp(connection.transport, undefined);157 } else {158 this.transportFactory = connection.factory;159 if (connection.endpoint !== undefined) this.openEndpoint(connection.endpoint);160 }161 }162163 /** Wire a transport's handlers (no `init`). */164 private attach(transport: Transport): void {165 transport.onMessage((msg) => this.onServerMsg(msg));166 // Heal a dropped/restarted connection (same endpoint): on reconnect, replay init + re-subscribe.167 transport.onReconnect?.(() => {168 // The held ticket was useful for routing this ws handshake, but HTTP re-leases must wait for169 // THIS connection's first affinity frame. Otherwise an expired/rotated persisted ticket can170 // independently re-pin the control leg before the fresh frame arrives.171 this.affinityStore()?.connectionPending();172 this.resync();173 });174 // Sustained outage on this endpoint: re-lease (the router may move us off a dead follower, §3).175 transport.onDown?.(() => this.onDown());176 }177178 /** Make `transport` the current one, announce identity, and (re)subscribe anything deferred. */179 private bringUp(transport: Transport, endpoint: string | undefined): void {180 this.transport = transport;181 this.currentEndpoint = endpoint;182 // `WsTransport` has already evaluated the ticket thunk while constructing this connection.183 // From this point the old/persisted ticket is handshake-only until the follower confirms the184 // selected machine with its first affinity frame.185 this.affinityStore()?.connectionPending();186 this.attach(transport);187 transport.send({ t: "init", clientID: this.clientID });188 this.flushDeferred();189 }190191 /** Build + bring up a fresh transport to `endpoint` (replaceable mode only). */192 private openEndpoint(endpoint: string): void {193 if (!this.transportFactory) return;194 this.bringUp(this.transportFactory(endpoint), endpoint);195 }196197 /** Migrate the whole session to a new follower (§2.3): build the new transport, tear the old one198 * down, and re-subscribe EVERY active query there (re-leasing — the old tokens are199 * follower-local and invalid on the new node). */200 private migrate(endpoint: string): void {201 const old = this.transport;202 this.openEndpoint(endpoint);203 old?.close();204 this.resubscribeAll();205 }206207 /** Re-subscribe every live query on the current transport (each re-resolves its lease). A208 * re-subscribe can synchronously trigger a `migrate` (lease names a new endpoint), whose own209 * re-subscribe pass supersedes this one — the generation check then aborts this pass so a query210 * is never re-subscribed (and re-leased) twice. */211 private resubscribeAll(): void {212 const gen = ++this.resubscribeGen;213 for (const [qid, s] of this.subs) {214 if (this.resubscribeGen !== gen) return; // a nested migrate/resubscribe took over — stop215 s.subscriber = null;216 s.resubscribing = true;217 this.subscribe(qid, s.remote);218 }219 }220221 /** Flush subscribes deferred until a transport existed (e.g. the lmid query in pure-lazy mode). */222 private flushDeferred(): void {223 if (this.deferred.size === 0) return;224 const qids = [...this.deferred];225 this.deferred.clear();226 for (const qid of qids) {227 const s = this.subs.get(qid);228 if (s) this.subscribe(qid, s.remote);229 }230 }231232 /** The current follower's ws is sustainedly down — re-lease every query. The router returns a233 * (possibly new) `wsEndpoint`: a changed one migrates the session; an unchanged one re-subscribes234 * over the reconnecting transport (READ-ROUTER-DESIGN.md §3). No-op for an UNROUTED daemon (no235 * lease ever carried a `wsEndpoint`) — there is nowhere to move, so we keep the pre-router236 * behavior and let the transport's own reconnect→resync recover. */237 private onDown(): void {238 if (this.closed) return;239 const affinityStore = this.affinityStore();240 if (affinityStore) {241 // Affinity: the pinned follower is gone (sustained outage). Drop the ticket so the transport's242 // ongoing reconnects go TICKETLESS — the fleet edge then selects a live follower, which mints243 // a fresh ticket, and that reconnect's `onReconnect` → resync re-leases there (FOLLOWER-AFFINITY244 // §8, one bounded reassignment). Nothing to migrate: the ws host is fixed; the edge routes by ticket.245 affinityStore.clear();246 return;247 }248 if (!this.sawRoutedEndpoint) return;249 this.resubscribeAll();250 }251252 /** Tear down the current transport and make any in-flight lease resolution inert (a late lease253 * must NOT open a new transport after the consumer closed the client). */254 close(): void {255 this.closed = true;256 this.transport?.close();257 for (const s of this.subs.values()) {258 if (s.retryTimer !== undefined) clearTimeout(s.retryTimer);259 }260 this.subs.clear();261 this.deferred.clear();262 this.pendingPushes.length = 0;263 }264265 /** Register a handler fired when the DAEMON restarts (a new boot id) — the backend resets its266 * `cv` watermark so the new daemon's reset `cv` sequence is accepted instead of dropped. */267 onRestart(handler: () => void): void {268 this.restartHandler = handler;269 }270271 expectClientSchema(tables: NormalizedTableSchema[]): void {272 this.clientTables = tables;273 }274275 registerQuery(qid: QueryId, remote: RemoteQuery): void {276 this.subs.set(qid, {277 remote,278 subscriber: null,279 epoch: 0,280 resubscribing: false,281 subscribeTicket: 0,282 retryTimer: undefined,283 retryAttempt: 0,284 authed: false,285 });286 this.subscribe(qid, remote);287 }288289 unregisterQuery(qid: QueryId): void {290 const s = this.subs.get(qid);291 if (s?.retryTimer !== undefined) clearTimeout(s.retryTimer);292 this.subs.delete(qid);293 this.deferred.delete(qid);294 this.transport?.send({ t: "unsubscribe", queryId: qid });295 }296297 pushMutation(envelope: MutationEnvelope): Promise<void> {298 if (this.pushMutationSender) return Promise.resolve(this.pushMutationSender(envelope));299 // A lease-auth session that just reconnected is not yet re-authenticated (the shell's300 // `pushMutation` subject gate would refuse) — hold the envelope until the first token301 // re-subscribe's hello, then flush in order (H-v §7.5 rule 3).302 if (this.awaitingAuthedHello) {303 this.pendingPushes.push(envelope);304 return Promise.resolve();305 }306 this.transport?.send({ t: "pushMutation", envelope });307 return Promise.resolve();308 }309310 onNormalized(handler: (qid: QueryId, ev: NormalizedEvent) => void): void {311 this.handler = handler;312 }313314 onProgress(handler: (frame: ProgressFrame) => void): void {315 this.progressHandler = handler;316 }317318 /** The room deopt handshake's verdict stream (H-v). Dispatched OUT-OF-BAND on arrival — see319 * {@link onServerMsg}'s `mutationOutcome` arm for why it must never wait behind the cv buffer. */320 onMutationOutcome(handler: (frame: MutationOutcomeFrame) => void): void {321 this.outcomeHandler = handler;322 }323324 /** Fired once per re-established session, SYNCHRONOUSLY inside {@link resync} — before any325 * post-reconnect frame can release (the §7.5 rule-3 window: a replayed lmid snapshot must not326 * retire an entry whose outcome frame died with the old socket before the re-send captured327 * it). The backend re-sends the domain's unconfirmed pending envelopes with their original328 * mids; on a lease-auth session their DELIVERY is deferred until the first token hello329 * re-authenticates the socket ({@link pendingPushes}). */330 onResync(handler: () => void): void {331 this.resyncHandler = handler;332 }333334 // --- internals ---------------------------------------------------------------335336 private subscribe(qid: QueryId, remote: RemoteQuery): void {337 const s = this.subs.get(qid);338 if (!s) return;339 // A fresh subscribe (reconnect resync, gap recovery, the retry timer itself) supersedes any340 // scheduled retryable-error retry — never leave two subscribe paths racing for one query.341 if (s.retryTimer !== undefined) {342 clearTimeout(s.retryTimer);343 s.retryTimer = undefined;344 }345 const request = { queryId: qid, remote, mode: "normalized" as const };346 const ticket = ++s.subscribeTicket;347 const send = (target: SubscribeTarget) => {348 if (this.closed) return; // a lease that resolved after close() must not (re)open anything349 const cur = this.subs.get(qid);350 if (cur !== s || cur.subscribeTicket !== ticket) return;351 const endpoint = "leaseToken" in target ? target.wsEndpoint : undefined;352 s.authed = "leaseToken" in target; // a token subscribe (re-)authenticates the socket (H-v)353 if (endpoint !== undefined) this.sawRoutedEndpoint = true;354 if (this.transport) {355 if (endpoint !== undefined && endpoint !== this.currentEndpoint && this.transportFactory) {356 // The router placed this key on a DIFFERENT follower — migrate the whole session there.357 // `migrate` re-subscribes every query (incl. this one) over the new transport, so return.358 this.migrate(endpoint);359 return;360 }361 this.transport.send(subscribeMessage(request, target));362 return;363 }364 // No transport yet (pure-lazy): the first lease naming an endpoint opens it.365 if (endpoint !== undefined && this.transportFactory) {366 this.openEndpoint(endpoint);367 this.transport!.send(subscribeMessage(request, target));368 return;369 }370 // Endpoint-less with no transport (the lmid system query before the first lease): defer until371 // a transport comes up, then re-run this subscribe. An APP lease landing here means the372 // server never names an endpoint (e.g. an api-server with an explicit `daemon` and no373 // `rindle.wsUrl`) while the client has no `wsUrl` of its own — the deferral would be374 // permanent and silent, so say it once.375 if ("leaseToken" in target && !this.warnedEndpointlessLease) {376 this.warnedEndpointlessLease = true;377 console.warn(378 "[rindle-remote] a query lease returned no wsEndpoint and no daemon.wsUrl/transport is " +379 "configured — the live subscription cannot open. Configure the API server's " +380 "rindle.wsUrl (or pass daemon.wsUrl to createRindleClient).",381 );382 }383 this.deferred.add(qid);384 };385 const fail = (err: unknown) => {386 const cur = this.subs.get(qid);387 if (cur !== s || cur.subscribeTicket !== ticket) return;388 s.resubscribing = false;389 console.error(390 `[rindle-remote] optimistic query ${qid} subscribe resolution failed: ${String((err as Error)?.message ?? err)}`,391 );392 };393 try {394 // The reserved lmid system query is part of the optimistic WIRE contract, not an app395 // query: the server resolves it from the connection's own `init` identity. It must396 // never route through the app's subscribe resolver (the API server has no such named397 // query, and a lease for it would be meaningless).398 const target =399 remote.name === LMID_QUERY_NAME ? defaultSubscribeTarget(request) : this.resolveSubscribe(request);400 if (isThenable(target)) void target.then(send, fail);401 else send(target);402 } catch (err) {403 fail(err);404 }405 }406407 private onServerMsg(msg: ServerMsg): void {408 if (msg.t === "affinity") {409 // Connection-level: the follower minted/refreshed this connection's placement ticket. Persist410 // it (via the store) so the next connect offers it as a subprotocol and the lease POST forwards411 // it — both legs then pin THIS follower (§4). Off ⇒ no store ⇒ dropped.412 this.affinityStore()?.set(msg.ticket);413 return;414 }415 if (msg.t === "progress") {416 this.progressHandler(msg.frame);417 return;418 }419 if (msg.t === "mutationOutcome") {420 // OUT-OF-BAND BY DESIGN (H-v): the frame has no `cv`, so it must NEVER be routed through the421 // backend's cv buffer — dispatch immediately. A deopt has to migrate its pending entry to422 // the daemon stream BEFORE the buffered lmid release that would otherwise retire it as a423 // success (silence + lmid coverage ⇒ applied), and the §7.3 hold-back trigger — keyed on the424 // entry's confirming domain — would then park its staged writes the wrong way.425 this.outcomeHandler({426 mid: msg.mid,427 kind: msg.kind,428 ...(msg.reason !== undefined ? { reason: msg.reason } : {}),429 ...(msg.name !== undefined ? { name: msg.name } : {}),430 ...("args" in msg ? { args: msg.args } : {}),431 });432 return;433 }434 if (msg.t === "queryError") {435 this.onQueryError(msg.queryId, msg);436 return;437 }438 if (msg.t !== "nhello" && msg.t !== "nbatch") return;439 // Restart detection rides every nhello (connection-level) and runs BEFORE this query's440 // snapshot buffers, so the backend's reset clears stale state ahead of the fresh hydrate.441 if (msg.t === "nhello") this.observeBootId(msg.bootId);442 const s = this.subs.get(msg.queryId);443 if (!s) return; // unsubscribed / unknown query444 if (msg.t === "nhello") this.openSubscriber(msg.queryId, s, msg.hello);445 else this.applyBatch(msg.queryId, s, msg.batch);446 }447448 /** On reconnect: re-announce identity, fire the `onResync` re-send, and re-subscribe every live449 * query (each re-resolves its lease, so a restarted daemon re-materializes + re-leases on the450 * transiently). The re-send fires HERE — synchronously, before any post-reconnect frame can be451 * processed — because the §7.5 rule-3 window closes fast: the re-subscribed lmid stream's452 * fresh snapshot may cover a mid whose outcome frame died with the OLD socket, and once the453 * release retires that entry as an apparent success there is nothing left to re-send (the454 * lost-deopt write would silently vanish). Firing now captures the in-flight set intact; on a455 * lease-auth session the envelopes themselves are HELD ({@link pendingPushes}) until the first456 * token re-subscribe's hello re-authenticates the socket, then flush in order — so the shell's457 * subject gate never refuses them, and its re-answer (a recorded outcome for any non-applied458 * mid) resolves even an already-retired entry via the handshake's not-found arm. */459 private resync(): void {460 if (this.closed) return;461 this.transport?.send({ t: "init", clientID: this.clientID });462 // Lease-auth session ⇒ hold pushes until re-authed. A stale queue from a superseded resync is463 // cleared first: every held envelope maps to a still-pending mutation, and THIS resync's464 // re-send below regenerates whatever still matters (no loss, no stale duplicates).465 this.pendingPushes.length = 0;466 this.awaitingAuthedHello = [...this.subs.values()].some((s) => s.authed);467 this.resyncHandler();468 this.resubscribeAll();469 }470471 /** Track the daemon's boot id; a change (after the first) means it restarted — fire onRestart. */472 private observeBootId(bootId: string | undefined): void {473 if (!bootId) return;474 if (this.lastBootId !== undefined && bootId !== this.lastBootId) this.restartHandler();475 this.lastBootId = bootId;476 }477478 /** Route a `queryError` by its 101 §5 classification: retryable ⇒ keep the QState (and the479 * rows already folded downstream — 101 §6) and re-subscribe after a jittered backoff480 * honoring `retryAfterMs`; terminal (or pre-classification servers) ⇒ drop the481 * subscription, exactly as before. Fixes the stranded-client gap: a worker fault's or a482 * shedding follower's error now heals end-to-end (FOLLOWER-LAG-SHED §6.3). */483 private onQueryError(qid: QueryId, err: { message: string; code?: string; retryable?: boolean; retryAfterMs?: number }): void {484 const s = this.subs.get(qid);485 if (!s) return;486 if (err.retryable !== true) {487 if (s.retryTimer !== undefined) clearTimeout(s.retryTimer);488 this.subs.delete(qid);489 console.error(`[rindle-remote] optimistic query ${qid} subscription rejected: ${err.message}`);490 return;491 }492 if (s.retryTimer !== undefined) return; // a retry is already scheduled — don't stack them493 s.subscriber = null; // stop validating the dead epoch; recovery is a fresh seq-0 hydrate494 s.resubscribing = true;495 const delay = retryDelayMs(s.retryAttempt++, err.retryAfterMs);496 console.warn(497 `[rindle-remote] optimistic query ${qid} ${err.code ?? "error"} (retryable): re-subscribing in ${delay}ms: ${err.message}`,498 );499 const timer = setTimeout(() => {500 const cur = this.subs.get(qid);501 if (cur !== s) return;502 s.retryTimer = undefined;503 this.subscribe(qid, s.remote);504 }, delay);505 // Node returns a Timeout (unref keeps a retiring process from being pinned by a retry);506 // browsers return a number, where the optional call is a no-op.507 (timer as { unref?: () => void }).unref?.();508 s.retryTimer = timer;509 }510511 private openSubscriber(qid: QueryId, s: QState, hello: NormalizedHello): void {512 try {513 s.subscriber = new NormalizedSubscriber(hello, (ev) => this.handler(qid, ev), this.clientTables);514 s.epoch = hello.epoch;515 s.resubscribing = false;516 s.retryAttempt = 0; // a successful hello resets the retryable-error backoff517 // H-v §7.5 rule 3: the first AUTHENTICATED hello after a reconnect proves the socket is518 // re-authorized (the shell set its subject from the verified token — pushMutation-ready):519 // flush the held envelopes, in order. A token-less hello (the lmid system query) does not520 // qualify — an envelope racing ahead of the lease-token subscribe would be refused.521 if (this.awaitingAuthedHello && s.authed) {522 this.awaitingAuthedHello = false;523 for (const envelope of this.pendingPushes.splice(0)) {524 this.transport?.send({ t: "pushMutation", envelope });525 }526 }527 } catch (e) {528 // A comparator/fp mismatch at hello is unrecoverable (a code-contract divergence).529 s.subscriber = null;530 console.error(`[rindle-remote] optimistic query ${qid} subscription rejected: ${(e as Error).message}`);531 }532 }533534 private applyBatch(qid: QueryId, s: QState, batch: NormalizedBatch): void {535 if (!s.subscriber) return; // no hello yet (or mid re-hydrate)536 if (batch.epoch < s.epoch) return; // a stale batch from a superseded epoch — drop537 try {538 s.subscriber.apply(batch);539 } catch (e) {540 if (!(e instanceof ProtocolError)) throw e;541 if (s.resubscribing) return; // already recovering542 // Gap / drift → re-hydrate under a new epoch; the fresh snapshot arrives `cv`-stamped543 // and releases (re-hydrating the footprint) at the server's accompanying progress frame.544 s.resubscribing = true;545 s.subscriber = null;546 this.subscribe(qid, s.remote);547 }548 }549}550551/** Convenience: a `RemoteOptimisticSource` over a ws URL or a custom transport. A URL becomes a552 * replaceable connection seeded at that endpoint (so a routed lease can still migrate it); a553 * pre-built transport stays fixed. */554export function createRemoteOptimisticSource(555 urlOrTransport: string | Transport,556 clientID: string,557 opts: RemoteOptimisticSourceOptions = {},558): RemoteOptimisticSource {559 const connection: RemoteOptimisticConnection =560 typeof urlOrTransport === "string"561 ? { factory: (endpoint) => new WsTransport(endpoint), endpoint: urlOrTransport }562 : urlOrTransport;563 return new RemoteOptimisticSource(connection, clientID, opts);564}565566function isTransport(value: RemoteOptimisticConnection): value is Transport {567 return typeof (value as Transport).send === "function" && typeof (value as Transport).onMessage === "function";568}569