( db: Database, remote: SyncRPC, backend: string | undefined, retried: boolean, )
| 83 | // A second divergence after the reset is a real protocol break, |
| 84 | // not a recoverable race, so we throw to surface it. |
| 85 | async function pullOnceImpl( |
| 86 | db: Database, |
| 87 | remote: SyncRPC, |
| 88 | backend: string | undefined, |
| 89 | retried: boolean, |
| 90 | ): Promise<ApplyResult> { |
| 91 | const after = readFetchCursor(db, backend); |
| 92 | const localPushRev = readWatermark(db, "pushRev", backend); |
| 93 | // fetchChanges hands back the remote's currentCursor (cursor we |
| 94 | // advance to after a clean drain), its appliedPushCursor (cross-side |
| 95 | // invariant check on the pull path), and the entry stream itself. |
| 96 | // One round-trip instead of the previous currentRev() + |
| 97 | // fetchChanges() pair. |
| 98 | // |
| 99 | // pullOnce owns the fetchChanges result envelope: it wraps a stream |
| 100 | // stub that sits in the exports table until disposed. The stream is |
| 101 | // fully consumed within this call (drained, cancelled, or abandoned |
| 102 | // on a throw), so a try/finally disposing the envelope covers every |
| 103 | // exit — the clean drain, the early-complete return, the cross-side |
| 104 | // invariant trip, and any throw inside the batch loop. Disposing the |
| 105 | // envelope tears down the contained stream stub, releasing the |
| 106 | // remote iterator. |
| 107 | const fetchResult = await remote.fetchChanges({ after }); |
| 108 | try { |
| 109 | const { currentCursor, appliedPushCursor } = fetchResult; |
| 110 | // Cross-side watermark divergence. Two shapes are recoverable: |
| 111 | // * appliedPushCursor.rev < localPushRev: the remote forgot |
| 112 | // what we pushed (typically a process-lifetime computerd restart |
| 113 | // while the WebSocket survived, so reconcileWatermarks on |
| 114 | // connect never re-ran). |
| 115 | // * currentCursor < after: the remote's log is shorter than we |
| 116 | // remember — same root cause, different symptom. |
| 117 | // Both are the inline equivalent of reconcileWatermarks: reset |
| 118 | // the divergent cursor to 0, cancel the in-flight stream, and |
| 119 | // retry once. The rev-0 baseline path in fetchChanges + pushOnce |
| 120 | // re-ships everything incrementally and the receiver's |
| 121 | // alreadyApplied() check absorbs the redundant work. |
| 122 | // |
| 123 | // The divergence test is rev-only on purpose. A same-rev partial |
| 124 | // appliedPushCursor (the remote applied part of the rev we |
| 125 | // pushed, then tore down mid-apply) is not a recoverable race — |
| 126 | // it means the receiver lost state inside a rev it told us it |
| 127 | // had. Resetting and replaying cannot mend that, so we let the |
| 128 | // assertion below surface it instead of looping. |
| 129 | // |
| 130 | // A second divergence after a reset is a real protocol break: |
| 131 | // surface it via the assertion below rather than loop. |
| 132 | const pushDiverged = appliedPushCursor.rev < localPushRev; |
| 133 | const fetchDiverged = compareChangeCursors(currentCursor, after) < 0; |
| 134 | if (!retried && (pushDiverged || fetchDiverged)) { |
| 135 | // Cancel the stream before disposing the envelope. For a real |
| 136 | // capnweb envelope the dispose alone is enough to tear down the |
| 137 | // backing stub, but the in-process server returns a plain |
| 138 | // ReadableStream wired to an async generator; without an |
| 139 | // explicit cancel the generator stays advanced (queue size 0 |
| 140 | // plus high-water mark 1 means pull() has already been called) |
| 141 | // and its query results sit in memory until GC. Cancel is |
| 142 | // best-effort: a real envelope may have already torn the stream |
no test coverage detected