seaweedfs/weed/server/volume_server_block.go

package weed_server

import (
	"fmt"
	"hash/fnv"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"

	engine "github.com/seaweedfs/seaweedfs/sw-block/engine/replication"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/storage"
	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol"
	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/iscsi"
	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/nvme"
	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/v2bridge"
)

// volReplState tracks active replication addresses per volume.
type volReplState struct {
	replicaDataAddr string
	replicaCtrlAddr string
	// allReplicas stores the full replica set for multi-replica idempotence.
	allReplicas       []blockvol.ReplicaAddr
	roleApplied       bool
	receiverReady     bool
	shipperConfigured bool
	replicaEligible   bool
	publishHealthy    bool
}

// BlockReadinessSnapshot names the assignment-to-publication closure at the
// BlockService boundary. These flags are owned by the service/adapter layer,
// not by blockvol's local storage mechanics.
//
// Important:
// PublishHealthy here is still an adapter-local publication bit used by current
// `weed/server` surfaces. It is NOT the semantic owner for Phase 14 core
// publication health; that owner is `engine.PublicationView`.
type BlockReadinessSnapshot struct {
	RoleApplied       bool
	ReceiverReady     bool
	ShipperConfigured bool
	ShipperConnected  bool
	ReplicaEligible   bool
	PublishHealthy    bool
}

// NVMeConfig holds NVMe/TCP target configuration passed from CLI flags.
type NVMeConfig struct {
	Enabled     bool
	ListenAddr  string
	Portal      string // reserved for heartbeat/CSI integration (CP10-2)
	NQNPrefix   string
	MaxIOQueues int
}

// BlockService manages block volumes and the iSCSI/NVMe target servers.
type BlockService struct {
	blockStore     *storage.BlockVolumeStore
	targetServer   *iscsi.TargetServer
	nvmeServer     *nvme.Server
	iqnPrefix      string
	nqnPrefix      string
	blockDir       string
	listenAddr     string
	nvmeListenAddr string

	// Replication state (CP6-3).
	replMu     sync.RWMutex
	replStates map[string]*volReplState // keyed by volume path

	// V2 engine bridge (Phase 08 P1).
	v2Bridge       *v2bridge.ControlBridge
	v2Orchestrator *engine.RecoveryOrchestrator
	v2Core         *engine.CoreEngine
	v2Recovery     *RecoveryManager
	coreProjMu     sync.RWMutex
	coreProj       map[string]engine.PublicationProjection
	coreExecMu     sync.RWMutex
	coreExec       map[string][]string

	// P3: last-applied assignment per volume path for idempotence.
	lastAssignMu sync.RWMutex
	lastAssign   map[string]lastAppliedAssignment
	// localServerID: stable identity for this volume server.
	// May be an opaque string (from -id flag) or ip:port (default fallback).
	// NOT guaranteed to be a routable address — do not use for transport endpoints.
	localServerID string

	// advertisedHost: routable host for this volume server (from -ip flag or auto-detected).
	// Used by CP13-2 to canonicalize wildcard-bind replica listener addresses to
	// routable host:port. This is the -ip value (IP or resolvable hostname),
	// never an opaque server identity from -id.
	advertisedHost string
}

// V2Orchestrator returns the V2 engine orchestrator for inspection/testing.
func (bs *BlockService) V2Orchestrator() *engine.RecoveryOrchestrator {
	return bs.v2Orchestrator
}

// V2Core returns the explicit Phase 14/15 core shell if wired.
func (bs *BlockService) V2Core() *engine.CoreEngine {
	return bs.v2Core
}

// CoreProjection returns the latest adapter-cached projection emitted by the
// explicit V2 core on the narrow live path.
func (bs *BlockService) CoreProjection(path string) (engine.PublicationProjection, bool) {
	bs.coreProjMu.RLock()
	defer bs.coreProjMu.RUnlock()
	if bs.coreProj == nil {
		return engine.PublicationProjection{}, false
	}
	proj, ok := bs.coreProj[path]
	return proj, ok
}

// ExecutedCoreCommands returns the bounded list of core commands executed on the
// current integrated path for one volume. Intended for focused runtime-ownership
// proofs in Phase 16.
func (bs *BlockService) ExecutedCoreCommands(path string) []string {
	bs.coreExecMu.RLock()
	defer bs.coreExecMu.RUnlock()
	if bs.coreExec == nil {
		return nil
	}
	cmds := bs.coreExec[path]
	out := make([]string, len(cmds))
	copy(out, cmds)
	return out
}

// CoreProjectionMismatches reports fields that should already agree on the
// narrow Phase 15A path but do not. It intentionally excludes adapter-local
// `PublishHealthy`, which is not yet rebound to the core publication owner.
func (bs *BlockService) CoreProjectionMismatches(path string) []string {
	proj, ok := bs.CoreProjection(path)
	if !ok {
		return []string{"missing_core_projection"}
	}
	readiness := bs.ReadinessSnapshot(path)
	var mismatches []string
	if readiness.RoleApplied != proj.Readiness.RoleApplied {
		mismatches = append(mismatches, "role_applied")
	}
	if readiness.ReceiverReady != proj.Readiness.ReceiverReady {
		mismatches = append(mismatches, "receiver_ready")
	}
	if readiness.ShipperConfigured != proj.Readiness.ShipperConfigured {
		mismatches = append(mismatches, "shipper_configured")
	}
	if readiness.ShipperConnected != proj.Readiness.ShipperConnected {
		mismatches = append(mismatches, "shipper_connected")
	}
	return mismatches
}

// SetServerID sets the stable server identity for V2 control semantics.
// This may be an opaque string (from -id flag) — not guaranteed routable.
func (bs *BlockService) SetServerID(id string) {
	bs.localServerID = id
}

// SetAdvertisedHost sets the routable host for replica endpoint canonicalization.
// This is the -ip flag value (IP address or resolvable hostname), never an
// opaque server identity from -id. Called at startup from volume.go.
func (bs *BlockService) SetAdvertisedHost(host string) {
	bs.advertisedHost = host
}

// WireStateChangeNotify sets up shipper state change callbacks on all
// registered volumes so that degradation/recovery triggers an immediate
// heartbeat via the provided channel. Non-blocking send (buffered chan 1).
func (bs *BlockService) WireStateChangeNotify(ch chan bool) {
	bs.blockStore.IterateBlockVolumes(func(path string, vol *blockvol.BlockVol) {
		vol.SetOnShipperStateChange(func(from, to blockvol.ReplicaState) {
			select {
			case ch <- true:
			default: // already pending
			}
		})
	})
}

// StartBlockService scans blockDir for .blk files, opens them as block volumes,
// registers them with iSCSI and optionally NVMe target servers, and starts listening.
// Returns nil if blockDir is empty (feature disabled).
func StartBlockService(listenAddr, blockDir, iqnPrefix, portalAddr string, nvmeCfg NVMeConfig) *BlockService {
	if blockDir == "" {
		return nil
	}

	if iqnPrefix == "" {
		iqnPrefix = "iqn.2024-01.com.seaweedfs:vol."
	}
	nqnPrefix := nvmeCfg.NQNPrefix
	if nqnPrefix == "" {
		nqnPrefix = "nqn.2024-01.com.seaweedfs:vol."
	}

	bs := &BlockService{
		blockStore:     storage.NewBlockVolumeStore(),
		iqnPrefix:      iqnPrefix,
		nqnPrefix:      nqnPrefix,
		blockDir:       blockDir,
		listenAddr:     listenAddr,
		nvmeListenAddr: nvmeCfg.ListenAddr,
		v2Bridge:       v2bridge.NewControlBridge(),
		v2Orchestrator: engine.NewRecoveryOrchestrator(),
		v2Core:         engine.NewCoreEngine(),
		localServerID:  listenAddr, // INTERIM: transport-shaped, see field doc
		coreProj:       make(map[string]engine.PublicationProjection),
	}
	bs.v2Recovery = NewRecoveryManager(bs)

	// iSCSI target setup.
	logger := log.New(os.Stderr, "iscsi: ", log.LstdFlags)

	config := iscsi.DefaultTargetConfig()
	config.TargetName = iqnPrefix + "default"

	bs.targetServer = iscsi.NewTargetServer(listenAddr, config, logger)
	if portalAddr != "" {
		bs.targetServer.SetPortalAddr(portalAddr)
	}

	// NVMe/TCP target setup (optional).
	if nvmeCfg.Enabled {
		maxQ := uint16(4)
		if nvmeCfg.MaxIOQueues >= 1 && nvmeCfg.MaxIOQueues <= 128 {
			maxQ = uint16(nvmeCfg.MaxIOQueues)
		}
		bs.nvmeServer = nvme.NewServer(nvme.Config{
			ListenAddr:  nvmeCfg.ListenAddr,
			NQNPrefix:   nqnPrefix,
			MaxIOQueues: maxQ,
			Enabled:     true,
		})
	}

	// Scan blockDir for .blk files.
	entries, err := os.ReadDir(blockDir)
	if err != nil {
		glog.Warningf("block service: cannot read dir %s: %v", blockDir, err)
		return bs
	}

	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".blk") {
			continue
		}
		path := filepath.Join(blockDir, entry.Name())
		vol, err := bs.blockStore.AddBlockVolume(path, "")
		if err != nil {
			// Auto-initialize raw files (e.g. created via truncate).
			info, serr := entry.Info()
			if serr == nil && info.Size() > 0 {
				glog.V(0).Infof("block service: auto-creating blockvol %s (%d bytes)", path, info.Size())
				os.Remove(path) // remove raw file so CreateBlockVol can use O_EXCL
				created, cerr := blockvol.CreateBlockVol(path, blockvol.CreateOptions{
					VolumeSize: uint64(info.Size()),
				})
				if cerr != nil {
					glog.Warningf("block service: auto-create %s: %v", path, cerr)
					continue
				}
				created.Close()
				vol, err = bs.blockStore.AddBlockVolume(path, "")
				if err != nil {
					glog.Warningf("block service: skip %s after auto-create: %v", path, err)
					continue
				}
			} else {
				glog.Warningf("block service: skip %s: %v", path, err)
				continue
			}
		}

		name := strings.TrimSuffix(entry.Name(), ".blk")
		bs.registerVolume(vol, name)
	}

	// Start iSCSI target in background.
	go func() {
		if err := bs.targetServer.ListenAndServe(); err != nil {
			glog.Warningf("block service: iSCSI target stopped: %v", err)
		}
	}()
	glog.V(0).Infof("block service: iSCSI target started on %s", listenAddr)

	// Start NVMe/TCP target in background (if enabled).
	if bs.nvmeServer != nil {
		if err := bs.nvmeServer.ListenAndServe(); err != nil {
			glog.Warningf("block service: NVMe/TCP target failed to start: %v (iSCSI continues)", err)
			bs.nvmeServer = nil // disable NVMe, iSCSI continues
		} else {
			glog.V(0).Infof("block service: NVMe/TCP target started on %s", nvmeCfg.ListenAddr)
		}
	}

	return bs
}

// registerVolume adds a volume to both iSCSI and NVMe targets.
func (bs *BlockService) registerVolume(vol *blockvol.BlockVol, name string) {
	iqn := bs.iqnPrefix + blockvol.SanitizeIQN(name)
	adapter := blockvol.NewBlockVolAdapter(vol)
	bs.targetServer.AddVolume(iqn, adapter)

	if bs.nvmeServer != nil {
		nqn := blockvol.BuildNQN(bs.nqnPrefix, name)
		nvmeAdapter := nvme.NewNVMeAdapter(vol)
		bs.nvmeServer.AddVolume(nqn, nvmeAdapter, nvmeAdapter.DeviceNGUID())
	}

	glog.V(0).Infof("block service: registered %s", name)
}

// Store returns the underlying BlockVolumeStore.
func (bs *BlockService) Store() *storage.BlockVolumeStore {
	return bs.blockStore
}

// BlockDir returns the block volume data directory.
func (bs *BlockService) BlockDir() string {
	return bs.blockDir
}

// ListenAddr returns the iSCSI target listen address.
func (bs *BlockService) ListenAddr() string {
	return bs.listenAddr
}

// NvmeListenAddr returns the configured NVMe/TCP target listen address, or empty if NVMe is disabled.
func (bs *BlockService) NvmeListenAddr() string {
	if bs.nvmeServer != nil {
		return bs.nvmeListenAddr
	}
	return ""
}

// NQN returns the NVMe subsystem NQN for a volume name.
func (bs *BlockService) NQN(name string) string {
	return blockvol.BuildNQN(bs.nqnPrefix, name)
}

// CreateBlockVol creates a new .blk file, registers it with BlockVolumeStore
// and iSCSI TargetServer. Returns path, IQN, iSCSI addr.
// Idempotent: if volume already exists with same or larger size, returns existing info.
func (bs *BlockService) CreateBlockVol(name string, sizeBytes uint64, diskType string, durabilityMode string) (path, iqn, iscsiAddr string, err error) {
	sanitized := blockvol.SanitizeFilename(name)
	path = filepath.Join(bs.blockDir, sanitized+".blk")
	iqn = bs.iqnPrefix + blockvol.SanitizeIQN(name)
	iscsiAddr = bs.listenAddr

	// Check if already registered.
	if vol, ok := bs.blockStore.GetBlockVolume(path); ok {
		info := vol.Info()
		if info.VolumeSize < sizeBytes {
			return "", "", "", fmt.Errorf("block volume %q exists with size %d (requested %d)",
				name, info.VolumeSize, sizeBytes)
		}
		// Re-add to targets in case they were cleared (crash recovery).
		// AddVolume is idempotent — no-op if already registered.
		adapter := blockvol.NewBlockVolAdapter(vol)
		bs.targetServer.AddVolume(iqn, adapter)
		if bs.nvmeServer != nil {
			nqn := blockvol.BuildNQN(bs.nqnPrefix, name)
			nvmeAdapter := nvme.NewNVMeAdapter(vol)
			bs.nvmeServer.AddVolume(nqn, nvmeAdapter, nvmeAdapter.DeviceNGUID())
		}
		return path, iqn, iscsiAddr, nil
	}

	// F2: VS-side validation — reject invalid mode strings (defense-in-depth).
	var durMode blockvol.DurabilityMode
	if durabilityMode != "" {
		var perr error
		durMode, perr = blockvol.ParseDurabilityMode(durabilityMode)
		if perr != nil {
			return "", "", "", fmt.Errorf("invalid durability mode: %w", perr)
		}
	}

	// Create the .blk file.
	if err := os.MkdirAll(bs.blockDir, 0755); err != nil {
		return "", "", "", fmt.Errorf("create block dir: %w", err)
	}
	created, err := blockvol.CreateBlockVol(path, blockvol.CreateOptions{
		VolumeSize:     sizeBytes,
		DurabilityMode: durMode,
	})
	if err != nil {
		return "", "", "", fmt.Errorf("create block volume: %w", err)
	}
	created.Close()

	// Open and register.
	vol, err := bs.blockStore.AddBlockVolume(path, diskType)
	if err != nil {
		os.Remove(path)
		return "", "", "", fmt.Errorf("register block volume: %w", err)
	}

	adapter := blockvol.NewBlockVolAdapter(vol)
	bs.targetServer.AddVolume(iqn, adapter)

	if bs.nvmeServer != nil {
		nqn := blockvol.BuildNQN(bs.nqnPrefix, name)
		nvmeAdapter := nvme.NewNVMeAdapter(vol)
		bs.nvmeServer.AddVolume(nqn, nvmeAdapter, nvmeAdapter.DeviceNGUID())
	}

	glog.V(0).Infof("block service: created %s as %s (%d bytes)", path, iqn, sizeBytes)
	return path, iqn, iscsiAddr, nil
}

// DeleteBlockVol disconnects iSCSI sessions, closes the volume, and removes the .blk file.
// Idempotent: returns nil if volume not found.
func (bs *BlockService) DeleteBlockVol(name string) error {
	sanitized := blockvol.SanitizeFilename(name)
	path := filepath.Join(bs.blockDir, sanitized+".blk")
	iqn := bs.iqnPrefix + blockvol.SanitizeIQN(name)

	// Disconnect active iSCSI sessions and remove target entry.
	if bs.targetServer != nil {
		bs.targetServer.DisconnectVolume(iqn)
	}

	// Remove from NVMe target.
	if bs.nvmeServer != nil {
		nqn := blockvol.BuildNQN(bs.nqnPrefix, name)
		bs.nvmeServer.RemoveVolume(nqn)
	}

	// Close and unregister.
	if err := bs.blockStore.RemoveBlockVolume(path); err != nil {
		// Not found is OK (idempotent).
		if !strings.Contains(err.Error(), "not found") {
			return fmt.Errorf("remove block volume: %w", err)
		}
	}

	// Remove the .blk file and any snapshot files.
	os.Remove(path)
	matches, _ := filepath.Glob(path + ".snap.*")
	for _, m := range matches {
		os.Remove(m)
	}

	glog.V(0).Infof("block service: deleted %s", path)
	return nil
}

// ProcessAssignments applies assignments from master, including replication setup.
// V2 bridge: also delivers each assignment to the V2 engine for recovery ownership.
func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAssignment) {
	_ = bs.ApplyAssignments(assignments)
}

// ApplyAssignments applies assignments through the single authoritative
// BlockService lifecycle: role apply, replication wiring, and publication
// readiness bookkeeping. Returns per-assignment errors parallel to the input.
func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssignment) []error {
	errs := make([]error, len(assignments))

	// V2 bridge: convert and deliver to engine orchestrator (Phase 08 P1).
	// P3: skip V2 processing for repeated unchanged assignments.
	// P4: RecoveryManager starts/cancels recovery goroutines based on results.
	if bs.v2Bridge != nil && bs.v2Orchestrator != nil {
		for _, a := range assignments {
			// P3 idempotence: skip V2 processing if this assignment is
			// materially unchanged from the last one applied for this path.
			if bs.isAssignmentUnchanged(a) {
				continue
			}
			bs.recordAppliedAssignment(a)

			intent := bs.v2Bridge.ConvertAssignment(a, bs.localServerID)
			result := bs.v2Orchestrator.ProcessAssignment(intent)
			glog.V(1).Infof("v2bridge: assignment %s epoch=%d → added=%d removed=%d sessions=%d",
				a.Path, a.Epoch, len(result.Added), len(result.Removed),
				len(result.SessionsCreated)+len(result.SessionsSuperseded))

			// P4: drive live recovery execution based on engine result.
			if bs.v2Recovery != nil && (len(result.SessionsCreated) > 0 || len(result.SessionsSuperseded) > 0 || len(result.Removed) > 0) {
				bs.v2Recovery.HandleAssignmentResult(result, assignments)
			}
		}
	}

	// V1 processing (requires blockStore).
	if bs.blockStore == nil {
		return errs
	}
	for i, a := range assignments {
		role := blockvol.RoleFromWire(a.Role)
		bs.recordAppliedAssignment(a)
		if err := bs.applyCoreAssignmentEvent(a); err != nil {
			errs[i] = err
			glog.Warningf("block service: assignment %s epoch=%d role=%s: %v", a.Path, a.Epoch, role, err)
			continue
		}

		// 2. Replication setup based on role + addresses.
		switch role {
		case blockvol.RolePrimary:
		case blockvol.RoleReplica:
		case blockvol.RoleRebuilding:
			if a.RebuildAddr != "" {
				bs.startRebuild(a.Path, a.RebuildAddr, a.Epoch)
			}
		}
	}
	return errs
}

func (bs *BlockService) applyCoreAssignmentEvent(a blockvol.BlockVolumeAssignment) error {
	if bs == nil || bs.v2Core == nil {
		return bs.applyRoleAssignment(a)
	}
	ev, ok := bs.coreAssignmentEvent(a)
	if !ok {
		return nil
	}
	result := bs.v2Core.ApplyEvent(ev)
	return bs.applyCoreCommandsWithAssignment(result.Commands, &a)
}

func (bs *BlockService) applyCoreEvent(ev engine.Event) {
	if bs == nil || bs.v2Core == nil {
		return
	}
	result := bs.v2Core.ApplyEvent(ev)
	bs.applyCoreCommands(result.Commands)
}

func (bs *BlockService) applyCoreCommands(cmds []engine.Command) {
	_ = bs.applyCoreCommandsWithAssignment(cmds, nil)
}

func (bs *BlockService) applyCoreCommandsWithAssignment(cmds []engine.Command, assignment *blockvol.BlockVolumeAssignment) error {
	for _, cmd := range cmds {
		switch v := cmd.(type) {
		case engine.ApplyRoleCommand:
			if err := bs.executeApplyRoleCommand(v, assignment); err != nil {
				return err
			}
		case engine.StartReceiverCommand:
			if err := bs.executeStartReceiverCommand(v, assignment); err != nil {
				return err
			}
		case engine.ConfigureShipperCommand:
			if err := bs.executeConfigureShipperCommand(v); err != nil {
				return err
			}
		case engine.InvalidateSessionCommand:
			if err := bs.executeInvalidateSessionCommand(v); err != nil {
				return err
			}
		case engine.StartCatchUpCommand:
			if err := bs.executeStartCatchUpCommand(v); err != nil {
				return err
			}
		case engine.StartRebuildCommand:
			if err := bs.executeStartRebuildCommand(v); err != nil {
				return err
			}
		case engine.PublishProjectionCommand:
			proj := v.Projection
			if latest, ok := bs.V2Core().Projection(v.VolumeID); ok {
				proj = latest
			}
			bs.coreProjMu.Lock()
			if bs.coreProj == nil {
				bs.coreProj = make(map[string]engine.PublicationProjection)
			}
			bs.coreProj[v.VolumeID] = proj
			bs.coreProjMu.Unlock()
		}
	}
	return nil
}

func (bs *BlockService) executeApplyRoleCommand(cmd engine.ApplyRoleCommand, assignment *blockvol.BlockVolumeAssignment) error {
	if assignment == nil {
		return nil
	}
	if assignment.Path != cmd.VolumeID {
		return fmt.Errorf("block service: core apply_role path mismatch %q != %q", assignment.Path, cmd.VolumeID)
	}
	if err := bs.applyRoleAssignment(*assignment); err != nil {
		return err
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "apply_role")
	return nil
}

func (bs *BlockService) executeStartReceiverCommand(cmd engine.StartReceiverCommand, assignment *blockvol.BlockVolumeAssignment) error {
	if assignment == nil {
		return nil
	}
	if assignment.Path != cmd.VolumeID {
		return fmt.Errorf("block service: core start_receiver path mismatch %q != %q", assignment.Path, cmd.VolumeID)
	}
	if assignment.ReplicaDataAddr == "" || assignment.ReplicaCtrlAddr == "" {
		return nil
	}
	if err := bs.setupReplicaReceiver(assignment.Path, assignment.ReplicaDataAddr, assignment.ReplicaCtrlAddr); err != nil {
		return err
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "start_receiver")
	bs.applyCoreEvent(engine.ReceiverReadyObserved{ID: cmd.VolumeID})
	return nil
}

func (bs *BlockService) executeConfigureShipperCommand(cmd engine.ConfigureShipperCommand) error {
	addrs := make([]blockvol.ReplicaAddr, 0, len(cmd.Replicas))
	for _, replica := range cmd.Replicas {
		if replica.Endpoint.DataAddr == "" || replica.Endpoint.CtrlAddr == "" {
			continue
		}
		addrs = append(addrs, blockvol.ReplicaAddr{
			ServerID: replica.ReplicaID,
			DataAddr: replica.Endpoint.DataAddr,
			CtrlAddr: replica.Endpoint.CtrlAddr,
		})
	}
	if len(addrs) == 0 {
		return nil
	}
	if len(addrs) == 1 {
		if err := bs.setupPrimaryReplication(cmd.VolumeID, addrs[0].DataAddr, addrs[0].CtrlAddr); err != nil {
			return err
		}
	} else {
		if err := bs.setupPrimaryReplicationMulti(cmd.VolumeID, addrs); err != nil {
			return err
		}
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "configure_shipper")
	bs.applyCoreEvent(engine.ShipperConfiguredObserved{ID: cmd.VolumeID})
	if bs.isPrimaryShipperConnected(cmd.VolumeID) {
		bs.applyCoreEvent(engine.ShipperConnectedObserved{ID: cmd.VolumeID})
	}
	return nil
}

func (bs *BlockService) executeInvalidateSessionCommand(cmd engine.InvalidateSessionCommand) error {
	if bs == nil || bs.v2Orchestrator == nil || bs.v2Core == nil {
		return nil
	}
	proj, ok := bs.v2Core.Projection(cmd.VolumeID)
	if !ok {
		return nil
	}
	for _, replicaID := range proj.ReplicaIDs {
		sender := bs.v2Orchestrator.Registry.Sender(replicaID)
		if sender == nil {
			continue
		}
		sender.InvalidateSession(cmd.Reason, engine.StateDisconnected)
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "invalidate_session")
	return nil
}

func (bs *BlockService) executeStartCatchUpCommand(cmd engine.StartCatchUpCommand) error {
	if bs == nil || bs.v2Recovery == nil {
		return nil
	}
	if err := bs.v2Recovery.ExecutePendingCatchUp(cmd.VolumeID, cmd.TargetLSN); err != nil {
		return err
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "start_catchup")
	return nil
}

func (bs *BlockService) executeStartRebuildCommand(cmd engine.StartRebuildCommand) error {
	if bs == nil || bs.v2Recovery == nil {
		return nil
	}
	if err := bs.v2Recovery.ExecutePendingRebuild(cmd.VolumeID, cmd.TargetLSN); err != nil {
		return err
	}
	bs.recordExecutedCoreCommand(cmd.VolumeID, "start_rebuild")
	return nil
}

func (bs *BlockService) applyRoleAssignment(a blockvol.BlockVolumeAssignment) error {
	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 {
		return err
	}
	bs.noteRoleApplied(a.Path, role)
	bs.applyCoreEvent(engine.RoleApplied{ID: a.Path})
	return nil
}

func (bs *BlockService) recordExecutedCoreCommand(path, name string) {
	bs.coreExecMu.Lock()
	defer bs.coreExecMu.Unlock()
	if bs.coreExec == nil {
		bs.coreExec = make(map[string][]string)
	}
	bs.coreExec[path] = append(bs.coreExec[path], name)
}

func (bs *BlockService) coreAssignmentEvent(a blockvol.BlockVolumeAssignment) (engine.AssignmentDelivered, bool) {
	role := blockvol.RoleFromWire(a.Role)
	ev := engine.AssignmentDelivered{
		ID:    a.Path,
		Epoch: a.Epoch,
	}
	switch role {
	case blockvol.RolePrimary:
		ev.Role = engine.RolePrimary
		if len(a.ReplicaAddrs) > 0 {
			ev.Replicas = make([]engine.ReplicaAssignment, 0, len(a.ReplicaAddrs))
			for _, ra := range a.ReplicaAddrs {
				if ra.ServerID == "" {
					continue
				}
				ev.Replicas = append(ev.Replicas, engine.ReplicaAssignment{
					ReplicaID: fmt.Sprintf("%s/%s", a.Path, ra.ServerID),
					Endpoint: engine.Endpoint{
						DataAddr: ra.DataAddr,
						CtrlAddr: ra.CtrlAddr,
					},
				})
			}
		} else if a.ReplicaServerID != "" && a.ReplicaDataAddr != "" {
			ev.Replicas = []engine.ReplicaAssignment{{
				ReplicaID: fmt.Sprintf("%s/%s", a.Path, a.ReplicaServerID),
				Endpoint: engine.Endpoint{
					DataAddr: a.ReplicaDataAddr,
					CtrlAddr: a.ReplicaCtrlAddr,
				},
			}}
		}
		return ev, true
	case blockvol.RoleReplica:
		ev.Role = engine.RoleReplica
		ev.Replicas = []engine.ReplicaAssignment{{
			ReplicaID: fmt.Sprintf("%s/%s", a.Path, bs.localServerID),
			Endpoint: engine.Endpoint{
				DataAddr: a.ReplicaDataAddr,
				CtrlAddr: a.ReplicaCtrlAddr,
			},
		}}
		return ev, true
	default:
		return engine.AssignmentDelivered{}, false
	}
}

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
}

// setupPrimaryReplication configures WAL shipping from primary to replica
// and starts the rebuild server (R1-2).
func (bs *BlockService) setupPrimaryReplication(path, replicaDataAddr, replicaCtrlAddr string) error {
	// P3 idempotence: skip if replica state is unchanged.
	bs.replMu.RLock()
	existing := bs.replStates[path]
	bs.replMu.RUnlock()
	if existing != nil && existing.replicaDataAddr == replicaDataAddr && existing.replicaCtrlAddr == replicaCtrlAddr {
		// Unchanged repeated assignment — idempotent, no side effects.
		bs.markPrimaryTransportConfigured(path, []blockvol.ReplicaAddr{{
			DataAddr: replicaDataAddr,
			CtrlAddr: replicaCtrlAddr,
		}})
		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 {
		glog.Warningf("block service: setup primary replication %s: %v", path, err)
		return err
	}
	bs.markPrimaryTransportConfigured(path, []blockvol.ReplicaAddr{{
		DataAddr: replicaDataAddr,
		CtrlAddr: replicaCtrlAddr,
	}})
	glog.V(0).Infof("block service: primary %s shipping WAL to %s/%s (rebuild=%s)", path, replicaDataAddr, replicaCtrlAddr, rebuildAddr)
	return nil
}

// setupPrimaryReplicationMulti configures WAL shipping from primary to N replicas
// using SetReplicaAddrs (CP8-2: multi-replica support).
func (bs *BlockService) setupPrimaryReplicationMulti(path string, addrs []blockvol.ReplicaAddr) error {
	// P3 idempotence: skip if ALL replica addresses unchanged.
	// Compare full replica set, not just the first entry.
	if len(addrs) > 0 {
		bs.replMu.RLock()
		existing := bs.replStates[path]
		bs.replMu.RUnlock()
		if existing != nil && bs.multiReplicaUnchanged(path, addrs) {
			bs.markPrimaryTransportConfigured(path, addrs)
			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.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 {
		glog.Warningf("block service: setup primary replication (multi) %s: %v", path, err)
		return err
	}
	bs.markPrimaryTransportConfigured(path, addrs)
	glog.V(0).Infof("block service: primary %s shipping WAL to %d replicas (rebuild=%s)", path, len(addrs), rebuildAddr)
	return nil
}

// 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 {
		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)
	return nil
}

// startRebuild starts a rebuild in the background.
// R2-F7: Rebuild success/failure is logged but not reported back to master.
// Future work: VS could report rebuild completion via heartbeat so master
// can update registry state (e.g., promote from Rebuilding to Replica).
func (bs *BlockService) startRebuild(path, rebuildAddr string, epoch uint64) {
	go func() {
		vol, ok := bs.blockStore.GetBlockVolume(path)
		if !ok {
			glog.Warningf("block service: rebuild %s: volume not found", path)
			return
		}
		if err := blockvol.StartRebuild(vol, rebuildAddr, 0, epoch); err != nil {
			glog.Warningf("block service: rebuild %s from %s: %v", path, rebuildAddr, err)
			return
		}
		glog.V(0).Infof("block service: rebuild %s from %s completed", path, rebuildAddr)
	}()
}

// SnapshotBlockVol creates a snapshot on the named volume.
func (bs *BlockService) SnapshotBlockVol(name string, snapID uint32) (createdAt int64, sizeBytes uint64, err error) {
	path := bs.volumePath(name)
	err = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		if serr := vol.CreateSnapshot(snapID); serr != nil {
			return serr
		}
		// Find the snapshot we just created for its metadata.
		for _, s := range vol.ListSnapshots() {
			if s.ID == snapID {
				createdAt = s.CreatedAt.Unix()
				sizeBytes = vol.Info().VolumeSize
				return nil
			}
		}
		return fmt.Errorf("snapshot %d created but not found in list", snapID)
	})
	return
}

// DeleteBlockSnapshot deletes a snapshot on the named volume.
// Idempotent: returns nil if snapshot does not exist.
func (bs *BlockService) DeleteBlockSnapshot(name string, snapID uint32) error {
	path := bs.volumePath(name)
	return bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		err := vol.DeleteSnapshot(snapID)
		if err != nil && err.Error() == "blockvol: snapshot not found" {
			return nil // idempotent
		}
		return err
	})
}

// RestoreBlockSnapshot restores the named volume to the specified snapshot.
// This is a destructive operation: all writes after the snapshot are lost.
func (bs *BlockService) RestoreBlockSnapshot(name string, snapID uint32) error {
	path := bs.volumePath(name)
	return bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		return vol.RestoreSnapshot(snapID)
	})
}

// ListBlockSnapshots lists all snapshots on the named volume.
func (bs *BlockService) ListBlockSnapshots(name string) ([]blockvol.SnapshotInfo, uint64, error) {
	path := bs.volumePath(name)
	var infos []blockvol.SnapshotInfo
	var volSize uint64
	err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		infos = vol.ListSnapshots()
		volSize = vol.Info().VolumeSize
		return nil
	})
	return infos, volSize, err
}

// ExpandBlockVol expands the named volume to newSize bytes.
func (bs *BlockService) ExpandBlockVol(name string, newSize uint64) (uint64, error) {
	path := bs.volumePath(name)
	var actualSize uint64
	err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		if eerr := vol.Expand(newSize); eerr != nil {
			return eerr
		}
		actualSize = vol.Info().VolumeSize
		return nil
	})
	return actualSize, err
}

// PrepareExpandBlockVol prepares an expand on the named volume without committing.
func (bs *BlockService) PrepareExpandBlockVol(name string, newSize, expandEpoch uint64) error {
	path := bs.volumePath(name)
	return bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		return vol.PrepareExpand(newSize, expandEpoch)
	})
}

// CommitExpandBlockVol commits a prepared expand on the named volume.
func (bs *BlockService) CommitExpandBlockVol(name string, expandEpoch uint64) (uint64, error) {
	path := bs.volumePath(name)
	var actualSize uint64
	err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		if eerr := vol.CommitExpand(expandEpoch); eerr != nil {
			return eerr
		}
		actualSize = vol.Info().VolumeSize
		return nil
	})
	return actualSize, err
}

// CancelExpandBlockVol cancels a prepared expand on the named volume.
func (bs *BlockService) CancelExpandBlockVol(name string, expandEpoch uint64) error {
	path := bs.volumePath(name)
	return bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		return vol.CancelExpand(expandEpoch)
	})
}

// volumePath converts a volume name to its .blk file path.
func (bs *BlockService) volumePath(name string) string {
	sanitized := blockvol.SanitizeFilename(name)
	return filepath.Join(bs.blockDir, sanitized+".blk")
}

// GetReplState returns the replication state for a volume path.
func (bs *BlockService) GetReplState(path string) (dataAddr, ctrlAddr string) {
	bs.replMu.RLock()
	defer bs.replMu.RUnlock()
	if s, ok := bs.replStates[path]; ok {
		return s.replicaDataAddr, s.replicaCtrlAddr
	}
	return "", ""
}

// CollectBlockVolumeHeartbeat returns heartbeat info for all block volumes,
// with replication addresses filled in from BlockService state (R1-4).
func (bs *BlockService) CollectBlockVolumeHeartbeat() []blockvol.BlockVolumeInfoMessage {
	msgs := bs.blockStore.CollectBlockVolumeHeartbeat()
	bs.replMu.RLock()
	defer bs.replMu.RUnlock()
	for i := range msgs {
		if s, ok := bs.replStates[msgs[i].Path]; ok {
			msgs[i].ReplicaDataAddr, msgs[i].ReplicaCtrlAddr = bs.heartbeatReplicaAddrs(msgs[i].Path, s)
		}
		msgs[i].ReplicaDegraded = bs.heartbeatReplicaDegraded(msgs[i].Path, msgs[i].ReplicaDegraded)
		// NVMe publication: report nvme_addr and nqn if NVMe target is running.
		if bs.nvmeListenAddr != "" {
			msgs[i].NvmeAddr = bs.nvmeListenAddr
			// Derive volume name from path for NQN construction.
			name := filepath.Base(msgs[i].Path)
			name = strings.TrimSuffix(name, ".blk")
			msgs[i].NQN = bs.NQN(name)
		}
	}
	return msgs
}

// heartbeatReplicaAddrs returns the scalar replica transport addresses that
// should be exposed on the current heartbeat surface. On the Phase 15 live path
// it prefers the explicit core projection when present, while preserving the
// older adapter-local fallback for unrebound paths.
func (bs *BlockService) heartbeatReplicaAddrs(path string, state *volReplState) (string, string) {
	if state == nil {
		return "", ""
	}
	if proj, ok := bs.CoreProjection(path); ok {
		switch proj.Role {
		case engine.RolePrimary:
			if proj.Readiness.ShipperConfigured {
				return state.replicaDataAddr, state.replicaCtrlAddr
			}
		case engine.RoleReplica:
			if proj.Readiness.ReceiverReady {
				return state.replicaDataAddr, state.replicaCtrlAddr
			}
		}
		return "", ""
	}
	if state.publishHealthy {
		return state.replicaDataAddr, state.replicaCtrlAddr
	}
	return "", ""
}

// heartbeatReplicaDegraded returns the bounded degraded bit for the current
// heartbeat surface. On the Phase 15 live path it prefers the core mode when
// present, then falls back to the runtime-local status bit.
func (bs *BlockService) heartbeatReplicaDegraded(path string, current bool) bool {
	if proj, ok := bs.CoreProjection(path); ok {
		switch proj.Mode.Name {
		case engine.ModeDegraded, engine.ModeNeedsRebuild:
			return true
		default:
			return false
		}
	}
	return current
}

// multiReplicaUnchanged checks if the full replica set is unchanged.
func (bs *BlockService) multiReplicaUnchanged(path string, addrs []blockvol.ReplicaAddr) bool {
	bs.replMu.RLock()
	defer bs.replMu.RUnlock()
	existing, ok := bs.replStates[path]
	if !ok || existing == nil {
		return false
	}
	if len(existing.allReplicas) != len(addrs) {
		return false
	}
	for i := range addrs {
		if existing.allReplicas[i].DataAddr != addrs[i].DataAddr ||
			existing.allReplicas[i].CtrlAddr != addrs[i].CtrlAddr ||
			existing.allReplicas[i].ServerID != addrs[i].ServerID {
			return false
		}
	}
	return true
}

func (bs *BlockService) ensureReplStateLocked(path string) *volReplState {
	if bs.replStates == nil {
		bs.replStates = make(map[string]*volReplState)
	}
	state := bs.replStates[path]
	if state == nil {
		state = &volReplState{}
		bs.replStates[path] = state
	}
	return state
}

func (bs *BlockService) noteRoleApplied(path string, role blockvol.Role) {
	bs.replMu.Lock()
	defer bs.replMu.Unlock()
	state := bs.ensureReplStateLocked(path)
	state.roleApplied = true
	switch role {
	case blockvol.RoleReplica:
		state.receiverReady = false
		state.shipperConfigured = false
		state.replicaEligible = false
		state.publishHealthy = false
	case blockvol.RolePrimary:
		state.receiverReady = false
		state.shipperConfigured = false
		state.replicaEligible = false
		state.publishHealthy = true
	case blockvol.RoleRebuilding:
		state.receiverReady = false
		state.shipperConfigured = false
		state.replicaEligible = false
		state.publishHealthy = false
	default:
		state.receiverReady = false
		state.shipperConfigured = false
		state.replicaEligible = false
		state.publishHealthy = false
		state.replicaDataAddr = ""
		state.replicaCtrlAddr = ""
		state.allReplicas = nil
	}
}

func (bs *BlockService) markPrimaryTransportConfigured(path string, addrs []blockvol.ReplicaAddr) {
	bs.replMu.Lock()
	defer bs.replMu.Unlock()
	state := bs.ensureReplStateLocked(path)
	state.shipperConfigured = len(addrs) > 0
	state.publishHealthy = true
	state.replicaEligible = false
	state.receiverReady = false
	if len(addrs) == 0 {
		state.replicaDataAddr = ""
		state.replicaCtrlAddr = ""
		state.allReplicas = nil
		return
	}
	copied := make([]blockvol.ReplicaAddr, len(addrs))
	copy(copied, addrs)
	state.allReplicas = copied
	state.replicaDataAddr = addrs[0].DataAddr
	state.replicaCtrlAddr = addrs[0].CtrlAddr
}

func (bs *BlockService) markReceiverReady(path, dataAddr, ctrlAddr string) {
	bs.replMu.Lock()
	defer bs.replMu.Unlock()
	state := bs.ensureReplStateLocked(path)
	state.receiverReady = true
	state.replicaEligible = true
	state.publishHealthy = true
	state.shipperConfigured = false
	state.replicaDataAddr = dataAddr
	state.replicaCtrlAddr = ctrlAddr
	state.allReplicas = nil
}

// ReadinessSnapshot reports the service-owned assignment/readiness closure for
// one volume. On the Phase 15 live path it prefers the explicit core projection
// for the aligned readiness subset, while `PublishHealthy` remains adapter-local
// until publication ownership is fully rebound.
func (bs *BlockService) ReadinessSnapshot(path string) BlockReadinessSnapshot {
	snap := BlockReadinessSnapshot{}
	bs.replMu.RLock()
	state := bs.replStates[path]
	if state != nil {
		snap.RoleApplied = state.roleApplied
		snap.ReceiverReady = state.receiverReady
		snap.ShipperConfigured = state.shipperConfigured
		snap.ReplicaEligible = state.replicaEligible
		snap.PublishHealthy = state.publishHealthy
	}
	bs.replMu.RUnlock()
	if !snap.ShipperConfigured || bs.blockStore == nil {
		if proj, ok := bs.CoreProjection(path); ok {
			snap.RoleApplied = proj.Readiness.RoleApplied
			snap.ReceiverReady = proj.Readiness.ReceiverReady
			snap.ShipperConfigured = proj.Readiness.ShipperConfigured
			snap.ShipperConnected = proj.Readiness.ShipperConnected
			snap.ReplicaEligible = proj.Readiness.ReplicaReady
		}
		return snap
	}
	_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
		snap.ShipperConnected = len(vol.ReplicaShipperStates()) > 0 && !vol.Status().ReplicaDegraded
		return nil
	})
	if proj, ok := bs.CoreProjection(path); ok {
		snap.RoleApplied = proj.Readiness.RoleApplied
		snap.ReceiverReady = proj.Readiness.ReceiverReady
		snap.ShipperConfigured = proj.Readiness.ShipperConfigured
		snap.ShipperConnected = proj.Readiness.ShipperConnected
		snap.ReplicaEligible = proj.Readiness.ReplicaReady
	}
	return snap
}

// --- P3: Assignment idempotence ---

// lastAppliedAssignment stores the full assignment for idempotence comparison.
// Keyed by volume path.
type lastAppliedAssignment struct {
	Path            string
	Epoch           uint64
	Role            uint32
	ReplicaServerID string
	ReplicaDataAddr string
	ReplicaCtrlAddr string
	ReplicaAddrs    []blockvol.ReplicaAddr
}

func lastAppliedFrom(a blockvol.BlockVolumeAssignment) lastAppliedAssignment {
	// Copy the slice to avoid aliasing.
	var addrs []blockvol.ReplicaAddr
	if len(a.ReplicaAddrs) > 0 {
		addrs = make([]blockvol.ReplicaAddr, len(a.ReplicaAddrs))
		copy(addrs, a.ReplicaAddrs)
	}
	return lastAppliedAssignment{
		Path:            a.Path,
		Epoch:           a.Epoch,
		Role:            a.Role,
		ReplicaServerID: a.ReplicaServerID,
		ReplicaDataAddr: a.ReplicaDataAddr,
		ReplicaCtrlAddr: a.ReplicaCtrlAddr,
		ReplicaAddrs:    addrs,
	}
}

func (la lastAppliedAssignment) equals(a blockvol.BlockVolumeAssignment) bool {
	if la.Path != a.Path || la.Epoch != a.Epoch || la.Role != a.Role {
		return false
	}
	if la.ReplicaServerID != a.ReplicaServerID || la.ReplicaDataAddr != a.ReplicaDataAddr || la.ReplicaCtrlAddr != a.ReplicaCtrlAddr {
		return false
	}
	if len(la.ReplicaAddrs) != len(a.ReplicaAddrs) {
		return false
	}
	for i := range la.ReplicaAddrs {
		if la.ReplicaAddrs[i].DataAddr != a.ReplicaAddrs[i].DataAddr ||
			la.ReplicaAddrs[i].CtrlAddr != a.ReplicaAddrs[i].CtrlAddr ||
			la.ReplicaAddrs[i].ServerID != a.ReplicaAddrs[i].ServerID {
			return false
		}
	}
	return true
}

func (bs *BlockService) isAssignmentUnchanged(a blockvol.BlockVolumeAssignment) bool {
	bs.lastAssignMu.RLock()
	defer bs.lastAssignMu.RUnlock()
	if bs.lastAssign == nil {
		return false
	}
	last, ok := bs.lastAssign[a.Path]
	if !ok {
		return false
	}
	return last.equals(a)
}

func (bs *BlockService) recordAppliedAssignment(a blockvol.BlockVolumeAssignment) {
	bs.lastAssignMu.Lock()
	defer bs.lastAssignMu.Unlock()
	if bs.lastAssign == nil {
		bs.lastAssign = make(map[string]lastAppliedAssignment)
	}
	bs.lastAssign[a.Path] = lastAppliedFrom(a)
}

// ReplicationPorts computes deterministic replication ports for a volume.
// Ports are derived from a hash of the volume path offset from the iSCSI base port.
func (bs *BlockService) ReplicationPorts(volPath string) (dataPort, ctrlPort, rebuildPort int) {
	basePort := 3260
	if idx := strings.LastIndex(bs.listenAddr, ":"); idx >= 0 {
		var p int
		if _, err := fmt.Sscanf(bs.listenAddr[idx+1:], "%d", &p); err == nil && p > 0 {
			basePort = p
		}
	}
	h := fnv.New32a()
	h.Write([]byte(volPath))
	offset := int(h.Sum32()%500) * 3
	dataPort = basePort + 1000 + offset
	ctrlPort = dataPort + 1
	rebuildPort = dataPort + 2
	return
}

// Shutdown gracefully stops the iSCSI and NVMe targets and closes all block volumes.
func (bs *BlockService) Shutdown() {
	if bs == nil {
		return
	}
	glog.V(0).Infof("block service: shutting down...")
	// P4: drain active recovery goroutines before closing volumes.
	if bs.v2Recovery != nil {
		bs.v2Recovery.Shutdown()
	}
	if bs.nvmeServer != nil {
		bs.nvmeServer.Close()
	}
	if bs.targetServer != nil {
		bs.targetServer.Close()
	}
	bs.blockStore.Close()
	glog.V(0).Infof("block service: shut down")
}

// SetBlockService wires a BlockService into the VolumeServer so that
// heartbeats include block volume info and the server is marked block-capable.
// Also wires shipper state change callbacks for immediate heartbeat on degradation.
func (vs *VolumeServer) SetBlockService(bs *BlockService) {
	vs.blockService = bs
	bs.WireStateChangeNotify(vs.blockStateChangeChan)
}