mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-01 01:40:44 +01:00
refactor: Batch 5 — recovery binding factory moved to v2bridge
New v2bridge.BuildRecoveryBundle(vol, rebuildAddr) assembles all recovery bindings (Reader + Pinner + StorageAdapter + Executor) from a real BlockVol instance in one call. block_recovery.go changes: - Removed local recoveryBundle type - buildRecoveryBundle now delegates to v2bridge.BuildRecoveryBundle inside WithVolume, returns (driver, executor, err) - Removed direct v2bridge.NewReader/NewPinner/NewExecutor construction - Removed bridge import (no longer needed) - runCatchUp/runRebuild use (driver, executor, err) directly block_recovery.go no longer knows how to construct Reader, Pinner, StorageAdapter, or Executor. It only knows: resolve volPath, ask the factory for bindings, plan, branch to legacy or core-present path. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
ded84b25e6
commit
263611004e
2 files changed
+50
-33
No files matched your search
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
bridge "github.com/seaweedfs/seaweedfs/sw-block/bridge/blockvol"
|
||||
engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
|
||||
rt "github.com/seaweedfs/seaweedfs/sw-block/engine/replication/runtime"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
@@ -235,32 +234,19 @@ func (rm *RecoveryManager) runRecovery(ctx context.Context, task *recoveryTask,
|
||||
}
|
||||
}
|
||||
|
||||
// recoveryBundle holds the concrete bindings built from a BlockVol instance
|
||||
// for one recovery execution. This is the thin host-side factory output.
|
||||
type recoveryBundle struct {
|
||||
driver *engine.RecoveryDriver
|
||||
executor *v2bridge.Executor
|
||||
}
|
||||
|
||||
// buildRecoveryBundle creates the StorageAdapter + Executor + RecoveryDriver
|
||||
// from a real BlockVol. This is the shared host-side setup for both catch-up
|
||||
// and rebuild paths.
|
||||
func (rm *RecoveryManager) buildRecoveryBundle(volPath, rebuildAddr string) (*recoveryBundle, error) {
|
||||
var sa engine.StorageAdapter
|
||||
var executor *v2bridge.Executor
|
||||
// buildRecoveryBundle creates recovery bindings from a real BlockVol via the
|
||||
// backend-binding factory. The host shell only calls WithVolume and delegates
|
||||
// Reader/Pinner/Adapter/Executor construction to v2bridge.BuildRecoveryBundle.
|
||||
func (rm *RecoveryManager) buildRecoveryBundle(volPath, rebuildAddr string) (*engine.RecoveryDriver, *v2bridge.Executor, error) {
|
||||
var bundle *v2bridge.RecoveryBundle
|
||||
if err := rm.bs.blockStore.WithVolume(volPath, func(vol *blockvol.BlockVol) error {
|
||||
reader := v2bridge.NewReader(vol)
|
||||
pinner := v2bridge.NewPinner(vol)
|
||||
sa = bridge.NewStorageAdapter(reader, pinner)
|
||||
executor = v2bridge.NewExecutor(vol, rebuildAddr)
|
||||
bundle = v2bridge.BuildRecoveryBundle(vol, rebuildAddr)
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
return &recoveryBundle{
|
||||
driver: &engine.RecoveryDriver{Orchestrator: rm.bs.v2Orchestrator, Storage: sa},
|
||||
executor: executor,
|
||||
}, nil
|
||||
driver := &engine.RecoveryDriver{Orchestrator: rm.bs.v2Orchestrator, Storage: bundle.Storage}
|
||||
return driver, bundle.Executor, nil
|
||||
}
|
||||
|
||||
func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAddr string) {
|
||||
@@ -271,7 +257,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd
|
||||
return
|
||||
}
|
||||
|
||||
bundle, err := rm.buildRecoveryBundle(volPath, rebuildAddr)
|
||||
driver, executor, err := rm.buildRecoveryBundle(volPath, rebuildAddr)
|
||||
if err != nil {
|
||||
glog.Warningf("recovery: cannot access volume %s: %v", volPath, err)
|
||||
return
|
||||
@@ -289,8 +275,6 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd
|
||||
return
|
||||
}
|
||||
|
||||
driver := bundle.driver
|
||||
|
||||
plan, err := driver.PlanRecovery(replicaID, replicaFlushedLSN)
|
||||
if err != nil {
|
||||
glog.Warningf("recovery: plan failed for %s: %v", replicaID, err)
|
||||
@@ -299,7 +283,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd
|
||||
switch plan.Outcome {
|
||||
case engine.OutcomeCatchUp:
|
||||
if bs.v2Core == nil {
|
||||
rm.executeLegacyCatchUp(ctx, volPath, replicaID, driver, plan, bundle.executor)
|
||||
rm.executeLegacyCatchUp(ctx, volPath, replicaID, driver, plan, executor)
|
||||
return
|
||||
}
|
||||
rm.coord.Store(volPath, &rt.PendingExecution{
|
||||
@@ -308,7 +292,7 @@ func (rm *RecoveryManager) runCatchUp(ctx context.Context, replicaID, rebuildAdd
|
||||
CatchUpTarget: plan.CatchUpTarget,
|
||||
Driver: driver,
|
||||
Plan: plan,
|
||||
CatchUpIO: bundle.executor,
|
||||
CatchUpIO: executor,
|
||||
})
|
||||
bs.applyCoreEvent(engine.CatchUpPlanned{ID: volPath, TargetLSN: plan.CatchUpTarget})
|
||||
if rm.coord.Has(volPath) {
|
||||
@@ -338,7 +322,7 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd
|
||||
return
|
||||
}
|
||||
|
||||
bundle, err := rm.buildRecoveryBundle(volPath, rebuildAddr)
|
||||
driver, executor, err := rm.buildRecoveryBundle(volPath, rebuildAddr)
|
||||
if err != nil {
|
||||
glog.Warningf("recovery: cannot access volume %s: %v", volPath, err)
|
||||
return
|
||||
@@ -348,15 +332,13 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd
|
||||
return
|
||||
}
|
||||
|
||||
driver := bundle.driver
|
||||
|
||||
plan, err := driver.PlanRebuild(replicaID)
|
||||
if err != nil {
|
||||
glog.Warningf("recovery: rebuild plan failed for %s: %v", replicaID, err)
|
||||
return
|
||||
}
|
||||
if bs.v2Core == nil {
|
||||
rm.executeLegacyRebuild(ctx, volPath, replicaID, driver, plan, bundle.executor)
|
||||
rm.executeLegacyRebuild(ctx, volPath, replicaID, driver, plan, executor)
|
||||
return
|
||||
}
|
||||
pe := &rt.PendingExecution{
|
||||
@@ -365,7 +347,7 @@ func (rm *RecoveryManager) runRebuild(ctx context.Context, replicaID, rebuildAdd
|
||||
RebuildTargetLSN: plan.RebuildTargetLSN,
|
||||
Driver: driver,
|
||||
Plan: plan,
|
||||
RebuildIO: bundle.executor,
|
||||
RebuildIO: executor,
|
||||
}
|
||||
rm.coord.Store(volPath, pe)
|
||||
if rm.OnPendingExecution != nil {
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
package v2bridge
|
||||
|
||||
import (
|
||||
bridge "github.com/seaweedfs/seaweedfs/sw-block/bridge/blockvol"
|
||||
engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
|
||||
)
|
||||
|
||||
// RecoveryBundle holds the concrete bindings built from a BlockVol instance
|
||||
// for one recovery execution. The host calls BuildRecoveryBundle once per
|
||||
// recovery attempt; the bundle is then consumed by the runtime coordinator.
|
||||
type RecoveryBundle struct {
|
||||
// Storage is the engine-facing storage adapter for recovery planning.
|
||||
Storage engine.StorageAdapter
|
||||
|
||||
// Executor implements both CatchUpIO and RebuildIO for this volume.
|
||||
Executor *Executor
|
||||
}
|
||||
|
||||
// BuildRecoveryBundle assembles all recovery bindings from a real BlockVol.
|
||||
// This is the backend-binding factory — it knows how to create Reader,
|
||||
// Pinner, StorageAdapter, and Executor from a concrete BlockVol instance.
|
||||
//
|
||||
// The host shell (block_recovery.go) calls this inside WithVolume and
|
||||
// does not need to know about Reader/Pinner/StorageAdapter construction.
|
||||
func BuildRecoveryBundle(vol *blockvol.BlockVol, rebuildAddr string) *RecoveryBundle {
|
||||
reader := NewReader(vol)
|
||||
pinner := NewPinner(vol)
|
||||
sa := bridge.NewStorageAdapter(reader, pinner)
|
||||
executor := NewExecutor(vol, rebuildAddr)
|
||||
return &RecoveryBundle{
|
||||
Storage: sa,
|
||||
Executor: executor,
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user