refactor: Task E — reader returns bridge.BlockVolState directly

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) <noreply@anthropic.com>

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) {

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,