fix(cli): a broadcast reaches every session once, whichever copy lands first
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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<InboxRow, "id" | "body" | "recipient_pubkey" | "recipient_kind"> | null {
|
||||
return db.prepare(
|
||||
`SELECT id, body, recipient_pubkey, recipient_kind FROM inbox WHERE client_message_id = ?`,
|
||||
).get<Pick<InboxRow, "id" | "body" | "recipient_pubkey" | "recipient_kind">>(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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, unknown>, 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,
|
||||
|
||||
51
apps/cli/tests/unit/inbound-multicast.test.ts
Normal file
51
apps/cli/tests/unit/inbound-multicast.test.ts
Normal file
@@ -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 });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user