// Bidirectional repo creation/discovery between mirror pairs (#16).
//
// Today repo existence replicates ONE way: source → mirror. A repo created
// on the mirror is stranded there because the source never opens the mirror's
// RepoEventLog. The fix reuses the shared IdentityRegistry back-channel: the
// mirror announces its OWN origin coordinates (`__server__` = its serverKey,
// `__repoeventlog__` = its RepoEventLog key) alongside its mirror-writer keys,
// and the source's watcher boots a REVERSE RepoPuller toward it.
//
// Two proofs:
// 1. RepoPuller respects a custom `remoteName` so multiple pullers coexist
// (the source runs one per peer). Pre-fix the name was hardcoded
// 'primary' → a second puller would clobber the first's remote entry.
// 2. A source daemon, seeing a mirror-writers entry that carries the meta
// namespaces, boots a reverse RepoPuller keyed by the peer's serverKey —
// and correctly SKIPS its own serverKey and the forward origin it
// already pulls from.
import test from 'brittle'
import { mkdtempSync } 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 { RepoEventLog } from '../lib/services/repo-event-log.js'
import { RepoPuller } from '../lib/services/repo-mirror.js'
import { startPearGit } from '../index.js'
let n = 0
function tmpDir () {
return mkdtempSync(join(tmpdir(), 'pgv3-bidir-' + (n++) + '-'))
}
function tmpStorage () {
return mkdtempSync(join(tmpdir(), 'pgv3-bidir-st-' + (n++) + '-'))
}
function tmpSockPath () {
return join(mkdtempSync(join(tmpdir(), 'pgv3-bidir-sock-' + (n++) + '-')), 'svc.sock')
}
async function waitFor (predicate, timeoutMs = 8000) {
const t0 = Date.now()
while (Date.now() - t0 < timeoutMs) {
try { if (await predicate()) return true } catch {}
await new Promise((r) => setTimeout(r, 50))
}
return false
}
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 }) }
}
test('RepoPuller: custom remoteName is used for addRemote + pull (multiple pullers coexist)', async (t) => {
t.timeout(60000)
const testnet = await createTestnet(3, t.teardown)
// One shared sync server, two pullers each toward a different origin under
// a distinct remote name — this is the source running a reverse puller per
// mirror peer.
const sync = new StubSync()
async function makePuller (repoName, serverKey, remoteName) {
const store = new Corestore(join(tmpDir(), 'store'))
await store.ready()
const originLog = new RepoEventLog({ corestore: store })
await originLog.ready()
await originLog.append({ op: 0, name: repoName })
// Reader replica of the same log (opened by key against a fresh store).
const readerStore = new Corestore(join(tmpDir(), 'reader'))
await readerStore.ready()
const readerLog = new RepoEventLog({ corestore: readerStore, key: originLog.keyHex })
await readerLog.ready()
t.teardown(async () => {
try { await originLog.close() } catch {}
try { await readerLog.close() } catch {}
try { await store.close() } catch {}
try { await readerStore.close() } catch {}
})
// Copy the block across directly (no swarm needed — we only care that the
// puller streams entries and pulls under the right remote name).
const puller = new RepoPuller({
sync,
log: originLog, // use the writable log directly so entries stream locally
primaryServerKey: serverKey,
remoteName,
cursorPath: join(tmpDir(), 'cursor.json')
})
t.teardown(async () => { try { await puller.close() } catch {} })
return puller
}
const p1 = await makePuller('repo-from-peer-1', 'a'.repeat(64), 'peer-aaaaaaaaaaaa')
const p2 = await makePuller('repo-from-peer-2', 'b'.repeat(64), 'peer-bbbbbbbbbbbb')
await p1.start()
await p2.start()
const both = await waitFor(() => sync.calls.length >= 2, 10000)
t.ok(both, 'both pullers fired a pull')
// Distinct remotes registered — neither clobbered the other.
t.is(sync.remotes.get('peer-aaaaaaaaaaaa'), 'a'.repeat(64), 'peer-1 remote registered under its own name')
t.is(sync.remotes.get('peer-bbbbbbbbbbbb'), 'b'.repeat(64), 'peer-2 remote registered under its own name')
// Each repo was pulled under the CORRECT remote name.
const c1 = sync.calls.find((c) => c.repoName === 'repo-from-peer-1')
const c2 = sync.calls.find((c) => c.repoName === 'repo-from-peer-2')
t.is(c1 && c1.remote, 'peer-aaaaaaaaaaaa', 'repo-from-peer-1 pulled via peer-1 remote')
t.is(c2 && c2.remote, 'peer-bbbbbbbbbbbb', 'repo-from-peer-2 pulled via peer-2 remote')
})
test('source boots a reverse RepoPuller when a peer announces __server__/__repoeventlog__ (and skips self + forward origin)', async (t) => {
t.timeout(90000)
const testnet = await createTestnet(3, t.teardown)
const topic = b4a.alloc(32, 'pgv3-bidir-reverse')
// Source daemon WITH a sync service (required to boot a reverse puller).
const source = await startPearGit({
storage: tmpStorage(),
uds: tmpSockPath(),
http: { port: 0 },
swarm: { topic, swarm: { bootstrap: testnet.bootstrap } },
deploy: false,
sync: true,
reposDir: tmpDir()
})
t.teardown(async () => { try { await source.close() } catch {} })
t.ok(source.svc._sync, 'source has a sync service')
const ownServerKey = source.identity.publicKeyHex
// Build a real RepoEventLog whose key the "mirror" would announce. The
// source opens it by key when it boots the reverse puller.
const peerStore = new Corestore(join(tmpDir(), 'peer-log'))
await peerStore.ready()
const peerLog = new RepoEventLog({ corestore: peerStore })
await peerLog.ready()
t.teardown(async () => {
try { await peerLog.close() } catch {}
try { await peerStore.close() } catch {}
})
const peerServerKey = 'c'.repeat(64)
const peerLogKey = peerLog.keyHex
// Simulate the mirror's announcement landing in the shared IdentityRegistry
// view. The source IS the identity-autobase writer, so it can append the
// entry itself — this is exactly the record the mirror's _doAnnounce
// produces (devicePubkey = the mirror's identity, keys carry the meta ns).
await source.svc.identityRegistry.registerMirrorWriters('d'.repeat(64), {
__server__: peerServerKey,
__repoeventlog__: peerLogKey
})
// The watcher (wired to the identity base 'update' event) should boot a
// reverse puller keyed by the peer's serverKey — no restart, no re-announce.
const booted = await waitFor(
() => source.svc._reversePullers instanceof Map &&
source.svc._reversePullers.has(peerServerKey),
20000
)
t.ok(booted, 'source booted a reverse RepoPuller toward the announcing peer')
// It registered a UNIQUE per-peer remote in the sync server.
const remoteName = 'peer-' + peerServerKey.slice(0, 12)
t.ok(source.svc._sync.remotes.has(remoteName), 'reverse puller registered a per-peer sync remote')
// ── Negative guards ──────────────────────────────────────────────
// (a) An announcement carrying the source's OWN serverKey must NOT boot a
// puller (a daemon never pulls from itself).
await source.svc.identityRegistry.registerMirrorWriters('e'.repeat(64), {
__server__: ownServerKey,
__repoeventlog__: peerLogKey
})
// Give the watcher a beat to (not) act.
await new Promise((r) => setTimeout(r, 1500))
t.absent(source.svc._reversePullers.has(ownServerKey.toLowerCase()),
'no reverse puller booted toward our own serverKey')
// (b) Idempotent: a re-announce with the SAME coordinates must not add a
// second puller.
const before = source.svc._reversePullers.size
await source.svc.identityRegistry.registerMirrorWriters('d'.repeat(64), {
__server__: peerServerKey,
__repoeventlog__: peerLogKey
})
await new Promise((r) => setTimeout(r, 1500))
t.is(source.svc._reversePullers.size, before, 're-announce did not boot a duplicate puller')
})