From 7abf90081502328f4ea20ab585c338c77d50f1eb Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 20:24:15 +0100 Subject: [PATCH] fix(core): forward to-device onward to a relay-data frame's receiver relay-hub's own relay-data handling read a sender's explicit to-device field to pick the right pairing among several a connection can hold at once, then discarded it before relaying the frame onward -- the receiving connection got only from-device, never to-device. A connection fronting more than one locally-addressable device behind a single hub connection (a gateway advertising several local peers, wire-mesh#170) has no other way to tell which of its own devices a frame was actually meant for, since every relay-data frame arrives multiplexed over that one connection regardless of target. IncomingManageRequest.toDevice already promised exactly this disambiguation, read from the frame's own field, but the hub never populated it because relay-hub.ts dropped the field on the way through. Splits relay-hub.unit.test.ts's multiplexed-pairing coverage into its own file (relay-hub-multiplexing.unit.test.ts) with shared fixtures in relay-hub-test-helpers.ts, under this repo's max-lines cap once the new to-device assertions pushed the combined file over it. --- ts/packages/core/src/domain/relay-hub.ts | 4 +- .../test/relay-hub-multiplexing.unit.test.ts | 380 ++++++++++++++ .../core/test/relay-hub-test-helpers.ts | 101 ++++ ts/packages/core/test/relay-hub.unit.test.ts | 465 ++---------------- 4 files changed, 523 insertions(+), 427 deletions(-) create mode 100644 ts/packages/core/test/relay-hub-multiplexing.unit.test.ts create mode 100644 ts/packages/core/test/relay-hub-test-helpers.ts diff --git a/ts/packages/core/src/domain/relay-hub.ts b/ts/packages/core/src/domain/relay-hub.ts index 4de9956..3002e72 100644 --- a/ts/packages/core/src/domain/relay-hub.ts +++ b/ts/packages/core/src/domain/relay-hub.ts @@ -1,6 +1,6 @@ // The hub's relay role, expressed purely against core's Transport port and the generated frame schemas -- no Worker-specific or WebSocket-specific type appears here, so the same logic runs under the TCP adapter in tests or any future transport. A connection's device-id is learned from its own gossiped peer-advert (the only spec frame that carries a device-id over a plain connection; TLS-cert identity extraction is deliberately out of scope for the WebSocket-ingress first pass, noted in the README). Registry semantics: last gossip wins for a device-id, and a mapping is only removed on disconnect if it still points at the connection that registered it, so a re-announcement by a newer connection isn't clobbered by an older one leaving. // -// Multiplexing: a connection may hold more than one relay pairing at once (a peer fanning a message out to several members of a group, all of whom are only reachable through this same hub). Pairings are therefore a symmetric adjacency map keyed by *connection*, not resolved through the device registry at forward time -- device-ids here are gossip-asserted, not certificate-verified, and re-resolving through the registry per frame would let a pairing silently re-attach to whichever connection most recently claimed a device-id, a spoofing vector. Keying by connection preserves the existing behaviour that an established pairing survives its peer's device mapping moving to a fresher connection, and dies only with the connection itself. `relay-data-frame`'s `to-device`/`from-device` fields disambiguate which pairing a frame belongs to now that a connection can hold several; a connection with exactly one pairing may omit `to-device`, which every legacy peer already does, so an unaddressed frame under multiplexing routes to the most recently established pairing -- the same "last one wins" semantics the old single-pairing hub already had. +// Multiplexing: a connection may hold more than one relay pairing at once (a peer fanning a message out to several members of a group, all of whom are only reachable through this same hub). Pairings are therefore a symmetric adjacency map keyed by *connection*, not resolved through the device registry at forward time -- device-ids here are gossip-asserted, not certificate-verified, and re-resolving through the registry per frame would let a pairing silently re-attach to whichever connection most recently claimed a device-id, a spoofing vector. Keying by connection preserves the existing behaviour that an established pairing survives its peer's device mapping moving to a fresher connection, and dies only with the connection itself. `relay-data-frame`'s `to-device`/`from-device` fields disambiguate which pairing a frame belongs to now that a connection can hold several; a connection with exactly one pairing may omit `to-device`, which every legacy peer already does, so an unaddressed frame under multiplexing routes to the most recently established pairing -- the same "last one wins" semantics the old single-pairing hub already had. `to-device` is also echoed onward, unmodified, in the frame this hub delivers to the receiving connection, never just consumed for the routing lookup above: a receiving connection that itself fronts more than one locally-addressable device (a gateway advertising several local peers over one hub connection, wire-mesh#170) has no other way to tell which of its own devices a given frame was actually addressed to, since from its side every relay-data frame arrives multiplexed over the same single connection regardless of target. // // Gossip forwarding and catch-up (wire-mesh#110): every `gossip-frame` this hub receives triggers two things, beyond registering the advert locally. First, the frame is re-broadcast, unmodified, to every *other* currently-connected client (never back to the sender) -- unconditionally, on every frame, with no dedup against a previously-seen advert. This is deliberately not the same "never touches payload" blindness `relay-data` gets: the hub already parses a gossip frame's `peers` to populate its own device registry above, so forwarding it is extending existing, already-established visibility, not opening a new blind spot -- `relay-data`'s ciphertext payload stays untouched and unforwarded-to-third-parties, which is the invariant that actually matters for end-to-end confidentiality. Unconditional (no per-advert or per-recipient cache) is a deliberate choice, not an oversight: `peer-advert.snapshot-seconds`/`addresses` are meant to keep propagating on every re-gossip (a liveness heartbeat, an address change), so suppressing a "duplicate" would silently stop legitimate freshness updates from reaching other clients. Loop-safety needs no extra bookkeeping either: this hub only ever re-broadcasts to its own directly-connected clients (a star topology, one hop), and nothing here re-gossips a frame it received back onto the wire on its own initiative, so there is no path for a forwarded frame to cycle back through this hub a second time. A forward that fails (the target connection has died) is swallowed per-recipient so one dead peer never aborts delivery to the rest of the fan-out, or the sending connection's own frame processing -- that peer's own `handleConnection` loop notices the same death independently via its `receive()` stream and cleans up through the ordinary disconnect path. // @@ -192,6 +192,8 @@ export function createRelayHub(): RelayHub { type: "relay-data", payload: frame.payload, "from-device": senderDevice, + // Echoed straight through from the frame this hub just received, not merely consumed for its own routing use above: a receiving connection fronting more than one locally-addressable device (a gateway) has no other way to learn which of its own devices the sender actually meant, since it's on the far side of a single multiplexed connection from the hub's own perspective. Omitted when the sender left it unaddressed, matching the legacy single-pairing convention (module header, "may omit to-device"). + ...(toDevice !== undefined ? { "to-device": toDevice } : {}), }); return; } diff --git a/ts/packages/core/test/relay-hub-multiplexing.unit.test.ts b/ts/packages/core/test/relay-hub-multiplexing.unit.test.ts new file mode 100644 index 0000000..6f772ce --- /dev/null +++ b/ts/packages/core/test/relay-hub-multiplexing.unit.test.ts @@ -0,0 +1,380 @@ +// relay-hub's own multi-pairing coverage: a connection holding several concurrent relay pairings at once, and the to-device addressing that disambiguates which pairing (and, per wire-mesh#170's gateway-forwarding use, which of the receiver's own several locally-fronted devices) a given relay-data frame belongs to. Split out of relay-hub.unit.test.ts under this repo's max-lines cap once the receiver-side to-device echo-through (wire-mesh#170) pushed the combined file over it. + +import { describe, expect, it } from "vitest"; +import { deviceIdFromFillHex, bytesFromHex } from "./hex.js"; +import { createRelayHub } from "../src/domain/relay-hub.js"; +import { + FakeConnection, + gossipFor, + gossipForMany, + tick, +} from "./relay-hub-test-helpers.js"; + +const deviceA = deviceIdFromFillHex("11"); +const deviceB = deviceIdFromFillHex("22"); +const relayPayload = bytesFromHex("deadbeef"); + +describe("createRelayHub -- multiplexed pairings and to-device addressing", () => { + it("a second relay-connect from the same initiator ADDS a pairing rather than replacing the first -- both stay live and route independently", async () => { + const hub = createRelayHub(); + const a = new FakeConnection(); + const b = new FakeConnection(); + const c = new FakeConnection(); + const handling = [ + hub.handleConnection(a.connection), + hub.handleConnection(b.connection), + hub.handleConnection(c.connection), + ]; + + const deviceC = deviceIdFromFillHex("44"); + a.push(gossipFor(deviceA)); + b.push(gossipFor(deviceB)); + c.push(gossipFor(deviceC)); + await tick(); + + a.push({ type: "relay-connect", "target-device": deviceB }); + await tick(); + a.push({ type: "relay-connect", "target-device": deviceC }); + await tick(); + + // b and c each also received the other two connections' gossip forwarded to them (own device excluded), then their own combined catch-up, before the relay-inbound. + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceC), + gossipForMany(deviceA, deviceC), + { type: "relay-inbound", "source-device": deviceA }, + ]); + expect(c.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceB), + gossipForMany(deviceA, deviceB), + { type: "relay-inbound", "source-device": deviceA }, + ]); + + // b's relay-data (single pairing on b's own side) still reaches a -- the b<->a pairing was never torn down + b.push({ type: "relay-data", payload: relayPayload }); + await tick(); + expect(a.sent).toEqual([ + gossipFor(deviceB), + gossipFor(deviceC), + gossipForMany(deviceB, deviceC), + { type: "relay-data", payload: relayPayload, "from-device": deviceB }, + ]); + + // c's relay-data reaches a too, correctly attributed and not mixed up with b's + c.push({ type: "relay-data", payload: relayPayload }); + await tick(); + expect(a.sent).toEqual([ + gossipFor(deviceB), + gossipFor(deviceC), + gossipForMany(deviceB, deviceC), + { type: "relay-data", payload: relayPayload, "from-device": deviceB }, + { type: "relay-data", payload: relayPayload, "from-device": deviceC }, + ]); + + // a, holding two pairings, addresses each explicitly via to-device and both routes work independently + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceB, + }); + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceC, + }); + await tick(); + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceC), + gossipForMany(deviceA, deviceC), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceB, + }, + ]); + expect(c.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceB), + gossipForMany(deviceA, deviceB), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceC, + }, + ]); + + await Promise.all([a.end(), b.end(), c.end()]); + await Promise.all(handling); + }); + + it("an initiator that was already a target keeps both pairings live when it connects out", async () => { + const hub = createRelayHub(); + const x = new FakeConnection(); + const a = new FakeConnection(); + const b = new FakeConnection(); + const handling = [ + hub.handleConnection(x.connection), + hub.handleConnection(a.connection), + hub.handleConnection(b.connection), + ]; + + const deviceX = deviceIdFromFillHex("55"); + x.push(gossipFor(deviceX)); + a.push(gossipFor(deviceA)); + b.push(gossipFor(deviceB)); + await tick(); + + // x dials a: a becomes the target of x -> a + x.push({ type: "relay-connect", "target-device": deviceA }); + await tick(); + // a also received x's and b's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. + expect(a.sent).toEqual([ + gossipFor(deviceX), + gossipFor(deviceB), + gossipForMany(deviceX, deviceB), + { type: "relay-inbound", "source-device": deviceX }, + ]); + + // a now also initiates its own pipe to b -- the x <-> a pairing stays live alongside the new a <-> b one + a.push({ type: "relay-connect", "target-device": deviceB }); + await tick(); + // b also received x's and a's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. + expect(b.sent).toEqual([ + gossipFor(deviceX), + gossipFor(deviceA), + gossipForMany(deviceX, deviceA), + { type: "relay-inbound", "source-device": deviceA }, + ]); + + // x's data (x holds one pairing, no to-device needed) still reaches a + x.push({ type: "relay-data", payload: relayPayload }); + await tick(); + expect(a.sent).toEqual([ + gossipFor(deviceX), + gossipFor(deviceB), + gossipForMany(deviceX, deviceB), + { type: "relay-inbound", "source-device": deviceX }, + { type: "relay-data", payload: relayPayload, "from-device": deviceX }, + ]); + + // a, now holding two pairings, must address b explicitly + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceB, + }); + b.push({ type: "relay-data", payload: relayPayload }); + await tick(); + expect(a.sent).toEqual([ + gossipFor(deviceX), + gossipFor(deviceB), + gossipForMany(deviceX, deviceB), + { type: "relay-inbound", "source-device": deviceX }, + { type: "relay-data", payload: relayPayload, "from-device": deviceX }, + { type: "relay-data", payload: relayPayload, "from-device": deviceB }, + ]); + expect(b.sent).toEqual([ + gossipFor(deviceX), + gossipFor(deviceA), + gossipForMany(deviceX, deviceA), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceB, + }, + ]); + + await Promise.all([x.end(), a.end(), b.end()]); + await Promise.all(handling); + }); + + it("a target that was already an initiator keeps both pairings live when dialed", async () => { + const hub = createRelayHub(); + const a = new FakeConnection(); + const b = new FakeConnection(); + const y = new FakeConnection(); + const handling = [ + hub.handleConnection(a.connection), + hub.handleConnection(b.connection), + hub.handleConnection(y.connection), + ]; + + const deviceY = deviceIdFromFillHex("66"); + a.push(gossipFor(deviceA)); + b.push(gossipFor(deviceB)); + y.push(gossipFor(deviceY)); + await tick(); + + // b dials y: b becomes the initiator of b -> y + b.push({ type: "relay-connect", "target-device": deviceY }); + await tick(); + // y also received a's and b's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. + expect(y.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceB), + gossipForMany(deviceA, deviceB), + { type: "relay-inbound", "source-device": deviceB }, + ]); + + // a now dials b -- the b <-> y pairing stays live alongside the new a <-> b one + a.push({ type: "relay-connect", "target-device": deviceB }); + await tick(); + // b also received a's and y's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceY), + gossipForMany(deviceA, deviceY), + { type: "relay-inbound", "source-device": deviceA }, + ]); + + // y's data (single pairing on y's own side) still reaches b + y.push({ type: "relay-data", payload: relayPayload }); + await tick(); + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceY), + gossipForMany(deviceA, deviceY), + { type: "relay-inbound", "source-device": deviceA }, + { type: "relay-data", payload: relayPayload, "from-device": deviceY }, + ]); + + // b, now holding two pairings, must address a explicitly + a.push({ type: "relay-data", payload: relayPayload }); + b.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceA, + }); + await tick(); + expect(a.sent).toEqual([ + gossipFor(deviceB), + gossipFor(deviceY), + gossipForMany(deviceB, deviceY), + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceB, + "to-device": deviceA, + }, + ]); + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceY), + gossipForMany(deviceA, deviceY), + { type: "relay-inbound", "source-device": deviceA }, + { type: "relay-data", payload: relayPayload, "from-device": deviceY }, + { type: "relay-data", payload: relayPayload, "from-device": deviceA }, + ]); + + await Promise.all([a.end(), b.end(), y.end()]); + await Promise.all(handling); + }); + + it("one initiator fans out to three targets over the same relay, each attributed correctly in both directions", async () => { + const hub = createRelayHub(); + const a = new FakeConnection(); + const b = new FakeConnection(); + const c = new FakeConnection(); + const d = new FakeConnection(); + const handling = [ + hub.handleConnection(a.connection), + hub.handleConnection(b.connection), + hub.handleConnection(c.connection), + hub.handleConnection(d.connection), + ]; + + const deviceC = deviceIdFromFillHex("44"); + const deviceD = deviceIdFromFillHex("77"); + a.push(gossipFor(deviceA)); + b.push(gossipFor(deviceB)); + c.push(gossipFor(deviceC)); + d.push(gossipFor(deviceD)); + await tick(); + + a.push({ type: "relay-connect", "target-device": deviceB }); + a.push({ type: "relay-connect", "target-device": deviceC }); + a.push({ type: "relay-connect", "target-device": deviceD }); + await tick(); + + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceB, + }); + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceC, + }); + a.push({ + type: "relay-data", + payload: relayPayload, + "to-device": deviceD, + }); + b.push({ type: "relay-data", payload: relayPayload }); + c.push({ type: "relay-data", payload: relayPayload }); + d.push({ type: "relay-data", payload: relayPayload }); + await tick(); + + // Each target also received the other two targets' (and a's) gossip forwarded to it, own device excluded, then its own combined catch-up, before the relay-inbound. + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceC), + gossipFor(deviceD), + gossipForMany(deviceA, deviceC, deviceD), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceB, + }, + ]); + expect(c.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceB), + gossipFor(deviceD), + gossipForMany(deviceA, deviceB, deviceD), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceC, + }, + ]); + expect(d.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceB), + gossipFor(deviceC), + gossipForMany(deviceA, deviceB, deviceC), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceD, + }, + ]); + expect(a.sent).toEqual([ + gossipFor(deviceB), + gossipFor(deviceC), + gossipFor(deviceD), + gossipForMany(deviceB, deviceC, deviceD), + { type: "relay-data", payload: relayPayload, "from-device": deviceB }, + { type: "relay-data", payload: relayPayload, "from-device": deviceC }, + { type: "relay-data", payload: relayPayload, "from-device": deviceD }, + ]); + + await Promise.all([a.end(), b.end(), c.end(), d.end()]); + await Promise.all(handling); + }); +}); diff --git a/ts/packages/core/test/relay-hub-test-helpers.ts b/ts/packages/core/test/relay-hub-test-helpers.ts new file mode 100644 index 0000000..c1edb39 --- /dev/null +++ b/ts/packages/core/test/relay-hub-test-helpers.ts @@ -0,0 +1,101 @@ +// Shared fixtures for relay-hub.unit.test.ts and relay-hub-multiplexing.unit.test.ts -- split out under this repo's max-lines cap the same way relay-hub-multiplexing.unit.test.ts's own multi-pairing/to-device coverage was split from the general single-pairing hub behaviour it grew alongside. + +import type { DeviceId, Frame, PeerAdvert } from "../src/generated/protocol.js"; +import type { Connection } from "../src/ports/transport.js"; + +/** One macrotask turn, letting the hub drain frames already queued on its connections. */ +export async function tick(): Promise { + return new Promise((resolve) => { + setTimeout(resolve, 0); + }); +} + +/** An in-memory Connection driving the hub through the port contract: queued inbound frames the test pushes, and a record of everything the hub sends back. */ +export class FakeConnection { + inbound: Frame[] = []; + sent: Frame[] = []; + /** When set, every subsequent send() rejects with this error instead of recording the frame -- simulates a peer whose own connection has died from the hub's perspective, without needing a second connection class. */ + sendRejection: Error | null = null; + private closed = false; + private readonly wakeWaiters: (() => void)[] = []; + + get connection(): Readonly { + return { + send: async (frame: Frame): Promise => { + if (this.sendRejection) { + throw this.sendRejection; + } + this.sent.push(frame); + return Promise.resolve(); + }, + receive: () => this.stream(), + close: async (): Promise => { + this.closed = true; + this.wake(); + return Promise.resolve(); + }, + }; + } + + push(frame: Frame): void { + this.inbound.push(frame); + this.wake(); + } + + /** Ends the inbound stream (simulating disconnect) while leaving sent readable. */ + async end(): Promise { + this.closed = true; + this.wake(); + return Promise.resolve(); + } + + private wake(): void { + for (const wake of this.wakeWaiters.splice(0)) { + wake(); + } + } + + private stream(): AsyncIterable { + return { + [Symbol.asyncIterator]: () => ({ + next: async (): Promise> => this.nextFrame(), + }), + }; + } + + private async nextFrame(): Promise> { + return this.drain(); + } + + private async drain(): Promise> { + for (;;) { + const next = this.inbound.shift(); + if (next !== undefined) { + return { value: next, done: false }; + } + if (this.closed) { + return { value: undefined, done: true }; + } + await new Promise((resolve) => { + this.wakeWaiters.push(resolve); + }); + } + } +} + +export function peerAdvertFor(device: DeviceId): PeerAdvert { + return { + device, + addresses: ["203.0.113.5:4433"], + "snapshot-seconds": 1861833600, + }; +} + +export function gossipFor(device: DeviceId): Frame { + return { type: "gossip", peers: [peerAdvertFor(device)] }; +} + +/** A gossip-frame bundling several peer-adverts at once -- the exact shape relay-hub's own catch-up mechanism sends back to a gossiping connection, listing every other currently-known device in one frame rather than one frame per device. */ +export function gossipForMany(...devices: readonly DeviceId[]): Frame { + return { type: "gossip", peers: devices.map(peerAdvertFor) }; +} diff --git a/ts/packages/core/test/relay-hub.unit.test.ts b/ts/packages/core/test/relay-hub.unit.test.ts index 6f0b8b4..9964a6d 100644 --- a/ts/packages/core/test/relay-hub.unit.test.ts +++ b/ts/packages/core/test/relay-hub.unit.test.ts @@ -1,111 +1,20 @@ import { describe, expect, it } from "vitest"; -import type { DeviceId, Frame, PeerAdvert } from "../src/generated/protocol.js"; +import type { Frame } from "../src/generated/protocol.js"; import type { Connection } from "../src/ports/transport.js"; import { createRelayHub } from "../src/domain/relay-hub.js"; import { deviceIdFromFillHex, bytesFromHex } from "./hex.js"; +import { + FakeConnection, + gossipFor, + gossipForMany, + tick, +} from "./relay-hub-test-helpers.js"; const deviceA = deviceIdFromFillHex("11"); const deviceB = deviceIdFromFillHex("22"); const relayPayload = bytesFromHex("deadbeef"); const orphanPayload = bytesFromHex("aa"); -/** One macrotask turn, letting the hub drain frames already queued on its connections. */ -async function tick(): Promise { - return new Promise((resolve) => { - setTimeout(resolve, 0); - }); -} - -/** An in-memory Connection driving the hub through the port contract: queued inbound frames the test pushes, and a record of everything the hub sends back. */ -class FakeConnection { - inbound: Frame[] = []; - sent: Frame[] = []; - /** When set, every subsequent send() rejects with this error instead of recording the frame -- simulates a peer whose own connection has died from the hub's perspective, without needing a second connection class. */ - sendRejection: Error | null = null; - private closed = false; - private readonly wakeWaiters: (() => void)[] = []; - - get connection(): Readonly { - return { - send: async (frame: Frame): Promise => { - if (this.sendRejection) { - throw this.sendRejection; - } - this.sent.push(frame); - return Promise.resolve(); - }, - receive: () => this.stream(), - close: async (): Promise => { - this.closed = true; - this.wake(); - return Promise.resolve(); - }, - }; - } - - push(frame: Frame): void { - this.inbound.push(frame); - this.wake(); - } - - /** Ends the inbound stream (simulating disconnect) while leaving sent readable. */ - async end(): Promise { - this.closed = true; - this.wake(); - return Promise.resolve(); - } - - private wake(): void { - for (const wake of this.wakeWaiters.splice(0)) { - wake(); - } - } - - private stream(): AsyncIterable { - return { - [Symbol.asyncIterator]: () => ({ - next: async (): Promise> => this.nextFrame(), - }), - }; - } - - private async nextFrame(): Promise> { - return this.drain(); - } - - private async drain(): Promise> { - for (;;) { - const next = this.inbound.shift(); - if (next !== undefined) { - return { value: next, done: false }; - } - if (this.closed) { - return { value: undefined, done: true }; - } - await new Promise((resolve) => { - this.wakeWaiters.push(resolve); - }); - } - } -} - -function peerAdvertFor(device: DeviceId): PeerAdvert { - return { - device, - addresses: ["203.0.113.5:4433"], - "snapshot-seconds": 1861833600, - }; -} - -function gossipFor(device: DeviceId): Frame { - return { type: "gossip", peers: [peerAdvertFor(device)] }; -} - -/** A gossip-frame bundling several peer-adverts at once -- the exact shape relay-hub's own catch-up mechanism sends back to a gossiping connection, listing every other currently-known device in one frame rather than one frame per device. */ -function gossipForMany(...devices: readonly DeviceId[]): Frame { - return { type: "gossip", peers: devices.map(peerAdvertFor) }; -} - /** A Connection whose receive() stream delivers pushed frames until rejectNow(), then rejects -- the mid-stream hostile-input failure the real adapter produces for undecodable bytes. */ class RejectingAfterFramesConnection { inbound: Frame[] = []; @@ -222,6 +131,38 @@ describe("createRelayHub", () => { await Promise.all(handling); }); + it("echoes the sender's own explicit to-device onward to the receiver, not just from-device -- a receiving connection fronting more than one locally-addressable device (a gateway) needs this to disambiguate which of its own devices a frame was actually meant for, since the hub's routing use of to-device on the way in otherwise leaves no trace once the frame reaches the other side", async () => { + const hub = createRelayHub(); + const a = new FakeConnection(); + const b = new FakeConnection(); + const handling = [ + hub.handleConnection(a.connection), + hub.handleConnection(b.connection), + ]; + + a.push(gossipFor(deviceA)); + b.push(gossipFor(deviceB)); + await tick(); + a.push({ type: "relay-connect", "target-device": deviceB }); + await tick(); + a.push({ type: "relay-data", payload: relayPayload, "to-device": deviceB }); + await tick(); + + expect(b.sent).toEqual([ + gossipFor(deviceA), + gossipFor(deviceA), + { type: "relay-inbound", "source-device": deviceA }, + { + type: "relay-data", + payload: relayPayload, + "from-device": deviceA, + "to-device": deviceB, + }, + ]); + await Promise.all([a.end(), b.end()]); + await Promise.all(handling); + }); + it("ignores a relay-connect for a device not registered on this hub", async () => { const hub = createRelayHub(); const a = new FakeConnection(); @@ -354,334 +295,6 @@ describe("createRelayHub", () => { await bHandling; }); - it("a second relay-connect from the same initiator ADDS a pairing rather than replacing the first -- both stay live and route independently", async () => { - const hub = createRelayHub(); - const a = new FakeConnection(); - const b = new FakeConnection(); - const c = new FakeConnection(); - const handling = [ - hub.handleConnection(a.connection), - hub.handleConnection(b.connection), - hub.handleConnection(c.connection), - ]; - - const deviceC = deviceIdFromFillHex("44"); - a.push(gossipFor(deviceA)); - b.push(gossipFor(deviceB)); - c.push(gossipFor(deviceC)); - await tick(); - - a.push({ type: "relay-connect", "target-device": deviceB }); - await tick(); - a.push({ type: "relay-connect", "target-device": deviceC }); - await tick(); - - // b and c each also received the other two connections' gossip forwarded to them (own device excluded), then their own combined catch-up, before the relay-inbound. - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceC), - gossipForMany(deviceA, deviceC), - { type: "relay-inbound", "source-device": deviceA }, - ]); - expect(c.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceB), - gossipForMany(deviceA, deviceB), - { type: "relay-inbound", "source-device": deviceA }, - ]); - - // b's relay-data (single pairing on b's own side) still reaches a -- the b<->a pairing was never torn down - b.push({ type: "relay-data", payload: relayPayload }); - await tick(); - expect(a.sent).toEqual([ - gossipFor(deviceB), - gossipFor(deviceC), - gossipForMany(deviceB, deviceC), - { type: "relay-data", payload: relayPayload, "from-device": deviceB }, - ]); - - // c's relay-data reaches a too, correctly attributed and not mixed up with b's - c.push({ type: "relay-data", payload: relayPayload }); - await tick(); - expect(a.sent).toEqual([ - gossipFor(deviceB), - gossipFor(deviceC), - gossipForMany(deviceB, deviceC), - { type: "relay-data", payload: relayPayload, "from-device": deviceB }, - { type: "relay-data", payload: relayPayload, "from-device": deviceC }, - ]); - - // a, holding two pairings, addresses each explicitly via to-device and both routes work independently - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceB, - }); - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceC, - }); - await tick(); - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceC), - gossipForMany(deviceA, deviceC), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - expect(c.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceB), - gossipForMany(deviceA, deviceB), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - - await Promise.all([a.end(), b.end(), c.end()]); - await Promise.all(handling); - }); - - it("an initiator that was already a target keeps both pairings live when it connects out", async () => { - const hub = createRelayHub(); - const x = new FakeConnection(); - const a = new FakeConnection(); - const b = new FakeConnection(); - const handling = [ - hub.handleConnection(x.connection), - hub.handleConnection(a.connection), - hub.handleConnection(b.connection), - ]; - - const deviceX = deviceIdFromFillHex("55"); - x.push(gossipFor(deviceX)); - a.push(gossipFor(deviceA)); - b.push(gossipFor(deviceB)); - await tick(); - - // x dials a: a becomes the target of x -> a - x.push({ type: "relay-connect", "target-device": deviceA }); - await tick(); - // a also received x's and b's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. - expect(a.sent).toEqual([ - gossipFor(deviceX), - gossipFor(deviceB), - gossipForMany(deviceX, deviceB), - { type: "relay-inbound", "source-device": deviceX }, - ]); - - // a now also initiates its own pipe to b -- the x <-> a pairing stays live alongside the new a <-> b one - a.push({ type: "relay-connect", "target-device": deviceB }); - await tick(); - // b also received x's and a's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. - expect(b.sent).toEqual([ - gossipFor(deviceX), - gossipFor(deviceA), - gossipForMany(deviceX, deviceA), - { type: "relay-inbound", "source-device": deviceA }, - ]); - - // x's data (x holds one pairing, no to-device needed) still reaches a - x.push({ type: "relay-data", payload: relayPayload }); - await tick(); - expect(a.sent).toEqual([ - gossipFor(deviceX), - gossipFor(deviceB), - gossipForMany(deviceX, deviceB), - { type: "relay-inbound", "source-device": deviceX }, - { type: "relay-data", payload: relayPayload, "from-device": deviceX }, - ]); - - // a, now holding two pairings, must address b explicitly - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceB, - }); - b.push({ type: "relay-data", payload: relayPayload }); - await tick(); - expect(a.sent).toEqual([ - gossipFor(deviceX), - gossipFor(deviceB), - gossipForMany(deviceX, deviceB), - { type: "relay-inbound", "source-device": deviceX }, - { type: "relay-data", payload: relayPayload, "from-device": deviceX }, - { type: "relay-data", payload: relayPayload, "from-device": deviceB }, - ]); - expect(b.sent).toEqual([ - gossipFor(deviceX), - gossipFor(deviceA), - gossipForMany(deviceX, deviceA), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - - await Promise.all([x.end(), a.end(), b.end()]); - await Promise.all(handling); - }); - - it("a target that was already an initiator keeps both pairings live when dialed", async () => { - const hub = createRelayHub(); - const a = new FakeConnection(); - const b = new FakeConnection(); - const y = new FakeConnection(); - const handling = [ - hub.handleConnection(a.connection), - hub.handleConnection(b.connection), - hub.handleConnection(y.connection), - ]; - - const deviceY = deviceIdFromFillHex("66"); - a.push(gossipFor(deviceA)); - b.push(gossipFor(deviceB)); - y.push(gossipFor(deviceY)); - await tick(); - - // b dials y: b becomes the initiator of b -> y - b.push({ type: "relay-connect", "target-device": deviceY }); - await tick(); - // y also received a's and b's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. - expect(y.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceB), - gossipForMany(deviceA, deviceB), - { type: "relay-inbound", "source-device": deviceB }, - ]); - - // a now dials b -- the b <-> y pairing stays live alongside the new a <-> b one - a.push({ type: "relay-connect", "target-device": deviceB }); - await tick(); - // b also received a's and y's gossip forwarded to it, then its own combined catch-up, before the relay-inbound. - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceY), - gossipForMany(deviceA, deviceY), - { type: "relay-inbound", "source-device": deviceA }, - ]); - - // y's data (single pairing on y's own side) still reaches b - y.push({ type: "relay-data", payload: relayPayload }); - await tick(); - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceY), - gossipForMany(deviceA, deviceY), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceY }, - ]); - - // b, now holding two pairings, must address a explicitly - a.push({ type: "relay-data", payload: relayPayload }); - b.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceA, - }); - await tick(); - expect(a.sent).toEqual([ - gossipFor(deviceB), - gossipFor(deviceY), - gossipForMany(deviceB, deviceY), - { type: "relay-data", payload: relayPayload, "from-device": deviceB }, - ]); - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceY), - gossipForMany(deviceA, deviceY), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceY }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - - await Promise.all([a.end(), b.end(), y.end()]); - await Promise.all(handling); - }); - - it("one initiator fans out to three targets over the same relay, each attributed correctly in both directions", async () => { - const hub = createRelayHub(); - const a = new FakeConnection(); - const b = new FakeConnection(); - const c = new FakeConnection(); - const d = new FakeConnection(); - const handling = [ - hub.handleConnection(a.connection), - hub.handleConnection(b.connection), - hub.handleConnection(c.connection), - hub.handleConnection(d.connection), - ]; - - const deviceC = deviceIdFromFillHex("44"); - const deviceD = deviceIdFromFillHex("77"); - a.push(gossipFor(deviceA)); - b.push(gossipFor(deviceB)); - c.push(gossipFor(deviceC)); - d.push(gossipFor(deviceD)); - await tick(); - - a.push({ type: "relay-connect", "target-device": deviceB }); - a.push({ type: "relay-connect", "target-device": deviceC }); - a.push({ type: "relay-connect", "target-device": deviceD }); - await tick(); - - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceB, - }); - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceC, - }); - a.push({ - type: "relay-data", - payload: relayPayload, - "to-device": deviceD, - }); - b.push({ type: "relay-data", payload: relayPayload }); - c.push({ type: "relay-data", payload: relayPayload }); - d.push({ type: "relay-data", payload: relayPayload }); - await tick(); - - // Each target also received the other two targets' (and a's) gossip forwarded to it, own device excluded, then its own combined catch-up, before the relay-inbound. - expect(b.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceC), - gossipFor(deviceD), - gossipForMany(deviceA, deviceC, deviceD), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - expect(c.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceB), - gossipFor(deviceD), - gossipForMany(deviceA, deviceB, deviceD), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - expect(d.sent).toEqual([ - gossipFor(deviceA), - gossipFor(deviceB), - gossipFor(deviceC), - gossipForMany(deviceA, deviceB, deviceC), - { type: "relay-inbound", "source-device": deviceA }, - { type: "relay-data", payload: relayPayload, "from-device": deviceA }, - ]); - expect(a.sent).toEqual([ - gossipFor(deviceB), - gossipFor(deviceC), - gossipFor(deviceD), - gossipForMany(deviceB, deviceC, deviceD), - { type: "relay-data", payload: relayPayload, "from-device": deviceB }, - { type: "relay-data", payload: relayPayload, "from-device": deviceC }, - { type: "relay-data", payload: relayPayload, "from-device": deviceD }, - ]); - - await Promise.all([a.end(), b.end(), c.end(), d.end()]); - await Promise.all(handling); - }); - it("forwards a received gossip-frame, unmodified, to every other currently-connected client but not back to the sender", async () => { const hub = createRelayHub(); const a = new FakeConnection();