fix(core): harden room RBAC recovery

Author: hmans

cli/internal/core/core.go

@@ -289,16 +289,16 @@ func (c *ChattoCore) Run(ctx context.Context) error {
 		if err := c.applyConfigOwners(gctx); err != nil {
 			return fmt.Errorf("apply config owners: %w", err)
 		}
-		// Adopt every existing room before committing the server defaults
-		// marker. This rollout boundary lets later unmarked rooms recover
-		// interrupted creation-time initialization without reapplying
-		// defaults that operators cleared on pre-marker rooms.
-		if err := c.EnsureDefaultChannelRoomPermissions(gctx); err != nil {
-			return fmt.Errorf("ensure default channel room permissions: %w", err)
-		}
+		// Commit the server marker with a durable evt.room.> cutoff before
+		// adopting rooms. Every replica then classifies rooms against the
+		// same boundary, including rooms created concurrently by an older
+		// serving replica during a rolling deployment.
 		if err := c.EnsureDefaultRolePermissions(gctx); err != nil {
 			return fmt.Errorf("ensure default role permissions: %w", err)
 		}
+		if err := c.EnsureDefaultChannelRoomPermissions(gctx); err != nil {
+			return fmt.Errorf("ensure default channel room permissions: %w", err)
+		}
 		// Seed the default room group and ensure every existing
 		// channel room belongs to a set (ADR-031). Idempotent —
 		// runs on every boot. Has to happen AFTER projectors are

cli/internal/core/permission_ops.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 
+	"hmans.de/chatto/internal/events"
 	corev1 "hmans.de/chatto/internal/pb/chatto/core/v1"
 )
 
@@ -283,39 +284,50 @@ func (c *ChattoCore) SeedDefaultChannelRoomPermissions(ctx context.Context, room
 		ScopeRoom,
 		roomID,
 		roomRBACDefaultsVersion,
-		true,
+		rbacDefaultsFillMissing,
 		defaultChannelRoomDecisions(roomID, roomName),
+		0,
 	)
 }
 
-// EnsureDefaultChannelRoomPermissions adopts existing rooms without changing
-// their decisions before the server initialization marker exists. On later
-// boots, an unmarked room was created after that rollout boundary and receives
-// its defaults plus marker atomically.
+// EnsureDefaultChannelRoomPermissions adopts rooms at or before the durable
+// server cutoff without changing their decisions. An unmarked room created
+// after that cutoff receives any missing defaults plus its marker atomically.
 func (c *ChattoCore) EnsureDefaultChannelRoomPermissions(ctx context.Context) error {
+	serverInitialized := c.RBAC.DefaultsVersion(ScopeServer, "") >= serverRBACDefaultsVersion
+	roomStreamCutoff := c.RBAC.ServerDefaultsRoomStreamCutoff()
+	if serverInitialized && roomStreamCutoff > 0 {
+		if err := c.RoomDirectoryProjector.WaitFor(ctx, events.SubjectPosition(events.RoomSubjectFilter(), roomStreamCutoff)); err != nil {
+			return fmt.Errorf("wait for channel rooms through defaults cutoff: %w", err)
+		}
+	}
 	rooms, err := c.ListRooms(ctx, KindChannel)
 	if err != nil {
 		return fmt.Errorf("list channel rooms: %w", err)
 	}
-	serverMarkerSeq := c.RBAC.DefaultsInitializedSeq(ScopeServer, "")
 	for _, room := range rooms {
+		seedMode := rbacDefaultsAdoptOnly
+		if c.shouldRecoverUnmarkedRoom(room.Id, serverInitialized, roomStreamCutoff) {
+			seedMode = rbacDefaultsFillMissing
+		}
 		if err := c.ensureRBACDefaultsInitialized(
 			ctx,
 			ScopeRoom,
 			room.Id,
 			roomRBACDefaultsVersion,
-			c.shouldSeedUnmarkedRoom(room.Id, serverMarkerSeq),
+			seedMode,
 			defaultChannelRoomDecisions(room.Id, room.Name),
+			0,
 		); err != nil {
 			return fmt.Errorf("ensure room permissions for %s: %w", room.Id, err)
 		}
 	}
 	return nil
 }
 
-func (c *ChattoCore) shouldSeedUnmarkedRoom(roomID string, serverMarkerSeq uint64) bool {
+func (c *ChattoCore) shouldRecoverUnmarkedRoom(roomID string, serverInitialized bool, roomStreamCutoff uint64) bool {
 	createdSeq, ok := c.RoomCatalog.CreatedSeq(roomID)
-	return serverMarkerSeq > 0 && ok && createdSeq > serverMarkerSeq
+	return serverInitialized && ok && createdSeq > roomStreamCutoff
 }
 
 // ============================================================================
@@ -362,13 +374,18 @@ func (c *ChattoCore) InitDefaultPermissions(ctx context.Context) error {
 // Current defaults are included only when RBAC has no permission decisions at
 // any scope; existing installations are marked without changing their state.
 func (c *ChattoCore) EnsureDefaultRolePermissions(ctx context.Context) error {
+	roomStreamCutoff, err := c.EventPublisher.LastSubjectSeq(ctx, events.RoomSubjectFilter())
+	if err != nil {
+		return fmt.Errorf("read room stream cutoff for RBAC defaults: %w", err)
+	}
 	return c.ensureRBACDefaultsInitialized(
 		ctx,
 		ScopeServer,
 		"",
 		serverRBACDefaultsVersion,
-		true,
+		rbacDefaultsSeedWhenEmpty,
 		defaultRBACDecisions(),
+		roomStreamCutoff,
 	)
 }
 

cli/internal/core/projection_admin.go

@@ -259,7 +259,7 @@ func (p *RBACProjection) adminProjectionEstimate() (int64, int64, []ProjectionAd
 	}
 	var defaultsBytes int64
 	for key := range p.defaults {
-		defaultsBytes += projectionMapEntryOverhead + int64(len(key.scope)+len(key.scopeID)) + 4 + 8
+		defaultsBytes += projectionMapEntryOverhead + int64(len(key.scope)+len(key.scopeID)) + 4 + 8 + 8
 	}
 	retainedEventIDs := p.replayGuard.retainedEventIDs()
 	retainedEventIDsBytes := estimateStringSetBytes(retainedEventIDs)

cli/internal/core/rbac_defaults.go

@@ -16,6 +16,14 @@ const (
 	roomRBACDefaultsVersion   uint32 = 1
 )
 
+type rbacDefaultsSeedMode uint8
+
+const (
+	rbacDefaultsAdoptOnly rbacDefaultsSeedMode = iota
+	rbacDefaultsSeedWhenEmpty
+	rbacDefaultsFillMissing
+)
+
 // ensureRBACDefaultsInitialized atomically writes any selected defaults and a
 // durable version marker under the deployment-wide RBAC OCC filter. The
 // decision check is repeated after every conflict so a concurrent operator or
@@ -26,8 +34,9 @@ func (c *ChattoCore) ensureRBACDefaultsInitialized(
 	scope PermissionScope,
 	scopeID string,
 	version uint32,
-	seedWhenEmpty bool,
+	seedMode rbacDefaultsSeedMode,
 	defaults []rbacSeedDecision,
+	roomStreamCutoff uint64,
 ) error {
 	filter := events.RBACSubjectFilter()
 
@@ -45,10 +54,29 @@ func (c *ChattoCore) ensureRBACDefaultsInitialized(
 		}
 
 		entries := make([]events.BatchEntry, 0, len(defaults)+1)
-		if currentVersion == 0 && seedWhenEmpty && !c.hasPermissionDecisionsForDefaults(scope, scopeID) {
-			entries = append(entries, rbacSeedEntries(nil, nil, append([]rbacSeedDecision(nil), defaults...))...)
+		if currentVersion == 0 {
+			var selected []rbacSeedDecision
+			switch seedMode {
+			case rbacDefaultsSeedWhenEmpty:
+				if !c.hasPermissionDecisionsForDefaults(scope, scopeID) {
+					selected = append(selected, defaults...)
+				}
+			case rbacDefaultsFillMissing:
+				for _, decision := range defaults {
+					if c.RBAC.getTypedDecision(
+						decision.scope,
+						decision.scopeID,
+						decision.subjectKind,
+						decision.subject,
+						decision.permission,
+					) == DecisionNone {
+						selected = append(selected, decision)
+					}
+				}
+			}
+			entries = append(entries, rbacSeedEntries(nil, nil, selected)...)
 		}
-		entries = append(entries, rbacDefaultsInitializedEntry(scope, scopeID, version))
+		entries = append(entries, rbacDefaultsInitializedEntry(scope, scopeID, version, roomStreamCutoff))
 		entries[0].HasOCC = true
 		entries[0].ExpectedSeq = filterSeq
 		entries[0].FilterSubject = filter
@@ -82,8 +110,8 @@ func (c *ChattoCore) hasPermissionDecisionsForDefaults(scope PermissionScope, sc
 	return c.RBAC.HasPermissionDecisions(scope, scopeID)
 }
 
-func rbacDefaultsInitializedEntry(scope PermissionScope, scopeID string, version uint32) events.BatchEntry {
-	marker := &corev1.RbacDefaultsInitializedEvent{Version: version}
+func rbacDefaultsInitializedEntry(scope PermissionScope, scopeID string, version uint32, roomStreamCutoff uint64) events.BatchEntry {
+	marker := &corev1.RbacDefaultsInitializedEvent{Version: version, RoomStreamCutoff: roomStreamCutoff}
 	switch scope {
 	case ScopeServer:
 		marker.Scope = &corev1.RbacDefaultsInitializedEvent_Server{Server: &corev1.RbacDefaultsInitializedEvent_ServerScope{}}

cli/internal/core/rbac_defaults_test.go

@@ -38,11 +38,11 @@ func TestEnsureDefaultRolePermissions_SeedsEmptyRBACAtomically(t *testing.T) {
 }
 
 func TestRBACDefaultsInitializedEntry_UsesScopedAggregate(t *testing.T) {
-	server := rbacDefaultsInitializedEntry(ScopeServer, "", serverRBACDefaultsVersion)
+	server := rbacDefaultsInitializedEntry(ScopeServer, "", serverRBACDefaultsVersion, 17)
 	if want := events.RBACServerAggregate().Subject(events.EventRBACDefaultsInitialized); server.Subject != want {
 		t.Fatalf("server marker subject = %q, want %q", server.Subject, want)
 	}
-	room := rbacDefaultsInitializedEntry(ScopeRoom, "Rabc123", roomRBACDefaultsVersion)
+	room := rbacDefaultsInitializedEntry(ScopeRoom, "Rabc123", roomRBACDefaultsVersion, 0)
 	if want := events.RBACScopedAggregate("Rabc123").Subject(events.EventRBACDefaultsInitialized); room.Subject != want {
 		t.Fatalf("room marker subject = %q, want %q", room.Subject, want)
 	}
@@ -79,10 +79,10 @@ func TestRBACDefaultsVersionAdvance_DoesNotReapplyDefaults(t *testing.T) {
 	core := newRBACDefaultsTestCore(t, harness)
 	ctx := testContext(t)
 
-	if err := core.ensureRBACDefaultsInitialized(ctx, ScopeServer, "", 1, false, defaultRBACDecisions()); err != nil {
+	if err := core.ensureRBACDefaultsInitialized(ctx, ScopeServer, "", 1, rbacDefaultsAdoptOnly, defaultRBACDecisions(), 0); err != nil {
 		t.Fatalf("write version 1 marker: %v", err)
 	}
-	if err := core.ensureRBACDefaultsInitialized(ctx, ScopeServer, "", 2, true, defaultRBACDecisions()); err != nil {
+	if err := core.ensureRBACDefaultsInitialized(ctx, ScopeServer, "", 2, rbacDefaultsSeedWhenEmpty, defaultRBACDecisions(), 0); err != nil {
 		t.Fatalf("advance to version 2: %v", err)
 	}
 
@@ -121,6 +121,86 @@ func TestSeedDefaultChannelRoomPermissions_MarkerPreservesClear(t *testing.T) {
 	}
 }
 
+func TestSeedDefaultChannelRoomPermissions_FillsMissingDefaultsAlongsideExistingDecision(t *testing.T) {
+	harness := newTestEventHarness(t)
+	core := newRBACDefaultsTestCore(t, harness)
+	ctx := testContext(t)
+	roomID := "Rpartial"
+
+	if err := core.DenyRoomPermission(ctx, SystemActorID, roomID, RoleEveryone, PermMessageReact); err != nil {
+		t.Fatalf("DenyRoomPermission unrelated override: %v", err)
+	}
+	if err := core.SeedDefaultChannelRoomPermissions(ctx, roomID, AnnouncementsRoomName); err != nil {
+		t.Fatalf("SeedDefaultChannelRoomPermissions: %v", err)
+	}
+
+	if got := core.RBAC.GetDecision(ScopeRoom, roomID, RoleEveryone, PermMessageReact); got != DecisionDeny {
+		t.Fatalf("existing unrelated decision = %s, want %s", got, DecisionDeny)
+	}
+	if got := core.RBAC.GetDecision(ScopeRoom, roomID, RoleEveryone, PermMessagePost); got != DecisionDeny {
+		t.Fatalf("missing announcements default = %s, want %s", got, DecisionDeny)
+	}
+}
+
+func TestSeedDefaultChannelRoomPermissions_PreservesExistingSameKeyDecision(t *testing.T) {
+	harness := newTestEventHarness(t)
+	core := newRBACDefaultsTestCore(t, harness)
+	ctx := testContext(t)
+	roomID := "Roverride"
+
+	if err := core.GrantRoomPermission(ctx, SystemActorID, roomID, RoleEveryone, PermMessagePost); err != nil {
+		t.Fatalf("GrantRoomPermission override: %v", err)
+	}
+	if err := core.SeedDefaultChannelRoomPermissions(ctx, roomID, AnnouncementsRoomName); err != nil {
+		t.Fatalf("SeedDefaultChannelRoomPermissions: %v", err)
+	}
+
+	if got := core.RBAC.GetDecision(ScopeRoom, roomID, RoleEveryone, PermMessagePost); got != DecisionAllow {
+		t.Fatalf("same-key decision = %s, want preserved %s", got, DecisionAllow)
+	}
+}
+
+func TestEnsureDefaultChannelRoomPermissions_RecoversPostCutoffRoomWithExistingDecision(t *testing.T) {
+	core, _ := setupTestCore(t)
+	ctx := testContext(t)
+	roomID := "Rpostcutoff"
+	roomEvent := newEvent(SystemActorID, &corev1.Event{Event: &corev1.Event_RoomCreated{
+		RoomCreated: &corev1.RoomCreatedEvent{
+			RoomId: roomID,
+			Kind:   corev1.RoomKind_ROOM_KIND_CHANNEL,
+			Name:   AnnouncementsRoomName,
+		},
+	}})
+	filterSeq, err := core.EventPublisher.LastSubjectSeq(ctx, events.RoomSubjectFilter())
+	if err != nil {
+		t.Fatalf("read room tail: %v", err)
+	}
+	roomSubject := events.RoomAggregate(roomID).SubjectFor(roomEvent)
+	roomSeq, err := core.EventPublisher.AppendAtFilter(ctx, roomSubject, roomEvent, events.RoomSubjectFilter(), filterSeq)
+	if err != nil {
+		t.Fatalf("append post-cutoff room: %v", err)
+	}
+	if err := core.RoomDirectoryProjector.WaitFor(ctx, events.SubjectPosition(roomSubject, roomSeq)); err != nil {
+		t.Fatalf("wait for post-cutoff room: %v", err)
+	}
+	if err := core.DenyRoomPermission(ctx, SystemActorID, roomID, RoleEveryone, PermMessageReact); err != nil {
+		t.Fatalf("write unrelated partial decision: %v", err)
+	}
+
+	if err := core.EnsureDefaultChannelRoomPermissions(ctx); err != nil {
+		t.Fatalf("EnsureDefaultChannelRoomPermissions: %v", err)
+	}
+	if got := core.RBAC.GetDecision(ScopeRoom, roomID, RoleEveryone, PermMessageReact); got != DecisionDeny {
+		t.Fatalf("existing unrelated decision = %s, want %s", got, DecisionDeny)
+	}
+	if got := core.RBAC.GetDecision(ScopeRoom, roomID, RoleEveryone, PermMessagePost); got != DecisionDeny {
+		t.Fatalf("recovered announcements default = %s, want %s", got, DecisionDeny)
+	}
+	if got := core.RBAC.DefaultsVersion(ScopeRoom, roomID); got != roomRBACDefaultsVersion {
+		t.Fatalf("room defaults version = %d, want %d", got, roomRBACDefaultsVersion)
+	}
+}
+
 func TestExistingRoomAdoption_PreservesCompletelyClearedScope(t *testing.T) {
 	harness := newTestEventHarness(t)
 	core := newRBACDefaultsTestCore(t, harness)
@@ -138,8 +218,9 @@ func TestExistingRoomAdoption_PreservesCompletelyClearedScope(t *testing.T) {
 		ScopeRoom,
 		roomID,
 		roomRBACDefaultsVersion,
-		false,
+		rbacDefaultsAdoptOnly,
 		defaultChannelRoomDecisions(roomID, AnnouncementsRoomName),
+		0,
 	); err != nil {
 		t.Fatalf("adopt existing room: %v", err)
 	}
@@ -165,18 +246,40 @@ func TestRoomDefaultsRolloutBoundaryUsesCreationSequence(t *testing.T) {
 	}
 
 	applyRoomCreated("Rbefore", 10)
-	marker := rbacDefaultsInitializedEntry(ScopeServer, "", serverRBACDefaultsVersion)
-	if err := core.RBAC.Apply(marker.Event, 20); err != nil {
+	applyRoomCreated("Rbetween", 30)
+	marker := rbacDefaultsInitializedEntry(ScopeServer, "", serverRBACDefaultsVersion, 20)
+	if err := core.RBAC.Apply(marker.Event, 40); err != nil {
 		t.Fatalf("apply server marker: %v", err)
 	}
-	applyRoomCreated("Rafter", 30)
 
-	markerSeq := core.RBAC.DefaultsInitializedSeq(ScopeServer, "")
-	if core.shouldSeedUnmarkedRoom("Rbefore", markerSeq) {
-		t.Fatal("room created before the rollout marker would receive defaults")
+	cutoff := core.RBAC.ServerDefaultsRoomStreamCutoff()
+	if core.shouldRecoverUnmarkedRoom("Rbefore", true, cutoff) {
+		t.Fatal("room created before the room cutoff would receive defaults")
+	}
+	if !core.shouldRecoverUnmarkedRoom("Rbetween", true, cutoff) {
+		t.Fatal("room created after the cutoff but before the server marker would not recover defaults")
+	}
+}
+
+func TestEnsureDefaultRolePermissions_PersistsRoomStreamCutoff(t *testing.T) {
+	harness := newTestEventHarness(t)
+	core := newRBACDefaultsTestCore(t, harness)
+	ctx := testContext(t)
+	roomID := "Rcutoff"
+	roomEvent := newEvent(SystemActorID, &corev1.Event{Event: &corev1.Event_RoomCreated{
+		RoomCreated: &corev1.RoomCreatedEvent{RoomId: roomID, Kind: corev1.RoomKind_ROOM_KIND_CHANNEL},
+	}})
+	roomSubject := events.RoomAggregate(roomID).SubjectFor(roomEvent)
+	roomSeq, err := harness.publisher.AppendAtFilter(ctx, roomSubject, roomEvent, events.RoomSubjectFilter(), 0)
+	if err != nil {
+		t.Fatalf("append room creation: %v", err)
+	}
+
+	if err := core.EnsureDefaultRolePermissions(ctx); err != nil {
+		t.Fatalf("EnsureDefaultRolePermissions: %v", err)
 	}
-	if !core.shouldSeedUnmarkedRoom("Rafter", markerSeq) {
-		t.Fatal("room created after the rollout marker would not receive defaults")
+	if got := core.RBAC.ServerDefaultsRoomStreamCutoff(); got != roomSeq {
+		t.Fatalf("room stream cutoff = %d, want %d", got, roomSeq)
 	}
 }