diff --git a/sw-block/.private/phase/phase-17-log.md b/sw-block/.private/phase/phase-17-log.md index 33aa9eebb..326d3c2c6 100644 --- a/sw-block/.private/phase/phase-17-log.md +++ b/sw-block/.private/phase/phase-17-log.md @@ -369,3 +369,119 @@ Open next seam: 1. no additional ownership move is currently justified on the readiness-state path + +--- + +### `17E` Logging Format Note + +Date: 2026-04-05 +Scope: bounded failover-completion evidence loop + +Use this section format for every `17E` run summary. + +Intent: + +1. keep `Phase 17` as the main semantic/product boundary home +2. keep each run summary short in the phase log +3. move full evidence details into a dedicated result document +4. make the next action explicit after every run + +Required entry shape: + +### `17E` Run `#N` Summary + +Date: +Scenario: +Commit / binary: +Environment: +Classification: + +Allowed classification values: + +1. `pure V2 core evidence` +2. `integrated runtime under V2 semantics` + +Result: + +1. `PASS` +2. `FAIL` +3. `PARTIAL` + +Key finding: + +1. one sentence only +2. state the bounded semantic conclusion, not only the symptom + +What this run proves: + +1. keep to one or two bounded points + +What this run does NOT prove: + +1. keep exclusions explicit + +Result document: + +1. reference one dedicated result md + +Next action: + +1. exact next step +2. owner + +Current recommended result-doc template: + +1. `learn/test/phase-17e-run-result-template.md` + +Recommended usage note: + +1. if a run only proves `primary changed + I/O resumed`, log it as `PARTIAL` +2. if historical readback was not reached, say exactly where the run stopped +3. if the finding is about live `weed/server` + `blockvol`, classify it as + `integrated runtime under V2 semantics`, not as pure `V2 core` + +--- + +### `17E` Run `#1` Summary + +Date: 2026-04-05 +Scenario: `internal/recovery-baseline-failover` +Commit / binary: exact binary identity not yet pinned from the returned run +bundle +Environment: Windows launcher with `sw-test-runner` SSH orchestration to Linux +`m01` / `m02` +Classification: `integrated runtime under V2 semantics` + +Result: + +1. `FAIL` + +Key finding: + +1. `wait_volume_healthy` is more truthful now, but `block_promote + + wait_volume_healthy` still does not guarantee immediate `sync_all` + barrier-ready writes on the promoted primary + +What this run proves: + +1. the runner now exposes the bootstrap/publish transition more honestly before + declaring healthy +2. the current integrated runtime still has a post-promote stability gap that + can surface before the intended auto-failover evidence section begins + +What this run does NOT prove: + +1. it does not yet prove auto-failover historical-read continuity +2. it does not yet prove that the upgraded failover scenario bundle is green on + the chosen path + +Result document: + +1. `learn/test/phase-17e-run-01-recovery-baseline-failover-2026-04-05.md` + +Next action: + +1. collect the remote bundle and node logs, then separate setup-promote + stability from the auto-failover baseline so the next run can test the real + failover continuity claim +2. owner: shared diff --git a/sw-block/design/README.md b/sw-block/design/README.md index 52cb9609e..a99590b47 100644 --- a/sw-block/design/README.md +++ b/sw-block/design/README.md @@ -7,7 +7,14 @@ Historical planning/review documents were moved to `../docs/archive/design/` to ## Read First - `v2-protocol-truths.md` +- `v2-capability-map.md` +- `v2-pure-runtime-rf1-bootstrap.md` +- `v2-volumev2-single-node-mvp.md` +- `v2-proof-and-retest-pyramid.md` - `v2-protocol-claim-and-evidence.md` +- `v2-two-loop-protocol.md` +- `v2-automata-ownership-map.md` +- `v2-loop1-surface-draft.md` - `v2-product-completion-overview.md` - `v2-phase-development-plan.md` - `v2-semantic-methodology.zh.md` diff --git a/sw-block/design/v2-product-completion-overview.md b/sw-block/design/v2-product-completion-overview.md index 5dcd5665c..3e385ae05 100644 --- a/sw-block/design/v2-product-completion-overview.md +++ b/sw-block/design/v2-product-completion-overview.md @@ -23,8 +23,9 @@ This document is the product-completion view. It complements: 1. `v2-protocol-truths.md` for accepted semantics -2. `v2-phase-development-plan.md` for the current phase ladder -3. `../.private/phase/phase-16.md` for the active bounded runtime-closure contract +2. `v2-capability-map.md` for the capability-expansion and test-closure view +3. `v2-phase-development-plan.md` for the current phase ladder +4. `../.private/phase/phase-16.md` for the active bounded runtime-closure contract ## Current Position diff --git a/weed/server/master_block_assignment_queue.go b/weed/server/master_block_assignment_queue.go index d55e83cdf..a24604482 100644 --- a/weed/server/master_block_assignment_queue.go +++ b/weed/server/master_block_assignment_queue.go @@ -84,7 +84,9 @@ func (q *BlockAssignmentQueue) Confirm(server string, path string, epoch uint64) } // ConfirmFromHeartbeat batch-confirms assignments that match reported heartbeat info. -// An assignment is confirmed if the VS reports (path, epoch) that matches. +// Same-epoch refresh assignments that carry replica transport are only confirmed +// once the heartbeat reflects that transport, so they are not dropped before +// the promoted VS actually applies them. func (q *BlockAssignmentQueue) ConfirmFromHeartbeat(server string, infos []blockvol.BlockVolumeInfoMessage) { if len(infos) == 0 { return @@ -97,26 +99,42 @@ func (q *BlockAssignmentQueue) ConfirmFromHeartbeat(server string, infos []block return } - // Build a set of reported (path, epoch) pairs. - type key struct { - path string - epoch uint64 - } - reported := make(map[key]bool, len(infos)) - for _, info := range infos { - reported[key{info.Path, info.Epoch}] = true - } - // Keep only assignments not confirmed. kept := pending[:0] for _, a := range pending { - if !reported[key{a.Path, a.Epoch}] { + if !assignmentConfirmedByHeartbeat(a, infos) { kept = append(kept, a) } } q.queues[server] = kept } +func assignmentConfirmedByHeartbeat(a blockvol.BlockVolumeAssignment, infos []blockvol.BlockVolumeInfoMessage) bool { + for _, info := range infos { + if info.Path != a.Path || info.Epoch != a.Epoch { + continue + } + expectedData, expectedCtrl, requiresReplicaTransport := assignmentReplicaTransport(a) + if !requiresReplicaTransport { + return true + } + if info.ReplicaDataAddr == expectedData && info.ReplicaCtrlAddr == expectedCtrl { + return true + } + } + return false +} + +func assignmentReplicaTransport(a blockvol.BlockVolumeAssignment) (dataAddr, ctrlAddr string, ok bool) { + if a.ReplicaDataAddr != "" || a.ReplicaCtrlAddr != "" { + return a.ReplicaDataAddr, a.ReplicaCtrlAddr, true + } + if len(a.ReplicaAddrs) == 1 { + return a.ReplicaAddrs[0].DataAddr, a.ReplicaAddrs[0].CtrlAddr, true + } + return "", "", false +} + // Pending returns the number of pending assignments for the server. func (q *BlockAssignmentQueue) Pending(server string) int { q.mu.Lock() diff --git a/weed/server/master_block_assignment_queue_test.go b/weed/server/master_block_assignment_queue_test.go index d5cb2e9f8..31dd8851d 100644 --- a/weed/server/master_block_assignment_queue_test.go +++ b/weed/server/master_block_assignment_queue_test.go @@ -143,6 +143,50 @@ func TestQueue_ConfirmFromHeartbeat_PrunesConfirmed(t *testing.T) { } } +func TestQueue_ConfirmFromHeartbeat_SameEpochRefreshWaitsForReplicaTransport(t *testing.T) { + q := NewBlockAssignmentQueue() + initial := mkAssign("/a.blk", 5, 1) + refresh := mkAssign("/a.blk", 5, 1) + refresh.ReplicaAddrs = []blockvol.ReplicaAddr{{ + DataAddr: "10.0.0.2:14260", + CtrlAddr: "10.0.0.2:14261", + ServerID: "vs2", + }} + q.Enqueue("s1", initial) + q.Enqueue("s1", refresh) + + // Old heartbeat confirms the epoch/role but does not yet report the refreshed + // replica transport, so only the original promote assignment is confirmed. + q.ConfirmFromHeartbeat("s1", []blockvol.BlockVolumeInfoMessage{{ + Path: "/a.blk", + Epoch: 5, + }}) + + got := q.Peek("s1") + if len(got) != 1 { + t.Fatalf("expected refresh assignment to remain pending, got %d: %+v", len(got), got) + } + if got[0].Path != "/a.blk" || got[0].Epoch != 5 { + t.Fatalf("wrong remaining assignment: %+v", got[0]) + } + if len(got[0].ReplicaAddrs) != 1 { + t.Fatalf("remaining refresh assignment lost replica addrs: %+v", got[0]) + } + + // Once the promoted VS reports the refreshed replica transport in heartbeat, + // the same-epoch refresh is confirmed and removed. + q.ConfirmFromHeartbeat("s1", []blockvol.BlockVolumeInfoMessage{{ + Path: "/a.blk", + Epoch: 5, + ReplicaDataAddr: "10.0.0.2:14260", + ReplicaCtrlAddr: "10.0.0.2:14261", + }}) + + if q.Pending("s1") != 0 { + t.Fatalf("expected refresh assignment to be confirmed after transport appears, got %d pending", q.Pending("s1")) + } +} + func TestQueue_PeekPrunesStaleEpochs(t *testing.T) { q := NewBlockAssignmentQueue() q.Enqueue("s1", mkAssign("/a.blk", 1, 1)) // stale diff --git a/weed/server/master_block_failover.go b/weed/server/master_block_failover.go index 87782ece5..f2d1efac4 100644 --- a/weed/server/master_block_failover.go +++ b/weed/server/master_block_failover.go @@ -2,6 +2,9 @@ package weed_server import ( "context" + "fmt" + "hash/fnv" + "strings" "sync" "time" @@ -258,6 +261,7 @@ func (ms *MasterServer) promoteReplica(volumeName string) { oldPrimary := entry.VolumeServer oldPath := entry.Path + oldPrimaryISCSIAddr := entry.ISCSIAddr // CP8-2: Use PromoteBestReplica (picks by health score, tie-break by WALHeadLSN). newEpoch, err := ms.blockRegistry.PromoteBestReplica(volumeName) @@ -266,13 +270,13 @@ func (ms *MasterServer) promoteReplica(volumeName string) { return } - ms.finalizePromotion(volumeName, oldPrimary, oldPath, newEpoch) + 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). -func (ms *MasterServer) finalizePromotion(volumeName, oldPrimary, oldPath string, newEpoch uint64) { +func (ms *MasterServer) finalizePromotion(volumeName, oldPrimary, oldPath, oldPrimaryISCSIAddr string, newEpoch uint64) { // Re-read entry after promotion. entry, ok := ms.blockRegistry.Lookup(volumeName) if !ok { @@ -303,11 +307,14 @@ func (ms *MasterServer) finalizePromotion(volumeName, oldPrimary, oldPath string ms.blockAssignmentQueue.Enqueue(entry.VolumeServer, assignment) // Record pending rebuild for when dead server reconnects. + replicaDataAddr, replicaCtrlAddr := deterministicReplicaAddrsForReplicaPath(oldPath, oldPrimary, oldPrimaryISCSIAddr) ms.recordPendingRebuild(oldPrimary, pendingRebuild{ VolumeName: volumeName, OldPath: oldPath, NewPrimary: entry.VolumeServer, Epoch: newEpoch, + ReplicaDataAddr: replicaDataAddr, + ReplicaCtrlAddr: replicaCtrlAddr, }) ms.blockRegistry.PromotionsTotal.Add(1) @@ -315,6 +322,35 @@ func (ms *MasterServer) finalizePromotion(volumeName, oldPrimary, oldPath string volumeName, entry.VolumeServer, newEpoch, oldPrimary) } +// deterministicReplicaAddrsForReplicaPath mirrors the volume-server-side +// ReplicationPorts derivation so the master can preserve the reconnect catch-up +// path even before the restarted replica emits a second heartbeat with explicit +// receiver addresses. +func deterministicReplicaAddrsForReplicaPath(path, serverAddr, iscsiAddr string) (dataAddr, ctrlAddr string) { + host := serverAddr + if idx := strings.LastIndex(host, ":"); idx >= 0 { + host = host[:idx] + } + if host == "" { + return "", "" + } + + basePort := 3260 + if idx := strings.LastIndex(iscsiAddr, ":"); idx >= 0 { + var p int + if _, err := fmt.Sscanf(iscsiAddr[idx+1:], "%d", &p); err == nil && p > 0 { + basePort = p + } + } + + h := fnv.New32a() + _, _ = h.Write([]byte(path)) + offset := int(h.Sum32()%500) * 3 + dataPort := basePort + 1000 + offset + ctrlPort := dataPort + 1 + return fmt.Sprintf("%s:%d", host, dataPort), fmt.Sprintf("%s:%d", host, ctrlPort) +} + // recordPendingRebuild stores a pending rebuild for a dead server. func (ms *MasterServer) recordPendingRebuild(deadServer string, rb pendingRebuild) { if ms.blockFailover == nil { diff --git a/weed/server/master_block_registry.go b/weed/server/master_block_registry.go index 42865e2c2..1bfb7fb73 100644 --- a/weed/server/master_block_registry.go +++ b/weed/server/master_block_registry.go @@ -86,6 +86,7 @@ type BlockVolumeEntry struct { HasHeartbeatVolumeMode bool // whether the current primary heartbeat carried explicit outward volume_mode truth HeartbeatVolumeReason string // explicit primary heartbeat outward volume_mode_reason truth when present HasHeartbeatVolumeReason bool // whether the current primary heartbeat carried explicit outward volume_mode_reason truth + PendingPrimaryHeartbeat bool // promotion selected a new primary, but it has not yet heartbeated as primary on the new epoch // CP13-9: Normalized volume mode for external surfaces. // Computed by recomputeReplicaState from the current entry state. @@ -206,6 +207,12 @@ func (e *BlockVolumeEntry) computeVolumeMode() string { } } + // A registry-driven promotion has selected a winner, but outward publication + // should not read as complete until that winner heartbeats as the new primary. + if e.PendingPrimaryHeartbeat { + return "bootstrap_pending" + } + // Prefer explicit primary heartbeat publish_healthy truth when present. if e.HasPublishHealthy && e.PublishHealthy { return "publish_healthy" @@ -679,6 +686,7 @@ func (r *BlockVolumeRegistry) applyPrimaryHeartbeatObservation(existing *BlockVo existing.Role = info.Role existing.Status = StatusActive existing.LastLeaseGrant = time.Now() + existing.PendingPrimaryHeartbeat = false existing.HealthScore = info.HealthScore existing.TransportDegraded = info.ReplicaDegraded applyExplicitPrimaryTruthFromHeartbeat(existing, info, true) @@ -737,6 +745,8 @@ func (r *BlockVolumeRegistry) applyReplicaHeartbeatObservation(existing *BlockVo oldCtrl := existing.Replicas[i].CtrlAddr dataChanged := info.ReplicaDataAddr != "" && oldData != "" && oldData != info.ReplicaDataAddr ctrlChanged := info.ReplicaCtrlAddr != "" && oldCtrl != "" && oldCtrl != info.ReplicaCtrlAddr + dataBecameKnown := oldData == "" && info.ReplicaDataAddr != "" + ctrlBecameKnown := oldCtrl == "" && info.ReplicaCtrlAddr != "" if dataChanged || ctrlChanged { result.AddrChanges = append(result.AddrChanges, ReplicaAddrChange{ VolumeName: existingName, @@ -753,6 +763,17 @@ func (r *BlockVolumeRegistry) applyReplicaHeartbeatObservation(existing *BlockVo if info.ReplicaCtrlAddr != "" { existing.Replicas[i].CtrlAddr = info.ReplicaCtrlAddr } + if dataBecameKnown || ctrlBecameKnown { + existing.NeedsPrimaryRefresh = true + } + } + if len(existing.Replicas) > 0 && existing.Replicas[0].Server == server { + existing.ReplicaServer = existing.Replicas[0].Server + existing.ReplicaPath = existing.Replicas[0].Path + existing.ReplicaISCSIAddr = existing.Replicas[0].ISCSIAddr + existing.ReplicaIQN = existing.Replicas[0].IQN + existing.ReplicaDataAddr = existing.Replicas[0].DataAddr + existing.ReplicaCtrlAddr = existing.Replicas[0].CtrlAddr } break } @@ -1005,7 +1026,15 @@ func (r *BlockVolumeRegistry) upsertServerAsReplica(name string, existing *Block for i := range existing.Replicas { if existing.Replicas[i].Server == newServer { // Update in place — force RoleReplica regardless of heartbeat claim. + oldData := existing.Replicas[i].DataAddr + oldCtrl := existing.Replicas[i].CtrlAddr existing.Replicas[i].Path = info.Path + if info.ReplicaDataAddr != "" { + existing.Replicas[i].DataAddr = info.ReplicaDataAddr + } + if info.ReplicaCtrlAddr != "" { + existing.Replicas[i].CtrlAddr = info.ReplicaCtrlAddr + } existing.Replicas[i].HealthScore = info.HealthScore existing.Replicas[i].WALHeadLSN = info.WalHeadLsn existing.Replicas[i].LastHeartbeat = time.Now() @@ -1013,6 +1042,17 @@ func (r *BlockVolumeRegistry) upsertServerAsReplica(name string, existing *Block existing.Replicas[i].NvmeAddr = info.NvmeAddr existing.Replicas[i].NQN = info.Nqn applyReplicaReadyFromHeartbeat(&existing.Replicas[i], info, true) + if len(existing.Replicas) > 0 && existing.Replicas[0].Server == newServer { + existing.ReplicaServer = existing.Replicas[0].Server + existing.ReplicaPath = existing.Replicas[0].Path + existing.ReplicaISCSIAddr = existing.Replicas[0].ISCSIAddr + existing.ReplicaIQN = existing.Replicas[0].IQN + existing.ReplicaDataAddr = existing.Replicas[0].DataAddr + existing.ReplicaCtrlAddr = existing.Replicas[0].CtrlAddr + } + if (oldData == "" && existing.Replicas[i].DataAddr != "") || (oldCtrl == "" && existing.Replicas[i].CtrlAddr != "") { + existing.NeedsPrimaryRefresh = true + } return } } @@ -1563,9 +1603,20 @@ func (r *BlockVolumeRegistry) applyPromotionLocked(entry *BlockVolumeEntry, name entry.Epoch = newEpoch entry.Role = blockvol.RoleToWire(blockvol.RolePrimary) entry.LastLeaseGrant = time.Now() + entry.PendingPrimaryHeartbeat = true + entry.HealthScore = candidate.HealthScore + entry.WALHeadLSN = candidate.WALHeadLSN // Clear stale rebuild/publication metadata from old primary (B-11 partial fix). entry.RebuildListenAddr = "" + entry.NeedsRebuild = false + entry.HasNeedsRebuild = false + entry.PublishHealthy = false + entry.HasPublishHealthy = false + entry.HeartbeatVolumeMode = "" + entry.HasHeartbeatVolumeMode = false + entry.HeartbeatVolumeReason = "" + entry.HasHeartbeatVolumeReason = false // Remove promoted from Replicas. Others stay. entry.Replicas = append(entry.Replicas[:candidateIdx], entry.Replicas[candidateIdx+1:]...) @@ -1590,6 +1641,7 @@ func (r *BlockVolumeRegistry) applyPromotionLocked(entry *BlockVolumeEntry, name // Update byServer index: new primary server now hosts this volume. r.addToServer(entry.VolumeServer, name) + entry.recomputeReplicaState() return newEpoch } diff --git a/weed/server/master_block_registry_test.go b/weed/server/master_block_registry_test.go index f3b3a2f15..ea1ba6306 100644 --- a/weed/server/master_block_registry_test.go +++ b/weed/server/master_block_registry_test.go @@ -1266,6 +1266,143 @@ func TestRegistry_PromoteBestReplica_ClearsRebuildAddr(t *testing.T) { } } +func TestRegistry_PromoteBestReplica_ClearsStalePrimaryTruthUntilNewPrimaryHeartbeat(t *testing.T) { + r := NewBlockVolumeRegistry() + r.MarkBlockCapable("replica1") + r.MarkBlockCapable("replica2") + r.Register(&BlockVolumeEntry{ + Name: "vol1", + VolumeServer: "primary", + Path: "/data/vol1.blk", + Epoch: 5, + ReplicaFactor: 3, + PublishHealthy: true, + HasPublishHealthy: true, + HeartbeatVolumeMode: "publish_healthy", + HasHeartbeatVolumeMode: true, + HeartbeatVolumeReason: "", + HasHeartbeatVolumeReason: true, + WALHeadLSN: 120, + Replicas: []ReplicaInfo{ + { + Server: "replica1", + Path: "/r1.blk", + HealthScore: 1.0, + WALHeadLSN: 110, + LastHeartbeat: time.Now(), + Role: blockvol.RoleToWire(blockvol.RoleReplica), + Ready: true, + HasExplicitReady: true, + }, + { + Server: "replica2", + Path: "/r2.blk", + HealthScore: 0.9, + WALHeadLSN: 109, + LastHeartbeat: time.Now(), + Role: blockvol.RoleToWire(blockvol.RoleReplica), + Ready: true, + HasExplicitReady: true, + }, + }, + }) + + newEpoch, err := r.PromoteBestReplica("vol1") + if err != nil { + t.Fatalf("PromoteBestReplica: %v", err) + } + if newEpoch != 6 { + t.Fatalf("epoch=%d, want 6", newEpoch) + } + + entry, _ := r.Lookup("vol1") + if entry.VolumeServer != "replica1" { + t.Fatalf("expected replica1 promoted, got %q", entry.VolumeServer) + } + if !entry.PendingPrimaryHeartbeat { + t.Fatalf("expected promotion to wait for new primary heartbeat, entry=%+v", entry) + } + if entry.HasPublishHealthy || entry.PublishHealthy { + t.Fatalf("stale publish_healthy truth should be cleared on promotion, entry=%+v", entry) + } + if entry.HasHeartbeatVolumeMode || entry.HeartbeatVolumeMode != "" { + t.Fatalf("stale heartbeat volume_mode should be cleared on promotion, entry=%+v", entry) + } + if entry.HasHeartbeatVolumeReason || entry.HeartbeatVolumeReason != "" { + t.Fatalf("stale heartbeat volume_mode_reason should be cleared on promotion, entry=%+v", entry) + } + if entry.WALHeadLSN != 110 { + t.Fatalf("expected promoted primary WALHeadLSN rebased to candidate, got %d", entry.WALHeadLSN) + } + if entry.VolumeMode != "bootstrap_pending" { + t.Fatalf("expected outward mode bootstrap_pending until new primary heartbeat, got %q", entry.VolumeMode) + } +} + +func TestRegistry_PromoteBestReplica_NewPrimaryHeartbeatClearsPendingAndRestoresExplicitTruth(t *testing.T) { + r := NewBlockVolumeRegistry() + r.MarkBlockCapable("replica1") + r.MarkBlockCapable("replica2") + r.Register(&BlockVolumeEntry{ + Name: "vol1", + VolumeServer: "primary", + Path: "/data/vol1.blk", + Epoch: 5, + ReplicaFactor: 3, + Replicas: []ReplicaInfo{ + { + Server: "replica1", + Path: "/r1.blk", + HealthScore: 1.0, + WALHeadLSN: 110, + LastHeartbeat: time.Now(), + Role: blockvol.RoleToWire(blockvol.RoleReplica), + Ready: true, + HasExplicitReady: true, + }, + { + Server: "replica2", + Path: "/r2.blk", + HealthScore: 0.9, + WALHeadLSN: 109, + LastHeartbeat: time.Now(), + Role: blockvol.RoleToWire(blockvol.RoleReplica), + Ready: true, + HasExplicitReady: true, + }, + }, + }) + + if _, err := r.PromoteBestReplica("vol1"); err != nil { + t.Fatalf("PromoteBestReplica: %v", err) + } + + publishHealthy := true + mode := "publish_healthy" + reason := "" + r.UpdateFullHeartbeat("replica1", []*master_pb.BlockVolumeInfoMessage{{ + Path: "/r1.blk", + Epoch: 6, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + WalHeadLsn: 110, + PublishHealthy: &publishHealthy, + VolumeMode: &mode, + VolumeModeReason: &reason, + VolumeSize: 1 << 30, + }}, "") + + entry, _ := r.Lookup("vol1") + if entry.PendingPrimaryHeartbeat { + t.Fatalf("expected new primary heartbeat to clear pending state, entry=%+v", entry) + } + if !entry.HasPublishHealthy || !entry.PublishHealthy { + t.Fatalf("expected explicit publish_healthy truth from promoted primary heartbeat, entry=%+v", entry) + } + if entry.VolumeMode != "publish_healthy" { + t.Fatalf("expected outward mode publish_healthy after promoted primary heartbeat, got %q", entry.VolumeMode) + } +} + // --- LeaseGrants --- func TestRegistry_LeaseGrants_PrimaryOnly(t *testing.T) { diff --git a/weed/server/master_server_handlers_block.go b/weed/server/master_server_handlers_block.go index 77d948be2..1e9114f85 100644 --- a/weed/server/master_server_handlers_block.go +++ b/weed/server/master_server_handlers_block.go @@ -328,6 +328,14 @@ func (ms *MasterServer) blockVolumePromoteHandler(w http.ResponseWriter, r *http return } + // Capture the old primary's iSCSI base port before promotion so the shared + // finalize path can derive deterministic receiver ports for the reconnecting + // old primary. + oldPrimaryISCSIAddr := "" + if entry, ok := ms.blockRegistry.Lookup(name); ok { + oldPrimaryISCSIAddr = entry.ISCSIAddr + } + // ManualPromote captures oldPrimary/oldPath under lock to avoid TOCTOU (BUG-T5-2). newEpoch, oldPrimary, oldPath, pf, err := ms.blockRegistry.ManualPromote(name, req.TargetServer, req.Force) if err != nil { @@ -352,7 +360,7 @@ func (ms *MasterServer) blockVolumePromoteHandler(w http.ResponseWriter, r *http } // Post-promotion orchestration (same as auto path). - ms.finalizePromotion(name, oldPrimary, oldPath, newEpoch) + ms.finalizePromotion(name, oldPrimary, oldPath, oldPrimaryISCSIAddr, newEpoch) if req.Reason != "" { glog.V(0).Infof("manual promote %q: reason=%q", name, req.Reason) diff --git a/weed/server/qa_block_cp11b3_adversarial_test.go b/weed/server/qa_block_cp11b3_adversarial_test.go index f2377aa4f..c6058a951 100644 --- a/weed/server/qa_block_cp11b3_adversarial_test.go +++ b/weed/server/qa_block_cp11b3_adversarial_test.go @@ -1324,7 +1324,7 @@ func TestQA_T5_PromoteHandler_HTTP(t *testing.T) { } // Simulate finalizePromotion. - ms.finalizePromotion("vol1", oldPrimary, oldPath, newEpoch) + ms.finalizePromotion("vol1", oldPrimary, oldPath, "", newEpoch) // Verify. entry, _ := ms.blockRegistry.Lookup("vol1") @@ -1401,7 +1401,7 @@ func TestQA_T5_PromotionsTotal_CountsBothAutoAndManual(t *testing.T) { if err != nil { t.Fatalf("manual promote: %v", err) } - ms.finalizePromotion("vol2", oldPrimary, oldPath, newEpoch) + ms.finalizePromotion("vol2", oldPrimary, oldPath, "", newEpoch) afterManual := ms.blockRegistry.PromotionsTotal.Load() if afterManual != afterAuto+1 { t.Fatalf("manual promote should increment PromotionsTotal: afterAuto=%d afterManual=%d", afterAuto, afterManual) diff --git a/weed/server/qa_promote_rejoin_live_test.go b/weed/server/qa_promote_rejoin_live_test.go new file mode 100644 index 000000000..671905c2d --- /dev/null +++ b/weed/server/qa_promote_rejoin_live_test.go @@ -0,0 +1,243 @@ +// Reproduces the live cp11b3-manual-promote failure path: +// create RF=2 → kill primary ��� promote replica → restart killed VS → verify +// that the promoted primary eventually gets an assignment with replica addresses. +// +// This test simulates the exact heartbeat/assignment sequence observed in glog +// to find where replicas are dropped. +package weed_server + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + blockvol "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +func TestPromote_LivePath_RestartedVSRejoinsAndPrimaryGetsReplica(t *testing.T) { + ms := testMasterServerForFailover(t) + + pathVS1 := "/data/vs1/promote-test.blk" + pathVS2 := "/data/vs2/promote-test.blk" + + ms.blockRegistry.MarkBlockCapable("vs1:18192") + ms.blockRegistry.MarkBlockCapable("vs2:18193") + + // Step 1: Create RF=2 volume. VS1=primary, VS2=replica. + ms.blockRegistry.Register(&BlockVolumeEntry{ + Name: "promote-test", + VolumeServer: "vs1:18192", + Path: pathVS1, + ISCSIAddr: "vs1:3279", + SizeBytes: 50 << 20, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + Status: StatusActive, + LeaseTTL: 30 * time.Second, + LastLeaseGrant: time.Now(), + ReplicaFactor: 2, + ReplicaServer: "vs2:18193", + ReplicaPath: pathVS2, + Replicas: []ReplicaInfo{ + { + Server: "vs2:18193", + Path: pathVS2, + ISCSIAddr: "vs2:3280", + HealthScore: 1.0, + Role: blockvol.RoleToWire(blockvol.RoleReplica), + LastHeartbeat: time.Now(), + DataAddr: "vs2:14260", + CtrlAddr: "vs2:14261", + }, + }, + }) + + t.Log("Step 1: created RF=2, vs1=primary, vs2=replica") + + // Step 2: Kill VS1 (simulate disconnect). + ms.blockRegistry.UnmarkBlockCapable("vs1:18192") + + // Step 3: Expire lease and promote VS2. + ms.blockRegistry.UpdateEntry("promote-test", func(e *BlockVolumeEntry) { + e.LastLeaseGrant = time.Now().Add(-1 * time.Minute) + }) + ms.failoverBlockVolumes("vs1:18192") + + entry := lookupEntryT(t, ms.blockRegistry, "promote-test") + if entry.VolumeServer != "vs2:18193" { + t.Fatalf("promote failed: primary=%s", entry.VolumeServer) + } + t.Logf("Step 3: promoted vs2, epoch=%d, replicas=%d", entry.Epoch, len(entry.Replicas)) + + // Drain the promote assignment. + promoteAssignments := ms.blockAssignmentQueue.Peek("vs2:18193") + for _, a := range promoteAssignments { + ms.blockAssignmentQueue.Confirm("vs2:18193", a.Path, a.Epoch) + } + t.Logf("Step 3: drained %d promote assignments for vs2", len(promoteAssignments)) + + // Step 4: VS1 restarts. First heartbeat: epoch=1, role=primary (stale). + ms.blockRegistry.MarkBlockCapable("vs1:18192") + result := ms.blockRegistry.UpdateFullHeartbeat("vs1:18192", []*master_pb.BlockVolumeInfoMessage{ + { + Path: pathVS1, + VolumeSize: 50 << 20, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + // First heartbeat: no replica addresses yet. + }, + }, "") + + entry = lookupEntryT(t, ms.blockRegistry, "promote-test") + t.Logf("Step 4a: after vs1 first heartbeat: primary=%s replicas=%d", entry.VolumeServer, len(entry.Replicas)) + for i, ri := range entry.Replicas { + t.Logf(" replica[%d]: server=%s data=%s ctrl=%s", i, ri.Server, ri.DataAddr, ri.CtrlAddr) + } + + // Process any primary refresh from first heartbeat. + for _, refresh := range result.PrimaryRefreshNeeded { + ms.enqueuePrimaryRefresh(refresh) + t.Logf("Step 4a: primary refresh triggered for %s", refresh.Name) + } + + // Check if vs2 got a refresh assignment with replicas. + vs2Assignments := ms.blockAssignmentQueue.Peek("vs2:18193") + t.Logf("Step 4a: vs2 pending assignments: %d", len(vs2Assignments)) + for _, a := range vs2Assignments { + t.Logf(" assignment: path=%s epoch=%d role=%d replicaDataAddr=%s replicaAddrs=%d", + a.Path, a.Epoch, a.Role, a.ReplicaDataAddr, len(a.ReplicaAddrs)) + } + + // Step 5: VS1 second heartbeat: now with replica data/ctrl addresses. + result2 := ms.blockRegistry.UpdateFullHeartbeat("vs1:18192", []*master_pb.BlockVolumeInfoMessage{ + { + Path: pathVS1, + VolumeSize: 50 << 20, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + ReplicaDataAddr: "vs1:14262", + ReplicaCtrlAddr: "vs1:14263", + }, + }, "") + + entry = lookupEntryT(t, ms.blockRegistry, "promote-test") + t.Logf("Step 5: after vs1 second heartbeat (with addrs): primary=%s replicas=%d", entry.VolumeServer, len(entry.Replicas)) + for i, ri := range entry.Replicas { + t.Logf(" replica[%d]: server=%s data=%s ctrl=%s", i, ri.Server, ri.DataAddr, ri.CtrlAddr) + } + + // Process any primary refresh from second heartbeat. + for _, refresh := range result2.PrimaryRefreshNeeded { + ms.enqueuePrimaryRefresh(refresh) + t.Logf("Step 5: primary refresh triggered for %s", refresh.Name) + } + + // Step 6: Final check — vs2 should now have a pending assignment with replica addresses. + vs2Final := ms.blockAssignmentQueue.Peek("vs2:18193") + t.Logf("Step 6: vs2 final pending assignments: %d", len(vs2Final)) + + hasReplicaAddrs := false + for _, a := range vs2Final { + t.Logf(" assignment: path=%s epoch=%d role=%d replicaDataAddr=%s replicaAddrs=%d", + a.Path, a.Epoch, a.Role, a.ReplicaDataAddr, len(a.ReplicaAddrs)) + if a.ReplicaDataAddr != "" || len(a.ReplicaAddrs) > 0 { + hasReplicaAddrs = true + } + } + + if !hasReplicaAddrs { + t.Fatalf("BUG: after promote + vs1 rejoin with addresses, vs2 STILL has no assignment with replica addresses.\n"+ + "This reproduces the live cp11b3-manual-promote failure: promoted primary never gets shipper configuration.") + } + t.Log("SUCCESS: promoted primary has pending assignment with replica addresses after rejoin") +} + +func TestPromote_LivePath_FirstHeartbeatThenRecoverUsesDeterministicReplicaAddrs(t *testing.T) { + ms := testMasterServerForFailover(t) + + pathVS1 := "/data/vs1/promote-test.blk" + pathVS2 := "/data/vs2/promote-test.blk" + + ms.blockRegistry.MarkBlockCapable("vs1:18192") + ms.blockRegistry.MarkBlockCapable("vs2:18193") + + ms.blockRegistry.Register(&BlockVolumeEntry{ + Name: "promote-test", + VolumeServer: "vs1:18192", + Path: pathVS1, + ISCSIAddr: "192.168.1.184:3279", + SizeBytes: 50 << 20, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + Status: StatusActive, + LeaseTTL: 30 * time.Second, + LastLeaseGrant: time.Now().Add(-1 * time.Minute), + ReplicaFactor: 2, + ReplicaServer: "vs2:18193", + ReplicaPath: pathVS2, + Replicas: []ReplicaInfo{{ + Server: "vs2:18193", + Path: pathVS2, + ISCSIAddr: "192.168.1.184:3280", + HealthScore: 1.0, + Role: blockvol.RoleToWire(blockvol.RoleReplica), + LastHeartbeat: time.Now(), + DataAddr: "192.168.1.184:4571", + CtrlAddr: "192.168.1.184:4572", + }}, + }) + + ms.blockRegistry.UnmarkBlockCapable("vs1:18192") + ms.failoverBlockVolumes("vs1:18192") + + for _, a := range ms.blockAssignmentQueue.Peek("vs2:18193") { + ms.blockAssignmentQueue.Confirm("vs2:18193", a.Path, a.Epoch) + } + + // The restarted old primary first reports only a stale heartbeat with no + // receiver addrs, matching the live manual-promote evidence. + ms.blockRegistry.MarkBlockCapable("vs1:18192") + ms.blockRegistry.UpdateFullHeartbeat("vs1:18192", []*master_pb.BlockVolumeInfoMessage{{ + Path: pathVS1, + VolumeSize: 50 << 20, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + }}, "") + + // Live path: after processing the heartbeat, the master runs reconnect + // recovery immediately. The pending rebuild must already carry deterministic + // replica addrs so recoverBlockVolumes can take the catch-up-first path + // without waiting for a second heartbeat. + ms.recoverBlockVolumes("vs1:18192") + + vs1Assignments := ms.blockAssignmentQueue.Peek("vs1:18192") + foundReplicaAssign := false + for _, a := range vs1Assignments { + if a.Path != pathVS1 { + continue + } + if blockvol.RoleFromWire(a.Role) == blockvol.RoleReplica && a.ReplicaDataAddr != "" && a.ReplicaCtrlAddr != "" { + foundReplicaAssign = true + } + if blockvol.RoleFromWire(a.Role) == blockvol.RoleRebuilding { + t.Fatalf("expected catch-up replica assignment, got rebuild assignment: %+v", a) + } + } + if !foundReplicaAssign { + t.Fatalf("restarted VS1 did not receive replica assignment with deterministic receiver addrs") + } + + vs2Assignments := ms.blockAssignmentQueue.Peek("vs2:18193") + foundPrimaryRefresh := false + for _, a := range vs2Assignments { + if a.Path != pathVS2 { + continue + } + if blockvol.RoleFromWire(a.Role) == blockvol.RolePrimary && (a.ReplicaDataAddr != "" || len(a.ReplicaAddrs) > 0) { + foundPrimaryRefresh = true + } + } + if !foundPrimaryRefresh { + t.Fatalf("promoted primary did not receive refreshed assignment with replica membership after reconnect recovery") + } +} diff --git a/weed/server/qa_promote_replication_test.go b/weed/server/qa_promote_replication_test.go index fcb044a69..6cf4b41e5 100644 --- a/weed/server/qa_promote_replication_test.go +++ b/weed/server/qa_promote_replication_test.go @@ -5,9 +5,9 @@ // replication dead. sync_all barrier passes vacuously with 0 shippers. // // This test suite verifies: -// 1. After promote + old primary re-register, the new primary's assignment -// includes the re-registered replica's addresses -// 2. sync_all with 0 shippers and RF>1 is detected as a gap +// 1. After promote + old primary re-register, the new primary's assignment +// includes the re-registered replica's addresses +// 2. sync_all with 0 shippers and RF>1 is detected as a gap package weed_server import ( @@ -127,14 +127,112 @@ func TestPromote_AssignmentHasReplicaAddrs(t *testing.T) { } if !foundReplicaAddrs { - t.Fatalf("BUG: after promote + re-register, new primary vs2 has NO assignment "+ - "with replica addresses.\n"+ - "This means the shipper will never be configured and replication is dead.\n"+ + t.Fatalf("BUG: after promote + re-register, new primary vs2 has NO assignment " + + "with replica addresses.\n" + + "This means the shipper will never be configured and replication is dead.\n" + "The master must send an updated assignment to vs2 after vs1 re-registers as replica.") } t.Log("post-promote assignment has replica addresses — shipper will be configured") } +// TestPromote_ReRegisterAddressUpgradeTriggersPrimaryRefresh verifies the +// rejoin path where the first stale heartbeat re-registers the old primary as a +// replica before it has receiver addresses, and a later heartbeat upgrades that +// replica entry with real addresses. The master must send a fresh Primary +// assignment once the addresses become known. +func TestPromote_ReRegisterAddressUpgradeTriggersPrimaryRefresh(t *testing.T) { + ms := testMasterServerForFailover(t) + + pathA := "/data/vs1/vol1.blk" + pathB := "/data/vs2/vol1.blk" + + ms.blockRegistry.MarkBlockCapable("vs1") + ms.blockRegistry.MarkBlockCapable("vs2") + + ms.blockRegistry.Register(&BlockVolumeEntry{ + Name: "vol1", + VolumeServer: "vs1", + Path: pathA, + ISCSIAddr: "vs1:3260", + SizeBytes: 1 << 30, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + Status: StatusActive, + LeaseTTL: 30 * time.Second, + LastLeaseGrant: time.Now().Add(-1 * time.Minute), + ReplicaServer: "vs2", + ReplicaPath: pathB, + Replicas: []ReplicaInfo{{ + Server: "vs2", + Path: pathB, + ISCSIAddr: "vs2:3260", + HealthScore: 1.0, + Role: blockvol.RoleToWire(blockvol.RoleReplica), + LastHeartbeat: time.Now(), + DataAddr: "vs2:14260", + CtrlAddr: "vs2:14261", + }}, + }) + + ms.failoverBlockVolumes("vs1") + + // Drain promotion assignment. + for _, a := range ms.blockAssignmentQueue.Peek("vs2") { + ms.blockAssignmentQueue.Confirm("vs2", a.Path, a.Epoch) + } + + // First stale heartbeat from the old primary re-registers it as a replica, + // but before it has receiver addresses. + firstHB := ms.blockRegistry.UpdateFullHeartbeat("vs1", []*master_pb.BlockVolumeInfoMessage{{ + Path: pathA, + VolumeSize: 1 << 30, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + }}, "") + for _, refreshEntry := range firstHB.PrimaryRefreshNeeded { + ms.enqueuePrimaryRefresh(refreshEntry) + } + for _, a := range ms.blockAssignmentQueue.Peek("vs2") { + ms.blockAssignmentQueue.Confirm("vs2", a.Path, a.Epoch) + } + + entry := lookupEntryT(t, ms.blockRegistry, "vol1") + if len(entry.Replicas) != 1 { + t.Fatalf("after stale re-register: replicas=%d, want 1", len(entry.Replicas)) + } + if entry.Replicas[0].DataAddr != "" || entry.Replicas[0].CtrlAddr != "" { + t.Fatalf("stale re-register should not have addresses yet, entry=%+v", entry.Replicas[0]) + } + + // Second heartbeat arrives after the replica assignment was applied and the + // receiver endpoints are now known. This must trigger a fresh Primary refresh. + secondHB := ms.blockRegistry.UpdateFullHeartbeat("vs1", []*master_pb.BlockVolumeInfoMessage{{ + Path: pathA, + VolumeSize: 1 << 30, + Epoch: entry.Epoch, + Role: blockvol.RoleToWire(blockvol.RoleReplica), + ReplicaDataAddr: "vs1:14260", + ReplicaCtrlAddr: "vs1:14261", + }}, "") + for _, refreshEntry := range secondHB.PrimaryRefreshNeeded { + ms.enqueuePrimaryRefresh(refreshEntry) + } + + updatedAssignments := ms.blockAssignmentQueue.Peek("vs2") + foundReplicaAddrs := false + for _, a := range updatedAssignments { + if a.Path != entry.Path { + continue + } + if a.ReplicaDataAddr != "" || len(a.ReplicaAddrs) > 0 { + foundReplicaAddrs = true + } + } + if !foundReplicaAddrs { + t.Fatalf("address-upgrade heartbeat did not trigger primary refresh with replica addresses") + } +} + // TestPromote_ReplicasEmptyAfterPromote documents the current behavior: // after PromoteBestReplica, entry.Replicas is empty. func TestPromote_ReplicasEmptyAfterPromote(t *testing.T) { diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index 98a03ed2b..7dda0369d 100644 --- a/weed/server/volume_server_block.go +++ b/weed/server/volume_server_block.go @@ -559,7 +559,7 @@ func (bs *BlockService) applyCoreAssignmentEvent(a blockvol.BlockVolumeAssignmen if !ok { return nil } - result := bs.v2Core.ApplyEvent(ev) + result := bs.coreApplyAndLog(ev) return bs.applyCoreCommandsWithAssignment(result.Commands, &a) } @@ -567,10 +567,25 @@ func (bs *BlockService) applyCoreEvent(ev engine.Event) { if bs == nil || bs.v2Core == nil { return } - result := bs.v2Core.ApplyEvent(ev) + result := bs.coreApplyAndLog(ev) bs.applyCoreCommands(result.Commands) } +// coreApplyAndLog applies an event to the V2 core and logs the transition. +// All core event paths (assignment-driven and observation-driven) must use this +// so the VS log contains a complete trace for post-run diagnosis. +func (bs *BlockService) coreApplyAndLog(ev engine.Event) engine.ApplyResult { + result := bs.v2Core.ApplyEvent(ev) + glog.V(0).Infof("core [%s]: event=%T mode=%s pub=%v reason=%q readiness={applied=%v shipper_cfg=%v shipper_conn=%v recv=%v} boundary={durable=%d committed=%d} cmds=%d", + ev.VolumeID(), ev, result.Projection.Mode.Name, + result.Projection.Publication.Healthy, result.Projection.Publication.Reason, + result.Projection.Readiness.RoleApplied, result.Projection.Readiness.ShipperConfigured, + result.Projection.Readiness.ShipperConnected, result.Projection.Readiness.ReceiverReady, + result.Projection.Boundary.DurableLSN, result.Projection.Boundary.CommittedLSN, + len(result.Commands)) + return result +} + func (bs *BlockService) applyCoreCommands(cmds []engine.Command) { _ = bs.applyCoreCommandsWithAssignment(cmds, nil) } @@ -713,10 +728,11 @@ func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (e if len(a.ReplicaAddrs) > 0 { ev.Replicas = make([]engine.ReplicaAssignment, 0, len(a.ReplicaAddrs)) for _, ra := range a.ReplicaAddrs { - if ra.ServerID == "" { + replicaServerID := legacyReplicaServerID(ra.ServerID, ra.DataAddr, ra.CtrlAddr) + if replicaServerID == "" { continue } - ev.Replicas = append(ev.Replicas, bridgeblockvol.ReplicaAssignmentForServer(a.Path, ra.ServerID, engine.Endpoint{ + ev.Replicas = append(ev.Replicas, bridgeblockvol.ReplicaAssignmentForServer(a.Path, replicaServerID, engine.Endpoint{ DataAddr: ra.DataAddr, CtrlAddr: ra.CtrlAddr, })) @@ -724,9 +740,9 @@ func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (e if len(ev.Replicas) > 0 { ev.RecoveryTarget = engine.SessionCatchUp } - } else if a.ReplicaServerID != "" && a.ReplicaDataAddr != "" { + } else if replicaServerID := legacyReplicaServerID(a.ReplicaServerID, a.ReplicaDataAddr, a.ReplicaCtrlAddr); replicaServerID != "" && a.ReplicaDataAddr != "" { ev.Replicas = []engine.ReplicaAssignment{ - bridgeblockvol.ReplicaAssignmentForServer(a.Path, a.ReplicaServerID, engine.Endpoint{ + bridgeblockvol.ReplicaAssignmentForServer(a.Path, replicaServerID, engine.Endpoint{ DataAddr: a.ReplicaDataAddr, CtrlAddr: a.ReplicaCtrlAddr, }), @@ -758,6 +774,20 @@ func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (e } } +// legacyReplicaServerID keeps the V2 core wired even when a legacy scalar +// assignment path does not carry explicit ReplicaServerID. Prefer the stable +// server identity when present; otherwise fall back to transport identity so +// the primary still models replica membership and configures shippers. +func legacyReplicaServerID(serverID, dataAddr, ctrlAddr string) string { + if serverID != "" { + return serverID + } + if dataAddr != "" { + return dataAddr + } + return ctrlAddr +} + func (bs *BlockService) isPrimaryShipperConnected(path string) bool { if bs == nil || bs.blockStore == nil { return false diff --git a/weed/server/volume_server_block_debug.go b/weed/server/volume_server_block_debug.go index 42a2282ce..fc1beca0f 100644 --- a/weed/server/volume_server_block_debug.go +++ b/weed/server/volume_server_block_debug.go @@ -5,6 +5,7 @@ import ( "net/http" "time" + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" ) @@ -17,21 +18,24 @@ type ShipperDebugInfo struct { // BlockVolumeDebugInfo is the real-time block volume state. type BlockVolumeDebugInfo struct { - Path string `json:"path"` - Role string `json:"role"` - Mode string `json:"mode,omitempty"` - Epoch uint64 `json:"epoch"` - HeadLSN uint64 `json:"head_lsn"` - Degraded bool `json:"degraded"` - RoleApplied bool `json:"role_applied"` - ReceiverReady bool `json:"receiver_ready"` - ShipperConfigured bool `json:"shipper_configured"` - ShipperConnected bool `json:"shipper_connected"` - ReplicaEligible bool `json:"replica_eligible"` - PublishHealthy bool `json:"publish_healthy"` - PublicationReason string `json:"publication_reason,omitempty"` - Shippers []ShipperDebugInfo `json:"shippers,omitempty"` - Timestamp string `json:"timestamp"` + Path string `json:"path"` + Role string `json:"role"` + Mode string `json:"mode,omitempty"` + Epoch uint64 `json:"epoch"` + HeadLSN uint64 `json:"head_lsn"` + Degraded bool `json:"degraded"` + RoleApplied bool `json:"role_applied"` + ReceiverReady bool `json:"receiver_ready"` + ShipperConfigured bool `json:"shipper_configured"` + ShipperConnected bool `json:"shipper_connected"` + ReplicaEligible bool `json:"replica_eligible"` + PublishHealthy bool `json:"publish_healthy"` + PublicationReason string `json:"publication_reason,omitempty"` + Shippers []ShipperDebugInfo `json:"shippers,omitempty"` + CoreProjection *engine.PublicationProjection `json:"core_projection,omitempty"` + ExecutedCoreCommands []string `json:"executed_core_commands,omitempty"` + ProjectionMismatches []string `json:"projection_mismatches,omitempty"` + Timestamp string `json:"timestamp"` } // DebugInfoForVolume returns the current debug surface for one volume. When the @@ -64,6 +68,14 @@ func (bs *BlockService) DebugInfoForVolume(path string, vol *blockvol.BlockVol) info.ReplicaEligible = proj.Readiness.ReplicaReady info.PublishHealthy = proj.Publication.Healthy info.PublicationReason = proj.Publication.Reason + projCopy := proj + info.CoreProjection = &projCopy + } + if cmds := bs.ExecutedCoreCommands(path); len(cmds) > 0 { + info.ExecutedCoreCommands = cmds + } + if mismatches := bs.CoreProjectionMismatches(path); len(mismatches) > 0 { + info.ProjectionMismatches = mismatches } return info } diff --git a/weed/server/volume_server_block_test.go b/weed/server/volume_server_block_test.go index 21a3df4d3..6cfa2f716 100644 --- a/weed/server/volume_server_block_test.go +++ b/weed/server/volume_server_block_test.go @@ -296,6 +296,42 @@ func TestBlockService_ApplyAssignments_UpdatesCoreProjectionPrimaryPath(t *testi } } +func TestBlockService_ApplyAssignments_PrimaryScalarReplicaAddrWithoutServerID(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "vol-core-primary-scalar") + + errs := bs.ApplyAssignments([]blockvol.BlockVolumeAssignment{ + { + Path: path, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + LeaseTtlMs: 30000, + ReplicaDataAddr: "10.0.0.2:4260", + ReplicaCtrlAddr: "10.0.0.2:4261", + }, + }) + if len(errs) != 1 { + t.Fatalf("errs len=%d", len(errs)) + } + if errs[0] != nil { + t.Fatalf("apply assignment: %v", errs[0]) + } + + proj, ok := bs.CoreProjection(path) + if !ok { + t.Fatal("expected core projection to be cached on narrow live path") + } + if !proj.Readiness.RoleApplied { + t.Fatalf("role_applied should be observed even on scalar fallback path, projection=%+v", proj) + } + if len(proj.ReplicaIDs) != 1 { + t.Fatalf("replica_ids=%v", proj.ReplicaIDs) + } + if proj.Mode.Name != engine.ModeBootstrapPending { + t.Fatalf("mode=%s", proj.Mode.Name) + } +} + func TestBlockService_ApplyAssignments_RepeatedUnchangedStaysInSyncWithCore(t *testing.T) { bs := newTestBlockServiceDirect(t) path := createTestVolDirect(t, bs, "vol-core-repeat") @@ -962,6 +998,18 @@ func TestBlockService_DebugInfoForVolume_UsesCoreProjectionPrimaryPath(t *testin if info.PublicationReason != proj.Publication.Reason { t.Fatalf("publication_reason=%q projection_reason=%q", info.PublicationReason, proj.Publication.Reason) } + if info.CoreProjection == nil { + t.Fatal("expected embedded core projection in debug info") + } + if info.CoreProjection.VolumeID != proj.VolumeID || info.CoreProjection.Mode.Name != proj.Mode.Name { + t.Fatalf("embedded core projection diverged: got=%+v want=%+v", info.CoreProjection, proj) + } + if !reflect.DeepEqual(info.ExecutedCoreCommands, bs.ExecutedCoreCommands(path)) { + t.Fatalf("executed_core_commands=%v want=%v", info.ExecutedCoreCommands, bs.ExecutedCoreCommands(path)) + } + if len(info.ProjectionMismatches) != 0 { + t.Fatalf("projection_mismatches=%v", info.ProjectionMismatches) + } if info.PublishHealthy { t.Fatalf("debug surface must not overclaim healthy on primary path without durable boundary: %+v", info) } @@ -1012,6 +1060,18 @@ func TestBlockService_DebugInfoForVolume_UsesCoreProjectionReplicaPath(t *testin if info.PublicationReason != proj.Publication.Reason { t.Fatalf("publication_reason=%q projection_reason=%q", info.PublicationReason, proj.Publication.Reason) } + if info.CoreProjection == nil { + t.Fatal("expected embedded core projection in debug info") + } + if info.CoreProjection.VolumeID != proj.VolumeID || info.CoreProjection.Role != proj.Role { + t.Fatalf("embedded core projection diverged: got=%+v want=%+v", info.CoreProjection, proj) + } + if !reflect.DeepEqual(info.ExecutedCoreCommands, bs.ExecutedCoreCommands(path)) { + t.Fatalf("executed_core_commands=%v want=%v", info.ExecutedCoreCommands, bs.ExecutedCoreCommands(path)) + } + if len(info.ProjectionMismatches) != 0 { + t.Fatalf("projection_mismatches=%v", info.ProjectionMismatches) + } } func TestBlockService_CollectBlockVolumeHeartbeat_PrimaryUsesCoreReadinessGate(t *testing.T) { diff --git a/weed/storage/blockvol/test/component/fresh_volume_write_test.go b/weed/storage/blockvol/test/component/fresh_volume_write_test.go new file mode 100644 index 000000000..e6c22e355 --- /dev/null +++ b/weed/storage/blockvol/test/component/fresh_volume_write_test.go @@ -0,0 +1,124 @@ +package component + +import ( + "bytes" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +// TestFreshRF2SyncAll_WriteImmediatelyAfterCreate tests whether a fresh +// RF=2 sync_all volume can accept writes immediately, or if it needs +// the shipper to bootstrap first. +// +// This reproduces the recovery-baseline-failover dd_write failure: +// fio 4K random writes succeed, but dd_write (2MB sequential) fails +// on a fresh volume without wait_volume_healthy. +func TestFreshRF2SyncAll_WriteImmediatelyAfterCreate(t *testing.T) { + primaryPath := t.TempDir() + "/primary.blk" + replicaPath := t.TempDir() + "/replica.blk" + + primary, err := blockvol.CreateBlockVol(primaryPath, blockvol.CreateOptions{ + VolumeSize: 64 * 1024 * 1024, + BlockSize: 4096, + WALSize: 16 * 1024 * 1024, + DurabilityMode: blockvol.DurabilitySyncAll, + }) + if err != nil { + t.Fatal(err) + } + defer primary.Close() + + replica, err := blockvol.CreateBlockVol(replicaPath, blockvol.CreateOptions{ + VolumeSize: 64 * 1024 * 1024, + BlockSize: 4096, + WALSize: 16 * 1024 * 1024, + DurabilityMode: blockvol.DurabilitySyncAll, + }) + if err != nil { + t.Fatal(err) + } + defer replica.Close() + + // Assign roles. + primary.HandleAssignment(1, blockvol.RolePrimary, 30*time.Second) + replica.HandleAssignment(1, blockvol.RoleReplica, 30*time.Second) + + // Start replica receiver. + if err := replica.StartReplicaReceiver(":0", ":0"); err != nil { + t.Fatal(err) + } + recvAddr := replica.ReplicaReceiverAddr() + t.Logf("replica receiver: data=%s ctrl=%s", recvAddr.DataAddr, recvAddr.CtrlAddr) + + // Wire shipper. + primary.SetReplicaAddr(recvAddr.DataAddr, recvAddr.CtrlAddr) + + // Test 1: Immediate 4K write (like fio). + small := bytes.Repeat([]byte{0xAA}, 4096) + if err := primary.WriteLBA(0, small); err != nil { + t.Fatalf("immediate 4K write failed: %v", err) + } + t.Log("4K write: OK") + + // Test 2: Immediate 2MB write (like dd_write at offset). + big := make([]byte, 2*1024*1024) + for i := range big { + big[i] = byte(i & 0xFF) + } + if err := primary.WriteLBA(1024, big); err != nil { + t.Fatalf("immediate 2MB write failed: %v", err) + } + t.Log("2MB write: OK") + + // Test 3: After short delay, another 2MB write. + time.Sleep(2 * time.Second) + if err := primary.WriteLBA(2048, big); err != nil { + t.Fatalf("delayed 2MB write failed: %v", err) + } + t.Log("delayed 2MB write: OK") + + // Test 4: Read back from primary. + data, err := primary.ReadLBA(0, 4096) + if err != nil { + t.Fatalf("read 4K: %v", err) + } + if data[0] != 0xAA { + t.Fatalf("read mismatch: 0x%02x, want 0xAA", data[0]) + } + t.Log("readback: OK") +} + +// TestFreshRF2SyncAll_WriteDuringSyncAllBarrier tests write behavior +// when the sync_all barrier is active but the shipper hasn't connected yet. +func TestFreshRF2SyncAll_WriteDuringSyncAllBarrier(t *testing.T) { + path := t.TempDir() + "/primary.blk" + vol, err := blockvol.CreateBlockVol(path, blockvol.CreateOptions{ + VolumeSize: 64 * 1024 * 1024, + BlockSize: 4096, + WALSize: 16 * 1024 * 1024, + DurabilityMode: blockvol.DurabilitySyncAll, + }) + if err != nil { + t.Fatal(err) + } + defer vol.Close() + + // Assign as primary with replicas configured but NO receiver started. + // This simulates the window where the shipper knows about a replica + // but can't connect yet. + vol.HandleAssignment(1, blockvol.RolePrimary, 30*time.Second) + vol.SetReplicaAddr("127.0.0.1:1", "127.0.0.1:2") // dead addresses + + // Write should either succeed (degraded mode) or fail with a clear error. + data := bytes.Repeat([]byte{0xBB}, 4096) + err = vol.WriteLBA(0, data) + if err != nil { + t.Logf("write with dead replica: %v (sync_all barrier may block)", err) + // This is the expected behavior — sync_all with unreachable replica + // should return an error, not hang. + } else { + t.Log("write with dead replica succeeded — barrier must have degraded") + } +} diff --git a/weed/storage/blockvol/testrunner/actions/devops.go b/weed/storage/blockvol/testrunner/actions/devops.go index f256e90d0..9b6b5f651 100644 --- a/weed/storage/blockvol/testrunner/actions/devops.go +++ b/weed/storage/blockvol/testrunner/actions/devops.go @@ -713,6 +713,11 @@ func blockPromote(ctx context.Context, actx *tr.ActionContext, act tr.Action) (m if act.SaveAs != "" { actx.Vars[act.SaveAs+"_server"] = resp.NewPrimary actx.Vars[act.SaveAs+"_epoch"] = strconv.FormatUint(resp.Epoch, 10) + actx.Vars[act.SaveAs+"_reason"] = resp.Reason + actx.Vars[act.SaveAs+"_rejections_count"] = strconv.Itoa(len(resp.Rejections)) + if raw, err := json.Marshal(resp.Rejections); err == nil { + actx.Vars[act.SaveAs+"_rejections_json"] = string(raw) + } } return map[string]string{"value": resp.NewPrimary}, nil } @@ -798,13 +803,35 @@ func waitVolumeHealthy(ctx context.Context, actx *tr.ActionContext, act tr.Actio continue } + if ready, reason := volumeHealthyReady(info); !ready { + actx.Log(" poll %d: %s", poll, reason) + continue + } + actx.Log(" volume %q healthy after %d polls (RF=%d, mode=%s, degraded=%v)", - name, poll, info.ReplicaFactor, info.DurabilityMode, info.ReplicaDegraded) + name, poll, info.ReplicaFactor, info.VolumeMode, info.ReplicaDegraded) return map[string]string{"value": "healthy"}, nil } } } +func volumeHealthyReady(info *blockapi.VolumeInfo) (bool, string) { + if info == nil { + return false, "volume info missing" + } + if info.ReplicaFactor > 1 && info.DurabilityMode == "sync_all" && info.VolumeMode != "publish_healthy" { + mode := info.VolumeMode + if mode == "" { + mode = "unknown" + } + if info.VolumeModeReason != "" { + return false, fmt.Sprintf("volume_mode=%s (%s), waiting for publish_healthy", mode, info.VolumeModeReason) + } + return false, fmt.Sprintf("volume_mode=%s, waiting for publish_healthy", mode) + } + return true, "" +} + // discoverPrimary looks up a block volume and maps the primary's IP to a topology node name. // This solves the "which node to kill?" problem for degraded-mode and failover scenarios. // diff --git a/weed/storage/blockvol/testrunner/actions/devops_test.go b/weed/storage/blockvol/testrunner/actions/devops_test.go index d862da846..1f62ad321 100644 --- a/weed/storage/blockvol/testrunner/actions/devops_test.go +++ b/weed/storage/blockvol/testrunner/actions/devops_test.go @@ -2,9 +2,11 @@ package actions import ( "sort" + "strings" "testing" tr "github.com/seaweedfs/seaweedfs/weed/storage/blockvol/testrunner" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol/testrunner/internal/blockapi" ) func TestDevOpsActions_Registration(t *testing.T) { @@ -176,3 +178,60 @@ func TestK8sActions_TierGating(t *testing.T) { t.Errorf("k8s enabled: %v", err) } } + +func TestVolumeHealthyReady_AllowsSyncAllOnlyAfterPublishHealthy(t *testing.T) { + tests := []struct { + name string + info *blockapi.VolumeInfo + wantReady bool + wantReason string + }{ + { + name: "sync_all_waits_for_publish_healthy", + info: &blockapi.VolumeInfo{ + ReplicaFactor: 2, + DurabilityMode: "sync_all", + VolumeMode: "bootstrap_pending", + VolumeModeReason: "awaiting_shipper_connected", + }, + wantReady: false, + wantReason: "publish_healthy", + }, + { + name: "sync_all_publish_healthy_passes", + info: &blockapi.VolumeInfo{ + ReplicaFactor: 2, + DurabilityMode: "sync_all", + VolumeMode: "publish_healthy", + }, + wantReady: true, + }, + { + name: "best_effort_not_blocked_by_publish_mode", + info: &blockapi.VolumeInfo{ + ReplicaFactor: 2, + DurabilityMode: "best_effort", + VolumeMode: "bootstrap_pending", + }, + wantReady: true, + }, + { + name: "nil_info_rejected", + info: nil, + wantReady: false, + wantReason: "missing", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotReady, gotReason := volumeHealthyReady(tt.info) + if gotReady != tt.wantReady { + t.Fatalf("ready=%v, want %v (reason=%q)", gotReady, tt.wantReady, gotReason) + } + if tt.wantReason != "" && !strings.Contains(gotReason, tt.wantReason) { + t.Fatalf("reason=%q, want substring %q", gotReason, tt.wantReason) + } + }) + } +} diff --git a/weed/storage/blockvol/testrunner/scenarios/cp11b3-manual-promote.yaml b/weed/storage/blockvol/testrunner/scenarios/cp11b3-manual-promote.yaml index 4d9dadf30..d30d55d31 100644 --- a/weed/storage/blockvol/testrunner/scenarios/cp11b3-manual-promote.yaml +++ b/weed/storage/blockvol/testrunner/scenarios/cp11b3-manual-promote.yaml @@ -122,8 +122,25 @@ phases: name: "promote-test" reason: "T7 integration test: manual failover" save_as: promote_result + - action: assert_equal + actual: "{{ promote_result_rejections_count }}" + expected: "0" + # Restart the killed VS so it re-registers as replica. + # Without this, RF=2 with 1 VS alive can never reach publish_healthy. + - action: start_weed_volume + node: target_node + port: "18192" + master: "localhost:9435" + dir: "/tmp/sw-b3m-vs1" + extra_args: "-block.dir=/tmp/sw-b3m-vs1/blocks -block.listen=:3279 -ip=192.168.1.184" + save_as: vs1_new_pid + - action: sleep + duration: 10s + - action: wait_volume_healthy + name: "promote-test" + timeout: 120s - action: print - msg: "promoted to {{ promote_result_server }} epoch={{ promote_result_epoch }}" + msg: "promoted to {{ promote_result_server }} epoch={{ promote_result_epoch }} reason={{ promote_result_reason }}" # Phase 5: Verify promoted state - name: verify_promoted @@ -179,6 +196,10 @@ phases: node: target_node pid: "{{ vs1_pid }}" ignore_error: true + - action: stop_weed + node: target_node + pid: "{{ vs1_new_pid }}" + ignore_error: true - action: stop_weed node: target_node pid: "{{ master_pid }}" diff --git a/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml b/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml index af90c3ca9..7f1be37c4 100644 --- a/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml +++ b/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml @@ -4,8 +4,8 @@ timeout: 10m # Robust dimension: automatic failover after primary death. # # Flow: -# 1. Create RF=2 sync_all volume, record primary -# 2. Write data, disconnect iSCSI +# 1. Create RF=2 sync_all volume on the natural primary +# 2. Bootstrap first barrier with a real write, then record primary # 3. Kill primary VS (SIGKILL) # 4. Wait for lease expiry (30s TTL + margin) # 5. Verify: master auto-promotes replica to primary (no manual promote) @@ -90,24 +90,6 @@ phases: replica_factor: "2" durability_mode: "sync_all" - - action: wait_volume_healthy - name: "{{ volume_name }}" - timeout: 60s - - # Force primary to m02 so we have a known node to kill. - - action: block_promote - name: "{{ volume_name }}" - target_server: "10.0.0.3:18480" - force: "true" - reason: "failover-setup" - - - action: sleep - duration: 5s - - - action: wait_volume_healthy - name: "{{ volume_name }}" - timeout: 60s - - name: record-before actions: - action: discover_primary @@ -117,7 +99,9 @@ phases: - action: print msg: "Before: primary={{ before }} ({{ before_server }}), replica={{ before_replica_node }}" - # Write data so volume has real state. + # Bootstrap sync_all with a real write before requiring publish_healthy. + # Fresh RF=2 volumes do not become publish_healthy until the first + # barrier succeeds and establishes durable truth. - action: lookup_block_volume name: "{{ volume_name }}" save_as: vol @@ -139,6 +123,30 @@ phases: time_based: "true" name: pre-write + - action: dd_write + node: m01 + device: "{{ device }}" + bs: 1M + count: "2" + seek: "16" + save_as: pre_failover_md5 + + - action: dd_read_md5 + node: m01 + device: "{{ device }}" + bs: 1M + count: "2" + skip: "16" + save_as: pre_failover_verify + + - action: assert_equal + actual: "{{ pre_failover_verify }}" + expected: "{{ pre_failover_md5 }}" + + - action: wait_volume_healthy + name: "{{ volume_name }}" + timeout: 60s + - action: iscsi_cleanup node: m01 ignore_error: true @@ -146,8 +154,11 @@ phases: - name: kill-primary actions: - action: print - msg: "=== Killing primary on m02 ({{ before_server }}) ===" + msg: "=== Killing primary ({{ before_server }}) ===" + # The cluster starts m02 before m01, so the natural initial primary is + # the first server (m02 / vs1_pid). Keep the kill target fixed here and + # use discover_primary only as an evidence check. - action: exec node: m02 cmd: "kill -9 {{ vs1_pid }}" @@ -171,6 +182,10 @@ phases: timeout: 60s save_as: after + - action: wait_volume_healthy + name: "{{ volume_name }}" + timeout: 60s + - action: discover_primary name: "{{ volume_name }}" save_as: new_pri @@ -178,12 +193,6 @@ phases: - action: print msg: "After auto-failover: primary={{ new_pri }} ({{ new_pri_server }})" - # Verify primary actually changed. - - action: assert_block_field - name: "{{ volume_name }}" - field: volume_server - expected: "10.0.0.1:18480" - - action: print msg: "AUTO-FAILOVER VERIFIED: {{ before_server }} → {{ new_pri_server }}" @@ -198,6 +207,18 @@ phases: iqn: "{{ vol_iqn }}" save_as: device2 + - action: dd_read_md5 + node: m01 + device: "{{ device2 }}" + bs: 1M + count: "2" + skip: "16" + save_as: post_failover_md5 + + - action: assert_equal + actual: "{{ post_failover_md5 }}" + expected: "{{ pre_failover_md5 }}" + - action: fio_json node: m01 device: "{{ device2 }}" diff --git a/weed/storage/blockvol/testrunner/scenarios/internal/suite-ha-failover.yaml b/weed/storage/blockvol/testrunner/scenarios/internal/suite-ha-failover.yaml index 6a5336309..70a9a611d 100644 --- a/weed/storage/blockvol/testrunner/scenarios/internal/suite-ha-failover.yaml +++ b/weed/storage/blockvol/testrunner/scenarios/internal/suite-ha-failover.yaml @@ -114,16 +114,18 @@ phases: - name: wait-failover actions: - - action: sleep - duration: 40s + - action: wait_block_primary + name: "{{ volume_name }}" + not: "{{ primary_before }}" + timeout: 60s + save_as: primary_after - action: assert_block_field name: "{{ volume_name }}" field: epoch save_as: epoch_after - - action: assert_block_field + - action: wait_volume_healthy name: "{{ volume_name }}" - field: volume_server - save_as: primary_after + timeout: 60s - action: print msg: "After failover: primary={{ primary_after }} epoch={{ epoch_after }}" - action: assert_greater diff --git a/weed/storage/blockvol/testrunner/scenarios/public/cp11b3-manual-promote.yaml b/weed/storage/blockvol/testrunner/scenarios/public/cp11b3-manual-promote.yaml index 4d9dadf30..51a44d1c9 100644 --- a/weed/storage/blockvol/testrunner/scenarios/public/cp11b3-manual-promote.yaml +++ b/weed/storage/blockvol/testrunner/scenarios/public/cp11b3-manual-promote.yaml @@ -122,8 +122,14 @@ phases: name: "promote-test" reason: "T7 integration test: manual failover" save_as: promote_result + - action: assert_equal + actual: "{{ promote_result_rejections_count }}" + expected: "0" + - action: wait_volume_healthy + name: "promote-test" + timeout: 60s - action: print - msg: "promoted to {{ promote_result_server }} epoch={{ promote_result_epoch }}" + msg: "promoted to {{ promote_result_server }} epoch={{ promote_result_epoch }} reason={{ promote_result_reason }}" # Phase 5: Verify promoted state - name: verify_promoted