(input: {
senderRev: number;
changes: ReadableStream<ChangeEntry>;
})
| 101 | } |
| 102 | |
| 103 | async push(input: { |
| 104 | senderRev: number; |
| 105 | changes: ReadableStream<ChangeEntry>; |
| 106 | }): Promise<{ rev: number; appliedPushCursor: ChangeCursor }> { |
| 107 | const entries: ChangeEntry[] = []; |
| 108 | const reader = input.changes.getReader(); |
| 109 | try { |
| 110 | while (true) { |
| 111 | const { value, done } = await reader.read(); |
| 112 | if (done) break; |
| 113 | entries.push(value); |
| 114 | } |
| 115 | } finally { |
| 116 | reader.releaseLock(); |
| 117 | } |
| 118 | // senderRev > 0 — the caller is a sync peer with its |
| 119 | // own rev space; advance the fetch cursor to that point so |
| 120 | // subsequent pulls and the cross-side invariant check see the |
| 121 | // right appliedPushCursor. The apply path's alreadyApplied() |
| 122 | // check is what stops the entries from ping-ponging back |
| 123 | // through the sender's own coalesce + apply loop on the next |
| 124 | // round trip. |
| 125 | // |
| 126 | // senderRev === 0 — the caller is an external writer |
| 127 | // (an orchestrator using the wire as a transport, the |
| 128 | // soak script, a manual curl). Treat the entries as |
| 129 | // local writes: bump rev through the normal apply path, |
| 130 | // leave pushRev untouched so the outbound sync loop |
| 131 | // ships them upstream on the next tick. |
| 132 | const isPeer = input.senderRev > 0; |
| 133 | // Wrap the whole batch in a single transactionSync so a |
| 134 | // mid-stream failure (e.g. a missing chunk in applyChangesSync's |
| 135 | // assembly step) rolls back every prior entry. Without this |
| 136 | // wrapper the receiver could be left with a subset of the |
| 137 | // pushed entries committed. |
| 138 | this.db.transactionSync(() => { |
| 139 | applyChangesSync(this.db, entries, new Map(), { |
| 140 | source: isPeer ? "upstream" : "local", |
| 141 | }); |
| 142 | if (isPeer) { |
| 143 | const nextCursor = { rev: input.senderRev, path: null }; |
| 144 | if (compareChangeCursors(nextCursor, readFetchCursor(this.db)) > 0) { |
| 145 | writeFetchCursor(this.db, nextCursor); |
| 146 | } |
| 147 | } |
| 148 | }); |
| 149 | if (this.options.afterApply !== undefined && entries.length > 0) { |
| 150 | try { |
| 151 | await this.options.afterApply(); |
| 152 | } catch (err) { |
| 153 | // Settle hook failures must not surface as push failures — |
| 154 | // the entries are already committed. Log so the operator |
| 155 | // notices a wedged shim, then return success. |
| 156 | console.warn("[SyncRPCServer] afterApply hook failed:", err); |
| 157 | } |
| 158 | } |
| 159 | return { |
| 160 | rev: currentRev(this.db), |
nothing calls this directly
no test coverage detected