From 30619c33a9c8b8e22537364397cbc748f702be6c Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 11:15:07 -0700 Subject: [PATCH 1/7] feat(p2p): durable checkpoints + reliable pull/relay reconnection Persist per-peer/per-collection replication checkpoints to a debounced JSON file under userData so a relaunch resumes incrementally instead of full-resyncing every collection (#40). Stop silently dropping changes on pull timeout: a timed-out pull now rejects and does not advance the checkpoint, and the responder advances to the newest returned doc's updatedAt rather than wall-clock now. Act on the previously-dropped pull-response on the requester side. Replace fixed 10s relay retries with exponential backoff + jitter, a single alternate-relay fallback, and a liveness heartbeat for half-open links (#41). Wire protocol id unchanged. Covered by checkpoint-store, backoff, relay-manager and replication suites (#32). --- app/src/shared/p2p-config.ts | 41 +- app/src/utility/__tests__/backoff.test.ts | 45 +++ .../__tests__/checkpoint-store.test.ts | 121 ++++++ .../utility/__tests__/relay-manager.test.ts | 181 +++++++++ app/src/utility/backoff.ts | 49 +++ app/src/utility/checkpoint-store.ts | 164 ++++++++ app/src/utility/p2p-service.ts | 90 +++-- .../utility/protocols/__tests__/harness.ts | 145 +++++++ .../protocols/__tests__/replication.test.ts | 115 ++++++ app/src/utility/protocols/replication.ts | 49 ++- app/src/utility/relay-manager.ts | 379 +++++++++++------- 11 files changed, 1216 insertions(+), 163 deletions(-) create mode 100644 app/src/utility/__tests__/backoff.test.ts create mode 100644 app/src/utility/__tests__/checkpoint-store.test.ts create mode 100644 app/src/utility/__tests__/relay-manager.test.ts create mode 100644 app/src/utility/backoff.ts create mode 100644 app/src/utility/checkpoint-store.ts create mode 100644 app/src/utility/protocols/__tests__/harness.ts create mode 100644 app/src/utility/protocols/__tests__/replication.test.ts 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__/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/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; + } + } +} From 9138704e7f823446e3c292a96509bb18e260cf91 Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 11:15:12 -0700 Subject: [PATCH 2/7] fix(p2p): skew-aware LWW with convergent tie-break Replace the brittle string comparison of updatedAt in replication-handler with epoch-ms parsing (lww.ts). Unparseable/missing timestamps resolve to oldest rather than throwing. Equal-timestamp ties are broken by a deterministic contentKey projection that strips id, updatedAt/addedAt and RxDB-internal underscore fields identically on both sides, so two peers comparing a transmitted data payload against a full toJSON elect the SAME winner and converge (acceptance #42). Covered by lww suite incl. a cross-peer convergence test (#32). --- app/src/renderer/db/__tests__/lww.test.ts | 174 +++++++++++++++++++++ app/src/renderer/db/lww.ts | 154 ++++++++++++++++++ app/src/renderer/db/replication-handler.ts | 19 ++- 3 files changed, 342 insertions(+), 5 deletions(-) create mode 100644 app/src/renderer/db/__tests__/lww.test.ts create mode 100644 app/src/renderer/db/lww.ts 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..4c92edb --- /dev/null +++ b/app/src/renderer/db/__tests__/lww.test.ts @@ -0,0 +1,174 @@ +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); + }); +}); + +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..bbf8fc0 --- /dev/null +++ b/app/src/renderer/db/lww.ts @@ -0,0 +1,154 @@ +/** + * 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". + */ + +/** + * 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. + * 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']); + +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 { From 94c85d076c92be8517fed9018720fc196fac0e7a Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 11:15:23 -0700 Subject: [PATCH 3/7] fix(p2p): backpressure-safe file-chunk sends Route serve-stream writes through a per-stream serialized queue that awaits the libp2p stream drain signal before overrunning the write buffer, so a slow reader paces the sender instead of triggering the stream reset that made >10MB transfers flaky. Removes the standing backpressure TODO (#47). Covered by a large-file (>10MB) reassembly + sha256 integrity test and a drain-wait unit test (#32). --- .../protocols/__tests__/file-transfer.test.ts | 149 ++++++++++++++++++ app/src/utility/protocols/file-transfer.ts | 57 ++++++- 2 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 app/src/utility/protocols/__tests__/file-transfer.test.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/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 } From 98f6f2bf22a3013a84c30b731b54a5d88ebd50a5 Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 11:15:24 -0700 Subject: [PATCH 4/7] test(p2p): handshake protocol coverage via shared harness Add handshake capability-exchange tests (success, capability-mismatch surfaced not rejected, empty-stream no-op) on the shared in-memory libp2p harness, closing the last untested protocol path (#32). --- .../protocols/__tests__/handshake.test.ts | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 app/src/utility/protocols/__tests__/handshake.test.ts 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(); + }); +}); From a9154248c453829f7db2f72e11f0aae15468007d Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 11:15:28 -0700 Subject: [PATCH 5/7] docs(p2p): document replication reliability hardening Update the RxDB-Replication concept page to reflect durable checkpoints, the non-silent pull-timeout policy, relay backoff/fallback/heartbeat, and skew-aware LWW with the convergent tie-break. Records the explicit decision NOT to bump the wire-protocol version (stays /whatnext/rxdb-replication/1.0.0) per the epic's no-silent-bump constraint. --- .../01 concepts/RxDB-Replication.md | 64 ++++++++++++++----- 1 file changed, 49 insertions(+), 15 deletions(-) diff --git a/WhatNext - docs/01 concepts/RxDB-Replication.md b/WhatNext - docs/01 concepts/RxDB-Replication.md index 13cc214..9e48a68 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,44 @@ 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. +- **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 +136,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 +149,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) From e44e828b829784f2e339a061ca91b22108c53b7c Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Sat, 27 Jun 2026 21:11:16 -0700 Subject: [PATCH 6/7] fix(p2p): exclude device-local fields from LWW tie-break key The sender strips device-local fields (localFilePath, localFileSize, albumArtLocalPath, coverArtLocalPath) before transmission, but the receiver computed the equal-timestamp content tie-break key over the unstripped stored doc. On an exact-ms tie the existing-side key carried localFilePath (which sorts before name), so the incoming version won on both peers -- they swapped user content and oscillated indefinitely, violating cross-peer convergence (AC #42). Introduce a single DEVICE_LOCAL_FIELDS constant in schemas.ts consumed by both the sender strip and lww.contentKey so the two field sets cannot drift. Receiver-side only: wire format and protocol id are unchanged. --- .claude/cycle/impl-notes.md | 161 ++++++++++++++++++ .../01 concepts/RxDB-Replication.md | 1 + app/src/renderer/db/__tests__/lww.test.ts | 60 +++++++ app/src/renderer/db/lww.ts | 18 +- app/src/renderer/db/schemas.ts | 20 +++ .../renderer/hooks/useSessionReplication.ts | 16 +- 6 files changed, 266 insertions(+), 10 deletions(-) create mode 100644 .claude/cycle/impl-notes.md diff --git a/.claude/cycle/impl-notes.md b/.claude/cycle/impl-notes.md new file mode 100644 index 0000000..9279c8d --- /dev/null +++ b/.claude/cycle/impl-notes.md @@ -0,0 +1,161 @@ +# Implementation notes + +**Brief:** .claude/cycle/brief.md (cycle started 2026-06-27) +**Epic:** WhatNext - docs/08 specs/epic-replication-reliability.md +**Lane:** replication-reliability (Wave 1, P2P-approved) +**Architect run:** 2026-06-27 + +## What I built +Hardened the five reliability surfaces of the P2P RxDB replication stack: durable +per-peer/per-collection checkpoints (#40), a non-silent pull-timeout plus +exponential-backoff relay reconnection with single-fallback and a liveness +heartbeat (#41), skew-aware epoch-ms LWW conflict resolution with a deterministic, +cross-peer-convergent tie-break (#42), backpressure-safe file-chunk sends so +>10MB transfers no longer reset the stream (#47), and a from-scratch Vitest P2P +harness with suites covering all of the above (#32). The `[[RxDB-Replication]]` +concept page is updated to document the new behavior. **No wire-protocol version +bump** — the protocol id stays `/whatnext/rxdb-replication/1.0.0`. + +Note on provenance: I inherited a substantial, uncommitted draft of this work +already present in the worktree from a prior Architect pass. I reviewed every line, +fixed the defects I found (a flaky test, type errors, and a real LWW convergence +bug — see below), added missing coverage, verified, documented, and committed it. + +## Design decisions + +- **Decision:** Checkpoint store = a single debounced JSON file under `userData`, + owned/written by the utility process (path derived from platform conventions, + `WHATNEXT_CHECKPOINT_PATH` env override for tests). + **Alternatives considered:** a dedicated RxDB collection in the renderer (the + spec's other candidate). + **Rationale:** a flat file is simpler and decoupled from the very DB being + replicated; the renderer-collection option couples checkpoint durability to + renderer liveness and the replicated dataset. + **Tradeoff:** the utility process can't call `app.getPath('userData')`, so it + re-derives the path from platform conventions — a small duplication of Electron's + own logic that could drift if the app name changes. Mitigated by the env override + and a corrupt/missing-file → full-resync fallback (degraded, never broken). + +- **Decision:** On pull timeout, REJECT and echo the requester's incoming + checkpoint back (do not advance); responder advances the checkpoint to the + newest returned doc's `updatedAt`, never wall-clock `now`. + **Alternatives considered:** keep "resolve empty + fresh checkpoint" (the bug); + advance to `now` on success. + **Rationale:** advancing on a timeout or to `now` silently skips unsent docs — + the exact data-loss bug. Echoing the incoming checkpoint re-pulls next time. + **Tradeoff:** a permanently-slow peer re-attempts the same pull each cycle + (no progress) rather than declaring false success — correct, but it trades a + silent wrong answer for a visible retry loop. Acceptable and observable. + +- **Decision:** Relay backoff/fallback/heartbeat as pure, separately-tested + helpers (`backoff.ts`, `nextFallbackAddress`, `checkRelayLiveness`). + **Rationale:** keeps the schedule and ring-fallback policy unit-testable without + a real libp2p network. + **Tradeoff:** fallback is single-hop only (spec-scoped); multi-relay + orchestration is explicitly out of scope. + +- **Decision (correctness fix beyond the inherited draft):** LWW equal-timestamp + tie-break uses a `contentKey()` projection that strips `id`, `updatedAt`, + `addedAt`, and all `_`-prefixed RxDB-internal fields before stable + serialization, applied identically to both the incoming `data` payload and the + existing `RxDocument.toJSON()`. + **Alternatives considered:** the inherited code compared + `stableStringify(doc.data)` against `stableStringify(existingData)` — i.e. a + transmitted user-fields payload against a full toJSON carrying `_rev`/`_meta`/ + `id`/`updatedAt`. That is heterogeneous and **non-symmetric across peers**, so + on an exact-ms tie peer A and peer B could elect different winners and diverge — + violating acceptance #42 ("identical on both peers"). + **Rationale:** every stripped key is non-discriminating at a tie (id equal, + timestamps equal/creation-time) and removed identically from both sides, so + `contentKey(data) === contentKey(toJSON)` for the same user content → symmetric + and convergent. + **Tradeoff:** relies on the residual assumption that the transmitted `data` + carries the same user fields the stored doc does (which must hold for + replication to function at all). Flagged under Uncertainty. + +## Files changed +- `app/src/utility/checkpoint-store.ts` — NEW: durable, debounced, atomic JSON checkpoint store (#40). +- `app/src/utility/backoff.ts` — NEW: pure exponential-backoff-with-jitter helper (#41). +- `app/src/utility/relay-manager.ts` — backoff + ring fallback + liveness heartbeat (#41). +- `app/src/shared/p2p-config.ts` — new `RELAY` backoff/heartbeat constants + `REPLICATION.PULL_TIMEOUT`; deprecate `RETRY_INTERVAL` (#41). +- `app/src/utility/p2p-service.ts` — wire in the checkpoint store; reject pull timeouts; act on pull-response (#40, #41). +- `app/src/utility/protocols/replication.ts` — `newestCheckpoint()` + `onPullResponse` requester handler (#40/#41). +- `app/src/renderer/db/lww.ts` — NEW: skew-aware parse + symmetric `contentKey` tie-break (#42). +- `app/src/renderer/db/replication-handler.ts` — use `incomingWins`/`contentKey` instead of string compare (#42). +- `app/src/utility/protocols/file-transfer.ts` — `sendWithBackpressure` + per-stream serialized send queue; TODO removed (#47). +- `WhatNext - docs/01 concepts/RxDB-Replication.md` — document all of the above; explicit "no version bump" note (P2P doc requirement). +- Tests (#32): `app/src/utility/protocols/__tests__/{harness.ts,handshake.test.ts,replication.test.ts,file-transfer.test.ts}`, `app/src/utility/__tests__/{backoff,checkpoint-store,relay-manager}.test.ts`, `app/src/renderer/db/__tests__/lww.test.ts`. + +## Tests +- **lww.test.ts** (26 cases): epoch parse of mixed precision/missing/numeric; updatedAt-over-addedAt; strictly-later wins; differing-precision equal instants as a tie; deterministic content tie-break; `contentKey` strips id/timestamps/`_`-fields; **explicit cross-peer convergence test** proving the two peers elect the same winner despite the data-vs-toJSON shape asymmetry. +- **checkpoint-store.test.ts**: round-trip + restart simulation; unknown-key null; corrupt/absent/non-string-value graceful degradation; unwritable-dir no-crash; debounce coalescing; no-op set; path resolution + env override. +- **backoff.test.ts**: monotonic growth, ceiling cap, jitter bounds, zero-jitter determinism. +- **relay-manager.test.ts**: `nextFallbackAddress` ring logic; backoff retry fires repeatedly; fallback to next relay on exhaustion; heartbeat detects half-open + reports live. +- **handshake.test.ts**: metadata exchange + reply; capability-mismatch surfaced not rejected; empty-stream no-op. +- **replication.test.ts**: `newestCheckpoint` selection/preservation/epoch-fallback; pull-request→pull-response; push→push-ack; pull-response forwarded to `onPullResponse`; backward-compat when callback omitted. +- **file-transfer.test.ts**: `sendWithBackpressure` waits for drain; serialized ordered writes; drop-with-no-stream no-throw; **>10MB reassembly with sha256 integrity** over the real receive path. + +A shared `harness.ts` provides in-memory `MockStream`/`MockConnection`/`MockLibp2p` +(length-prefixed framing identical to the protocols) plus an `asLibp2p()` cast +helper that centralizes the one unavoidable structural mock→type assertion. + +## Verification run +- `npm run test` (Vitest) — **PASS: 252 passed / 0 failed (16 files)**, stable across 9 consecutive full runs after fixes. +- Type-check (`tsc --noEmit`) — **src/ is CLEAN**. The only remaining `tsc` errors are environmental: type-resolution noise inside the *symlinked* `node_modules` (vite `#types/*` subpaths, libp2p-yamux `PromiseWithResolvers`) and `vite.config.ts` resolving `@tailwindcss/vite` through the symlink. These reproduce on the untouched main checkout and are the symlink artifact the dispatch brief anticipated. +- `npm run lint` (ESLint) — could not execute end-to-end: `eslint.config.js` uses ESM `import` but `app/package.json` sets `"type": "commonjs"`, so under this shell's Node v26 the flat config fails to load. **This is pre-existing and environmental** — it fails identically on the clean main checkout at HEAD. I verified my code by running eslint against a temporary `.mjs` copy of the config: **my changed files introduce zero new lint errors/warnings.** The only lint hits in my files are two pre-existing `(process as any).parentPort` casts in `p2p-service.ts` (present verbatim in HEAD; line numbers merely shifted by my additions) — I left them untouched (out of scope, not mine). + +## Deviations from brief +- **Fixed a real LWW convergence bug in the inherited draft** (heterogeneous, non-symmetric tie-break) by adding `contentKey` and a convergence test. This is squarely within #42's acceptance ("identical on both peers") — I treat it as completing the issue, not exceeding it. Called out explicitly here per the no-silent-scope-creep rule. +- **Hardened the flaky `relay-manager.test.ts`** (test-only): under the concurrent suite + fake timers, the source's established `await import('@multiformats/multiaddr')` pattern didn't settle within the timer-advance window, so retries were missed ~50% of runs. Fixed in the TEST (not the source) with `vi.dynamicImportSettled()` + a bounded poll + a module stub. The production dynamic-import pattern is left intact (it mirrors `p2p-service.ts:835`). +- No wire-protocol version bump (spec-mandated; documented explicitly). + +## Uncertainty +- **`contentKey` residual assumption** — convergence holds as long as the + transmitted `data` carries the same user fields the stored doc does. If a future + schema adds a server/RxDB-default-populated field that lands in `toJSON()` but is + omitted from the replicated `data`, the tie-break could desync again. This only + bites on an *exact-millisecond* timestamp tie (astronomically rare for + independently-edited docs). Worth an Inspector eye on the exact push-side `data` + shape (`useSessionReplication.ts` / wherever `pushChanges` builds documents). +- **Checkpoint path derivation** duplicates Electron's `userData` convention in the + utility process. If the packaged app name/id ever differs from `WhatNext`, the + derived path would miss and silently full-resync. Main could inject + `WHATNEXT_CHECKPOINT_PATH` to remove the guess — deferred (not required by #40, + and the degradation is graceful). +- **Heartbeat reuse** — I used a connection-membership check (`getConnections`) + rather than extending the `ping` protocol as the spec floated. Lighter and + test-deterministic, but it detects a dropped libp2p connection, not an + application-level half-dead-but-connected peer. Adequate for the half-open-link + case the spec named; flagged in case a true keepalive is wanted later. +- **Lint not run clean end-to-end** due to the pre-existing ESM-config/Node-version + mismatch (see Verification). If the reviewer's environment runs Node v24 (the + project's pinned version) lint should load normally. + +## Deliberately deferred +- True `ping`-protocol keepalive heartbeat (membership check used instead). +- Multi-relay selection/orchestration (single-hop fallback only, per spec scope). +- Main-injected checkpoint path (env override hook is in place). + +--- + +## Fix-pass (2026-06-27) — inspector blocking items + +Focused pass resolving exactly the two blockers from `replication-verdict.md`. No other changes. + +### Blocker [1] — non-convergent LWW equal-timestamp tie-break +- **Root cause:** sender strips device-local fields (`localFilePath`, `localFileSize`, `albumArtLocalPath`, `coverArtLocalPath`) from the transmitted `data`, but the receiver computed `contentKey(existing.toJSON())` over the *un*stripped stored doc. On an exact-ms tie the existing-side key carried e.g. `localFilePath` (sorts before `name`), so the incoming side always won on BOTH peers → content swap / oscillation. +- **Fix:** introduced a single shared constant `DEVICE_LOCAL_FIELDS` in `app/src/renderer/db/schemas.ts` (source of truth). `lww.contentKey` now drops these fields (merged into `NON_CONTENT_KEYS`), and the sender in `useSessionReplication.ts` now strips by iterating that same constant (replacing the per-collection hard-coded `delete`s). Sender-strip and tiebreak-strip can no longer drift. +- **Scope:** receiver-side tie-break only. Wire format, message types, and protocol id `/whatnext/rxdb-replication/1.0.0` unchanged. + +### Blocker [2] — convergence regression test +- Added two cases to `lww.test.ts`: + 1. `contentKey` equality when the stored doc carries all four device-local fields (+ `_rev`) vs the stripped transmitted payload. + 2. Full convergence assertion on an equal-timestamp tie where only one peer has `localFilePath`/`localFileSize` set — asserts both peers elect the same winner (`qWinsOnP === !pWinsOnQ`). This is the exact production shape the prior fixture omitted. + +### Doc reconciliation +- `WhatNext - docs/01 concepts/RxDB-Replication.md` § Conflict Resolution: added a bullet stating device-local fields are excluded from the content tie-break key, with the single-source-of-truth note. + +### Verification +- `npm run typecheck` — clean for `src/`; only the pre-existing `vite.config.ts` / `@tailwindcss/vite` node_modules baseline error remains (known, out of scope). +- `npx vitest run` (full suite) — **254 passed / 0 failed (16 files)** (252 baseline + 2 new lww cases). +- `npm run lint` — not run (pre-existing env-broken eslint flat-config, out of scope per brief). diff --git a/WhatNext - docs/01 concepts/RxDB-Replication.md b/WhatNext - docs/01 concepts/RxDB-Replication.md index 9e48a68..7bb8654 100644 --- a/WhatNext - docs/01 concepts/RxDB-Replication.md +++ b/WhatNext - docs/01 concepts/RxDB-Replication.md @@ -97,6 +97,7 @@ Last-Write-Wins based on the `updatedAt` timestamp, **hardened to be skew-aware - 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. diff --git a/app/src/renderer/db/__tests__/lww.test.ts b/app/src/renderer/db/__tests__/lww.test.ts index 4c92edb..a348cd8 100644 --- a/app/src/renderer/db/__tests__/lww.test.ts +++ b/app/src/renderer/db/__tests__/lww.test.ts @@ -155,6 +155,66 @@ describe('contentKey (cross-peer tie-break symmetry)', () => { // 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', () => { diff --git a/app/src/renderer/db/lww.ts b/app/src/renderer/db/lww.ts index bbf8fc0..0ded797 100644 --- a/app/src/renderer/db/lww.ts +++ b/app/src/renderer/db/lww.ts @@ -16,6 +16,8 @@ * path is CRDTs (Phase 2). See [[RxDB-Replication]] "Conflict Resolution". */ +import { DEVICE_LOCAL_FIELDS } from './schemas'; + /** * Parse a timestamp value into epoch milliseconds. * @@ -110,14 +112,26 @@ export function incomingWins(incoming: LwwCandidate, existing: LwwCandidate): bo * `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. + * `_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']); +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)) { 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, From 8ac964eef6de5953cad767b383a7348e71bcad67 Mon Sep 17 00:00:00 2001 From: Joseph Madigan Date: Thu, 16 Jul 2026 13:24:19 -0700 Subject: [PATCH 7/7] chore: drop agent impl-notes scratch from branch The Architect handoff notes are working artifacts of the agent cycle, not project source. They were the only add/add conflict when integrating the Wave-1 lanes, and they do not belong in the reviewable diff. Removing them here keeps the PR diff to real changes; .claude/ scratch is gitignored on mvp. --- .claude/cycle/impl-notes.md | 161 ------------------------------------ 1 file changed, 161 deletions(-) delete mode 100644 .claude/cycle/impl-notes.md diff --git a/.claude/cycle/impl-notes.md b/.claude/cycle/impl-notes.md deleted file mode 100644 index 9279c8d..0000000 --- a/.claude/cycle/impl-notes.md +++ /dev/null @@ -1,161 +0,0 @@ -# Implementation notes - -**Brief:** .claude/cycle/brief.md (cycle started 2026-06-27) -**Epic:** WhatNext - docs/08 specs/epic-replication-reliability.md -**Lane:** replication-reliability (Wave 1, P2P-approved) -**Architect run:** 2026-06-27 - -## What I built -Hardened the five reliability surfaces of the P2P RxDB replication stack: durable -per-peer/per-collection checkpoints (#40), a non-silent pull-timeout plus -exponential-backoff relay reconnection with single-fallback and a liveness -heartbeat (#41), skew-aware epoch-ms LWW conflict resolution with a deterministic, -cross-peer-convergent tie-break (#42), backpressure-safe file-chunk sends so ->10MB transfers no longer reset the stream (#47), and a from-scratch Vitest P2P -harness with suites covering all of the above (#32). The `[[RxDB-Replication]]` -concept page is updated to document the new behavior. **No wire-protocol version -bump** — the protocol id stays `/whatnext/rxdb-replication/1.0.0`. - -Note on provenance: I inherited a substantial, uncommitted draft of this work -already present in the worktree from a prior Architect pass. I reviewed every line, -fixed the defects I found (a flaky test, type errors, and a real LWW convergence -bug — see below), added missing coverage, verified, documented, and committed it. - -## Design decisions - -- **Decision:** Checkpoint store = a single debounced JSON file under `userData`, - owned/written by the utility process (path derived from platform conventions, - `WHATNEXT_CHECKPOINT_PATH` env override for tests). - **Alternatives considered:** a dedicated RxDB collection in the renderer (the - spec's other candidate). - **Rationale:** a flat file is simpler and decoupled from the very DB being - replicated; the renderer-collection option couples checkpoint durability to - renderer liveness and the replicated dataset. - **Tradeoff:** the utility process can't call `app.getPath('userData')`, so it - re-derives the path from platform conventions — a small duplication of Electron's - own logic that could drift if the app name changes. Mitigated by the env override - and a corrupt/missing-file → full-resync fallback (degraded, never broken). - -- **Decision:** On pull timeout, REJECT and echo the requester's incoming - checkpoint back (do not advance); responder advances the checkpoint to the - newest returned doc's `updatedAt`, never wall-clock `now`. - **Alternatives considered:** keep "resolve empty + fresh checkpoint" (the bug); - advance to `now` on success. - **Rationale:** advancing on a timeout or to `now` silently skips unsent docs — - the exact data-loss bug. Echoing the incoming checkpoint re-pulls next time. - **Tradeoff:** a permanently-slow peer re-attempts the same pull each cycle - (no progress) rather than declaring false success — correct, but it trades a - silent wrong answer for a visible retry loop. Acceptable and observable. - -- **Decision:** Relay backoff/fallback/heartbeat as pure, separately-tested - helpers (`backoff.ts`, `nextFallbackAddress`, `checkRelayLiveness`). - **Rationale:** keeps the schedule and ring-fallback policy unit-testable without - a real libp2p network. - **Tradeoff:** fallback is single-hop only (spec-scoped); multi-relay - orchestration is explicitly out of scope. - -- **Decision (correctness fix beyond the inherited draft):** LWW equal-timestamp - tie-break uses a `contentKey()` projection that strips `id`, `updatedAt`, - `addedAt`, and all `_`-prefixed RxDB-internal fields before stable - serialization, applied identically to both the incoming `data` payload and the - existing `RxDocument.toJSON()`. - **Alternatives considered:** the inherited code compared - `stableStringify(doc.data)` against `stableStringify(existingData)` — i.e. a - transmitted user-fields payload against a full toJSON carrying `_rev`/`_meta`/ - `id`/`updatedAt`. That is heterogeneous and **non-symmetric across peers**, so - on an exact-ms tie peer A and peer B could elect different winners and diverge — - violating acceptance #42 ("identical on both peers"). - **Rationale:** every stripped key is non-discriminating at a tie (id equal, - timestamps equal/creation-time) and removed identically from both sides, so - `contentKey(data) === contentKey(toJSON)` for the same user content → symmetric - and convergent. - **Tradeoff:** relies on the residual assumption that the transmitted `data` - carries the same user fields the stored doc does (which must hold for - replication to function at all). Flagged under Uncertainty. - -## Files changed -- `app/src/utility/checkpoint-store.ts` — NEW: durable, debounced, atomic JSON checkpoint store (#40). -- `app/src/utility/backoff.ts` — NEW: pure exponential-backoff-with-jitter helper (#41). -- `app/src/utility/relay-manager.ts` — backoff + ring fallback + liveness heartbeat (#41). -- `app/src/shared/p2p-config.ts` — new `RELAY` backoff/heartbeat constants + `REPLICATION.PULL_TIMEOUT`; deprecate `RETRY_INTERVAL` (#41). -- `app/src/utility/p2p-service.ts` — wire in the checkpoint store; reject pull timeouts; act on pull-response (#40, #41). -- `app/src/utility/protocols/replication.ts` — `newestCheckpoint()` + `onPullResponse` requester handler (#40/#41). -- `app/src/renderer/db/lww.ts` — NEW: skew-aware parse + symmetric `contentKey` tie-break (#42). -- `app/src/renderer/db/replication-handler.ts` — use `incomingWins`/`contentKey` instead of string compare (#42). -- `app/src/utility/protocols/file-transfer.ts` — `sendWithBackpressure` + per-stream serialized send queue; TODO removed (#47). -- `WhatNext - docs/01 concepts/RxDB-Replication.md` — document all of the above; explicit "no version bump" note (P2P doc requirement). -- Tests (#32): `app/src/utility/protocols/__tests__/{harness.ts,handshake.test.ts,replication.test.ts,file-transfer.test.ts}`, `app/src/utility/__tests__/{backoff,checkpoint-store,relay-manager}.test.ts`, `app/src/renderer/db/__tests__/lww.test.ts`. - -## Tests -- **lww.test.ts** (26 cases): epoch parse of mixed precision/missing/numeric; updatedAt-over-addedAt; strictly-later wins; differing-precision equal instants as a tie; deterministic content tie-break; `contentKey` strips id/timestamps/`_`-fields; **explicit cross-peer convergence test** proving the two peers elect the same winner despite the data-vs-toJSON shape asymmetry. -- **checkpoint-store.test.ts**: round-trip + restart simulation; unknown-key null; corrupt/absent/non-string-value graceful degradation; unwritable-dir no-crash; debounce coalescing; no-op set; path resolution + env override. -- **backoff.test.ts**: monotonic growth, ceiling cap, jitter bounds, zero-jitter determinism. -- **relay-manager.test.ts**: `nextFallbackAddress` ring logic; backoff retry fires repeatedly; fallback to next relay on exhaustion; heartbeat detects half-open + reports live. -- **handshake.test.ts**: metadata exchange + reply; capability-mismatch surfaced not rejected; empty-stream no-op. -- **replication.test.ts**: `newestCheckpoint` selection/preservation/epoch-fallback; pull-request→pull-response; push→push-ack; pull-response forwarded to `onPullResponse`; backward-compat when callback omitted. -- **file-transfer.test.ts**: `sendWithBackpressure` waits for drain; serialized ordered writes; drop-with-no-stream no-throw; **>10MB reassembly with sha256 integrity** over the real receive path. - -A shared `harness.ts` provides in-memory `MockStream`/`MockConnection`/`MockLibp2p` -(length-prefixed framing identical to the protocols) plus an `asLibp2p()` cast -helper that centralizes the one unavoidable structural mock→type assertion. - -## Verification run -- `npm run test` (Vitest) — **PASS: 252 passed / 0 failed (16 files)**, stable across 9 consecutive full runs after fixes. -- Type-check (`tsc --noEmit`) — **src/ is CLEAN**. The only remaining `tsc` errors are environmental: type-resolution noise inside the *symlinked* `node_modules` (vite `#types/*` subpaths, libp2p-yamux `PromiseWithResolvers`) and `vite.config.ts` resolving `@tailwindcss/vite` through the symlink. These reproduce on the untouched main checkout and are the symlink artifact the dispatch brief anticipated. -- `npm run lint` (ESLint) — could not execute end-to-end: `eslint.config.js` uses ESM `import` but `app/package.json` sets `"type": "commonjs"`, so under this shell's Node v26 the flat config fails to load. **This is pre-existing and environmental** — it fails identically on the clean main checkout at HEAD. I verified my code by running eslint against a temporary `.mjs` copy of the config: **my changed files introduce zero new lint errors/warnings.** The only lint hits in my files are two pre-existing `(process as any).parentPort` casts in `p2p-service.ts` (present verbatim in HEAD; line numbers merely shifted by my additions) — I left them untouched (out of scope, not mine). - -## Deviations from brief -- **Fixed a real LWW convergence bug in the inherited draft** (heterogeneous, non-symmetric tie-break) by adding `contentKey` and a convergence test. This is squarely within #42's acceptance ("identical on both peers") — I treat it as completing the issue, not exceeding it. Called out explicitly here per the no-silent-scope-creep rule. -- **Hardened the flaky `relay-manager.test.ts`** (test-only): under the concurrent suite + fake timers, the source's established `await import('@multiformats/multiaddr')` pattern didn't settle within the timer-advance window, so retries were missed ~50% of runs. Fixed in the TEST (not the source) with `vi.dynamicImportSettled()` + a bounded poll + a module stub. The production dynamic-import pattern is left intact (it mirrors `p2p-service.ts:835`). -- No wire-protocol version bump (spec-mandated; documented explicitly). - -## Uncertainty -- **`contentKey` residual assumption** — convergence holds as long as the - transmitted `data` carries the same user fields the stored doc does. If a future - schema adds a server/RxDB-default-populated field that lands in `toJSON()` but is - omitted from the replicated `data`, the tie-break could desync again. This only - bites on an *exact-millisecond* timestamp tie (astronomically rare for - independently-edited docs). Worth an Inspector eye on the exact push-side `data` - shape (`useSessionReplication.ts` / wherever `pushChanges` builds documents). -- **Checkpoint path derivation** duplicates Electron's `userData` convention in the - utility process. If the packaged app name/id ever differs from `WhatNext`, the - derived path would miss and silently full-resync. Main could inject - `WHATNEXT_CHECKPOINT_PATH` to remove the guess — deferred (not required by #40, - and the degradation is graceful). -- **Heartbeat reuse** — I used a connection-membership check (`getConnections`) - rather than extending the `ping` protocol as the spec floated. Lighter and - test-deterministic, but it detects a dropped libp2p connection, not an - application-level half-dead-but-connected peer. Adequate for the half-open-link - case the spec named; flagged in case a true keepalive is wanted later. -- **Lint not run clean end-to-end** due to the pre-existing ESM-config/Node-version - mismatch (see Verification). If the reviewer's environment runs Node v24 (the - project's pinned version) lint should load normally. - -## Deliberately deferred -- True `ping`-protocol keepalive heartbeat (membership check used instead). -- Multi-relay selection/orchestration (single-hop fallback only, per spec scope). -- Main-injected checkpoint path (env override hook is in place). - ---- - -## Fix-pass (2026-06-27) — inspector blocking items - -Focused pass resolving exactly the two blockers from `replication-verdict.md`. No other changes. - -### Blocker [1] — non-convergent LWW equal-timestamp tie-break -- **Root cause:** sender strips device-local fields (`localFilePath`, `localFileSize`, `albumArtLocalPath`, `coverArtLocalPath`) from the transmitted `data`, but the receiver computed `contentKey(existing.toJSON())` over the *un*stripped stored doc. On an exact-ms tie the existing-side key carried e.g. `localFilePath` (sorts before `name`), so the incoming side always won on BOTH peers → content swap / oscillation. -- **Fix:** introduced a single shared constant `DEVICE_LOCAL_FIELDS` in `app/src/renderer/db/schemas.ts` (source of truth). `lww.contentKey` now drops these fields (merged into `NON_CONTENT_KEYS`), and the sender in `useSessionReplication.ts` now strips by iterating that same constant (replacing the per-collection hard-coded `delete`s). Sender-strip and tiebreak-strip can no longer drift. -- **Scope:** receiver-side tie-break only. Wire format, message types, and protocol id `/whatnext/rxdb-replication/1.0.0` unchanged. - -### Blocker [2] — convergence regression test -- Added two cases to `lww.test.ts`: - 1. `contentKey` equality when the stored doc carries all four device-local fields (+ `_rev`) vs the stripped transmitted payload. - 2. Full convergence assertion on an equal-timestamp tie where only one peer has `localFilePath`/`localFileSize` set — asserts both peers elect the same winner (`qWinsOnP === !pWinsOnQ`). This is the exact production shape the prior fixture omitted. - -### Doc reconciliation -- `WhatNext - docs/01 concepts/RxDB-Replication.md` § Conflict Resolution: added a bullet stating device-local fields are excluded from the content tie-break key, with the single-source-of-truth note. - -### Verification -- `npm run typecheck` — clean for `src/`; only the pre-existing `vite.config.ts` / `@tailwindcss/vite` node_modules baseline error remains (known, out of scope). -- `npx vitest run` (full suite) — **254 passed / 0 failed (16 files)** (252 baseline + 2 new lww cases). -- `npm run lint` — not run (pre-existing env-broken eslint flat-config, out of scope per brief).