feat: Phase 20 T2 — promotion evidence semantics + selection substrate

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) <noreply@anthropic.com>

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.

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

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

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

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