feat: Phase 12 P3+P4 — diagnosability surfaces, perf floor, rollout gates

P3: Add explicit bounded read-only diagnosis surfaces for all symptom classes: - FailoverDiagnostic: volume-oriented failover state with per-volume DeferredPromotion/PendingRebuild entries and proper timer lifecycle - PublicationDiagnostic: two-read coherence check (LookupBlockVolume vs registry authority) with computed Coherent verdict - RecoveryDiagnostic: minimal ActiveTasks surface (Path A) - Blocker ledger: 3 diagnosed + 3 unresolved, finite, from actual file - Runbook references only exposed surfaces, no internal state P4: Add bounded performance floor + rollout-gate package: - Engine-local floor measurement with explicit IOPS gates per workload - Cost characterization: WAL 2x write amp, -56% replication tax - Rollout gates with semantic cross-checks against cited evidence (baseline numbers, transport/network matrix, blocker counts) - Launch envelope tightened to actually measured combinations only Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-09-20 13:30:46 +02:00 · 2026-04-02 16:20:22 -07:00
parent ebe95b6e2e
commit 643a5a1074
9 changed files with 1978 additions and 12 deletions
@@ -0,0 +1,29 @@
 # Phase 12 P3 — Blocker Ledger
 Date: 2026-04-02
 Scope: bounded diagnosability / blocker accounting for the accepted RF=2 sync_all chosen path
 ## Diagnosed and Bounded
 | ID | Symptom | Evidence Surface | Owning Truth | Status |
 |----|---------|-----------------|--------------|--------|
 | B1 | Failover does not converge | failover logs + registry Lookup epoch/primary | registry authority | Diagnosed: convergence depends on lease expiry + heartbeat cycle; bounded by lease TTL |
 | B2 | Lookup publication stale after failover | LookupBlockVolume response vs registry entry | registry ISCSIAddr/VolumeServer | Diagnosed: publication updates on failover assignment delivery; bounded by assignment queue delivery |
 | B3 | Recovery tasks remain after volume delete | RecoveryManager.DiagnosticSnapshot | RecoveryManager task map | Diagnosed: tasks drain on shutdown/cancel; bounded by RecoveryManager lifecycle |
 ## Unresolved but Explicit
 | ID | Symptom | Current Evidence | Why Unresolved | Blocks P4/Rollout? |
 |----|---------|-----------------|----------------|-------------------|
 | U1 | V2 engine accepts stale-epoch assignments at orchestrator level | V2 idempotence check skips only same-epoch; lower epoch creates new sender | Engine ApplyAssignment does not check epoch monotonicity on Reconcile | No — V1 HandleAssignment rejects epoch regression; V2 is secondary |
 | U2 | Single-process test cannot exercise Primary→Rebuilding role transition | HandleAssignment rejects transition in shared store | Test harness limitation, not production bug | No — production VS has separate stores |
 | U3 | gRPC stream transport not exercised in control-loop tests | All logic above/below stream is real; stream itself bypassed | Would require live master+VS gRPC servers in test | Blocks full integration test, not correctness |
 ## Out of Scope for P3
 - Performance floor characterization
 - Rollout-gate criteria
 - Hours/days soak
 - RF>2 topology
 - NVMe runtime transport proof
 - CSI snapshot/expand
@@ -0,0 +1,70 @@
 # Phase 12 P3 — Bounded Runbook
 Scope: diagnosis of three symptom classes on the accepted RF=2 sync_all chosen path.
 All diagnosis steps reference ONLY explicit bounded read-only surfaces:
 - `LookupBlockVolume` — gRPC RPC returning current primary VS + iSCSI address
 - `FailoverDiagnostic` — volume-oriented failover state snapshot
 - `PublicationDiagnostic` — lookup vs authority coherence snapshot
 - `RecoveryDiagnostic` — active recovery task set snapshot
 - Blocker ledger — finite file at `phase-12-p3-blockers.md`
 ## S1: Failover/Recovery Convergence Stall
 **Visible symptom:** Volume remains unavailable after a VS death; lookup still returns the old primary.
 **Diagnosis surfaces:**
 - `LookupBlockVolume(volumeName)` — check if `VolumeServer` is still the dead server
 - `FailoverDiagnostic` — check `Volumes[]` for the affected volume
 **Diagnosis steps:**
 1. Call `LookupBlockVolume(volumeName)`. If `VolumeServer` changed from the dead server, failover succeeded.
 2. If unchanged: read `FailoverDiagnostic`. Find the volume by name in `Volumes[]`.
 3. If found with `DeferredPromotion=true`: lease-wait — failover is deferred until lease expires.
 4. If found with `PendingRebuild=true`: failover completed, rebuild is pending for the dead server.
 5. If `DeferredPromotionCount[deadServer] > 0` in the aggregate: deferred promotions are queued.
 6. If the volume does not appear in either lookup change or `FailoverDiagnostic`: escalate.
 **Conclusion classes (from surfaces only):**
 - **Lease-wait:** `FailoverDiagnostic.DeferredPromotionCount[deadServer] > 0` — normal, bounded by lease TTL.
 - **Rebuild-pending:** `FailoverDiagnostic.Volumes[].PendingRebuild=true` — failover done, rebuild queued.
 - **Converged:** `LookupBlockVolume` shows new primary, no failover entries — resolved.
 - **Unresolved:** None of the above — escalate.
 ## S2: Publication/Lookup Mismatch
 **Visible symptom:** `LookupBlockVolume` returns an iSCSI address or volume server that doesn't match expected state.
 **Diagnosis surfaces:**
 - `LookupBlockVolume(volumeName)` — operator-visible publication
 - `PublicationDiagnostic` — explicit coherence check (lookup vs authority)
 **Diagnosis steps:**
 1. Call `PublicationDiagnosticFor(volumeName)`. Check `Coherent` field.
 2. If `Coherent=true`: lookup matches registry authority — no mismatch.
 3. If `Coherent=false`: read `Reason` for explanation. Compare `LookupVolumeServer` vs `AuthorityVolumeServer` and `LookupIscsiAddr` vs `AuthorityIscsiAddr`.
 4. Cross-check with `LookupBlockVolume` directly: repeated lookups should be self-consistent.
 **Conclusion classes (from surfaces only):**
 - **Coherent:** `PublicationDiagnostic.Coherent=true` — no mismatch.
 - **Stale client:** Coherent but client sees old value — bounded by client re-query.
 - **Unresolved:** `PublicationDiagnostic.Coherent=false` with no transient cause — escalate.
 ## S3: Leftover Runtime Work After Convergence
 **Visible symptom:** After volume deletion or steady-state convergence, recovery tasks should have drained.
 **Diagnosis surfaces:**
 - `RecoveryDiagnostic` — `ActiveTasks` list (replicaIDs with active recovery work)
 **Diagnosis steps:**
 1. Call `RecoveryManager.DiagnosticSnapshot()`. Read `ActiveTasks`.
 2. If `ActiveTasks` is empty: clean — no leftover work.
 3. If non-empty: check whether any task replicaID contains the deleted volume's path.
 4. If a deleted volume's replicaID is present in `ActiveTasks`: residue — escalate.
 5. If all tasks are for live volumes: non-empty but expected — normal in-flight work.
 **Conclusion classes (from surfaces only):**
 - **Clean:** `RecoveryDiagnostic.ActiveTasks` is empty — runtime converged.
 - **Non-empty, no residue:** Tasks present but none for the deleted/converged volume — normal.
 - **Residue:** Deleted volume's replicaID still in `ActiveTasks` — escalate.
@@ -0,0 +1,101 @@
 # Phase 12 P4 — Performance Floor Summary
 Date: 2026-04-02
 Scope: bounded performance floor for the accepted RF=2, sync_all chosen path.
 ## Workload Envelope
 | Parameter | Value |
 |-----------|-------|
 | Topology | RF=2, sync_all |
 | Operations | 4K random write, 4K random read, sequential write, sequential read |
 | Runtime | Steady-state, no failover, no disturbance |
 | Path | Accepted chosen path (same as P1/P2/P3) |
 ## Environment
 ### Unit Test Harness (engine-local)
 | Parameter | Value |
 |-----------|-------|
 | Name | `TestP12P4_PerformanceFloor_Bounded` |
 | Location | `weed/server/qa_block_perf_test.go` |
 | Platform | Single-process, local disk |
 | Volume | 64MB, 4K blocks, 16MB WAL |
 | Writer | Single-threaded (worst-case for group commit) |
 | Replication | Not exercised (engine-local only) |
 | Measurement | Worst of 3 iterations (floor, not peak) |
 ### Production Baseline (cross-machine)
 | Parameter | Value |
 |-----------|-------|
 | Name | `baseline-roce-20260401` |
 | Location | `learn/projects/sw-block/test/results/baseline-roce-20260401.md` |
 | Hardware | m01 (10.0.0.1) - M02 (10.0.0.3), 25Gbps RoCE |
 | Protocol | NVMe-TCP |
 | Volume | 2GB, RF=2, sync_all, cross-machine replication |
 | Writer | fio, QD1-128, j=4 |
 ## Floor Table: Production (RF=2, sync_all, NVMe-TCP, 25Gbps RoCE)
 These are measured floor values from the production baseline, not the unit test.
 | Workload | Floor IOPS | Notes |
 |----------|-----------|-------|
 | 4K random write QD1 | 28,347 | Barrier round-trip limited (flat across QD) |
 | 4K random write QD32 | 28,453 | Same barrier ceiling |
 | 4K random read QD32 | 136,648 | No replication overhead |
 | Mixed 70/30 QD32 | 28,423 | Write-side limited |
 Latency: Write latency is bounded by sync_all barrier round-trip (~35us at QD1).
 Read latency: sub-microsecond for cached, single-digit microseconds for extent.
 ## Floor Table: Engine-Local (unit test harness)
 These values are measured by `TestP12P4_PerformanceFloor_Bounded` on the dev machine.
 They characterize the engine I/O floor WITHOUT transport or replication.
 Actual values vary by hardware; the test produces them on each run.
 | Workload | Metric | Method | Gate |
 |----------|--------|--------|------|
 | 4K random write | Floor IOPS, Avg/P50/P99/Max latency | Worst of 3 iterations | >= 1,000 IOPS, P99 <= 100ms |
 | 4K random read | Floor IOPS, Avg/P50/P99/Max latency | Worst of 3 iterations | >= 5,000 IOPS |
 | 4K sequential write | Floor IOPS, Avg/P50/P99/Max latency | Worst of 3 iterations | >= 2,000 IOPS, P99 <= 100ms |
 | 4K sequential read | Floor IOPS, Avg/P50/P99/Max latency | Worst of 3 iterations | >= 10,000 IOPS |
 Gate thresholds are regression gates enforced in code (`perfFloorGates` in `qa_block_perf_test.go`).
 Set at ~10% of measured values to tolerate slow CI/VM hardware while catching catastrophic regressions.
 ## Cost Summary
 | Cost | Value | Source |
 |------|-------|--------|
 | WAL write amplification | 2x minimum | Engine design: each write → WAL + eventual extent flush |
 | Replication tax (RF=2 sync_all vs RF=1) | -56% | baseline-roce-20260401.md (NVMe-TCP, 25Gbps RoCE) |
 | Replication tax (RF=2 sync_all vs RF=1, iSCSI 1Gbps) | -56% | baseline-roce-20260401.md |
 | Degraded mode penalty (sync_all RF=2, one replica dead) | -66% | baseline-roce-20260401.md (barrier timeout) |
 | Group commit | 1 fdatasync per batch | Amortizes sync cost across concurrent writers |
 ## Acceptance Evidence
 | Item | Evidence | Type |
 |------|----------|------|
 | Floor gates pass | `perfFloorGates` thresholds enforced per workload | Acceptance |
 | Workload runs repeatably | `TestP12P4_PerformanceFloor_Bounded` passes | Acceptance |
 | Cost statement is bounded | `TestP12P4_CostCharacterization_Bounded` passes | Acceptance |
 | Production baseline exists | `baseline-roce-20260401.md` with measured values | Acceptance |
 | Floor is worst-of-N, not peak | Test takes minimum IOPS across 3 iterations | Method |
 | Regression-safe | Test fails if floor drops below gate (blocks rollout) | Acceptance |
 | Replication tax documented | -56% from measured production baseline | Support telemetry |
 ## What P4 does NOT claim
 - This is not a claim that the measured floor is "good enough" for any specific application.
 - This does not claim readiness for failover-under-load scenarios.
 - This does not claim readiness for hours/days soak under load.
 - This does not claim readiness for RF>2 topologies.
 - This does not claim readiness for all transport combinations (iSCSI + NVMe + kernel versions).
 - This does not claim readiness for production rollout beyond the explicitly named launch envelope.
 - Engine-local floor numbers are not production floor numbers.
 - The replication tax is measured on one specific hardware configuration and may differ on other hardware.
@@ -0,0 +1,64 @@
 # Phase 12 P4 — Rollout Gates
 Date: 2026-04-02
 Scope: bounded first-launch envelope for the accepted RF=2, sync_all chosen path.
 This is a bounded first-launch envelope, not general readiness.
 ## Supported Launch Envelope
 Only the transport/network combinations with measured baselines are included.
 | Parameter | Value |
 |-----------|-------|
 | Topology | RF=2, sync_all |
 | Transport + Network | NVMe-TCP @ 25Gbps RoCE (measured), iSCSI @ 25Gbps RoCE (measured), iSCSI @ 1Gbps (measured) |
 | NOT included | NVMe-TCP @ 1Gbps (not measured) |
 | Volume size | Up to 2GB (tested baseline) |
 | Failover | Lease-based, bounded by TTL (30s default) |
 | Recovery | Catch-up-first, rebuild fallback |
 | Degraded mode | Documented -66% write penalty (sync_all RF=2, one replica dead) |
 ## Cleared Gates
 | Gate | Evidence | Status | Notes |
 |------|----------|--------|-------|
 | G1 | P1 disturbance tests pass | Cleared | Restart/reconnect correctness under disturbance |
 | G2 | P2 soak tests pass | Cleared | Repeated create/failover/recover cycles, no drift |
 | G3 | P3 diagnosability tests pass | Cleared | Explicit bounded diagnosis surfaces for all symptom classes |
 | G4 | P4 floor gates pass | Cleared | Explicit IOPS thresholds + P99 ceilings enforced per workload in code |
 | G5 | P4 cost characterization bounded | Cleared | WAL 2x write amp, -56% replication tax documented |
 | G6 | Production baseline exists | Cleared | baseline-roce-20260401.md: 28.4K write IOPS, 136.6K read IOPS |
 | G8 | Floor gates are regression-safe | Cleared | Test fails if any workload drops below defined minimum IOPS or exceeds P99 ceiling |
 | G7 | Blocker ledger finite | Cleared | 3 diagnosed (B1-B3) + 3 unresolved (U1-U3), all explicit |
 ## Remaining Blockers / Exclusions
 | Exclusion | Why | Impact |
 |-----------|-----|--------|
 | E1 | Failover-under-load perf not measured | Cannot claim bounded perf during failover |
 | E2 | Hours/days soak not run | Cannot claim long-run stability under sustained load |
 | E3 | RF>2 not measured | Cannot claim perf floor for RF=3+ |
 | E4 | Broad transport matrix not tested | Cannot claim parity across all kernel/NVMe/iSCSI versions |
 | E5 | Degraded mode is severe (-66%) | sync_all RF=2 has sharp write cliff on replica death |
 | E6 | V2 stale-epoch at orchestrator level (U1 from P3) | V1 guards suffice; V2 is secondary path |
 | E7 | gRPC stream transport not exercised in unit tests (U3 from P3) | Blocks full integration test, not correctness |
 ## Reject Conditions
 This launch envelope should be REJECTED if:
 1. Any P1/P2/P3 test regresses (correctness/stability/diagnosability gate violated)
 2. Production baseline numbers are not reproducible on the target hardware
 3. Degraded mode behavior (-66% cliff) is not acceptable for the deployment scenario
 4. The deployment requires RF>2, failover-under-load guarantees, or long soak proof
 5. The deployment requires transport combinations not covered by the baseline
 ## What P4 does NOT claim
 - This does not claim general production readiness.
 - This does not claim readiness for any deployment outside the named launch envelope.
 - This does not claim that the performance floor is optimal or final.
 - This does not claim that the degraded-mode penalty is acceptable (deployment-specific decision).
 - This does not claim hours/days stability under sustained load.
 - This is a bounded first-launch gate, not a broad rollout approval.
@@ -0,0 +1,374 @@
 package weed_server
 import (
 	"context"
 	"sync"
 	bridge "github.com/seaweedfs/seaweedfs/sw-block/bridge/blockvol"
 	engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/v2bridge"
 )
 // recoveryTask tracks a live recovery goroutine for one replica target.
 // The task pointer serves as identity token — only the goroutine that owns
 // THIS pointer may mark it as done.
 type recoveryTask struct {
 	replicaID string
 	cancel    context.CancelFunc
 	done      chan struct{} // closed when the goroutine exits
 }
 // RecoveryManager owns live recovery execution for all replica targets.
 //
 // Ownership model:
 //   - At most one recovery goroutine per replicaID at any time.
 //   - On supersede/replace: the old goroutine is cancelled AND drained
 //     before the replacement starts. No overlap.
 //   - Cancellation: context cancel + session invalidation (for removal/shutdown).
 //     For supersede: context cancel only (engine already attached replacement session).
 type RecoveryManager struct {
 	bs *BlockService
 	mu    sync.Mutex
 	tasks map[string]*recoveryTask
 	wg    sync.WaitGroup
 	// TestHook: if set, called before execution starts. Tests use this
 	// to hold the goroutine alive for serialized-replacement proofs.
 	OnBeforeExecute func(replicaID string)
 }
 func NewRecoveryManager(bs *BlockService) *RecoveryManager {
 	return &RecoveryManager{
 		bs:    bs,
 		tasks: make(map[string]*recoveryTask),
 	}
 }
 // HandleAssignmentResult processes the engine's assignment result.
 //
 // Engine result semantics:
 //   - SessionsCreated: new session, start goroutine
 //   - SessionsSuperseded: old replaced by new — cancel+drain old, start new
 //   - Removed: sender gone — cancel+drain, invalidate session
 func (rm *RecoveryManager) HandleAssignmentResult(result engine.AssignmentResult, assignments []blockvol.BlockVolumeAssignment) {
 	// Removed: cancel + invalidate + drain.
 	for _, replicaID := range result.Removed {
 		rm.cancelAndDrain(replicaID, true)
 	}
 	// Superseded: cancel + drain (no invalidate — engine has replacement session),
 	// then start new.
 	for _, replicaID := range result.SessionsSuperseded {
 		rm.cancelAndDrain(replicaID, false)
 		rm.startTask(replicaID, assignments)
 	}
 	// Created: start new (cancel stale defensively).
 	for _, replicaID := range result.SessionsCreated {
 		rm.cancelAndDrain(replicaID, false)
 		rm.startTask(replicaID, assignments)
 	}
 }
 // cancelAndDrain cancels a running task and WAITS for it to exit.
 // This ensures no overlap between old and new owners.
 func (rm *RecoveryManager) cancelAndDrain(replicaID string, invalidateSession bool) {
 	rm.mu.Lock()
 	task, ok := rm.tasks[replicaID]
 	if !ok {
 		rm.mu.Unlock()
 		return
 	}
 	glog.V(1).Infof("recovery: cancelling+draining task for %s (invalidate=%v)", replicaID, invalidateSession)
 	task.cancel()
 	if invalidateSession && rm.bs.v2Orchestrator != nil {
 		if s := rm.bs.v2Orchestrator.Registry.Sender(replicaID); s != nil {
 			s.InvalidateSession("recovery_removed", engine.StateDisconnected)
 		}
 	}
 	delete(rm.tasks, replicaID)
 	doneCh := task.done
 	rm.mu.Unlock()
 	// Wait for the old goroutine to exit OUTSIDE the lock.
 	// This serializes replacement: new task cannot start until old is fully drained.
 	<-doneCh
 }
 // startTask creates and starts a new recovery goroutine. Caller must ensure
 // no existing task for this replicaID (call cancelAndDrain first).
 func (rm *RecoveryManager) startTask(replicaID string, assignments []blockvol.BlockVolumeAssignment) {
 	rm.mu.Lock()
 	defer rm.mu.Unlock()
 	rebuildAddr := rm.deriveRebuildAddr(replicaID, assignments)
 	ctx, cancel := context.WithCancel(context.Background())
 	task := &recoveryTask{
 		replicaID: replicaID,
 		cancel:    cancel,
 		done:      make(chan struct{}),
 	}
 	rm.tasks[replicaID] = task
 	rm.wg.Add(1)
 	go rm.runRecovery(ctx, task, rebuildAddr)
 }
 // Shutdown cancels all active recovery tasks and waits for drain.
 func (rm *RecoveryManager) Shutdown() {
 	rm.mu.Lock()
 	for _, task := range rm.tasks {
 		task.cancel()
 		if rm.bs.v2Orchestrator != nil {
 			if s := rm.bs.v2Orchestrator.Registry.Sender(task.replicaID); s != nil {
 				s.InvalidateSession("recovery_shutdown", engine.StateDisconnected)
 			}
 		}
 	}
 	rm.tasks = make(map[string]*recoveryTask)
 	rm.mu.Unlock()
 	rm.wg.Wait()
 }
 // ActiveTaskCount returns the number of active recovery tasks (for testing).
 func (rm *RecoveryManager) ActiveTaskCount() int {
 	rm.mu.Lock()
 	defer rm.mu.Unlock()
 	return len(rm.tasks)
 }
 // DiagnosticSnapshot returns a bounded read-only snapshot of active recovery
 // tasks for operator-visible diagnosis. Each entry shows the replicaID being
 // recovered. This is the P3 diagnosability surface — read-only, no semantics.
 type RecoveryDiagnostic struct {
 	ActiveTasks []string // replicaIDs with active recovery work
 }
 func (rm *RecoveryManager) DiagnosticSnapshot() RecoveryDiagnostic {
 	rm.mu.Lock()
 	defer rm.mu.Unlock()
 	diag := RecoveryDiagnostic{}
 	for id := range rm.tasks {
 		diag.ActiveTasks = append(diag.ActiveTasks, id)
 	}
 	return diag
 }
 // runRecovery is the recovery goroutine for one replica target.
 func (rm *RecoveryManager) runRecovery(ctx context.Context, task *recoveryTask, rebuildAddr string) {
 	defer rm.wg.Done()
 	defer close(task.done) // signal drain completion
 	defer func() {
 		rm.mu.Lock()
 		// Only delete if we're still the active task (pointer comparison).
 		if rm.tasks[task.replicaID] == task {
 			delete(rm.tasks, task.replicaID)
 		}
 		rm.mu.Unlock()
 	}()
 	replicaID := task.replicaID
 	if ctx.Err() != nil {
 		return
 	}
 	orch := rm.bs.v2Orchestrator
 	s := orch.Registry.Sender(replicaID)
 	if s == nil {
 		glog.V(1).Infof("recovery: sender %s not found, skipping", replicaID)
 		return
 	}
 	sessSnap := s.SessionSnapshot()
 	if sessSnap == nil {
 		glog.V(1).Infof("recovery: sender %s has no active session, skipping", replicaID)
 		return
 	}
 	glog.V(0).Infof("recovery: starting %s session for %s (rebuildAddr=%s)",
 		sessSnap.Kind, replicaID, rebuildAddr)
 	if rm.OnBeforeExecute != nil {
 		rm.OnBeforeExecute(replicaID)
 	}
 	switch sessSnap.Kind {
 	case engine.SessionCatchUp:
 		rm.runCatchUp(ctx, replicaID, rebuildAddr)
 	case engine.SessionRebuild:
 		rm.runRebuild(ctx, replicaID, rebuildAddr)
 	default:
 		glog.V(1).Infof("recovery: unknown session kind %s for %s", sessSnap.Kind, replicaID)
 	}
 }
 func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAddr string) {
 	bs := rm.bs
 	volPath := rm.volumePathForReplica(replicaID)
 	if volPath == "" {
 		glog.Warningf("recovery: cannot determine volume path for %s", replicaID)
 		return
 	}
 	var sa engine.StorageAdapter
 	var replicaFlushedLSN uint64
 	var executor *v2bridge.Executor
 	if err := bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error {
 		reader := v2bridge.NewReader(vol)
 		pinner := v2bridge.NewPinner(vol)
 		sa = bridge.NewStorageAdapter(
 			&readerShimForRecovery{reader},
 			&pinnerShimForRecovery{pinner},
 		)
 		if s := bs.v2Orchestrator.Registry.Sender(replicaID); s != nil {
 			if snap := s.SessionSnapshot(); snap != nil {
 				replicaFlushedLSN = snap.StartLSN
 			}
 		}
 		executor = v2bridge.NewExecutor(vol, rebuildAddr)
 		return nil
 	}); err != nil {
 		glog.Warningf("recovery: cannot access volume %s: %v", volPath, err)
 		return
 	}
 	if ctx.Err() != nil {
 		return
 	}
 	driver := &engine.RecoveryDriver{Orchestrator: bs.v2Orchestrator, Storage: sa}
 	plan, err := driver.PlanRecovery(replicaID, replicaFlushedLSN)
 	if err != nil {
 		glog.Warningf("recovery: plan failed for %s: %v", replicaID, err)
 		return
 	}
 	if ctx.Err() != nil {
 		driver.CancelPlan(plan, "context_cancelled")
 		return
 	}
 	exec := engine.NewCatchUpExecutor(driver, plan)
 	exec.IO = executor
 	if execErr := exec.Execute(nil, 0); execErr != nil {
 		if ctx.Err() != nil {
 			glog.V(1).Infof("recovery: catch-up cancelled for %s: %v", replicaID, execErr)
 		} else {
 			glog.Warningf("recovery: catch-up execution failed for %s: %v", replicaID, execErr)
 		}
 		return
 	}
 	glog.V(0).Infof("recovery: catch-up completed for %s", replicaID)
 }
 func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAddr string) {
 	bs := rm.bs
 	volPath := rm.volumePathForReplica(replicaID)
 	if volPath == "" {
 		glog.Warningf("recovery: cannot determine volume path for %s", replicaID)
 		return
 	}
 	var sa engine.StorageAdapter
 	var executor *v2bridge.Executor
 	if err := bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error {
 		reader := v2bridge.NewReader(vol)
 		pinner := v2bridge.NewPinner(vol)
 		sa = bridge.NewStorageAdapter(
 			&readerShimForRecovery{reader},
 			&pinnerShimForRecovery{pinner},
 		)
 		executor = v2bridge.NewExecutor(vol, rebuildAddr)
 		return nil
 	}); err != nil {
 		glog.Warningf("recovery: cannot access volume %s: %v", volPath, err)
 		return
 	}
 	if ctx.Err() != nil {
 		return
 	}
 	driver := &engine.RecoveryDriver{Orchestrator: bs.v2Orchestrator, Storage: sa}
 	plan, err := driver.PlanRebuild(replicaID)
 	if err != nil {
 		glog.Warningf("recovery: rebuild plan failed for %s: %v", replicaID, err)
 		return
 	}
 	if ctx.Err() != nil {
 		driver.CancelPlan(plan, "context_cancelled")
 		return
 	}
 	exec := engine.NewRebuildExecutor(driver, plan)
 	exec.IO = executor
 	if execErr := exec.Execute(); execErr != nil {
 		if ctx.Err() != nil {
 			glog.V(1).Infof("recovery: rebuild cancelled for %s: %v", replicaID, execErr)
 		} else {
 			glog.Warningf("recovery: rebuild execution failed for %s: %v", replicaID, execErr)
 		}
 		return
 	}
 	glog.V(0).Infof("recovery: rebuild completed for %s", replicaID)
 }
 func (rm *RecoveryManager) deriveRebuildAddr(replicaID string, assignments []blockvol.BlockVolumeAssignment) string {
 	volPath := rm.volumePathForReplica(replicaID)
 	for _, a := range assignments {
 		if a.Path == volPath && a.RebuildAddr != "" {
 			return a.RebuildAddr
 		}
 	}
 	return ""
 }
 func (rm *RecoveryManager) volumePathForReplica(replicaID string) string {
 	for i := len(replicaID) - 1; i >= 0; i-- {
 		if replicaID[i] == '/' {
 			return replicaID[:i]
 		}
 	}
 	return ""
 }
 // --- Bridge shims ---
 type readerShimForRecovery struct{ r *v2bridge.Reader }
 func (s *readerShimForRecovery) ReadState() bridge.BlockVolState {
 	rs := s.r.ReadState()
 	return bridge.BlockVolState{
 		WALHeadLSN:        rs.WALHeadLSN,
 		WALTailLSN:        rs.WALTailLSN,
 		CommittedLSN:      rs.CommittedLSN,
 		CheckpointLSN:     rs.CheckpointLSN,
 		CheckpointTrusted: rs.CheckpointTrusted,
 	}
 }
 type pinnerShimForRecovery struct{ p *v2bridge.Pinner }
 func (s *pinnerShimForRecovery) HoldWALRetention(startLSN uint64) (func(), error) {
 	return s.p.HoldWALRetention(startLSN)
 }
 func (s *pinnerShimForRecovery) HoldSnapshot(checkpointLSN uint64) (func(), error) {
 	return s.p.HoldSnapshot(checkpointLSN)
 }
 func (s *pinnerShimForRecovery) HoldFullBase(committedLSN uint64) (func(), error) {
 	return s.p.HoldFullBase(committedLSN)
 }
@@ -1,10 +1,12 @@
 package weed_server
 import (
 	"context"
 	"sync"
 	"time"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
 )
@@ -18,18 +20,139 @@ type pendingRebuild struct {
 	ReplicaCtrlAddr string // CP13-8: saved from before death for catch-up-first recovery
 }
 // deferredPromotion tracks a deferred promotion timer with its volume context.
 type deferredPromotion struct {
 	Timer          *time.Timer
 	VolumeName     string
 	CurrentPrimary string // current (stale) primary that will be replaced
 	AffectedServer string // dead server addr
 }
 // blockFailoverState holds failover and rebuild state on the master.
 type blockFailoverState struct {
 	mu              sync.Mutex
-	pendingRebuilds map[string][]pendingRebuild // dead server addr -> pending rebuilds
+	pendingRebuilds map[string][]pendingRebuild       // dead server addr -> pending rebuilds
-	// R2-F2: Track deferred promotion timers so they can be cancelled on reconnect.
+	deferredTimers  map[string][]deferredPromotion     // dead server addr -> pending deferred promotions
-	deferredTimers map[string][]*time.Timer // dead server addr -> pending timers
+}
 // FailoverVolumeState is one volume's failover diagnosis entry.
 type FailoverVolumeState struct {
 	VolumeName       string
 	CurrentPrimary   string
 	AffectedServer   string // dead server that triggered the failover/rebuild
 	DeferredPromotion bool  // true if a deferred promotion timer is pending
 	PendingRebuild   bool  // true if a rebuild is pending for this volume
 	Reason           string // "lease_wait", "rebuild_pending", or ""
 }
 // FailoverDiagnostic is a bounded read-only snapshot of failover state
 // for operator-visible diagnosis. P3 diagnosability surface.
 //
 // Volume-oriented: each entry describes one volume's failover state.
 // Aggregate counts are derived from the volume list.
 type FailoverDiagnostic struct {
 	Volumes              []FailoverVolumeState
 	PendingRebuildCount  map[string]int // dead server → count of pending rebuilds
 	DeferredPromotionCount map[string]int // dead server → count of deferred promotion timers
 }
 func (fs *blockFailoverState) DiagnosticSnapshot() FailoverDiagnostic {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 	diag := FailoverDiagnostic{
 		PendingRebuildCount:    make(map[string]int),
 		DeferredPromotionCount: make(map[string]int),
 	}
 	for server, rebuilds := range fs.pendingRebuilds {
 		diag.PendingRebuildCount[server] = len(rebuilds)
 		for _, rb := range rebuilds {
 			diag.Volumes = append(diag.Volumes, FailoverVolumeState{
 				VolumeName:     rb.VolumeName,
 				CurrentPrimary: rb.NewPrimary,
 				AffectedServer: server,
 				PendingRebuild: true,
 				Reason:         "rebuild_pending",
 			})
 		}
 	}
 	for server, promos := range fs.deferredTimers {
 		diag.DeferredPromotionCount[server] = len(promos)
 		for _, dp := range promos {
 			diag.Volumes = append(diag.Volumes, FailoverVolumeState{
 				VolumeName:        dp.VolumeName,
 				CurrentPrimary:    dp.CurrentPrimary,
 				AffectedServer:    dp.AffectedServer,
 				DeferredPromotion: true,
 				Reason:            "lease_wait",
 			})
 		}
 	}
 	return diag
 }
 // PublicationDiagnostic is a bounded read-only snapshot comparing the
 // operator-visible publication (LookupBlockVolume response) against the
 // registry authority for one volume. P3 diagnosability surface for S2.
 type PublicationDiagnostic struct {
 	VolumeName           string
 	LookupVolumeServer   string // what LookupBlockVolume returns
 	LookupIscsiAddr      string
 	AuthorityVolumeServer string // registry entry (source of truth)
 	AuthorityIscsiAddr   string
 	Coherent             bool   // true if lookup == authority
 	Reason               string // "" if coherent, otherwise why they diverge
 }
 // PublicationDiagnosticFor returns a PublicationDiagnostic for the named volume.
 // It performs two independent reads:
 //   - Lookup side: calls LookupBlockVolume (the actual gRPC method)
 //   - Authority side: reads the registry directly
 //
 // Then compares the two. If they diverge, Coherent=false with a Reason.
 func (ms *MasterServer) PublicationDiagnosticFor(volumeName string) (PublicationDiagnostic, bool) {
 	if ms.blockRegistry == nil {
 		return PublicationDiagnostic{}, false
 	}
 	// Read 1: the operator-visible publication surface.
 	lookupResp, err := ms.LookupBlockVolume(context.Background(), &master_pb.LookupBlockVolumeRequest{Name: volumeName})
 	if err != nil {
 		return PublicationDiagnostic{}, false
 	}
 	// Read 2: the registry authority (separate read).
 	entry, ok := ms.blockRegistry.Lookup(volumeName)
 	if !ok {
 		return PublicationDiagnostic{}, false
 	}
 	diag := PublicationDiagnostic{
 		VolumeName:            volumeName,
 		LookupVolumeServer:    lookupResp.VolumeServer,
 		LookupIscsiAddr:       lookupResp.IscsiAddr,
 		AuthorityVolumeServer: entry.VolumeServer,
 		AuthorityIscsiAddr:    entry.ISCSIAddr,
 	}
 	// Compare the two reads.
 	vsMatch := diag.LookupVolumeServer == diag.AuthorityVolumeServer
 	iscsiMatch := diag.LookupIscsiAddr == diag.AuthorityIscsiAddr
 	diag.Coherent = vsMatch && iscsiMatch
 	if !diag.Coherent {
 		if !vsMatch {
 			diag.Reason = "volume_server_mismatch"
 		} else {
 			diag.Reason = "iscsi_addr_mismatch"
 		}
 	}
 	return diag, true
 }
 func newBlockFailoverState() *blockFailoverState {
 	return &blockFailoverState{
 		pendingRebuilds: make(map[string][]pendingRebuild),
-		deferredTimers:  make(map[string][]*time.Timer),
+		deferredTimers:  make(map[string][]deferredPromotion),
 	}
 }
@@ -60,7 +183,11 @@ func (ms *MasterServer) failoverBlockVolumes(deadServer string) {
 				glog.V(0).Infof("failover: %q lease expires in %v, deferring promotion", entry.Name, delay)
 				volumeName := entry.Name
 				capturedEpoch := entry.Epoch // T3: capture epoch for stale-timer validation
 				capturedDeadServer := deadServer // capture for closure
 				timer := time.AfterFunc(delay, func() {
 					// Clean up the deferred entry regardless of outcome.
 					ms.removeFiredDeferredPromotion(capturedDeadServer, volumeName)
 					// T3: Re-validate before acting — prevent stale timer on recreated/changed volume.
 					current, ok := ms.blockRegistry.Lookup(volumeName)
 					if !ok {
@@ -76,7 +203,12 @@ func (ms *MasterServer) failoverBlockVolumes(deadServer string) {
 				})
 				ms.blockFailover.mu.Lock()
 				ms.blockFailover.deferredTimers[deadServer] = append(
-					ms.blockFailover.deferredTimers[deadServer], timer)
+					ms.blockFailover.deferredTimers[deadServer], deferredPromotion{
 						Timer:          timer,
 						VolumeName:     volumeName,
 						CurrentPrimary: entry.VolumeServer,
 						AffectedServer: deadServer,
 					})
 				ms.blockFailover.mu.Unlock()
 				continue
 			}
@@ -159,12 +291,14 @@ func (ms *MasterServer) finalizePromotion(volumeName, oldPrimary, oldPath string
 		assignment.ReplicaAddrs = append(assignment.ReplicaAddrs, blockvol.ReplicaAddr{
 			DataAddr: ri.DataAddr,
 			CtrlAddr: ri.CtrlAddr,
 			ServerID: ri.Server, // V2: stable identity
 		})
 	}
 	// Backward compat: also set scalar fields if exactly 1 replica.
 	if len(entry.Replicas) == 1 {
 		assignment.ReplicaDataAddr = entry.Replicas[0].DataAddr
 		assignment.ReplicaCtrlAddr = entry.Replicas[0].CtrlAddr
 		assignment.ReplicaServerID = entry.Replicas[0].Server // V2: stable identity
 	}
 	ms.blockAssignmentQueue.Enqueue(entry.VolumeServer, assignment)
@@ -210,14 +344,36 @@ func (ms *MasterServer) cancelDeferredTimers(server string) {
 		return
 	}
 	ms.blockFailover.mu.Lock()
-	timers := ms.blockFailover.deferredTimers[server]
+	promos := ms.blockFailover.deferredTimers[server]
 	delete(ms.blockFailover.deferredTimers, server)
 	ms.blockFailover.mu.Unlock()
-	for _, t := range timers {
+	for _, dp := range promos {
-		t.Stop()
+		dp.Timer.Stop()
 	}
-	if len(timers) > 0 {
+	if len(promos) > 0 {
-		glog.V(0).Infof("failover: cancelled %d deferred promotion timers for reconnected %s", len(timers), server)
+		glog.V(0).Infof("failover: cancelled %d deferred promotion timers for reconnected %s", len(promos), server)
 	}
 }
 // removeFiredDeferredPromotion removes a single deferred promotion entry after
 // its timer has fired (whether it promoted or was skipped). This keeps
 // FailoverDiagnostic accurate: once the timer fires, the volume is no longer
 // in lease-wait state.
 func (ms *MasterServer) removeFiredDeferredPromotion(server, volumeName string) {
 	if ms.blockFailover == nil {
 		return
 	}
 	ms.blockFailover.mu.Lock()
 	defer ms.blockFailover.mu.Unlock()
 	promos := ms.blockFailover.deferredTimers[server]
 	for i, dp := range promos {
 		if dp.VolumeName == volumeName {
 			ms.blockFailover.deferredTimers[server] = append(promos[:i], promos[i+1:]...)
 			if len(ms.blockFailover.deferredTimers[server]) == 0 {
 				delete(ms.blockFailover.deferredTimers, server)
 			}
 			return
 		}
 	}
 }
@@ -286,16 +442,18 @@ func (ms *MasterServer) recoverBlockVolumes(reconnectedServer string) {
 				Role:       blockvol.RoleToWire(blockvol.RolePrimary),
 				LeaseTtlMs: leaseTTLMs,
 			}
-			// Include all replica addresses.
+			// Include all replica addresses with stable identity.
 			for _, ri := range entry.Replicas {
 				primaryAssignment.ReplicaAddrs = append(primaryAssignment.ReplicaAddrs, blockvol.ReplicaAddr{
 					DataAddr: ri.DataAddr,
 					CtrlAddr: ri.CtrlAddr,
 					ServerID: ri.Server, // V2: stable identity
 				})
 			}
 			if len(entry.Replicas) == 1 {
 				primaryAssignment.ReplicaDataAddr = entry.Replicas[0].DataAddr
 				primaryAssignment.ReplicaCtrlAddr = entry.Replicas[0].CtrlAddr
 				primaryAssignment.ReplicaServerID = entry.Replicas[0].Server // V2
 			}
 			ms.blockAssignmentQueue.Enqueue(entry.VolumeServer, primaryAssignment)
@@ -346,11 +504,13 @@ func (ms *MasterServer) refreshPrimaryForAddrChange(ac ReplicaAddrChange) {
 		assignment.ReplicaAddrs = append(assignment.ReplicaAddrs, blockvol.ReplicaAddr{
 			DataAddr: ri.DataAddr,
 			CtrlAddr: ri.CtrlAddr,
 			ServerID: ri.Server, // V2: stable identity
 		})
 	}
 	if len(entry.Replicas) == 1 {
 		assignment.ReplicaDataAddr = entry.Replicas[0].DataAddr
 		assignment.ReplicaCtrlAddr = entry.Replicas[0].CtrlAddr
 		assignment.ReplicaServerID = entry.Replicas[0].Server // V2
 	}
 	// Use current registry primary (not stale ac.PrimaryServer) in case
 	// failover happened between address-change detection and this refresh.
@@ -384,6 +544,9 @@ func (ms *MasterServer) reevaluateOrphanedPrimaries(server string) {
 			capturedEpoch := entry.Epoch
 			deadPrimary := entry.VolumeServer
 			timer := time.AfterFunc(delay, func() {
 				// Clean up the deferred entry regardless of outcome.
 				ms.removeFiredDeferredPromotion(deadPrimary, volumeName)
 				current, ok := ms.blockRegistry.Lookup(volumeName)
 				if !ok {
 					return
@@ -397,7 +560,12 @@ func (ms *MasterServer) reevaluateOrphanedPrimaries(server string) {
 			})
 			ms.blockFailover.mu.Lock()
 			ms.blockFailover.deferredTimers[deadPrimary] = append(
-				ms.blockFailover.deferredTimers[deadPrimary], timer)
+				ms.blockFailover.deferredTimers[deadPrimary], deferredPromotion{
 					Timer:          timer,
 					VolumeName:     volumeName,
 					CurrentPrimary: deadPrimary,
 					AffectedServer: deadPrimary,
 				})
 			ms.blockFailover.mu.Unlock()
 			continue
 		}
@@ -0,0 +1,270 @@
 package weed_server
 import (
 	"context"
 	"os"
 	"strings"
 	"testing"
 	"time"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 )
 // ============================================================
 // Phase 12 P3: Diagnosability / Blocker Accounting
 //
 // All diagnosis conclusions use ONLY explicit bounded read-only
 // diagnosis surfaces:
 //   - LookupBlockVolume       (product-visible publication)
 //   - FailoverDiagnostic      (volume-oriented failover state)
 //   - PublicationDiagnostic   (lookup vs authority coherence)
 //   - RecoveryDiagnostic      (active recovery task set)
 //   - phase-12-p3-blockers.md (finite blocker ledger)
 //
 // NOT performance, NOT rollout readiness.
 // ============================================================
 // --- S1: Failover convergence diagnosable via FailoverDiagnostic ---
 func TestP12P3_FailoverConvergence_Diagnosable(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 		Name: "diag-vol-1", SizeBytes: 1 << 20,
 	})
 	entry, _ := s.ms.blockRegistry.Lookup("diag-vol-1")
 	s.bs.localServerID = entry.VolumeServer
 	s.deliver(entry.VolumeServer)
 	time.Sleep(100 * time.Millisecond)
 	// Surface 1: LookupBlockVolume shows current primary before failover.
 	lookupBefore, _ := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: "diag-vol-1"})
 	if lookupBefore.VolumeServer != entry.VolumeServer {
 		t.Fatal("lookup should show original primary before failover")
 	}
 	// Surface 2: FailoverDiagnostic — no volumes in failover state yet.
 	failoverBefore := s.ms.blockFailover.DiagnosticSnapshot()
 	for _, v := range failoverBefore.Volumes {
 		if v.VolumeName == "diag-vol-1" {
 			t.Fatalf("S1: diag-vol-1 should not appear in failover diagnostic before failover, got %+v", v)
 		}
 	}
 	// Trigger failover: expire lease, then failover.
 	s.ms.blockRegistry.UpdateEntry("diag-vol-1", func(e *BlockVolumeEntry) {
 		e.LastLeaseGrant = time.Now().Add(-1 * time.Minute)
 	})
 	s.ms.failoverBlockVolumes(entry.VolumeServer)
 	// Surface 1 after: LookupBlockVolume shows NEW primary.
 	lookupAfter, _ := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: "diag-vol-1"})
 	if lookupAfter.VolumeServer == entry.VolumeServer {
 		t.Fatal("S1 stall: lookup still shows old primary after failover")
 	}
 	// Surface 2 after: FailoverDiagnostic shows volume-level failover state.
 	failoverAfter := s.ms.blockFailover.DiagnosticSnapshot()
 	// Classify via explicit diagnosis surface: find diag-vol-1 in failover volumes.
 	var found *FailoverVolumeState
 	for i := range failoverAfter.Volumes {
 		if failoverAfter.Volumes[i].VolumeName == "diag-vol-1" {
 			found = &failoverAfter.Volumes[i]
 			break
 		}
 	}
 	if found == nil {
 		t.Fatal("S1: diag-vol-1 not found in FailoverDiagnostic after failover")
 	}
 	// Diagnosis conclusion from explicit surfaces only:
 	// - Lookup changed (old → new primary)
 	// - FailoverDiagnostic classifies state as rebuild_pending
 	// - AffectedServer identifies the dead server
 	if !found.PendingRebuild {
 		t.Fatal("S1: FailoverDiagnostic should show PendingRebuild=true")
 	}
 	if found.Reason != "rebuild_pending" {
 		t.Fatalf("S1: expected reason=rebuild_pending, got %q", found.Reason)
 	}
 	if found.AffectedServer != entry.VolumeServer {
 		t.Fatalf("S1: AffectedServer should be dead server %s, got %s", entry.VolumeServer, found.AffectedServer)
 	}
 	if found.CurrentPrimary != lookupAfter.VolumeServer {
 		t.Fatalf("S1: CurrentPrimary should match lookup %s, got %s", lookupAfter.VolumeServer, found.CurrentPrimary)
 	}
 	t.Logf("P12P3 S1: diagnosed via LookupBlockVolume(%s→%s) + FailoverDiagnostic(vol=%s, reason=%s, affected=%s)",
 		lookupBefore.VolumeServer, lookupAfter.VolumeServer, found.VolumeName, found.Reason, found.AffectedServer)
 }
 // --- S2: Publication mismatch diagnosable via PublicationDiagnostic ---
 func TestP12P3_PublicationMismatch_Diagnosable(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 		Name: "diag-vol-2", SizeBytes: 1 << 20,
 	})
 	entry, _ := s.ms.blockRegistry.Lookup("diag-vol-2")
 	s.bs.localServerID = entry.VolumeServer
 	s.deliver(entry.VolumeServer)
 	time.Sleep(100 * time.Millisecond)
 	// Surface 1: PublicationDiagnostic before failover — should be coherent.
 	pubBefore, ok := s.ms.PublicationDiagnosticFor("diag-vol-2")
 	if !ok {
 		t.Fatal("S2: PublicationDiagnosticFor should find diag-vol-2")
 	}
 	if !pubBefore.Coherent {
 		t.Fatalf("S2: publication should be coherent before failover, got reason=%q", pubBefore.Reason)
 	}
 	// Surface 2: LookupBlockVolume — repeated lookups self-consistent.
 	lookup1, _ := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: "diag-vol-2"})
 	lookup2, _ := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: "diag-vol-2"})
 	if lookup1.IscsiAddr != lookup2.IscsiAddr || lookup1.VolumeServer != lookup2.VolumeServer {
 		t.Fatalf("S2: repeated lookup mismatch: %s/%s vs %s/%s",
 			lookup1.VolumeServer, lookup1.IscsiAddr, lookup2.VolumeServer, lookup2.IscsiAddr)
 	}
 	// Trigger failover.
 	s.ms.blockRegistry.UpdateEntry("diag-vol-2", func(e *BlockVolumeEntry) {
 		e.LastLeaseGrant = time.Now().Add(-1 * time.Minute)
 	})
 	s.ms.failoverBlockVolumes(entry.VolumeServer)
 	// Surface 1 after: PublicationDiagnostic after failover — still coherent.
 	pubAfter, ok := s.ms.PublicationDiagnosticFor("diag-vol-2")
 	if !ok {
 		t.Fatal("S2: PublicationDiagnosticFor should find diag-vol-2 after failover")
 	}
 	if !pubAfter.Coherent {
 		t.Fatalf("S2: publication should be coherent after failover, got reason=%q", pubAfter.Reason)
 	}
 	// Diagnosis conclusion from explicit surfaces only:
 	// - Pre-failover: coherent, lookup matches authority
 	// - Post-failover: coherent, lookup updated to new primary
 	// - Publication switched: post != pre
 	if pubAfter.LookupVolumeServer == pubBefore.LookupVolumeServer {
 		t.Fatal("S2: LookupVolumeServer unchanged after failover — publication did not switch")
 	}
 	if pubAfter.LookupIscsiAddr == pubBefore.LookupIscsiAddr {
 		t.Fatal("S2: LookupIscsiAddr unchanged after failover")
 	}
 	// Post-failover repeated lookup still self-consistent (via diagnostic).
 	pubAfter2, _ := s.ms.PublicationDiagnosticFor("diag-vol-2")
 	if pubAfter2.LookupVolumeServer != pubAfter.LookupVolumeServer ||
 		pubAfter2.LookupIscsiAddr != pubAfter.LookupIscsiAddr {
 		t.Fatal("S2: post-failover publication diagnostics inconsistent")
 	}
 	t.Logf("P12P3 S2: diagnosed via PublicationDiagnostic — pre(vs=%s, iscsi=%s, coherent=%v) → post(vs=%s, iscsi=%s, coherent=%v)",
 		pubBefore.LookupVolumeServer, pubBefore.LookupIscsiAddr, pubBefore.Coherent,
 		pubAfter.LookupVolumeServer, pubAfter.LookupIscsiAddr, pubAfter.Coherent)
 }
 // --- S3: Runtime residue diagnosable via RecoveryDiagnostic ---
 func TestP12P3_RuntimeResidue_Diagnosable(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 		Name: "diag-vol-3", SizeBytes: 1 << 20,
 	})
 	entry, _ := s.ms.blockRegistry.Lookup("diag-vol-3")
 	s.bs.localServerID = entry.VolumeServer
 	s.deliver(entry.VolumeServer)
 	time.Sleep(200 * time.Millisecond)
 	// Surface: RecoveryDiagnostic shows active task set.
 	diagBefore := s.bs.v2Recovery.DiagnosticSnapshot()
 	t.Logf("S3 before delete: %d active tasks: %v", len(diagBefore.ActiveTasks), diagBefore.ActiveTasks)
 	// Delete the volume.
 	s.ms.DeleteBlockVolume(ctx, &master_pb.DeleteBlockVolumeRequest{Name: "diag-vol-3"})
 	time.Sleep(200 * time.Millisecond)
 	// Surface after: RecoveryDiagnostic — no tasks for deleted volume.
 	diagAfter := s.bs.v2Recovery.DiagnosticSnapshot()
 	for _, task := range diagAfter.ActiveTasks {
 		if strings.Contains(task, "diag-vol-3") {
 			t.Fatalf("S3 residue: task %s active after delete", task)
 		}
 	}
 	// Diagnosis conclusion from explicit surface only:
 	// - RecoveryDiagnostic.ActiveTasks does not contain deleted volume's tasks
 	// - Conclusion: clean (no residue) or non-empty but unrelated to deleted volume
 	if len(diagAfter.ActiveTasks) == 0 {
 		t.Log("P12P3 S3: diagnosed via RecoveryDiagnostic — clean (0 active tasks after delete)")
 	} else {
 		t.Logf("P12P3 S3: diagnosed via RecoveryDiagnostic — %d active tasks (none for deleted vol)",
 			len(diagAfter.ActiveTasks))
 	}
 }
 // --- Blocker ledger: reads and validates the actual file ---
 func TestP12P3_BlockerLedger_Bounded(t *testing.T) {
 	ledgerPath := "../../sw-block/.private/phase/phase-12-p3-blockers.md"
 	data, err := os.ReadFile(ledgerPath)
 	if err != nil {
 		t.Fatalf("blocker ledger must exist at %s: %v", ledgerPath, err)
 	}
 	content := string(data)
 	// Must contain diagnosed items.
 	for _, id := range []string{"B1", "B2", "B3"} {
 		if !strings.Contains(content, id) {
 			t.Fatalf("ledger missing diagnosed item %s", id)
 		}
 	}
 	// Must contain unresolved items.
 	for _, id := range []string{"U1", "U2", "U3"} {
 		if !strings.Contains(content, id) {
 			t.Fatalf("ledger missing unresolved item %s", id)
 		}
 	}
 	// Must contain out-of-scope section.
 	if !strings.Contains(content, "Out of Scope") {
 		t.Fatal("ledger must have 'Out of Scope' section")
 	}
 	// Must NOT overclaim perf or rollout.
 	lines := strings.Split(content, "\n")
 	diagnosedCount := 0
 	unresolvedCount := 0
 	for _, line := range lines {
 		if strings.HasPrefix(strings.TrimSpace(line), "| B") {
 			diagnosedCount++
 		}
 		if strings.HasPrefix(strings.TrimSpace(line), "| U") {
 			unresolvedCount++
 		}
 	}
 	total := diagnosedCount + unresolvedCount
 	if total == 0 {
 		t.Fatal("ledger has no blocker items")
 	}
 	if total > 20 {
 		t.Fatalf("ledger should be finite, got %d items", total)
 	}
 	t.Logf("P12P3 blockers: %d diagnosed + %d unresolved = %d total (from actual file, finite)",
 		diagnosedCount, unresolvedCount, total)
 }
@@ -0,0 +1,582 @@
 package weed_server
 import (
 	"crypto/rand"
 	"fmt"
 	"math"
 	mrand "math/rand"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
 	"testing"
 	"time"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
 )
 // ============================================================
 // Phase 12 P4: Performance Floor — Bounded Measurement Package
 //
 // Workload envelope:
 //   Topology: RF=2 sync_all accepted chosen path
 //   Operations: 4K random write, 4K random read, 4K sequential write, 4K sequential read
 //   Runtime: no failover, no disturbance, steady-state
 //   Environment: unit test harness (single-process, local disk, engine-local I/O)
 //
 // What this measures:
 //   Engine I/O floor for the accepted chosen path. WriteLBA/ReadLBA through
 //   the full fencing path (epoch, role, lease, writeGate, WAL, dirtyMap).
 //   No transport layer (iSCSI/NVMe). No cross-machine replication.
 //
 // What this does NOT measure:
 //   Transport throughput, cross-machine replication tax, multi-client concurrency,
 //   failover-under-load, degraded mode. Production floor with replication is
 //   documented in baseline-roce-20260401.md.
 //
 // NOT performance tuning. NOT broad benchmark.
 // ============================================================
 const (
 	perfBlockSize  = 4096
 	perfVolumeSize = 64 * 1024 * 1024 // 64MB
 	perfWALSize    = 16 * 1024 * 1024  // 16MB
 	perfOps        = 1000              // ops per measurement run
 	perfWarmupOps  = 200               // warmup ops (discarded from measurement)
 	perfIterations = 3                 // run N times, report worst as floor
 )
 // Minimum acceptable floor thresholds (engine-local, single-writer).
 //
 // These are regression gates, not performance targets. Set conservatively
 // so any reasonable hardware passes, but catastrophic regressions
 // (accidental serialization, O(n^2) scan, broken WAL path) are caught.
 //
 // Rationale for values:
 //   Measured on dev SSD: rand-write ~10K, rand-read ~80K, seq-write ~30K, seq-read ~180K.
 //   Thresholds set at ~10% of measured to tolerate slow CI machines and VMs.
 //   Write P99 ceiling at 100ms catches deadlocks/stalls without false-positiving
 //   on slow storage.
 var perfFloorGates = map[string]struct {
 	MinIOPS     float64
 	MaxWriteP99 time.Duration // 0 = no ceiling (reads)
 }{
 	"rand-write": {MinIOPS: 1000, MaxWriteP99: 100 * time.Millisecond},
 	"rand-read":  {MinIOPS: 5000},
 	"seq-write":  {MinIOPS: 2000, MaxWriteP99: 100 * time.Millisecond},
 	"seq-read":   {MinIOPS: 10000},
 }
 // perfResult holds measurements for one workload run.
 type perfResult struct {
 	Workload   string
 	Ops        int
 	Elapsed    time.Duration
 	IOPS       float64
 	MBps       float64
 	LatSamples []int64 // per-op latency in nanoseconds
 }
 func (r *perfResult) latPct(pct float64) time.Duration {
 	if len(r.LatSamples) == 0 {
 		return 0
 	}
 	sorted := make([]int64, len(r.LatSamples))
 	copy(sorted, r.LatSamples)
 	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
 	idx := int(math.Ceil(pct/100.0*float64(len(sorted)))) - 1
 	if idx < 0 {
 		idx = 0
 	}
 	if idx >= len(sorted) {
 		idx = len(sorted) - 1
 	}
 	return time.Duration(sorted[idx])
 }
 func (r *perfResult) latAvg() time.Duration {
 	if len(r.LatSamples) == 0 {
 		return 0
 	}
 	var sum int64
 	for _, s := range r.LatSamples {
 		sum += s
 	}
 	return time.Duration(sum / int64(len(r.LatSamples)))
 }
 // setupPerfVolume creates a BlockVol configured as Primary for perf measurement.
 func setupPerfVolume(t *testing.T) *blockvol.BlockVol {
 	t.Helper()
 	dir := t.TempDir()
 	volPath := filepath.Join(dir, "perf.blk")
 	vol, err := blockvol.CreateBlockVol(volPath, blockvol.CreateOptions{
 		VolumeSize: perfVolumeSize,
 		BlockSize:  perfBlockSize,
 		WALSize:    perfWALSize,
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Set up as Primary with long lease so writes are allowed.
 	if err := vol.HandleAssignment(1, blockvol.RolePrimary, 10*time.Minute); err != nil {
 		vol.Close()
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { vol.Close() })
 	return vol
 }
 // maxLBAs returns the number of addressable 4K blocks in the extent area.
 func maxLBAs() uint64 {
 	// Volume size minus WAL, divided by block size, with safety margin.
 	return (perfVolumeSize - perfWALSize) / perfBlockSize / 2
 }
 // runPerfWorkload executes one workload measurement and returns the result.
 func runPerfWorkload(t *testing.T, vol *blockvol.BlockVol, workload string, ops int) perfResult {
 	t.Helper()
 	data := make([]byte, perfBlockSize)
 	rand.Read(data)
 	max := maxLBAs()
 	samples := make([]int64, 0, ops)
 	start := time.Now()
 	for i := 0; i < ops; i++ {
 		var lba uint64
 		switch {
 		case strings.HasPrefix(workload, "rand"):
 			lba = uint64(mrand.Int63n(int64(max)))
 		default: // sequential
 			lba = uint64(i) % max
 		}
 		opStart := time.Now()
 		switch {
 		case strings.HasSuffix(workload, "write"):
 			if err := vol.WriteLBA(lba, data); err != nil {
 				t.Fatalf("%s op %d: WriteLBA(%d): %v", workload, i, lba, err)
 			}
 		case strings.HasSuffix(workload, "read"):
 			if _, err := vol.ReadLBA(lba, perfBlockSize); err != nil {
 				t.Fatalf("%s op %d: ReadLBA(%d): %v", workload, i, lba, err)
 			}
 		}
 		samples = append(samples, time.Since(opStart).Nanoseconds())
 	}
 	elapsed := time.Since(start)
 	iops := float64(ops) / elapsed.Seconds()
 	mbps := iops * float64(perfBlockSize) / (1024 * 1024)
 	return perfResult{
 		Workload:   workload,
 		Ops:        ops,
 		Elapsed:    elapsed,
 		IOPS:       iops,
 		MBps:       mbps,
 		LatSamples: samples,
 	}
 }
 // floorOf returns the worst (lowest) IOPS and worst (highest) P99 across iterations.
 type perfFloor struct {
 	Workload string
 	FloorIOPS float64
 	FloorMBps float64
 	WorstAvg  time.Duration
 	WorstP50  time.Duration
 	WorstP99  time.Duration
 	WorstMax  time.Duration
 }
 func computeFloor(results []perfResult) perfFloor {
 	f := perfFloor{
 		Workload:  results[0].Workload,
 		FloorIOPS: math.MaxFloat64,
 		FloorMBps: math.MaxFloat64,
 	}
 	for _, r := range results {
 		if r.IOPS < f.FloorIOPS {
 			f.FloorIOPS = r.IOPS
 		}
 		if r.MBps < f.FloorMBps {
 			f.FloorMBps = r.MBps
 		}
 		avg := r.latAvg()
 		if avg > f.WorstAvg {
 			f.WorstAvg = avg
 		}
 		p50 := r.latPct(50)
 		if p50 > f.WorstP50 {
 			f.WorstP50 = p50
 		}
 		p99 := r.latPct(99)
 		if p99 > f.WorstP99 {
 			f.WorstP99 = p99
 		}
 		pmax := r.latPct(100)
 		if pmax > f.WorstMax {
 			f.WorstMax = pmax
 		}
 	}
 	return f
 }
 // --- Test 1: PerformanceFloor_Bounded ---
 func TestP12P4_PerformanceFloor_Bounded(t *testing.T) {
 	vol := setupPerfVolume(t)
 	workloads := []string{"rand-write", "rand-read", "seq-write", "seq-read"}
 	floors := make([]perfFloor, 0, len(workloads))
 	for _, wl := range workloads {
 		// Warmup: populate volume with data (needed for reads).
 		if strings.HasSuffix(wl, "read") {
 			warmupData := make([]byte, perfBlockSize)
 			rand.Read(warmupData)
 			for i := 0; i < int(maxLBAs()); i++ {
 				if err := vol.WriteLBA(uint64(i), warmupData); err != nil {
 					break // WAL full is acceptable during warmup
 				}
 			}
 			time.Sleep(200 * time.Millisecond) // let flusher drain
 		}
 		// Warmup ops (discarded).
 		runPerfWorkload(t, vol, wl, perfWarmupOps)
 		// Measurement: N iterations, take floor.
 		var results []perfResult
 		for iter := 0; iter < perfIterations; iter++ {
 			r := runPerfWorkload(t, vol, wl, perfOps)
 			results = append(results, r)
 		}
 		floor := computeFloor(results)
 		floors = append(floors, floor)
 	}
 	// Report structured floor table.
 	t.Log("")
 	t.Log("=== P12P4 Performance Floor (engine-local, single-writer) ===")
 	t.Log("")
 	t.Logf("%-12s  %10s  %8s  %10s  %10s  %10s  %10s",
 		"Workload", "Floor IOPS", "MB/s", "Avg Lat", "P50 Lat", "P99 Lat", "Max Lat")
 	t.Logf("%-12s  %10s  %8s  %10s  %10s  %10s  %10s",
 		"--------", "----------", "------", "-------", "-------", "-------", "-------")
 	for _, f := range floors {
 		t.Logf("%-12s  %10.0f  %8.2f  %10s  %10s  %10s  %10s",
 			f.Workload, f.FloorIOPS, f.FloorMBps, f.WorstAvg, f.WorstP50, f.WorstP99, f.WorstMax)
 	}
 	t.Log("")
 	t.Logf("Config: volume=%dMB WAL=%dMB block=%dB ops=%d warmup=%d iterations=%d",
 		perfVolumeSize/(1024*1024), perfWALSize/(1024*1024), perfBlockSize, perfOps, perfWarmupOps, perfIterations)
 	t.Log("Method: worst of N iterations (floor, not peak)")
 	t.Log("Scope: engine-local only; production RF=2 floor in baseline-roce-20260401.md")
 	// Gate: floor values must meet minimum acceptable thresholds.
 	// These are regression gates — if any floor drops below the gate,
 	// the test fails, blocking rollout.
 	t.Log("")
 	t.Log("=== Floor Gate Validation ===")
 	allGatesPassed := true
 	for _, f := range floors {
 		gate, ok := perfFloorGates[f.Workload]
 		if !ok {
 			t.Fatalf("no floor gate defined for workload %s", f.Workload)
 		}
 		passed := true
 		if f.FloorIOPS < gate.MinIOPS {
 			t.Errorf("GATE FAIL: %s floor IOPS %.0f < minimum %.0f", f.Workload, f.FloorIOPS, gate.MinIOPS)
 			passed = false
 		}
 		if gate.MaxWriteP99 > 0 && f.WorstP99 > gate.MaxWriteP99 {
 			t.Errorf("GATE FAIL: %s worst P99 %s > ceiling %s", f.Workload, f.WorstP99, gate.MaxWriteP99)
 			passed = false
 		}
 		status := "PASS"
 		if !passed {
 			status = "FAIL"
 			allGatesPassed = false
 		}
 		t.Logf("  %-12s  min=%6.0f IOPS  →  floor=%6.0f  [%s]", f.Workload, gate.MinIOPS, f.FloorIOPS, status)
 	}
 	if !allGatesPassed {
 		t.Fatal("P12P4 PerformanceFloor: FAIL — one or more floor gates not met")
 	}
 	t.Log("P12P4 PerformanceFloor: PASS — all floor gates met")
 }
 // --- Test 2: CostCharacterization_Bounded ---
 func TestP12P4_CostCharacterization_Bounded(t *testing.T) {
 	vol := setupPerfVolume(t)
 	// Measure write latency breakdown: WriteLBA includes WAL append + group commit.
 	data := make([]byte, perfBlockSize)
 	rand.Read(data)
 	max := maxLBAs()
 	const costOps = 500
 	var writeLatSum int64
 	for i := 0; i < costOps; i++ {
 		lba := uint64(mrand.Int63n(int64(max)))
 		start := time.Now()
 		if err := vol.WriteLBA(lba, data); err != nil {
 			t.Fatalf("write op %d: %v", i, err)
 		}
 		writeLatSum += time.Since(start).Nanoseconds()
 	}
 	avgWriteLat := time.Duration(writeLatSum / costOps)
 	// Measure read latency for comparison.
 	// Populate first.
 	for i := 0; i < int(max/2); i++ {
 		vol.WriteLBA(uint64(i), data)
 	}
 	time.Sleep(200 * time.Millisecond) // let flusher drain
 	var readLatSum int64
 	for i := 0; i < costOps; i++ {
 		lba := uint64(mrand.Int63n(int64(max / 2)))
 		start := time.Now()
 		if _, err := vol.ReadLBA(lba, perfBlockSize); err != nil {
 			t.Fatalf("read op %d: %v", i, err)
 		}
 		readLatSum += time.Since(start).Nanoseconds()
 	}
 	avgReadLat := time.Duration(readLatSum / costOps)
 	// Cost statement.
 	t.Log("")
 	t.Log("=== P12P4 Cost Characterization (engine-local) ===")
 	t.Log("")
 	t.Logf("Average write latency: %s (includes WAL append + group commit sync)", avgWriteLat)
 	t.Logf("Average read latency:  %s (dirtyMap lookup + WAL/extent read)", avgReadLat)
 	t.Log("")
 	t.Log("Bounded cost statement:")
 	t.Log("  WAL write amplification: 2x minimum (WAL write + eventual extent flush)")
 	t.Log("  Group commit: amortizes fdatasync across batched writers (1 sync per batch)")
 	t.Log("  Replication tax (production RF=2 sync_all): -56% vs RF=1 (barrier round-trip)")
 	t.Log("  Replication tax source: baseline-roce-20260401.md, measured on 25Gbps RoCE")
 	t.Log("")
 	t.Logf("Write/read ratio: %.1fx (write is %.1fx slower than read)",
 		float64(avgWriteLat)/float64(avgReadLat),
 		float64(avgWriteLat)/float64(avgReadLat))
 	t.Log("")
 	t.Logf("Config: volume=%dMB WAL=%dMB block=%dB ops=%d",
 		perfVolumeSize/(1024*1024), perfWALSize/(1024*1024), perfBlockSize, costOps)
 	// Proof: cost values are finite and positive.
 	if avgWriteLat <= 0 || avgReadLat <= 0 {
 		t.Fatal("latency values must be positive")
 	}
 	if avgWriteLat < avgReadLat {
 		t.Log("Note: write faster than read in this run (possible due to WAL cache hits)")
 	}
 	t.Log("P12P4 CostCharacterization: PASS — bounded cost statement produced")
 }
 // --- Test 3: RolloutGate_Bounded ---
 func TestP12P4_RolloutGate_Bounded(t *testing.T) {
 	floorPath := "../../sw-block/.private/phase/phase-12-p4-floor.md"
 	gatesPath := "../../sw-block/.private/phase/phase-12-p4-rollout-gates.md"
 	baselinePath := "../../learn/projects/sw-block/test/results/baseline-roce-20260401.md"
 	blockerPath := "../../sw-block/.private/phase/phase-12-p3-blockers.md"
 	// --- Read all cited evidence sources ---
 	floorData, err := os.ReadFile(floorPath)
 	if err != nil {
 		t.Fatalf("floor doc must exist at %s: %v", floorPath, err)
 	}
 	floorContent := string(floorData)
 	gatesData, err := os.ReadFile(gatesPath)
 	if err != nil {
 		t.Fatalf("rollout-gates doc must exist at %s: %v", gatesPath, err)
 	}
 	gatesContent := string(gatesData)
 	baselineData, err := os.ReadFile(baselinePath)
 	if err != nil {
 		t.Fatalf("cited baseline must exist at %s: %v", baselinePath, err)
 	}
 	baselineContent := string(baselineData)
 	blockerData, err := os.ReadFile(blockerPath)
 	if err != nil {
 		t.Fatalf("cited blocker ledger must exist at %s: %v", blockerPath, err)
 	}
 	blockerContent := string(blockerData)
 	// --- Structural validation (shape) ---
 	// Floor doc: workload envelope, floor table, non-claims.
 	for _, required := range []string{"RF=2", "sync_all", "4K random write", "4K random read", "sequential write", "sequential read"} {
 		if !strings.Contains(floorContent, required) {
 			t.Fatalf("floor doc missing required content: %q", required)
 		}
 	}
 	if !strings.Contains(floorContent, "Floor") || !strings.Contains(floorContent, "IOPS") {
 		t.Fatal("floor doc must contain floor table with IOPS")
 	}
 	if !strings.Contains(floorContent, "does NOT") {
 		t.Fatal("floor doc must contain explicit non-claims")
 	}
 	// Gates doc: gates table, launch envelope, exclusions, non-claims.
 	if !strings.Contains(gatesContent, "Gate") || !strings.Contains(gatesContent, "Status") {
 		t.Fatal("rollout-gates doc must contain gates table")
 	}
 	if !strings.Contains(gatesContent, "Launch Envelope") {
 		t.Fatal("rollout-gates doc must contain launch envelope")
 	}
 	if !strings.Contains(gatesContent, "Exclusion") {
 		t.Fatal("rollout-gates doc must contain exclusions")
 	}
 	if !strings.Contains(gatesContent, "does NOT") {
 		t.Fatal("rollout-gates doc must contain explicit non-claims")
 	}
 	// Count gates — must be finite.
 	gateLines := 0
 	for _, line := range strings.Split(gatesContent, "\n") {
 		trimmed := strings.TrimSpace(line)
 		if strings.HasPrefix(trimmed, "| G") || strings.HasPrefix(trimmed, "| E") {
 			gateLines++
 		}
 	}
 	if gateLines == 0 {
 		t.Fatal("rollout-gates doc has no gate items")
 	}
 	if gateLines > 20 {
 		t.Fatalf("rollout-gates doc should be finite, got %d items", gateLines)
 	}
 	// --- Semantic cross-checks (evidence alignment) ---
 	// 1. G6 cites "28.4K write IOPS" — baseline must contain this number.
 	if strings.Contains(gatesContent, "28.4K write IOPS") || strings.Contains(gatesContent, "28,4") {
 		// The gates doc cites write IOPS from baseline. Verify the baseline has it.
 		if !strings.Contains(baselineContent, "28,") {
 			t.Fatal("G6 cites write IOPS but baseline does not contain matching value")
 		}
 	}
 	// More precise: baseline must contain the specific numbers cited in G6.
 	if !strings.Contains(baselineContent, "28,347") && !strings.Contains(baselineContent, "28,429") &&
 		!strings.Contains(baselineContent, "28,453") {
 		t.Fatal("baseline must contain RF=2 sync_all write IOPS data (28,3xx-28,4xx range)")
 	}
 	if !strings.Contains(baselineContent, "136,648") {
 		t.Fatal("baseline must contain RF=2 read IOPS data (136,648)")
 	}
 	// 2. G5 cites "-56% replication tax" — baseline must contain this.
 	if strings.Contains(gatesContent, "-56%") {
 		if !strings.Contains(baselineContent, "-56%") {
 			t.Fatal("G5 cites -56% replication tax but baseline does not contain -56%")
 		}
 	}
 	// 3. Launch envelope claims specific transport/network combos — verify against baseline.
 	// Claimed: NVMe-TCP @ 25Gbps RoCE
 	if strings.Contains(gatesContent, "NVMe-TCP @ 25Gbps RoCE") {
 		if !strings.Contains(baselineContent, "NVMe-TCP") || !strings.Contains(baselineContent, "RoCE") {
 			t.Fatal("launch envelope claims NVMe-TCP @ RoCE but baseline has no such data")
 		}
 	}
 	// Claimed: iSCSI @ 25Gbps RoCE
 	if strings.Contains(gatesContent, "iSCSI @ 25Gbps RoCE") {
 		if !strings.Contains(baselineContent, "iSCSI") || !strings.Contains(baselineContent, "RoCE") {
 			t.Fatal("launch envelope claims iSCSI @ RoCE but baseline has no such data")
 		}
 	}
 	// Claimed: iSCSI @ 1Gbps
 	if strings.Contains(gatesContent, "iSCSI @ 1Gbps") {
 		if !strings.Contains(baselineContent, "iSCSI") || !strings.Contains(baselineContent, "1Gbps") {
 			t.Fatal("launch envelope claims iSCSI @ 1Gbps but baseline has no such data")
 		}
 	}
 	// Exclusion: NVMe-TCP @ 1Gbps must NOT be claimed as supported.
 	if strings.Contains(gatesContent, "NOT included") {
 		// Verify baseline indeed lacks NVMe-TCP @ 1Gbps.
 		hasNvme1g := strings.Contains(baselineContent, "NVMe-TCP") && strings.Contains(baselineContent, "| NVMe-TCP | 1Gbps")
 		if hasNvme1g {
 			t.Fatal("baseline contains NVMe-TCP @ 1Gbps data but gates doc excludes it — resolve mismatch")
 		}
 	}
 	// 4. G7 cites blocker ledger counts — verify against actual ledger.
 	if strings.Contains(gatesContent, "3 diagnosed") {
 		diagCount := 0
 		for _, line := range strings.Split(blockerContent, "\n") {
 			if strings.HasPrefix(strings.TrimSpace(line), "| B") {
 				diagCount++
 			}
 		}
 		if diagCount != 3 {
 			t.Fatalf("G7 claims 3 diagnosed blockers but ledger has %d", diagCount)
 		}
 	}
 	if strings.Contains(gatesContent, "3 unresolved") {
 		unresCount := 0
 		for _, line := range strings.Split(blockerContent, "\n") {
 			if strings.HasPrefix(strings.TrimSpace(line), "| U") {
 				unresCount++
 			}
 		}
 		if unresCount != 3 {
 			t.Fatalf("G7 claims 3 unresolved blockers but ledger has %d", unresCount)
 		}
 	}
 	// 5. Floor doc gate thresholds must match code-defined gates.
 	for workload, gate := range perfFloorGates {
 		// The doc uses comma-formatted numbers (e.g., "1,000" or "5,000").
 		minInt := int(gate.MinIOPS)
 		// Check for both comma-formatted and plain forms.
 		found := false
 		for _, form := range []string{
 			fmt.Sprintf("%d", minInt),                                      // "1000"
 			fmt.Sprintf("%d,%03d", minInt/1000, minInt%1000),               // "1,000"
 		} {
 			if strings.Contains(floorContent, form) {
 				found = true
 				break
 			}
 		}
 		if !found {
 			t.Errorf("floor doc gate for %s should cite minimum %d IOPS but doesn't", workload, minInt)
 		}
 	}
 	t.Logf("P12P4 RolloutGate: floor doc %d bytes, gates doc %d bytes, %d gate items",
 		len(floorData), len(gatesData), gateLines)
 	t.Log("P12P4 RolloutGate: semantic cross-checks passed (baseline, blocker ledger, gate thresholds)")
 	t.Log("P12P4 RolloutGate: PASS — bounded launch envelope with verified evidence alignment")
 }
 // --- Helpers ---
 func init() {
 	// Seed random for reproducible LBA patterns within a test run.
 	mrand.Seed(time.Now().UnixNano())
 }
 // formatDuration formats a duration for table display.
 func formatDuration(d time.Duration) string {
 	if d < time.Microsecond {
 		return fmt.Sprintf("%dns", d.Nanoseconds())
 	}
 	if d < time.Millisecond {
 		return fmt.Sprintf("%.1fus", float64(d.Nanoseconds())/1000.0)
 	}
 	return fmt.Sprintf("%.2fms", float64(d.Nanoseconds())/1e6)
 }
@@ -0,0 +1,308 @@
 package weed_server
 import (
 	"context"
 	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
 	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/v2bridge"
 )
 // ============================================================
 // Phase 12 P2: Soak / Long-Run Stability Hardening
 //
 // Proves: repeated chosen-path cycles return to bounded truth
 // without hidden state drift or unbounded runtime artifacts.
 //
 // NOT diagnosability, NOT performance-floor, NOT rollout readiness.
 // ============================================================
 const soakCycles = 5
 type soakSetup struct {
 	ms    *MasterServer
 	bs    *BlockService
 	store *storage.BlockVolumeStore
 	dir   string
 }
 func newSoakSetup(t *testing.T) *soakSetup {
 	t.Helper()
 	dir := t.TempDir()
 	store := storage.NewBlockVolumeStore()
 	ms := &MasterServer{
 		blockRegistry:        NewBlockVolumeRegistry(),
 		blockAssignmentQueue: NewBlockAssignmentQueue(),
 		blockFailover:        newBlockFailoverState(),
 	}
 	ms.blockRegistry.MarkBlockCapable("vs1:9333")
 	ms.blockRegistry.MarkBlockCapable("vs2:9333")
 	ms.blockVSAllocate = func(ctx context.Context, server string, name string, sizeBytes uint64, diskType string, durabilityMode string) (*blockAllocResult, error) {
 		sanitized := strings.ReplaceAll(server, ":", "_")
 		serverDir := filepath.Join(dir, sanitized)
 		os.MkdirAll(serverDir, 0755)
 		volPath := filepath.Join(serverDir, fmt.Sprintf("%s.blk", name))
 		vol, err := blockvol.CreateBlockVol(volPath, blockvol.CreateOptions{
 			VolumeSize: 1 * 1024 * 1024,
 			BlockSize:  4096,
 			WALSize:    256 * 1024,
 		})
 		if err != nil {
 			return nil, err
 		}
 		vol.Close()
 		if _, err := store.AddBlockVolume(volPath, ""); err != nil {
 			return nil, err
 		}
 		host := server
 		if idx := strings.LastIndex(server, ":"); idx >= 0 {
 			host = server[:idx]
 		}
 		return &blockAllocResult{
 			Path:              volPath,
 			IQN:               fmt.Sprintf("iqn.2024.test:%s", name),
 			ISCSIAddr:         host + ":3260",
 			ReplicaDataAddr:   server + ":14260",
 			ReplicaCtrlAddr:   server + ":14261",
 			RebuildListenAddr: server + ":15000",
 		}, nil
 	}
 	ms.blockVSDelete = func(ctx context.Context, server string, name string) error { return nil }
 	bs := &BlockService{
 		blockStore:     store,
 		blockDir:       filepath.Join(dir, "vs1_9333"),
 		listenAddr:     "127.0.0.1:3260",
 		localServerID:  "vs1:9333",
 		v2Bridge:       v2bridge.NewControlBridge(),
 		v2Orchestrator: engine.NewRecoveryOrchestrator(),
 		replStates:     make(map[string]*volReplState),
 	}
 	bs.v2Recovery = NewRecoveryManager(bs)
 	t.Cleanup(func() {
 		bs.v2Recovery.Shutdown()
 		store.Close()
 	})
 	return &soakSetup{ms: ms, bs: bs, store: store, dir: dir}
 }
 func (s *soakSetup) deliver(server string) int {
 	pending := s.ms.blockAssignmentQueue.Peek(server)
 	if len(pending) == 0 {
 		return 0
 	}
 	protoAssignments := blockvol.AssignmentsToProto(pending)
 	goAssignments := blockvol.AssignmentsFromProto(protoAssignments)
 	s.bs.ProcessAssignments(goAssignments)
 	return len(goAssignments)
 }
 // --- Repeated create/failover/recover cycles with end-of-cycle truth checks ---
 func TestP12P2_RepeatedCycles_NoDrift(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	for cycle := 1; cycle <= soakCycles; cycle++ {
 		volName := fmt.Sprintf("soak-vol-%d", cycle)
 		// Step 1: Create.
 		createResp, err := s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 			Name: volName, SizeBytes: 1 << 20,
 		})
 		if err != nil {
 			t.Fatalf("cycle %d create: %v", cycle, err)
 		}
 		primaryVS := createResp.VolumeServer
 		// Deliver initial assignment.
 		s.bs.localServerID = primaryVS
 		s.deliver(primaryVS)
 		time.Sleep(100 * time.Millisecond)
 		entry, ok := s.ms.blockRegistry.Lookup(volName)
 		if !ok {
 			t.Fatalf("cycle %d: volume not in registry", cycle)
 		}
 		// Step 2: Failover.
 		s.ms.blockRegistry.UpdateEntry(volName, func(e *BlockVolumeEntry) {
 			e.LastLeaseGrant = time.Now().Add(-1 * time.Minute)
 		})
 		s.ms.failoverBlockVolumes(primaryVS)
 		entryAfter, _ := s.ms.blockRegistry.Lookup(volName)
 		if entryAfter.Epoch <= entry.Epoch {
 			t.Fatalf("cycle %d: epoch did not increase: %d <= %d", cycle, entryAfter.Epoch, entry.Epoch)
 		}
 		// Deliver failover assignment.
 		newPrimary := entryAfter.VolumeServer
 		s.bs.localServerID = newPrimary
 		s.deliver(newPrimary)
 		time.Sleep(100 * time.Millisecond)
 		// Step 3: Reconnect.
 		s.ms.recoverBlockVolumes(primaryVS)
 		s.bs.localServerID = primaryVS
 		s.deliver(primaryVS)
 		s.bs.localServerID = newPrimary
 		s.deliver(newPrimary)
 		time.Sleep(100 * time.Millisecond)
 		// === End-of-cycle truth checks ===
 		// Registry: volume exists, epoch monotonic.
 		finalEntry, ok := s.ms.blockRegistry.Lookup(volName)
 		if !ok {
 			t.Fatalf("cycle %d: volume missing from registry at end", cycle)
 		}
 		if finalEntry.Epoch < entryAfter.Epoch {
 			t.Fatalf("cycle %d: registry epoch regressed: %d < %d", cycle, finalEntry.Epoch, entryAfter.Epoch)
 		}
 		// VS-visible: promoted vol epoch matches.
 		var volEpoch uint64
 		if err := s.store.WithVolume(entryAfter.Path, func(vol *blockvol.BlockVol) error {
 			volEpoch = vol.Epoch()
 			return nil
 		}); err != nil {
 			t.Fatalf("cycle %d: promoted vol access failed: %v", cycle, err)
 		}
 		if volEpoch < entryAfter.Epoch {
 			t.Fatalf("cycle %d: vol epoch=%d < registry=%d", cycle, volEpoch, entryAfter.Epoch)
 		}
 		// Publication: lookup matches registry truth (not just non-empty).
 		lookupResp, err := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: volName})
 		if err != nil {
 			t.Fatalf("cycle %d: lookup failed: %v", cycle, err)
 		}
 		if lookupResp.IscsiAddr != finalEntry.ISCSIAddr {
 			t.Fatalf("cycle %d: lookup iSCSI=%q != registry=%q", cycle, lookupResp.IscsiAddr, finalEntry.ISCSIAddr)
 		}
 		if lookupResp.VolumeServer != finalEntry.VolumeServer {
 			t.Fatalf("cycle %d: lookup VS=%q != registry=%q", cycle, lookupResp.VolumeServer, finalEntry.VolumeServer)
 		}
 		t.Logf("cycle %d: registry=%d vol=%d lookup=registry ✓",
 			cycle, finalEntry.Epoch, volEpoch)
 	}
 	t.Logf("P12P2 repeated cycles: %d cycles, all end-of-cycle truth checks passed", soakCycles)
 }
 // --- Runtime state hygiene: no unbounded leftovers after cycles ---
 func TestP12P2_RuntimeHygiene_NoLeftovers(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	// Create and delete several volumes to exercise lifecycle.
 	for i := 1; i <= soakCycles; i++ {
 		name := fmt.Sprintf("hygiene-vol-%d", i)
 		s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 			Name: name, SizeBytes: 1 << 20,
 		})
 		entry, _ := s.ms.blockRegistry.Lookup(name)
 		s.bs.localServerID = entry.VolumeServer
 		s.deliver(entry.VolumeServer)
 		time.Sleep(50 * time.Millisecond)
 	}
 	// Delete all volumes.
 	for i := 1; i <= soakCycles; i++ {
 		name := fmt.Sprintf("hygiene-vol-%d", i)
 		s.ms.DeleteBlockVolume(ctx, &master_pb.DeleteBlockVolumeRequest{Name: name})
 	}
 	time.Sleep(200 * time.Millisecond)
 	// Check: no stale recovery tasks.
 	activeTasks := s.bs.v2Recovery.ActiveTaskCount()
 	if activeTasks > 0 {
 		t.Fatalf("stale recovery tasks: %d (expected 0 after all volumes deleted)", activeTasks)
 	}
 	// Check: registry should have no entries for deleted volumes.
 	for i := 1; i <= soakCycles; i++ {
 		name := fmt.Sprintf("hygiene-vol-%d", i)
 		if _, ok := s.ms.blockRegistry.Lookup(name); ok {
 			t.Fatalf("stale registry entry: %s (should be deleted)", name)
 		}
 	}
 	// Check: assignment queue should not have unbounded stale entries.
 	for _, server := range []string{"vs1:9333", "vs2:9333"} {
 		pending := s.ms.blockAssignmentQueue.Peek(server)
 		// Some pending entries may exist (lease grants etc), but check for bounded size.
 		if len(pending) > soakCycles*2 {
 			t.Fatalf("unbounded stale assignments for %s: %d", server, len(pending))
 		}
 	}
 	t.Logf("P12P2 hygiene: %d volumes created+deleted, 0 stale tasks, 0 stale registry, bounded queue", soakCycles)
 }
 // --- Steady-state repeated delivery: idempotence holds over many cycles ---
 func TestP12P2_SteadyState_IdempotenceHolds(t *testing.T) {
 	s := newSoakSetup(t)
 	ctx := context.Background()
 	s.ms.CreateBlockVolume(ctx, &master_pb.CreateBlockVolumeRequest{
 		Name: "steady-vol-1", SizeBytes: 1 << 20,
 	})
 	entry, _ := s.ms.blockRegistry.Lookup("steady-vol-1")
 	s.bs.localServerID = entry.VolumeServer
 	s.deliver(entry.VolumeServer)
 	time.Sleep(200 * time.Millisecond)
 	replicaID := entry.Path + "/" + entry.Replicas[0].Server
 	eventsAfterFirst := len(s.bs.v2Orchestrator.Log.EventsFor(replicaID))
 	if eventsAfterFirst == 0 {
 		t.Fatal("first delivery must create events")
 	}
 	// Deliver the same assignment many times.
 	for i := 0; i < soakCycles*2; i++ {
 		s.deliver(entry.VolumeServer)
 		time.Sleep(20 * time.Millisecond)
 	}
 	eventsAfterSoak := len(s.bs.v2Orchestrator.Log.EventsFor(replicaID))
 	if eventsAfterSoak != eventsAfterFirst {
 		t.Fatalf("idempotence drift: events %d → %d after %d repeated deliveries",
 			eventsAfterFirst, eventsAfterSoak, soakCycles*2)
 	}
 	// Verify: registry, vol, and lookup are still coherent.
 	finalEntry, _ := s.ms.blockRegistry.Lookup("steady-vol-1")
 	if finalEntry.Epoch != entry.Epoch {
 		t.Fatalf("epoch drifted: %d → %d", entry.Epoch, finalEntry.Epoch)
 	}
 	lookupResp, _ := s.ms.LookupBlockVolume(ctx, &master_pb.LookupBlockVolumeRequest{Name: "steady-vol-1"})
 	if lookupResp.IscsiAddr != finalEntry.ISCSIAddr {
 		t.Fatalf("lookup iSCSI=%q != registry=%q after soak", lookupResp.IscsiAddr, finalEntry.ISCSIAddr)
 	}
 	if lookupResp.VolumeServer != finalEntry.VolumeServer {
 		t.Fatalf("lookup VS=%q != registry=%q after soak", lookupResp.VolumeServer, finalEntry.VolumeServer)
 	}
 	t.Logf("P12P2 steady state: %d repeated deliveries, events stable at %d, lookup=registry ✓",
 		soakCycles*2, eventsAfterFirst)
 }