diff --git a/weed/server/volume_grpc_block_activation_gate_test.go b/weed/server/volume_grpc_block_activation_gate_test.go new file mode 100644 index 000000000..cd9626843 --- /dev/null +++ b/weed/server/volume_grpc_block_activation_gate_test.go @@ -0,0 +1,181 @@ +package weed_server + +import ( + "testing" + + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +func TestT4_DegradedProjection_GatesActivation(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-degraded") + + // Inject degraded projection. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeDegraded, Reason: "barrier_timeout"}, + } + bs.coreProjMu.Unlock() + + bs.evaluateActivationGate(path) + + gated, reason := bs.IsActivationGated(path) + if !gated { + t.Fatal("expected activation gated for degraded projection") + } + if reason == "" { + t.Fatal("expected non-empty gate reason") + } +} + +func TestT4_NeedsRebuildProjection_GatesActivation(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-rebuild") + + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeNeedsRebuild, Reason: "gap_too_large"}, + } + bs.coreProjMu.Unlock() + + bs.evaluateActivationGate(path) + + gated, reason := bs.IsActivationGated(path) + if !gated { + t.Fatal("expected activation gated for needs_rebuild projection") + } + if reason == "" { + t.Fatal("expected non-empty gate reason") + } +} + +func TestT4_HealthyProjection_ClearsGate(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-healthy") + + // Start gated. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeDegraded, Reason: "barrier_timeout"}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + if gated, _ := bs.IsActivationGated(path); !gated { + t.Fatal("expected gated initially") + } + + // Transition to healthy. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModePublishHealthy}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + + if gated, _ := bs.IsActivationGated(path); gated { + t.Fatal("expected gate cleared for publish_healthy projection") + } +} + +func TestT4_GateEnforcedBeforeHeartbeat(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-pre-hb") + + // Inject degraded projection and evaluate gate. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeDegraded, Reason: "test"}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + + // Gate is set BEFORE any heartbeat. Verify the heartbeat carries + // the gated state. + msgs := bs.CollectBlockVolumeHeartbeat() + var found *blockvol.BlockVolumeInfoMessage + for i := range msgs { + if msgs[i].Path == path { + found = &msgs[i] + break + } + } + if found == nil { + t.Fatal("heartbeat message not found for gated volume") + } + if !found.ActivationGated { + t.Fatal("heartbeat should report ActivationGated=true") + } + if found.ActivationGateReason == "" { + t.Fatal("heartbeat should report non-empty ActivationGateReason") + } +} + +func TestT4_RecoveryFromGated_ReenablesServing(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-recovery") + + // Start gated (needs_rebuild). + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeNeedsRebuild, Reason: "gap"}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + if gated, _ := bs.IsActivationGated(path); !gated { + t.Fatal("expected gated") + } + + // Simulate recovery: projection transitions to replica_ready. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModeReplicaReady}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + if gated, _ := bs.IsActivationGated(path); gated { + t.Fatal("expected gate cleared after recovery to replica_ready") + } + + // And then to publish_healthy. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + Mode: engine.ModeView{Name: engine.ModePublishHealthy}, + } + bs.coreProjMu.Unlock() + bs.evaluateActivationGate(path) + if gated, _ := bs.IsActivationGated(path); gated { + t.Fatal("expected gate cleared after recovery to publish_healthy") + } +} + +func TestT4_ApplyCoreAssignment_GatesDegradedPrimary(t *testing.T) { + bs := newTestBlockServiceDirect(t) + path := createTestVolDirect(t, bs, "gate-assignment") + + // Pre-inject degraded projection so that after assignment processing + // the gate is evaluated. + bs.coreProjMu.Lock() + bs.coreProj[path] = engine.PublicationProjection{ + VolumeID: path, + Mode: engine.ModeView{Name: engine.ModeDegraded, Reason: "incomplete_reconstruction"}, + } + bs.coreProjMu.Unlock() + + // Process primary assignment through the core path. + bs.applyCoreAssignmentEvent(blockvol.BlockVolumeAssignment{ + Path: path, + Epoch: 5, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + LeaseTtlMs: 30000, + }) + + // The gate should have been evaluated after assignment. + gated, reason := bs.IsActivationGated(path) + if !gated { + t.Fatal("expected activation gated after primary assignment with degraded projection") + } + if reason == "" { + t.Fatal("expected non-empty gate reason") + } +} diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index f787279ee..37366964b 100644 --- a/weed/server/volume_server_block.go +++ b/weed/server/volume_server_block.go @@ -84,6 +84,11 @@ type BlockService struct { coreExecMu sync.RWMutex coreExec map[string][]string + // T4: activation gate — promoted primaries that have not passed + // reconstruction quality check are gated from serving. + activationGateMu sync.RWMutex + activationGated map[string]string // path → reason (non-empty = gated) + // P3: last-applied assignment per volume path for idempotence. lastAssignMu sync.RWMutex lastAssign map[string]lastAppliedAssignment @@ -234,6 +239,7 @@ func StartBlockService(listenAddr, blockDir, iqnPrefix, portalAddr string, nvmeC v2Core: engine.NewCoreEngine(), localServerID: listenAddr, // INTERIM: transport-shaped, see field doc coreProj: make(map[string]engine.PublicationProjection), + activationGated: make(map[string]string), blockInventoryAuthoritative: false, } bs.v2Recovery = NewRecoveryManager(bs) @@ -560,7 +566,62 @@ func (bs *BlockService) applyCoreAssignmentEvent(a blockvol.BlockVolumeAssignmen return nil } result := bs.coreApplyAndLog(ev) - return bs.applyCoreCommandsWithAssignment(result.Commands, &a) + if err := bs.applyCoreCommandsWithAssignment(result.Commands, &a); err != nil { + return err + } + // T4: After assignment execution, check the resulting V2 core projection + // and gate activation locally if the mode indicates the node should not + // serve. This is the enforcement point — it happens immediately after + // assignment, before the next heartbeat round-trip. + bs.evaluateActivationGate(a.Path) + return nil +} + +// evaluateActivationGate checks the current V2 core projection for a volume +// and gates or clears activation accordingly. This is the local enforcement +// point for T4 — the promoted node decides locally whether reconstruction +// quality allows serving. +func (bs *BlockService) evaluateActivationGate(path string) { + proj, ok := bs.CoreProjection(path) + if !ok { + return // no V2 core, no gate enforcement + } + bs.activationGateMu.Lock() + defer bs.activationGateMu.Unlock() + switch proj.Mode.Name { + case "publish_healthy", "replica_ready": + delete(bs.activationGated, path) + default: + reason := fmt.Sprintf("engine_projection_mode=%s", proj.Mode.Name) + if proj.Mode.Reason != "" { + reason += ": " + proj.Mode.Reason + } + bs.activationGated[path] = reason + glog.V(0).Infof("activation gated: %s — %s", path, reason) + } +} + +// IsActivationGated returns whether a volume is currently gated from serving +// and the reason. Used by iSCSI/NVMe adapter and heartbeat surface. +func (bs *BlockService) IsActivationGated(path string) (bool, string) { + if bs == nil { + return false, "" + } + bs.activationGateMu.RLock() + defer bs.activationGateMu.RUnlock() + reason, gated := bs.activationGated[path] + return gated, reason +} + +// ClearActivationGate removes the activation gate for a volume. Called when +// recovery completes and the projection reaches a serving-allowed state. +func (bs *BlockService) ClearActivationGate(path string) { + if bs == nil { + return + } + bs.activationGateMu.Lock() + defer bs.activationGateMu.Unlock() + delete(bs.activationGated, path) } func (bs *BlockService) applyCoreEvent(ev engine.Event) { @@ -1018,6 +1079,10 @@ func (bs *BlockService) CollectBlockVolumeHeartbeat() []blockvol.BlockVolumeInfo } msgs[i].ReplicaDegraded = bs.heartbeatReplicaDegraded(msgs[i].Path, msgs[i].ReplicaDegraded) msgs[i].EngineProjectionMode = bs.heartbeatEngineProjectionMode(msgs[i].Path) + if gated, reason := bs.IsActivationGated(msgs[i].Path); gated { + msgs[i].ActivationGated = true + msgs[i].ActivationGateReason = reason + } // NVMe publication: report nvme_addr and nqn if NVMe target is running. if bs.nvmeListenAddr != "" { msgs[i].NvmeAddr = bs.nvmeListenAddr diff --git a/weed/server/volume_server_block_test.go b/weed/server/volume_server_block_test.go index 6cfa2f716..ea311654a 100644 --- a/weed/server/volume_server_block_test.go +++ b/weed/server/volume_server_block_test.go @@ -76,8 +76,9 @@ func newTestBlockServiceDirect(t *testing.T) *BlockService { iqnPrefix: "iqn.2024-01.com.seaweedfs:vol.", replStates: make(map[string]*volReplState), v2Core: engine.NewCoreEngine(), - coreProj: make(map[string]engine.PublicationProjection), - localServerID: "vs-test", + coreProj: make(map[string]engine.PublicationProjection), + activationGated: make(map[string]string), + localServerID: "vs-test", } } diff --git a/weed/storage/blockvol/block_heartbeat.go b/weed/storage/blockvol/block_heartbeat.go index e437679a4..178acad61 100644 --- a/weed/storage/blockvol/block_heartbeat.go +++ b/weed/storage/blockvol/block_heartbeat.go @@ -33,6 +33,8 @@ type BlockVolumeInfoMessage struct { NQN string // NVMe subsystem NQN, empty if NVMe disabled ReplicaShipperStates []ReplicaShipperStatus // CP13-7: per-replica state from primary's shipper group EngineProjectionMode string // T1: pure V2 engine-derived local projection mode (distinct from VolumeMode) + ActivationGated bool // T4: true if this volume is gated from serving after promotion + ActivationGateReason string // T4: reason for activation gate (empty if not gated) } // BlockVolumeShortInfoMessage is used for delta heartbeats