fix: revert Bridge 2, fix Bridge 1 with per-replica identity

Revert detectAndEnqueueRebuildFromHeartbeat (Bridge 2) — master
should not drive rebuild assignments from heartbeat. The primary
owns data-control recovery per the V2 authority split.

Fix Bridge 1: NeedsRebuildObserved now carries per-replica identity.
resolveReplicaIDForShipper maps shipper DataAddr to ReplicaID via
the shipper group (works for RF=2 and RF=3+). The engine receives
the specific replica that needs rebuild, not a volume-level broadcast.

Primary-direct rebuild: the primary detects which replica needs
rebuild and will drive the session directly. The master learns about
it via subsequent heartbeat projection (needs_rebuild → rebuilding →
healthy). No master round-trip needed for the rebuild decision.

Added WALShipper.DataAddr() getter for address resolution.

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

weed/server/volume_server_block.go

@@ -1345,6 +1345,28 @@ func (bs *BlockService) observePrimaryShipperConnectivity(path string) {
 		// entries are available.
 		_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
 			connected = vol.TryReconnectShippers()
+			if !connected {
+				// Bridge 1: check if any shipper reached NeedsRebuild during
+				// the proactive reconnect. Emit per-replica NeedsRebuildObserved
+				// to the core, then initiate primary-direct rebuild for that
+				// specific replica. The master is NOT involved in the rebuild
+				// decision — the primary owns data-control recovery. The master
+				// sees the result via subsequent heartbeat projection.
+				for _, st := range vol.ReplicaShipperStates() {
+					if st.State == "needs_rebuild" && st.DataAddr != "" {
+						replicaID := bs.resolveReplicaIDForShipper(path, st.DataAddr)
+						glog.V(0).Infof("block service: shipper %s (replica=%s) needs rebuild — primary-direct rebuild",
+							st.DataAddr, replicaID)
+						if replicaID != "" {
+							bs.applyCoreEvent(engine.NeedsRebuildObserved{
+								ID:        path,
+								ReplicaID: replicaID,
+								Reason:    "gap_exceeds_retained_wal",
+							})
+						}
+					}
+				}
+			}
 			return nil
 		})
 	}
@@ -1353,6 +1375,31 @@ func (bs *BlockService) observePrimaryShipperConnectivity(path string) {
 	bs.observePrimaryShipperConnectivityStatus(path, connected)
 }
 
+// resolveReplicaIDForShipper maps a shipper's data address back to the
+// replica ID. Uses the shipper group's per-shipper ReplicaID which was
+// set via SetReplicaAddrs from the ServerID. Works for RF=2 and RF=3+.
+func (bs *BlockService) resolveReplicaIDForShipper(path, dataAddr string) string {
+	if bs == nil || bs.blockStore == nil || dataAddr == "" {
+		return ""
+	}
+	var replicaID string
+	_ = bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
+		sg := vol.GetShipperGroup()
+		if sg == nil {
+			return nil
+		}
+		for i := 0; i < sg.Len(); i++ {
+			s := sg.Shipper(i)
+			if s != nil && s.DataAddr() == dataAddr {
+				replicaID = s.ReplicaID()
+				return nil
+			}
+		}
+		return nil
+	})
+	return replicaID
+}
+
 func (bs *BlockService) observePrimaryShipperConnectivityStatus(path string, connected bool) {
 	if bs == nil || bs.v2Core == nil || !connected {
 		return

weed/storage/blockvol/wal_shipper.go

@@ -111,6 +111,11 @@ func (s *WALShipper) ReplicaID() string {
 	return s.replicaID
 }
 
+// DataAddr returns the shipper's configured data channel address.
+func (s *WALShipper) DataAddr() string {
+	return s.dataAddr
+}
+
 // SetLiveShippingPolicy installs a host-provided gate for current live-tail
 // shipping. The callback is consulted before any network dial or send occurs.
 func (s *WALShipper) SetLiveShippingPolicy(fn func(replicaID string, entryLSN uint64) (allow bool, reason string)) {