diff --git a/sw-block/engine/replication/runtime/pending.go b/sw-block/engine/replication/runtime/pending.go index fa19f671b..4eef5291e 100644 --- a/sw-block/engine/replication/runtime/pending.go +++ b/sw-block/engine/replication/runtime/pending.go @@ -3,29 +3,36 @@ // be used by any adapter shell. package runtime -import "sync" +import ( + "sync" + + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" +) // PendingExecution holds the state needed to execute a planned recovery // action. The coordinator stores one pending execution per volume and // matches it against incoming commands. +// +// All fields are typed — no interface{} handles. The host adapter builds +// these from concrete BlockVol bindings; the coordinator and execution +// helpers consume them without type assertions. type PendingExecution struct { VolumeID string ReplicaID string // CatchUpTarget is the target LSN for catch-up execution. - // Used by ExecutePendingCatchUp for fail-closed target matching. + // Used by TakeCatchUp for fail-closed target matching. CatchUpTarget uint64 // RebuildTargetLSN is the target LSN for rebuild execution. - // Used by ExecutePendingRebuild for fail-closed target matching. + // Used by TakeRebuild for fail-closed target matching. RebuildTargetLSN uint64 - // Opaque handles — the coordinator stores these but doesn't interpret them. - // The host adapter supplies concrete types (driver, plan, IO bindings). - Driver interface{} - Plan interface{} - CatchUpIO interface{} - RebuildIO interface{} + // Typed recovery handles — no interface{} drift. + Driver *engine.RecoveryDriver + Plan *engine.RecoveryPlan + CatchUpIO engine.CatchUpIO + RebuildIO engine.RebuildIO } // CancelFunc is called when a pending execution is cancelled due to diff --git a/weed/server/block_recovery.go b/weed/server/block_recovery.go index 4cbab4d60..6e7c20cb7 100644 --- a/weed/server/block_recovery.go +++ b/weed/server/block_recovery.go @@ -53,11 +53,7 @@ func NewRecoveryManager(bs *BlockService) *RecoveryManager { } rm.coord = rt.NewPendingCoordinator(func(pe *rt.PendingExecution, reason string) { if pe != nil && pe.Driver != nil && pe.Plan != nil { - if drv, ok := pe.Driver.(*engine.RecoveryDriver); ok { - if plan, ok := pe.Plan.(*engine.RecoveryPlan); ok { - drv.CancelPlan(plan, reason) - } - } + pe.Driver.CancelPlan(pe.Plan, reason) } }) return rm @@ -377,30 +373,18 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd func (rm *RecoveryManager) ExecutePendingCatchUp(volumeID string, targetLSN uint64) error { pe := rm.coord.TakeCatchUp(volumeID, targetLSN) - if pe == nil { + if pe == nil || pe.Driver == nil || pe.Plan == nil { return nil } - drv, _ := pe.Driver.(*engine.RecoveryDriver) - plan, _ := pe.Plan.(*engine.RecoveryPlan) - io, _ := pe.CatchUpIO.(engine.CatchUpIO) - if drv == nil || plan == nil { - return nil - } - return rt.ExecuteCatchUpPlan(drv, plan, io, volumeID, rm) + return rt.ExecuteCatchUpPlan(pe.Driver, pe.Plan, pe.CatchUpIO, volumeID, rm) } func (rm *RecoveryManager) ExecutePendingRebuild(volumeID string, targetLSN uint64) error { pe := rm.coord.TakeRebuild(volumeID, targetLSN) - if pe == nil { + if pe == nil || pe.Driver == nil || pe.Plan == nil { return nil } - drv, _ := pe.Driver.(*engine.RecoveryDriver) - plan, _ := pe.Plan.(*engine.RecoveryPlan) - io, _ := pe.RebuildIO.(engine.RebuildIO) - if drv == nil || plan == nil { - return nil - } - return rt.ExecuteRebuildPlan(drv, plan, io, volumeID, rm) + return rt.ExecuteRebuildPlan(pe.Driver, pe.Plan, pe.RebuildIO, volumeID, rm) } // RecoveryCallbacks implementation — host-side completion notifications. diff --git a/weed/server/block_recovery_test.go b/weed/server/block_recovery_test.go index 9ea166133..a9fc844cb 100644 --- a/weed/server/block_recovery_test.go +++ b/weed/server/block_recovery_test.go @@ -346,9 +346,7 @@ func TestP16B_RunRebuild_UsesCoreStartRebuildCommandOnLivePath(t *testing.T) { if volumeID != volPath || pending == nil || pending.Plan == nil { return } - if plan, ok := pending.Plan.(*engine.RecoveryPlan); ok { - pending.RebuildIO = fakeRebuildIO{achievedLSN: plan.RebuildTargetLSN} - } + pending.RebuildIO = fakeRebuildIO{achievedLSN: pending.Plan.RebuildTargetLSN} } _, _, rebuildPort := bs.ReplicationPorts(volPath) rebuildAddr := fmt.Sprintf("127.0.0.1:%d", rebuildPort) diff --git a/weed/server/volume_server_block_test.go b/weed/server/volume_server_block_test.go index 56a16117d..c47c25c93 100644 --- a/weed/server/volume_server_block_test.go +++ b/weed/server/volume_server_block_test.go @@ -443,9 +443,7 @@ func TestBlockService_ApplyAssignments_PrimaryRole_UsesCoreStartRecoveryTaskForC bs.v2Recovery.OnPendingExecution = func(volumeID string, pending *rt.PendingExecution) { if volumeID == path && pending != nil && pending.Plan != nil { - if plan, ok := pending.Plan.(*engine.RecoveryPlan); ok { - pending.CatchUpIO = fakeCatchUpIO{transferredTo: plan.CatchUpTarget} - } + pending.CatchUpIO = fakeCatchUpIO{transferredTo: pending.Plan.CatchUpTarget} } } @@ -518,9 +516,7 @@ func TestBlockService_ApplyAssignments_RebuildingRole_UsesCoreRecoveryPathWithou } bs.v2Recovery.OnPendingExecution = func(volumeID string, pending *rt.PendingExecution) { if volumeID == path && pending != nil && pending.Plan != nil { - if plan, ok := pending.Plan.(*engine.RecoveryPlan); ok { - pending.RebuildIO = fakeRebuildIO{achievedLSN: plan.RebuildTargetLSN} - } + pending.RebuildIO = fakeRebuildIO{achievedLSN: pending.Plan.RebuildTargetLSN} } }