From 55d3f5c2b84bab22e9e507417e2be87a26b1e27c Mon Sep 17 00:00:00 2001 From: seal Date: Fri, 7 Aug 2026 00:37:17 -0400 Subject: [PATCH 1/2] refactor(comms): set-based delivery-cursor seed for mandatory channels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapse the two per-member seed N+1 loops (CreateChannel born-mandatory, SetChannelPolicy newly-mandatory) into one set-based INSERT ... SELECT each. A JOIN agent_accounts replaces the per-row WHERE EXISTS agent-only guard; the channel head is a same-channel constant computed once; ON CONFLICT DO NOTHING keeps re-subscribe idempotent. Byte-identical seed semantics (agent members only, seed-to-head, no replay) — a correctness-neutral parity cleanup surfaced by the SEA-1722 T4 review. The per-row seedDeliveryCursor helper is unchanged (still used by the membership-insert sites). Adds TestSetChannelPolicyMandatoryDoesNotSeedHumanMember pinning the JOIN's agent-only filter (a human member gets no cursor). Refs SEA-1836. Co-authored-by: Matt Wilkinson --- .../store/channel_policy_pgtest_test.go | 40 +++++++++++++++ go/internal/store/channels.go | 49 ++++++------------- go/internal/store/delivery_cursors.go | 30 ++++++++++++ 3 files changed, 85 insertions(+), 34 deletions(-) diff --git a/go/internal/store/channel_policy_pgtest_test.go b/go/internal/store/channel_policy_pgtest_test.go index e6a62dbf..e87b3a46 100644 --- a/go/internal/store/channel_policy_pgtest_test.go +++ b/go/internal/store/channel_policy_pgtest_test.go @@ -197,6 +197,46 @@ func TestSetChannelPolicySeedsCursorsForNewlyMandatory(t *testing.T) { } } +// TestSetChannelPolicyMandatoryDoesNotSeedHumanMember pins the set-based seed's +// agent-only guard (the JOIN agent_accounts): flipping mandatory on a channel +// with both a human and an agent member seeds ONLY the agent — the human member +// gets no cursor row. This fails if the JOIN were dropped or loosened to seed +// every member. +func TestSetChannelPolicyMandatoryDoesNotSeedHumanMember(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + owner := mustUser(t, s, "owner") + agent := mustAgent(t, s, owner.ID, "a1") + human := mustUser(t, s, "human") + // A non-mandatory channel with one agent and one human member, neither + // subscribed — so neither has a delivery cursor row yet. + ch := mustPolicyChannel(t, s, owner.ID, "coord", ChannelPolicy{}, agent.ID, human.ID) + unsubscribeMember(t, s, ch.ID, agent.ID) + unsubscribeMember(t, s, ch.ID, human.ID) + + // Precondition: no cursor rows exist for either member on this channel. + for _, a := range []AccountID{agent.ID, human.ID} { + if _, _, ok := readCursor(t, s, a, ch.ID); ok { + t.Fatalf("precondition: member %s already has a cursor on %s", a, ch.ID) + } + } + + if _, err := s.SetChannelPolicy(ctx, owner.ID, ch.ID, ChannelPolicy{ + MandatorySubscription: true, + }); err != nil { + t.Fatalf("SetChannelPolicy: %v", err) + } + + // The agent member is seeded (a delivery target); the human member is NOT — + // the agent-only JOIN admits no human_account row. + if _, _, ok := readCursor(t, s, agent.ID, ch.ID); !ok { + t.Fatalf("agent %s has no cursor after mandatory flip — an un-seeded delivery target", agent.ID) + } + if _, _, ok := readCursor(t, s, human.ID, ch.ID); ok { + t.Fatalf("human %s got a cursor after mandatory flip — the agent-only JOIN was bypassed", human.ID) + } +} + // TestCreateChannelBornMandatorySeedsCursors pins the create-with-policy path's // D2-hazard closure: a channel created with Policy.MandatorySubscription=true // makes every member a delivery target (D1 disjunct), so CreateChannel MUST seed diff --git a/go/internal/store/channels.go b/go/internal/store/channels.go index 152acf3f..31ed0945 100644 --- a/go/internal/store/channels.go +++ b/go/internal/store/channels.go @@ -165,16 +165,16 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel // agent member's delivery cursor MUST be seeded in this same tx — an // un-seeded delivery target is the fail-DANGEROUS D2 hazard // (compass-notification-delivery/design.md:293-311). Symmetric with - // SetChannelPolicy's newly-mandatory seed. seedDeliveryCursor is - // self-guarding (agent-only) and idempotent, so seeding every member is safe - // (a human member is a no-op). A non-mandatory channel seeds nothing here — - // its members seed at subscribe time (addOrUpdateMember), the pre-substrate - // behavior. + // SetChannelPolicy's newly-mandatory seed. One set-based statement seeds + // every agent member of the channel; it is self-guarding (agent-only) and + // idempotent, so a human member is a no-op. The member INSERTs above have + // already landed in this tx's snapshot, so the statement's channel_members + // read sees exactly this channel's member set. A non-mandatory channel seeds + // nothing here — its members seed at subscribe time (addOrUpdateMember), the + // pre-substrate behavior. if c.Policy.MandatorySubscription { - for _, m := range members { - if err := seedDeliveryCursor(ctx, tx, m, ChannelID(id)); err != nil { - return Channel{}, err - } + if err := seedChannelDeliveryCursors(ctx, tx, ChannelID(id)); err != nil { + return Channel{}, err } } if err := tx.Commit(ctx); err != nil { @@ -692,32 +692,13 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID } // Newly-mandatory: every member becomes a delivery target, so seed each - // agent member's cursor in this same txn. seedDeliveryCursor is self-guarding - // (agent-only) and idempotent, so seeding every member is safe. + // agent member's cursor in this same txn — an un-seeded delivery target is + // the fail-DANGEROUS D2 hazard. One set-based statement seeds every agent + // member of the channel; it is self-guarding (agent-only) and idempotent, so + // seeding across the whole member set is safe (a human member is a no-op). if p.MandatorySubscription && !wasMandatory { - rows, err := tx.Query(ctx, - "SELECT account_id FROM channel_members WHERE channel_id = $1", string(channelID)) - if err != nil { - return Channel{}, fmt.Errorf("store: list channel members for seed: %w", err) - } - var members []AccountID - for rows.Next() { - var m string - if err := rows.Scan(&m); err != nil { - rows.Close() - return Channel{}, fmt.Errorf("store: scan member for seed: %w", err) - } - members = append(members, AccountID(m)) - } - if err := rows.Err(); err != nil { - rows.Close() - return Channel{}, fmt.Errorf("store: iterate members for seed: %w", err) - } - rows.Close() - for _, m := range members { - if err := seedDeliveryCursor(ctx, tx, m, channelID); err != nil { - return Channel{}, err - } + if err := seedChannelDeliveryCursors(ctx, tx, channelID); err != nil { + return Channel{}, err } } diff --git a/go/internal/store/delivery_cursors.go b/go/internal/store/delivery_cursors.go index 40ca5d83..0fac0f19 100644 --- a/go/internal/store/delivery_cursors.go +++ b/go/internal/store/delivery_cursors.go @@ -35,6 +35,36 @@ func seedDeliveryCursor(ctx context.Context, tx pgx.Tx, agent AccountID, channel return nil } +// seedChannelDeliveryCursorsSQL seeds EVERY agent member of the channel to the +// current channel head in one statement — MAX(seq) over the channel's messages, +// 0 if empty, a same-channel constant across all seeded rows — with NO history +// replay (design record D2). The JOIN agent_accounts is the agent-only guard +// (the set form of the per-row WHERE EXISTS in seedDeliveryCursorSQL): a human +// member has no agent_accounts row and so is a silent no-op rather than an FK +// violation. ON CONFLICT DO NOTHING keeps a re-subscribe / re-run idempotent — +// it never resets an existing cursor. $1 is the channel id. +const seedChannelDeliveryCursorsSQL = ` + INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq) + SELECT cm.account_id, $1, + COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $1), 0) + FROM channel_members cm + JOIN agent_accounts aa ON aa.account_id = cm.account_id + WHERE cm.channel_id = $1 + ON CONFLICT (agent_account_id, channel_id) DO NOTHING` + +// seedChannelDeliveryCursors is the set-based counterpart to seedDeliveryCursor: +// it seeds all agent members of the channel in a single statement (collapsing the +// per-member seed loop), riding the caller's existing transaction so a missed +// seed is a loud failure in that same commit. Self-guarding (agent-only, see +// seedChannelDeliveryCursorsSQL) and idempotent, so it is safe to call for a +// channel whose member set includes humans. +func seedChannelDeliveryCursors(ctx context.Context, tx pgx.Tx, channel ChannelID) error { + if _, err := tx.Exec(ctx, seedChannelDeliveryCursorsSQL, string(channel)); err != nil { + return fmt.Errorf("store: seed channel delivery cursors: %w", err) + } + return nil +} + // SeedDeliveryCursor seeds acked_seq to the current channel head (MAX(seq) over // the channel's messages, 0 if empty) — NO history replay. It MUST be called in // the SAME txn as the channel_members row insert (the seed rides that commit, so From 4f3ac91da23396285f87bb17830c324d25566e50 Mon Sep 17 00:00:00 2001 From: seal Date: Fri, 7 Aug 2026 01:14:26 -0400 Subject: [PATCH 2/2] test(comms): pin agent-only seed on the born-mandatory create path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a human member to TestCreateChannelBornMandatorySeedsCursors and assert it gets no delivery cursor after a born-mandatory create — symmetric with the SetChannelPolicy human-member test. Closes the review-flagged coverage gap: the set-based seed's agent-only JOIN was pinned only on the policy-flip path, not the create path (both reach the same helper, so this is coverage parity, not a fix). Refs SEA-1836. Co-authored-by: Matt Wilkinson --- go/internal/store/channel_policy_pgtest_test.go | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/go/internal/store/channel_policy_pgtest_test.go b/go/internal/store/channel_policy_pgtest_test.go index e87b3a46..db512db1 100644 --- a/go/internal/store/channel_policy_pgtest_test.go +++ b/go/internal/store/channel_policy_pgtest_test.go @@ -241,17 +241,22 @@ func TestSetChannelPolicyMandatoryDoesNotSeedHumanMember(t *testing.T) { // D2-hazard closure: a channel created with Policy.MandatorySubscription=true // makes every member a delivery target (D1 disjunct), so CreateChannel MUST seed // each agent member's cursor in the create txn — else the channel is born with -// un-seeded delivery targets. Symmetric with the SetChannelPolicy flip. +// un-seeded delivery targets — and a human member gets none (the seed is +// agent-only). Symmetric with the SetChannelPolicy flip and its human test. func TestCreateChannelBornMandatorySeedsCursors(t *testing.T) { ctx := context.Background() s := newTestStore(t) owner := mustUser(t, s, "owner") a1 := mustAgent(t, s, owner.ID, "a1") a2 := mustAgent(t, s, owner.ID, "a2") + // A human member proves the born-mandatory seed is agent-only on the create + // path too (symmetric with the SetChannelPolicy human test): the JOIN + // agent_accounts admits no human row. + human := mustUser(t, s, "human") ch := mustPolicyChannel(t, s, owner.ID, "coord", ChannelPolicy{ MandatorySubscription: true, - }, a1.ID, a2.ID) + }, a1.ID, a2.ID, human.ID) // Every agent member has a seeded cursor from birth (at head 0, an empty // channel) and is caught-up (owed nothing) — no un-seeded delivery target, @@ -273,6 +278,12 @@ func TestCreateChannelBornMandatorySeedsCursors(t *testing.T) { t.Fatalf("agent %s owed %d messages after born-mandatory seed, want 0", a, len(owed[ch.ID])) } } + + // The human member is NOT a delivery target — the agent-only JOIN admits no + // human row, so it has no cursor (symmetric with the SetChannelPolicy path). + if _, _, ok := readCursor(t, s, human.ID, ch.ID); ok { + t.Fatalf("human %s got a cursor after born-mandatory create — the agent-only JOIN was bypassed", human.ID) + } } // TestSetChannelPolicyNonOwnerOnOwnedChannelIsNotFound pins Matt's owner-only