diff --git a/WhatNext - docs/01 concepts/RxDB-Replication.md b/WhatNext - docs/01 concepts/RxDB-Replication.md index 13cc214..7bb8654 100644 --- a/WhatNext - docs/01 concepts/RxDB-Replication.md +++ b/WhatNext - docs/01 concepts/RxDB-Replication.md @@ -3,7 +3,7 @@ tags: - data/rxdb/replication - core/net/p2p/protocols/replication date created: Saturday, February 14th 2026, 11:36:37 am -date modified: Monday, March 9th 2026, 12:20:45 am +date modified: Saturday, June 27th 2026, 12:00:00 pm --- # RxDB Replication Protocol @@ -12,11 +12,13 @@ date modified: Monday, March 9th 2026, 12:20:45 am A custom P2P replication protocol (`/whatnext/rxdb-replication/1.0.0`) that synchronizes RxDB documents between WhatNext peers over libp2p streams. It uses checkpoint-based sync with Last-Write-Wins (LWW) conflict resolution. -> ⚠️ **Reliability status (2026-06-27)** — the protocol works on the happy path but has un-hardened failure modes (see [[report-260627-mvp-state-of-the-union]] §3, tracked as issues N5–N7): -> - **Checkpoints are in-memory only** (`p2p-service.ts:71`) → **full resync of every collection on each app launch.** -> - **5s pull timeout resolves empty silently** (`p2p-service.ts:553–556`) → slow/loaded peers **silently drop remote changes.** -> - **LWW uses string comparison, not timestamp parsing** (`replication-handler.ts:60`) → works for ISO-8601 by coincidence; brittle to format drift or clock skew. -> - **Thin reconnection**: retries the same relay only (`relay-manager.ts:103`, 10s fixed, max 5) — no backoff, alternate-relay fallback, or heartbeat. **Zero tests.** +> ✅ **Reliability hardening (2026-06-27, epic [[epic-replication-reliability]], #40/#41/#42/#47/#32)** — the failure modes flagged in [[report-260627-mvp-state-of-the-union]] §3 are now addressed. **The wire protocol id stays `/whatnext/rxdb-replication/1.0.0` — no version bump.** The message types and on-wire shapes are unchanged; all fixes are local behavior (durability, timeout policy, conflict parsing, flow control) plus a previously-missing *requester-side* handler for the existing `pull-response` message. Specifics below: +> - **Durable checkpoints** — persisted to a debounced JSON file under `userData`; relaunch resumes incrementally instead of full-resyncing every collection (§ Checkpoint-Based Sync). +> - **No silent empty-resolve on pull timeout** — a timed-out pull now rejects and does **not** advance the checkpoint, so missed changes are re-pulled (§ Pull Timeout Policy). +> - **Skew-aware LWW** — timestamps are parsed to epoch-ms with a deterministic tie-break, not string-compared (§ Conflict Resolution). +> - **Relay reconnect** — exponential backoff + jitter, single alternate-relay fallback, and a liveness heartbeat for half-open links (§ Relay Reconnection). +> - **Fileshare backpressure** — chunk sends respect stream drain so >10MB transfers don't reset (`app/src/utility/protocols/file-transfer.ts`). +> - **Test coverage** — a Vitest P2P harness now covers handshake, replication, LWW, checkpoints, backoff/fallback/heartbeat, and large-file integrity. ## Why We Use It @@ -59,19 +61,45 @@ Remote Renderer (RxDB) | `push` | Outbound | Send local changes to remote peer | | `push-ack` | Inbound | Acknowledgment of received push | +> These four message types and their shapes are **unchanged** by the 2026-06-27 hardening — the protocol id remains `/whatnext/rxdb-replication/1.0.0`. Previously the requester *opened* `pull-request` streams but never consumed the resulting `pull-response` (it was logged and dropped), so pulled documents never reached the renderer and the checkpoint never advanced. The handler now invokes an `onPullResponse` callback that forwards the documents to the renderer and persists the returned checkpoint. This is a behavior fix on an existing message, not a new message type. + ### Checkpoint-Based Sync -Each peer tracks a checkpoint (ISO timestamp) per collection per remote peer. On pull: +Each peer tracks a checkpoint (ISO timestamp) per collection per remote peer, keyed `"peerId:collection"`. On pull: 1. Send `pull-request` with the last known checkpoint -2. Remote peer returns all documents modified after that checkpoint -3. Local peer updates its checkpoint to the returned value +2. Remote peer (responder) returns documents modified after that checkpoint, and a new checkpoint set to the **newest `updatedAt` among the returned documents** (not wall-clock "now", which would skip docs written between the newest returned doc and now). If no document carries a usable timestamp, the incoming checkpoint is echoed back so it never moves backward. +3. Local peer (requester) applies the documents and persists the returned checkpoint. + +**Durability (#40).** Checkpoints are persisted so a relaunch resumes incrementally instead of full-resyncing every collection from every peer: +- **Store**: a single JSON file, `replication-checkpoints.json` (flat map of `"peerId:collection"` → checkpoint string), in the Electron `userData` directory (the same location as `relay-config.json`). +- **Ownership**: the *utility* process owns reads/writes (it holds the in-flight checkpoint state). It cannot call `app.getPath('userData')` (no `electron` module), so it derives the `userData` path from platform conventions for app name `WhatNext`, with the `WHATNEXT_CHECKPOINT_PATH` env var as an explicit override (used by tests; available to main if it ever wants to inject the exact path). Decision rationale: a small JSON file is simpler and decoupled from the very DB being replicated — choosing the alternative (a dedicated RxDB collection in the renderer) would couple checkpoint durability to renderer liveness and the replicated dataset. See [[epic-replication-reliability]] §#40 open question. +- **Write policy**: writes are **debounced** (default 1s) and done atomically (temp file + rename) so a crash mid-write can't corrupt the file. Pending writes are flushed on node stop. +- **Degradation**: a missing or corrupt file is tolerated — the store starts empty, which falls back to the original full-resync behavior. Degraded, never a crash. + +### Pull Timeout Policy (#41) + +When a peer answers a `pull-request`, the responder asks its own renderer (over IPC) for the matching documents. That round-trip is bounded by `P2P_CONFIG.REPLICATION.PULL_TIMEOUT` (**15000 ms**, deliberately longer than the old hardcoded 5s to tolerate slow/large collections). + +The old behavior resolved the timeout with an **empty document array and a fresh `now` checkpoint**, which made a slow peer look like it had *no changes* and silently advanced the requester past unsent data. Now a timeout **rejects**: the responder returns an empty document set **and echoes back the requester's incoming checkpoint** (it does **not** advance it), so the missed changes are simply re-pulled on the next attempt rather than dropped. + +### Relay Reconnection (#41) + +Relay reconnect (see [[Circuit-Relay]], `relay-manager.ts`) no longer hammers a downed relay on a fixed 10s interval: +- **Exponential backoff with jitter** — `delay = min(RETRY_MAX_DELAY, RETRY_BASE_DELAY · BACKOFF_FACTOR^attempt)`, then "equal jitter" randomizes within `[d/2, d]` so many peers don't reconnect in lockstep. Defaults: base 1s, factor 2, ceiling 30s, jitter 0.5. +- **Single alternate-relay fallback** — after exhausting `MAX_RETRIES` (5) against the active relay *and* with no live relay, the next configured relay address is tried (ring order). Multi-relay orchestration/selection policy remains out of scope. +- **Liveness heartbeat** — every `HEARTBEAT_INTERVAL` (15s) the manager checks the active relay is still among the node's live connections; a half-open link that never fired a `close` event is treated as a disconnect and triggers reconnect. + +> `RETRY_INTERVAL` (the old fixed 10s) is retained in config as `@deprecated` for backward-compat but is no longer read by `RelayManager`. ### Conflict Resolution: LWW -Currently using Last-Write-Wins based on the `updatedAt` timestamp: -- When two peers modify the same document, the version with the later `updatedAt` wins -- This is simple but can lose data in rare concurrent-edit scenarios -- Future: migrate to CRDTs for true eventual consistency without data loss +Last-Write-Wins based on the `updatedAt` timestamp, **hardened to be skew-aware (#42)** in `app/src/renderer/db/lww.ts`: +- Both sides' `updatedAt` (with `addedAt` as a fallback for older docs) are **parsed to epoch milliseconds** before comparison — not string-compared. String comparison only sorted ISO-8601 correctly when format, timezone, and fractional-second precision matched exactly across peers; any drift silently corrupted the merge. +- **Unparseable/missing timestamps** resolve to `0` (oldest) rather than throwing, so they always lose to a real timestamp and never crash replication. +- **Tie-break**: on an exact timestamp tie, a deterministic content key (`stableStringify` — JSON with sorted keys, computed identically on every peer) is compared lexicographically; the larger key wins. Equal timestamp **and** equal key ⇒ indistinguishable ⇒ keep existing (no write). This guarantees two peers converge on the *same* winner instead of ping-ponging. + - **Device-local fields are excluded from the content key.** The sender strips device-local fields (`localFilePath`, `localFileSize`, `albumArtLocalPath`, `coverArtLocalPath`) from the transmitted payload, but they survive on the receiver's stored doc. `contentKey` therefore drops the same `DEVICE_LOCAL_FIELDS` set (single source of truth in `schemas.ts`, consumed by both the sender and `lww.contentKey`) so both peers compute the key over the *identical* field set. Without this, the existing-side key would carry e.g. `localFilePath` (which sorts before `name`), making both peers elect the incoming version on a tie — they would swap user content and oscillate. +- **Clock-skew posture**: LWW fundamentally trusts wall clocks. We do **not** clamp implausibly-future timestamps in the MVP — a peer with a fast clock can win conflicts it shouldn't. This is an accepted limitation; the migration path is CRDTs (Phase 2). This loses data only in rare concurrent-edit / skewed-clock scenarios. +- Future: migrate to CRDTs for true eventual consistency without data loss. ## Key Patterns @@ -109,7 +137,8 @@ window.electron.replication.onReplicationChanges((data) => { - __Data lives in renderer__: RxDB runs in the renderer process, but P2P runs in the utility process. All data must be relayed through IPC (renderer -> main -> utility and back). - __Stream-per-message__: Each replication message opens a new stream. This is simple but may not scale well for high-frequency updates. Future optimization: use persistent streams. -- __Clock skew__: LWW depends on accurate timestamps. If peer clocks are significantly skewed, the wrong version may win. +- __Clock skew__: LWW depends on accurate timestamps. Comparison is now epoch-ms-parsed and tie-broken deterministically (see § Conflict Resolution), but if peer clocks are significantly skewed the wrong version can still win — that is inherent to LWW and is the motivation for the CRDT migration. +- __Checkpoint must not jump to "now"__: when answering a pull, advance the checkpoint to the newest *returned document's* `updatedAt`, never wall-clock `now` — otherwise docs written between the newest returned doc and now are skipped. On timeout, do not advance at all. ## Related Concepts @@ -121,8 +150,14 @@ window.electron.replication.onReplicationChanges((data) => { ## References -- Protocol implementation: `app/src/utility/protocols/replication.ts` +- Protocol implementation: `app/src/utility/protocols/replication.ts` (`newestCheckpoint`, `onPullResponse`) +- Pull timeout + checkpoint persistence wiring: `app/src/utility/p2p-service.ts` +- Durable checkpoint store: `app/src/utility/checkpoint-store.ts` +- Skew-aware LWW: `app/src/renderer/db/lww.ts`, applied in `app/src/renderer/db/replication-handler.ts` +- Relay backoff/fallback/heartbeat: `app/src/utility/relay-manager.ts`, `app/src/utility/backoff.ts`; config in `app/src/shared/p2p-config.ts` (`P2P_CONFIG.RELAY`, `P2P_CONFIG.REPLICATION`) +- Tests: `app/src/utility/**/__tests__/`, `app/src/renderer/db/__tests__/lww.test.ts` - IPC types: `app/src/shared/core/ipc-protocol.ts` (ReplicationPushPayload, etc.) - Preload API: `app/src/main/preload.ts` (replication namespace) +- Epic: [[epic-replication-reliability]] (#40, #41, #42, #47, #32) - [RxDB Replication Protocol docs](https://rxdb.info/replication.html) diff --git a/app/src/renderer/db/__tests__/lww.test.ts b/app/src/renderer/db/__tests__/lww.test.ts new file mode 100644 index 0000000..a348cd8 --- /dev/null +++ b/app/src/renderer/db/__tests__/lww.test.ts @@ -0,0 +1,234 @@ +import { describe, it, expect } from 'vitest'; +import { parseTimestampMs, resolveTimestampMs, incomingWins, stableStringify, contentKey } from '../lww'; + +describe('parseTimestampMs', () => { + it('parses an ISO-8601 string to epoch ms', () => { + expect(parseTimestampMs('2026-06-27T00:00:00.000Z')).toBe(Date.parse('2026-06-27T00:00:00.000Z')); + }); + + it('parses ISO strings of differing precision to the same instant', () => { + // The exact bug string comparison missed: same instant, different precision. + const a = parseTimestampMs('2026-06-27T12:00:00Z'); + const b = parseTimestampMs('2026-06-27T12:00:00.000Z'); + expect(a).toBe(b); + }); + + it('accepts a numeric epoch-ms value as-is', () => { + expect(parseTimestampMs(1_700_000_000_000)).toBe(1_700_000_000_000); + }); + + it.each([undefined, null, '', ' ', 'not-a-date', {}, NaN])( + 'maps unparseable/missing value %p to 0', + (value) => { + expect(parseTimestampMs(value as unknown)).toBe(0); + } + ); +}); + +describe('resolveTimestampMs', () => { + it('prefers updatedAt over addedAt', () => { + const ms = resolveTimestampMs({ + updatedAt: '2026-06-27T00:00:00.000Z', + addedAt: '2020-01-01T00:00:00.000Z', + }); + expect(ms).toBe(Date.parse('2026-06-27T00:00:00.000Z')); + }); + + it('falls back to addedAt when updatedAt is missing', () => { + const ms = resolveTimestampMs({ addedAt: '2020-01-01T00:00:00.000Z' }); + expect(ms).toBe(Date.parse('2020-01-01T00:00:00.000Z')); + }); + + it('returns 0 when neither field is parseable', () => { + expect(resolveTimestampMs({ updatedAt: 'x', addedAt: 'y' })).toBe(0); + }); +}); + +describe('incomingWins', () => { + it('lets the strictly-later timestamp win', () => { + expect( + incomingWins( + { updatedAt: '2026-06-27T00:00:01.000Z' }, + { updatedAt: '2026-06-27T00:00:00.000Z' } + ) + ).toBe(true); + }); + + it('keeps the existing version when incoming is older', () => { + expect( + incomingWins( + { updatedAt: '2026-06-27T00:00:00.000Z' }, + { updatedAt: '2026-06-27T00:00:01.000Z' } + ) + ).toBe(false); + }); + + it('treats differing-precision equal instants as a tie (not a win)', () => { + // Pure string compare would treat these as unequal; epoch-ms sees a tie. + expect( + incomingWins( + { updatedAt: '2026-06-27T12:00:00Z', tiebreak: 'a' }, + { updatedAt: '2026-06-27T12:00:00.000Z', tiebreak: 'a' } + ) + ).toBe(false); + }); + + it('breaks an exact timestamp tie deterministically by content key', () => { + const older = { updatedAt: '2026-06-27T12:00:00.000Z', tiebreak: 'aaa' }; + const newerKey = { updatedAt: '2026-06-27T12:00:00.000Z', tiebreak: 'zzz' }; + // The larger key wins, and the decision is symmetric: whichever side is + // "incoming", the SAME version (zzz) ends up the winner → peers converge. + expect(incomingWins(newerKey, older)).toBe(true); + expect(incomingWins(older, newerKey)).toBe(false); + }); + + it('does not overwrite when timestamps and tiebreak are identical', () => { + const v = { updatedAt: '2026-06-27T12:00:00.000Z', tiebreak: 'same' }; + expect(incomingWins({ ...v }, { ...v })).toBe(false); + }); + + it('treats a missing/unparseable incoming timestamp as oldest (loses)', () => { + expect( + incomingWins({ updatedAt: undefined }, { updatedAt: '2026-06-27T12:00:00.000Z' }) + ).toBe(false); + }); +}); + +describe('contentKey (cross-peer tie-break symmetry)', () => { + it('drops id, timestamps, and RxDB-internal underscore fields', () => { + // The exact asymmetry the bug had: a transmitted `data` payload (no id, + // no updatedAt) vs a full RxDocument.toJSON() (id, updatedAt, addedAt, + // _rev, _meta) must project to the SAME key. + const transmitted = { name: 'Road Trip', tracks: 3 }; + const stored = { + id: 'pl-1', + name: 'Road Trip', + tracks: 3, + updatedAt: '2026-06-27T12:00:00.000Z', + addedAt: '2026-01-01T00:00:00.000Z', + _rev: '5-abc', + _meta: { lwt: 123 }, + _attachments: {}, + _deleted: false, + }; + expect(contentKey(transmitted)).toBe(contentKey(stored)); + }); + + it('still distinguishes genuinely different content', () => { + expect(contentKey({ name: 'A' })).not.toBe(contentKey({ name: 'B' })); + }); + + it('is order-independent for user keys', () => { + expect(contentKey({ a: 1, b: 2 })).toBe(contentKey({ b: 2, a: 1 })); + }); + + it('makes the LWW tie-break converge across peers despite shape asymmetry', () => { + // Two versions, identical timestamp. Peer P holds vP and receives vQ; + // peer Q holds vQ and receives vP. The transmitted payload is user-only; + // the stored doc carries RxDB internals. Both peers must elect the SAME + // winner. Without contentKey, P and Q compared heterogeneous strings and + // could diverge. + const ts = '2026-06-27T12:00:00.000Z'; + const vP_data = { name: 'P-edit' }; + const vQ_data = { name: 'Q-edit' }; + const stored = (data: Record) => ({ + id: 'pl-1', + ...data, + updatedAt: ts, + _rev: '9-zzz', + _meta: { lwt: 999 }, + }); + + // On P: incoming = vQ payload, existing = stored vP. + const qWinsOnP = incomingWins( + { updatedAt: ts, tiebreak: contentKey(vQ_data) }, + { updatedAt: ts, tiebreak: contentKey(stored(vP_data)) } + ); + // On Q: incoming = vP payload, existing = stored vQ. + const pWinsOnQ = incomingWins( + { updatedAt: ts, tiebreak: contentKey(vP_data) }, + { updatedAt: ts, tiebreak: contentKey(stored(vQ_data)) } + ); + + // Convergence: exactly one side's version is the winner on BOTH peers. + // qWinsOnP === true ⟺ pWinsOnQ === false (they agree vQ wins), and + // qWinsOnP === false ⟺ pWinsOnQ === true (they agree vP wins). + expect(qWinsOnP).toBe(!pWinsOnQ); + }); + + it('excludes device-local fields so the existing-side key matches the stripped incoming key', () => { + // The sender strips device-local fields (localFilePath etc.) from the + // transmitted payload; the receiver's stored doc retains them. Both keys + // must still match for the same user content. + const transmitted = { name: 'Song', artist: 'Band' }; + const stored = { + id: 'tr-1', + name: 'Song', + artist: 'Band', + updatedAt: '2026-06-27T12:00:00.000Z', + localFilePath: '/home/peer/music/song.flac', + localFileSize: 4096, + albumArtLocalPath: '/home/peer/art/song.jpg', + coverArtLocalPath: '/home/peer/art/cover.jpg', + _rev: '3-xyz', + }; + expect(contentKey(transmitted)).toBe(contentKey(stored)); + }); + + it('converges on a tie even when only one peer has device-local fields set', () => { + // Regression for the non-convergent tie-break: peer P has downloaded the + // file (localFilePath set on its STORED doc) while peer Q has not. Each + // peer receives the other's stripped payload at the exact same timestamp. + // Both must elect the SAME winner. Before the fix, the existing-side key + // carried localFilePath (sorting before `name`), so the incoming side + // always won on BOTH peers — they swapped content and oscillated forever. + const ts = '2026-06-27T12:00:00.000Z'; + const vP_data = { name: 'P-edit', artist: 'Band' }; + const vQ_data = { name: 'Q-edit', artist: 'Band' }; + // P has the local file downloaded; Q does not. + const storedP = { + id: 'tr-1', + ...vP_data, + updatedAt: ts, + localFilePath: '/home/p/song.flac', + localFileSize: 4096, + _rev: '7-aaa', + }; + const storedQ = { + id: 'tr-1', + ...vQ_data, + updatedAt: ts, + _rev: '7-bbb', + }; + + // On P: incoming = vQ stripped payload, existing = stored vP (with localFilePath). + const qWinsOnP = incomingWins( + { updatedAt: ts, tiebreak: contentKey(vQ_data) }, + { updatedAt: ts, tiebreak: contentKey(storedP) } + ); + // On Q: incoming = vP stripped payload, existing = stored vQ (no localFilePath). + const pWinsOnQ = incomingWins( + { updatedAt: ts, tiebreak: contentKey(vP_data) }, + { updatedAt: ts, tiebreak: contentKey(storedQ) } + ); + + // Convergence: the peers must agree on a single winner. + expect(qWinsOnP).toBe(!pWinsOnQ); + }); +}); + +describe('stableStringify', () => { + it('produces identical output regardless of key order', () => { + expect(stableStringify({ a: 1, b: 2 })).toBe(stableStringify({ b: 2, a: 1 })); + }); + + it('sorts nested object keys too', () => { + expect(stableStringify({ outer: { y: 1, x: 2 } })).toBe( + stableStringify({ outer: { x: 2, y: 1 } }) + ); + }); + + it('preserves array order (arrays are not reordered)', () => { + expect(stableStringify([3, 1, 2])).toBe('[3,1,2]'); + }); +}); diff --git a/app/src/renderer/db/lww.ts b/app/src/renderer/db/lww.ts new file mode 100644 index 0000000..0ded797 --- /dev/null +++ b/app/src/renderer/db/lww.ts @@ -0,0 +1,168 @@ +/** + * Last-Write-Wins (LWW) conflict resolution helpers. + * + * Extracted from replication-handler.ts (#42) so the comparison logic is pure, + * unit-testable, and skew-aware. The previous implementation compared the raw + * `updatedAt` STRINGS with `>`, which only sorts correctly for ISO-8601 values + * that share identical format, timezone, and fractional-second precision. Any + * drift (a missing `Z`, different precision, a non-ISO source) silently + * corrupted the merge. These helpers parse timestamps to epoch milliseconds + * before comparing and define a deterministic tie-break so two peers converge + * on the SAME winner. + * + * Clock-skew posture: LWW fundamentally trusts wall clocks. We do NOT clamp + * implausibly-future timestamps in the MVP — a peer with a fast clock can win + * conflicts it should not. This is an accepted limitation of LWW; the migration + * path is CRDTs (Phase 2). See [[RxDB-Replication]] "Conflict Resolution". + */ + +import { DEVICE_LOCAL_FIELDS } from './schemas'; + +/** + * Parse a timestamp value into epoch milliseconds. + * + * Accepts ISO-8601 strings (any precision/timezone that `Date.parse` handles) + * and numeric epoch-ms values. Anything missing, empty, or unparseable maps to + * `0` (treated as the oldest possible time) rather than throwing — an + * unparseable timestamp must never crash replication, and "oldest" is the safe + * default because it loses to any real timestamp. + */ +export function parseTimestampMs(value: unknown): number { + if (typeof value === 'number' && Number.isFinite(value)) { + return value; + } + if (typeof value !== 'string') { + return 0; + } + const trimmed = value.trim(); + if (trimmed === '') { + return 0; + } + const ms = Date.parse(trimmed); + return Number.isNaN(ms) ? 0 : ms; +} + +/** + * A candidate document version for LWW comparison. + * + * `updatedAt` is the primary clock; `addedAt` is a fallback for documents that + * predate the `updatedAt` field. `tiebreak` is a deterministic, content-derived + * string used ONLY when two versions carry the exact same timestamp — it must be + * computed identically on every peer (see {@link stableStringify}) so the merge + * converges instead of ping-ponging. + */ +export interface LwwCandidate { + updatedAt?: unknown; + addedAt?: unknown; + tiebreak?: string; +} + +/** + * Resolve the effective epoch-ms for a candidate: prefer `updatedAt`, fall back + * to `addedAt` only when `updatedAt` is missing/unparseable (resolves to 0). + */ +export function resolveTimestampMs(candidate: LwwCandidate): number { + const primary = parseTimestampMs(candidate.updatedAt); + if (primary > 0) { + return primary; + } + return parseTimestampMs(candidate.addedAt); +} + +/** + * Decide whether the incoming version should overwrite the existing one. + * + * Rules (deterministic and identical on both peers): + * 1. Compare parsed epoch-ms. The strictly-later timestamp wins. + * 2. On an exact tie, compare the deterministic `tiebreak` content key + * lexicographically; the larger key wins. + * 3. If timestamps AND tiebreak keys are equal, the versions are + * indistinguishable — keep the existing one (no write). + */ +export function incomingWins(incoming: LwwCandidate, existing: LwwCandidate): boolean { + const incomingMs = resolveTimestampMs(incoming); + const existingMs = resolveTimestampMs(existing); + + if (incomingMs !== existingMs) { + return incomingMs > existingMs; + } + + const incomingKey = incoming.tiebreak ?? ''; + const existingKey = existing.tiebreak ?? ''; + if (incomingKey === existingKey) { + return false; + } + return incomingKey > existingKey; +} + +/** + * Derive the deterministic equal-timestamp tie-break key for a document version. + * + * CRITICAL for cross-peer convergence: both peers must compute the SAME key for + * the SAME logical content. The incoming side carries a transmitted `data` + * payload while the existing side is a full `RxDocument.toJSON()`. Those two + * shapes differ in exactly the fields the replication envelope pulls out or RxDB + * manages — never in user content — so comparing them raw would be + * apples-to-oranges and could make two peers pick DIFFERENT winners on a tie + * (divergence). We therefore project both sides down to the same canonical user + * content by dropping {@link NON_CONTENT_KEYS}: + * - `id` — pulled out of `data` (`{id, data, updatedAt}`) and identical across + * the two competing versions anyway; + * - `updatedAt` / `addedAt` — the LWW clocks, pulled out of `data`; at a tie + * `updatedAt` is equal by definition and `addedAt` is the doc's creation time + * (equal across versions of the same id) — neither discriminates; + * - any `_`-prefixed field — RxDB internals (`_rev`, `_meta`, `_attachments`, + * `_deleted`) that never cross the wire; + * - any {@link DEVICE_LOCAL_FIELDS} (`localFilePath`, `localFileSize`, + * `albumArtLocalPath`, `coverArtLocalPath`) — the SENDER strips these from the + * transmitted `data`, but they SURVIVE on the receiver's `existing.toJSON()`. + * Excluding them here keeps the receiver's existing-side key over the SAME + * field set the incoming side already lacks; otherwise the existing-side key + * would carry e.g. `localFilePath` (sorting before `name`) and the two peers + * would elect opposite winners on a tie — an oscillating divergence. + * Every dropped key is non-discriminating at a tie and removed identically from + * both sides, so `contentKey(incoming.data)` and `contentKey(existing.toJSON())` + * reduce to the same string for the same user content — symmetric and + * convergent. (Residual assumption: the transmitted `data` carries the same user + * fields the stored doc does, which must hold for replication to work at all.) + */ +const NON_CONTENT_KEYS = new Set([ + 'id', + 'updatedAt', + 'addedAt', + ...DEVICE_LOCAL_FIELDS, +]); + +export function contentKey(value: unknown): string { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return stableStringify(value); + } + const projected: Record = {}; + for (const [k, v] of Object.entries(value as Record)) { + if (NON_CONTENT_KEYS.has(k) || k.startsWith('_')) { + continue; + } + projected[k] = v; + } + return stableStringify(projected); +} + +/** + * Deterministically serialize a value to a string with object keys sorted, so + * two peers produce byte-identical output for equal content. Used to derive the + * equal-timestamp tie-break key. Not a hash — collisions are irrelevant here + * because identical keys simply mean "keep existing". + */ +export function stableStringify(value: unknown): string { + return JSON.stringify(value, (_key, val) => { + if (val && typeof val === 'object' && !Array.isArray(val)) { + return Object.keys(val as Record) + .sort() + .reduce>((acc, k) => { + acc[k] = (val as Record)[k]; + return acc; + }, {}); + } + return val; + }); +} diff --git a/app/src/renderer/db/replication-handler.ts b/app/src/renderer/db/replication-handler.ts index ca66188..75516d1 100644 --- a/app/src/renderer/db/replication-handler.ts +++ b/app/src/renderer/db/replication-handler.ts @@ -25,6 +25,7 @@ import { getDatabase } from './database'; import type { WhatNextCollections } from './schemas'; +import { incomingWins, contentKey } from './lww'; type CollectionName = keyof WhatNextCollections; @@ -51,13 +52,21 @@ export async function applyReplicatedChanges( await existing.remove(); } } else { - // LWW: only update if incoming is newer + // LWW: only update if incoming wins. Comparison is skew-aware — + // timestamps are parsed to epoch-ms (not string-compared) and ties + // are broken deterministically by content so peers converge. See lww.ts. const existing = await col.findOne(doc.id).exec(); if (existing) { - const existingTime = (existing as unknown as Record).updatedAt as string - || (existing as unknown as Record).addedAt as string - || ''; - if (doc.updatedAt > existingTime) { + const existingData = existing.toJSON() as Record; + const winner = incomingWins( + { updatedAt: doc.updatedAt, tiebreak: contentKey(doc.data) }, + { + updatedAt: existingData.updatedAt, + addedAt: existingData.addedAt, + tiebreak: contentKey(existingData), + } + ); + if (winner) { await existing.update({ $set: doc.data }); } } else { diff --git a/app/src/renderer/db/schemas.ts b/app/src/renderer/db/schemas.ts index 864e3ed..45e7107 100644 --- a/app/src/renderer/db/schemas.ts +++ b/app/src/renderer/db/schemas.ts @@ -12,6 +12,26 @@ import type { } from 'rxdb'; import type { PurchaseLink } from '../../shared/core/download-types'; +// ======================================== +// Device-local field contract +// ======================================== + +/** + * Fields that describe a document's state on THIS device only (cached file + * locations/sizes) and are meaningless to other peers. They are stripped before + * a document is transmitted (see useSessionReplication push) and MUST also be + * excluded from the LWW equal-timestamp content tie-break key (see lww.contentKey) + * so both peers compute the key over the identical field set and converge on the + * same winner. This is the single source of truth so the sender-strip and the + * tiebreak-strip cannot drift apart. + */ +export const DEVICE_LOCAL_FIELDS = [ + 'localFilePath', + 'localFileSize', + 'albumArtLocalPath', + 'coverArtLocalPath', +] as const; + // ======================================== // User/Peer Schema // ======================================== diff --git a/app/src/renderer/hooks/useSessionReplication.ts b/app/src/renderer/hooks/useSessionReplication.ts index 76823cc..75407c6 100644 --- a/app/src/renderer/hooks/useSessionReplication.ts +++ b/app/src/renderer/hooks/useSessionReplication.ts @@ -23,6 +23,7 @@ import { useEffect, useRef } from 'react'; import { getDatabase } from '../db/database'; +import { DEVICE_LOCAL_FIELDS } from '../db/schemas'; import type { ReplicationPullRequestPayload, } from '../../shared/core/ipc-protocol'; @@ -89,14 +90,13 @@ export function useSessionReplication(enabled: boolean) { return { id, data: {}, updatedAt: new Date().toISOString(), deleted: true }; } const data = { ...doc.toJSON() } as Record; - // Strip device-local fields before sending to peers - if (col === 'tracks') { - delete data.localFilePath; - delete data.localFileSize; - delete data.albumArtLocalPath; - } - if (col === 'playlists') { - delete data.coverArtLocalPath; + // Strip device-local fields before sending to peers. + // Sourced from the shared DEVICE_LOCAL_FIELDS so this + // strip and the LWW tiebreak strip (lww.contentKey) + // can never drift apart. Field names are unique per + // collection, so deleting the full set is safe here. + for (const field of DEVICE_LOCAL_FIELDS) { + delete data[field]; } return { id, diff --git a/app/src/shared/p2p-config.ts b/app/src/shared/p2p-config.ts index 9e033e8..3646092 100644 --- a/app/src/shared/p2p-config.ts +++ b/app/src/shared/p2p-config.ts @@ -124,10 +124,47 @@ export const P2P_CONFIG = { RELAY: { /** Auto-connect to configured relays on startup */ AUTO_CONNECT: true, - /** Milliseconds between retry attempts after relay connection failure */ + /** + * @deprecated Superseded by exponential backoff (RETRY_BASE_DELAY / + * RETRY_MAX_DELAY / BACKOFF_FACTOR). Retained only for backward compat + * with any external consumer of P2P_CONFIG; RelayManager no longer reads it. + */ RETRY_INTERVAL: 10000, - /** Maximum number of connection attempts per relay address */ + /** Maximum number of connection attempts per relay address before falling back */ MAX_RETRIES: 5, + /** Base delay (ms) for exponential reconnect backoff: delay = BASE * FACTOR^attempt */ + RETRY_BASE_DELAY: 1000, + /** Ceiling (ms) for a single backoff delay, regardless of attempt count */ + RETRY_MAX_DELAY: 30000, + /** Multiplier applied per attempt for exponential backoff */ + BACKOFF_FACTOR: 2, + /** + * Jitter fraction in [0, 1]. "Equal jitter": the delay is randomized in + * [d/2, d] where d is the computed backoff. Spreads reconnect storms so + * many peers don't hammer a recovering relay in lockstep. + */ + BACKOFF_JITTER: 0.5, + /** + * Heartbeat interval (ms) for proactive relay-liveness checks. Detects + * half-open links that never emit a 'close' event by verifying the active + * relay is still among the node's live connections; if not, reconnect. + */ + HEARTBEAT_INTERVAL: 15000, + }, + + /** + * RxDB Replication Settings + */ + REPLICATION: { + /** + * How long (ms) a pull responder waits for the renderer to supply its + * documents before giving up. On timeout the responder does NOT advance + * the requester's checkpoint (it echoes the incoming checkpoint back) so + * the missed changes are re-pulled on the next attempt — this replaces + * the old hardcoded 5s "resolve empty + fresh checkpoint" that silently + * dropped changes (#41). Longer than 5s to tolerate slow/large collections. + */ + PULL_TIMEOUT: 15000, }, } as const; diff --git a/app/src/utility/__tests__/backoff.test.ts b/app/src/utility/__tests__/backoff.test.ts new file mode 100644 index 0000000..528cdad --- /dev/null +++ b/app/src/utility/__tests__/backoff.test.ts @@ -0,0 +1,45 @@ +import { describe, it, expect } from 'vitest'; +import { computeBackoffDelay, type BackoffOptions } from '../backoff'; + +const OPTS: BackoffOptions = { baseMs: 1000, maxMs: 30000, factor: 2, jitter: 0 }; + +describe('computeBackoffDelay', () => { + it('grows exponentially with no jitter', () => { + expect(computeBackoffDelay(0, OPTS)).toBe(1000); + expect(computeBackoffDelay(1, OPTS)).toBe(2000); + expect(computeBackoffDelay(2, OPTS)).toBe(4000); + expect(computeBackoffDelay(3, OPTS)).toBe(8000); + }); + + it('caps at maxMs', () => { + expect(computeBackoffDelay(100, OPTS)).toBe(30000); + }); + + it('treats negative attempts as attempt 0', () => { + expect(computeBackoffDelay(-5, OPTS)).toBe(1000); + }); + + it('keeps jittered delays within [d/2, d] of the capped value', () => { + const jittered: BackoffOptions = { ...OPTS, jitter: 1 }; + const d = 4000; // attempt 2 capped value + for (let i = 0; i < 100; i++) { + const delay = computeBackoffDelay(2, jittered); + expect(delay).toBeGreaterThanOrEqual(d / 2); + expect(delay).toBeLessThanOrEqual(d); + } + }); + + it('uses the injected rng deterministically (equal jitter)', () => { + const jittered: BackoffOptions = { ...OPTS, jitter: 1 }; + // rng=1 -> top of the range == capped value; rng=0 -> bottom == d/2. + expect(computeBackoffDelay(2, jittered, () => 1)).toBe(4000); + expect(computeBackoffDelay(2, jittered, () => 0)).toBe(2000); + }); + + it('produces strictly increasing un-jittered intervals across attempts', () => { + const delays = [0, 1, 2, 3, 4].map((a) => computeBackoffDelay(a, OPTS)); + for (let i = 1; i < delays.length; i++) { + expect(delays[i]).toBeGreaterThan(delays[i - 1]); + } + }); +}); diff --git a/app/src/utility/__tests__/checkpoint-store.test.ts b/app/src/utility/__tests__/checkpoint-store.test.ts new file mode 100644 index 0000000..70fb5cf --- /dev/null +++ b/app/src/utility/__tests__/checkpoint-store.test.ts @@ -0,0 +1,121 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { CheckpointStore, resolveCheckpointPath } from '../checkpoint-store'; + +let tmpDir: string; +let filePath: string; + +beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'wn-ckpt-')); + filePath = path.join(tmpDir, 'replication-checkpoints.json'); +}); + +afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); +}); + +describe('CheckpointStore round-trip', () => { + it('persists checkpoints and reloads them in a fresh instance (restart sim)', async () => { + const store = new CheckpointStore(filePath, 0); + await store.load(); + store.set('peerA:playlists', '2026-06-27T00:00:00.000Z'); + store.set('peerA:tracks', '2026-06-27T00:00:01.000Z'); + await store.flush(); + + // Simulate a process restart: brand-new instance reading the same file. + const reloaded = new CheckpointStore(filePath, 0); + await reloaded.load(); + expect(reloaded.get('peerA:playlists')).toBe('2026-06-27T00:00:00.000Z'); + expect(reloaded.get('peerA:tracks')).toBe('2026-06-27T00:00:01.000Z'); + }); + + it('returns null for unknown keys', async () => { + const store = new CheckpointStore(filePath, 0); + await store.load(); + expect(store.get('nope:playlists')).toBeNull(); + }); + + it('overwrites an existing checkpoint', async () => { + const store = new CheckpointStore(filePath, 0); + await store.load(); + store.set('k', 'v1'); + store.set('k', 'v2'); + await store.flush(); + expect(store.get('k')).toBe('v2'); + }); +}); + +describe('CheckpointStore graceful degradation', () => { + it('starts empty when the file is absent (cold start → full resync)', async () => { + const store = new CheckpointStore(path.join(tmpDir, 'does-not-exist.json'), 0); + await store.load(); + expect(store.entries()).toEqual([]); + }); + + it('starts empty when the file is corrupt JSON, without throwing', async () => { + fs.writeFileSync(filePath, '{ this is not json'); + const store = new CheckpointStore(filePath, 0); + await expect(store.load()).resolves.toBeUndefined(); + expect(store.entries()).toEqual([]); + }); + + it('ignores non-string values in the persisted record', async () => { + fs.writeFileSync(filePath, JSON.stringify({ good: 'x', bad: 123, nested: {} })); + const store = new CheckpointStore(filePath, 0); + await store.load(); + expect(store.get('good')).toBe('x'); + expect(store.get('bad')).toBeNull(); + expect(store.get('nested')).toBeNull(); + }); + + it('does not crash flushing into an unwritable directory', async () => { + const store = new CheckpointStore('/this/path/should/not/exist/ckpt.json', 0); + await store.load(); + store.set('k', 'v'); + await expect(store.flush()).resolves.toBeUndefined(); + }); +}); + +describe('CheckpointStore debounce', () => { + it('coalesces rapid set() calls into a single write', async () => { + const store = new CheckpointStore(filePath, 50); + await store.load(); + store.set('a', '1'); + store.set('b', '2'); + store.set('c', '3'); + // Nothing written yet (debounce window still open). + expect(fs.existsSync(filePath)).toBe(false); + await new Promise((r) => setTimeout(r, 80)); + expect(fs.existsSync(filePath)).toBe(true); + const written = JSON.parse(fs.readFileSync(filePath, 'utf-8')); + expect(written).toEqual({ a: '1', b: '2', c: '3' }); + }); + + it('set() with an unchanged value does not schedule a write', async () => { + const store = new CheckpointStore(filePath, 10); + await store.load(); + store.set('a', '1'); + await store.flush(); + // Remove file, then set the SAME value — no new write should occur. + fs.rmSync(filePath); + store.set('a', '1'); + await new Promise((r) => setTimeout(r, 30)); + expect(fs.existsSync(filePath)).toBe(false); + }); +}); + +describe('resolveCheckpointPath', () => { + it('honors the WHATNEXT_CHECKPOINT_PATH override', () => { + expect(resolveCheckpointPath({ WHATNEXT_CHECKPOINT_PATH: '/custom/ckpt.json' })).toBe( + '/custom/ckpt.json' + ); + }); + + it('derives a WhatNext-scoped path under a config dir otherwise', () => { + const resolved = resolveCheckpointPath({}); + expect(resolved).toMatch(/WhatNext/); + expect(resolved.endsWith('replication-checkpoints.json')).toBe(true); + }); +}); diff --git a/app/src/utility/__tests__/relay-manager.test.ts b/app/src/utility/__tests__/relay-manager.test.ts new file mode 100644 index 0000000..5350dda --- /dev/null +++ b/app/src/utility/__tests__/relay-manager.test.ts @@ -0,0 +1,181 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { Libp2p } from 'libp2p'; +import { RelayManager, nextFallbackAddress } from '../relay-manager'; +import { P2P_CONFIG } from '../../shared/p2p-config'; + +// RelayManager dials via `await import('@multiformats/multiaddr')` (the codebase's +// established dynamic-import pattern for this ESM-only package, mirrored in +// p2p-service.ts). Under fake timers + the concurrent suite, the real module +// loader resolves on a macrotask that `advanceTimersByTimeAsync` doesn't pump, so +// scheduled retries fire before the import settles and dials are missed. Stubbing +// it makes resolution a pure microtask (deterministic) without touching source; +// the stub preserves `toString()` so the dial-target assertions still hold. +vi.mock('@multiformats/multiaddr', () => ({ + multiaddr: (addr: string) => ({ toString: () => addr }), +})); + +const ADDR_A = '/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; +const ADDR_B = '/ip4/127.0.0.1/tcp/4002/p2p/12D3KooWBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB'; + +interface FakeConnection { + remotePeer: { toString(): string }; + addEventListener(type: 'close', cb: () => void): void; + _close(): void; +} + +function makeConnection(peerId: string): FakeConnection { + let closeCb: (() => void) | null = null; + return { + remotePeer: { toString: () => peerId }, + addEventListener: (_type, cb) => { + closeCb = cb; + }, + _close: () => closeCb?.(), + }; +} + +interface FakeNode { + dial: ReturnType; + getConnections: ReturnType; +} + +function makeNode(dialImpl: (addr: string) => Promise): FakeNode { + return { + dial: vi.fn(async (ma: { toString(): string }) => dialImpl(ma.toString())), + getConnections: vi.fn(() => [] as Array<{ remotePeer: { toString(): string } }>), + }; +} + +/** + * Repeatedly advance fake time past a backoff window and flush microtasks until + * `predicate` holds or `maxRounds` is reached. Robust against retries that pause + * on a dynamic `import()` the timer advance can't pump in a single step. + */ +async function advanceUntil(predicate: () => boolean, maxRounds = 12): Promise { + for (let i = 0; i < maxRounds && !predicate(); i++) { + await vi.advanceTimersByTimeAsync(P2P_CONFIG.RELAY.RETRY_MAX_DELAY); + // Each retry awaits `import('@multiformats/multiaddr')`; under fake timers + // that import job is not pumped by advancing time, so wait for it explicitly + // (the vitest-documented escape hatch) before checking the predicate again. + await vi.dynamicImportSettled(); + } +} + +describe('nextFallbackAddress', () => { + it('returns null when there is no alternative', () => { + expect(nextFallbackAddress([ADDR_A], ADDR_A)).toBeNull(); + expect(nextFallbackAddress([], ADDR_A)).toBeNull(); + }); + + it('returns the next address after the failed one', () => { + expect(nextFallbackAddress([ADDR_A, ADDR_B], ADDR_A)).toBe(ADDR_B); + }); + + it('wraps around the ring', () => { + expect(nextFallbackAddress([ADDR_A, ADDR_B], ADDR_B)).toBe(ADDR_A); + }); + + it('falls back to the first address when the failed one is unknown', () => { + expect(nextFallbackAddress([ADDR_A, ADDR_B], 'unknown')).toBe(ADDR_A); + }); +}); + +describe('RelayManager backoff + reconnect', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('retries a failing relay with backoff (multiple dial attempts)', async () => { + const node = makeNode(async () => { + throw new Error('connection refused'); + }); + const mgr = new RelayManager(node as unknown as Libp2p, [ADDR_A], () => {}); + + await mgr.connectAll(); + expect(node.dial).toHaveBeenCalledTimes(1); + + // Drive scheduled retries. Each retry awaits a dynamic `import()` whose + // job the fake-timer advance doesn't deterministically pump, so we POLL: + // advance past the capped backoff window, flush microtasks, and repeat + // until a second dial lands (bounded) rather than guessing an iteration + // count. This is deterministic — the retry WILL eventually fire. + await advanceUntil(() => node.dial.mock.calls.length > 1); + expect(node.dial.mock.calls.length).toBeGreaterThan(1); + + mgr.dispose(); + }); + + it('falls back to the next relay after exhausting the first', async () => { + const node = makeNode(async (addr) => { + if (addr === ADDR_A) throw new Error('A down'); + return makeConnection('relayB-peer'); + }); + const statuses: Array<[boolean, string | null]> = []; + const mgr = new RelayManager( + node as unknown as Libp2p, + [ADDR_A, ADDR_B], + (connected, ma) => statuses.push([connected, ma]) + ); + + await mgr.connectAll(); + // Drive all of A's retries to exhaustion; fallback should then dial B. + // Poll (see note above) so the dynamic-import-gated retries all settle. + await advanceUntil( + () => + node.dial.mock.calls.some( + (c) => (c[0] as { toString(): string }).toString() === ADDR_B + ), + P2P_CONFIG.RELAY.MAX_RETRIES + 4 + ); + + const dialedB = node.dial.mock.calls.some( + (c) => (c[0] as { toString(): string }).toString() === ADDR_B + ); + expect(dialedB).toBe(true); + expect(statuses.some(([connected, ma]) => connected && ma === ADDR_B)).toBe(true); + + mgr.dispose(); + }); +}); + +describe('RelayManager heartbeat liveness', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('detects a half-open relay and triggers reconnect', async () => { + const conn = makeConnection('relayA-peer'); + const node = makeNode(async () => conn); + const statuses: Array<[boolean, string | null]> = []; + const mgr = new RelayManager( + node as unknown as Libp2p, + [ADDR_A], + (connected, ma) => statuses.push([connected, ma]) + ); + + await mgr.connectAll(); + expect(mgr.currentRelayMultiaddr).toBe(ADDR_A); + + // Node reports the relay peer is NOT among live connections → half-open. + node.getConnections.mockReturnValue([]); + const live = mgr.checkRelayLiveness(); + + expect(live).toBe(false); + expect(mgr.currentRelayMultiaddr).toBeNull(); + expect(statuses).toContainEqual([false, null]); + + mgr.dispose(); + }); + + it('reports live when the relay peer is still connected', async () => { + const conn = makeConnection('relayA-peer'); + const node = makeNode(async () => conn); + const mgr = new RelayManager(node as unknown as Libp2p, [ADDR_A], () => {}); + + await mgr.connectAll(); + node.getConnections.mockReturnValue([{ remotePeer: { toString: () => 'relayA-peer' } }]); + + expect(mgr.checkRelayLiveness()).toBe(true); + expect(mgr.currentRelayMultiaddr).toBe(ADDR_A); + + mgr.dispose(); + }); +}); diff --git a/app/src/utility/backoff.ts b/app/src/utility/backoff.ts new file mode 100644 index 0000000..ba5a91d --- /dev/null +++ b/app/src/utility/backoff.ts @@ -0,0 +1,49 @@ +/** + * Exponential backoff with jitter (#41). + * + * Pure, dependency-free helper so the schedule is unit-testable independent of + * any libp2p/relay wiring. Replaces the fixed-interval relay reconnect that + * hammered a downed relay every 10s in lockstep. + */ + +export interface BackoffOptions { + /** Base delay in ms (the delay before/at attempt 0, pre-jitter). */ + baseMs: number; + /** Hard ceiling for a single computed delay, pre-jitter. */ + maxMs: number; + /** Multiplier applied per attempt (e.g. 2 doubles each time). */ + factor: number; + /** + * Jitter fraction in [0, 1]. With "equal jitter" the returned delay is + * randomized within [d/2, d] when jitter is 1, scaling linearly: the random + * portion spans `jitter * d / 2`. 0 disables jitter (deterministic). + */ + jitter: number; +} + +/** + * Compute the delay (ms) before the given retry `attempt` (0-indexed). + * + * The deterministic component is `min(maxMs, baseMs * factor^attempt)`. Jitter + * then subtracts up to `jitter * d / 2` so concurrent peers desynchronize. The + * `rng` parameter is injectable for deterministic tests (defaults to Math.random). + */ +export function computeBackoffDelay( + attempt: number, + options: BackoffOptions, + rng: () => number = Math.random +): number { + const safeAttempt = attempt < 0 ? 0 : attempt; + const raw = options.baseMs * Math.pow(options.factor, safeAttempt); + const capped = Math.min(options.maxMs, raw); + + const jitter = Math.max(0, Math.min(1, options.jitter)); + if (jitter === 0) { + return capped; + } + + // Equal-jitter: keep the lower half fixed, randomize the upper portion. + const randomSpan = (jitter * capped) / 2; + const delay = capped - randomSpan + rng() * randomSpan; + return Math.round(delay); +} diff --git a/app/src/utility/checkpoint-store.ts b/app/src/utility/checkpoint-store.ts new file mode 100644 index 0000000..dded807 --- /dev/null +++ b/app/src/utility/checkpoint-store.ts @@ -0,0 +1,164 @@ +/** + * Durable replication checkpoint store (#40). + * + * Replaces the in-memory `Map` in p2p-service.ts that died with + * the utility process and forced a full resync of every collection from every + * peer on each launch. Checkpoints are keyed `"peerId:collection"` and persisted + * to a small JSON file so a relaunch resumes from the saved checkpoint and pulls + * only an incremental delta. + * + * Ownership & location: the file lives in the Electron `userData` directory + * (the same place relay-config.json lives — see relay-config-store.ts). The + * utility process cannot call `app.getPath('userData')` (no `electron` module), + * so the directory is resolved here from platform conventions, with the + * `WHATNEXT_CHECKPOINT_PATH` env var as an explicit override (used by tests and + * available to main if it ever wants to inject the exact path). If the derived + * path is wrong or the file is corrupt/missing, the store starts empty and the + * system degrades to a full resync — degraded, never broken (acceptance #40). + * + * Writes are debounced to avoid write amplification (one fs write per document). + */ + +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; + +const CHECKPOINT_FILE = 'replication-checkpoints.json'; +const APP_DIR_NAME = 'WhatNext'; +const DEFAULT_DEBOUNCE_MS = 1000; + +/** On-disk shape: a flat map of `"peerId:collection"` → checkpoint string. */ +type CheckpointRecord = Record; + +/** + * Resolve the Electron `userData` directory using platform conventions, matching + * Electron's own defaults for app name `WhatNext`. An explicit override via + * `WHATNEXT_CHECKPOINT_PATH` wins (it points at the *file*, not the dir). + */ +export function resolveCheckpointPath(env: NodeJS.ProcessEnv = process.env): string { + const override = env.WHATNEXT_CHECKPOINT_PATH; + if (override && override.trim() !== '') { + return override; + } + + const home = os.homedir(); + let dir: string; + switch (process.platform) { + case 'win32': + dir = path.join(env.APPDATA ?? path.join(home, 'AppData', 'Roaming'), APP_DIR_NAME); + break; + case 'darwin': + dir = path.join(home, 'Library', 'Application Support', APP_DIR_NAME); + break; + default: + dir = path.join(env.XDG_CONFIG_HOME ?? path.join(home, '.config'), APP_DIR_NAME); + break; + } + return path.join(dir, CHECKPOINT_FILE); +} + +export class CheckpointStore { + private readonly filePath: string; + private readonly debounceMs: number; + private cache: Map = new Map(); + private writeTimer: ReturnType | null = null; + private loaded = false; + + constructor(filePath: string, debounceMs: number = DEFAULT_DEBOUNCE_MS) { + this.filePath = filePath; + this.debounceMs = debounceMs; + } + + /** + * Load checkpoints from disk into memory. Tolerant of a missing or corrupt + * file: on any failure the cache starts empty (→ full resync), it does not throw. + */ + async load(): Promise { + this.loaded = true; + try { + const raw = await fs.promises.readFile(this.filePath, 'utf-8'); + const parsed = JSON.parse(raw) as unknown; + if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) { + this.cache = new Map( + Object.entries(parsed as CheckpointRecord).filter( + ([k, v]) => typeof k === 'string' && typeof v === 'string' + ) + ); + } else { + this.cache = new Map(); + } + } catch { + // Missing file (cold start) or corrupt JSON — start fresh. + this.cache = new Map(); + } + } + + /** Get the checkpoint for `"peerId:collection"`, or null if none stored. */ + get(key: string): string | null { + return this.cache.get(key) ?? null; + } + + /** All stored entries (for seeding / inspection). */ + entries(): Array<[string, string]> { + return [...this.cache.entries()]; + } + + /** + * Advance a checkpoint. No-ops if the value is unchanged (avoids needless + * writes). Schedules a debounced flush to disk. + */ + set(key: string, checkpoint: string): void { + if (this.cache.get(key) === checkpoint) { + return; + } + this.cache.set(key, checkpoint); + this.scheduleFlush(); + } + + private scheduleFlush(): void { + if (this.writeTimer) { + return; + } + this.writeTimer = setTimeout(() => { + this.writeTimer = null; + void this.flush(); + }, this.debounceMs); + // Don't keep the event loop alive solely for a pending checkpoint write. + if (typeof this.writeTimer === 'object' && 'unref' in this.writeTimer) { + this.writeTimer.unref(); + } + } + + /** + * Write the current cache to disk immediately (also cancels any pending + * debounced write). Failures are swallowed — a checkpoint that fails to + * persist costs at most a redundant resync, never a crash. + */ + async flush(): Promise { + if (this.writeTimer) { + clearTimeout(this.writeTimer); + this.writeTimer = null; + } + const record: CheckpointRecord = Object.fromEntries(this.cache); + try { + await fs.promises.mkdir(path.dirname(this.filePath), { recursive: true }); + // Atomic-ish write: temp file + rename so a crash mid-write can't + // leave a half-written (corrupt) checkpoint file. + const tmp = `${this.filePath}.tmp`; + await fs.promises.writeFile(tmp, JSON.stringify(record, null, 2), 'utf-8'); + await fs.promises.rename(tmp, this.filePath); + } catch (err) { + console.warn(`[CheckpointStore] Failed to persist checkpoints: ${err}`); + } + } + + /** Flush any pending write and stop the debounce timer. */ + async dispose(): Promise { + await this.flush(); + } + + /** Whether load() has been called (used to guard double-loads). */ + get isLoaded(): boolean { + return this.loaded; + } +} diff --git a/app/src/utility/p2p-service.ts b/app/src/utility/p2p-service.ts index acd837a..f8867ea 100644 --- a/app/src/utility/p2p-service.ts +++ b/app/src/utility/p2p-service.ts @@ -39,7 +39,7 @@ import { import { P2P_CONFIG } from '../shared/p2p-config'; import { FILE_TRANSFER_CAPABILITY } from '../shared/core/file-transfer-types'; import { registerHandshakeProtocol, initiateHandshake, type HandshakeData } from './protocols/handshake'; -import { registerReplicationProtocol, pushToRemotePeer, pullFromRemotePeer, type ReplicationDocument } from './protocols/replication'; +import { registerReplicationProtocol, pushToRemotePeer, pullFromRemotePeer, newestCheckpoint, type ReplicationDocument } from './protocols/replication'; import { registerPingProtocol, startPresenceTracking } from './protocols/ping'; import { registerFileTransferProtocol, @@ -52,6 +52,7 @@ import { } from './protocols/file-transfer'; import type { FileTransferMessage } from '../shared/core/file-transfer-types'; import { RelayManager } from './relay-manager'; +import { CheckpointStore, resolveCheckpointPath } from './checkpoint-store'; import type { PeerDiscoveryEvent, PeerConnectionEvent, @@ -68,7 +69,9 @@ class P2PService { private libp2pNode: Libp2p | null = null; private isStarted = false; private connectedPeerNames: Map = new Map(); - private replicationCheckpoints: Map = new Map(); // "peerId:collection" -> checkpoint + // Durable per-peer/per-collection checkpoints ("peerId:collection" -> checkpoint). + // Persisted to disk so a relaunch resumes incrementally instead of full-resyncing (#40). + private checkpointStore: CheckpointStore = new CheckpointStore(resolveCheckpointPath()); // User identity (set via IPC from main process, used in handshake) private userDisplayName: string | null = null; @@ -367,6 +370,13 @@ class P2PService { try { this.log('info', 'Starting libp2p node...'); + // Load durable replication checkpoints before any peer connects so the + // first pull resumes from the saved checkpoint (incremental, not full resync). + if (!this.checkpointStore.isLoaded) { + await this.checkpointStore.load(); + this.log('info', `Loaded ${this.checkpointStore.entries().length} persisted checkpoint(s)`); + } + // LEARNING: Minimal libp2p configuration for desktop-to-desktop connections // We start with WebRTC and mDNS discovery // @@ -484,6 +494,9 @@ class P2PService { try { this.log('info', 'Stopping libp2p node...'); + // Flush any pending checkpoint writes before the process can exit. + await this.checkpointStore.dispose(); + // Clean up presence tracking this.stopPresenceTracking?.(); this.stopPresenceTracking = null; @@ -543,35 +556,48 @@ class P2PService { // Register replication handler registerReplicationProtocol( this.libp2pNode, - // onPullRequest: request data from renderer via main (async bridge) + // onPullRequest (responder): request our data from the renderer via main. async (collection, checkpoint, limit) => { this.log('info', `Pull request for ${collection} (checkpoint: ${checkpoint})`); - // Generate a correlation ID and wait for main to relay back renderer's data + // Generate a correlation ID and wait for main to relay back renderer's data. const requestId = `${Date.now()}-${Math.random().toString(36).slice(2)}`; - const documents = await new Promise((resolve, reject) => { - const timer = setTimeout(() => { - this.pendingPullRequests.delete(requestId); - // Resolve empty rather than reject — partial sync is OK for MVP - resolve([]); - }, 5000); - - this.pendingPullRequests.set(requestId, { - resolve: (docs) => { clearTimeout(timer); resolve(docs); }, - reject: (err) => { clearTimeout(timer); reject(err); }, + try { + const documents = await new Promise((resolve, reject) => { + // On timeout REJECT (don't resolve empty). Resolving empty + + // a fresh checkpoint made a slow peer look like it had no + // changes, silently advancing the requester past unsent data (#41). + const timer = setTimeout(() => { + this.pendingPullRequests.delete(requestId); + reject(new Error(`Pull request for ${collection} timed out after ${P2P_CONFIG.REPLICATION.PULL_TIMEOUT}ms`)); + }, P2P_CONFIG.REPLICATION.PULL_TIMEOUT); + + this.pendingPullRequests.set(requestId, { + resolve: (docs) => { clearTimeout(timer); resolve(docs); }, + reject: (err) => { clearTimeout(timer); reject(err); }, + }); + + this.sendToMain(UtilityToMainMessageType.REPLICATION_PULL_REQUEST, { + requestId, + collection, + checkpoint, + limit, + }); }); - this.sendToMain(UtilityToMainMessageType.REPLICATION_PULL_REQUEST, { - requestId, - collection, - checkpoint, - limit, - }); - }); - - return { documents, checkpoint: new Date().toISOString() }; + // Advance the checkpoint to the newest doc we're actually sending, + // not to "now" — "now" would skip docs written between the newest + // returned doc and wall-clock time. + return { documents, checkpoint: newestCheckpoint(documents, checkpoint) }; + } catch (err) { + // Surface the timeout/failure and DO NOT advance the requester's + // checkpoint: echo back the checkpoint they sent so the missed + // changes are re-pulled next time rather than silently dropped. + this.log('warn', `Pull for ${collection} failed, not advancing checkpoint: ${err}`); + return { documents: [], checkpoint: checkpoint ?? new Date(0).toISOString() }; + } }, - // onPushReceived: forward changes to main -> renderer + // onPushReceived: forward changes to main -> renderer. async (collection, documents) => { this.log('info', `Received ${documents.length} docs for ${collection}`); this.sendToMain(UtilityToMainMessageType.REPLICATION_CHANGES, { @@ -579,6 +605,20 @@ class P2PService { documents, checkpoint: new Date().toISOString(), }); + }, + // onPullResponse (requester): apply pulled docs and persist the checkpoint (#40). + (remotePeerId, collection, documents, checkpoint) => { + this.log('info', `Pull-response from ${remotePeerId.slice(0, 12)}: ${documents.length} docs for ${collection}`); + if (documents.length > 0) { + this.sendToMain(UtilityToMainMessageType.REPLICATION_CHANGES, { + collection, + documents, + checkpoint: checkpoint ?? new Date().toISOString(), + }); + } + if (checkpoint) { + this.checkpointStore.set(`${remotePeerId}:${collection}`, checkpoint); + } } ); @@ -861,7 +901,7 @@ class P2PService { for (const collection of SESSION_COLLECTIONS) { const checkpointKey = `${peerId}:${collection}`; - const checkpoint = this.replicationCheckpoints.get(checkpointKey) ?? null; + const checkpoint = this.checkpointStore.get(checkpointKey); // pullFromRemotePeer sends a pull-request; the response arrives via // the replication protocol handler (pull-response case in replication.ts). diff --git a/app/src/utility/protocols/__tests__/file-transfer.test.ts b/app/src/utility/protocols/__tests__/file-transfer.test.ts new file mode 100644 index 0000000..e984a69 --- /dev/null +++ b/app/src/utility/protocols/__tests__/file-transfer.test.ts @@ -0,0 +1,149 @@ +import { describe, it, expect, vi } from 'vitest'; +import { createHash, randomBytes } from 'node:crypto'; +import { + registerFileTransferProtocol, + sendWithBackpressure, + sendFileChunk, + requestFile, + type FileTransferCallbacks, +} from '../file-transfer'; +import { FILE_TRANSFER_CONFIG, type FileTransferMessage } from '../../../shared/core/file-transfer-types'; +import { MockStream, MockConnection, MockLibp2p, encodeFrame, asLibp2p } from './harness'; + +const PROTOCOL = FILE_TRANSFER_CONFIG.PROTOCOL_ID; + +function noopCallbacks(overrides: Partial = {}): FileTransferCallbacks { + return { + onManifestRequest: vi.fn(async () => ({ peerId: 'p', playlistId: 'pl', files: [], generatedAt: '' })), + onFileRequest: vi.fn(), + onFileChunkReceived: vi.fn(), + onFileComplete: vi.fn(), + onFileError: vi.fn(), + onTransferCancel: vi.fn(), + ...overrides, + }; +} + +describe('sendWithBackpressure (#47)', () => { + it('resolves immediately when the stream accepts the write', async () => { + const stream = new MockStream(); + await expect(sendWithBackpressure(stream as never, new Uint8Array([1, 2, 3]))).resolves.toBeUndefined(); + expect(stream.sent).toHaveLength(1); + }); + + it('waits for drain when the stream signals backpressure', async () => { + const stream = new MockStream(); + stream.failSends = 1; // first send returns false + sets writableNeedsDrain + + let resolved = false; + const p = sendWithBackpressure(stream as never, new Uint8Array([9])).then(() => { + resolved = true; + }); + + // Give microtasks a chance — it must NOT resolve while the buffer is full. + await Promise.resolve(); + expect(resolved).toBe(false); + + stream.drain(); + await p; + expect(resolved).toBe(true); + }); +}); + +describe('sendFileChunk', () => { + it('drops a message when there is no active serve stream (no throw)', async () => { + const node = new MockLibp2p(); + const peerId = { toString: () => 'peer-no-stream' }; + const msg: FileTransferMessage = { type: 'file-chunk', sha256: 'abc', offset: 0, data: '' }; + await expect( + sendFileChunk(node as never, peerId as never, msg) + ).resolves.toBeUndefined(); + }); + + it('serializes ordered writes onto the registered serve stream', async () => { + const node = new MockLibp2p(); + const callbacks = noopCallbacks(); + registerFileTransferProtocol(asLibp2p(node), callbacks); + const handler = node.handlers.get(PROTOCOL)!; + + const sha = 'deadbeef'; + const peerId = 'peer-serve'; + // Drive an inbound file-request so the serve stream gets registered. + const serveStream = new MockStream([ + encodeFrame({ type: 'file-request', sha256: sha, offsetBytes: 0 } satisfies FileTransferMessage), + ]); + await handler(serveStream, new MockConnection(peerId)); + expect(callbacks.onFileRequest).toHaveBeenCalledWith(peerId, sha, 0); + + const peer = { toString: () => peerId }; + await sendFileChunk(node as never, peer as never, { + type: 'file-header', sha256: sha, totalBytes: 10, chunkSize: 5, + }); + await sendFileChunk(node as never, peer as never, { + type: 'file-chunk', sha256: sha, offset: 0, data: 'AAAA', + }); + + const frames = serveStream.sentFrames(); + expect(frames.map((f) => f.type)).toEqual(['file-header', 'file-chunk']); + }); +}); + +describe('file-transfer integrity over the receive path (#47)', () => { + it('reassembles chunks to a byte-identical file with matching sha256', async () => { + // Build a >10MB payload so the test exercises the large-file path. + const original = randomBytes(11 * 1024 * 1024); + const sha = createHash('sha256').update(original).digest('hex'); + + // Frame it the way a provider would: file-header → N chunks → file-complete. + const chunkSize = FILE_TRANSFER_CONFIG.CHUNK_SIZE; + const frames: Uint8Array[] = []; + frames.push( + encodeFrame({ type: 'file-header', sha256: sha, totalBytes: original.length, chunkSize } satisfies FileTransferMessage) + ); + for (let offset = 0; offset < original.length; offset += chunkSize) { + const slice = original.subarray(offset, offset + chunkSize); + frames.push( + encodeFrame({ + type: 'file-chunk', + sha256: sha, + offset, + data: Buffer.from(slice).toString('base64'), + } satisfies FileTransferMessage) + ); + } + frames.push(encodeFrame({ type: 'file-complete', sha256: sha } satisfies FileTransferMessage)); + + // Collect received chunks via callbacks and reassemble. Drive the real + // requester receive path (requestFile → handleIncomingFileStream). + const received: Buffer[] = []; + const done = new Promise((resolve) => { + const callbacks = noopCallbacks({ + onFileChunkReceived: (_peer, _sha, offset, data) => { + received[offset / chunkSize] = Buffer.from(data, 'base64'); + }, + onFileComplete: () => resolve(), + onFileError: (_p, _s, err) => { + throw new Error(`unexpected file error: ${err}`); + }, + }); + + // The provider's responses arrive on the stream requestFile opens. + const providerStream = new MockStream(frames); + const node = { + dialProtocol: vi.fn(async () => providerStream), + }; + void requestFile( + node as never, + { toString: () => 'provider-peer' } as never, + sha, + 0, + callbacks + ); + }); + + await done; + const reassembled = Buffer.concat(received); + expect(reassembled.length).toBe(original.length); + expect(createHash('sha256').update(reassembled).digest('hex')).toBe(sha); + }); +}); diff --git a/app/src/utility/protocols/__tests__/handshake.test.ts b/app/src/utility/protocols/__tests__/handshake.test.ts new file mode 100644 index 0000000..3b7feea --- /dev/null +++ b/app/src/utility/protocols/__tests__/handshake.test.ts @@ -0,0 +1,76 @@ +import { describe, it, expect, vi } from 'vitest'; +import { registerHandshakeProtocol, type HandshakeData } from '../handshake'; +import { P2P_CONFIG } from '../../../shared/p2p-config'; +import { MockStream, MockConnection, MockLibp2p, encodeFrame, asLibp2p } from './harness'; + +const PROTOCOL = P2P_CONFIG.PROTOCOLS.HANDSHAKE; + +const localData: HandshakeData = { + displayName: 'Local User', + userId: 'local-uuid', + version: '1.0.0', + capabilities: ['playlist-sync', 'rxdb-replication', 'file-transfer/1.0.0'], + peerId: 'local-peer', +}; + +function remote(overrides: Partial = {}): HandshakeData { + return { + displayName: 'Remote User', + userId: 'remote-uuid', + version: '1.0.0', + capabilities: ['playlist-sync', 'rxdb-replication'], + peerId: 'remote-peer', + ...overrides, + }; +} + +describe('handshake protocol', () => { + it('exchanges metadata: reads remote handshake and replies with local data', async () => { + const node = new MockLibp2p(); + const onHandshake = vi.fn(); + registerHandshakeProtocol(asLibp2p(node), localData, onHandshake); + const handler = node.handlers.get(PROTOCOL)!; + + const remoteData = remote(); + const stream = new MockStream([encodeFrame(remoteData)]); + const conn = new MockConnection('remote-peer'); + + await handler(stream, conn); + + // onHandshake fires with the remote's parsed data. + expect(onHandshake).toHaveBeenCalledWith('remote-peer', remoteData); + // We replied with OUR handshake on a fresh stream. + const [reply] = conn.newStreams[0].sentFrames(); + expect(reply).toEqual(localData); + }); + + it('surfaces a capability mismatch to the app layer rather than rejecting it', async () => { + // The protocol does not enforce capabilities (MVP) — it exchanges them and + // lets the higher layer decide. This test pins that contract: a peer missing + // 'rxdb-replication' still completes the handshake, with its (reduced) + // capability set delivered so the app can choose how to treat it. + const node = new MockLibp2p(); + const onHandshake = vi.fn(); + registerHandshakeProtocol(asLibp2p(node), localData, onHandshake); + const handler = node.handlers.get(PROTOCOL)!; + + const mismatched = remote({ capabilities: ['playlist-sync'], version: '0.9.0' }); + await handler(new MockStream([encodeFrame(mismatched)]), new MockConnection('remote-peer')); + + expect(onHandshake).toHaveBeenCalledTimes(1); + const delivered = onHandshake.mock.calls[0][1] as HandshakeData; + expect(delivered.capabilities).toEqual(['playlist-sync']); + expect(delivered.capabilities).not.toContain('rxdb-replication'); + }); + + it('does not call onHandshake when the stream yields no data', async () => { + const node = new MockLibp2p(); + const onHandshake = vi.fn(); + registerHandshakeProtocol(asLibp2p(node), localData, onHandshake); + const handler = node.handlers.get(PROTOCOL)!; + + // Empty stream → readMessage throws → handler swallows, no handshake. + await handler(new MockStream([]), new MockConnection('remote-peer')); + expect(onHandshake).not.toHaveBeenCalled(); + }); +}); diff --git a/app/src/utility/protocols/__tests__/harness.ts b/app/src/utility/protocols/__tests__/harness.ts new file mode 100644 index 0000000..09e58ec --- /dev/null +++ b/app/src/utility/protocols/__tests__/harness.ts @@ -0,0 +1,145 @@ +/** + * Shared P2P test harness (#32). + * + * Minimal in-memory mocks for libp2p `Stream` / `Connection` / `Libp2p` so the + * protocol handlers can be exercised without a real two-node libp2p network. + * Frames are encoded with the same 4-byte big-endian length prefix the protocols + * use, so what we feed in is exactly what they parse. + * + * NOT a test file itself (no `.test.ts` suffix) — imported by the suites. + */ + +import type { Libp2p } from 'libp2p'; + +/** Encode a JSON value with the protocols' 4-byte big-endian length prefix. */ +export function encodeFrame(data: unknown): Uint8Array { + const json = new TextEncoder().encode(JSON.stringify(data)); + const frame = new Uint8Array(4 + json.length); + new DataView(frame.buffer).setUint32(0, json.length, false); + frame.set(json, 4); + return frame; +} + +/** Decode a single length-prefixed frame back into a JSON value. */ +export function decodeFrame(bytes: Uint8Array): T { + const view = new DataView(bytes.buffer, bytes.byteOffset); + const length = view.getUint32(0, false); + const body = bytes.slice(4, 4 + length); + return JSON.parse(new TextDecoder().decode(body)) as T; +} + +/** + * A mock libp2p message stream. Yields the supplied inbound frames when iterated + * and records everything written via `send`. Models backpressure via + * `writableNeedsDrain` + a resolvable `onDrain`. + */ +export class MockStream { + sent: Uint8Array[] = []; + closed = false; + aborted = false; + writableNeedsDrain = false; + /** Number of `send` calls to return `false` for (simulating a full buffer). */ + failSends = 0; + + private inbound: Uint8Array[]; + private drainResolvers: Array<() => void> = []; + + constructor(inbound: Uint8Array[] = []) { + this.inbound = inbound; + } + + async *[Symbol.asyncIterator](): AsyncIterator { + for (const chunk of this.inbound) { + yield chunk; + } + } + + send(data: Uint8Array): boolean { + this.sent.push(data); + if (this.failSends > 0) { + this.failSends -= 1; + this.writableNeedsDrain = true; + return false; + } + return true; + } + + onDrain(): Promise { + if (!this.writableNeedsDrain) { + return Promise.resolve(); + } + return new Promise((resolve) => { + this.drainResolvers.push(resolve); + }); + } + + /** Test hook: signal the buffer drained, releasing any pending onDrain awaiters. */ + drain(): void { + this.writableNeedsDrain = false; + const resolvers = this.drainResolvers; + this.drainResolvers = []; + for (const r of resolvers) r(); + } + + async close(): Promise { + this.closed = true; + } + + abort(): void { + this.aborted = true; + } + + /** Decode all frames written to this stream. */ + sentFrames(): T[] { + return this.sent.map((b) => decodeFrame(b)); + } +} + +/** A mock connection whose `newStream` hands back capturable streams. */ +export class MockConnection { + newStreams: MockStream[] = []; + + constructor(public readonly remotePeerId: string) {} + + get remotePeer() { + return { toString: () => this.remotePeerId }; + } + + async newStream(): Promise { + const stream = new MockStream(); + this.newStreams.push(stream); + return stream; + } +} + +/** + * A mock libp2p node that records the handler registered via `handle` and lets + * the test drive `getConnections()`. + */ +export class MockLibp2p { + handlers: Map unknown> = new Map(); + connections: Array<{ remotePeer: { toString(): string } }> = []; + + handle(protocol: string, handler: (stream: MockStream, connection: MockConnection) => unknown): void { + this.handlers.set(protocol, handler); + } + + getConnections() { + return this.connections; + } + + setConnectedPeers(peerIds: string[]): void { + this.connections = peerIds.map((id) => ({ remotePeer: { toString: () => id } })); + } +} + +/** + * Cast a {@link MockLibp2p} to the `Libp2p` type the protocol registrars expect. + * The mock implements only the handful of methods the protocol handlers touch + * (`handle`, `getConnections`); the registrars' parameter type demands the full + * surface. Centralizing the single unavoidable structural cast here keeps the + * test bodies clean and the unsafe assertion explained in exactly one place. + */ +export function asLibp2p(mock: MockLibp2p): Libp2p { + return mock as unknown as Libp2p; +} diff --git a/app/src/utility/protocols/__tests__/replication.test.ts b/app/src/utility/protocols/__tests__/replication.test.ts new file mode 100644 index 0000000..8ba1249 --- /dev/null +++ b/app/src/utility/protocols/__tests__/replication.test.ts @@ -0,0 +1,115 @@ +import { describe, it, expect, vi } from 'vitest'; +import { + registerReplicationProtocol, + newestCheckpoint, + type ReplicationMessage, + type ReplicationDocument, +} from '../replication'; +import { P2P_CONFIG } from '../../../shared/p2p-config'; +import { MockStream, MockConnection, MockLibp2p, encodeFrame, asLibp2p } from './harness'; + +const PROTOCOL = P2P_CONFIG.PROTOCOLS.RXDB_REPLICATION; + +function getHandler(node: MockLibp2p) { + const handler = node.handlers.get(PROTOCOL); + if (!handler) throw new Error('replication handler not registered'); + return handler; +} + +const doc = (id: string, updatedAt: string): ReplicationDocument => ({ + id, + data: { id, updatedAt, name: id }, + updatedAt, +}); + +describe('newestCheckpoint', () => { + it('returns the newest updatedAt among the documents', () => { + const docs = [doc('a', '2026-06-27T00:00:00.000Z'), doc('b', '2026-06-27T00:00:05.000Z')]; + expect(newestCheckpoint(docs, null)).toBe('2026-06-27T00:00:05.000Z'); + }); + + it('preserves the incoming checkpoint when no doc has a usable timestamp', () => { + expect(newestCheckpoint([], '2026-06-27T00:00:00.000Z')).toBe('2026-06-27T00:00:00.000Z'); + expect(newestCheckpoint([doc('a', 'bad')], '2026-06-27T00:00:00.000Z')).toBe( + '2026-06-27T00:00:00.000Z' + ); + }); + + it('never returns wall-clock now for an empty pull (epoch fallback only)', () => { + expect(newestCheckpoint([], null)).toBe(new Date(0).toISOString()); + }); +}); + +describe('replication protocol handler', () => { + it('answers a pull-request with a pull-response carrying the resolved docs', async () => { + const node = new MockLibp2p(); + const docs = [doc('a', '2026-06-27T00:00:01.000Z')]; + const onPullRequest = vi.fn(async () => ({ documents: docs, checkpoint: 'cp-1' })); + registerReplicationProtocol(asLibp2p(node), onPullRequest, vi.fn(), vi.fn()); + + const request: ReplicationMessage = { type: 'pull-request', collection: 'playlists', checkpoint: null }; + const stream = new MockStream([encodeFrame(request)]); + const conn = new MockConnection('peer-1'); + + await getHandler(node)(stream, conn); + + expect(onPullRequest).toHaveBeenCalledWith('playlists', null, 100); + expect(conn.newStreams).toHaveLength(1); + const [response] = conn.newStreams[0].sentFrames(); + expect(response.type).toBe('pull-response'); + expect(response.collection).toBe('playlists'); + expect(response.documents).toEqual(docs); + expect(response.checkpoint).toBe('cp-1'); + }); + + it('applies a push and acks it', async () => { + const node = new MockLibp2p(); + const onPushReceived = vi.fn(async () => {}); + registerReplicationProtocol(asLibp2p(node), vi.fn(), onPushReceived, vi.fn()); + + const docs = [doc('a', '2026-06-27T00:00:01.000Z')]; + const push: ReplicationMessage = { type: 'push', collection: 'tracks', documents: docs }; + const stream = new MockStream([encodeFrame(push)]); + const conn = new MockConnection('peer-2'); + + await getHandler(node)(stream, conn); + + expect(onPushReceived).toHaveBeenCalledWith('tracks', docs); + const [ack] = conn.newStreams[0].sentFrames(); + expect(ack.type).toBe('push-ack'); + expect(ack.collection).toBe('tracks'); + }); + + it('forwards a pull-response to onPullResponse with the remote peer id (#40)', async () => { + const node = new MockLibp2p(); + const onPullResponse = vi.fn(); + registerReplicationProtocol(asLibp2p(node), vi.fn(), vi.fn(), onPullResponse); + + const docs = [doc('a', '2026-06-27T00:00:09.000Z')]; + const response: ReplicationMessage = { + type: 'pull-response', + collection: 'comments', + documents: docs, + checkpoint: 'cp-9', + }; + const stream = new MockStream([encodeFrame(response)]); + const conn = new MockConnection('peer-xyz'); + + await getHandler(node)(stream, conn); + + expect(onPullResponse).toHaveBeenCalledWith('peer-xyz', 'comments', docs, 'cp-9'); + }); + + it('does not throw when no onPullResponse is provided (backward compat)', async () => { + const node = new MockLibp2p(); + registerReplicationProtocol(asLibp2p(node), vi.fn(), vi.fn()); + const response: ReplicationMessage = { + type: 'pull-response', + collection: 'users', + documents: [], + checkpoint: 'cp', + }; + const stream = new MockStream([encodeFrame(response)]); + await expect(getHandler(node)(stream, new MockConnection('p'))).resolves.toBeUndefined(); + }); +}); diff --git a/app/src/utility/protocols/file-transfer.ts b/app/src/utility/protocols/file-transfer.ts index 6275994..67a1a04 100644 --- a/app/src/utility/protocols/file-transfer.ts +++ b/app/src/utility/protocols/file-transfer.ts @@ -43,6 +43,50 @@ function encodeFramed(data: unknown): Uint8Array { return frame } +/** + * Backpressure-aware send (#47). + * + * libp2p v2 `stream.send()` returns `false` when the underlying write buffer is + * full; continuing to push data anyway grows an internal buffer that, once it + * exceeds `maxWriteBufferLength`, RESETS the stream — the exact failure mode that + * made >10MB transfers flaky. Here we respect that signal: if the stream needs + * to drain we await `onDrain()` before returning, so the caller naturally paces + * itself to the speed of a slow reader instead of overflowing the buffer. + */ +export async function sendWithBackpressure(stream: Stream, data: Uint8Array): Promise { + const accepted = stream.send(data); + if (accepted === false || stream.writableNeedsDrain) { + // Wait until the stream signals it can accept more data (or rejects if + // it closes/resets first — propagated to the caller). + await stream.onDrain(); + } +} + +/** + * Per-stream send serialization (#47). + * + * Multiple FILE_TRANSFER_SERVE_CHUNK messages can arrive concurrently from main; + * without ordering, their async backpressure awaits could interleave and corrupt + * the framed byte order on the wire. Each serve key gets a promise chain so sends + * are strictly ordered AND backpressure-paced. + */ +const serveSendChains: Map> = new Map() + +function enqueueSend(key: string, stream: Stream, data: Uint8Array): Promise { + const prior = serveSendChains.get(key) ?? Promise.resolve() + const next = prior + .catch(() => { /* a prior send failure must not block subsequent sends */ }) + .then(() => sendWithBackpressure(stream, data)) + serveSendChains.set(key, next) + // Once this is the tail of the chain, drop it to avoid unbounded map growth. + void next.finally(() => { + if (serveSendChains.get(key) === next) { + serveSendChains.delete(key) + } + }) + return next +} + /** * Accumulate raw bytes from a stream until at least `needed` bytes are available. * Returns `null` if the stream ends before enough bytes arrive. @@ -460,10 +504,9 @@ export async function requestFile( * * If the transfer is a file-complete or file-error, the stream is closed after sending. * - * TODO: stream.send() backpressure is not handled. Large files (>10MB) may - * overflow the stream write buffer, causing stream resets. Needs flow control - * between main process chunk dispatch and utility process stream writes. - * See: https://github.com/libp2p/js-libp2p/blob/main/doc/migrations/v1.0.0-v2.0.0.md#streams + * Backpressure (#47): writes go through a per-stream serialized queue that awaits + * the stream's drain signal, so a slow reader paces the sender instead of + * overflowing the write buffer (which would reset the stream on files >10MB). */ export async function sendFileChunk( libp2p: Libp2p, @@ -480,15 +523,19 @@ export async function sendFileChunk( } try { - stream.send(encodeFramed(message)) + // Serialized + backpressure-aware: ordering is preserved across concurrent + // calls and we wait for the stream to drain before overrunning its buffer. + await enqueueSend(key, stream, encodeFramed(message)) if (message.type === 'file-complete' || message.type === 'file-error') { activeServeStreams.delete(key) + serveSendChains.delete(key) try { await stream.close() } catch { /* ignore */ } } } catch (err) { console.error(`[FileTransfer] Error writing to serve stream: ${err}`) activeServeStreams.delete(key) + serveSendChains.delete(key) try { await stream.close() } catch { /* ignore */ } throw err } diff --git a/app/src/utility/protocols/replication.ts b/app/src/utility/protocols/replication.ts index 2b47d75..8a3b3ad 100644 --- a/app/src/utility/protocols/replication.ts +++ b/app/src/utility/protocols/replication.ts @@ -36,6 +36,31 @@ export interface ReplicationDocument { deleted?: boolean; } +/** + * Compute the checkpoint to report after returning `documents` for a pull. + * + * Returns the newest parseable `updatedAt` among the documents. If no document + * carries a usable timestamp, the incoming `checkpoint` is preserved (so the + * checkpoint never moves backward and never jumps to wall-clock "now", which + * would skip docs written between the newest returned doc and now). Exported for + * unit testing (#32). + */ +export function newestCheckpoint( + documents: ReplicationDocument[], + incoming: string | null +): string { + let newestMs = -1; + let newest = incoming; + for (const doc of documents) { + const ms = Date.parse(doc.updatedAt); + if (!Number.isNaN(ms) && ms > newestMs) { + newestMs = ms; + newest = doc.updatedAt; + } + } + return newest ?? new Date(0).toISOString(); +} + /** * Encode a message with a 4-byte big-endian length prefix. */ @@ -123,6 +148,20 @@ export type OnPushReceived = ( documents: ReplicationDocument[] ) => Promise; +/** + * Called on the REQUESTER side when a `pull-response` arrives. Previously the + * pull-response was dropped (logged only), so pulled documents never reached the + * renderer and the checkpoint never advanced — meaning every pull silently + * achieved nothing and the next launch full-resynced again (#40/#41). The handler + * forwards the documents to the renderer and persists the returned checkpoint. + */ +export type OnPullResponse = ( + remotePeerId: string, + collection: string, + documents: ReplicationDocument[], + checkpoint: string | null +) => void; + /** * Register replication protocol handler */ @@ -130,6 +169,7 @@ export function registerReplicationProtocol( node: Libp2p, onPullRequest: OnPullRequest, onPushReceived: OnPushReceived, + onPullResponse?: OnPullResponse, ): void { node.handle(P2P_CONFIG.PROTOCOLS.RXDB_REPLICATION, async (stream: Stream, connection: Connection) => { try { @@ -170,8 +210,15 @@ export function registerReplicationProtocol( } case 'pull-response': { - // This is handled by the initiator - forward to main process + // Requester side: forward pulled docs to the renderer and + // advance the persisted checkpoint. Without this, pulls are no-ops. console.log(`[Replication] Got pull-response with ${message.documents?.length ?? 0} docs`); + onPullResponse?.( + connection.remotePeer.toString(), + message.collection, + message.documents ?? [], + message.checkpoint ?? null + ); break; } diff --git a/app/src/utility/relay-manager.ts b/app/src/utility/relay-manager.ts index f0e9149..3cc6c24 100644 --- a/app/src/utility/relay-manager.ts +++ b/app/src/utility/relay-manager.ts @@ -1,135 +1,244 @@ -/** - * Relay Manager - * - * Manages circuit relay v2 connections for NAT traversal. - * Handles: - * - Connecting to configured relay addresses with retry logic - * - Tracking which relay is currently active - * - Notifying the service when relay status changes - * - * Relay addresses are user-configured (from relay-config-store) and passed - * in at startup — they are NOT baked into the compiled code. This preserves - * user sovereignty: users pick the relay infrastructure for their sessions. - */ - -import type { Libp2p } from 'libp2p'; -import { P2P_CONFIG } from '../shared/p2p-config'; - -export type RelayStatusCallback = (connected: boolean, relayMultiaddr: string | null, relayPeerId: string | null) => void; - -export class RelayManager { - private node: Libp2p; - private addresses: string[]; - private onStatusChange: RelayStatusCallback; - private activeRelayMultiaddr: string | null = null; - private activeRelayPeerId: string | null = null; - private retryTimers: Map> = new Map(); - - constructor(node: Libp2p, addresses: string[], onStatusChange: RelayStatusCallback) { - this.node = node; - this.addresses = addresses; - this.onStatusChange = onStatusChange; - } - - /** Attempt to connect to all configured relay addresses. */ - async connectAll(): Promise { - if (this.addresses.length === 0) { - console.log('[RelayManager] No relay addresses configured'); - return; - } - - for (const addr of this.addresses) { - await this.connectWithRetry(addr, 0); - } - } - - /** Update the list of relay addresses and reconnect. */ - async updateAddresses(newAddresses: string[]): Promise { - // Compute diff BEFORE overwriting this.addresses — - // otherwise added would always be empty (comparing newAddresses to itself). - const removed = this.addresses.filter((a) => !newAddresses.includes(a)); - const added = newAddresses.filter((a) => !this.addresses.includes(a)); - - // Cancel pending retries for addresses that have been removed - for (const addr of removed) { - this.clearRetry(addr); - } - - this.addresses = newAddresses; - - // Connect to newly added addresses - for (const addr of added) { - await this.connectWithRetry(addr, 0); - } - } - - get currentRelayMultiaddr(): string | null { - return this.activeRelayMultiaddr; - } - - get currentRelayPeerId(): string | null { - return this.activeRelayPeerId; - } - - private async connectWithRetry(addr: string, attempt: number): Promise { - if (attempt >= P2P_CONFIG.RELAY.MAX_RETRIES) { - console.warn(`[RelayManager] Gave up connecting to relay ${addr} after ${attempt} attempts`); - return; - } - - try { - console.log(`[RelayManager] Connecting to relay: ${addr} (attempt ${attempt + 1}/${P2P_CONFIG.RELAY.MAX_RETRIES})`); - const { multiaddr } = await import('@multiformats/multiaddr'); - const ma = multiaddr(addr); - const connection = await this.node.dial(ma); - - const peerId = connection.remotePeer.toString(); - console.log(`[RelayManager] Connected to relay: ${addr} (peer: ${peerId})`); - - this.activeRelayMultiaddr = addr; - this.activeRelayPeerId = peerId; - this.clearRetry(addr); - this.onStatusChange(true, addr, peerId); - - // Listen for relay disconnection - connection.addEventListener('close', () => { - console.log(`[RelayManager] Relay disconnected: ${addr}`); - if (this.activeRelayMultiaddr === addr) { - this.activeRelayMultiaddr = null; - this.activeRelayPeerId = null; - this.onStatusChange(false, null, null); - } - // Schedule reconnect - this.scheduleRetry(addr, 0); - }); - } catch (err) { - console.warn(`[RelayManager] Failed to connect to relay ${addr}: ${err}`); - this.scheduleRetry(addr, attempt + 1); - } - } - - private scheduleRetry(addr: string, attempt: number): void { - this.clearRetry(addr); - if (attempt >= P2P_CONFIG.RELAY.MAX_RETRIES) return; - - const timer = setTimeout( - () => this.connectWithRetry(addr, attempt), - P2P_CONFIG.RELAY.RETRY_INTERVAL - ); - this.retryTimers.set(addr, timer); - } - - private clearRetry(addr: string): void { - const existing = this.retryTimers.get(addr); - if (existing) { - clearTimeout(existing); - this.retryTimers.delete(addr); - } - } - - dispose(): void { - for (const [addr] of this.retryTimers) { - this.clearRetry(addr); - } - } -} +/** + * Relay Manager + * + * Manages circuit relay v2 connections for NAT traversal. + * Handles: + * - Connecting to configured relay addresses with exponential backoff (#41) + * - Falling back to the next configured relay when one is exhausted (#41) + * - A periodic heartbeat that detects half-open relay links (#41) + * - Tracking which relay is currently active + * - Notifying the service when relay status changes + * + * Relay addresses are user-configured (from relay-config-store) and passed + * in at startup — they are NOT baked into the compiled code. This preserves + * user sovereignty: users pick the relay infrastructure for their sessions. + */ + +import type { Libp2p } from 'libp2p'; +import { P2P_CONFIG } from '../shared/p2p-config'; +import { computeBackoffDelay } from './backoff'; + +export type RelayStatusCallback = (connected: boolean, relayMultiaddr: string | null, relayPeerId: string | null) => void; + +/** + * Pick the next relay address to try when `failedAddr` has exhausted its + * retries. Returns the first OTHER configured address after `failedAddr` + * (wrapping around), or null when no alternative exists. Pure so the fallback + * policy is unit-testable without any libp2p wiring. + */ +export function nextFallbackAddress(addresses: readonly string[], failedAddr: string): string | null { + if (addresses.length <= 1) { + return null; + } + const idx = addresses.indexOf(failedAddr); + if (idx === -1) { + // Unknown address — fall back to the first configured one. + return addresses[0] ?? null; + } + // Walk the ring starting after the failed address; return the first distinct entry. + for (let i = 1; i < addresses.length; i++) { + const candidate = addresses[(idx + i) % addresses.length]; + if (candidate !== failedAddr) { + return candidate; + } + } + return null; +} + +export class RelayManager { + private node: Libp2p; + private addresses: string[]; + private onStatusChange: RelayStatusCallback; + private activeRelayMultiaddr: string | null = null; + private activeRelayPeerId: string | null = null; + private retryTimers: Map> = new Map(); + private heartbeatTimer: ReturnType | null = null; + private disposed = false; + + constructor(node: Libp2p, addresses: string[], onStatusChange: RelayStatusCallback) { + this.node = node; + this.addresses = addresses; + this.onStatusChange = onStatusChange; + } + + /** Attempt to connect to all configured relay addresses. */ + async connectAll(): Promise { + if (this.addresses.length === 0) { + console.log('[RelayManager] No relay addresses configured'); + return; + } + + this.startHeartbeat(); + + for (const addr of this.addresses) { + await this.connectWithRetry(addr, 0); + } + } + + /** Update the list of relay addresses and reconnect. */ + async updateAddresses(newAddresses: string[]): Promise { + // Compute diff BEFORE overwriting this.addresses — + // otherwise added would always be empty (comparing newAddresses to itself). + const removed = this.addresses.filter((a) => !newAddresses.includes(a)); + const added = newAddresses.filter((a) => !this.addresses.includes(a)); + + // Cancel pending retries for addresses that have been removed + for (const addr of removed) { + this.clearRetry(addr); + } + + this.addresses = newAddresses; + + // Connect to newly added addresses + for (const addr of added) { + await this.connectWithRetry(addr, 0); + } + } + + get currentRelayMultiaddr(): string | null { + return this.activeRelayMultiaddr; + } + + get currentRelayPeerId(): string | null { + return this.activeRelayPeerId; + } + + private async connectWithRetry(addr: string, attempt: number): Promise { + if (this.disposed) return; + + if (attempt >= P2P_CONFIG.RELAY.MAX_RETRIES) { + console.warn(`[RelayManager] Gave up connecting to relay ${addr} after ${attempt} attempts`); + this.tryFallback(addr); + return; + } + + try { + console.log(`[RelayManager] Connecting to relay: ${addr} (attempt ${attempt + 1}/${P2P_CONFIG.RELAY.MAX_RETRIES})`); + const { multiaddr } = await import('@multiformats/multiaddr'); + const ma = multiaddr(addr); + const connection = await this.node.dial(ma); + + const peerId = connection.remotePeer.toString(); + console.log(`[RelayManager] Connected to relay: ${addr} (peer: ${peerId})`); + + this.activeRelayMultiaddr = addr; + this.activeRelayPeerId = peerId; + this.clearRetry(addr); + this.onStatusChange(true, addr, peerId); + + // Listen for relay disconnection + connection.addEventListener('close', () => { + console.log(`[RelayManager] Relay disconnected: ${addr}`); + if (this.activeRelayMultiaddr === addr) { + this.activeRelayMultiaddr = null; + this.activeRelayPeerId = null; + this.onStatusChange(false, null, null); + } + // Schedule reconnect (attempt 0 — start the backoff schedule fresh) + this.scheduleRetry(addr, 0); + }); + } catch (err) { + console.warn(`[RelayManager] Failed to connect to relay ${addr}: ${err}`); + this.scheduleRetry(addr, attempt + 1); + } + } + + /** + * On exhausting retries for `failedAddr`, try the next configured relay — + * but only if we don't already have a live relay. This is the single-fallback + * hook the spec asks for (multi-relay orchestration remains out of scope). + */ + private tryFallback(failedAddr: string): void { + if (this.disposed || this.activeRelayMultiaddr) return; + const fallback = nextFallbackAddress(this.addresses, failedAddr); + if (fallback && !this.retryTimers.has(fallback)) { + console.log(`[RelayManager] Falling back from ${failedAddr} to ${fallback}`); + void this.connectWithRetry(fallback, 0); + } + } + + private scheduleRetry(addr: string, attempt: number): void { + this.clearRetry(addr); + if (this.disposed) return; + if (attempt >= P2P_CONFIG.RELAY.MAX_RETRIES) { + this.tryFallback(addr); + return; + } + + const delay = computeBackoffDelay(attempt, { + baseMs: P2P_CONFIG.RELAY.RETRY_BASE_DELAY, + maxMs: P2P_CONFIG.RELAY.RETRY_MAX_DELAY, + factor: P2P_CONFIG.RELAY.BACKOFF_FACTOR, + jitter: P2P_CONFIG.RELAY.BACKOFF_JITTER, + }); + console.log(`[RelayManager] Retry ${addr} (attempt ${attempt + 1}) in ${delay}ms`); + const timer = setTimeout(() => this.connectWithRetry(addr, attempt), delay); + // Don't keep the event loop alive solely for a pending relay retry. + if (typeof timer === 'object' && 'unref' in timer) { + (timer as { unref: () => void }).unref(); + } + this.retryTimers.set(addr, timer); + } + + private clearRetry(addr: string): void { + const existing = this.retryTimers.get(addr); + if (existing) { + clearTimeout(existing); + this.retryTimers.delete(addr); + } + } + + /** + * Start the relay liveness heartbeat. Some half-open links never emit a + * 'close' event (NAT timeout, silent peer death), so the disconnect handler + * never fires and we believe we're still relayed. The heartbeat proactively + * verifies the active relay is still among the node's live connections and, + * if not, treats it as a disconnect and reconnects. + */ + private startHeartbeat(): void { + if (this.heartbeatTimer) return; + this.heartbeatTimer = setInterval(() => { + this.checkRelayLiveness(); + }, P2P_CONFIG.RELAY.HEARTBEAT_INTERVAL); + if (typeof this.heartbeatTimer === 'object' && 'unref' in this.heartbeatTimer) { + (this.heartbeatTimer as { unref: () => void }).unref(); + } + } + + /** + * One heartbeat tick. Returns true if the active relay is still live; false + * if a half-open link was detected (and a reconnect was scheduled). Exposed + * for unit testing — callers should not need to invoke it directly. + */ + checkRelayLiveness(): boolean { + if (this.disposed || !this.activeRelayPeerId || !this.activeRelayMultiaddr) { + return true; // nothing to check + } + const stillConnected = this.node + .getConnections() + .some((c) => c.remotePeer.toString() === this.activeRelayPeerId); + + if (stillConnected) { + return true; + } + + console.warn(`[RelayManager] Heartbeat: active relay ${this.activeRelayMultiaddr} is half-open — reconnecting`); + const addr = this.activeRelayMultiaddr; + this.activeRelayMultiaddr = null; + this.activeRelayPeerId = null; + this.onStatusChange(false, null, null); + this.scheduleRetry(addr, 0); + return false; + } + + dispose(): void { + this.disposed = true; + for (const [addr] of this.retryTimers) { + this.clearRetry(addr); + } + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + } +}