From 567ae4874e3469d2c2f57b0d30dcdb0bd2f89841 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alejandro=20Guti=C3=A9rrez?= <35082514+alezmad@users.noreply.github.com> Date: Sat, 26 Sep 2026 21:10:52 +0100 Subject: [PATCH] fix(cli): a broadcast reaches every session once, whichever copy lands first MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A multicast reaches the daemon twice — on the member-WS (visible to all sessions) and on a session-WS (visible to one). The inbox deduped on client_message_id alone, so when the session copy landed first the message was hidden from every other session. Now the rows converge on the member-wide copy: a session copy arriving after the member copy is dropped; a member copy arriving after a session copy promotes the row and publishes it to the other sessions, excluding the one that already rendered it (exclude_session_pubkey in the SSE filter). Spec: .artifacts/specs/2026-09-26-session-identity-and-routing.md §3 Co-Authored-By: Claude Opus 5.5 --- apps/cli/src/daemon/db/inbox.ts | 15 ++++++ apps/cli/src/daemon/events.ts | 2 + apps/cli/src/daemon/inbound.ts | 37 +++++++++++++- apps/cli/tests/unit/inbound-multicast.test.ts | 51 +++++++++++++++++++ 4 files changed, 104 insertions(+), 1 deletion(-) create mode 100644 apps/cli/tests/unit/inbound-multicast.test.ts diff --git a/apps/cli/src/daemon/db/inbox.ts b/apps/cli/src/daemon/db/inbox.ts index 496e33e..ccdd879 100644 --- a/apps/cli/src/daemon/db/inbox.ts +++ b/apps/cli/src/daemon/db/inbox.ts @@ -111,6 +111,21 @@ export function insertIfNew( return after?.id === row.id ? row.id : null; } +/** 1.38.0: the row a client_message_id already landed as, if any. */ +export function findByClientMessageId( + db: SqliteDb, + clientMessageId: string, +): Pick | null { + return db.prepare( + `SELECT id, body, recipient_pubkey, recipient_kind FROM inbox WHERE client_message_id = ?`, + ).get>(clientMessageId) ?? null; +} + +/** 1.38.0: widen a session-scoped row to the whole member (multicast). */ +export function promoteToMember(db: SqliteDb, id: string, memberPubkey: string): void { + db.prepare(`UPDATE inbox SET recipient_kind = 'member', recipient_pubkey = ? WHERE id = ?`).run(memberPubkey, id); +} + export interface ListInboxParams { since?: number; // received_at >= since topic?: string; diff --git a/apps/cli/src/daemon/events.ts b/apps/cli/src/daemon/events.ts index 9f95525..c842f57 100644 --- a/apps/cli/src/daemon/events.ts +++ b/apps/cli/src/daemon/events.ts @@ -98,6 +98,8 @@ export function shouldDeliver(e: DaemonEvent, f: SseFilterOptions): boolean { return !!f.sessionPubkey && f.sessionPubkey.toLowerCase() === recipientPubkey; } if (recipientKind === "member") { + const excluded = typeof e.data.exclude_session_pubkey === "string" ? e.data.exclude_session_pubkey.toLowerCase() : null; + if (excluded && f.sessionPubkey && f.sessionPubkey.toLowerCase() === excluded) return false; return !!f.memberPubkey && f.memberPubkey.toLowerCase() === recipientPubkey; } return true; diff --git a/apps/cli/src/daemon/inbound.ts b/apps/cli/src/daemon/inbound.ts index 424ad2c..51e63ff 100644 --- a/apps/cli/src/daemon/inbound.ts +++ b/apps/cli/src/daemon/inbound.ts @@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto"; import type { SqliteDb } from "./db/sqlite.js"; -import { insertIfNew } from "./db/inbox.js"; +import { findByClientMessageId, insertIfNew, promoteToMember } from "./db/inbox.js"; import type { EventBus } from "./events.js"; import { decryptDirect } from "~/services/crypto/facade.js"; import { decodeBase64Utf8 } from "~/utils/text.js"; @@ -118,6 +118,41 @@ export async function handleBrokerPush(msg: Record, ctx: Inboun } } + // 1.38.0 (spec 2026-09-26 §3): a multicast (broadcast / group / topic) + // reaches this daemon twice — on the member-WS (visible to every session) + // and on a session-WS (visible to that session). Dedupe used to keep + // whichever landed first, so a session copy winning hid the message from + // every other session. Converge on the member-wide row instead. + const existing = ctx.recipientKind + ? findByClientMessageId(ctx.db, clientMessageId) + : null; + if (existing && existing.recipient_kind && existing.recipient_kind !== ctx.recipientKind) { + ctx.ackClientMessage?.(clientMessageId, brokerMessageId); + if (ctx.recipientKind === "member" && existing.recipient_kind === "session" && ctx.recipientPubkey) { + promoteToMember(ctx.db, existing.id, ctx.recipientPubkey); + ctx.bus.publish("message", { + id: existing.id, + mesh: ctx.meshSlug, + client_message_id: clientMessageId, + broker_message_id: brokerMessageId, + sender_pubkey: senderPubkey, + sender_member_pubkey: senderMemberPk, + sender_name: senderName, + topic, + reply_to_id: replyToId, + priority, + ...(subtype ? { subtype } : {}), + body: existing.body, + created_at: createdAt, + recipient_pubkey: ctx.recipientPubkey, + recipient_kind: "member", + // the session that already rendered it must not see it twice + ...(existing.recipient_pubkey ? { exclude_session_pubkey: existing.recipient_pubkey } : {}), + }); + } + return; + } + const id = randomUUID(); const inserted = insertIfNew(ctx.db, { id, diff --git a/apps/cli/tests/unit/inbound-multicast.test.ts b/apps/cli/tests/unit/inbound-multicast.test.ts new file mode 100644 index 0000000..f523e39 --- /dev/null +++ b/apps/cli/tests/unit/inbound-multicast.test.ts @@ -0,0 +1,51 @@ +/** Spec 2026-09-26 §3: a broadcast arriving on both the member-WS and a + * session-WS must reach every session exactly once, whatever lands first. */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { openSqlite, type SqliteDb } from "~/daemon/db/sqlite.js"; +import { migrateInbox } from "~/daemon/db/inbox.js"; +import { EventBus, shouldDeliver, type DaemonEvent } from "~/daemon/events.js"; +import { handleBrokerPush } from "~/daemon/inbound.js"; + +const MEMBER = "a".repeat(64); +const S1 = "1".repeat(64); +const S2 = "2".repeat(64); +const push = { type: "push", messageId: "b1", client_message_id: "cid-1", ciphertext: Buffer.from("hola").toString("base64") }; + +describe("inbound: multicast convergence", () => { + let dir: string; + let db: SqliteDb; + let events: DaemonEvent[]; + let bus: EventBus; + beforeEach(async () => { + dir = mkdtempSync(join(tmpdir(), "cmh-mc-")); + db = await openSqlite(join(dir, "inbox.db")); + migrateInbox(db); + events = []; + bus = new EventBus(); + bus.subscribe((e) => { if (e.kind === "message") events.push(e); }); + }); + afterEach(() => { try { db.close(); } catch { /* ignore */ } rmSync(dir, { recursive: true, force: true }); }); + + const seenBy = (session: string) => + events.filter((e) => shouldDeliver(e, { sessionPubkey: session, memberPubkey: MEMBER, meshSlug: "m" })).length; + + it("session copy first, member copy second → every session sees it once", async () => { + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: S1, recipientKind: "session" }); + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: MEMBER, recipientKind: "member" }); + expect(seenBy(S1)).toBe(1); + expect(seenBy(S2)).toBe(1); + }); + + it("member copy first → the session copy is dropped", async () => { + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: MEMBER, recipientKind: "member" }); + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: S1, recipientKind: "session" }); + expect(seenBy(S1)).toBe(1); + expect(seenBy(S2)).toBe(1); + expect(db.prepare("SELECT COUNT(*) AS n FROM inbox").get()).toEqual({ n: 1 }); + }); +});