From 680b530314e298148ce2e7b1b1231222515e59cf Mon Sep 17 00:00:00 2001 From: pingqiu Date: Sat, 4 Apr 2026 00:43:30 -0700 Subject: [PATCH] =?UTF-8?q?refactor:=20Task=20E=20=E2=80=94=20reader=20ret?= =?UTF-8?q?urns=20bridge.BlockVolState=20directly?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reader backend-binding extraction: - v2bridge/reader.go: Reader.ReadState() now returns bridge.BlockVolState directly instead of a local v2bridge.BlockVolState mirror type. Removed the local BlockVolState type entirely. - block_recovery.go: removed readerShimForRecovery (12 lines of 1:1 field copying). Reader is now passed directly as bridge.BlockVolReader. Before: v2bridge.Reader → v2bridge.BlockVolState → readerShim → bridge.BlockVolState After: v2bridge.Reader → bridge.BlockVolState (direct) v2bridge now imports sw-block/bridge/blockvol for the contract type (control.go already did this, reader.go now follows the same pattern). Validation: - go test ./sw-block/bridge/blockvol/... → PASS - go test ./weed/storage/blockvol/v2bridge/ -run "TestReader_" → PASS - go test ./weed/server/ -run "TestP4_|TestP16B_" → PASS (8 tests) Co-Authored-By: Claude Opus 4.6 (1M context) --- weed/server/block_recovery.go | 48 +++++++++++------------- weed/storage/blockvol/v2bridge/reader.go | 43 ++++++++------------- 2 files changed, 36 insertions(+), 55 deletions(-) diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 4761075b8..8f815fa27 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -62,32 +62,39 @@ func NewRecoveryManager(bs *BlockService) *RecoveryManager { } } -// 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 +// HandleAssignmentResult preserves the pre-16D behavior for no-core paths and +// older tests: session creation/supersede results directly start recovery +// goroutines. Core-present paths should use StartRecoveryTask instead. 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) } } +// HandleRemovedAssignments drains tasks for senders removed by registry +// reconciliation. Recovery task startup is handled separately by core command +// execution on the bounded live path. +func (rm *RecoveryManager) HandleRemovedAssignments(result engine.AssignmentResult) { + for _, replicaID := range result.Removed { + rm.cancelAndDrain(replicaID, true) + } +} + +// StartRecoveryTask starts one bounded recovery goroutine from a core-emitted +// command. Any stale task for the same replica is drained first. +func (rm *RecoveryManager) StartRecoveryTask(replicaID string, assignments []blockvol.BlockVolumeAssignment) { + 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) { @@ -244,7 +251,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd reader := v2bridge.NewReader(vol) pinner := v2bridge.NewPinner(vol) sa = bridge.NewStorageAdapter( - &readerShimForRecovery{reader}, + reader, &pinnerShimForRecovery{pinner}, ) if s := bs.v2Orchestrator.Registry.Sender(replicaID); s != nil { @@ -324,7 +331,7 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd reader := v2bridge.NewReader(vol) pinner := v2bridge.NewPinner(vol) sa = bridge.NewStorageAdapter( - &readerShimForRecovery{reader}, + reader, &pinnerShimForRecovery{pinner}, ) executor = v2bridge.NewExecutor(vol, rebuildAddr) @@ -514,19 +521,6 @@ func (rm *RecoveryManager) volumePathForReplica(replicaID string) string { // --- 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) { diff --git a/weed/storage/blockvol/v2bridge/reader.go b/weed/storage/blockvol/v2bridge/reader.go index ec28f28b0..69ffb6146 100644 --- a/weed/storage/blockvol/v2bridge/reader.go +++ b/weed/storage/blockvol/v2bridge/reader.go @@ -4,28 +4,18 @@ // // Import direction: // v2bridge → blockvol (real state) -// v2bridge exports types matching sw-block/bridge/blockvol/ contracts -// v2bridge does NOT import sw-block/ (to avoid cross-module dependency) -// -// The sw-block/bridge/blockvol/ package consumes these implementations -// through its contract interfaces (BlockVolReader, BlockVolPinner, etc.). +// v2bridge → sw-block/bridge/blockvol (contract types) +// v2bridge → sw-block/engine/replication (engine types) package v2bridge import ( + bridge "github.com/seaweedfs/seaweedfs/sw-block/bridge/blockvol" "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" ) -// BlockVolState mirrors the contract type from sw-block/bridge/blockvol/. -// Fields are populated from real blockvol internals. -type BlockVolState struct { - WALHeadLSN uint64 - WALTailLSN uint64 - CommittedLSN uint64 - CheckpointLSN uint64 - CheckpointTrusted bool -} - -// Reader implements BlockVolReader by reading real blockvol fields. +// Reader implements bridge.BlockVolReader by reading real blockvol fields. +// This is a thin backend binding — it fetches a snapshot from real BlockVol +// and returns the contract type directly. No state-shaping logic here. type Reader struct { vol *blockvol.BlockVol } @@ -36,20 +26,17 @@ func NewReader(vol *blockvol.BlockVol) *Reader { } // ReadState reads current blockvol state from real fields. -// Each field maps to a specific blockvol source: +// Returns bridge.BlockVolState directly (no intermediate local type). // -// WALHeadLSN ← vol.nextLSN - 1 (last written LSN) -// WALTailLSN ← vol.super.WALCheckpointLSN (LSN boundary, not byte offset) -// CommittedLSN ← vol.flusher.CheckpointLSN() (V1 interim: committed = checkpointed) -// CheckpointLSN ← vol.super.WALCheckpointLSN -// CheckpointTrusted ← vol.super.Validate() == nil (superblock integrity) -// -// Note: CommittedLSN maps to CheckpointLSN in the current V1 model where -// barrier-confirmed = flusher-checkpointed. In V2, these may diverge when -// distributed commit is separated from local flush. -func (r *Reader) ReadState() BlockVolState { +// Field mapping: +// WALHeadLSN ← StatusSnapshot().WALHeadLSN +// WALTailLSN ← StatusSnapshot().WALTailLSN +// CommittedLSN ← StatusSnapshot().CommittedLSN +// CheckpointLSN ← StatusSnapshot().CheckpointLSN +// CheckpointTrusted ← StatusSnapshot().CheckpointTrusted +func (r *Reader) ReadState() bridge.BlockVolState { snap := r.vol.StatusSnapshot() - return BlockVolState{ + return bridge.BlockVolState{ WALHeadLSN: snap.WALHeadLSN, WALTailLSN: snap.WALTailLSN, CommittedLSN: snap.CommittedLSN,