diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index f6a706082..ca03c1eb1 100644 --- a/weed/filer/meta_aggregator.go +++ b/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 } diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go index 40c3b04a7..8cb4512b4 100644 --- a/weed/pb/filer_pb_tail.go +++ b/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) } } } diff --git a/weed/pb/filer_pb_tail_test.go b/weed/pb/filer_pb_tail_test.go index e647165a6..224d323bd 100644 --- a/weed/pb/filer_pb_tail_test.go +++ b/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) + } +}