fix: rebuild base-only completion + protocol handshake + direct ack events

Three fixes for the remote rebuild path:

1. Base-only completion: when BaseLSN == TargetLSN, the base image covers
   all data — no WAL tail needed. MarkBaseComplete now auto-satisfies the
   WAL condition and calls TryComplete so the session completes immediately
   after the base transfer finishes.

2. Base lane protocol handshake: runBaseLaneClient now sends MsgRebuildReq
   {Type: RebuildSessionBase} before reading. The RebuildServer requires
   this handshake to dispatch to ServeBaseBlocks. Without it, the server
   received raw frames it couldn't understand.

3. Direct ack events: OnAck emits engine events directly (SessionCompleted,
   SessionProgressObserved, SessionFailed) instead of routing through
   ObserveReplicaRebuildSessionAck which requires the sender in the
   orchestrator registry. The remote coordinator owns the session — no
   registry lookup needed.

Also adds diagnostic logging on both sides:
- Replica: logs parsed RebuildAddr and base lane client start
- Primary: logs sender state after installSession

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

weed/server/block_recovery.go

@@ -862,22 +862,56 @@ func (rm *RecoveryManager) buildRemoteRebuildIO(replicaID, volPath, rebuildAddr
 		Epoch:           epoch,
 		SessionID:       sessionID,
 		OnAck: func(ack blockvol.SessionAckMsg) error {
-			err := bs.ObserveReplicaRebuildSessionAck(volPath, replicaID, ack)
-			// On completion, store the replica's achieved LSN so
-			// readRebuildStatus uses the replica's proof, not the primary's vol.
-			if err == nil && ack.Phase == blockvol.SessionAckCompleted {
-				achieved := ack.AchievedLSN
-				if achieved == 0 {
-					achieved = ack.WALAppliedLSN
+			// Emit engine events directly. The remote coordinator owns the
+			// session — no sender registry lookup needed. This avoids the
+			// "sender not found" failure when the registry is reconciled
+			// between session install and ack arrival.
+			if bs == nil || bs.v2Core == nil {
+				return nil
+			}
+			achieved := ack.AchievedLSN
+			if achieved == 0 {
+				achieved = ack.WALAppliedLSN
+			}
+			switch ack.Phase {
+			case blockvol.SessionAckAccepted:
+				// No engine event needed — shipper transition handled by caller.
+			case blockvol.SessionAckRunning, blockvol.SessionAckBaseComplete:
+				if achieved > 0 {
+					bs.applyCoreEvent(engine.SessionProgressObserved{
+						ID:          volPath,
+						ReplicaID:   replicaID,
+						Kind:        engine.SessionRebuild,
+						AchievedLSN: achieved,
+					})
 				}
+			case blockvol.SessionAckCompleted:
+				// Store achieved LSN for OnRebuildCompleted (skip local vol read).
 				rm.mu.Lock()
 				if rm.remoteRebuildAchieved == nil {
 					rm.remoteRebuildAchieved = make(map[string]uint64)
 				}
 				rm.remoteRebuildAchieved[replicaID] = achieved
 				rm.mu.Unlock()
+				bs.applyCoreEvent(engine.SessionCompleted{
+					ID:          volPath,
+					ReplicaID:   replicaID,
+					Kind:        engine.SessionRebuild,
+					AchievedLSN: achieved,
+				})
+			case blockvol.SessionAckFailed:
+				reason := "rebuild_failed"
+				if ack.BaseComplete {
+					reason = "rebuild_failed_post_base"
+				}
+				bs.applyCoreEvent(engine.SessionFailed{
+					ID:        volPath,
+					ReplicaID: replicaID,
+					Kind:      engine.SessionRebuild,
+					Reason:    reason,
+				})
 			}
-			return err
+			return nil
 		},
 		TransitionShipper: func(state blockvol.ReplicaState) {
 			if shipperRef != nil {

weed/storage/blockvol/rebuild_session.go

@@ -217,9 +217,18 @@ func (s *RebuildSession) MarkBaseComplete(totalBlocks uint64) {
 	if s.phase == RebuildPhaseRunning {
 		s.phase = RebuildPhaseBaseComplete
 	}
+	// When BaseLSN == TargetLSN, the base image covers all data — no WAL
+	// tail needed. Auto-satisfy the WAL condition so TryComplete succeeds
+	// immediately after base transfer.
+	if s.config.BaseLSN == s.config.TargetLSN && s.walAppliedLSN < s.config.TargetLSN {
+		s.walAppliedLSN = s.config.TargetLSN
+	}
 	ack := s.sessionAckLocked()
 	s.mu.Unlock()
 	s.vol.emitRebuildSessionAck(ack)
+	// Try to complete immediately — covers the BaseLSN == TargetLSN case
+	// where no WAL entries will arrive.
+	s.TryComplete()
 }
 
 // TryComplete checks if both completion conditions are met:

weed/storage/blockvol/replica_barrier.go

@@ -67,6 +67,8 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr
 	}
 	switch ctrl.Command {
 	case SessionCmdStartRebuild:
+		log.Printf("replica: handleSessionControl start_rebuild session=%d epoch=%d base=%d target=%d rebuildAddr=%q",
+			ctrl.SessionID, ctrl.Epoch, ctrl.BaseLSN, ctrl.TargetLSN, ctrl.RebuildAddr)
 		r.vol.SetOnRebuildSessionAck(func(ack SessionAckMsg) {
 			if ack.SessionID != ctrl.SessionID {
 				return
@@ -111,12 +113,26 @@ func (r *ReplicaReceiver) handleSessionControl(conn net.Conn, payload []byte, wr
 // session control message. On failure, the rebuild session will eventually time
 // out or be cancelled by the primary.
 func (r *ReplicaReceiver) runBaseLaneClient(sessionID uint64, rebuildAddr string) {
+	log.Printf("replica: base lane client starting session=%d addr=%s", sessionID, rebuildAddr)
 	conn, err := net.DialTimeout("tcp", rebuildAddr, 5*time.Second)
 	if err != nil {
 		log.Printf("replica: base lane dial %s: %v", rebuildAddr, err)
 		return
 	}
 	defer conn.Close()
+
+	// Send rebuild request so the RebuildServer dispatches to ServeBaseBlocks.
+	epoch := r.vol.Epoch()
+	req := RebuildRequest{
+		Type:    RebuildSessionBase,
+		Epoch:   epoch,
+		FromLSN: 0, // full base
+	}
+	if err := WriteFrame(conn, MsgRebuildReq, EncodeRebuildRequest(req)); err != nil {
+		log.Printf("replica: base lane send request to %s: %v", rebuildAddr, err)
+		return
+	}
+
 	client := NewRebuildTransportClient(r.vol, sessionID)
 	blocks, err := client.ReceiveBaseBlocks(conn)
 	if err != nil {