import test from 'brittle'
import Corestore from 'corestore'
import { Room, jsonSchema } from 'hyper-rooms'
import createTestnet from '@hyperswarm/testnet'
import { mkdtempSync } from '#node:fs'
import { join } from '#node:path'
import { tmpdir } from '#node:os'
import b4a from 'b4a'
import { Service, Client, encodeFrame, FrameDecoder } from '../index.js'
// ── Helpers ────────────────────────────────────────────────────
function tmpStore () {
return new Corestore(mkdtempSync(join(tmpdir(), 'hyper-svc-store-')))
}
function tmpSockPath () {
return join(mkdtempSync(join(tmpdir(), 'hyper-svc-sock-')), 'svc.sock')
}
// A trivial Room schema for swarm replication tests.
const counterSchema = jsonSchema({
validate (op) {
if (op.type !== 'inc') throw new Error('only inc supported')
},
async apply (batch, op) {
const cur = await batch.get('counter')
const next = (cur ? cur.value.n : 0) + (op.delta || 1)
await batch.put('counter', { n: next })
}
})
class CounterRoom {
constructor (corestore, opts = {}) {
this._room = new Room(corestore, {
schema: counterSchema,
namespace: 'counter',
bootstrapKey: opts.bootstrapKey || null
})
}
async ready () { await this._room.ready() }
async close () { await this._room.close() }
get keyHex () { return this._room.keyHex }
get base () { return this._room.base }
async inc (delta = 1) {
await this._room.append({ type: 'inc', delta })
}
async value () {
await this._room.base.update()
const e = await this._room.view.get('counter')
return e ? e.value.n : 0
}
}
// ── Framing ───────────────────────────────────────────────────
test('frame round-trip', (t) => {
const original = { id: 1, method: 'list', params: { limit: 10 } }
const frame = encodeFrame(original)
let decoded
const dec = new FrameDecoder((obj) => { decoded = obj })
dec.push(frame)
t.alike(decoded, original)
})
test('frame decoder handles multiple frames in one chunk', (t) => {
const a = encodeFrame({ id: 1, method: 'one' })
const b = encodeFrame({ id: 2, method: 'two' })
const seen = []
const dec = new FrameDecoder((obj) => seen.push(obj))
dec.push(b4a.concat([a, b]))
t.is(seen.length, 2)
t.is(seen[0].id, 1)
t.is(seen[1].id, 2)
})
test('frame decoder handles a frame split across chunks', (t) => {
const frame = encodeFrame({ id: 1, method: 'split', params: { x: 'hello world' } })
const seen = []
const dec = new FrameDecoder((obj) => seen.push(obj))
// Split arbitrarily
dec.push(frame.subarray(0, 3)) // partial header
t.is(seen.length, 0)
dec.push(frame.subarray(3, 6)) // rest of header + tiny payload
t.is(seen.length, 0)
dec.push(frame.subarray(6)) // rest of payload
t.is(seen.length, 1)
t.is(seen[0].method, 'split')
})
test('frame decoder rejects malformed JSON', (t) => {
const dec = new FrameDecoder(() => {})
// Length 5, payload "{bad}" — invalid JSON
const bad = b4a.alloc(9)
bad.writeUInt32BE(5, 0)
bad.write('{bad}', 4, 5, 'utf8')
// SyntaxError is a "native" error so we use exception.all
t.exception.all(() => dec.push(bad))
})
// ── In-process dispatch ──────────────────────────────────────
test('in-process dispatch (no transports)', async (t) => {
const svc = new Service({ name: 'test' })
svc.handle('echo', async (ctx, params) => params)
svc.handle('hello', async (ctx, params) => `hi ${params.name}`)
await svc.ready()
t.teardown(async () => { try { await svc.close() } catch {} })
t.alike(await svc.dispatch('echo', { foo: 1 }), { foo: 1 })
t.is(await svc.dispatch('hello', { name: 'alice' }), 'hi alice')
})
test('dispatch unknown method throws ENOMETHOD', async (t) => {
const svc = new Service({ name: 'test' })
await svc.ready()
t.teardown(async () => { try { await svc.close() } catch {} })
try {
await svc.dispatch('nope', {})
t.fail('should have thrown')
} catch (err) {
t.is(err.code, 'ENOMETHOD')
t.ok(err.message.includes('nope'))
}
})
test('handle rejects duplicate names', async (t) => {
const svc = new Service({ name: 'test' })
svc.handle('foo', () => 1)
t.exception(() => svc.handle('foo', () => 2))
await svc.close()
})
test('listHandlers + listMounts', async (t) => {
const svc = new Service({ name: 'test' })
svc.handle('a', () => 1)
svc.handle('b', () => 2)
svc.mount('m1', { hello: 'world' })
t.alike(svc.listHandlers(), ['a', 'b'])
t.alike(svc.listMounts(), ['m1'])
t.is(svc.modules.m1.hello, 'world')
await svc.close()
})
// ── UDS transport ────────────────────────────────────────────
test('UDS round-trip', async (t) => {
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock })
svc.handle('echo', async (ctx, params) => {
return { from: ctx.transport, params }
})
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({ uds: sock })
t.is(client.kind, 'uds')
t.teardown(async () => { try { await client.close() } catch {} })
const result = await client.call('echo', { x: 42 })
t.is(result.from, 'uds')
t.alike(result.params, { x: 42 })
})
test('UDS handler error propagates', async (t) => {
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock })
svc.handle('boom', async () => { throw new Error('kaboom') })
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({ uds: sock })
t.teardown(async () => { try { await client.close() } catch {} })
try {
await client.call('boom', {})
t.fail('should have thrown')
} catch (err) {
t.ok(err.message.includes('kaboom'))
}
})
test('UDS unknown method propagates', async (t) => {
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock })
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({ uds: sock })
t.teardown(async () => { try { await client.close() } catch {} })
try {
await client.call('missing', {})
t.fail('should have thrown')
} catch (err) {
t.ok(err.message.includes('missing') || err.message.includes('unknown'))
}
})
test('UDS pipelined requests', async (t) => {
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock })
svc.handle('slow', async (ctx, params) => {
await new Promise(r => setTimeout(r, 30))
return params.n * 2
})
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({ uds: sock })
t.teardown(async () => { try { await client.close() } catch {} })
// Fire 5 calls in parallel — they should pipeline and all return correct values
const results = await Promise.all([
client.call('slow', { n: 1 }),
client.call('slow', { n: 2 }),
client.call('slow', { n: 3 }),
client.call('slow', { n: 4 }),
client.call('slow', { n: 5 })
])
t.alike(results, [2, 4, 6, 8, 10])
})
test('UDS server reaps disconnected connections (no fd/buffer leak)', async (t) => {
// Regression for the 2026-06-09 pgh-sidecar leak: bare-net defaults
// allowHalfOpen:true and never emits 'close' on its own, so every client
// disconnect used to pin a socket (+ its 64KB read buffer) in the server's
// _connections set forever — 205k leaked sockets / 14GB RSS in production.
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock })
svc.handle('echo', async (ctx, params) => params)
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const N = 30
for (let i = 0; i < N; i++) {
const client = await Client.discover({ uds: sock })
await client.call('echo', { i })
await client.close()
}
// Give the server a tick to process the disconnects.
await new Promise((r) => setTimeout(r, 200))
t.is(svc.uds._connections.size, 0, 'all disconnected connections were reaped')
})
// ── HTTP transport ───────────────────────────────────────────
test('HTTP round-trip', async (t) => {
const svc = new Service({ name: 'test', http: { port: 0 } })
svc.handle('echo', async (ctx, params) => {
return { from: ctx.transport, params }
})
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const port = svc.http.port
const client = await Client.discover({ http: `http://127.0.0.1:${port}` })
t.is(client.kind, 'http')
t.teardown(async () => { try { await client.close() } catch {} })
const result = await client.call('echo', { y: 'hi' })
t.is(result.from, 'http')
t.alike(result.params, { y: 'hi' })
})
test('HTTP unknown method returns 404', async (t) => {
const svc = new Service({ name: 'test', http: { port: 0 } })
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({ http: `http://127.0.0.1:${svc.http.port}` })
t.teardown(async () => { try { await client.close() } catch {} })
try {
await client.call('nope', {})
t.fail('should have thrown')
} catch (err) {
t.ok(err.message.toLowerCase().includes('unknown') || err.message.includes('nope'))
}
})
test('HTTP prefix strips and dispatches /rpc/* under /<prefix>/rpc/*', async (t) => {
// Regression: when a daemon is reverse-proxied OR accessed directly under
// a path prefix that the browser bakes into outgoing fetch URLs (e.g.
// pear-git-v3 served at /git/), direct hits to `<host>/git/rpc/*` used to
// hang — hyper-svc only matched '/rpc/' and the consumer's listener
// strips '/git' but skips '/rpc/' assuming hyper-svc handled it.
const svc = new Service({ name: 'test', http: { port: 0, prefix: '/git' } })
svc.handle('echo', async (ctx, params) => ({ from: ctx.transport, params }))
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const port = svc.http.port
// Prefixed path must dispatch to the RPC handler.
const prefixed = await fetchJson(`http://127.0.0.1:${port}/git/rpc/echo`, { y: 'hi' })
t.is(prefixed.status, 200)
t.is(prefixed.body.result.from, 'http')
t.alike(prefixed.body.result.params, { y: 'hi' })
// Bare path must continue to work too — proxies that strip the prefix
// before forwarding remain supported.
const bare = await fetchJson(`http://127.0.0.1:${port}/rpc/echo`, { y: 'hi' })
t.is(bare.status, 200)
t.is(bare.body.result.from, 'http')
// A URL that *starts with* the prefix string but isn't a prefix boundary
// ('/gitfoo') must NOT be treated as prefixed — that would silently
// mangle unrelated paths.
const notPrefix = await fetchJson(`http://127.0.0.1:${port}/gitfoo/rpc/echo`, {})
t.is(notPrefix.status, 404, '/gitfoo/* falls through to the 404 handler')
})
test('HTTP app routes coexist with /rpc', async (t) => {
const svc = new Service({ name: 'test', http: { port: 0 } })
svc.handle('ping', async () => 'pong')
// App registers its own SSR route
svc.http.server.on('request', (req, res) => {
if (req.url === '/hello') {
res.writeHead(200, { 'Content-Type': 'text/plain' })
res.end('hello world')
}
})
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
// RPC still works
const client = await Client.discover({ http: `http://127.0.0.1:${svc.http.port}` })
t.teardown(async () => { try { await client.close() } catch {} })
t.is(await client.call('ping', {}), 'pong')
// App route also works
const text = await fetchText(`http://127.0.0.1:${svc.http.port}/hello`)
t.is(text, 'hello world')
})
// ── Discovery ────────────────────────────────────────────────
test('discover prefers UDS when both are reachable', async (t) => {
const sock = tmpSockPath()
const svc = new Service({ name: 'test', uds: sock, http: { port: 0 } })
svc.handle('echo', async (ctx) => ctx.transport)
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({
uds: sock,
http: `http://127.0.0.1:${svc.http.port}`
})
t.is(client.kind, 'uds')
t.is(await client.call('echo', {}), 'uds')
await client.close()
})
test('discover falls back to HTTP when UDS unreachable', async (t) => {
// Daemon only listens on HTTP
const svc = new Service({ name: 'test', http: { port: 0 } })
svc.handle('echo', async (ctx) => ctx.transport)
await svc.listen()
t.teardown(async () => { try { await svc.close() } catch {} })
const client = await Client.discover({
uds: '/tmp/this-socket-does-not-exist-' + Date.now() + '.sock',
http: `http://127.0.0.1:${svc.http.port}`
})
t.is(client.kind, 'http')
t.is(await client.call('echo', {}), 'http')
await client.close()
})
test('discover throws when nothing is reachable', async (t) => {
try {
await Client.discover({})
t.fail('should have thrown')
} catch (err) {
t.ok(err.message.includes('no daemon') || err.message.includes('reachable'))
}
})
// ── Mounts ───────────────────────────────────────────────────
test('mounted modules accessible via svc.modules', async (t) => {
const svc = new Service({ name: 'test' })
const orgs = { greet: () => 'hi from orgs' }
svc.mount('orgs', orgs)
svc.handle('greet', async () => svc.modules.orgs.greet())
await svc.ready()
t.teardown(async () => { try { await svc.close() } catch {} })
t.is(await svc.dispatch('greet', {}), 'hi from orgs')
})
// ── Swarm replication ────────────────────────────────────────
test('swarm replicates a mounted Room across two services', async (t) => {
const testnet = await createTestnet(3, t.teardown)
const topic = b4a.alloc(32, 'topic-svc-test-rep')
// Service A — owns the bootstrap room
const storeA = tmpStore()
const counterA = new CounterRoom(storeA)
await counterA.ready()
await counterA.inc(1)
const svcA = new Service({
name: 'a',
corestore: storeA,
swarm: {
topic,
swarm: { bootstrap: testnet.bootstrap }
}
})
svcA.mount('counter', counterA)
await svcA.listen()
// Service B — joins with the same key
const storeB = tmpStore()
const counterB = new CounterRoom(storeB, { bootstrapKey: counterA.keyHex })
await counterB.ready()
const svcB = new Service({
name: 'b',
corestore: storeB,
swarm: {
topic,
swarm: { bootstrap: testnet.bootstrap }
}
})
svcB.mount('counter', counterB)
await svcB.listen()
t.teardown(async () => {
try { await svcA.close() } catch {}
try { await svcB.close() } catch {}
try { await counterA.close() } catch {}
try { await counterB.close() } catch {}
try { await storeA.close() } catch {}
try { await storeB.close() } catch {}
})
// Wait for B to receive A's writes (replication is async)
const ok = await waitFor(async () => (await counterB.value()) === 1, 5000)
t.ok(ok, 'B received A\'s data via swarm replication')
// Now A increments again, B should also see it
await counterA.inc(2)
const ok2 = await waitFor(async () => (await counterB.value()) === 3, 5000)
t.ok(ok2, 'B received A\'s second write')
})
// ── Helpers ───────────────────────────────────────────────────
async function waitFor (predicate, timeoutMs) {
const start = Date.now()
while (Date.now() - start < timeoutMs) {
if (await predicate()) return true
await new Promise(r => setTimeout(r, 50))
}
return false
}
function fetchText (url) {
return new Promise((resolve, reject) => {
import('#node:http').then(({ default: http }) => {
http.get(url, (res) => {
const parts = []
res.on('data', (c) => parts.push(c))
res.on('end', () => resolve(b4a.concat(parts).toString('utf8')))
res.on('error', reject)
}).on('error', reject)
})
})
}
function fetchJson (url, body) {
return new Promise((resolve, reject) => {
import('#node:http').then(({ default: http }) => {
const u = new URL(url)
const req = http.request({
method: 'POST',
hostname: u.hostname,
port: u.port,
path: u.pathname + u.search,
headers: { 'Content-Type': 'application/json' }
}, (res) => {
const parts = []
res.on('data', (c) => parts.push(c))
res.on('end', () => {
const text = b4a.concat(parts).toString('utf8')
let parsed = null
try { parsed = JSON.parse(text) } catch {}
resolve({ status: res.statusCode, body: parsed, text })
})
res.on('error', reject)
})
req.on('error', reject)
// 4 s ceiling — the bug we're guarding against is a hang, so keep
// the failure mode loud rather than letting brittle's default
// run-out-the-clock behavior swallow it.
req.setTimeout(4000, () => req.destroy(new Error('request timed out')))
req.end(JSON.stringify(body || {}))
})
})
}
// ── Swarm-RPC ─────────────────────────────────────────────────
test('swarm-RPC: client calls handler on server via Hyperswarm', async (t) => {
const crypto = (await import('hypercore-crypto')).default
const testnet = await createTestnet(2, t.teardown)
// Derive a shared topic the same way Client.connectSwarm does:
// topic = hash(serverKey + salt)
const serverKey = b4a.alloc(32, 'rpc-test-server-key')
const salt = 'pear-git-v3'
const topic = crypto.hash(b4a.concat([serverKey, b4a.from(salt)]))
const store = tmpStore()
const svc = new Service({
name: 'rpc-server',
corestore: store,
swarm: {
topic,
swarm: { bootstrap: testnet.bootstrap }
}
})
svc.handle('ping', async () => 'pong')
svc.handle('add', async (ctx, params) => params.a + params.b)
svc.handle('whoami-transport', async (ctx) => ctx.transport)
await svc.listen()
t.teardown(async () => {
try { await svc.close() } catch {}
try { await store.close() } catch {}
})
// Client connects via swarm — derives the same topic from serverKey + salt
const client = await Client.connectSwarm({
serverKey,
topicSalt: salt,
swarm: { bootstrap: testnet.bootstrap }
})
t.teardown(async () => { try { await client.close() } catch {} })
t.is(client.kind, 'swarm', 'client kind is swarm')
t.is(await client.call('ping'), 'pong', 'ping-pong over swarm-RPC')
t.is(await client.call('add', { a: 3, b: 4 }), 7, 'params round-trip')
t.is(await client.call('whoami-transport'), 'swarm-rpc', 'transport is swarm-rpc')
// Error propagation
try {
await client.call('no-such-method')
t.fail('should throw for unknown method')
} catch (err) {
t.ok(err.message.includes('unknown method'), 'error message propagated')
}
})