OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-overlay commits / 0158f84

reconcile: drive SyncEngine.diff() into applied merge actions (rev-aware)
SyncEngine computes a 3-way diff but never applied it. reconcile() is the missing orchestration: for a peer, run diff(), resolve conflicts by strategy, apply pull/delete via injected effects (prod: overlay fetchContent/unlink), markSynced for idempotence. Adds a 'rev-wins' strategy that resolves by atproto rev (TID, time-ordered) — the local-first record-merge layer for the atproto virtual device. SyncEngine: carry an optional `rev` through logPut + buildChangeMap (nullable, back-compat). Bumps hyper-overlay 0.3.2 -> 0.3.3 (workspace consumers rebundle). Offline 2-device proof (test, 5/5 · 15/15, bare): disjoint writes converge; concurrent same-key edits resolve by higher rev on both sides, order-independent; deletes propagate; reconcile idempotent once converged. sync-engine 20/20 + file-index 28/28 green. Tracks hyper-social #8. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
0158f844b58f39b6833307bfc3e92692108aa088
by plankatron <fleeky@gmail.com> · 2026-07-19T16:27:32.000Z

Changed files (5) +206 -3

index.js +1 -0
……@@ -32,6 +32,7 @@ export {
3232export { ChunkScheduler } from './lib/chunk-scheduler.js'
3333export { FileIndex, CONTENT_REGISTER_PREFIX } from './lib/file-index.js'
3434export { SyncEngine, STRATEGIES } from './lib/sync-engine.js'
35export { reconcile } from './lib/sync-reconciler.js'
3536export {
3637 chunk,
3738 chunkStream,
lib/sync-engine.js +2 -1
……@@ -74,6 +74,7 @@ export class SyncEngine extends ReadyResource {
7474 hash: meta.contentHash,
7575 size: meta.size,
7676 mtime: meta.mtime,
77 rev: meta.rev ?? null, // optional app version (e.g. atproto TID) for rev-based conflict resolution; null for generic callers
7778 timestamp: Date.now()
7879 })
7980 }
……@@ -151,7 +152,7 @@ export class SyncEngine extends ReadyResource {
151152 buildChangeMap (changes) {
152153 const map = new Map()
153154 for (const c of changes) {
154 map.set(c.path, { seq: c.seq, op: c.op, hash: c.hash, size: c.size, mtime: c.mtime })
155 map.set(c.path, { seq: c.seq, op: c.op, hash: c.hash, size: c.size, mtime: c.mtime, rev: c.rev ?? null })
155156 }
156157 return map
157158 }
lib/sync-reconciler.js +98 -0
……@@ -0,0 +1,98 @@
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
23import { 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.
27function 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 */
47export 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}
package.json +2 -2
……@@ -1,7 +1,7 @@
11{
22 "name": "hyper-overlay",
3 "version": "0.3.2",
4 "description": "Content-addressed P2P file distribution over Protomux — FastCDC chunking, multi-source fetch, resumable transfers, zero blob duplication. Files live on disk; the overlay is a thin index and transfer protocol.",
3 "version": "0.3.3",
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.",
55 "type": "module",
66 "main": "index.js",
77 "imports": {
test/sync-reconciler.test.js +103 -0
……@@ -0,0 +1,103 @@
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
6import test from 'brittle'
7import { tmpStore } from './helpers.js'
8import { FileIndex } from '../lib/file-index.js'
9import { SyncEngine } from '../lib/sync-engine.js'
10import { reconcile } from '../lib/sync-reconciler.js'
11
12async 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).
29function 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
34async 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
40const eq = (t, a, b, msg) => t.alike([...a.state.entries()].sort(), [...b.state.entries()].sort(), msg)
41
42test('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
53test('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
67test('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
81test('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
93test('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})