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

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")
+	}
+}

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
+}

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
 }
 

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)
+}