OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-overlay commits / 4978c1b

attachReplication: generic file-tree convergence over one connection
The reusable record-replication capability any shim sits on (atproto records, a hyperdrive, a synced folder). Two channels on one mux: file BYTES over the overlay content protocol (attachProtocol/fetchFile); the CHANGE LIST (pointer deltas) EXCHANGED as JSON messages — NOT corestore.replicate — so the change log never contends with the content protocol for the mux (sidesteps the documented shared-mux content-channel hazard). Received deltas feed an in-memory peer-feed, so SyncEngine.diff + reconcile run unchanged. Path-aware pull fetches bytes to the record path + registers. registerFile now passes rev through to logPut. Offline two-peer proof (bare, 3/3): disjoint records converge with real content transfer; divergent same-key records resolve by atproto rev (higher wins) on both sides; deletes propagate. Regression green: fetch, file-index, sync-engine, reconciler. Bumps 0.3.3 -> 0.3.4. Tracks hyper-social #8. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
4978c1bef108a3bcfc46489ced02d0ef1fcff1df
by plankatron <fleeky@gmail.com> · 2026-07-19T17:40:21.000Z

Changed files (5) +200 -2

index.js +1 -0
……@@ -33,6 +33,7 @@ export { ChunkScheduler } from './lib/chunk-scheduler.js'
3333export { FileIndex, CONTENT_REGISTER_PREFIX } from './lib/file-index.js'
3434export { SyncEngine, STRATEGIES } from './lib/sync-engine.js'
3535export { reconcile } from './lib/sync-reconciler.js'
36export { attachReplication } from './lib/overlay-replicator.js'
3637export {
3738 chunk,
3839 chunkStream,
lib/overlay-replicator.js +110 -0
……@@ -0,0 +1,110 @@
1/**
2 * attachReplication — keep an overlay's file tree converged with ONE peer over a connection.
3 *
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:
6 *
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.
9 *
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.
14 *
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).
17 *
18 * `fileRoot`: diskPath = join(fileRoot, overlayPath). `strategy`: reconcile conflict strategy
19 * ('rev-wins' for atproto TIDs, 'newest-wins' generic). Single peer per connection.
20 */
21
22import Protomux from 'protomux'
23import c from 'compact-encoding'
24import fs from 'bare-fs'
25import { join, dirname } from 'bare-path'
26import { reconcile } from './sync-reconciler.js'
27
28const SYNC_PROTOCOL = 'hyper-overlay/sync-exchange/v1'
29
30// In-memory feed backed by change entries received over the wire — the subset SyncEngine.diff() uses.
31function 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}
41
42export async function attachReplication (overlay, conn, { strategy = 'newest-wins', fileRoot, onConverge } = {}) {
43 if (!fileRoot) throw new Error('attachReplication: fileRoot required')
44 await overlay._ensure() // initialize the overlay stack (content protocol + sync engine)
45 const engine = overlay._sync
46 const mux = Protomux.from(conn)
47 overlay.attachProtocol(mux) // content-transfer channel (bytes)
48
49 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 const pull = async (overlayPath, remote) => {
53 const disk = diskFor(overlayPath)
54 fs.mkdirSync(dirname(disk), { recursive: true })
55 const r = await overlay.fetchFile(remote.hash, { destPath: disk })
56 if (!r) throw new Error('attachReplication: content fetch failed for ' + remote.hash)
57 await overlay.registerFile(overlayPath, disk, { contentHash: remote.hash, size: remote.size, rev: remote.rev })
58 }
59 const remove = async (overlayPath) => {
60 try { fs.unlinkSync(diskFor(overlayPath)) } catch {}
61 await engine.logDel(overlayPath)
62 }
63
64 let peerKey = null
65 const peer = memFeed()
66 let running = false; let again = false
67 const runReconcile = async () => {
68 if (!peerKey) return
69 if (running) { again = true; return } // coalesce concurrent triggers; re-run once if one arrived mid-flight
70 running = true
71 try { const s = await reconcile(engine, peerKey, { strategy, pull, remove }); if (onConverge) onConverge(s) } catch {}
72 finally { running = false; if (again) { again = false; runReconcile() } }
73 }
74
75 // The sync-exchange channel: receive the peer's delta, feed the in-memory peer-feed, reconcile.
76 const chan = mux.createChannel({ protocol: SYNC_PROTOCOL })
77 const msg = chan.addMessage({
78 encoding: c.json,
79 onmessage: async (data) => {
80 if (!data || !Array.isArray(data.entries)) return
81 if (!peerKey && data.feedKey) { peerKey = String(data.feedKey); engine._peerFeeds.set(peerKey, peer) }
82 for (const e of data.entries) peer.add(e)
83 await runReconcile()
84 }
85 })
86 chan.open()
87
88 // Announce OUR change list: the full log on connect, then only new entries on each append.
89 let sentLen = 0
90 const sendDelta = async () => {
91 await engine.feed.ready()
92 const end = engine.feed.length
93 if (end <= sentLen) return
94 const entries = []
95 for (let i = sentLen; i < end; i++) entries.push(await engine.feed.get(i))
96 sentLen = end
97 try { msg.send({ feedKey: engine.feedKey, entries }) } catch {}
98 }
99 await sendDelta()
100 const onAppend = () => { sendDelta().catch(() => {}) }
101 engine.feed.on('append', onAppend)
102
103 return {
104 reconcileNow: runReconcile,
105 close: () => {
106 try { engine.feed.off('append', onAppend) } catch {}
107 if (peerKey) { try { engine._peerFeeds.delete(peerKey) } catch {} }
108 }
109 }
110}
lib/overlay.js +1 -1
……@@ -205,7 +205,7 @@ export class HyperOverlay extends ReadyResource {
205205 if (meta.prepare !== false) await this._transfer.prepareFile(diskPath, overlayPath)
206206 // Record in the local change feed (used by the protocol's markSynced
207207 // bookkeeping; with autoSync off it is not broadcast).
208 await this._sync.logPut(overlayPath, { contentHash, size, mtime: stat.mtimeMs })
208 await this._sync.logPut(overlayPath, { contentHash, size, mtime: stat.mtimeMs, rev: meta.rev }) // rev optional (app version, e.g. atproto TID) for rev-based conflict resolution
209209 return { contentHash, size }
210210 }
211211
package.json +1 -1
……@@ -1,6 +1,6 @@
11{
22 "name": "hyper-overlay",
3 "version": "0.3.3",
3 "version": "0.3.4",
44 "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.",
55 "type": "module",
66 "main": "index.js",
test/overlay-replicator.test.js +87 -0
……@@ -0,0 +1,87 @@
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
7import test from 'brittle'
8import { Duplex } from 'streamx'
9import { tmpStore, tmpDir, fs, path } from './helpers.js'
10import { HyperOverlay } from '../lib/overlay.js'
11import { attachReplication } from '../lib/overlay-replicator.js'
12
13function 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}
20const settle = (ms = 900) => new Promise((r) => setTimeout(r, ms))
21
22async 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}
28async 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}
34async 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}
38const readText = (dev, overlayPath) => { try { return fs.readFileSync(path.join(dev.root, overlayPath.replace(/^\/+/, ''))).toString() } catch { return null } }
39
40async 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
47test('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
60test('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
75test('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})