| … | … | @@ -32,8 +32,6 @@ export { |
| 32 | 32 | export { ChunkScheduler } from './lib/chunk-scheduler.js' |
| 33 | 33 | export { FileIndex, CONTENT_REGISTER_PREFIX } from './lib/file-index.js' |
| 34 | 34 | export { SyncEngine, STRATEGIES } from './lib/sync-engine.js' |
| 35 | export { reconcile } from './lib/sync-reconciler.js' | |
| 36 | export { attachReplication } from './lib/overlay-replicator.js' | |
| 37 | 35 | export { |
| 38 | 36 | chunk, |
| 39 | 37 | chunkStream, |
| … | … | @@ -1,113 +0,0 @@ |
| 1 | /** | |
| 2 | * attachReplication — keep an overlay's file tree converged with ONE peer over a connection, | |
| 3 | * using hyper-overlay's NATIVE convergence (SyncEngine change-feeds that replicate as Hypercores). | |
| 4 | * | |
| 5 | * The overlay's own design (sync-engine.js): "Each peer has a Hypercore feed logging file change | |
| 6 | * operations. When peers connect, feeds replicate automatically (Hypercore handles this)." So the | |
| 7 | * change list is NOT hand-shipped over a bespoke channel — it rides the host's corestore | |
| 8 | * replication like any other core, and diff() reads the REAL peer feed. Two things cross the wire: | |
| 9 | * | |
| 10 | * 1. the CHANGE-FEED (pointer ops) → the peer's sync-feed Hypercore, replicated by the HOST's | |
| 11 | * corestore.replicate(conn) (the daemon calls store.replicate(conn) once per connection; a test | |
| 12 | * harness does the same). We put a standing download on the peer feed so its blocks actively pull | |
| 13 | * and the core re-attaches direction-free on reconnect (the did-peer.js:74 / membrane lesson). | |
| 14 | * 2. file BYTES → the overlay content protocol (attachProtocol + fetchFile), a request/response | |
| 15 | * protomux channel that reads chunks from disk on demand. Unchanged, proven. | |
| 16 | * | |
| 17 | * The ONLY thing exchanged as a message here is the peer's sync-feed KEY (once, at connect) so we can | |
| 18 | * open its feed by key — everything after is native Hypercore replication + SyncEngine.diff. | |
| 19 | * | |
| 20 | * HOST CONTRACT: the caller MUST call `corestore.replicate(conn)` on the overlay's corestore for this | |
| 21 | * connection (the multi-DID daemon already does `store.replicate(conn)` in its connection handler). | |
| 22 | * attachReplication does NOT replicate the store itself — that would double-replicate on the daemon. | |
| 23 | * | |
| 24 | * `fileRoot`: diskPath = join(fileRoot, overlayPath). `strategy`: reconcile conflict strategy | |
| 25 | * ('rev-wins' for atproto TIDs, 'newest-wins' generic). Single peer per connection. | |
| 26 | */ | |
| 27 | ||
| 28 | import Protomux from 'protomux' | |
| 29 | import c from 'compact-encoding' | |
| 30 | import fs from 'bare-fs' | |
| 31 | import { join, dirname } from 'bare-path' | |
| 32 | import { reconcile } from './sync-reconciler.js' | |
| 33 | ||
| 34 | const DBG = typeof process !== 'undefined' && process.env && process.env.HODEBUG ? (o) => { try { console.error('[repl] ' + JSON.stringify(o)) } catch {} } : () => {} | |
| 35 | ||
| 36 | // One tiny message per connection: exchange our sync-feed key. id = overlay._key so N overlays | |
| 37 | // (a multi-DID daemon: one overlay per DID) don't collide on one shared mux. | |
| 38 | const FEEDKEY_PROTOCOL = 'hyper-overlay/sync-feedkey/v1' | |
| 39 | ||
| 40 | export async function attachReplication (overlay, conn, { strategy = 'newest-wins', fileRoot, onConverge } = {}) { | |
| 41 | if (!fileRoot) throw new Error('attachReplication: fileRoot required') | |
| 42 | await overlay._ensure() // build the overlay stack (content protocol + sync engine) if not already | |
| 43 | const engine = overlay._sync | |
| 44 | const mux = Protomux.from(conn) | |
| 45 | ||
| 46 | // Content byte channel (bytes). MUST be attached synchronously per connection — protomux won't pair | |
| 47 | // a channel opened after the remote's (overlay.js:158). The sync-FEEDS ride the host's | |
| 48 | // corestore.replicate(conn), NOT this channel. | |
| 49 | overlay.attachProtocol(mux) | |
| 50 | ||
| 51 | const diskFor = (p) => join(fileRoot, String(p).replace(/^\/+/, '')) | |
| 52 | // Pull: fetch the bytes by content-hash to the record path (fetchFile fans out to every connected | |
| 53 | // content peer), then register (servable + logged to OUR feed so we announce we now hold it). | |
| 54 | const pull = async (overlayPath, remote) => { | |
| 55 | const disk = diskFor(overlayPath) | |
| 56 | fs.mkdirSync(dirname(disk), { recursive: true }) | |
| 57 | const r = await overlay.fetchFile(remote.hash, { destPath: disk }) | |
| 58 | if (!r) throw new Error('attachReplication: content fetch failed for ' + remote.hash) | |
| 59 | await overlay.registerFile(overlayPath, disk, { contentHash: remote.hash, size: remote.size, rev: remote.rev }) | |
| 60 | } | |
| 61 | const remove = async (overlayPath) => { | |
| 62 | try { fs.unlinkSync(diskFor(overlayPath)) } catch {} | |
| 63 | await engine.logDel(overlayPath) | |
| 64 | } | |
| 65 | ||
| 66 | let peerKey = null | |
| 67 | let running = false; let again = false | |
| 68 | const runReconcile = async () => { | |
| 69 | if (!peerKey) return | |
| 70 | if (running) { again = true; return } // coalesce concurrent triggers; re-run once if one arrived mid-flight | |
| 71 | running = true | |
| 72 | try { const s = await reconcile(engine, peerKey, { strategy, pull, remove }); DBG({ ev: 'reconcile', peerLen: peerFeed ? peerFeed.length : null, pulled: s.pulled, pushed: s.pushed, skipped: s.skipped, failed: s.failed.length, err: s.failed[0] && s.failed[0].error }); if (onConverge) onConverge(s) } catch (e) { DBG({ ev: 'reconcile-throw', error: e && e.message }) } | |
| 73 | finally { running = false; if (again) { again = false; runReconcile() } } | |
| 74 | } | |
| 75 | ||
| 76 | // Feed-key handshake: send our sync-feed key ONCE; on the peer's key, open its feed by key (it | |
| 77 | // replicates over the host's corestore.replicate), put a STANDING download so blocks pull | |
| 78 | // direction-free + the core re-attaches on reconnect, and reconcile on every append as replicated | |
| 79 | // blocks land. This is the native path — diff() then reads the real, replicated peer feed. | |
| 80 | let peerFeed = null | |
| 81 | let onPeerAppend = null | |
| 82 | const chan = mux.createChannel({ protocol: FEEDKEY_PROTOCOL, id: overlay._key || null }) | |
| 83 | const msg = chan.addMessage({ | |
| 84 | encoding: c.json, | |
| 85 | onmessage: async (data) => { | |
| 86 | if (!data || !data.feedKey || peerKey) return | |
| 87 | peerKey = String(data.feedKey) | |
| 88 | DBG({ ev: 'recv-feedkey', peerKey: peerKey.slice(0, 12) }) | |
| 89 | peerFeed = engine.addPeerFeed(peerKey) | |
| 90 | try { await peerFeed.ready() } catch (e) { DBG({ ev: 'peerfeed-ready-throw', error: e && e.message }); return } | |
| 91 | try { peerFeed.download({ start: 0, end: -1 }) } catch {} // standing pull (did-peer.js:74 lesson) | |
| 92 | onPeerAppend = () => { DBG({ ev: 'peerfeed-append', len: peerFeed.length }); runReconcile().catch(() => {}) } | |
| 93 | peerFeed.on('append', onPeerAppend) // fires as the peer's feed replicates block-by-block | |
| 94 | await runReconcile() | |
| 95 | } | |
| 96 | }) | |
| 97 | chan.open() | |
| 98 | ||
| 99 | await engine.feed.ready() | |
| 100 | try { msg.send({ feedKey: engine.feedKey }); DBG({ ev: 'sent-feedkey', feedKey: engine.feedKey.slice(0, 12) }) } catch (e) { DBG({ ev: 'send-feedkey-throw', error: e && e.message }) } | |
| 101 | // Local authoring grows our feed; the peer pulls it on ITS reconcile (its copy of our feed appends). | |
| 102 | // We also reconcile locally so a pending pull/conflict is re-evaluated the moment we change state. | |
| 103 | const onLocalAppend = () => { runReconcile().catch(() => {}) } | |
| 104 | engine.feed.on('append', onLocalAppend) | |
| 105 | ||
| 106 | return { | |
| 107 | reconcileNow: runReconcile, | |
| 108 | close: () => { | |
| 109 | try { engine.feed.off('append', onLocalAppend) } catch {} | |
| 110 | try { if (peerFeed && onPeerAppend) peerFeed.off('append', onPeerAppend) } catch {} | |
| 111 | } | |
| 112 | } | |
| 113 | } |
| … | … | @@ -219,9 +219,9 @@ export class HyperOverlay extends ReadyResource { |
| 219 | 219 | // ONCE, persists the content-addressed chunk map (durable in FileIndex), and |
| 220 | 220 | // returns the content hash. Replaces a hash-only pass plus a lazy fetch-time |
| 221 | 221 | // re-chunk with a single read, so the first fetcher never waits on chunk indexing. |
| 222 | // Stop serving a path AND record the deletion in the change feed, so the delete propagates to peers | |
| 223 | // (attachReplication's remote peers see the del op and remove their copy). The authoring counterpart | |
| 224 | // to a shim deleting a record. Idempotent. | |
| 222 | // Stop serving a path AND record the deletion in the local change feed (bookkeeping; autoSync is off | |
| 223 | // so it is not broadcast — convergence is an application concern, not this library's). The authoring | |
| 224 | // counterpart to a shim deleting a record. Idempotent. | |
| 225 | 225 | async unregisterFile (overlayPath) { |
| 226 | 226 | await this._ensure() |
| 227 | 227 | this._filePaths.delete(overlayPath) |
| … | … | @@ -1,98 +0,0 @@ |
| 1 | /** | |
| 2 | * reconcile — drive SyncEngine.diff() into APPLIED merge actions. | |
| 3 | * | |
| 4 | * SyncEngine computes a 3-way diff (mine / theirs / common-ancestor) but does not itself | |
| 5 | * apply the result — this is the missing orchestration. For one connected peer: run diff(), | |
| 6 | * resolve conflicts by a strategy, apply pulls/deletes via injected effects, and record | |
| 7 | * markSynced so the next round is stable (idempotent once converged). | |
| 8 | * | |
| 9 | * Effects are INJECTED so this is testable offline and reusable in prod: | |
| 10 | * pull(path, remote) — fetch remote content + write the file locally. | |
| 11 | * In prod this is overlay.fetchContent(remote.hash) → registerFile → logPut, | |
| 12 | * so the local feed reflects the adopted content (+ its rev). | |
| 13 | * remove(path) — delete the local file (prod: unlink + logDel). | |
| 14 | * | |
| 15 | * Strategy: any SyncEngine STRATEGY, PLUS 'rev-wins' — resolve a conflict by the atproto `rev` | |
| 16 | * (a TID, lexicographically time-ordered; higher wins) carried on each change entry, falling | |
| 17 | * back to seq-based 'newest-wins' when either side lacks a rev. | |
| 18 | * | |
| 19 | * Symmetry: each peer runs reconcile against the other. A 'push'/'delete-remote' is a local | |
| 20 | * no-op (the peer PULLS it on its own reconcile) — we only markSynced so we don't re-flag it. | |
| 21 | */ | |
| 22 | ||
| 23 | import { STRATEGIES } from './sync-engine.js' | |
| 24 | ||
| 25 | // Resolve a conflict to a concrete action. 'rev-wins' compares the atproto rev (TID); everything | |
| 26 | // else delegates to SyncEngine.resolveConflict. | |
| 27 | function resolveConflict (engine, conflict, strategy) { | |
| 28 | if (strategy === 'rev-wins') { | |
| 29 | const lr = conflict.local && conflict.local.rev | |
| 30 | const rr = conflict.remote && conflict.remote.rev | |
| 31 | if (lr != null && rr != null) { | |
| 32 | // TIDs sort lexicographically by time; the higher rev is the newer write and wins. | |
| 33 | return String(lr) >= String(rr) ? { action: 'push', winner: 'local' } : { action: 'pull', winner: 'remote' } | |
| 34 | } | |
| 35 | return engine.resolveConflict(conflict, 'newest-wins') // a side has no rev → fall back to feed seq | |
| 36 | } | |
| 37 | return engine.resolveConflict(conflict, strategy) | |
| 38 | } | |
| 39 | ||
| 40 | /** | |
| 41 | * Reconcile local state with one peer. | |
| 42 | * @param {SyncEngine} engine - the local sync engine | |
| 43 | * @param {string} peerKey - hex feed key of the peer (already added via engine.addPeerFeed) | |
| 44 | * @param {{ strategy?: string, pull: Function, remove?: Function }} opts | |
| 45 | * @returns {Promise<{pulled,removed,pushed,skipped,conflicts,keptBoth,deferred,failed}>} | |
| 46 | */ | |
| 47 | export async function reconcile (engine, peerKey, opts = {}) { | |
| 48 | const { strategy = 'newest-wins', pull, remove } = opts | |
| 49 | if (typeof pull !== 'function') throw new Error('reconcile: opts.pull(path, remote) is required') | |
| 50 | if (strategy !== 'rev-wins' && !STRATEGIES.includes(strategy)) throw new Error('reconcile: unknown strategy ' + strategy) | |
| 51 | ||
| 52 | const actions = await engine.diff(peerKey) | |
| 53 | const summary = { pulled: 0, removed: 0, pushed: 0, skipped: 0, conflicts: 0, keptBoth: 0, deferred: 0, failed: [] } | |
| 54 | ||
| 55 | for (const a of actions) { | |
| 56 | let action = a.action | |
| 57 | if (action === 'conflict') { | |
| 58 | summary.conflicts++ | |
| 59 | action = resolveConflict(engine, a, strategy).action | |
| 60 | } | |
| 61 | try { | |
| 62 | switch (action) { | |
| 63 | case 'pull': { | |
| 64 | await pull(a.path, a.remote) | |
| 65 | await engine.markSynced(peerKey, a.path, a.remote ? a.remote.hash : null, a.remote ? a.remote.seq : -1) | |
| 66 | summary.pulled++ | |
| 67 | break | |
| 68 | } | |
| 69 | case 'delete-local': { | |
| 70 | if (remove) await remove(a.path) | |
| 71 | await engine.markSynced(peerKey, a.path, null, a.remote ? a.remote.seq : -1) | |
| 72 | summary.removed++ | |
| 73 | break | |
| 74 | } | |
| 75 | case 'push': | |
| 76 | case 'delete-remote': { | |
| 77 | // The peer pulls this from us on its own reconcile; agree on our current state so we don't re-flag it. | |
| 78 | await engine.markSynced(peerKey, a.path, a.local ? a.local.hash : null, a.local ? a.local.seq : -1) | |
| 79 | summary.pushed++ | |
| 80 | break | |
| 81 | } | |
| 82 | case 'skip': { | |
| 83 | const hash = (a.local && a.local.hash) || (a.remote && a.remote.hash) || null | |
| 84 | const seq = Math.max(a.local ? a.local.seq : -1, a.remote ? a.remote.seq : -1) | |
| 85 | await engine.markSynced(peerKey, a.path, hash, seq) | |
| 86 | summary.skipped++ | |
| 87 | break | |
| 88 | } | |
| 89 | case 'keep-both': { summary.keptBoth++; break } // app resolves (rename); no markSynced here | |
| 90 | default: { summary.deferred++; break } // 'prompt' / resolved-to-skip: leave for later | |
| 91 | } | |
| 92 | } catch (err) { | |
| 93 | // No markSynced on failure → the action is retried on the next reconcile pass. | |
| 94 | summary.failed.push({ path: a.path, action, error: err.message }) | |
| 95 | } | |
| 96 | } | |
| 97 | return summary | |
| 98 | } |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "hyper-overlay", |
| 3 | "version": "0.3.7", | |
| 3 | "version": "0.3.8", | |
| 4 | 4 | "description": "Content-addressed P2P file distribution over Protomux \u2014 FastCDC chunking, multi-source fetch, resumable transfers, zero blob duplication. Files live on disk; the overlay is a thin index and transfer protocol.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -1,87 +0,0 @@ |
| 1 | // Offline proof of generic file-tree convergence: two HyperOverlay peers over an in-memory pipe. | |
| 2 | // The change list (pointer deltas) is EXCHANGED over a dedicated channel; bytes ride the content | |
| 3 | // protocol; reconcile drives convergence with atproto rev-based conflict resolution. No | |
| 4 | // corestore.replicate (sidesteps the shared-mux content-channel hazard). This is the reusable | |
| 5 | // record-replication capability the atproto virtual device (and any other shim) sits on top of. | |
| 6 | ||
| 7 | import test from 'brittle' | |
| 8 | import { Duplex } from 'streamx' | |
| 9 | import { tmpStore, tmpDir, fs, path } from './helpers.js' | |
| 10 | import { HyperOverlay } from '../lib/overlay.js' | |
| 11 | import { attachReplication } from '../lib/overlay-replicator.js' | |
| 12 | ||
| 13 | function makeDuplex () { | |
| 14 | let aw, bw | |
| 15 | const a = new Duplex({ write (d, cb) { bw(d); cb() }, read () {} }) | |
| 16 | const b = new Duplex({ write (d, cb) { aw(d); cb() }, read () {} }) | |
| 17 | aw = (d) => a.push(d); bw = (d) => b.push(d) | |
| 18 | return [a, b] | |
| 19 | } | |
| 20 | const settle = (ms = 900) => new Promise((r) => setTimeout(r, ms)) | |
| 21 | ||
| 22 | async function device (name) { | |
| 23 | const store = tmpStore('repl-' + name) | |
| 24 | const overlay = new HyperOverlay(store, { namespace: 'overlay-v2', destDir: tmpDir('repl-' + name + '-dest') }) | |
| 25 | const root = tmpDir('repl-' + name + '-files') | |
| 26 | return { name, store, overlay, root } | |
| 27 | } | |
| 28 | async function author (dev, overlayPath, text, rev) { | |
| 29 | const disk = path.join(dev.root, overlayPath.replace(/^\/+/, '')) | |
| 30 | fs.mkdirSync(path.dirname(disk), { recursive: true }) | |
| 31 | fs.writeFileSync(disk, Buffer.from(text)) | |
| 32 | await dev.overlay.registerFile(overlayPath, disk, { size: text.length, rev }) | |
| 33 | } | |
| 34 | async function removeRecord (dev, overlayPath) { // simulate a shim deleting a record (unlink + logDel) | |
| 35 | try { fs.unlinkSync(path.join(dev.root, overlayPath.replace(/^\/+/, ''))) } catch {} | |
| 36 | await dev.overlay._sync.logDel(overlayPath) | |
| 37 | } | |
| 38 | const readText = (dev, overlayPath) => { try { return fs.readFileSync(path.join(dev.root, overlayPath.replace(/^\/+/, ''))).toString() } catch { return null } } | |
| 39 | ||
| 40 | async function connect (A, B) { | |
| 41 | const [sa, sb] = makeDuplex() | |
| 42 | const ra = await attachReplication(A.overlay, sa, { strategy: 'rev-wins', fileRoot: A.root }) | |
| 43 | const rb = await attachReplication(B.overlay, sb, { strategy: 'rev-wins', fileRoot: B.root }) | |
| 44 | return { ra, rb } | |
| 45 | } | |
| 46 | ||
| 47 | test('two peers converge disjoint records — real content transfer over the pipe', async (t) => { | |
| 48 | const A = await device('A'); const B = await device('B') | |
| 49 | t.teardown(async () => { try { await A.overlay.close() } catch {} try { await B.overlay.close() } catch {} }) | |
| 50 | await connect(A, B) | |
| 51 | ||
| 52 | await author(A, '/records/app.bsky.feed.post/x.dagcbor', 'post-x-content', '3t01') | |
| 53 | await author(B, '/records/app.bsky.feed.post/y.dagcbor', 'post-y-content', '3t02') | |
| 54 | await settle() | |
| 55 | ||
| 56 | t.is(readText(B, '/records/app.bsky.feed.post/x.dagcbor'), 'post-x-content', "B pulled A's record x (bytes)") | |
| 57 | t.is(readText(A, '/records/app.bsky.feed.post/y.dagcbor'), 'post-y-content', "A pulled B's record y (bytes)") | |
| 58 | }) | |
| 59 | ||
| 60 | test('divergent records resolve by rev — higher TID wins on both sides', async (t) => { | |
| 61 | const A = await device('A'); const B = await device('B') | |
| 62 | t.teardown(async () => { try { await A.overlay.close() } catch {} try { await B.overlay.close() } catch {} }) | |
| 63 | ||
| 64 | // Two devices independently authored different versions of the SAME record while apart, then meet | |
| 65 | // (the local-first offline-edit case). Author BEFORE connecting so the edits are truly concurrent. | |
| 66 | await author(A, '/records/c/z.dagcbor', 'z-vA', '3t05') | |
| 67 | await author(B, '/records/c/z.dagcbor', 'z-vB', '3t09') // LATER rev | |
| 68 | await connect(A, B) | |
| 69 | await settle() | |
| 70 | ||
| 71 | t.is(readText(A, '/records/c/z.dagcbor'), 'z-vB', 'A adopted the higher-rev (B) version') | |
| 72 | t.is(readText(B, '/records/c/z.dagcbor'), 'z-vB', 'B kept its higher-rev version') | |
| 73 | }) | |
| 74 | ||
| 75 | test('delete propagates to the peer', async (t) => { | |
| 76 | const A = await device('A'); const B = await device('B') | |
| 77 | t.teardown(async () => { try { await A.overlay.close() } catch {} try { await B.overlay.close() } catch {} }) | |
| 78 | await connect(A, B) | |
| 79 | ||
| 80 | await author(A, '/records/c/d.dagcbor', 'd-content', '3t01') | |
| 81 | await settle() | |
| 82 | t.is(readText(B, '/records/c/d.dagcbor'), 'd-content', 'B pulled d') | |
| 83 | ||
| 84 | await removeRecord(A, '/records/c/d.dagcbor') | |
| 85 | await settle() | |
| 86 | t.is(readText(B, '/records/c/d.dagcbor'), null, 'delete propagated — B removed d') | |
| 87 | }) |
| … | … | @@ -1,103 +0,0 @@ |
| 1 | // Offline 2-device convergence proof for the reconcile driver (records-as-files merge layer). | |
| 2 | // Two SyncEngines author independently, cross-read each other's feeds (no network), and reconcile() | |
| 3 | // pulls/merges/deletes until both devices' applied state converges — with atproto rev-based conflict | |
| 4 | // resolution. This is the local-first record-merge layer for the atproto virtual device. | |
| 5 | ||
| 6 | import test from 'brittle' | |
| 7 | import { tmpStore } from './helpers.js' | |
| 8 | import { FileIndex } from '../lib/file-index.js' | |
| 9 | import { SyncEngine } from '../lib/sync-engine.js' | |
| 10 | import { reconcile } from '../lib/sync-reconciler.js' | |
| 11 | ||
| 12 | async function device (name) { | |
| 13 | const store = tmpStore('recon-' + name) | |
| 14 | const index = new FileIndex(store) | |
| 15 | await index.ready() | |
| 16 | const engine = new SyncEngine(index, store) | |
| 17 | await engine.ready() | |
| 18 | const state = new Map() // path -> hash: the "applied files" this device holds | |
| 19 | // author locally: updates applied state + logs to the feed (as registerFile does in prod) | |
| 20 | const write = async (path, hash, rev) => { state.set(path, hash); await engine.logPut(path, { contentHash: hash, size: 1, mtime: 1, rev }) } | |
| 21 | const del = async (path) => { state.delete(path); await engine.logDel(path) } | |
| 22 | // pull effect (mirrors overlay fetchContent → registerFile → logPut): adopt remote content + its rev | |
| 23 | const pull = async (path, remote) => { state.set(path, remote.hash); await engine.logPut(path, { contentHash: remote.hash, size: remote.size || 1, mtime: remote.mtime || 1, rev: remote.rev }) } | |
| 24 | const remove = async (path) => { state.delete(path); await engine.logDel(path) } | |
| 25 | return { name, store, index, engine, state, write, del, pull, remove } | |
| 26 | } | |
| 27 | ||
| 28 | // Offline cross-link: each engine reads the other's writable feed object directly (same process). | |
| 29 | function link (a, b) { | |
| 30 | a.engine._peerFeeds.set(b.engine.feedKey, b.engine.feed) | |
| 31 | b.engine._peerFeeds.set(a.engine.feedKey, a.engine.feed) | |
| 32 | } | |
| 33 | ||
| 34 | async function syncBoth (a, b, strategy = 'rev-wins') { | |
| 35 | const ra = await reconcile(a.engine, b.engine.feedKey, { strategy, pull: a.pull, remove: a.remove }) | |
| 36 | const rb = await reconcile(b.engine, a.engine.feedKey, { strategy, pull: b.pull, remove: b.remove }) | |
| 37 | return { ra, rb } | |
| 38 | } | |
| 39 | ||
| 40 | const eq = (t, a, b, msg) => t.alike([...a.state.entries()].sort(), [...b.state.entries()].sort(), msg) | |
| 41 | ||
| 42 | test('disjoint writes converge on both devices', async (t) => { | |
| 43 | const A = await device('A'); const B = await device('B'); link(A, B) | |
| 44 | await A.write('/records/app.bsky.feed.post/x', 'hx', '3t01') | |
| 45 | await B.write('/records/app.bsky.feed.post/y', 'hy', '3t02') | |
| 46 | await syncBoth(A, B) | |
| 47 | eq(t, A, B, 'states converged') | |
| 48 | t.is(A.state.get('/records/app.bsky.feed.post/x'), 'hx') | |
| 49 | t.is(A.state.get('/records/app.bsky.feed.post/y'), 'hy') | |
| 50 | await A.engine.close(); await B.engine.close() | |
| 51 | }) | |
| 52 | ||
| 53 | test('concurrent edit to the same key resolves by rev (higher TID wins) on both sides', async (t) => { | |
| 54 | const A = await device('A'); const B = await device('B'); link(A, B) | |
| 55 | await A.write('/records/c/z', 'h0', '3t00') | |
| 56 | await syncBoth(A, B) // B pulls z@h0; ancestor set on both | |
| 57 | eq(t, A, B, 'converged on h0') | |
| 58 | await A.write('/records/c/z', 'hA', '3t05') // A edits offline | |
| 59 | await B.write('/records/c/z', 'hB', '3t09') // B edits offline — LATER rev | |
| 60 | await syncBoth(A, B) | |
| 61 | t.is(A.state.get('/records/c/z'), 'hB', 'A adopted the higher-rev (B) version') | |
| 62 | t.is(B.state.get('/records/c/z'), 'hB', 'B kept its higher-rev version') | |
| 63 | eq(t, A, B, 'converged on hB') | |
| 64 | await A.engine.close(); await B.engine.close() | |
| 65 | }) | |
| 66 | ||
| 67 | test('lower-rev local edit yields to a higher-rev remote regardless of sync order', async (t) => { | |
| 68 | // run reconcile B-first to prove order-independence of the rev merge | |
| 69 | const A = await device('A'); const B = await device('B'); link(A, B) | |
| 70 | await A.write('/records/c/z', 'h0', '3t00') | |
| 71 | await reconcile(B.engine, A.engine.feedKey, { strategy: 'rev-wins', pull: B.pull, remove: B.remove }) // B pulls h0 | |
| 72 | await A.write('/records/c/z', 'hA', '3t09') // A now has the HIGHER rev | |
| 73 | await B.write('/records/c/z', 'hB', '3t04') | |
| 74 | await reconcile(B.engine, A.engine.feedKey, { strategy: 'rev-wins', pull: B.pull, remove: B.remove }) | |
| 75 | await reconcile(A.engine, B.engine.feedKey, { strategy: 'rev-wins', pull: A.pull, remove: A.remove }) | |
| 76 | t.is(A.state.get('/records/c/z'), 'hA', 'A keeps its higher-rev version') | |
| 77 | t.is(B.state.get('/records/c/z'), 'hA', 'B adopts the higher-rev version') | |
| 78 | await A.engine.close(); await B.engine.close() | |
| 79 | }) | |
| 80 | ||
| 81 | test('delete propagates to the peer', async (t) => { | |
| 82 | const A = await device('A'); const B = await device('B'); link(A, B) | |
| 83 | await A.write('/records/c/d', 'hd', '3t01') | |
| 84 | await syncBoth(A, B) // B pulls d | |
| 85 | t.is(B.state.get('/records/c/d'), 'hd') | |
| 86 | await A.del('/records/c/d') | |
| 87 | await syncBoth(A, B) // B applies delete-local | |
| 88 | t.absent(B.state.has('/records/c/d'), 'delete propagated to B') | |
| 89 | eq(t, A, B, 'converged (both without d)') | |
| 90 | await A.engine.close(); await B.engine.close() | |
| 91 | }) | |
| 92 | ||
| 93 | test('reconcile is idempotent — a converged 2nd pass pulls nothing', async (t) => { | |
| 94 | const A = await device('A'); const B = await device('B'); link(A, B) | |
| 95 | await A.write('/records/c/x', 'hx', '3t01') | |
| 96 | await B.write('/records/c/y', 'hy', '3t02') | |
| 97 | await syncBoth(A, B) | |
| 98 | const second = await syncBoth(A, B) | |
| 99 | t.is(second.ra.pulled, 0, 'A pulls nothing on the 2nd pass') | |
| 100 | t.is(second.rb.pulled, 0, 'B pulls nothing on the 2nd pass') | |
| 101 | t.is(second.ra.failed.length + second.rb.failed.length, 0, 'no failures') | |
| 102 | await A.engine.close(); await B.engine.close() | |
| 103 | }) |