Skip to content

fix(core): stop dropping broadcasts sent while peer connections are still dialling #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Mearman merged 2 commits into from
Sep 9, 2026
Merged

fix(core): stop dropping broadcasts sent while peer connections are still dialling #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6bde895
fix(core): keep the certificate serial a minimal DER integer
Mearman Sep 9, 2026
b75023e
fix(core): stop dropping broadcasts sent while peer connections are s…
Mearman Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@
"generate:server-json": "tsx scripts/sync-release-metadata.ts",
"publish:mcp-registry": "tsx scripts/publish-mcp-registry.ts",
"lint": "eslint .",
"test": "node --test dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js",
"test": "node --test dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js",
"test:visibility": "node --test dist/test/visibility.integration.test.js",
"test:delivery": "node dist/test/delivery-receipt.runner.js",
"test:all": "pnpm test && pnpm test:delivery",
Expand Down
2 changes: 2 additions & 0 deletions src/core/identity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,8 @@ export function generateIdentity(): PeerIdentity {
const serial = Buffer.from(serialBytes);
const firstSerialByte = serial[0];
if (firstSerialByte !== undefined) serial[0] = firstSerialByte & 0x7f;
// DER INTEGERs are minimally encoded: a leading zero byte is only legal when the following byte's high bit is set. Clearing the sign bit above can leave 0x00 here, which OpenSSL rejects as illegal padding when the certificate is loaded (tls.createServer then fails despite retries), so pin it to a minimal non-zero value.
if (serial[0] === 0) serial[0] = 1;

// Validity period: now through CERTIFICATE_VALIDITY_MS from now
const now = new Date();
Expand Down
71 changes: 70 additions & 1 deletion src/core/tcp-transport.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,12 @@ const CONNECT_TIMEOUT_MS = 1000;
// Async socket write helper (not exported)
// ---------------------------------------------------------------------------

/**
* Safety valve so a dial that never completes cannot grow its outbound queue
* unbounded; oldest entries are dropped first (#23).
*/
const MAX_PENDING_PER_PEER = 100;

function writeAsync(socket: net.Socket, data: string): Promise<void> {
return new Promise((resolve, reject) => {
if (socket.destroyed) {
Expand Down Expand Up @@ -80,12 +86,19 @@ export class TcpTransport implements MeshTransport {
// -- Coordinator client socket (connection to the coordinator) --
private coordinatorSocket: net.Socket | undefined;

// -- Coordinator introduction handshake (resolved on the peer list) --
private resolveCoordinatorHandshake: (() => void) | undefined;
private coordinatorHandshakeTimer: ReturnType<typeof setTimeout> | undefined;

// -- Peer data connections (peer ID → socket + buffer) --
private peerConnections = new Map<
string,
{ socket: net.Socket; buffer: MessageBuffer }
>();

// -- Messages queued for peers whose dial is still in flight (#23) --
private pendingOutbound = new Map<string, MeshMessage[]>();

// -- All sockets accepted by the data server (for shutdown cleanup) --
private dataServerSockets = new Set<net.Socket>();

Expand Down Expand Up @@ -196,7 +209,16 @@ export class TcpTransport implements MeshTransport {
});

clearTimeout(timer);
resolve();
// Resolve on the coordinator's peer list rather than on sending the
// introduction: MeshStore.init() then returns only after the post-join
// peer dials have started, so the first broadcasts are queued for the
// dialling peers instead of dropped (#23). A timeout keeps today's
// degraded behaviour for a coordinator that never answers.
this.resolveCoordinatorHandshake = resolve;
this.coordinatorHandshakeTimer = setTimeout(() => {
this.resolveCoordinatorHandshake = undefined;
resolve();
}, CONNECT_TIMEOUT_MS);
});

const timer = setTimeout(() => {
Expand Down Expand Up @@ -387,6 +409,10 @@ export class TcpTransport implements MeshTransport {
async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise<void> {
if (this.peerConnections.has(peer.id)) return;

// Queue broadcasts until the connection registers: messages sent in the
// dial window previously had nowhere to go and were silently dropped.
this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []);

await new Promise<void>((resolve) => {
const socket = net.createConnection(
{ port: peer.port, host: COORDINATOR_HOST },
Expand All @@ -398,6 +424,8 @@ export class TcpTransport implements MeshTransport {
const pong: MeshMessage = { method: "pong", peerId: ownPeerId };
socket.write(encode(pong));

void this.flushPending(peer.id, socket);

// Wire up ongoing message handling
socket.on("data", (data) => {
const items = buffer.append(data.toString());
Expand Down Expand Up @@ -428,6 +456,7 @@ export class TcpTransport implements MeshTransport {

socket.on("close", onDisconnect);
socket.on("error", () => {
this.pendingOutbound.delete(peer.id);
onDisconnect();
socket.destroy();
resolve();
Expand Down Expand Up @@ -513,15 +542,41 @@ export class TcpTransport implements MeshTransport {
}),
);
}
for (const queue of this.pendingOutbound.values()) {
queue.push(message);
if (queue.length > MAX_PENDING_PER_PEER) queue.shift();
}
await Promise.all(writes);
}

/** Send messages queued while the peer's connection was being dialled. */
private async flushPending(
peerId: string,
socket: net.Socket,
): Promise<void> {
const queue = this.pendingOutbound.get(peerId);
this.pendingOutbound.delete(peerId);
if (queue === undefined) return;
for (const message of queue) {
const sent = await writeAsync(socket, encode(message)).then(
() => true,
() => false,
);
if (!sent) return; // connection is dying; close/error listeners clean up
}
}

// -----------------------------------------------------------------------
// MeshTransport — Shutdown / unref
// -----------------------------------------------------------------------

shutdown(): Promise<void> {
this.shutDown = true;
if (this.coordinatorHandshakeTimer !== undefined) {
clearTimeout(this.coordinatorHandshakeTimer);
this.coordinatorHandshakeTimer = undefined;
}
this.resolveCoordinatorHandshake = undefined;

// Destroy the coordinator client socket
this.coordinatorSocket?.unref();
Expand Down Expand Up @@ -598,14 +653,27 @@ export class TcpTransport implements MeshTransport {
if (this.shutDown) return;

if (msg.method === "peer_list") {
// Fire onPeerList first: it starts the post-join dials (and their
// broadcast queues) before init() resolves.
this.events.onPeerList(msg.peers);
this.completeCoordinatorHandshake();
} else if (msg.method === "peer_joined") {
this.events.onPeerJoined(msg.peer);
} else if (msg.method === "become_coordinator") {
this.events.onBecomeCoordinator(msg.peerList);
}
}

/** Complete connectToCoordinator's handshake after the peer list arrives. */
private completeCoordinatorHandshake(): void {
if (this.coordinatorHandshakeTimer !== undefined) {
clearTimeout(this.coordinatorHandshakeTimer);
this.coordinatorHandshakeTimer = undefined;
}
this.resolveCoordinatorHandshake?.();
this.resolveCoordinatorHandshake = undefined;
}

// -----------------------------------------------------------------------
// Internal — Data message dispatch
// -----------------------------------------------------------------------
Expand Down Expand Up @@ -717,6 +785,7 @@ export class TcpTransport implements MeshTransport {
if (!this.peerConnections.has(peerId)) {
this.peerConnections.set(peerId, { socket, buffer });
}
void this.flushPending(peerId, socket);
const handle: ConnectionHandle = { id: peerId };
const info: PeerInfo = {
id: peerId,
Expand Down
71 changes: 70 additions & 1 deletion src/core/tls-transport.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,12 @@ function retryCreateTlsServer(
// Async socket write helper (not exported)
// ---------------------------------------------------------------------------

/**
* Safety valve so a dial that never completes cannot grow its outbound queue
* unbounded; oldest entries are dropped first (#23).
*/
const MAX_PENDING_PER_PEER = 100;

function writeAsync(
socket: net.Socket | tls.TLSSocket,
data: string,
Expand Down Expand Up @@ -109,6 +115,13 @@ export class TlsTransport {
// -- Coordinator client socket (TLS connection to the coordinator) --
private coordinatorSocket: tls.TLSSocket | undefined;

// -- Messages queued for peers whose dial is still in flight (#23) --
private pendingOutbound = new Map<string, MeshMessage[]>();

// -- Coordinator introduction handshake (resolved on the peer list) --
private resolveCoordinatorHandshake: (() => void) | undefined;
private coordinatorHandshakeTimer: ReturnType<typeof setTimeout> | undefined;

// -- Peer data connections (peer ID → socket + buffer) --
private peerConnections = new Map<
string,
Expand Down Expand Up @@ -249,7 +262,16 @@ export class TlsTransport {
});

clearTimeout(timer);
resolve();
// Resolve on the coordinator's peer list rather than on sending the
// introduction: MeshStore.init() then returns only after the post-join
// peer dials have started, so the first broadcasts are queued for the
// dialling peers instead of dropped (#23). A timeout keeps today's
// degraded behaviour for a coordinator that never answers.
this.resolveCoordinatorHandshake = resolve;
this.coordinatorHandshakeTimer = setTimeout(() => {
this.resolveCoordinatorHandshake = undefined;
resolve();
}, CONNECT_TIMEOUT_MS);
});

const timer = setTimeout(() => {
Expand Down Expand Up @@ -441,6 +463,10 @@ export class TlsTransport {
async connectToPeer(peer: PeerInfo, ownPeerId: string): Promise<void> {
if (this.peerConnections.has(peer.id)) return;

// Queue broadcasts until the connection registers: messages sent in the
// dial window previously had nowhere to go and were silently dropped.
this.pendingOutbound.set(peer.id, this.pendingOutbound.get(peer.id) ?? []);

await new Promise<void>((resolve) => {
const socket = tls.connect(
{ ...this.connectOptions, host: COORDINATOR_HOST, port: peer.port },
Expand All @@ -452,6 +478,8 @@ export class TlsTransport {
const pong: MeshMessage = { method: "pong", peerId: ownPeerId };
socket.write(encode(pong));

void this.flushPending(peer.id, socket);

// Wire up ongoing message handling
socket.on("data", (data) => {
const items = buffer.append(data.toString());
Expand Down Expand Up @@ -482,6 +510,7 @@ export class TlsTransport {

socket.on("close", onDisconnect);
socket.on("error", () => {
this.pendingOutbound.delete(peer.id);
onDisconnect();
socket.destroy();
resolve();
Expand Down Expand Up @@ -567,15 +596,41 @@ export class TlsTransport {
}),
);
}
for (const queue of this.pendingOutbound.values()) {
queue.push(message);
if (queue.length > MAX_PENDING_PER_PEER) queue.shift();
}
await Promise.all(writes);
}

/** Send messages queued while the peer's connection was being dialled. */
private async flushPending(
peerId: string,
socket: tls.TLSSocket,
): Promise<void> {
const queue = this.pendingOutbound.get(peerId);
this.pendingOutbound.delete(peerId);
if (queue === undefined) return;
for (const message of queue) {
const sent = await writeAsync(socket, encode(message)).then(
() => true,
() => false,
);
if (!sent) return; // connection is dying; close/error listeners clean up
}
}

// -----------------------------------------------------------------------
// MeshTransport — Shutdown / unref
// -----------------------------------------------------------------------

shutdown(): Promise<void> {
this.shutDown = true;
if (this.coordinatorHandshakeTimer !== undefined) {
clearTimeout(this.coordinatorHandshakeTimer);
this.coordinatorHandshakeTimer = undefined;
}
this.resolveCoordinatorHandshake = undefined;

// Destroy the coordinator client socket
this.coordinatorSocket?.unref();
Expand Down Expand Up @@ -647,14 +702,27 @@ export class TlsTransport {
if (this.shutDown) return;

if (msg.method === "peer_list") {
// Fire onPeerList first: it starts the post-join dials (and their
// broadcast queues) before init() resolves.
this.events.onPeerList(msg.peers);
this.completeCoordinatorHandshake();
} else if (msg.method === "peer_joined") {
this.events.onPeerJoined(msg.peer);
} else if (msg.method === "become_coordinator") {
this.events.onBecomeCoordinator(msg.peerList);
}
}

/** Complete connectToCoordinator's handshake after the peer list arrives. */
private completeCoordinatorHandshake(): void {
if (this.coordinatorHandshakeTimer !== undefined) {
clearTimeout(this.coordinatorHandshakeTimer);
this.coordinatorHandshakeTimer = undefined;
}
this.resolveCoordinatorHandshake?.();
this.resolveCoordinatorHandshake = undefined;
}

// -----------------------------------------------------------------------
// Internal — Data message dispatch
// -----------------------------------------------------------------------
Expand Down Expand Up @@ -756,6 +824,7 @@ export class TlsTransport {
if (!this.peerConnections.has(peerId)) {
this.peerConnections.set(peerId, { socket, buffer });
}
void this.flushPending(peerId, socket);
const handle: ConnectionHandle = { id: peerId };
const info: PeerInfo = {
id: peerId,
Expand Down
Loading