feat: Phase 20 T3 — durability-first V2 promotion in real failover path

Wire V2 promotion into the real master failover decision path:
promoteReplica() now dispatches to promoteReplicaV2() when
blockV2Promotion flag is true. V2 path queries each candidate for
fresh evidence via pluggable BlockPromotionEvidenceQuerier, selects
by CommittedLSN (durability-first), and fail-closes when no eligible
candidate exists. No silent fallback to V1.

Feature flag: blockV2Promotion bool on MasterServer. When false,
existing promoteReplicaV1() (health-score-first) is used unchanged.
Flag is explicit and observable, not a hidden rescue path.

Registry: add PromoteReplicaByServer() for V2 path where master
already knows the winner. Clear stale EngineProjectionMode in
applyPromotionLocked (complements T1 turnover fix).

T2 fix: fail-closed when V2 core projection is absent —
Eligible=false with reason "missing_engine_projection". CommittedLSN
from core used unconditionally (no WALHeadLSN overstatement).

5 T3 integration tests: higher CommittedLSN wins, all-ineligible
fail-closed, evidence-failure fail-closed, flag-off uses legacy,
epoch bump + assignment enqueue only after selection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

weed/server/master_block_evidence.go

@@ -36,6 +36,7 @@ func queryAllCandidateEvidence(querier BlockPromotionEvidenceQuerier, candidates
 			errs = append(errs, fmt.Errorf("evidence query %s %s: %w", c.server, c.path, err))
 			continue
 		}
+		ev.Server = c.server
 		evidence = append(evidence, ev)
 	}
 	return evidence, errs

weed/server/master_block_failover.go

@@ -250,6 +250,8 @@ func (ms *MasterServer) failoverBlockVolumes(deadServer string) {
 
 // promoteReplica promotes the best replica to primary for the named volume,
 // enqueues an assignment for the new primary, and records a pending rebuild.
+// When blockV2Promotion is true, uses fresh on-demand evidence and
+// durability-first selection. When false, uses legacy health-score-first.
 func (ms *MasterServer) promoteReplica(volumeName string) {
 	entry, ok := ms.blockRegistry.Lookup(volumeName)
 	if !ok {
@@ -259,6 +261,16 @@ func (ms *MasterServer) promoteReplica(volumeName string) {
 		return
 	}
 
+	if ms.blockV2Promotion && ms.blockVSQueryEvidence != nil {
+		ms.promoteReplicaV2(volumeName, entry)
+		return
+	}
+	ms.promoteReplicaV1(volumeName, entry)
+}
+
+// promoteReplicaV1 is the legacy promotion path: health-score-first,
+// heartbeat-stale data, no fresh evidence query.
+func (ms *MasterServer) promoteReplicaV1(volumeName string, entry BlockVolumeEntry) {
 	oldPrimary := entry.VolumeServer
 	oldPath := entry.Path
 	oldPrimaryISCSIAddr := entry.ISCSIAddr
@@ -273,6 +285,56 @@ func (ms *MasterServer) promoteReplica(volumeName string) {
 	ms.finalizePromotion(volumeName, oldPrimary, oldPath, oldPrimaryISCSIAddr, newEpoch)
 }
 
+// promoteReplicaV2 queries each candidate for fresh evidence, selects by
+// CommittedLSN (durability-first), and fail-closes when no eligible candidate
+// exists. Does NOT silently fall back to V1 — if evidence fails, promotion
+// does not proceed.
+func (ms *MasterServer) promoteReplicaV2(volumeName string, entry BlockVolumeEntry) {
+	oldPrimary := entry.VolumeServer
+	oldPath := entry.Path
+	oldPrimaryISCSIAddr := entry.ISCSIAddr
+
+	// Collect candidates from registry membership.
+	var candidates []promotionCandidate
+	for _, ri := range entry.Replicas {
+		candidates = append(candidates, promotionCandidate{
+			server:        ri.Server,
+			path:          ri.Path,
+			expectedEpoch: entry.Epoch,
+		})
+	}
+	if len(candidates) == 0 {
+		glog.Warningf("failover V2: %q has no replica candidates", volumeName)
+		return
+	}
+
+	// Query each candidate for fresh evidence.
+	evidence, errs := queryAllCandidateEvidence(ms.blockVSQueryEvidence, candidates)
+	for _, err := range errs {
+		glog.Warningf("failover V2: %s", err)
+	}
+
+	// Durability-first selection. Fail-closed if no eligible candidate.
+	best, err := selectDurabilityFirstCandidate(evidence)
+	if err != nil {
+		glog.Warningf("failover V2: %q: %v (queried %d, errors %d)",
+			volumeName, err, len(candidates), len(errs))
+		return
+	}
+
+	// Apply promotion in registry using the selected server.
+	newEpoch, err := ms.blockRegistry.PromoteReplicaByServer(volumeName, best.Server)
+	if err != nil {
+		glog.Warningf("failover V2: PromoteReplicaByServer %q %s: %v", volumeName, best.Server, err)
+		return
+	}
+
+	glog.V(0).Infof("failover V2: %q selected %s (CommittedLSN=%d WALHeadLSN=%d HealthScore=%.2f)",
+		volumeName, best.Server, best.CommittedLSN, best.WALHeadLSN, best.HealthScore)
+
+	ms.finalizePromotion(volumeName, oldPrimary, oldPath, oldPrimaryISCSIAddr, newEpoch)
+}
+
 // finalizePromotion performs post-registry promotion steps:
 // enqueue assignment for new primary, record pending rebuild for old primary, bump metrics.
 // Called by both promoteReplica (auto) and blockVolumePromoteHandler (manual).

weed/server/master_block_failover_test.go

@@ -1178,3 +1178,169 @@ func TestT4_RebuildEmptyAddr_StillQueued(t *testing.T) {
 		t.Fatal("rebuild assignment should still be queued even with empty addr")
 	}
 }
+
+// T3 V2 promotion tests.
+
+func testMasterServerV2(t *testing.T, querier BlockPromotionEvidenceQuerier) *MasterServer {
+	t.Helper()
+	ms := testMasterServerForFailover(t)
+	ms.blockV2Promotion = true
+	ms.blockVSQueryEvidence = querier
+	return ms
+}
+
+func TestFailoverV2_HigherCommittedLSNWins(t *testing.T) {
+	querier := func(_ context.Context, server, path string, epoch uint64) (BlockPromotionEvidence, error) {
+		switch server {
+		case "vs2":
+			return BlockPromotionEvidence{
+				Server: "vs2", Path: path, Epoch: 1,
+				CommittedLSN: 10, WALHeadLSN: 10, HealthScore: 1.0,
+				EngineProjectionMode: "publish_healthy", Eligible: true,
+			}, nil
+		case "vs3":
+			return BlockPromotionEvidence{
+				Server: "vs3", Path: path, Epoch: 1,
+				CommittedLSN: 20, WALHeadLSN: 20, HealthScore: 0.5,
+				EngineProjectionMode: "publish_healthy", Eligible: true,
+			}, nil
+		}
+		return BlockPromotionEvidence{}, fmt.Errorf("unknown server %s", server)
+	}
+	ms := testMasterServerV2(t, querier)
+
+	// Register with two replicas.
+	ms.blockRegistry.MarkBlockCapable("vs1")
+	ms.blockRegistry.MarkBlockCapable("vs2")
+	ms.blockRegistry.MarkBlockCapable("vs3")
+	if err := ms.blockRegistry.Register(&BlockVolumeEntry{
+		Name: "vol-v2-lsn", VolumeServer: "vs1",
+		Path: "/data/vol-v2-lsn.blk", Epoch: 1,
+		Role: blockvol.RoleToWire(blockvol.RolePrimary), Status: StatusActive,
+		LeaseTTL: 1 * time.Second, LastLeaseGrant: time.Now().Add(-5 * time.Second),
+		Replicas: []ReplicaInfo{
+			{Server: "vs2", Path: "/data/vol-v2-lsn.blk", HealthScore: 1.0, Role: blockvol.RoleToWire(blockvol.RoleReplica), LastHeartbeat: time.Now()},
+			{Server: "vs3", Path: "/data/vol-v2-lsn.blk", HealthScore: 0.5, Role: blockvol.RoleToWire(blockvol.RoleReplica), LastHeartbeat: time.Now()},
+		},
+	}); err != nil {
+		t.Fatalf("register: %v", err)
+	}
+
+	ms.failoverBlockVolumes("vs1")
+
+	entry, ok := ms.blockRegistry.Lookup("vol-v2-lsn")
+	if !ok {
+		t.Fatal("entry missing")
+	}
+	// vs3 has higher CommittedLSN (20 > 10) despite lower health (0.5 < 1.0).
+	if entry.VolumeServer != "vs3" {
+		t.Fatalf("promoted %q, want vs3 (higher CommittedLSN)", entry.VolumeServer)
+	}
+	if entry.Epoch != 2 {
+		t.Fatalf("epoch=%d, want 2", entry.Epoch)
+	}
+	// Assignment should be enqueued.
+	pending := ms.blockAssignmentQueue.Peek("vs3")
+	if len(pending) == 0 {
+		t.Fatal("expected assignment enqueued for promoted vs3")
+	}
+}
+
+func TestFailoverV2_AllIneligible_NoPromotion(t *testing.T) {
+	querier := func(_ context.Context, server, path string, epoch uint64) (BlockPromotionEvidence, error) {
+		return BlockPromotionEvidence{
+			Server: server, Path: path, Epoch: 1,
+			CommittedLSN: 10, WALHeadLSN: 10,
+			EngineProjectionMode: "needs_rebuild", Eligible: false, Reason: "needs_rebuild",
+		}, nil
+	}
+	ms := testMasterServerV2(t, querier)
+	registerVolumeWithReplica(t, ms, "vol-v2-nopromo", "vs1", "vs2", 1, 1*time.Second)
+
+	ms.failoverBlockVolumes("vs1")
+
+	entry, ok := ms.blockRegistry.Lookup("vol-v2-nopromo")
+	if !ok {
+		t.Fatal("entry missing")
+	}
+	// No promotion should have occurred — fail-closed.
+	if entry.VolumeServer != "vs1" {
+		t.Fatalf("primary changed to %q, want vs1 (fail-closed, no eligible)", entry.VolumeServer)
+	}
+	if entry.Epoch != 1 {
+		t.Fatalf("epoch=%d, want 1 (unchanged)", entry.Epoch)
+	}
+}
+
+func TestFailoverV2_EvidenceQueryFailure_NoPromotion(t *testing.T) {
+	querier := func(_ context.Context, server, path string, epoch uint64) (BlockPromotionEvidence, error) {
+		return BlockPromotionEvidence{}, fmt.Errorf("connection refused")
+	}
+	ms := testMasterServerV2(t, querier)
+	registerVolumeWithReplica(t, ms, "vol-v2-fail", "vs1", "vs2", 1, 1*time.Second)
+
+	ms.failoverBlockVolumes("vs1")
+
+	entry, ok := ms.blockRegistry.Lookup("vol-v2-fail")
+	if !ok {
+		t.Fatal("entry missing")
+	}
+	// Evidence failure in V2 mode → no promotion, not silent V1 fallback.
+	if entry.VolumeServer != "vs1" {
+		t.Fatalf("primary changed to %q, want vs1 (fail-closed on evidence failure)", entry.VolumeServer)
+	}
+}
+
+func TestFailoverV2_FlagOff_UsesLegacy(t *testing.T) {
+	ms := testMasterServerForFailover(t)
+	ms.blockV2Promotion = false // explicitly off
+	registerVolumeWithReplica(t, ms, "vol-legacy", "vs1", "vs2", 1, 1*time.Second)
+
+	ms.failoverBlockVolumes("vs1")
+
+	entry, ok := ms.blockRegistry.Lookup("vol-legacy")
+	if !ok {
+		t.Fatal("entry missing")
+	}
+	// Legacy path should promote (V1 health-score-first).
+	if entry.VolumeServer != "vs2" {
+		t.Fatalf("promoted %q, want vs2 (legacy path)", entry.VolumeServer)
+	}
+	if entry.Epoch != 2 {
+		t.Fatalf("epoch=%d, want 2", entry.Epoch)
+	}
+}
+
+func TestFailoverV2_EpochBumpAndAssignmentOnlyAfterSelection(t *testing.T) {
+	querier := func(_ context.Context, server, path string, epoch uint64) (BlockPromotionEvidence, error) {
+		return BlockPromotionEvidence{
+			Server: server, Path: path, Epoch: 1,
+			CommittedLSN: 15, WALHeadLSN: 15, HealthScore: 1.0,
+			EngineProjectionMode: "replica_ready", Eligible: true,
+		}, nil
+	}
+	ms := testMasterServerV2(t, querier)
+	registerVolumeWithReplica(t, ms, "vol-v2-epoch", "vs1", "vs2", 1, 1*time.Second)
+
+	// Before failover: epoch=1, no pending assignments.
+	if pending := ms.blockAssignmentQueue.Peek("vs2"); len(pending) > 0 {
+		t.Fatal("unexpected pending assignments before failover")
+	}
+
+	ms.failoverBlockVolumes("vs1")
+
+	entry, ok := ms.blockRegistry.Lookup("vol-v2-epoch")
+	if !ok {
+		t.Fatal("entry missing")
+	}
+	if entry.Epoch != 2 {
+		t.Fatalf("epoch=%d, want 2 after promotion", entry.Epoch)
+	}
+	pending := ms.blockAssignmentQueue.Peek("vs2")
+	if len(pending) == 0 {
+		t.Fatal("expected assignment enqueued after successful selection")
+	}
+	if pending[0].Epoch != 2 {
+		t.Fatalf("assignment epoch=%d, want 2", pending[0].Epoch)
+	}
+}

weed/server/master_block_registry.go

@@ -1648,6 +1648,8 @@ func (r *BlockVolumeRegistry) applyPromotionLocked(entry *BlockVolumeEntry, name
 	entry.HasHeartbeatVolumeMode = false
 	entry.HeartbeatVolumeReason = ""
 	entry.HasHeartbeatVolumeReason = false
+	entry.EngineProjectionMode = ""
+	entry.HasEngineProjectionMode = false
 
 	// Remove promoted from Replicas. Others stay.
 	entry.Replicas = append(entry.Replicas[:candidateIdx], entry.Replicas[candidateIdx+1:]...)
@@ -1703,6 +1705,26 @@ func (r *BlockVolumeRegistry) PromoteBestReplica(name string) (uint64, error) {
 	return newEpoch, nil
 }
 
+// PromoteReplicaByServer promotes a specific replica identified by server
+// address to primary. Used by T3 V2 path where the master already selected
+// the winner via fresh evidence. Bypasses V1 eligibility gates since the
+// caller (V2 evidence path) owns eligibility determination.
+func (r *BlockVolumeRegistry) PromoteReplicaByServer(name, server string) (uint64, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	entry, ok := r.volumes[name]
+	if !ok {
+		return 0, fmt.Errorf("block volume %q not found", name)
+	}
+	for i, ri := range entry.Replicas {
+		if ri.Server == server {
+			newEpoch := r.applyPromotionLocked(entry, name, ri, i)
+			return newEpoch, nil
+		}
+	}
+	return 0, fmt.Errorf("block volume %q: replica server %q not found", name, server)
+}
+
 // evaluateManualPromotionLocked evaluates promotion candidates for a manual promote request.
 // Caller must hold r.mu (read or write).
 //

weed/server/master_server.go

@@ -109,7 +109,9 @@ type MasterServer struct {
 	blockVSPrepareExpand func(ctx context.Context, server string, name string, newSize, expandEpoch uint64) error
 	blockVSCommitExpand  func(ctx context.Context, server string, name string, expandEpoch uint64) (uint64, error)
 	blockVSCancelExpand  func(ctx context.Context, server string, name string, expandEpoch uint64) error
+	blockVSQueryEvidence BlockPromotionEvidenceQuerier // T2: fresh on-demand promotion evidence
 	nextExpandEpoch      atomic.Uint64
+	blockV2Promotion     bool // T3: when true, use durability-first V2 promotion; when false, legacy V1
 
 	// Test-only hook: called after AcquireExpandInflight but before the
 	// re-read Lookup in coordinated expand. Nil in production.

weed/server/volume_grpc_block_evidence.go

@@ -15,6 +15,7 @@ import (
 // semantics). The master only consumes this evidence; it does not own or
 // synthesize it.
 type BlockPromotionEvidence struct {
+	Server               string  // server address that produced this evidence
 	Path                 string
 	Epoch                uint64
 	CommittedLSN         uint64