// Tests for RepoPuller cursor persistence + name coalescing.
//
// Bugs being fixed (issue #5 items 5 and 9):
// 1. RepoPuller opens its log stream at `{ live: true, start: 0 }`
// every boot. With ~1000 historical ref-update entries that's
// ~80 minutes of redundant `git fetch` per restart.
// 2. _drain re-queues duplicate-name entries instead of coalescing
// to the latest, so a flurry of pushes to the same repo produces
// a flurry of redundant pulls.
//
// These tests use a stub sync server (just records `pull` calls) and
// the real RepoEventLog over a real Hyperswarm testnet so the
// log-streaming path is exercised end-to-end.
import test from 'brittle'
import { mkdtempSync } from '#node:fs'
import { readFileSync, existsSync } from '#node:fs'
import { join } from '#node:path'
import { tmpdir } from '#node:os'
import b4a from 'b4a'
import Corestore from 'corestore'
import createTestnet from '@hyperswarm/testnet'
import Hyperswarm from 'hyperswarm'
import { RepoEventLog } from '../lib/services/repo-event-log.js'
import { RepoPuller } from '../lib/services/repo-mirror.js'
let n = 0
function tmpDir () {
return mkdtempSync(join(tmpdir(), 'pgv3-cursor-' + (n++) + '-'))
}
class StubSync {
constructor () {
this.calls = []
this.remotes = new Map()
}
addRemote (name, key) { this.remotes.set(name, key) }
async pull (remote, repoName) {
this.calls.push({ remote, repoName, at: Date.now() })
}
}
async function setupOriginAndMirror (testnet, t) {
const topic = b4a.alloc(32, 'pgv3-cursor-' + (n++))
// Origin (writer) — owns the RepoEventLog.
const originStore = new Corestore(join(tmpDir(), 'origin'))
await originStore.ready()
const originLog = new RepoEventLog({ corestore: originStore })
await originLog.ready()
const originSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
originSwarm.on('connection', (conn) => originStore.replicate(conn))
originSwarm.join(topic, { server: true, client: false })
await originSwarm.flush()
// Mirror (reader) — opens by key.
const mirrorStore = new Corestore(join(tmpDir(), 'mirror'))
await mirrorStore.ready()
const mirrorLog = new RepoEventLog({ corestore: mirrorStore, key: originLog.keyHex })
await mirrorLog.ready()
const mirrorSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
mirrorSwarm.on('connection', (conn) => mirrorStore.replicate(conn))
mirrorSwarm.join(topic, { server: false, client: true })
await mirrorSwarm.flush()
t.teardown(async () => {
try { await originSwarm.destroy() } catch {}
try { await mirrorSwarm.destroy() } catch {}
try { await originLog.close() } catch {}
try { await mirrorLog.close() } catch {}
try { await originStore.close() } catch {}
try { await mirrorStore.close() } catch {}
})
return { originLog, mirrorLog, mirrorStore }
}
async function waitFor (predicate, timeoutMs = 8000) {
const t0 = Date.now()
while (Date.now() - t0 < timeoutMs) {
if (await predicate()) return true
await new Promise((r) => setTimeout(r, 50))
}
return false
}
test('cursor: cold boot starts at 0, post-boot persists last processed seq', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
const { originLog, mirrorLog } = await setupOriginAndMirror(testnet, t)
// Origin appends 5 lifecycle entries for distinct repos.
for (let i = 0; i < 5; i++) {
await originLog.append({ op: 0, name: 'repo-' + i, head: 'abc' + i })
}
const cursorPath = join(tmpDir(), 'repo-puller-cursor.json')
const sync = new StubSync()
const puller = new RepoPuller({
sync,
log: mirrorLog,
primaryServerKey: 'a'.repeat(64),
cursorPath
})
t.teardown(async () => { try { await puller.close() } catch {} })
await puller.start()
const allProcessed = await waitFor(() => sync.calls.length === 5, 10000)
t.ok(allProcessed, 'all 5 entries processed on cold boot')
t.is(sync.calls.length, 5, 'exactly 5 pulls fired')
// After processing, cursor must be persisted at seq 4 (last index).
const written = await waitFor(() => existsSync(cursorPath), 4000)
t.ok(written, 'cursor file was written')
const cursor = JSON.parse(readFileSync(cursorPath, 'utf8'))
t.is(cursor[puller._primaryServerKey], 4, 'cursor records last processed seq (4)')
})
test('cursor: warm boot resumes from cursor, skips already-processed entries', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
const { originLog, mirrorLog, mirrorStore } = await setupOriginAndMirror(testnet, t)
// Phase 1: warm up the log with 3 entries, run the mirror once.
for (let i = 0; i < 3; i++) {
await originLog.append({ op: 0, name: 'r' + i })
}
const cursorPath = join(tmpDir(), 'repo-puller-cursor.json')
const sync1 = new StubSync()
const puller1 = new RepoPuller({
sync: sync1,
log: mirrorLog,
primaryServerKey: 'b'.repeat(64),
cursorPath
})
await puller1.start()
await waitFor(() => sync1.calls.length === 3, 8000)
t.is(sync1.calls.length, 3, 'phase 1: all 3 historical entries processed')
await puller1.close()
// Phase 2: a new RepoPuller pointing at the SAME log + cursor file.
// Reads cursor (=2), opens stream at 3 — must NOT replay the historical
// entries. Append 2 NEW entries; only those should fire.
await originLog.append({ op: 0, name: 'r3' })
await originLog.append({ op: 0, name: 'r4' })
// Reopen mirrorLog cleanly via the same corestore (cursor test scenario
// is "same daemon restart" — re-instantiate the log against the same
// corestore + same key).
const mirrorLog2 = new RepoEventLog({ corestore: mirrorStore, key: mirrorLog.keyHex })
await mirrorLog2.ready()
t.teardown(async () => { try { await mirrorLog2.close() } catch {} })
const sync2 = new StubSync()
const puller2 = new RepoPuller({
sync: sync2,
log: mirrorLog2,
primaryServerKey: 'b'.repeat(64),
cursorPath
})
t.teardown(async () => { try { await puller2.close() } catch {} })
await puller2.start()
const newOnes = await waitFor(() => sync2.calls.length === 2, 8000)
t.ok(newOnes, 'phase 2: only the 2 new entries processed (cursor honored)')
t.is(sync2.calls.length, 2, 'exactly 2 pulls — historical entries skipped')
const names = sync2.calls.map(c => c.repoName).sort()
t.alike(names, ['r3', 'r4'], 'pulled names are the new ones, not historical')
// Cursor advanced to 4.
const cursor = JSON.parse(readFileSync(cursorPath, 'utf8'))
t.is(cursor['b'.repeat(64)], 4, 'cursor advanced to last processed seq (4)')
})
test('coalesce: duplicate-name entries collapse to one pull (latest wins)', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
const { originLog, mirrorLog } = await setupOriginAndMirror(testnet, t)
// Append 4 entries — three for the SAME repo (sequential pushes) and
// one for another repo. With coalescing, only TWO pulls should fire
// (one for repo-a, one for repo-b), not four.
await originLog.append({ op: 0, name: 'repo-a', head: 'old1' })
await originLog.append({ op: 0, name: 'repo-a', head: 'old2' })
await originLog.append({ op: 0, name: 'repo-a', head: 'latest' })
await originLog.append({ op: 0, name: 'repo-b', head: 'b1' })
const cursorPath = join(tmpDir(), 'repo-puller-cursor.json')
// Slow sync: makes coalescing observable. With instant pull, all four
// entries process one-after-another and we never see the queue grow.
// With slow pull, the second/third repo-a entries arrive while the
// first is in-flight, so coalescing kicks in.
class SlowSync extends StubSync {
async pull (remote, repoName) {
await new Promise((r) => setTimeout(r, 200))
this.calls.push({ remote, repoName, at: Date.now() })
}
}
const sync = new SlowSync()
const completed = []
const puller = new RepoPuller({
sync,
log: mirrorLog,
primaryServerKey: 'c'.repeat(64),
cursorPath,
maxConcurrentClones: 1 // force serialization so coalescing is testable
})
puller._onProgress = ({ name, ok }) => { if (ok) completed.push(name) }
t.teardown(async () => { try { await puller.close() } catch {} })
await puller.start()
// Wait for at least one pull to fire (proves activity), THEN wait
// for steady state. Without the first wait, "drained" is true at
// t=0 because nothing has started yet.
const startedSomething = await waitFor(
() => completed.length >= 1 || sync.calls.length >= 1, 12000
)
t.ok(startedSomething, 'puller started processing entries')
const drained = await waitFor(() =>
puller.status().inflight === 0 && puller.status().queued === 0
&& puller.status().pendingRePull === 0,
12000
)
t.ok(drained, 'queue + pending re-pull drained')
const calls = sync.calls.map(c => c.repoName).sort()
// Expect at most 2 pulls (one per unique name). A flaky scheduler
// might fire repo-a twice if the duplicate landed before the
// in-flight check, but never four.
t.ok(calls.length <= 2, `coalesced: ${calls.length} pulls (≤2 unique names) — was ${JSON.stringify(calls)}`)
t.ok(calls.includes('repo-a'), 'repo-a was pulled')
t.ok(calls.includes('repo-b'), 'repo-b was pulled')
})
test('cursor: missing/corrupt cursor file falls back to start=0 without crashing', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
const { originLog, mirrorLog } = await setupOriginAndMirror(testnet, t)
await originLog.append({ op: 0, name: 'r' })
// Corrupt cursor file
const cursorPath = join(tmpDir(), 'repo-puller-cursor.json')
const fs = await import('#node:fs')
fs.writeFileSync(cursorPath, '{not valid json')
const sync = new StubSync()
const puller = new RepoPuller({
sync,
log: mirrorLog,
primaryServerKey: 'd'.repeat(64),
cursorPath
})
t.teardown(async () => { try { await puller.close() } catch {} })
let started = true
try { await puller.start() } catch { started = false }
t.ok(started, 'corrupt cursor does not crash start')
const got = await waitFor(() => sync.calls.length === 1, 8000)
t.ok(got, 'still processed the entry (fell back to start=0)')
})
// #66 — Option A decoupling: a created entry's tracker key must bind even when
// the git pull fails. This is what lets the repo-less `bootstrap` tracker reach
// an already-paired mirror via the repoEventLog it already follows (no
// re-pair). Before the reorder the bind sat AFTER `_sync.pull`, so a failing
// pull skipped it.
class FailingSync {
constructor () { this.calls = []; this.remotes = new Map() }
addRemote (name, key) { this.remotes.set(name, key) }
async pull (remote, repoName) {
this.calls.push(repoName)
throw new Error('no git repo to clone: ' + repoName)
}
}
test('bind: tracker key binds even when the git pull fails (decoupling — #66)', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
const { originLog, mirrorLog } = await setupOriginAndMirror(testnet, t)
const bound = []
const stubReg = { bindRepoBootstrapKey (name, key) { bound.push({ name, key }) } }
const keyHex = b4a.toString(b4a.alloc(32, 7), 'hex')
// A tracker-only lifecycle entry (the repo-less `bootstrap` tracker): carries
// an issueBootstrapKey, but there is no git repo to clone → pull throws.
await originLog.append({ op: 0, name: 'bootstrap', issueBootstrapKey: keyHex })
const sync = new FailingSync()
const puller = new RepoPuller({
sync,
issueRegistry: stubReg,
log: mirrorLog,
primaryServerKey: 'e'.repeat(64),
cursorPath: join(tmpDir(), 'bind-decouple-cursor.json')
})
t.teardown(async () => { try { await puller.close() } catch {} })
await puller.start()
// pull was attempted AND failed...
const pulled = await waitFor(() => sync.calls.includes('bootstrap'), 8000)
t.ok(pulled, 'git pull was attempted (and threw)')
// ...but the tracker key bound anyway (the decoupling).
const didBind = await waitFor(() => bound.length === 1, 8000)
t.ok(didBind, 'tracker key bound despite the failed pull')
t.is(bound[0] && bound[0].name, 'bootstrap', 'bound the bootstrap tracker key')
})