diff --git a/adapter/encryption_admin.go b/adapter/encryption_admin.go index cda27e627..3cc44cc84 100644 --- a/adapter/encryption_admin.go +++ b/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,6 +413,10 @@ 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, @@ -406,7 +424,7 @@ func (s *EncryptionAdminServer) GetSidecarState(_ context.Context, _ *pb.Empty) 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 diff --git a/adapter/encryption_admin_test.go b/adapter/encryption_admin_test.go index 659fb06eb..e7e097f2a 100644 --- a/adapter/encryption_admin_test.go +++ b/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 diff --git a/docs/design/2026_04_29_partial_data_at_rest_encryption.md b/docs/design/2026_04_29_partial_data_at_rest_encryption.md index 0d77a8784..2c53e0626 100644 --- a/docs/design/2026_04_29_partial_data_at_rest_encryption.md +++ b/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 | — | diff --git a/docs/design/2026_05_29_implemented_7c_confchange_time_registration.md b/docs/design/2026_05_29_implemented_7c_confchange_time_registration.md index 6e13370f9..84113a53e 100644 --- a/docs/design/2026_05_29_implemented_7c_confchange_time_registration.md +++ b/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. diff --git a/docs/design/2026_07_11_implemented_7d_sidecar_registry_projection.md b/docs/design/2026_07_11_implemented_7d_sidecar_registry_projection.md new file mode 100644 index 000000000..b84f51a9a --- /dev/null +++ b/docs/design/2026_07_11_implemented_7d_sidecar_registry_projection.md @@ -0,0 +1,52 @@ +# Implemented 7d: sidecar registry projection + +Status: Implemented +Author: bootjp +Date: 2026-07-11 + +## Scope + +This closes the remaining Stage 7 §5.5 recovery surface for +`WriterRegistryForCaller`. + +The prior Stage 7 slices already registered writers at process start, +storage-layer admission, runtime cutover/rotation, and membership +change time. The open gap was the compaction-fallback recovery RPC: +a node whose sidecar fell behind a compacted Raft log needs the +leader's per-DEK `last_seen_local_epoch` record for that specific +caller so it can choose a strictly monotonic replacement +`local_epoch`. + +## Implementation + +`EncryptionAdminServer` now accepts the production +`encryption.WriterRegistryStore` through +`WithEncryptionAdminWriterRegistry`. + +`GetSidecarState` projects registry rows for the local full node ID. +`ResyncSidecar` projects rows for `ResyncSidecarRequest.caller_full_node_id`. +Both responses keep returning a non-nil empty map when the registry is +not wired, preserving the encryption-disabled and test posture. + +For each DEK present in the sidecar, the server reads +`RegistryKey(dek_id, NodeID16(full_node_id))` and returns the decoded +`LastSeenLocalEpoch`. Missing rows are omitted instead of zero-filled, +because zero could be mistaken for a real registry observation. A +decoded row whose stored `FullNodeID` does not match the requested +caller is rejected as an internal node-ID collision. + +Production wiring stores the writer registry on each +`raftGroupRuntime` as the group is built, then passes it into the +per-shard `EncryptionAdmin` registration in `startRaftServers`. + +## Validation + +- `go test ./adapter -run 'TestEncryptionAdmin_(GetSidecarState|ResyncSidecar)' -count=1 -timeout=240s` +- `go test ./adapter -run TestEncryptionAdmin -count=1 -timeout=300s` +- `go test ./store ./internal/encryption . -count=1 -timeout=180s` +- `golangci-lint run ./adapter ./store ./internal/encryption . --timeout=5m` + +The broader `go test ./adapter ./store ./internal/encryption . -count=1 +-timeout=300s` run timed out in the full adapter package. The targeted +encryption admin tests and the directly affected non-adapter packages +passed. diff --git a/main.go b/main.go index 530dbf4b7..a20985bbe 100644 --- a/main.go +++ b/main.go @@ -1112,6 +1112,7 @@ func buildShardGroups( _ = st.Close() return nil, nil, errors.Wrapf(err, "failed to start raft group %d", g.id) } + runtime.writerRegistry = reg runtimes = append(runtimes, runtime) // Stage 6E-2c: route every shard group's TransactionManager // through NewLeaderProxyForShardGroup so the proposer chain @@ -1189,6 +1190,14 @@ func postCutoverProposerForRuntime(rt *raftGroupRuntime, shardGroups map[uint64] return proposerForGroup(rt, shardGroups) } +func writerRegistryForEncryptionAdmin(runtimes []*raftGroupRuntime, defaultGroup uint64) encryption.WriterRegistryStore { + defaultRuntime := findDefaultGroupRuntime(runtimes, defaultGroup) + if defaultRuntime == nil { + return nil + } + return defaultRuntime.writerRegistry +} + func appliedIndexForEngine(engine raftengine.Engine) func() uint64 { applied, ok := engine.(interface{ AppliedIndex() uint64 }) if !ok { @@ -1500,6 +1509,7 @@ func startServersAfterStartupRotation(waitRotateOnStartup startupRotationWaiter, encWiring: in.encWiring, redisApplyObserver: in.redisApplyObserver, dynamoAddress: *dynamoAddr, + defaultGroup: in.cfg.defaultGroup, leaderDynamo: in.cfg.leaderDynamo, s3Address: *s3Addr, leaderS3: in.cfg.leaderS3, @@ -2060,6 +2070,7 @@ func startRaftServers( forwardDeps adminForwardServerDeps, confChangeInterceptor internalraftadmin.MembershipChangeInterceptor, encWiring encryptionWriteWiring, + defaultGroup uint64, ) error { forwardLogger := slog.Default().With(slog.String("component", "admin")) // extraOptsCap reserves slots for the unary + stream admin interceptor @@ -2068,6 +2079,7 @@ func startRaftServers( const extraOptsCap = 2 enableMutators := encryptionMutatorsEnabled() encryptionCapabilityFanout := buildEncryptionCapabilityFanout(ctx, eg, runtimes, enableMutators) + adminWriterRegistry := writerRegistryForEncryptionAdmin(runtimes, defaultGroup) for _, rt := range runtimes { baseOpts := internalutil.GRPCServerOptions() opts := make([]grpc.ServerOption, 0, len(baseOpts)+extraOptsCap) @@ -2122,6 +2134,7 @@ func startRaftServers( enableMutators, rt.engine, encryptionCapabilityFanout, + adapter.WithEncryptionAdminWriterRegistry(adminWriterRegistry), adapter.WithEncryptionAdminLatestAppliedIndex(appliedIndexForEngine(rt.engine)), adapter.WithEncryptionAdminPostCutoverProposer(proposerForGroup(rt, shardGroups)), adapter.WithEncryptionAdminCutoverBarrier(encWiring.raftEnvelope.barrier()), @@ -2444,6 +2457,7 @@ type runtimeServerRunner struct { redisApplyObserver *adapter.RedisApplyObserver encWiring encryptionWriteWiring dynamoAddress string + defaultGroup uint64 leaderDynamo map[string]string s3Address string leaderS3 map[string]string @@ -2557,6 +2571,7 @@ func (r *runtimeServerRunner) startRaftTransport() error { forwardDeps, r.encryptionConfChangeInterceptor, r.encWiring, + r.defaultGroup, ); err != nil { return r.startupFailure(err) } diff --git a/main_encryption_admin_test.go b/main_encryption_admin_test.go index df98b0d1c..96a7743be 100644 --- a/main_encryption_admin_test.go +++ b/main_encryption_admin_test.go @@ -5,6 +5,7 @@ import ( "net" "testing" + "github.com/bootjp/elastickv/internal/encryption" "github.com/bootjp/elastickv/internal/raftengine" etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" pb "github.com/bootjp/elastickv/proto" @@ -41,6 +42,20 @@ func (stubEncryptionAdminEngine) LinearizableRead(context.Context) (uint64, erro return 0, nil } +type stubEncryptionAdminRegistry struct { + name string +} + +func (stubEncryptionAdminRegistry) GetRegistryRow([]byte) ([]byte, bool, error) { + return nil, false, nil +} + +func (stubEncryptionAdminRegistry) SetRegistryRow(_, _ []byte) error { + return nil +} + +var _ encryption.WriterRegistryStore = stubEncryptionAdminRegistry{} + // TestEncryptionAdminFullNodeID_DistinctPerRaftId pins the // PR760 r1 Codex P1 regression: full_node_id must be derived from // --raftId (per-node-stable) and NOT from the Raft group id @@ -65,6 +80,29 @@ func TestEncryptionAdminFullNodeID_DistinctPerRaftId(t *testing.T) { } } +func TestWriterRegistryForEncryptionAdminUsesDefaultGroup(t *testing.T) { + defaultRegistry := stubEncryptionAdminRegistry{name: "default"} + otherRegistry := stubEncryptionAdminRegistry{name: "other"} + runtimes := []*raftGroupRuntime{ + {spec: groupSpec{id: 2}, writerRegistry: otherRegistry}, + {spec: groupSpec{id: 1}, writerRegistry: defaultRegistry}, + } + + got := writerRegistryForEncryptionAdmin(runtimes, 1) + if got != defaultRegistry { + t.Fatalf("writerRegistryForEncryptionAdmin selected %#v, want default group registry %#v", got, defaultRegistry) + } +} + +func TestWriterRegistryForEncryptionAdminMissingDefaultGroup(t *testing.T) { + got := writerRegistryForEncryptionAdmin([]*raftGroupRuntime{ + {spec: groupSpec{id: 2}, writerRegistry: stubEncryptionAdminRegistry{name: "other"}}, + }, 1) + if got != nil { + t.Fatalf("writerRegistryForEncryptionAdmin selected %#v, want nil when default group runtime is missing", got) + } +} + // TestEncryptionAdmin_MutatingRPCRefusedWhenGateOff pins the // Stage 6B-2 double-gate. With either condition of the gate false // (enableMutators=false OR engine=nil), Proposer + LeaderView diff --git a/multiraft_runtime.go b/multiraft_runtime.go index 6c41fac80..c18443745 100644 --- a/multiraft_runtime.go +++ b/multiraft_runtime.go @@ -8,6 +8,7 @@ import ( "strings" "sync" + "github.com/bootjp/elastickv/internal/encryption" "github.com/bootjp/elastickv/internal/raftengine" "github.com/bootjp/elastickv/store" "github.com/cockroachdb/errors" @@ -27,8 +28,9 @@ type raftGroupRuntime struct { engineMu sync.RWMutex engine raftengine.Engine - store store.MVCCStore - stateMachine raftengine.StateMachine + store store.MVCCStore + writerRegistry encryption.WriterRegistryStore + stateMachine raftengine.StateMachine registerTransport func(grpc.ServiceRegistrar) closeFactory func() error // releases factory-created resources (transport, stores)