From 38b50429972d080b6810faac912aac190316e1a3 Mon Sep 17 00:00:00 2001 From: pingqiu Date: Sat, 4 Apr 2026 08:11:39 -0700 Subject: [PATCH] refactor: extract command bindings and service ops from volume server Move BlockVol-backed command bindings into v2bridge and move non-BlockVol command operations into weed/server/blockcmd. This keeps dispatch and host effects in weed/server, keeps backend binding in v2bridge, and further shrinks volume_server_block.go toward a host shell while preserving current command-driven proofs. Co-Authored-By: Claude Opus 4.6 (1M context) --- weed/server/blockcmd/dispatch_test.go | 122 +++++++++++++ weed/server/blockcmd/service_ops.go | 124 +++++++++++++ weed/server/volume_server_block.go | 171 +++++------------- .../blockvol/v2bridge/command_bindings.go | 131 ++++++++++++++ 4 files changed, 424 insertions(+), 124 deletions(-) create mode 100644 weed/server/blockcmd/service_ops.go create mode 100644 weed/storage/blockvol/v2bridge/command_bindings.go diff --git a/weed/server/blockcmd/dispatch_test.go b/weed/server/blockcmd/dispatch_test.go index 9465810c7..030d788be 100644 --- a/weed/server/blockcmd/dispatch_test.go +++ b/weed/server/blockcmd/dispatch_test.go @@ -184,3 +184,125 @@ func TestDispatcher_StopsOnFirstError(t *testing.T) { t.Fatal("expected error") } } + +type fakeRecoveryCoordinator struct { + startedReplica string + startedAssigns []blockvol.BlockVolumeAssignment + catchUpCalls []struct { + volumeID string + targetLSN uint64 + } + rebuildCalls []struct { + volumeID string + targetLSN uint64 + } +} + +func (f *fakeRecoveryCoordinator) StartRecoveryTask(replicaID string, assignments []blockvol.BlockVolumeAssignment) { + f.startedReplica = replicaID + f.startedAssigns = assignments +} + +func (f *fakeRecoveryCoordinator) ExecutePendingCatchUp(volumeID string, targetLSN uint64) error { + f.catchUpCalls = append(f.catchUpCalls, struct { + volumeID string + targetLSN uint64 + }{volumeID: volumeID, targetLSN: targetLSN}) + return nil +} + +func (f *fakeRecoveryCoordinator) ExecutePendingRebuild(volumeID string, targetLSN uint64) error { + f.rebuildCalls = append(f.rebuildCalls, struct { + volumeID string + targetLSN uint64 + }{volumeID: volumeID, targetLSN: targetLSN}) + return nil +} + +type fakeProjectionReader struct { + proj engine.PublicationProjection + ok bool +} + +func (f fakeProjectionReader) Projection(volumeID string) (engine.PublicationProjection, bool) { + if !f.ok || f.proj.VolumeID != volumeID { + return engine.PublicationProjection{}, false + } + return f.proj, true +} + +type fakeSessionInvalidator struct { + reasons []string + states []engine.ReplicaState +} + +func (f *fakeSessionInvalidator) InvalidateSession(reason string, targetState engine.ReplicaState) { + f.reasons = append(f.reasons, reason) + f.states = append(f.states, targetState) +} + +func TestServiceOps_StartRecoveryTaskUsesRecoveryCoordinator(t *testing.T) { + rec := &fakeRecoveryCoordinator{} + ops := NewServiceOps(fakeOps{}, rec, nil, nil) + assign := blockvol.BlockVolumeAssignment{Path: "vol1"} + executed, err := ops.StartRecoveryTask("vol1/vs2", assign) + if err != nil { + t.Fatal(err) + } + if !executed { + t.Fatal("expected executed") + } + if rec.startedReplica != "vol1/vs2" || len(rec.startedAssigns) != 1 || rec.startedAssigns[0].Path != "vol1" { + t.Fatalf("started=%q assigns=%v", rec.startedReplica, rec.startedAssigns) + } +} + +func TestServiceOps_InvalidateSessionUsesProjectionAndSenderResolver(t *testing.T) { + s1 := &fakeSessionInvalidator{} + s2 := &fakeSessionInvalidator{} + ops := NewServiceOps( + fakeOps{}, + nil, + fakeProjectionReader{ + ok: true, + proj: engine.PublicationProjection{ + VolumeID: "vol1", + ReplicaIDs: []string{"vol1/vs2", "vol1/vs3"}, + }, + }, + func(replicaID string) SessionInvalidator { + switch replicaID { + case "vol1/vs2": + return s1 + case "vol1/vs3": + return s2 + default: + return nil + } + }, + ) + executed, err := ops.InvalidateSession("vol1", "test_reason") + if err != nil { + t.Fatal(err) + } + if !executed { + t.Fatal("expected executed") + } + if !reflect.DeepEqual(s1.reasons, []string{"test_reason"}) || !reflect.DeepEqual(s2.reasons, []string{"test_reason"}) { + t.Fatalf("reasons1=%v reasons2=%v", s1.reasons, s2.reasons) + } + if len(s1.states) != 1 || s1.states[0] != engine.StateDisconnected { + t.Fatalf("states1=%v", s1.states) + } +} + +func TestServiceOps_StartRecoveryTask_NilRecoveryIsNoop(t *testing.T) { + ops := NewServiceOps(fakeOps{}, nil, nil, nil) + executed, err := ops.StartRecoveryTask("vol1/vs2", blockvol.BlockVolumeAssignment{Path: "vol1"}) + if err != nil { + t.Fatal(err) + } + if executed { + t.Fatal("expected noop when recovery is nil") + } +} diff --git a/weed/server/blockcmd/service_ops.go b/weed/server/blockcmd/service_ops.go new file mode 100644 index 000000000..2025d3ed3 --- /dev/null +++ b/weed/server/blockcmd/service_ops.go @@ -0,0 +1,124 @@ +package blockcmd + +import ( + engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +// BackendOps owns the concrete backend-facing operations that remain separate +// from server-side command orchestration. +type BackendOps interface { + ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) + StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) + ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (executed bool, shipperConnected bool, err error) +} + +// RecoveryCoordinator is the runtime recovery surface used by command ops. +type RecoveryCoordinator interface { + StartRecoveryTask(replicaID string, assignments []blockvol.BlockVolumeAssignment) + ExecutePendingCatchUp(volumeID string, targetLSN uint64) error + ExecutePendingRebuild(volumeID string, targetLSN uint64) error +} + +// ProjectionReader provides access to current core publication state. +type ProjectionReader interface { + Projection(volumeID string) (engine.PublicationProjection, bool) +} + +// SessionInvalidator invalidates an active sender session. +type SessionInvalidator interface { + InvalidateSession(reason string, targetState engine.ReplicaState) +} + +// SenderResolver resolves one sender/session invalidator by replica ID. +type SenderResolver func(replicaID string) SessionInvalidator + +// ServiceOps composes backend ops with runtime/server dependencies to satisfy +// the dispatcher Ops interface without requiring a full BlockService pointer. +type ServiceOps struct { + backend BackendOps + recovery RecoveryCoordinator + projection ProjectionReader + senderByID SenderResolver +} + +func NewServiceOps( + backend BackendOps, + recovery RecoveryCoordinator, + projection ProjectionReader, + senderByID SenderResolver, +) *ServiceOps { + return &ServiceOps{ + backend: backend, + recovery: recovery, + projection: projection, + senderByID: senderByID, + } +} + +func (ops *ServiceOps) ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops == nil || ops.backend == nil { + return false, nil + } + return ops.backend.ApplyRole(assignment) +} + +func (ops *ServiceOps) StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops == nil || ops.backend == nil { + return false, nil + } + return ops.backend.StartReceiver(assignment) +} + +func (ops *ServiceOps) ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { + if ops == nil || ops.backend == nil { + return false, false, nil + } + return ops.backend.ConfigureShipper(volumeID, replicas) +} + +func (ops *ServiceOps) StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error) { + if ops == nil || ops.recovery == nil { + return false, nil + } + ops.recovery.StartRecoveryTask(replicaID, []blockvol.BlockVolumeAssignment{assignment}) + return true, nil +} + +func (ops *ServiceOps) InvalidateSession(volumeID, reason string) (bool, error) { + if ops == nil || ops.projection == nil || ops.senderByID == nil { + return false, nil + } + proj, ok := ops.projection.Projection(volumeID) + if !ok { + return false, nil + } + for _, replicaID := range proj.ReplicaIDs { + sender := ops.senderByID(replicaID) + if sender == nil { + continue + } + sender.InvalidateSession(reason, engine.StateDisconnected) + } + return true, nil +} + +func (ops *ServiceOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { + if ops == nil || ops.recovery == nil { + return false, nil + } + if err := ops.recovery.ExecutePendingCatchUp(volumeID, targetLSN); err != nil { + return false, err + } + return true, nil +} + +func (ops *ServiceOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { + if ops == nil || ops.recovery == nil { + return false, nil + } + if err := ops.recovery.ExecutePendingRebuild(volumeID, targetLSN); err != nil { + return false, err + } + return true, nil +} diff --git a/weed/server/volume_server_block.go b/weed/server/volume_server_block.go index db7afb806..e20ba8ba3 100644 --- a/weed/server/volume_server_block.go +++ b/weed/server/volume_server_block.go @@ -572,14 +572,42 @@ func (bs *BlockService) applyCoreCommandsWithAssignment(cmds []engine.Command, a } func (bs *BlockService) coreCommandDispatcher() *blockcmd.Dispatcher { - return blockcmd.NewDispatcher(coreCommandOps{bs: bs}, coreCommandEffects{bs: bs}) + var recovery blockcmd.RecoveryCoordinator + if bs != nil && bs.v2Recovery != nil { + recovery = bs.v2Recovery + } + var projection blockcmd.ProjectionReader + if bs != nil && bs.v2Core != nil { + projection = bs.v2Core + } + return blockcmd.NewDispatcher( + blockcmd.NewServiceOps( + coreCommandBackend{bs: bs}, + recovery, + projection, + func(replicaID string) blockcmd.SessionInvalidator { + if bs == nil || bs.v2Orchestrator == nil { + return nil + } + return bs.v2Orchestrator.Registry.Sender(replicaID) + }, + ), + coreCommandEffects{bs: bs}, + ) } -type coreCommandOps struct { +func (bs *BlockService) commandBindings() *v2bridge.CommandBindings { + if bs == nil { + return nil + } + return v2bridge.NewCommandBindings(bs.blockStore, bs.listenAddr, bs.advertisedHost) +} + +type coreCommandBackend struct { bs *BlockService } -func (ops coreCommandOps) ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) { +func (ops coreCommandBackend) ApplyRole(assignment blockvol.BlockVolumeAssignment) (bool, error) { if ops.bs == nil { return false, nil } @@ -589,7 +617,7 @@ func (ops coreCommandOps) ApplyRole(assignment blockvol.BlockVolumeAssignment) ( return true, nil } -func (ops coreCommandOps) StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) { +func (ops coreCommandBackend) StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error) { if ops.bs == nil { return false, nil } @@ -602,7 +630,7 @@ func (ops coreCommandOps) StartReceiver(assignment blockvol.BlockVolumeAssignmen return true, nil } -func (ops coreCommandOps) ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { +func (ops coreCommandBackend) ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (bool, bool, error) { if ops.bs == nil { return false, false, nil } @@ -632,52 +660,6 @@ func (ops coreCommandOps) ConfigureShipper(volumeID string, replicas []engine.Re return true, ops.bs.isPrimaryShipperConnected(volumeID), nil } -func (ops coreCommandOps) StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error) { - if ops.bs == nil || ops.bs.v2Recovery == nil { - return false, nil - } - ops.bs.v2Recovery.StartRecoveryTask(replicaID, []blockvol.BlockVolumeAssignment{assignment}) - return true, nil -} - -func (ops coreCommandOps) InvalidateSession(volumeID, reason string) (bool, error) { - if ops.bs == nil || ops.bs.v2Orchestrator == nil || ops.bs.v2Core == nil { - return false, nil - } - proj, ok := ops.bs.v2Core.Projection(volumeID) - if !ok { - return false, nil - } - for _, replicaID := range proj.ReplicaIDs { - sender := ops.bs.v2Orchestrator.Registry.Sender(replicaID) - if sender == nil { - continue - } - sender.InvalidateSession(reason, engine.StateDisconnected) - } - return true, nil -} - -func (ops coreCommandOps) StartCatchUp(volumeID string, targetLSN uint64) (bool, error) { - if ops.bs == nil || ops.bs.v2Recovery == nil { - return false, nil - } - if err := ops.bs.v2Recovery.ExecutePendingCatchUp(volumeID, targetLSN); err != nil { - return false, err - } - return true, nil -} - -func (ops coreCommandOps) StartRebuild(volumeID string, targetLSN uint64) (bool, error) { - if ops.bs == nil || ops.bs.v2Recovery == nil { - return false, nil - } - if err := ops.bs.v2Recovery.ExecutePendingRebuild(volumeID, targetLSN); err != nil { - return false, err - } - return true, nil -} - type coreCommandEffects struct { bs *BlockService } @@ -719,13 +701,10 @@ func (bs *BlockService) applyRoleAssignment(a blockvol.BlockVolumeAssignment) er if bs == nil || bs.blockStore == nil { return nil } - role := blockvol.RoleFromWire(a.Role) - ttl := blockvol.LeaseTTLFromWire(a.LeaseTtlMs) - if err := bs.blockStore.WithVolume(a.Path, func(vol *blockvol.BlockVol) error { - return vol.HandleAssignment(a.Epoch, role, ttl) - }); err != nil { + if err := bs.commandBindings().ApplyRole(a); err != nil { return err } + role := blockvol.RoleFromWire(a.Role) bs.noteRoleApplied(a.Path, role) bs.applyCoreEvent(engine.RoleApplied{ID: a.Path}) return nil @@ -801,12 +780,7 @@ func (bs *BlockService) isPrimaryShipperConnected(path string) bool { if bs == nil || bs.blockStore == nil { return false } - connected := false - _ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error { - connected = len(vol.ReplicaShipperStates()) > 0 && !vol.Status().ReplicaDegraded - return nil - }) - return connected + return bs.commandBindings().IsPrimaryShipperConnected(path) } // setupPrimaryReplication configures WAL shipping from primary to replica @@ -825,23 +799,11 @@ func (bs *BlockService) setupPrimaryReplication(path, replicaDataAddr, replicaCt return nil } - // Compute deterministic rebuild listen address. - _, _, rebuildPort := bs.ReplicationPorts(path) - host := bs.listenAddr - if idx := strings.LastIndex(host, ":"); idx >= 0 { - host = host[:idx] - } - rebuildAddr := fmt.Sprintf("%s:%d", host, rebuildPort) - - if err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error { - vol.SetReplicaAddr(replicaDataAddr, replicaCtrlAddr) - // R1-2: Start rebuild server so replicas can catch up after failover. - if err := vol.StartRebuildServer(rebuildAddr); err != nil { - glog.Warningf("block service: start rebuild server %s on %s: %v", path, rebuildAddr, err) - // Non-fatal: WAL shipping can work without rebuild server. - } - return nil - }); err != nil { + rebuildAddr, err := bs.commandBindings().ConfigurePrimaryReplication(path, []blockvol.ReplicaAddr{{ + DataAddr: replicaDataAddr, + CtrlAddr: replicaCtrlAddr, + }}) + if err != nil { glog.Warningf("block service: setup primary replication %s: %v", path, err) return err } @@ -868,21 +830,8 @@ func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockv } } - // Compute deterministic rebuild listen address. - _, _, rebuildPort := bs.ReplicationPorts(path) - host := bs.listenAddr - if idx := strings.LastIndex(host, ":"); idx >= 0 { - host = host[:idx] - } - rebuildAddr := fmt.Sprintf("%s:%d", host, rebuildPort) - - if err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error { - vol.SetReplicaAddrs(addrs) - if err := vol.StartRebuildServer(rebuildAddr); err != nil { - glog.Warningf("block service: start rebuild server %s on %s: %v", path, rebuildAddr, err) - } - return nil - }); err != nil { + rebuildAddr, err := bs.commandBindings().ConfigurePrimaryReplication(path, addrs) + if err != nil { glog.Warningf("block service: setup primary replication (multi) %s: %v", path, err) return err } @@ -893,39 +842,13 @@ func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockv // setupReplicaReceiver starts the replica WAL receiver. 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 - advHost := bs.advertisedHost - if err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error { - if advHost != "" { - if err := vol.StartReplicaReceiver(dataAddr, ctrlAddr, advHost); err != nil { - return err - } - } else { - if err := vol.StartReplicaReceiver(dataAddr, ctrlAddr); err != nil { - return err - } - } - // Read back canonical addresses from the receiver. - if vol.ReplicaReceiverAddr() != nil { - canonDataAddr = vol.ReplicaReceiverAddr().DataAddr - canonCtrlAddr = vol.ReplicaReceiverAddr().CtrlAddr - } - return nil - }); err != nil { + endpoints, err := bs.commandBindings().StartReceiver(path, dataAddr, ctrlAddr) + if err != nil { glog.Warningf("block service: setup replica receiver %s: %v", path, err) return err } - // Fallback to assignment addresses if receiver didn't report. - if canonDataAddr == "" { - canonDataAddr = dataAddr - } - if canonCtrlAddr == "" { - canonCtrlAddr = ctrlAddr - } - bs.markReceiverReady(path, canonDataAddr, canonCtrlAddr) - glog.V(0).Infof("block service: replica %s receiving on %s/%s", path, canonDataAddr, canonCtrlAddr) + bs.markReceiverReady(path, endpoints.DataAddr, endpoints.CtrlAddr) + glog.V(0).Infof("block service: replica %s receiving on %s/%s", path, endpoints.DataAddr, endpoints.CtrlAddr) return nil } diff --git a/weed/storage/blockvol/v2bridge/command_bindings.go b/weed/storage/blockvol/v2bridge/command_bindings.go new file mode 100644 index 000000000..deab6a5b5 --- /dev/null +++ b/weed/storage/blockvol/v2bridge/command_bindings.go @@ -0,0 +1,131 @@ +package v2bridge + +import ( + "fmt" + "hash/fnv" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +type VolumeAccess interface { + WithVolume(path string, fn func(*blockvol.BlockVol) error) error +} + +type ReceiverEndpoints struct { + DataAddr string + CtrlAddr string +} + +// CommandBindings owns the concrete BlockVol-backed command operations. +// It performs backend binding only and does not own server-side event semantics. +type CommandBindings struct { + volumes VolumeAccess + listenAddr string + advertisedHost string +} + +func NewCommandBindings(volumes VolumeAccess, listenAddr, advertisedHost string) *CommandBindings { + return &CommandBindings{ + volumes: volumes, + listenAddr: listenAddr, + advertisedHost: advertisedHost, + } +} + +func (b *CommandBindings) ApplyRole(a blockvol.BlockVolumeAssignment) error { + if b == nil || b.volumes == nil { + return nil + } + role := blockvol.RoleFromWire(a.Role) + ttl := blockvol.LeaseTTLFromWire(a.LeaseTtlMs) + return b.volumes.WithVolume(a.Path, func(vol *blockvol.BlockVol) error { + return vol.HandleAssignment(a.Epoch, role, ttl) + }) +} + +func (b *CommandBindings) StartReceiver(path, dataAddr, ctrlAddr string) (ReceiverEndpoints, error) { + if b == nil || b.volumes == nil { + return ReceiverEndpoints{}, nil + } + var endpoints ReceiverEndpoints + if err := b.volumes.WithVolume(path, func(vol *blockvol.BlockVol) error { + if b.advertisedHost != "" { + if err := vol.StartReplicaReceiver(dataAddr, ctrlAddr, b.advertisedHost); err != nil { + return err + } + } else { + if err := vol.StartReplicaReceiver(dataAddr, ctrlAddr); err != nil { + return err + } + } + if vol.ReplicaReceiverAddr() != nil { + endpoints.DataAddr = vol.ReplicaReceiverAddr().DataAddr + endpoints.CtrlAddr = vol.ReplicaReceiverAddr().CtrlAddr + } + return nil + }); err != nil { + return ReceiverEndpoints{}, err + } + if endpoints.DataAddr == "" { + endpoints.DataAddr = dataAddr + } + if endpoints.CtrlAddr == "" { + endpoints.CtrlAddr = ctrlAddr + } + return endpoints, nil +} + +func (b *CommandBindings) ConfigurePrimaryReplication(path string, addrs []blockvol.ReplicaAddr) (string, error) { + if b == nil || b.volumes == nil || len(addrs) == 0 { + return "", nil + } + rebuildAddr := rebuildListenAddr(path, b.listenAddr) + if err := b.volumes.WithVolume(path, func(vol *blockvol.BlockVol) error { + if len(addrs) == 1 { + vol.SetReplicaAddr(addrs[0].DataAddr, addrs[0].CtrlAddr) + } else { + vol.SetReplicaAddrs(addrs) + } + if err := vol.StartRebuildServer(rebuildAddr); err != nil { + glog.Warningf("v2bridge: start rebuild server %s on %s: %v", path, rebuildAddr, err) + } + return nil + }); err != nil { + return "", err + } + return rebuildAddr, nil +} + +func (b *CommandBindings) IsPrimaryShipperConnected(path string) bool { + if b == nil || b.volumes == nil { + return false + } + connected := false + _ = b.volumes.WithVolume(path, func(vol *blockvol.BlockVol) error { + connected = len(vol.ReplicaShipperStates()) > 0 && !vol.Status().ReplicaDegraded + return nil + }) + return connected +} + +func rebuildListenAddr(path, listenAddr string) string { + basePort := 3260 + if idx := strings.LastIndex(listenAddr, ":"); idx >= 0 { + var p int + if _, err := fmt.Sscanf(listenAddr[idx+1:], "%d", &p); err == nil && p > 0 { + basePort = p + } + } + h := fnv.New32a() + h.Write([]byte(path)) + offset := int(h.Sum32()%500) * 3 + dataPort := basePort + 1000 + offset + rebuildPort := dataPort + 2 + host := listenAddr + if idx := strings.LastIndex(host, ":"); idx >= 0 { + host = host[:idx] + } + return fmt.Sprintf("%s:%d", host, rebuildPort) +}