| … | … | @@ -1,19 +1,25 @@ |
| 1 | 1 | /** |
| 2 | | * attachReplication — keep an overlay's file tree converged with ONE peer over a connection. |
| 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). |
| 3 | 4 | * |
| 4 | | * GENERIC: it knows nothing about what the files mean, so it's reusable by any shim (atproto records, |
| 5 | | * a hyperdrive, a plain synced folder). Two things cross the wire, on two channels of the same mux: |
| 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: |
| 6 | 9 | * |
| 7 | | * 1. file BYTES → the overlay content protocol (attachProtocol + fetchFile). Proven, unchanged. |
| 8 | | * 2. the CHANGE LIST (pointer deltas: {op,path,hash,rev,…}) → a dedicated JSON channel here. |
| 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. |
| 9 | 16 | * |
| 10 | | * The change list is EXCHANGED as messages — NOT replicated via corestore.replicate — precisely so it |
| 11 | | * never contends with the content protocol for the connection's mux (the documented shared-mux hazard: |
| 12 | | * corestore's own protocol claiming the mux can leave the content channel un-opened). Each side keeps |
| 13 | | * its manifest (FileIndex) local; only the delta travels, and only the differing bytes are pulled. |
| 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. |
| 14 | 19 | * |
| 15 | | * The received entries feed an in-memory peer-feed that satisfies the subset of the Hypercore interface |
| 16 | | * SyncEngine.diff() needs, so diff + reconcile work UNCHANGED (no hypercore for the peer's feed). |
| 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. |
| 17 | 23 | * |
| 18 | 24 | * `fileRoot`: diskPath = join(fileRoot, overlayPath). `strategy`: reconcile conflict strategy |
| 19 | 25 | * ('rev-wins' for atproto TIDs, 'newest-wins' generic). Single peer per connection. |
| … | … | @@ -25,37 +31,32 @@ import fs from 'bare-fs' |
| 25 | 31 | import { join, dirname } from 'bare-path' |
| 26 | 32 | import { reconcile } from './sync-reconciler.js' |
| 27 | 33 | |
| 28 | | const SYNC_PROTOCOL = 'hyper-overlay/sync-exchange/v1' |
| 34 | const DBG = typeof process !== 'undefined' && process.env && process.env.HODEBUG ? (o) => { try { console.error('[repl] ' + JSON.stringify(o)) } catch {} } : () => {} |
| 29 | 35 | |
| 30 | | // In-memory feed backed by change entries received over the wire — the subset SyncEngine.diff() uses. |
| 31 | | function memFeed () { |
| 32 | | const entries = [] |
| 33 | | return { |
| 34 | | get length () { return entries.length }, |
| 35 | | async ready () {}, |
| 36 | | async update () {}, |
| 37 | | async get (i) { return entries[i] }, |
| 38 | | add (e) { entries.push(e) } |
| 39 | | } |
| 40 | | } |
| 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' |
| 41 | 39 | |
| 42 | 40 | export async function attachReplication (overlay, conn, { strategy = 'newest-wins', fileRoot, onConverge } = {}) { |
| 43 | 41 | if (!fileRoot) throw new Error('attachReplication: fileRoot required') |
| 44 | | await overlay._ensure() // initialize the overlay stack (content protocol + sync engine) |
| 42 | await overlay._ensure() // build the overlay stack (content protocol + sync engine) if not already |
| 45 | 43 | const engine = overlay._sync |
| 46 | 44 | const mux = Protomux.from(conn) |
| 47 | | overlay.attachProtocol(mux) // content-transfer channel (bytes) |
| 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) |
| 48 | 50 | |
| 49 | 51 | const diskFor = (p) => join(fileRoot, String(p).replace(/^\/+/, '')) |
| 50 | | // Pull effect: fetch the bytes by content-hash to the record path, then register (servable + logged |
| 51 | | // to OUR feed, so we announce that we now hold it and never re-pull it). |
| 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). |
| 52 | 54 | const pull = async (overlayPath, remote) => { |
| 53 | 55 | const disk = diskFor(overlayPath) |
| 54 | 56 | fs.mkdirSync(dirname(disk), { recursive: true }) |
| 55 | 57 | const r = await overlay.fetchFile(remote.hash, { destPath: disk }) |
| 56 | 58 | if (!r) throw new Error('attachReplication: content fetch failed for ' + remote.hash) |
| 57 | 59 | await overlay.registerFile(overlayPath, disk, { contentHash: remote.hash, size: remote.size, rev: remote.rev }) |
| 58 | | try { console.log(JSON.stringify({ msg: 'overlay-repl pull', path: overlayPath, hash: String(remote.hash).slice(0, 12) })) } catch {} |
| 59 | 60 | } |
| 60 | 61 | const remove = async (overlayPath) => { |
| 61 | 62 | try { fs.unlinkSync(diskFor(overlayPath)) } catch {} |
| … | … | @@ -63,50 +64,50 @@ export async function attachReplication (overlay, conn, { strategy = 'newest-win |
| 63 | 64 | } |
| 64 | 65 | |
| 65 | 66 | let peerKey = null |
| 66 | | const peer = memFeed() |
| 67 | 67 | let running = false; let again = false |
| 68 | 68 | const runReconcile = async () => { |
| 69 | 69 | if (!peerKey) return |
| 70 | 70 | if (running) { again = true; return } // coalesce concurrent triggers; re-run once if one arrived mid-flight |
| 71 | 71 | running = true |
| 72 | | try { const s = await reconcile(engine, peerKey, { strategy, pull, remove }); try { console.log(JSON.stringify({ msg: 'overlay-repl reconcile', peer: String(peerKey).slice(0, 12), pulled: s && s.pulled, removed: s && s.removed, pushed: s && s.pushed, skipped: s && s.skipped, conflicts: s && s.conflicts, failed: s && s.failed && s.failed.length })) } catch {}; if (onConverge) onConverge(s) } catch (e) { try { console.log(JSON.stringify({ msg: 'overlay-repl reconcile ERR', error: e && e.message })) } catch {} } |
| 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 | 73 | finally { running = false; if (again) { again = false; runReconcile() } } |
| 74 | 74 | } |
| 75 | 75 | |
| 76 | | // The sync-exchange channel: receive the peer's delta, feed the in-memory peer-feed, reconcile. |
| 77 | | const chan = mux.createChannel({ protocol: SYNC_PROTOCOL }) |
| 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 }) |
| 78 | 83 | const msg = chan.addMessage({ |
| 79 | 84 | encoding: c.json, |
| 80 | 85 | onmessage: async (data) => { |
| 81 | | if (!data || !Array.isArray(data.entries)) return |
| 82 | | if (!peerKey && data.feedKey) { peerKey = String(data.feedKey); engine._peerFeeds.set(peerKey, peer) } |
| 83 | | try { console.log(JSON.stringify({ msg: 'overlay-repl recv', peer: String(peerKey).slice(0, 12), entries: data.entries.length })) } catch {} |
| 84 | | for (const e of data.entries) peer.add(e) |
| 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 |
| 85 | 94 | await runReconcile() |
| 86 | 95 | } |
| 87 | 96 | }) |
| 88 | 97 | chan.open() |
| 89 | 98 | |
| 90 | | // Announce OUR change list: the full log on connect, then only new entries on each append. |
| 91 | | let sentLen = 0 |
| 92 | | const sendDelta = async () => { |
| 93 | | await engine.feed.ready() |
| 94 | | const end = engine.feed.length |
| 95 | | if (end <= sentLen) return |
| 96 | | const entries = [] |
| 97 | | for (let i = sentLen; i < end; i++) entries.push(await engine.feed.get(i)) |
| 98 | | sentLen = end |
| 99 | | try { msg.send({ feedKey: engine.feedKey, entries }); console.log(JSON.stringify({ msg: 'overlay-repl send', entries: entries.length, feedKey: String(engine.feedKey).slice(0, 12) })) } catch (e) { try { console.log(JSON.stringify({ msg: 'overlay-repl send ERR', error: e && e.message })) } catch {} } |
| 100 | | } |
| 101 | | await sendDelta() |
| 102 | | const onAppend = () => { sendDelta().catch(() => {}) } |
| 103 | | engine.feed.on('append', onAppend) |
| 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) |
| 104 | 105 | |
| 105 | 106 | return { |
| 106 | 107 | reconcileNow: runReconcile, |
| 107 | 108 | close: () => { |
| 108 | | try { engine.feed.off('append', onAppend) } catch {} |
| 109 | | if (peerKey) { try { engine._peerFeeds.delete(peerKey) } catch {} } |
| 109 | try { engine.feed.off('append', onLocalAppend) } catch {} |
| 110 | try { if (peerFeed && onPeerAppend) peerFeed.off('append', onPeerAppend) } catch {} |
| 110 | 111 | } |
| 111 | 112 | } |
| 112 | 113 | } |