From 59b2e2d8f90f5d34f52a8c54d0f9cead5ab507cc Mon Sep 17 00:00:00 2001 From: pingqiu Date: Sun, 5 Apr 2026 16:15:54 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=2020=20T3=20=E2=80=94=20durabilit?= =?UTF-8?q?y-first=20V2=20promotion=20in=20real=20failover=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- weed/server/master_block_evidence.go | 1 + weed/server/master_block_failover.go | 62 ++++++++ weed/server/master_block_failover_test.go | 166 ++++++++++++++++++++++ weed/server/master_block_registry.go | 22 +++ weed/server/master_server.go | 2 + weed/server/volume_grpc_block_evidence.go | 1 + 6 files changed, 254 insertions(+) diff --git a/weed/server/master_block_evidence.go b/weed/server/master_block_evidence.go index 07309fe00..673d8b6ff 100644 --- a/weed/server/master_block_evidence.go +++ b/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 diff --git a/weed/server/master_block_failover.go b/weed/server/master_block_failover.go index f2d1efac4..44e5f2831 100644 --- a/weed/server/master_block_failover.go +++ b/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). diff --git a/weed/server/master_block_failover_test.go b/weed/server/master_block_failover_test.go index e89456182..29e4481aa 100644 --- a/weed/server/master_block_failover_test.go +++ b/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) + } +} diff --git a/weed/server/master_block_registry.go b/weed/server/master_block_registry.go index e47ecaa96..a98b63d2f 100644 --- a/weed/server/master_block_registry.go +++ b/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). // diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 5cd9f7b5d..a6756020c 100644 --- a/weed/server/master_server.go +++ b/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. diff --git a/weed/server/volume_grpc_block_evidence.go b/weed/server/volume_grpc_block_evidence.go index 3a1252f29..2a7234d62 100644 --- a/weed/server/volume_grpc_block_evidence.go +++ b/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