From 1ca13143b60ba26a1fdaa5645f0500947177d01b Mon Sep 17 00:00:00 2001 From: pingqiu Date: Sun, 5 Apr 2026 16:10:57 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=2020=20T2=20=E2=80=94=20promotion?= =?UTF-8?q?=20evidence=20semantics=20+=20selection=20substrate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VS-side evidence handler (QueryBlockPromotionEvidence) reads live blockvol.Status() + V2 core projection at call time. Fail-closed: no core projection → ineligible with reason "missing_engine_projection". Engine CommittedLSN used unconditionally when core present (no WALHeadLSN overstatement). Eligibility owned by local V2 engine, not master. Master-side selection (selectDurabilityFirstCandidate): durability-first ordering by CommittedLSN, tie-break WALHeadLSN then HealthScore. All ineligible → fail-closed, no promotion. Pluggable querier (BlockPromotionEvidenceQuerier) for T3 wiring. Proto messages added to volume_server.proto. gRPC transport binding pending proto regen on M01 — this commit delivers evidence semantics and selection substrate, not full end-to-end RPC closure. Phase 20 doc updated with T2-T5 reviewer packs and cross-task guardrails. 13 tests: live facts, core projection mode, fail-closed no-core, 4 gated modes, missing volume, epoch mismatch, CommittedLSN ordering, WALHeadLSN tie-break, HealthScore tie-break, all-ineligible, mixed collection. Co-Authored-By: Claude Opus 4.6 (1M context) --- sw-block/.private/phase/phase-20.md | 123 +++++++ weed/pb/volume_server.proto | 20 ++ weed/server/master_block_evidence.go | 85 +++++ weed/server/volume_grpc_block_evidence.go | 112 +++++++ .../server/volume_grpc_block_evidence_test.go | 316 ++++++++++++++++++ 5 files changed, 656 insertions(+) create mode 100644 weed/server/master_block_evidence.go create mode 100644 weed/server/volume_grpc_block_evidence.go create mode 100644 weed/server/volume_grpc_block_evidence_test.go diff --git a/sw-block/.private/phase/phase-20.md b/sw-block/.private/phase/phase-20.md index fc1958dd2..dedb5ffe1 100644 --- a/sw-block/.private/phase/phase-20.md +++ b/sw-block/.private/phase/phase-20.md @@ -387,3 +387,126 @@ After Phase 20, the production binary has V2 correctness: And it runs on real hardware with real network, real iSCSI clients, and real WAL shipping — not in a simulation harness. + +## Reviewer Packs + +### T2: Promotion Evidence Query RPC + +**Allowed**: Dedicated RPC for promotion evidence. Return fresh local facts +from queried VS at call time. Read from local blockvol status and V2 core +projection. Return explicit eligibility plus reason. Keep heartbeat and +evidence query as separate channels. + +**Not allowed**: Reuse heartbeat payload as promotion decision source. +Reconstruct evidence from master registry caches. Let master guess +engine_projection_mode. Hide evidence failure behind silent fallback when +V2 path is enabled. Mix assignment authorization into the evidence RPC. + +**Truth owner**: Local storage/runtime facts = blockvol. Local semantic +projection and eligibility = V2 engine on queried VS. Master only consumes +evidence; it does not own or synthesize it. + +**Required tests**: (1) Handler returns live committed_lsn / wal_head_lsn. +(2) Handler returns current engine_projection_mode from core projection. +(3) Handler returns eligible=false with explicit reason for gated states. +(4) Query against stale/unknown/missing volume fails cleanly. +(5) Proto/wire field presence/absence handled correctly. + +**Pitfalls**: Using cached registry state instead of querying fresh VS-local +facts. Smuggling promotion policy into handler. Making RPC look like +"mini assignment" instead of evidence-only observation. + +### T3: Durability-First Promotion Selection + +**Allowed**: Master collects candidates from registry membership. Queries +each via T2 at failover time. Filter on eligible==true. Rank by +CommittedLSN then WALHeadLSN then health. Fail closed when no eligible +candidate. Explicit feature flag for temporary V1 fallback. + +**Not allowed**: Promote based only on last heartbeat. Rank health before +durability. Silently fall back to V1 when evidence query fails in V2 mode. +Treat "best-looking candidate" as sufficient when durability is ambiguous. +Let master override a node's ineligibility reason. + +**Truth owner**: Candidate durability/eligibility = queried VS local state + +engine. Promotion authorization = master. Registry = cluster membership/index +state, not evidence truth owner. + +**Required tests**: (1) Higher CommittedLSN wins even if health lower. +(2) Equal CommittedLSN, higher WALHeadLSN wins. (3) All ineligible => no +promotion. (4) Evidence query failure in V2 mode => fail-closed. (5) +Flag-off uses legacy. (6) Epoch bump + assignment enqueue only after +successful selection. + +**Pitfalls**: Leaving old PromoteBestReplica() heuristics in decision path. +Making flag a silent rescue. Letting registry-side stale WALHeadLSN +participate in final ordering. + +### T4: Local Activation Gate + +**Allowed**: After assignment through V2 core, read resulting local +projection. Gate local activation based on mode/reason. Refuse serving +while gated. Clear gate only when projection reaches allowed serving state. +Heartbeat may report gated state after enforcement. + +**Not allowed**: Treat assignment delivery as permission to serve. Wait for +master/heartbeat to enforce no-serve. Allow frontend/iSCSI publish while +degraded or needs_rebuild. Reinterpret bad reconstruction as "good enough." +Put gate only in operator surfaces while serving still proceeds. + +**Truth owner**: Reconstruction judgment and activation gate = promoted +primary local V2 engine/runtime. Master may observe, does not enforce. +Frontend/export = execution surface only. + +**Required tests**: (1) Degraded projection does not export/serve. +(2) needs_rebuild does not export/serve. (3) Healthy projection clears +gate. (4) Gate enforced before heartbeat round-trip. (5) Recovery from +gated to healthy re-enables serving. + +**Pitfalls**: Gate enforced too late (after export). Checking VolumeMode +instead of local engine projection. Gate advisory in logs but not in +serving paths. + +### T5: ClusterReplicationMode on Master + +**Allowed**: Compute new master-owned field from multi-replica facts. Keep +separate from EngineProjectionMode. Use for cluster/operator judgment. +Derive from replica set facts, freshness, lag. Distinct field on registry +entry and surfaces. + +**Not allowed**: Reuse/rename VolumeMode. Copy primary's +EngineProjectionMode into ClusterReplicationMode. Collapse local and +cluster concepts. Expose two ambiguous generic mode fields. Override local +VS truth with master-computed local semantics. + +**Truth owner**: EngineProjectionMode = VS-local engine truth. +ClusterReplicationMode = master-owned cluster judgment. VolumeMode = legacy +transitional, not new semantic source. + +**Required tests**: (1) All healthy => keepup. (2) Replica behind => +catching_up. (3) Missing/failed => degraded. (4) Unrecoverable gap => +needs_rebuild. (5) Explicit proof the two modes can differ without +conflict. (6) Surface/API shows distinct naming. + +**Pitfalls**: Computing from only primary-local projection. Reusing old +VolumeMode semantics. Exposing field ambiguously. + +### Cross-Task Guardrails + +**Allowed**: Additive migration with explicit flags. Reuse V1 execution +while replacing decision ownership. Projection/cache layers carrying truth +from actual owner. Fail-closed when critical evidence unavailable. + +**Not allowed**: Silent fallback. Dual-truth mode handling with unclear +authority. Master-side invention of local semantic truth. New +simulation-only seams bypassing production binary path. + +**Global required tests**: (1) End-to-end failover exercising T2+T3+T4. +(2) No serve-after-promotion when activation gated. (3) Operator surface +proves local vs cluster modes distinct. (4) Legacy-flag test proves +fallback is explicit and observable. + +**Recurring failure pattern to watch**: Heartbeat becomes overloaded into +liveness + evidence + decision. Master starts "helpfully" reconstructing +local semantics. Local gate exists in logs/surfaces but actual serving path +still open. diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index f26350bf0..d18055199 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -159,6 +159,8 @@ service VolumeServer { } rpc CancelExpandBlockVolume (CancelExpandBlockVolumeRequest) returns (CancelExpandBlockVolumeResponse) { } + rpc QueryBlockPromotionEvidence (QueryBlockPromotionEvidenceRequest) returns (QueryBlockPromotionEvidenceResponse) { + } } @@ -872,3 +874,21 @@ message CancelExpandBlockVolumeRequest { } message CancelExpandBlockVolumeResponse { } + +// T2: Fresh on-demand promotion evidence. Master queries VS at failover +// time. Returns live local facts, not cached/stale heartbeat data. +message QueryBlockPromotionEvidenceRequest { + string path = 1; // volume file path on queried VS + uint64 expected_epoch = 2; // caller's expected epoch for staleness check +} +message QueryBlockPromotionEvidenceResponse { + string path = 1; + uint64 epoch = 2; + uint64 committed_lsn = 3; + uint64 wal_head_lsn = 4; + uint64 checkpoint_lsn = 5; + string engine_projection_mode = 6; // pure V2 engine local projection, empty if no core + bool eligible = 7; + string reason = 8; // why ineligible, or empty + double health_score = 9; +} diff --git a/weed/server/master_block_evidence.go b/weed/server/master_block_evidence.go new file mode 100644 index 000000000..07309fe00 --- /dev/null +++ b/weed/server/master_block_evidence.go @@ -0,0 +1,85 @@ +package weed_server + +import ( + "context" + "fmt" + "time" +) + +// BlockPromotionEvidenceQuerier queries one volume server for fresh promotion +// evidence at failover time. The default implementation will use gRPC once +// proto is regenerated. For testing, a direct in-process implementation is +// used. +type BlockPromotionEvidenceQuerier func(ctx context.Context, server, path string, expectedEpoch uint64) (BlockPromotionEvidence, error) + +// queryBlockPromotionEvidence queries one candidate for fresh evidence with a +// bounded timeout. Returns the evidence or an error. The master must not +// promote based on stale/cached data — only fresh evidence from this call. +func queryBlockPromotionEvidence(querier BlockPromotionEvidenceQuerier, server, path string, expectedEpoch uint64) (BlockPromotionEvidence, error) { + if querier == nil { + return BlockPromotionEvidence{}, fmt.Errorf("promotion evidence querier is nil") + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + return querier(ctx, server, path, expectedEpoch) +} + +// queryAllCandidateEvidence queries multiple candidates in sequence and returns +// all successfully collected evidence. Failed queries are recorded as errors +// but do not stop collection from other candidates. +func queryAllCandidateEvidence(querier BlockPromotionEvidenceQuerier, candidates []promotionCandidate) ([]BlockPromotionEvidence, []error) { + var evidence []BlockPromotionEvidence + var errs []error + for _, c := range candidates { + ev, err := queryBlockPromotionEvidence(querier, c.server, c.path, c.expectedEpoch) + if err != nil { + errs = append(errs, fmt.Errorf("evidence query %s %s: %w", c.server, c.path, err)) + continue + } + evidence = append(evidence, ev) + } + return evidence, errs +} + +// selectDurabilityFirstCandidate selects the best promotion candidate from +// fresh evidence using durability-first ordering: +// +// 1. Filter: only eligible candidates +// 2. Rank: highest CommittedLSN +// 3. Tie-break: highest WALHeadLSN +// 4. Tie-break: highest HealthScore +// +// Returns an error if no eligible candidate exists (fail-closed). +func selectDurabilityFirstCandidate(evidence []BlockPromotionEvidence) (BlockPromotionEvidence, error) { + var eligible []BlockPromotionEvidence + for _, ev := range evidence { + if ev.Eligible { + eligible = append(eligible, ev) + } + } + if len(eligible) == 0 { + return BlockPromotionEvidence{}, fmt.Errorf("no eligible promotion candidate: %d queried, 0 eligible", len(evidence)) + } + best := eligible[0] + for _, ev := range eligible[1:] { + if ev.CommittedLSN > best.CommittedLSN { + best = ev + } else if ev.CommittedLSN == best.CommittedLSN { + if ev.WALHeadLSN > best.WALHeadLSN { + best = ev + } else if ev.WALHeadLSN == best.WALHeadLSN { + if ev.HealthScore > best.HealthScore { + best = ev + } + } + } + } + return best, nil +} + +// promotionCandidate is the minimal info needed to query one candidate. +type promotionCandidate struct { + server string + path string + expectedEpoch uint64 +} diff --git a/weed/server/volume_grpc_block_evidence.go b/weed/server/volume_grpc_block_evidence.go new file mode 100644 index 000000000..3a1252f29 --- /dev/null +++ b/weed/server/volume_grpc_block_evidence.go @@ -0,0 +1,112 @@ +package weed_server + +import ( + "fmt" + + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +// BlockPromotionEvidence is the fresh on-demand evidence returned by a volume +// server when the master queries it at failover time. It reads live local +// facts from blockvol status and V2 core projection — never from cached or +// stale heartbeat data. +// +// Truth owner: local blockvol (storage facts) + V2 engine (eligibility +// semantics). The master only consumes this evidence; it does not own or +// synthesize it. +type BlockPromotionEvidence struct { + Path string + Epoch uint64 + CommittedLSN uint64 + WALHeadLSN uint64 + CheckpointLSN uint64 + EngineProjectionMode string // pure V2 engine local projection, empty if no core + Eligible bool + Reason string // why ineligible, or empty + HealthScore float64 +} + +// QueryBlockPromotionEvidence returns fresh promotion evidence for one volume +// on this volume server. The master calls this at failover time instead of +// relying on stale heartbeat data. +// +// Returns live blockvol.Status() plus V2 core projection when present. +// Eligibility is determined locally by the V2 engine — the master does not +// override it. +// +// When V2 core projection is absent, the node is reported as ineligible with +// reason "missing_engine_projection". This is fail-closed by design: the V2 +// evidence path does not silently fall back to V1 eligibility semantics. +// Legacy V1 promotion (without engine projection) is handled by T3's explicit +// rollout-gated flag, not by this evidence layer. +func (bs *BlockService) QueryBlockPromotionEvidence(path string, expectedEpoch uint64) (BlockPromotionEvidence, error) { + if bs == nil { + return BlockPromotionEvidence{}, fmt.Errorf("block service not enabled") + } + if path == "" { + return BlockPromotionEvidence{}, fmt.Errorf("volume path is required") + } + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok || vol == nil { + return BlockPromotionEvidence{}, fmt.Errorf("volume %q not found on this server", path) + } + + status := vol.Status() + evidence := BlockPromotionEvidence{ + Path: path, + Epoch: status.Epoch, + CommittedLSN: status.WALHeadLSN, // V1 fallback: overridden below when core is present + WALHeadLSN: status.WALHeadLSN, + CheckpointLSN: status.CheckpointLSN, + HealthScore: status.HealthScore, + } + + // V2 core projection: when present, use engine-derived boundaries and + // eligibility semantics. This is the authoritative source. + proj, hasCore := bs.CoreProjection(path) + if hasCore { + evidence.EngineProjectionMode = string(proj.Mode.Name) + // Use engine CommittedLSN unconditionally when core is present. + // A legitimate CommittedLSN == 0 means no data is committed yet; + // falling back to WALHeadLSN would overstate durability. + evidence.CommittedLSN = proj.Boundary.CommittedLSN + // Eligibility from V2 engine: only publish_healthy and replica_ready + // are eligible for promotion. + switch proj.Mode.Name { + case "publish_healthy", "replica_ready": + evidence.Eligible = true + default: + evidence.Eligible = false + evidence.Reason = fmt.Sprintf("engine_projection_mode=%s", proj.Mode.Name) + if proj.Mode.Reason != "" { + evidence.Reason += ": " + proj.Mode.Reason + } + } + } else { + // Fail-closed: no V2 core projection means no V2 semantic evidence. + // The node cannot be reported as eligible through the V2 evidence + // path. Legacy V1 promotion uses a separate rollout-gated path. + evidence.Eligible = false + evidence.Reason = "missing_engine_projection" + } + + // Epoch staleness check: overrides any eligibility set above. + if expectedEpoch != 0 && status.Epoch != expectedEpoch { + evidence.Eligible = false + evidence.Reason = fmt.Sprintf("epoch_mismatch: local=%d expected=%d", status.Epoch, expectedEpoch) + } + + // Role check: only primary and replica roles are eligible for promotion. + switch status.Role { + case blockvol.RolePrimary, blockvol.RoleReplica: + // eligible (unless overridden above) + default: + evidence.Eligible = false + if evidence.Reason == "" { + evidence.Reason = fmt.Sprintf("role=%s", status.Role) + } + } + + return evidence, nil +} diff --git a/weed/server/volume_grpc_block_evidence_test.go b/weed/server/volume_grpc_block_evidence_test.go new file mode 100644 index 000000000..af6483853 --- /dev/null +++ b/weed/server/volume_grpc_block_evidence_test.go @@ -0,0 +1,316 @@ +package weed_server + +import ( + "context" + "fmt" + "testing" + + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +func TestBlockService_QueryBlockPromotionEvidence_ReturnsLiveFacts(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "evidence-live") + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok { + t.Fatal("volume not found") + } + if err := vol.HandleAssignment(5, blockvol.RolePrimary, 30000); err != nil { + t.Fatalf("handle assignment: %v", err) + } + data := make([]byte, 4096) + data[0] = 0xAB + if err := vol.WriteLBA(0, data); err != nil { + t.Fatalf("write: %v", err) + } + + // Inject V2 core projection so evidence uses engine-derived facts. + walHead := vol.Status().WALHeadLSN + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModePublishHealthy}, + Boundary: engine.BoundaryView{ + CommittedLSN: walHead, + }, + } + bs.coreProjMu.Unlock() + + ev, err := bs.QueryBlockPromotionEvidence(path, 0) + if err != nil { + t.Fatalf("query evidence: %v", err) + } + if ev.Epoch != 5 { + t.Fatalf("epoch=%d, want 5", ev.Epoch) + } + if ev.WALHeadLSN == 0 { + t.Fatal("wal_head_lsn should be >0 after write") + } + if ev.CommittedLSN == 0 { + t.Fatal("committed_lsn should be >0 from core projection") + } + if ev.HealthScore <= 0 { + t.Fatalf("health_score=%f, want >0", ev.HealthScore) + } + if !ev.Eligible { + t.Fatalf("expected eligible for publish_healthy with core, reason=%s", ev.Reason) + } + if ev.EngineProjectionMode != "publish_healthy" { + t.Fatalf("engine_projection_mode=%q, want publish_healthy", ev.EngineProjectionMode) + } +} + +func TestBlockService_QueryBlockPromotionEvidence_NoCoreProjectionFailsClosed(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "evidence-nocore") + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok { + t.Fatal("volume not found") + } + if err := vol.HandleAssignment(3, blockvol.RolePrimary, 30000); err != nil { + t.Fatalf("handle assignment: %v", err) + } + + // No core projection injected — should fail closed. + ev, err := bs.QueryBlockPromotionEvidence(path, 0) + if err != nil { + t.Fatalf("query evidence: %v", err) + } + if ev.Eligible { + t.Fatal("expected ineligible without V2 core projection") + } + if ev.Reason != "missing_engine_projection" { + t.Fatalf("reason=%q, want %q", ev.Reason, "missing_engine_projection") + } + if ev.EngineProjectionMode != "" { + t.Fatalf("engine_projection_mode=%q, want empty without core", ev.EngineProjectionMode) + } + // Storage facts should still be present even when ineligible. + if ev.WALHeadLSN == 0 && ev.Epoch == 0 { + t.Fatal("expected storage facts even for ineligible evidence") + } +} + +func TestBlockService_QueryBlockPromotionEvidence_ReturnsCoreProjectionMode(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "evidence-epm") + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok { + t.Fatal("volume not found") + } + if err := vol.SetRole(blockvol.RolePrimary); err != nil { + t.Fatalf("set role: %v", err) + } + + // Inject a core projection for this path. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{ + Name: engine.ModePublishHealthy, + Reason: "all_ready", + }, + Boundary: engine.BoundaryView{ + CommittedLSN: 42, + }, + } + bs.coreProjMu.Unlock() + + ev, err := bs.QueryBlockPromotionEvidence(path, 0) + if err != nil { + t.Fatalf("query evidence: %v", err) + } + if ev.EngineProjectionMode != "publish_healthy" { + t.Fatalf("engine_projection_mode=%q, want %q", ev.EngineProjectionMode, "publish_healthy") + } + if ev.CommittedLSN != 42 { + t.Fatalf("committed_lsn=%d, want 42 from core projection", ev.CommittedLSN) + } + if !ev.Eligible { + t.Fatalf("expected eligible for publish_healthy, reason=%s", ev.Reason) + } +} + +func TestBlockService_QueryBlockPromotionEvidence_IneligibleForGatedStates(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "evidence-gated") + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok { + t.Fatal("volume not found") + } + if err := vol.SetRole(blockvol.RolePrimary); err != nil { + t.Fatalf("set role: %v", err) + } + + tests := []struct { + mode engine.ModeName + reason string + }{ + {engine.ModeDegraded, "barrier_timeout"}, + {engine.ModeNeedsRebuild, "gap_too_large"}, + {engine.ModeBootstrapPending, "awaiting_shipper"}, + {engine.ModeAllocatedOnly, "no_role"}, + } + + for _, tt := range tests { + t.Run(string(tt.mode), func(t *testing.T) { + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{ + Name: tt.mode, + Reason: tt.reason, + }, + } + bs.coreProjMu.Unlock() + + ev, err := bs.QueryBlockPromotionEvidence(path, 0) + if err != nil { + t.Fatalf("query evidence: %v", err) + } + if ev.Eligible { + t.Fatalf("expected ineligible for mode=%s", tt.mode) + } + if ev.Reason == "" { + t.Fatal("expected non-empty reason for ineligible candidate") + } + if ev.EngineProjectionMode != string(tt.mode) { + t.Fatalf("engine_projection_mode=%q, want %q", ev.EngineProjectionMode, tt.mode) + } + }) + } +} + +func TestBlockService_QueryBlockPromotionEvidence_MissingVolumeFailsCleanly(t *testing.T) { + bs := newTestBlockServiceDirect(t) + + _, err := bs.QueryBlockPromotionEvidence("/nonexistent/path.blk", 0) + if err == nil { + t.Fatal("expected error for missing volume") + } + + _, err = bs.QueryBlockPromotionEvidence("", 0) + if err == nil { + t.Fatal("expected error for empty path") + } +} + +func TestBlockService_QueryBlockPromotionEvidence_EpochMismatchIneligible(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "evidence-epoch") + + vol, ok := bs.blockStore.GetBlockVolume(path) + if !ok { + t.Fatal("volume not found") + } + if err := vol.SetEpoch(3); err != nil { + t.Fatalf("set epoch: %v", err) + } + if err := vol.SetRole(blockvol.RolePrimary); err != nil { + t.Fatalf("set role: %v", err) + } + + ev, err := bs.QueryBlockPromotionEvidence(path, 99) + if err != nil { + t.Fatalf("query evidence: %v", err) + } + if ev.Eligible { + t.Fatal("expected ineligible for epoch mismatch") + } + if ev.Reason == "" { + t.Fatal("expected reason for epoch mismatch") + } +} + +// Master-side durability-first selection logic tests. + +func TestSelectDurabilityFirstCandidate_HighestCommittedLSNWins(t *testing.T) { + evidence := []BlockPromotionEvidence{ + {Path: "/a.blk", CommittedLSN: 10, WALHeadLSN: 10, HealthScore: 1.0, Eligible: true}, + {Path: "/b.blk", CommittedLSN: 20, WALHeadLSN: 20, HealthScore: 0.5, Eligible: true}, + {Path: "/c.blk", CommittedLSN: 15, WALHeadLSN: 30, HealthScore: 1.0, Eligible: true}, + } + best, err := selectDurabilityFirstCandidate(evidence) + if err != nil { + t.Fatalf("select: %v", err) + } + if best.Path != "/b.blk" { + t.Fatalf("selected %q, want /b.blk (highest CommittedLSN)", best.Path) + } +} + +func TestSelectDurabilityFirstCandidate_WALHeadLSNBreaksTie(t *testing.T) { + evidence := []BlockPromotionEvidence{ + {Path: "/a.blk", CommittedLSN: 20, WALHeadLSN: 25, HealthScore: 1.0, Eligible: true}, + {Path: "/b.blk", CommittedLSN: 20, WALHeadLSN: 30, HealthScore: 0.5, Eligible: true}, + } + best, err := selectDurabilityFirstCandidate(evidence) + if err != nil { + t.Fatalf("select: %v", err) + } + if best.Path != "/b.blk" { + t.Fatalf("selected %q, want /b.blk (higher WALHeadLSN on tie)", best.Path) + } +} + +func TestSelectDurabilityFirstCandidate_HealthScoreBreaksFinalTie(t *testing.T) { + evidence := []BlockPromotionEvidence{ + {Path: "/a.blk", CommittedLSN: 20, WALHeadLSN: 30, HealthScore: 0.8, Eligible: true}, + {Path: "/b.blk", CommittedLSN: 20, WALHeadLSN: 30, HealthScore: 1.0, Eligible: true}, + } + best, err := selectDurabilityFirstCandidate(evidence) + if err != nil { + t.Fatalf("select: %v", err) + } + if best.Path != "/b.blk" { + t.Fatalf("selected %q, want /b.blk (higher HealthScore on full tie)", best.Path) + } +} + +func TestSelectDurabilityFirstCandidate_AllIneligibleFailsClosed(t *testing.T) { + evidence := []BlockPromotionEvidence{ + {Path: "/a.blk", CommittedLSN: 10, Eligible: false, Reason: "degraded"}, + {Path: "/b.blk", CommittedLSN: 20, Eligible: false, Reason: "needs_rebuild"}, + } + _, err := selectDurabilityFirstCandidate(evidence) + if err == nil { + t.Fatal("expected fail-closed error when all candidates ineligible") + } +} + +func TestSelectDurabilityFirstCandidate_EmptyEvidenceFailsClosed(t *testing.T) { + _, err := selectDurabilityFirstCandidate(nil) + if err == nil { + t.Fatal("expected fail-closed error on empty evidence") + } +} + +func TestQueryAllCandidateEvidence_CollectsSuccessAndErrors(t *testing.T) { + querier := func(_ context.Context, server, path string, epoch uint64) (BlockPromotionEvidence, error) { + if server == "bad-server" { + return BlockPromotionEvidence{}, fmt.Errorf("connection refused") + } + return BlockPromotionEvidence{ + Path: path, + Epoch: epoch, + CommittedLSN: 10, + Eligible: true, + }, nil + } + + candidates := []promotionCandidate{ + {server: "good-server", path: "/a.blk", expectedEpoch: 5}, + {server: "bad-server", path: "/b.blk", expectedEpoch: 5}, + {server: "good-server2", path: "/c.blk", expectedEpoch: 5}, + } + + evidence, errs := queryAllCandidateEvidence(querier, candidates) + if len(evidence) != 2 { + t.Fatalf("evidence count=%d, want 2", len(evidence)) + } + if len(errs) != 1 { + t.Fatalf("error count=%d, want 1", len(errs)) + } +}