feat: CP13-8 PASSES — real-workload validation on RF=2 sync_all

CP13-8 scenario results on m01/M02 (25Gbps RoCE):
  fsck_ext4:       CLEAN
  file count:      200 (assert_equal PASS)
  checksum match:  MATCH (assert_contains PASS)
  pgbench TPS:     565.69 (assert_greater PASS)
  auto-failover:   10.0.0.1:18480 → 10.0.0.3:18480

Code changes (tester + scenario):
- volume_server_block.go: readiness state, assignment lifecycle cleanup
- block_heartbeat_loop.go: readiness-aware heartbeat reporting
- store_blockvol.go: readiness tracking
- master_server_handlers_block.go: block API handler updates
- cp13-8-real-workload-validation.yaml: redesigned scenario
  (removed block_promote, use natural auto-failover flow,
  bootstrap write before wait_volume_healthy)
- testrunner/actions/devops.go: scenario action improvements
- replica_read_test.go: component-level replica read test

Phase docs: CP13-7 accepted, CP13-8/8A technical packs updated,
design docs updated for protocol closure evidence.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
pingqiu
2026-04-03 14:24:13 -07:00
co-authored by Claude Opus 4.6
parent 334c12664a
commit 4c7fbefe25
21 changed files with 2375 additions and 248 deletions
+2 -2
View File
@@ -93,7 +93,7 @@ func (c *BlockVolumeHeartbeatCollector) Run() {
select {
case <-ticker.C:
// Outbound: collect and report status.
msgs := c.blockService.Store().CollectBlockVolumeHeartbeat()
msgs := c.blockService.CollectBlockVolumeHeartbeat()
c.safeCallback(msgs)
// Inbound: process any pending assignments.
c.processAssignments()
@@ -115,7 +115,7 @@ func (c *BlockVolumeHeartbeatCollector) processAssignments() {
if len(assignments) == 0 {
return
}
errs := c.blockService.Store().ProcessBlockVolumeAssignments(assignments)
errs := c.blockService.ApplyAssignments(assignments)
c.cbMu.Lock()
cb := c.assignmentCallback
c.cbMu.Unlock()
+36
View File
@@ -463,6 +463,42 @@ func TestBlockAssign_NilSource(t *testing.T) {
}
}
// TestBlockAssign_CollectorUsesAuthoritativeLifecycle verifies the heartbeat
// collector now drives the full BlockService assignment path, not the store-only
// role path. A replica assignment must start the receiver and close publish
// readiness.
func TestBlockAssign_CollectorUsesAuthoritativeLifecycle(t *testing.T) {
bs := newTestBlockService(t)
path := testBlockVolPath(t, bs)
collector := NewBlockVolumeHeartbeatCollector(bs, 5*time.Millisecond)
collector.SetAssignmentSource(func() []blockvol.BlockVolumeAssignment {
return []blockvol.BlockVolumeAssignment{{
Path: path,
Epoch: 1,
Role: uint32(blockvol.RoleReplica),
ReplicaDataAddr: ":0",
ReplicaCtrlAddr: ":0",
}}
})
go collector.Run()
defer collector.Stop()
deadline := time.After(500 * time.Millisecond)
for {
dataAddr, ctrlAddr := bs.GetReplState(path)
readiness := bs.ReadinessSnapshot(path)
if dataAddr != "" && ctrlAddr != "" && readiness.ReceiverReady && readiness.PublishHealthy {
return
}
select {
case <-deadline:
t.Fatalf("collector did not start replica receiver: data=%q ctrl=%q readiness=%+v", dataAddr, ctrlAddr, readiness)
case <-time.After(10 * time.Millisecond):
}
}
}
// TestBlockAssign_MixedBatch verifies a batch with 1 success, 1 unknown volume,
// and 1 invalid transition returns parallel errors correctly.
func TestBlockAssign_MixedBatch(t *testing.T) {
@@ -148,7 +148,8 @@ func TestClusterHealthSummary(t *testing.T) {
Path: "/data/healthy.blk",
Role: blockvol.RoleToWire(blockvol.RolePrimary),
ReplicaFactor: 2,
Replicas: []ReplicaInfo{{Server: "vs2:9333", Role: blockvol.RoleToWire(blockvol.RoleReplica)}},
ReplicaReady: true,
Replicas: []ReplicaInfo{{Server: "vs2:9333", Role: blockvol.RoleToWire(blockvol.RoleReplica), Ready: true}},
Status: StatusActive,
})
@@ -188,7 +189,8 @@ func TestBlockStatusHandler_IncludesHealthCounts(t *testing.T) {
Path: "/data/status.blk",
Role: blockvol.RoleToWire(blockvol.RolePrimary),
ReplicaFactor: 2,
Replicas: []ReplicaInfo{{Server: "vs2:9333", Role: blockvol.RoleToWire(blockvol.RoleReplica)}},
ReplicaReady: true,
Replicas: []ReplicaInfo{{Server: "vs2:9333", Role: blockvol.RoleToWire(blockvol.RoleReplica), Ready: true}},
Status: StatusActive,
})
+42
View File
@@ -1965,3 +1965,45 @@ func TestRegistry_InflightBlocksAutoRegister(t *testing.T) {
t.Fatalf("replica health not updated after inflight released: %f", entry.Replicas[0].HealthScore)
}
}
func TestRegistry_ReplicaReadyRequiresReplicaHeartbeat(t *testing.T) {
r := NewBlockVolumeRegistry()
if err := r.Register(&BlockVolumeEntry{
Name: "vol-ready",
VolumeServer: "primary-server:8080",
Path: "/blocks/vol-ready.blk",
Status: StatusActive,
Replicas: []ReplicaInfo{{
Server: "replica-server:8080",
Path: "/blocks/vol-ready.blk",
}},
}); err != nil {
t.Fatalf("register: %v", err)
}
entry, _ := r.Lookup("vol-ready")
if entry.ReplicaReady {
t.Fatal("replica should not be ready before replica heartbeat confirms publication")
}
if !entry.ReplicaDegraded {
t.Fatal("volume should remain degraded until replica readiness closes")
}
r.UpdateFullHeartbeat("replica-server:8080", []*master_pb.BlockVolumeInfoMessage{{
Path: "/blocks/vol-ready.blk",
Epoch: 1,
Role: uint32(blockvol.RoleReplica),
VolumeSize: 1 << 30,
HealthScore: 0.9,
ReplicaDataAddr: "10.0.0.2:14260",
ReplicaCtrlAddr: "10.0.0.2:14261",
}}, "")
entry, _ = r.Lookup("vol-ready")
if !entry.Replicas[0].Ready {
t.Fatal("replica heartbeat with published receiver addresses should mark replica ready")
}
if !entry.ReplicaReady {
t.Fatal("aggregate replica readiness should become true after replica heartbeat")
}
}
@@ -394,6 +394,7 @@ func entryToVolumeInfo(e *BlockVolumeEntry, primaryAlive bool) blockapi.VolumeIn
ReplicaDataAddr: e.ReplicaDataAddr,
ReplicaCtrlAddr: e.ReplicaCtrlAddr,
ReplicaFactor: rf,
ReplicaReady: e.ReplicaReady,
HealthScore: e.HealthScore,
ReplicaDegraded: e.ReplicaDegraded,
DurabilityMode: durMode,
@@ -407,6 +408,7 @@ func entryToVolumeInfo(e *BlockVolumeEntry, primaryAlive bool) blockapi.VolumeIn
Server: ri.Server,
ISCSIAddr: ri.ISCSIAddr,
IQN: ri.IQN,
Ready: ri.Ready,
HealthScore: ri.HealthScore,
WALLag: ri.WALLag,
})
+168 -54
View File
@@ -24,6 +24,23 @@ type volReplState struct {
replicaCtrlAddr string
// allReplicas stores the full replica set for multi-replica idempotence.
allReplicas []blockvol.ReplicaAddr
roleApplied bool
receiverReady bool
shipperConfigured bool
replicaEligible bool
publishHealthy bool
}
// BlockReadinessSnapshot names the assignment-to-publication closure at the
// BlockService boundary. These flags are owned by the service/adapter layer,
// not by blockvol's local storage mechanics.
type BlockReadinessSnapshot struct {
RoleApplied bool
ReceiverReady bool
ShipperConfigured bool
ShipperConnected bool
ReplicaEligible bool
PublishHealthy bool
}
// NVMeConfig holds NVMe/TCP target configuration passed from CLI flags.
@@ -373,6 +390,15 @@ func (bs *BlockService) DeleteBlockVol(name string) error {
// ProcessAssignments applies assignments from master, including replication setup.
// V2 bridge: also delivers each assignment to the V2 engine for recovery ownership.
func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAssignment) {
_ = bs.ApplyAssignments(assignments)
}
// ApplyAssignments applies assignments through the single authoritative
// BlockService lifecycle: role apply, replication wiring, and publication
// readiness bookkeeping. Returns per-assignment errors parallel to the input.
func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssignment) []error {
errs := make([]error, len(assignments))
// V2 bridge: convert and deliver to engine orchestrator (Phase 08 P1).
// P3: skip V2 processing for repeated unchanged assignments.
// P4: RecoveryManager starts/cancels recovery goroutines based on results.
@@ -400,9 +426,9 @@ func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAss
// V1 processing (requires blockStore).
if bs.blockStore == nil {
return
return errs
}
for _, a := range assignments {
for i, a := range assignments {
role := blockvol.RoleFromWire(a.Role)
ttl := blockvol.LeaseTTLFromWire(a.LeaseTtlMs)
@@ -410,22 +436,30 @@ func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAss
if err := bs.blockStore.WithVolume(a.Path, func(vol *blockvol.BlockVol) error {
return vol.HandleAssignment(a.Epoch, role, ttl)
}); err != nil {
errs[i] = err
glog.Warningf("block service: assignment %s epoch=%d role=%s: %v", a.Path, a.Epoch, role, err)
continue
}
bs.noteRoleApplied(a.Path, role)
// 2. Replication setup based on role + addresses.
switch role {
case blockvol.RolePrimary:
// CP8-2: ReplicaAddrs (multi-replica) takes precedence over scalar fields.
if len(a.ReplicaAddrs) > 0 {
bs.setupPrimaryReplicationMulti(a.Path, a.ReplicaAddrs)
if err := bs.setupPrimaryReplicationMulti(a.Path, a.ReplicaAddrs); err != nil {
errs[i] = err
}
} else if a.ReplicaDataAddr != "" && a.ReplicaCtrlAddr != "" {
bs.setupPrimaryReplication(a.Path, a.ReplicaDataAddr, a.ReplicaCtrlAddr)
if err := bs.setupPrimaryReplication(a.Path, a.ReplicaDataAddr, a.ReplicaCtrlAddr); err != nil {
errs[i] = err
}
}
case blockvol.RoleReplica:
if a.ReplicaDataAddr != "" && a.ReplicaCtrlAddr != "" {
bs.setupReplicaReceiver(a.Path, a.ReplicaDataAddr, a.ReplicaCtrlAddr)
if err := bs.setupReplicaReceiver(a.Path, a.ReplicaDataAddr, a.ReplicaCtrlAddr); err != nil {
errs[i] = err
}
}
case blockvol.RoleRebuilding:
if a.RebuildAddr != "" {
@@ -433,18 +467,23 @@ func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAss
}
}
}
return errs
}
// setupPrimaryReplication configures WAL shipping from primary to replica
// and starts the rebuild server (R1-2).
func (bs *BlockService) setupPrimaryReplication(path, replicaDataAddr, replicaCtrlAddr string) {
func (bs *BlockService) setupPrimaryReplication(path, replicaDataAddr, replicaCtrlAddr string) error {
// P3 idempotence: skip if replica state is unchanged.
bs.replMu.RLock()
existing := bs.replStates[path]
bs.replMu.RUnlock()
if existing != nil && existing.replicaDataAddr == replicaDataAddr && existing.replicaCtrlAddr == replicaCtrlAddr {
// Unchanged repeated assignment — idempotent, no side effects.
return
bs.markPrimaryTransportConfigured(path, []blockvol.ReplicaAddr{{
DataAddr: replicaDataAddr,
CtrlAddr: replicaCtrlAddr,
}})
return nil
}
// Compute deterministic rebuild listen address.
@@ -465,27 +504,19 @@ func (bs *BlockService) setupPrimaryReplication(path, replicaDataAddr, replicaCt
return nil
}); err != nil {
glog.Warningf("block service: setup primary replication %s: %v", path, err)
return
return err
}
// Track replication state for heartbeat reporting (R1-4).
// These addresses are what the primary ships to — they come from the
// master's assignment. They should already be canonical (from
// AllocateBlockVolumeResponse), but if not, they'll be reported as-is.
bs.replMu.Lock()
if bs.replStates == nil {
bs.replStates = make(map[string]*volReplState)
}
bs.replStates[path] = &volReplState{
replicaDataAddr: replicaDataAddr,
replicaCtrlAddr: replicaCtrlAddr,
}
bs.replMu.Unlock()
bs.markPrimaryTransportConfigured(path, []blockvol.ReplicaAddr{{
DataAddr: replicaDataAddr,
CtrlAddr: replicaCtrlAddr,
}})
glog.V(0).Infof("block service: primary %s shipping WAL to %s/%s (rebuild=%s)", path, replicaDataAddr, replicaCtrlAddr, rebuildAddr)
return nil
}
// setupPrimaryReplicationMulti configures WAL shipping from primary to N replicas
// using SetReplicaAddrs (CP8-2: multi-replica support).
func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockvol.ReplicaAddr) {
func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockvol.ReplicaAddr) error {
// P3 idempotence: skip if ALL replica addresses unchanged.
// Compare full replica set, not just the first entry.
if len(addrs) > 0 {
@@ -493,7 +524,8 @@ func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockv
existing := bs.replStates[path]
bs.replMu.RUnlock()
if existing != nil && bs.multiReplicaUnchanged(path, addrs) {
return
bs.markPrimaryTransportConfigured(path, addrs)
return nil
}
}
@@ -513,30 +545,15 @@ func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockv
return nil
}); err != nil {
glog.Warningf("block service: setup primary replication (multi) %s: %v", path, err)
return
return err
}
// Track replication state for heartbeat reporting.
bs.replMu.Lock()
if bs.replStates == nil {
bs.replStates = make(map[string]*volReplState)
}
// Store full replica set + first replica for backward compat heartbeat.
if len(addrs) > 0 {
// Copy the addrs slice to avoid aliasing.
copied := make([]blockvol.ReplicaAddr, len(addrs))
copy(copied, addrs)
bs.replStates[path] = &volReplState{
replicaDataAddr: addrs[0].DataAddr,
replicaCtrlAddr: addrs[0].CtrlAddr,
allReplicas: copied,
}
}
bs.replMu.Unlock()
bs.markPrimaryTransportConfigured(path, addrs)
glog.V(0).Infof("block service: primary %s shipping WAL to %d replicas (rebuild=%s)", path, len(addrs), rebuildAddr)
return nil
}
// setupReplicaReceiver starts the replica WAL receiver.
func (bs *BlockService) setupReplicaReceiver(path, dataAddr, ctrlAddr string) {
func (bs *BlockService) setupReplicaReceiver(path, dataAddr, ctrlAddr string) error {
// CP13-2: Pass the routable advertisedIP (from -ip flag, NOT from -id/serverID)
// so wildcard-bind listeners resolve to a real IP, not an opaque identity string.
var canonDataAddr, canonCtrlAddr string
@@ -559,7 +576,7 @@ func (bs *BlockService) setupReplicaReceiver(path, dataAddr, ctrlAddr string) {
return nil
}); err != nil {
glog.Warningf("block service: setup replica receiver %s: %v", path, err)
return
return err
}
// Fallback to assignment addresses if receiver didn't report.
if canonDataAddr == "" {
@@ -568,16 +585,9 @@ func (bs *BlockService) setupReplicaReceiver(path, dataAddr, ctrlAddr string) {
if canonCtrlAddr == "" {
canonCtrlAddr = ctrlAddr
}
bs.replMu.Lock()
if bs.replStates == nil {
bs.replStates = make(map[string]*volReplState)
}
bs.replStates[path] = &volReplState{
replicaDataAddr: canonDataAddr,
replicaCtrlAddr: canonCtrlAddr,
}
bs.replMu.Unlock()
bs.markReceiverReady(path, canonDataAddr, canonCtrlAddr)
glog.V(0).Infof("block service: replica %s receiving on %s/%s", path, canonDataAddr, canonCtrlAddr)
return nil
}
// startRebuild starts a rebuild in the background.
@@ -722,8 +732,10 @@ func (bs *BlockService) CollectBlockVolumeHeartbeat() []blockvol.BlockVolumeInfo
defer bs.replMu.RUnlock()
for i := range msgs {
if s, ok := bs.replStates[msgs[i].Path]; ok {
msgs[i].ReplicaDataAddr = s.replicaDataAddr
msgs[i].ReplicaCtrlAddr = s.replicaCtrlAddr
if s.publishHealthy {
msgs[i].ReplicaDataAddr = s.replicaDataAddr
msgs[i].ReplicaCtrlAddr = s.replicaCtrlAddr
}
}
// NVMe publication: report nvme_addr and nqn if NVMe target is running.
if bs.nvmeListenAddr != "" {
@@ -758,6 +770,108 @@ func (bs *BlockService) multiReplicaUnchanged(path string, addrs []blockvol.Repl
return true
}
func (bs *BlockService) ensureReplStateLocked(path string) *volReplState {
if bs.replStates == nil {
bs.replStates = make(map[string]*volReplState)
}
state := bs.replStates[path]
if state == nil {
state = &volReplState{}
bs.replStates[path] = state
}
return state
}
func (bs *BlockService) noteRoleApplied(path string, role blockvol.Role) {
bs.replMu.Lock()
defer bs.replMu.Unlock()
state := bs.ensureReplStateLocked(path)
state.roleApplied = true
switch role {
case blockvol.RoleReplica:
state.receiverReady = false
state.shipperConfigured = false
state.replicaEligible = false
state.publishHealthy = false
case blockvol.RolePrimary:
state.receiverReady = false
state.shipperConfigured = false
state.replicaEligible = false
state.publishHealthy = true
case blockvol.RoleRebuilding:
state.receiverReady = false
state.shipperConfigured = false
state.replicaEligible = false
state.publishHealthy = false
default:
state.receiverReady = false
state.shipperConfigured = false
state.replicaEligible = false
state.publishHealthy = false
state.replicaDataAddr = ""
state.replicaCtrlAddr = ""
state.allReplicas = nil
}
}
func (bs *BlockService) markPrimaryTransportConfigured(path string, addrs []blockvol.ReplicaAddr) {
bs.replMu.Lock()
defer bs.replMu.Unlock()
state := bs.ensureReplStateLocked(path)
state.shipperConfigured = len(addrs) > 0
state.publishHealthy = true
state.replicaEligible = false
state.receiverReady = false
if len(addrs) == 0 {
state.replicaDataAddr = ""
state.replicaCtrlAddr = ""
state.allReplicas = nil
return
}
copied := make([]blockvol.ReplicaAddr, len(addrs))
copy(copied, addrs)
state.allReplicas = copied
state.replicaDataAddr = addrs[0].DataAddr
state.replicaCtrlAddr = addrs[0].CtrlAddr
}
func (bs *BlockService) markReceiverReady(path, dataAddr, ctrlAddr string) {
bs.replMu.Lock()
defer bs.replMu.Unlock()
state := bs.ensureReplStateLocked(path)
state.receiverReady = true
state.replicaEligible = true
state.publishHealthy = true
state.shipperConfigured = false
state.replicaDataAddr = dataAddr
state.replicaCtrlAddr = ctrlAddr
state.allReplicas = nil
}
// ReadinessSnapshot reports the service-owned assignment/readiness closure for
// one volume. It keeps v2 publication truth above blockvol's local mechanics.
func (bs *BlockService) ReadinessSnapshot(path string) BlockReadinessSnapshot {
snap := BlockReadinessSnapshot{}
bs.replMu.RLock()
state := bs.replStates[path]
if state != nil {
snap.RoleApplied = state.roleApplied
snap.ReceiverReady = state.receiverReady
snap.ShipperConfigured = state.shipperConfigured
snap.ReplicaEligible = state.replicaEligible
snap.PublishHealthy = state.publishHealthy
}
bs.replMu.RUnlock()
if !snap.ShipperConfigured || bs.blockStore == nil {
return snap
}
_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
snap.ShipperConnected = len(vol.ReplicaShipperStates()) > 0 && !vol.Status().ReplicaDegraded
return nil
})
return snap
}
// --- P3: Assignment idempotence ---
// lastAppliedAssignment stores the full assignment for idempotence comparison.
+26 -13
View File
@@ -17,13 +17,19 @@ type ShipperDebugInfo struct {
// BlockVolumeDebugInfo is the real-time block volume state.
type BlockVolumeDebugInfo struct {
Path string `json:"path"`
Role string `json:"role"`
Epoch uint64 `json:"epoch"`
HeadLSN uint64 `json:"head_lsn"`
Degraded bool `json:"degraded"`
Shippers []ShipperDebugInfo `json:"shippers,omitempty"`
Timestamp string `json:"timestamp"`
Path string `json:"path"`
Role string `json:"role"`
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"`
Shippers []ShipperDebugInfo `json:"shippers,omitempty"`
Timestamp string `json:"timestamp"`
}
// debugBlockShipperHandler returns real-time shipper state for all block volumes.
@@ -48,13 +54,20 @@ func (vs *VolumeServer) debugBlockShipperHandler(w http.ResponseWriter, r *http.
var infos []BlockVolumeDebugInfo
store.IterateBlockVolumes(func(path string, vol *blockvol.BlockVol) {
status := vol.Status()
readiness := vs.blockService.ReadinessSnapshot(path)
info := BlockVolumeDebugInfo{
Path: path,
Role: status.Role.String(),
Epoch: status.Epoch,
HeadLSN: status.WALHeadLSN,
Degraded: status.ReplicaDegraded,
Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
Path: path,
Role: status.Role.String(),
Epoch: status.Epoch,
HeadLSN: status.WALHeadLSN,
Degraded: status.ReplicaDegraded,
RoleApplied: readiness.RoleApplied,
ReceiverReady: readiness.ReceiverReady,
ShipperConfigured: readiness.ShipperConfigured,
ShipperConnected: readiness.ShipperConnected,
ReplicaEligible: readiness.ReplicaEligible,
PublishHealthy: readiness.PublishHealthy,
Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
}
// Get per-shipper state from ShipperGroup if available.