OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-svc commits / 0692e4a

Swarm-RPC transport — Protomux RPC channel over Hyperswarm
Adds request/response dispatch over swarm connections alongside existing replication. Uses a Protomux 'hyper-svc-rpc' channel with compact-encoding JSON messages. - lib/transports/swarm-rpc.js: attachRpcServer + attachRpcClient - lib/transports/swarm.js: pass dispatch to attachRpcServer on each connection when opts.dispatch is set (default when swarm enabled) - lib/service.js: forward dispatch to SwarmTransport (opt-out via swarm.rpc = false) - lib/client.js: Client.connectSwarm({ serverKey, topicSalt }) joins Hyperswarm, attaches RPC client channel, returns Client with kind=swarm - index.js: export attachRpcServer/attachRpcClient - test: swarm-RPC round-trip via testnet (ping, params, error propagation) 21/21 tests, 42 asserts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
0692e4a71ffa74912dcb13c6f279b73826e2edea
by plankatron <fleeky@gmail.com> · 2026-04-17T01:05:12.000Z

Changed files (6) +306 -2

index.js +3 -2
……@@ -1,10 +1,11 @@
11/**
22 * hyper-svc — reusable per-app daemon library
33 *
4 * Local-only RPC (UDS + HTTP) plus Hyperswarm replication for peer mirroring.
5 * No remote RPC: peers exchange data, not commands.
4 * RPC transports: UDS, HTTP, and Hyperswarm (Protomux channel).
5 * Replication: Hyperswarm topic join + corestore/autobase/mux replicate.
66 */
77
88export { Service } from './lib/service.js'
99export { Client } from './lib/client.js'
1010export { encodeFrame, FrameDecoder } from './lib/frame.js'
11export { attachRpcServer, attachRpcClient } from './lib/transports/swarm-rpc.js'
lib/client.js +61 -0
……@@ -25,7 +25,11 @@ import http from 'http'
2525import { URL } from 'url'
2626import b4a from 'b4a'
2727import { stat } from 'fs/promises'
28import Hyperswarm from 'hyperswarm'
29import Protomux from 'protomux'
30import crypto from 'hypercore-crypto'
2831import { encodeFrame, FrameDecoder } from './frame.js'
32import { attachRpcClient } from './transports/swarm-rpc.js'
2933
3034export class Client {
3135 /**
……@@ -80,10 +84,54 @@ export class Client {
8084 return new Client({ kind: 'http', base: base.replace(/\/+$/, '') })
8185 }
8286
87 /**
88 * Connect to a remote daemon via Hyperswarm. Joins the topic derived
89 * from the server's public key, waits for the first connection, then
90 * attaches a Protomux RPC channel.
91 *
92 * @param {object} opts
93 * @param {string|Buffer} opts.serverKey - server's Noise public key (hex or Buffer)
94 * @param {string} [opts.topicSalt='pear-git-v3'] - salt for topic derivation
95 * @param {object} [opts.swarm] - Hyperswarm constructor options (e.g. testnet bootstrap)
96 * @param {number} [opts.timeout=30000] - connection timeout ms
97 * @returns {Promise<Client>}
98 */
99 static async connectSwarm (opts = {}) {
100 if (!opts.serverKey) throw new Error('connectSwarm requires opts.serverKey')
101 const serverKey = typeof opts.serverKey === 'string'
102 ? b4a.from(opts.serverKey, 'hex')
103 : opts.serverKey
104 const salt = opts.topicSalt || 'pear-git-v3'
105 const topic = crypto.hash(b4a.concat([serverKey, b4a.from(salt)]))
106 const timeout = opts.timeout || 30_000
107
108 const swarm = new Hyperswarm(opts.swarm || {})
109 const discovery = swarm.join(topic, { server: false, client: true })
110
111 const conn = await new Promise((resolve, reject) => {
112 const timer = setTimeout(() => {
113 reject(new Error('swarm-rpc: connection timeout'))
114 }, timeout)
115 swarm.on('connection', (c) => {
116 clearTimeout(timer)
117 resolve(c)
118 })
119 discovery.flushed().catch(reject)
120 })
121
122 const mux = Protomux.from(conn)
123 const rpc = attachRpcClient(mux)
124
125 return new Client({ kind: 'swarm', swarm, conn, rpc })
126 }
127
83128 constructor (opts) {
84129 this._kind = opts.kind
85130 this._sock = opts.sock || null
86131 this._base = opts.base || null
132 this._swarm = opts.swarm || null
133 this._conn = opts.conn || null
134 this._rpc = opts.rpc || null
87135 this._closed = false
88136
89137 // UDS pending request map (id → {resolve, reject})
……@@ -120,6 +168,7 @@ export class Client {
120168
121169 if (this._kind === 'uds') return this._callUds(method, params)
122170 if (this._kind === 'http') return this._callHttp(method, params)
171 if (this._kind === 'swarm') return this._rpc.call(method, params)
123172 throw new Error('unknown client kind: ' + this._kind)
124173 }
125174
……@@ -202,6 +251,18 @@ export class Client {
202251 try { this._sock.destroy() } catch {}
203252 this._sock = null
204253 }
254 if (this._rpc) {
255 try { this._rpc.close() } catch {}
256 this._rpc = null
257 }
258 if (this._conn) {
259 try { this._conn.destroy() } catch {}
260 this._conn = null
261 }
262 if (this._swarm) {
263 try { await this._swarm.destroy() } catch {}
264 this._swarm = null
265 }
205266 }
206267}
207268
lib/service.js +1 -0
……@@ -92,6 +92,7 @@ export class Service extends ReadyResource {
9292 topic: this._swarmOpt.topic,
9393 corestore: this._corestore,
9494 mounts: () => this._listMounts(),
95 dispatch: this._swarmOpt.rpc !== false ? dispatch : null,
9596 swarm: this._swarmOpt.swarm,
9697 server: this._swarmOpt.server,
9798 client: this._swarmOpt.client
lib/transports/swarm-rpc.js +161 -0
……@@ -0,0 +1,161 @@
1/**
2 * Swarm-RPC transport for hyper-svc.
3 *
4 * Adds a Protomux RPC channel on every Hyperswarm connection. Requests
5 * and responses use the same length-prefixed JSON framing as UDS (see
6 * frame.js), multiplexed over the Noise-encrypted swarm stream.
7 *
8 * This transport extends the existing SwarmTransport (replication) with
9 * request/response dispatch. A single swarm connection carries:
10 * 1. Hypercore/Autobase replication (corestore.replicate)
11 * 2. Module mux channels (attachProtocol)
12 * 3. RPC frames (this module)
13 *
14 * Wire protocol over Protomux:
15 * Channel: 'hyper-svc-rpc'
16 * Message type 0 (request): { id, method, params }
17 * Message type 1 (response): { id, result } | { id, error }
18 *
19 * Auth: the caller's Noise public key is available on the connection.
20 * For pear-git-v3, envelope auth (Phase 2a) is layered on top — the
21 * RPC payload itself is a signed envelope, verified by the dispatch
22 * handler. The transport just provides the Noise tunnel.
23 */
24
25import Protomux from 'protomux'
26import c from 'compact-encoding'
27import b4a from 'b4a'
28
29const PROTOCOL = 'hyper-svc-rpc'
30
31/**
32 * Attach an RPC channel to a Protomux instance (one per swarm connection).
33 *
34 * Server-side: receives requests, calls dispatch, sends responses.
35 *
36 * @param {Protomux} mux — the protomux for this connection
37 * @param {Function} dispatch — (method, params, ctx) => result
38 * @param {object} [opts]
39 * @param {Buffer} [opts.remotePublicKey] — peer's Noise pubkey (for ctx)
40 * @returns {{ channel, close: () => void }}
41 */
42export function attachRpcServer (mux, dispatch, opts = {}) {
43 const channel = mux.createChannel({
44 protocol: PROTOCOL,
45 onopen () {},
46 onclose () {}
47 })
48
49 if (!channel) return null // protocol already claimed on this mux
50
51 const requestType = channel.addMessage({
52 encoding: c.json,
53 onmessage (msg) {
54 handleRequest(msg, channel, responseType, dispatch, opts)
55 }
56 })
57
58 const responseType = channel.addMessage({
59 encoding: c.json,
60 onmessage () {} // server doesn't receive responses
61 })
62
63 channel.open()
64
65 return { channel, close: () => channel.close() }
66}
67
68/**
69 * Attach an RPC channel to a Protomux instance (client-side).
70 *
71 * Client-side: sends requests, receives responses.
72 *
73 * @param {Protomux} mux
74 * @returns {{ call: (method, params) => Promise<any>, close: () => void }}
75 */
76export function attachRpcClient (mux) {
77 let nextId = 1
78 const pending = new Map() // id → { resolve, reject }
79
80 const channel = mux.createChannel({
81 protocol: PROTOCOL,
82 onopen () {},
83 onclose () {
84 // Reject all pending calls on channel close.
85 for (const [id, p] of pending) {
86 p.reject(new Error('swarm-rpc channel closed'))
87 pending.delete(id)
88 }
89 }
90 })
91
92 if (!channel) throw new Error('swarm-rpc: protocol already claimed')
93
94 const requestType = channel.addMessage({
95 encoding: c.json,
96 onmessage () {} // client doesn't receive requests
97 })
98
99 const responseType = channel.addMessage({
100 encoding: c.json,
101 onmessage (msg) {
102 const p = pending.get(msg.id)
103 if (!p) return // stale or unknown id
104 pending.delete(msg.id)
105 if (msg.error) {
106 const err = new Error(msg.error)
107 err.code = msg.code || null
108 p.reject(err)
109 } else {
110 p.resolve(msg.result)
111 }
112 }
113 })
114
115 channel.open()
116
117 return {
118 call (method, params) {
119 return new Promise((resolve, reject) => {
120 const id = nextId++
121 pending.set(id, { resolve, reject })
122 requestType.send({ id, method, params })
123 })
124 },
125 close () {
126 channel.close()
127 }
128 }
129}
130
131// --- internals ---
132
133async function handleRequest (msg, channel, responseType, dispatch, opts) {
134 const { id, method, params } = msg
135 if (!id || !method) {
136 responseType.send({ id: id || 0, error: 'invalid request (missing id or method)' })
137 return
138 }
139
140 const ctx = {
141 transport: 'swarm-rpc',
142 identity: {
143 type: 'swarm',
144 remotePublicKey: opts.remotePublicKey
145 ? b4a.toString(opts.remotePublicKey, 'hex')
146 : null,
147 trust: 'network'
148 }
149 }
150
151 try {
152 const result = await dispatch(method, params || {}, ctx)
153 responseType.send({ id, result: result !== undefined ? result : null })
154 } catch (err) {
155 responseType.send({
156 id,
157 error: err.message || String(err),
158 code: err.code || null
159 })
160 }
161}
lib/transports/swarm.js +27 -0
……@@ -21,6 +21,7 @@
2121import Hyperswarm from 'hyperswarm'
2222import Protomux from 'protomux'
2323import b4a from 'b4a'
24import { attachRpcServer } from './swarm-rpc.js'
2425
2526export class SwarmTransport {
2627 /**
……@@ -28,6 +29,7 @@ export class SwarmTransport {
2829 * @param {Buffer|string} opts.topic - 32-byte topic to join (hex string or buffer)
2930 * @param {object} opts.corestore - corestore to replicate
3031 * @param {() => Iterable<{ name: string, instance: object }>} opts.mounts - returns the current mount list (called on each connection so late mounts work)
32 * @param {Function} [opts.dispatch] - RPC dispatch function. When provided, a swarm-RPC Protomux channel is attached to every connection.
3133 * @param {object} [opts.swarm] - hyperswarm options (e.g., bootstrap nodes for testnet)
3234 * @param {boolean} [opts.server=true]
3335 * @param {boolean} [opts.client=true]
……@@ -42,6 +44,7 @@ export class SwarmTransport {
4244
4345 this._corestore = opts.corestore
4446 this._getMounts = opts.mounts
47 this._dispatch = opts.dispatch || null
4548 this._swarmOpts = opts.swarm || {}
4649 this._joinAsServer = opts.server !== false
4750 this._joinAsClient = opts.client !== false
……@@ -126,6 +129,30 @@ export class SwarmTransport {
126129 }
127130 }
128131 }
132
133 // Swarm-RPC channel — attach only when dispatch is configured.
134 // Multiplexes cleanly over the same Noise stream as replication
135 // and module protocols above.
136 if (this._dispatch) {
137 if (!mux) {
138 try {
139 mux = Protomux.from(conn)
140 } catch (err) {
141 // eslint-disable-next-line no-console
142 console.error('[hyper-svc/swarm] Protomux.from failed (rpc):', err.message)
143 }
144 }
145 if (mux) {
146 try {
147 attachRpcServer(mux, this._dispatch, {
148 remotePublicKey: conn.remotePublicKey || null
149 })
150 } catch (err) {
151 // eslint-disable-next-line no-console
152 console.error('[hyper-svc/swarm] attachRpcServer failed:', err.message)
153 }
154 }
155 }
129156 }
130157
131158 async close () {
test/basic.test.js +53 -0
……@@ -429,3 +429,56 @@ function fetchText (url) {
429429 })
430430 })
431431}
432
433// ── Swarm-RPC ─────────────────────────────────────────────────
434
435test('swarm-RPC: client calls handler on server via Hyperswarm', async (t) => {
436 const crypto = (await import('hypercore-crypto')).default
437 const testnet = await createTestnet(2, t.teardown)
438
439 // Derive a shared topic the same way Client.connectSwarm does:
440 // topic = hash(serverKey + salt)
441 const serverKey = b4a.alloc(32, 'rpc-test-server-key')
442 const salt = 'pear-git-v3'
443 const topic = crypto.hash(b4a.concat([serverKey, b4a.from(salt)]))
444
445 const store = tmpStore()
446 const svc = new Service({
447 name: 'rpc-server',
448 corestore: store,
449 swarm: {
450 topic,
451 swarm: { bootstrap: testnet.bootstrap }
452 }
453 })
454 svc.handle('ping', async () => 'pong')
455 svc.handle('add', async (ctx, params) => params.a + params.b)
456 svc.handle('whoami-transport', async (ctx) => ctx.transport)
457 await svc.listen()
458
459 t.teardown(async () => {
460 try { await svc.close() } catch {}
461 try { await store.close() } catch {}
462 })
463
464 // Client connects via swarm — derives the same topic from serverKey + salt
465 const client = await Client.connectSwarm({
466 serverKey,
467 topicSalt: salt,
468 swarm: { bootstrap: testnet.bootstrap }
469 })
470 t.teardown(async () => { try { await client.close() } catch {} })
471
472 t.is(client.kind, 'swarm', 'client kind is swarm')
473 t.is(await client.call('ping'), 'pong', 'ping-pong over swarm-RPC')
474 t.is(await client.call('add', { a: 3, b: 4 }), 7, 'params round-trip')
475 t.is(await client.call('whoami-transport'), 'swarm-rpc', 'transport is swarm-rpc')
476
477 // Error propagation
478 try {
479 await client.call('no-such-method')
480 t.fail('should throw for unknown method')
481 } catch (err) {
482 t.ok(err.message.includes('unknown method'), 'error message propagated')
483 }
484})