fix(filer.sync): guard batched events against nil EventNotification (#9877)

* fix(filer.sync): guard batched events against nil EventNotification

The server folds a backlog into one response: the first event in the
top-level fields, the rest in resp.Events, and the pipelined sender can
drain an idle heartbeat (nil EventNotification) into that tail. Only the
envelope got the freshness-signal guard, so a batched heartbeat reached
AddSyncJob and nil-derefed in IsEmpty while replaying a backlog buffered
during a peer outage.

Route every event, envelope and batched, through one handler that sends
freshness signals (nil heartbeat, empty marker) to OnIdleHeartbeat.

* fix(filer): guard MetaAggregator batched events against nil EventNotification

The peer subscription's envelope is nil-guarded but its batched tail was
not. The aggregator doesn't enable idle heartbeats today, so the server
can't fold a nil EventNotification into the batch yet, but make the two
loops consistent so it can't nil-deref if that changes.

weed/filer/meta_aggregator.go

@@ -307,8 +307,13 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 					return err
 				}
 			}
-			// Process any additional batched events
+			// Process any additional batched events. Mirror the envelope's nil
+			// guard: the server can fold a freshness signal (nil EventNotification)
+			// into the batched tail, and processOne dereferences it.
 			for _, batchedEvent := range resp.Events {
+				if batchedEvent.EventNotification == nil {
+					continue
+				}
 				if err := processOne(batchedEvent); err != nil {
 					return err
 				}

weed/pb/filer_pb_tail.go

@@ -109,6 +109,30 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc
 			}
 		}
 
+		// handleOneEvent processes a single response, whether it arrived as the
+		// batch envelope or inside resp.Events. Freshness signals carry a timestamp
+		// but no entry: the idle heartbeat (nil EventNotification) and the
+		// MaxUnsyncedEvents marker (empty one). Both route to OnIdleHeartbeat,
+		// else the marker pins sync_offset to the stale processed watermark and a
+		// nil heartbeat folded into a batch nil-derefs in the sync job path.
+		handleOneEvent := func(resp *filer_pb.SubscribeMetadataResponse) {
+			if resp.EventNotification == nil || filer_pb.IsEmpty(resp) {
+				if resp.TsNs > 0 && option.OnIdleHeartbeat != nil {
+					option.OnIdleHeartbeat(resp.TsNs)
+				}
+				// The marker advances the resume cursor past the filtered range; the
+				// heartbeat leaves StartTsNs put so a restart cannot outrun a straggler.
+				if resp.EventNotification != nil && resp.TsNs > 0 {
+					option.StartTsNs = resp.TsNs
+				}
+				return
+			}
+			if err := processEventFn(resp); err != nil {
+				handleErr(resp, err)
+			}
+			option.StartTsNs = resp.TsNs
+		}
+
 		var pendingRefs []*filer_pb.LogFileChunkRef
 
 		for {
@@ -145,36 +169,13 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc
 				pendingRefs = nil
 			}
 
-			// Freshness signals carry a timestamp but no entry: the idle heartbeat
-			// (nil EventNotification) and the MaxUnsyncedEvents marker (empty one).
-			// Route both to OnIdleHeartbeat, else the marker pins sync_offset to the
-			// stale processed watermark.
-			if len(resp.Events) == 0 && resp.TsNs > 0 && (resp.EventNotification == nil || filer_pb.IsEmpty(resp)) {
-				if option.OnIdleHeartbeat != nil {
-					option.OnIdleHeartbeat(resp.TsNs)
-				}
-				// Marker advances the resume cursor past the filtered range; the
-				// heartbeat leaves StartTsNs put so a restart cannot outrun a straggler.
-				if resp.EventNotification != nil {
-					option.StartTsNs = resp.TsNs
-				}
-				continue
-			}
-
-			// Process the first event (always present in top-level fields)
-			if resp.EventNotification != nil {
-				if err := processEventFn(resp); err != nil {
-					handleErr(resp, err)
-				}
-				option.StartTsNs = resp.TsNs
-			}
-
-			// Process any additional batched events
+			// Process the envelope event (top-level fields) and any batched tail.
+			// The server folds a backlog into one response: the first event lives in
+			// the top-level fields, the rest in resp.Events. Either slot can hold a
+			// freshness signal, so both go through the same handler.
+			handleOneEvent(resp)
 			for _, batchedEvent := range resp.Events {
-				if err := processEventFn(batchedEvent); err != nil {
-					handleErr(batchedEvent, err)
-				}
-				option.StartTsNs = batchedEvent.TsNs
+				handleOneEvent(batchedEvent)
 			}
 		}
 	}

weed/pb/filer_pb_tail_test.go

@@ -123,3 +123,72 @@ func TestFilerSyncOffsetStaysFreshOnFilteredMarker(t *testing.T) {
 		t.Fatalf("expected final gauge write fresh at %d, got %+v (spike is back if stale %d)", markerTs, last, oldEventTs)
 	}
 }
+
+// TestFilerSyncBatchedFreshnessSignalDoesNotCrash: while catching up after a
+// peer outage the server folds a backlog into one batched response: the first
+// event in the top-level fields, the rest in resp.Events. The drain can pull an
+// idle heartbeat (nil EventNotification) into that tail. The batched tail must
+// get the same freshness-signal handling as the envelope, else processEventFn
+// (filer.sync's AddSyncJob) nil-derefs in IsEmpty. The processEventFn here
+// mirrors that first call.
+func TestFilerSyncBatchedFreshnessSignalDoesNotCrash(t *testing.T) {
+	const ts1 = int64(1_000_000_000)      // envelope: real create
+	const ts2 = int64(1_000_000_001)      // batched: real create
+	const markerTs = int64(1_000_000_002) // batched: MaxUnsyncedEvents marker (empty entry)
+	hbTs := time.Now().UnixNano()         // batched: idle heartbeat (nil EventNotification), fresh
+
+	var realEvents []int64
+	var heartbeatTs []int64
+
+	// Mirrors AddSyncJob: IsEmpty nil-derefs on a nil EventNotification.
+	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if filer_pb.IsEmpty(resp) {
+			return nil
+		}
+		realEvents = append(realEvents, resp.TsNs)
+		return nil
+	}
+
+	option := &MetadataFollowOption{
+		ClientName:     "syncFrom_A_To_B",
+		StartTsNs:      ts1,
+		EventErrorType: DontLogError,
+		OnIdleHeartbeat: func(tsNs int64) {
+			heartbeatTs = append(heartbeatTs, tsNs)
+		},
+	}
+
+	// One batched response: envelope plus a tail mixing a real event, a marker,
+	// and a fresh heartbeat, as the server emits while draining a backlog. The
+	// heartbeat is last and carries the largest timestamp on purpose.
+	stream := &fakeSubscribeStream{
+		responses: []*filer_pb.SubscribeMetadataResponse{{
+			Directory:         "/watched",
+			TsNs:              ts1,
+			EventNotification: &filer_pb.EventNotification{NewEntry: &filer_pb.Entry{Name: "a"}},
+			Events: []*filer_pb.SubscribeMetadataResponse{
+				{Directory: "/watched", TsNs: ts2, EventNotification: &filer_pb.EventNotification{NewEntry: &filer_pb.Entry{Name: "b"}}},
+				{TsNs: markerTs, EventNotification: &filer_pb.EventNotification{}}, // marker: empty entry
+				{TsNs: hbTs}, // idle heartbeat: nil EventNotification
+			},
+		}},
+	}
+
+	if err := makeSubscribeMetadataFunc(option, processEventFn)(&fakeFilerClient{stream: stream}); err != nil {
+		t.Fatalf("follow: %v", err)
+	}
+
+	// Both real events reached processEventFn; neither freshness signal did.
+	if len(realEvents) != 2 || realEvents[0] != ts1 || realEvents[1] != ts2 {
+		t.Errorf("expected real events [%d %d], got %v", ts1, ts2, realEvents)
+	}
+	// The batched marker and heartbeat both fired OnIdleHeartbeat.
+	if len(heartbeatTs) != 2 || heartbeatTs[0] != markerTs || heartbeatTs[1] != hbTs {
+		t.Errorf("expected heartbeats [%d %d], got %v", markerTs, hbTs, heartbeatTs)
+	}
+	// The marker advances the resume cursor; the heartbeat does not, even though
+	// it is last and carries the largest timestamp. StartTsNs ends at the marker.
+	if option.StartTsNs != markerTs {
+		t.Errorf("expected StartTsNs %d (marker), got %d (heartbeat must not advance the cursor)", markerTs, option.StartTsNs)
+	}
+}