Project sidecar writer registry

adapter/encryption_admin.go

@@ -31,6 +31,7 @@ type EncryptionAdminServer struct {
 	sidecarPath        string
 	keystore           *encryption.Keystore
 	fullNodeID         uint64
+	writerRegistry     encryption.WriterRegistryStore
 	buildSHA           string
 	latestAppliedIndex func() uint64
 	// proposer is the raw raft proposer used for cleartext-only
@@ -195,6 +196,19 @@ func WithEncryptionAdminFullNodeID(id uint64) EncryptionAdminServerOption {
 	}
 }
 
+// WithEncryptionAdminWriterRegistry wires the §4.1 writer-registry
+// store that read-only sidecar recovery RPCs use to project
+// writer_registry_for_caller. A nil argument is a no-op so tests and
+// encryption-disabled nodes keep the pre-Stage-7 empty-map posture.
+func WithEncryptionAdminWriterRegistry(reg encryption.WriterRegistryStore) EncryptionAdminServerOption {
+	return func(s *EncryptionAdminServer) {
+		if reg == nil {
+			return
+		}
+		s.writerRegistry = reg
+	}
+}
+
 // WithEncryptionAdminBuildSHA overrides the auto-detected
 // runtime/debug build SHA. Tests use this to pin a deterministic
 // value; production wiring leaves it empty.
@@ -387,10 +401,10 @@ func (s *EncryptionAdminServer) GetCapability(_ context.Context, _ *pb.Empty) (*
 // pointers; the wrapped material is leakage-safe because it is
 // KEK-wrapped, which is the same property the on-disk sidecar has.
 //
-// The writer_registry_for_caller map is empty until Stage 7 wires
-// the registry. Callers in the §7.1 cutover path tolerate an empty
-// map because the §5.6 step 1a batch is sourced from the
-// GetCapability fan-out, not from this RPC.
+// When the writer registry is wired, writer_registry_for_caller
+// carries this node's recorded last_seen_local_epoch per sidecar DEK.
+// Unwired tests and encryption-disabled nodes keep the historical
+// empty non-nil map.
 func (s *EncryptionAdminServer) GetSidecarState(_ context.Context, _ *pb.Empty) (*pb.SidecarStateReport, error) {
 	if s.sidecarPath == "" {
 		return nil, grpcStatusError(codes.FailedPrecondition, "encryption: sidecar path is not configured on this node")
@@ -399,14 +413,18 @@ func (s *EncryptionAdminServer) GetSidecarState(_ context.Context, _ *pb.Empty)
 	if err != nil {
 		return nil, statusFromSidecarErr(err)
 	}
+	writerRegistry, err := s.writerRegistryForCaller(sc, s.fullNodeID, codes.Internal)
+	if err != nil {
+		return nil, err
+	}
 	resp := &pb.SidecarStateReport{
 		ActiveStorageId:          sc.Active.Storage,
 		ActiveRaftId:             sc.Active.Raft,
 		StorageEnvelopeActive:    sc.StorageEnvelopeActive,
 		RaftEnvelopeCutoverIndex: sc.RaftEnvelopeCutoverIndex,
 		LatestAppliedIndex:       s.appliedIndex(sc.RaftAppliedIndex),
 		WrappedDeksById:          wrappedDEKMap(sc),
-		WriterRegistryForCaller:  map[uint32]uint32{},
+		WriterRegistryForCaller:  writerRegistry,
 	}
 	return resp, nil
 }
@@ -423,14 +441,6 @@ func (s *EncryptionAdminServer) GetSidecarState(_ context.Context, _ *pb.Empty)
 // state to a recovering follower and silently overwrite recent
 // rotations.
 func (s *EncryptionAdminServer) ResyncSidecar(ctx context.Context, req *pb.ResyncSidecarRequest) (*pb.ResyncSidecarResponse, error) {
-	// req.CallerFullNodeId is intentionally unused for the
-	// recovery payload itself in PR-B; Stage 7 will use it to
-	// scope the writer-registry projection to that specific
-	// caller per §5.5. Recording it here keeps the field on the
-	// hot path so a future leader-side audit log can correlate
-	// resyncs to the requesting member without a wire-format
-	// change.
-	_ = req
 	if err := s.requireLeader(ctx); err != nil {
 		return nil, err
 	}
@@ -441,20 +451,62 @@ func (s *EncryptionAdminServer) ResyncSidecar(ctx context.Context, req *pb.Resyn
 	if err != nil {
 		return nil, statusFromSidecarErr(err)
 	}
+	var callerFullNodeID uint64
+	if req != nil {
+		callerFullNodeID = req.GetCallerFullNodeId()
+	}
+	writerRegistry, err := s.writerRegistryForCaller(sc, callerFullNodeID, codes.InvalidArgument)
+	if err != nil {
+		return nil, err
+	}
 	return &pb.ResyncSidecarResponse{
 		WrappedDeksById:          wrappedDEKMap(sc),
 		ActiveStorageId:          sc.Active.Storage,
 		ActiveRaftId:             sc.Active.Raft,
 		LeaderLatestAppliedIndex: s.appliedIndex(sc.RaftAppliedIndex),
-		// §5.5 follower-repair: leader's recorded
-		// last_seen_local_epoch per (dek_id, caller). Stage 7
-		// fills this from the writer registry. PR-A returns an
-		// empty non-nil map because a node recovering before
-		// the registry exists has nothing to re-derive.
-		WriterRegistryForCaller: map[uint32]uint32{},
+		WriterRegistryForCaller:  writerRegistry,
 	}, nil
 }
 
+func (s *EncryptionAdminServer) writerRegistryForCaller(sc *encryption.Sidecar, fullNodeID uint64, missingIDCode codes.Code) (map[uint32]uint32, error) {
+	out := map[uint32]uint32{}
+	if s.writerRegistry == nil {
+		return out, nil
+	}
+	if fullNodeID == 0 {
+		return nil, grpcStatusError(missingIDCode,
+			"encryption: full_node_id is required to project writer_registry_for_caller")
+	}
+	nodeID16 := encryption.NodeID16(fullNodeID)
+	for idStr := range sc.Keys {
+		dekID, err := parseSidecarKeyID(idStr)
+		if err != nil {
+			return nil, grpcStatusErrorf(codes.Internal,
+				"encryption: sidecar key id %q could not be projected into writer registry: %v", idStr, err)
+		}
+		raw, ok, err := s.writerRegistry.GetRegistryRow(encryption.RegistryKey(dekID, nodeID16))
+		if err != nil {
+			return nil, grpcStatusErrorf(codes.Internal,
+				"encryption: read writer registry row for dek_id=%d full_node_id=%#x: %v", dekID, fullNodeID, err)
+		}
+		if !ok {
+			continue
+		}
+		row, err := encryption.DecodeRegistryValue(raw)
+		if err != nil {
+			return nil, grpcStatusErrorf(codes.Internal,
+				"encryption: decode writer registry row for dek_id=%d full_node_id=%#x: %v", dekID, fullNodeID, err)
+		}
+		if row.FullNodeID != fullNodeID {
+			return nil, grpcStatusErrorf(codes.Internal,
+				"encryption: writer registry node_id collision for dek_id=%d caller_full_node_id=%#x registry_full_node_id=%#x",
+				dekID, fullNodeID, row.FullNodeID)
+		}
+		out[dekID] = uint32(row.LastSeenLocalEpoch)
+	}
+	return out, nil
+}
+
 func (s *EncryptionAdminServer) appliedIndex(sidecarValue uint64) uint64 {
 	if s.latestAppliedIndex == nil {
 		return sidecarValue

adapter/encryption_admin_test.go

@@ -159,6 +159,59 @@ func TestEncryptionAdmin_GetSidecarState_ShipsWrappedDEKs(t *testing.T) {
 	assertSidecarWrappedDEKs(t, got)
 }
 
+func TestEncryptionAdmin_GetSidecarState_ProjectsWriterRegistryForLocalNode(t *testing.T) {
+	t.Parallel()
+	const localFullNodeID uint64 = 0xCAFE_BABE_0000_1234
+	path := writeSidecarFixture(t, &encryption.Sidecar{
+		RaftAppliedIndex: 42,
+		Active:           encryption.ActiveKeys{Storage: 1, Raft: 2},
+		Keys: map[string]encryption.SidecarKey{
+			"1": {Purpose: "storage", Wrapped: []byte("wrapped-1")},
+			"2": {Purpose: "raft", Wrapped: []byte("wrapped-2")},
+			"3": {Purpose: "storage", Wrapped: []byte("wrapped-old-storage")},
+		},
+	})
+	reg := newTestWriterRegistry()
+	reg.seed(t, 1, localFullNodeID, 4, 9)
+	reg.seed(t, 2, localFullNodeID, 5, 10)
+	// No row for DEK 3: the projection omits absent rows instead of
+	// inventing a zero epoch that could be mistaken for a real record.
+	srv := NewEncryptionAdminServer(
+		WithEncryptionAdminSidecarPath(path),
+		WithEncryptionAdminFullNodeID(localFullNodeID),
+		WithEncryptionAdminWriterRegistry(reg),
+	)
+
+	got, err := srv.GetSidecarState(context.Background(), &pb.Empty{})
+	if err != nil {
+		t.Fatalf("GetSidecarState: %v", err)
+	}
+	want := map[uint32]uint32{1: 9, 2: 10}
+	if fmt.Sprint(got.WriterRegistryForCaller) != fmt.Sprint(want) {
+		t.Fatalf("WriterRegistryForCaller=%v, want %v", got.WriterRegistryForCaller, want)
+	}
+}
+
+func TestEncryptionAdmin_GetSidecarState_RejectsMissingLocalNodeIDAsInternal(t *testing.T) {
+	t.Parallel()
+	path := writeSidecarFixture(t, &encryption.Sidecar{
+		Active: encryption.ActiveKeys{Storage: 1, Raft: 2},
+		Keys: map[string]encryption.SidecarKey{
+			"1": {Purpose: "storage", Wrapped: []byte("wrapped-1")},
+			"2": {Purpose: "raft", Wrapped: []byte("wrapped-2")},
+		},
+	})
+	srv := NewEncryptionAdminServer(
+		WithEncryptionAdminSidecarPath(path),
+		WithEncryptionAdminWriterRegistry(newTestWriterRegistry()),
+	)
+
+	_, err := srv.GetSidecarState(context.Background(), &pb.Empty{})
+	if status.Code(err) != codes.Internal {
+		t.Fatalf("GetSidecarState status=%v, want Internal for missing local full node id (err=%v)", status.Code(err), err)
+	}
+}
+
 func assertSidecarHeader(t *testing.T, got *pb.SidecarStateReport) {
 	t.Helper()
 	if got.ActiveStorageId != 1 || got.ActiveRaftId != 2 {
@@ -184,10 +237,10 @@ func assertSidecarWrappedDEKs(t *testing.T, got *pb.SidecarStateReport) {
 		t.Errorf("wrapped[2]=%q, want %q", got.WrappedDeksById[2], "wrapped-2")
 	}
 	if got.WriterRegistryForCaller == nil {
-		t.Errorf("WriterRegistryForCaller=nil, want empty non-nil map (PR-A contract)")
+		t.Errorf("WriterRegistryForCaller=nil, want empty non-nil map when registry is unwired")
 	}
 	if len(got.WriterRegistryForCaller) != 0 {
-		t.Errorf("WriterRegistryForCaller=%v, want empty in PR-A", got.WriterRegistryForCaller)
+		t.Errorf("WriterRegistryForCaller=%v, want empty when registry is unwired", got.WriterRegistryForCaller)
 	}
 }
 
@@ -245,14 +298,88 @@ func TestEncryptionAdmin_ResyncSidecar_ShipsWrappedDEKs(t *testing.T) {
 	if string(got.WrappedDeksById[3]) != "ws" || string(got.WrappedDeksById[4]) != "wr" {
 		t.Errorf("wrapped=%v, want id3=ws id4=wr", got.WrappedDeksById)
 	}
-	// Mirror the GetSidecarState contract: non-nil empty map until
-	// Stage 7 wires the writer registry. Locks in the §5.5 promise
-	// so a future change to the field cannot silently degrade to nil.
 	if got.WriterRegistryForCaller == nil {
-		t.Errorf("WriterRegistryForCaller=nil, want empty non-nil map (PR-A contract)")
+		t.Errorf("WriterRegistryForCaller=nil, want empty non-nil map when registry is unwired")
 	}
 	if len(got.WriterRegistryForCaller) != 0 {
-		t.Errorf("WriterRegistryForCaller=%v, want empty in PR-A", got.WriterRegistryForCaller)
+		t.Errorf("WriterRegistryForCaller=%v, want empty when registry is unwired", got.WriterRegistryForCaller)
+	}
+}
+
+func TestEncryptionAdmin_ResyncSidecar_ProjectsWriterRegistryForCaller(t *testing.T) {
+	t.Parallel()
+	const callerFullNodeID uint64 = 0x1111_2222_3333_4444
+	path := writeSidecarFixture(t, &encryption.Sidecar{
+		RaftAppliedIndex: 17,
+		Active:           encryption.ActiveKeys{Storage: 3, Raft: 4},
+		Keys: map[string]encryption.SidecarKey{
+			"3": {Purpose: "storage", Wrapped: []byte("ws")},
+			"4": {Purpose: "raft", Wrapped: []byte("wr")},
+			"5": {Purpose: "storage", Wrapped: []byte("old")},
+		},
+	})
+	reg := newTestWriterRegistry()
+	reg.seed(t, 3, callerFullNodeID, 1, 6)
+	reg.seed(t, 4, callerFullNodeID, 2, 8)
+	srv := NewEncryptionAdminServer(
+		WithEncryptionAdminSidecarPath(path),
+		WithEncryptionAdminLeaderView(stubLeaderView{state: raftengine.StateLeader}),
+		WithEncryptionAdminWriterRegistry(reg),
+	)
+
+	got, err := srv.ResyncSidecar(context.Background(), &pb.ResyncSidecarRequest{CallerFullNodeId: callerFullNodeID})
+	if err != nil {
+		t.Fatalf("ResyncSidecar: %v", err)
+	}
+	want := map[uint32]uint32{3: 6, 4: 8}
+	if fmt.Sprint(got.WriterRegistryForCaller) != fmt.Sprint(want) {
+		t.Fatalf("WriterRegistryForCaller=%v, want %v", got.WriterRegistryForCaller, want)
+	}
+}
+
+func TestEncryptionAdmin_ResyncSidecar_RejectsMissingCallerNodeIDAsInvalidArgument(t *testing.T) {
+	t.Parallel()
+	path := writeSidecarFixture(t, &encryption.Sidecar{
+		Active: encryption.ActiveKeys{Storage: 3, Raft: 4},
+		Keys: map[string]encryption.SidecarKey{
+			"3": {Purpose: "storage", Wrapped: []byte("ws")},
+			"4": {Purpose: "raft", Wrapped: []byte("wr")},
+		},
+	})
+	srv := NewEncryptionAdminServer(
+		WithEncryptionAdminSidecarPath(path),
+		WithEncryptionAdminLeaderView(stubLeaderView{state: raftengine.StateLeader}),
+		WithEncryptionAdminWriterRegistry(newTestWriterRegistry()),
+	)
+
+	_, err := srv.ResyncSidecar(context.Background(), &pb.ResyncSidecarRequest{})
+	if status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("ResyncSidecar status=%v, want InvalidArgument for missing caller full node id (err=%v)", status.Code(err), err)
+	}
+}
+
+func TestEncryptionAdmin_ResyncSidecar_RejectsWriterRegistryNodeCollision(t *testing.T) {
+	t.Parallel()
+	const callerFullNodeID uint64 = 0x10001
+	const collidingFullNodeID uint64 = 0x20001
+	path := writeSidecarFixture(t, &encryption.Sidecar{
+		Active: encryption.ActiveKeys{Storage: 7, Raft: 8},
+		Keys: map[string]encryption.SidecarKey{
+			"7": {Purpose: "storage", Wrapped: []byte("ws")},
+			"8": {Purpose: "raft", Wrapped: []byte("wr")},
+		},
+	})
+	reg := newTestWriterRegistry()
+	reg.seed(t, 7, collidingFullNodeID, 1, 2)
+	srv := NewEncryptionAdminServer(
+		WithEncryptionAdminSidecarPath(path),
+		WithEncryptionAdminLeaderView(stubLeaderView{state: raftengine.StateLeader}),
+		WithEncryptionAdminWriterRegistry(reg),
+	)
+
+	_, err := srv.ResyncSidecar(context.Background(), &pb.ResyncSidecarRequest{CallerFullNodeId: callerFullNodeID})
+	if status.Code(err) != codes.Internal {
+		t.Fatalf("ResyncSidecar status=%v, want Internal for writer-registry node collision (err=%v)", status.Code(err), err)
 	}
 }
 
@@ -1115,6 +1242,48 @@ func writeSidecarFixture(t *testing.T, sc *encryption.Sidecar) string {
 	return path
 }
 
+type testWriterRegistry struct {
+	rows   map[string][]byte
+	getErr error
+}
+
+func newTestWriterRegistry() *testWriterRegistry {
+	return &testWriterRegistry{rows: map[string][]byte{}}
+}
+
+func (r *testWriterRegistry) GetRegistryRow(key []byte) ([]byte, bool, error) {
+	if r.getErr != nil {
+		return nil, false, r.getErr
+	}
+	raw, ok := r.rows[string(key)]
+	if !ok {
+		return nil, false, nil
+	}
+	out := append([]byte(nil), raw...)
+	return out, true, nil
+}
+
+func (r *testWriterRegistry) SetRegistryRow(key, value []byte) error {
+	if r.rows == nil {
+		r.rows = map[string][]byte{}
+	}
+	r.rows[string(key)] = append([]byte(nil), value...)
+	return nil
+}
+
+func (r *testWriterRegistry) seed(t *testing.T, dekID uint32, fullNodeID uint64, firstSeen, lastSeen uint16) {
+	t.Helper()
+	key := encryption.RegistryKey(dekID, encryption.NodeID16(fullNodeID))
+	val := encryption.EncodeRegistryValue(encryption.RegistryValue{
+		FullNodeID:          fullNodeID,
+		FirstSeenLocalEpoch: firstSeen,
+		LastSeenLocalEpoch:  lastSeen,
+	})
+	if err := r.SetRegistryRow(key, val); err != nil {
+		t.Fatalf("seed registry row: %v", err)
+	}
+}
+
 // fixedCapabilityFanout returns a closure that yields the supplied
 // result regardless of context — lets tests drive the §4 fan-out
 // branches deterministically without spinning real clients. A

docs/design/2026_04_29_partial_data_at_rest_encryption.md

@@ -1,6 +1,6 @@
 # Data-at-rest encryption for elastickv
 
-Status: Partial — Stages 0–6 and Stage 8 shipped (5E deferred); Stage 7 partially shipped; Stage 9 open
+Status: Partial — Stages 0–8 shipped (5E deferred); Stage 9 open
 Author: bootjp
 Date: 2026-04-29
 
@@ -30,7 +30,7 @@ Date: 2026-04-29
 | 6D | §6.6 `enable-storage-envelope` admin RPC + §7.1 Phase-1 storage cutover (§6.2 toggle ON) + Voters ∪ Learners capability gate + production storage-envelope write-path wiring | shipped | `2026_05_18_implemented_6d_enable_storage_envelope.md` + `2026_05_25_implemented_6d6c2_production_storage_envelope_wiring.md` |
 | 6E | §6.6 `enable-raft-envelope` admin RPC + §7.1 Phase-2 raft cutover + `raft_envelope_cutover_index` sidecar record + `internal/raftengine/etcd/engine.go` `applyNormalEntry` unwrap hook activation + `ErrRaftUnwrapFailed` HaltApply path + `kv/coordinator.go` / `kv/sharded_coordinator.go` wrap-on-propose switch (Phase-2 leader-side §6.3 proposal-payload wrap) + §7.1 steps 1–6 proposal quiescence barrier (block new user proposal intake, drain in-flight queue, source-tag exemption for the cutover entry itself) + production runtime wiring for startup/apply-time wrap install and post-cutover admin proposals + 6C-4 fail-closed guards. | shipped | 2026_05_31_implemented_6e_enable_raft_envelope.md |
 | 6F | §6.5 `--encryption-rotate-on-startup` request flag + leader-elected rotation proposal on the default encryption Raft group. The leader rotates every active DEK purpose once before listeners open; followers keep an in-memory pending request and fire only if they acquire leadership in the same process uptime. | shipped | this PR |
-| 7 | Writer registry + deterministic nonce (§4.1). Implemented slices: process-start registration, storage-layer registration gate, runtime re-registration for cutover and rotation, and conf-change-time pre-registration. Still open: §5.5 registry projection in `GetSidecarState` / `ResyncSidecar` (`WriterRegistryForCaller`). | partial | `2026_05_26_implemented_7a_process_start_registration.md` + `2026_05_26_implemented_7a2_storage_layer_registration_enforcement.md` + `2026_05_28_implemented_7b_runtime_reregistration.md` + `2026_05_28_implemented_7b_prime_runtime_reregistration_rotation.md` + `2026_05_29_implemented_7c_confchange_time_registration.md` |
+| 7 | Writer registry + deterministic nonce (§4.1). Covers process-start registration, storage-layer registration gate, runtime re-registration for cutover and rotation, conf-change-time pre-registration, and §5.5 registry projection in `GetSidecarState` / `ResyncSidecar` (`WriterRegistryForCaller`). | shipped | `2026_05_26_implemented_7a_process_start_registration.md` + `2026_05_26_implemented_7a2_storage_layer_registration_enforcement.md` + `2026_05_28_implemented_7b_runtime_reregistration.md` + `2026_05_28_implemented_7b_prime_runtime_reregistration_rotation.md` + `2026_05_29_implemented_7c_confchange_time_registration.md` + `2026_07_11_implemented_7d_sidecar_registry_projection.md` |
 | 8 | Snapshot header v2 (§4.4); WAL coverage closure (§4.3 / §4.6) | shipped | [`2026_05_29_implemented_8a_snapshot_header_v2.md`](2026_05_29_implemented_8a_snapshot_header_v2.md) + [`2026_06_01_implemented_8b_wal_coverage_closure.md`](2026_06_01_implemented_8b_wal_coverage_closure.md) |
 | 9 | KMS-backed wrappers, compression, rotation/retire/rewrite, Jepsen (§5.2, §5.4, §6.4, §8) | open | — |
 

docs/design/2026_05_29_implemented_7c_confchange_time_registration.md

@@ -463,8 +463,8 @@ that bypass the standard workflow) must rely on the
 `ErrNodeIDCollision` startup membership pre-check before issuing the
 ConfChange.
 
-This closes the 7c ConfChange-time registration slice. The parent
-Stage 7 remains partial until the §5.5 `WriterRegistryForCaller`
-projection is wired into `GetSidecarState` / `ResyncSidecar`.
+This closes the 7c ConfChange-time registration slice. The remaining
+Stage 7 §5.5 `WriterRegistryForCaller` projection is tracked by
+`2026_07_11_implemented_7d_sidecar_registry_projection.md`.
 Stage 8 (snapshot header v2) and Stage 9 (KMS + compress +
 rotation/retire/rewrite + Jepsen) follow.