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();