fix(server): subscription hardening — read-error subscription preservation + owner_id fence (PR #22 follow-ups)

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)
+	}
+}

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

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

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();

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();