From 10eda697b838b1d4f61679884407dd6e5e749057 Mon Sep 17 00:00:00 2001 From: ATCharpentier Date: Tue, 14 Jul 2026 19:52:18 -0400 Subject: [PATCH] fix(server): close two subscription-hardening follow-ups from PR #22 review Two non-blocking items Odin flagged while reviewing the concurrency work: 1. A transient read error in SubscribeAuthorizedStable no longer tears down the channel's live subscriptions. RemoveChannel on an indeterminate read dropped legit subscribers a concurrent connect had installed, stranding already- connected members until reconnect. It now returns the error and leaves the set intact; the caller broadcasts nothing and the committed row self-heals on the client's next channel-list fetch. RemoveChannel is kept only on the genuine !exists path. 2. Migration 019 extends the per-server event fence on the servers table from DELETE-only to also cover owner_id UPDATE. An ownership transfer changes channel-visibility authorization; no endpoint updates owner_id today, so the trigger fires never until one is added -- at which point the create-vs-authz race is already closed by construction. Future-proofing, zero current cost. Mutation-verified pins for both. Full Go -race suite + gofmt/vet/lint green. --- .../internal/api/attachment_migration_test.go | 57 +++++++++++++++++++ server/internal/realtime/hub.go | 12 +++- server/internal/realtime/hub_test.go | 30 ++++++++++ .../019_server_owner_fence.down.sql | 5 ++ .../migrations/019_server_owner_fence.up.sql | 12 ++++ 5 files changed, 114 insertions(+), 2 deletions(-) create mode 100644 server/migrations/019_server_owner_fence.down.sql create mode 100644 server/migrations/019_server_owner_fence.up.sql diff --git a/server/internal/api/attachment_migration_test.go b/server/internal/api/attachment_migration_test.go index 1ebe26f..3ac5fef 100644 --- a/server/internal/api/attachment_migration_test.go +++ b/server/internal/api/attachment_migration_test.go @@ -74,3 +74,60 @@ func TestMigration018BackfillsAttachmentPositions(t *testing.T) { t.Fatal("position column should be dropped after rollback") } } + +// TestMigration019FencesServerOwnerUpdate verifies that migration 019 extends the +// per-server event fence on the servers table from DELETE-only to also cover +// owner_id UPDATE (an ownership transfer changes channel-visibility authorization), +// and that the down migration reverts it to DELETE-only. +func TestMigration019FencesServerOwnerUpdate(t *testing.T) { + dsn := testutil.NewMigrationDB(t) + m := newMigrator(t, dsn) + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer pool.Close() + + fenceEvents := func() map[string]bool { + t.Helper() + rows, err := pool.Query(ctx, + `SELECT DISTINCT event_manipulation FROM information_schema.triggers + WHERE event_object_table = 'servers' + AND action_statement LIKE '%lock_bastion_server_events%'`) + if err != nil { + t.Fatalf("read servers fence events: %v", err) + } + defer rows.Close() + events := map[string]bool{} + for rows.Next() { + var e string + if err := rows.Scan(&e); err != nil { + t.Fatalf("scan event: %v", err) + } + events[e] = true + } + return events + } + + if err := m.Migrate(18); err != nil { + t.Fatalf("migrate to v18: %v", err) + } + if ev := fenceEvents(); !ev["DELETE"] || ev["UPDATE"] { + t.Fatalf("v18 servers fence events = %v, want DELETE only", ev) + } + + if err := m.Migrate(19); err != nil { + t.Fatalf("migrate to v19: %v", err) + } + if ev := fenceEvents(); !ev["DELETE"] || !ev["UPDATE"] { + t.Fatalf("v19 servers fence events = %v, want DELETE and UPDATE", ev) + } + + if err := m.Migrate(18); err != nil { + t.Fatalf("migrate down to v18: %v", err) + } + if ev := fenceEvents(); !ev["DELETE"] || ev["UPDATE"] { + t.Fatalf("after rollback servers fence events = %v, want DELETE only", ev) + } +} diff --git a/server/internal/realtime/hub.go b/server/internal/realtime/hub.go index a7ae669..930705b 100644 --- a/server/internal/realtime/hub.go +++ b/server/internal/realtime/hub.go @@ -328,11 +328,19 @@ func (h *Hub) SubscribeAuthorizedStable(channelID uuid.UUID, read func() (ids [] gen := h.ReconcileGen() ids, exists, err := read() if err != nil { - h.RemoveChannel(channelID) // fail closed on an indeterminate set + // Do NOT tear down the channel's subscriptions on a transient read + // error: the row is committed and any users a concurrent connect + // subscribed are authorized (connects only subscribe viewable + // channels), so dropping them would strand already-connected members + // until they reconnect. Return the error; the caller broadcasts nothing + // and the client self-heals on its next channel-list fetch. (Under the + // create fence this is a pass-0 error with nothing yet subscribed by us; + // the multi-pass fallback preserves the last authorized read's + // subscriptions, reconciled on the next authorization change.) return false, err } if !exists { - h.RemoveChannel(channelID) + h.RemoveChannel(channelID) // channel is gone: drop its subscriptions return false, nil } // Reconcile the ENTIRE live set, not merely users added by this call. A diff --git a/server/internal/realtime/hub_test.go b/server/internal/realtime/hub_test.go index 24ed2bc..a312ad4 100644 --- a/server/internal/realtime/hub_test.go +++ b/server/internal/realtime/hub_test.go @@ -425,6 +425,36 @@ func TestSubscribeAuthorizedStableConvergesUnderGenChurn(t *testing.T) { } } +// TestSubscribeAuthorizedStableReadErrorPreservesSubscriptions: a transient read +// error must NOT tear down the channel's existing subscriptions. A concurrent +// connect may have subscribed an authorized member (connects only subscribe +// viewable channels), and dropping them on an indeterminate read would strand an +// already-connected client until reconnect. The call returns the error and the +// caller broadcasts nothing; the committed row self-heals on the client's next +// channel-list fetch. +func TestSubscribeAuthorizedStableReadErrorPreservesSubscriptions(t *testing.T) { + hub := NewHub() + go hub.Run() + defer hub.Stop() + + userID := uuid.New() + channelID := uuid.New() + client := newDroppingClient(userID) + hub.RegisterUser(client) + hub.Subscribe(channelID, client) // a concurrent connect already subscribed this authorized member + + readErr := errors.New("transient db error") + _, err := hub.SubscribeAuthorizedStable(channelID, func() ([]uuid.UUID, bool, error) { + return nil, false, readErr + }) + if !errors.Is(err, readErr) { + t.Fatalf("expected the read error, got %v", err) + } + if !channelHas(hub, channelID, client) { + t.Fatal("a transient read error must not drop a pre-existing authorized subscription") + } +} + // TestSubscribeAuthorizedStableBestEffortOnFlood: if authorization genuinely never // stabilizes (the set differs on every read), the primitive does NOT error and // does NOT hang -- it best-efforts within its bounded attempts and returns. The diff --git a/server/migrations/019_server_owner_fence.down.sql b/server/migrations/019_server_owner_fence.down.sql new file mode 100644 index 0000000..0d8766e --- /dev/null +++ b/server/migrations/019_server_owner_fence.down.sql @@ -0,0 +1,5 @@ +DROP TRIGGER IF EXISTS fence_server_owner_events ON servers; + +CREATE TRIGGER fence_server_delete_events + BEFORE DELETE ON servers + FOR EACH ROW EXECUTE FUNCTION lock_bastion_server_events(); diff --git a/server/migrations/019_server_owner_fence.up.sql b/server/migrations/019_server_owner_fence.up.sql new file mode 100644 index 0000000..15de60c --- /dev/null +++ b/server/migrations/019_server_owner_fence.up.sql @@ -0,0 +1,12 @@ +-- Future-proof the server-event fence. authorizedMemberIDs reads servers.owner_id, +-- but migration 017 fenced the servers table on DELETE only. An ownership transfer +-- (UPDATE servers SET owner_id) would change channel-visibility authorization +-- without taking the exclusive per-server lock, reopening the create-vs-authz race +-- the fence closes. No endpoint updates owner_id today, so this trigger fires +-- NEVER until such an endpoint is added -- at which point the race is already +-- closed by construction. The lock function already resolves NEW.id on UPDATE. +DROP TRIGGER IF EXISTS fence_server_delete_events ON servers; + +CREATE TRIGGER fence_server_owner_events + BEFORE UPDATE OF owner_id OR DELETE ON servers + FOR EACH ROW EXECUTE FUNCTION lock_bastion_server_events();