OverviewFilesCommitsIssuesDocsReleasesPeers

ap-bridge commits / 83c3c9b

0.10.20: a peer that asks LATE on a long-lived connection is answered (phone claim: "channel closed")
The bridge opened its registration channel when a connection appeared; a peer whose feed it holds stays connected for hours and had no channel yet, so its protomux rejected ours and ours closed - and the peer's later open hit no channel and no pair() handler and was rejected: "channel closed" on every claim (measured on fleeky's phone). attachRegistration now registers mux.pair() and recreates the responder on demand. Test reproduces CHANNEL_CLOSED red; 388/388. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
83c3c9b7dea31ea579ad75b1a8ba17d4d33c469f
by plankatron <fleeky@gmail.com> · 2026-10-14T22:47:22.000Z

Changed files (4) +77 -17

lib/register-rpc.js +39 -13
……@@ -18,6 +18,7 @@
1818// one for the other.
1919
2020import ProtomuxRPC from 'protomux-rpc'
21import Protomux from 'protomux'
2122import c from 'compact-encoding'
2223import b4a from 'b4a'
2324
……@@ -87,11 +88,39 @@ export function serveRegistration ({ swarm, log = () => {}, ...deps }) {
8788 // The swarm keypair signs observer grants (admit.js): its public key IS the bridgeKey a peer claimed against.
8889 const handlers = registerHandlers({ bridgeKeyPair: swarm.keyPair || null, ...deps })
8990
90 const onconnection = (conn) => {
91 const onconnection = (conn) => { attachRegistration(conn, handlers) }
92
93 swarm.on('connection', onconnection)
94 const topic = swarm.keyPair.publicKey
95 const disc = swarm.join(topic, { server: true, client: true })
96
97 return {
98 key: b4a.toString(topic, 'hex'),
99 ready: () => disc.flushed(),
100 close: () => { try { swarm.off('connection', onconnection) } catch {} }
101 }
102}
103
104/**
105 * Serve the registration verbs on one connection.
106 *
107 * ⛔ AND KEEP SERVING A PEER THAT ASKS LATE (2026-10-14). A peer whose feed this bridge holds stays connected for
108 * hours. Our channel opens the moment the connection appears; the peer has none yet, so ITS protomux rejects that
109 * open (_requestSession → _rejectSession) and ours closes. When the user then presses "claim", the peer's open met
110 * no channel and no pair() handler here and was rejected too — "channel closed", every time, until the connection
111 * dropped (measured on a phone). So the responder is (re)created on demand: once now, and from pair() whenever the
112 * peer opens and no live channel remains. The same idiom as atproto-room's presence channel.
113 */
114export function attachRegistration (conn, handlers) {
115 let mux
116 try { mux = Protomux.from(conn) } catch { return null }
117 let live = null
118 const open = () => {
119 if (live && !live.closed) return live
91120 let rpc
92121 try {
93122 rpc = new ProtomuxRPC(conn, { protocol: PROTOCOL, valueEncoding: c.json })
94 } catch { return }
123 } catch { return null }
95124 for (const [name, fn] of Object.entries(handlers)) {
96125 rpc.respond(name, async (params) => {
97126 try {
……@@ -103,17 +132,14 @@ export function serveRegistration ({ swarm, log = () => {}, ...deps }) {
103132 }
104133 })
105134 }
135 rpc.on('close', () => { if (live === rpc) live = null })
136 live = rpc
137 return rpc
106138 }
107
108 swarm.on('connection', onconnection)
109 const topic = swarm.keyPair.publicKey
110 const disc = swarm.join(topic, { server: true, client: true })
111
112 return {
113 key: b4a.toString(topic, 'hex'),
114 ready: () => disc.flushed(),
115 close: () => { try { swarm.off('connection', onconnection) } catch {} }
116 }
139 const first = open()
140 try { mux.pair({ protocol: PROTOCOL }, () => { open() }) } catch {}
141 try { conn.once('close', () => { try { mux.unpair({ protocol: PROTOCOL }) } catch {} }) } catch {}
142 return first
117143}
118144
119export default { serveRegistration, registerHandlers, PROTOCOL }
145export default { serveRegistration, registerHandlers, attachRegistration, PROTOCOL }
package-lock.json +2 -2
……@@ -1,12 +1,12 @@
11{
22 "name": "ap-bridge",
3 "version": "0.10.19",
3 "version": "0.10.20",
44 "lockfileVersion": 3,
55 "requires": true,
66 "packages": {
77 "": {
88 "name": "ap-bridge",
9 "version": "0.10.19",
9 "version": "0.10.20",
1010 "license": "MIT",
1111 "dependencies": {
1212 "atproto-did-peer": "file:../atproto-did-peer",
package.json +3 -2
……@@ -1,6 +1,6 @@
11{
22 "name": "ap-bridge",
3 "version": "0.10.19",
3 "version": "0.10.20",
44 "description": "ActivityPub membrane \u2014 a second projection of the self-custody atproto peer",
55 "type": "module",
66 "main": "index.js",
……@@ -38,7 +38,8 @@
3838 "devDependencies": {
3939 "anproto": "file:../anproto",
4040 "brittle": "^3.0.0",
41 "keet-identity-key": "3.2.0"
41 "keet-identity-key": "3.2.0",
42 "@hyperswarm/secret-stream": "^6.9.1"
4243 },
4344 "license": "MIT",
4445 "pgh": {
test/register-rpc.test.js +33 -0
……@@ -158,3 +158,36 @@ test('no observer configured: the reply is unchanged (no grant field)', async (t
158158 t.is(r.status, 200)
159159 t.absent(r.body.observer)
160160})
161
162// ⛔ A PEER THAT ASKS LATE ON A LONG-LIVED CONNECTION MUST STILL BE ANSWERED. 2026-10-14: a phone holding fleeky's
163// feed stays connected to the bridge for hours (the bridge holds its feeds). The bridge opens its registration
164// channel the moment the connection appears; the phone has none yet, so ITS protomux rejects that open
165// (_requestSession → _rejectSession) and the bridge's channel closes. When the user later presses "claim", the
166// phone's open meets no channel and no pair() handler and is rejected too: "channel closed", every time, until the
167// connection drops. A fresh connection that asks at once (every earlier registration) never saw it.
168test('★ a client that opens the registration channel LATE, on an already-muxed connection, is answered', async (t) => {
169 const SecretStream = (await import('@hyperswarm/secret-stream')).default
170 const Protomux = (await import('protomux')).default
171 const { attachRegistration, PROTOCOL } = await import('../lib/register-rpc.js')
172 const ProtomuxRPC = (await import('protomux-rpc')).default
173 const c = (await import('compact-encoding')).default
174
175 const bridge = new SecretStream(true)
176 const peer = new SecretStream(false)
177 bridge.rawStream.pipe(peer.rawStream).pipe(bridge.rawStream)
178 t.teardown(() => { try { bridge.destroy() } catch {} try { peer.destroy() } catch {} })
179
180 // As in production: the peer's stream already carries a mux (feed replication), so the bridge's early open is
181 // processed — and rejected — on arrival.
182 Protomux.from(peer)
183 attachRegistration(bridge, registerHandlers(deps()))
184 await new Promise((resolve) => setTimeout(resolve, 300))
185
186 const rpc = new ProtomuxRPC(peer, { protocol: PROTOCOL, valueEncoding: c.json })
187 const answer = await Promise.race([
188 rpc.request('policy', {}).then((r) => ({ ok: r }), (e) => ({ err: e.message })),
189 new Promise((resolve) => setTimeout(() => resolve({ err: 'timeout' }), 3000))
190 ])
191 t.ok(answer.ok, 'answered (was: ' + (answer.err || 'ok') + ')')
192 t.is(answer.ok && answer.ok.policy, 'open')
193})