From 345e224bbbd65c72cfb6930cf7340e1d06b17629 Mon Sep 17 00:00:00 2001 From: Peter Broadhurst Date: Fri, 28 Aug 2026 14:33:37 -0400 Subject: [PATCH] Checkpoint loop fixes Signed-off-by: Peter Broadhurst --- internal/events/eventstream.go | 35 +++++- internal/events/eventstream_test.go | 188 +++++++++++++++++++++++++++- internal/events/listener.go | 19 ++- 3 files changed, 230 insertions(+), 12 deletions(-) diff --git a/internal/events/eventstream.go b/internal/events/eventstream.go index 039d5881..2d5666f4 100644 --- a/internal/events/eventstream.go +++ b/internal/events/eventstream.go @@ -470,7 +470,11 @@ func (es *eventStream) AddOrUpdateListener(ctx context.Context, id *fftypes.UUID } } else if isNew && startedState != nil { if l.spec.Type != nil && *l.spec.Type == apitypes.ListenerTypeBlocks { - return spec, l.es.confirmations.StartConfirmedBlockListener(ctx, l.spec.ID, *l.spec.FromBlock, nil /* new so no checkpoint */, es.batchChannel) + err := l.es.confirmations.StartConfirmedBlockListener(ctx, l.spec.ID, *l.spec.FromBlock, nil /* new so no checkpoint */, es.batchChannel) + if err == nil { + l.markStarted(true) + } + return spec, err } // Start the new listener - no checkpoint needed here return spec, l.start(startedState, nil) @@ -616,6 +620,14 @@ func (es *eventStream) start(ctx context.Context, apiManagedCheckpoint *apitypes return nil, err } + // The connector has accepted the stream start with all the initial event listeners - mark them + // started so the checkpoint loop knows it can query them (note we hold es.mux) + for _, l := range es.listeners { + if l.spec.Type == nil || *l.spec.Type != apitypes.ListenerTypeBlocks { + l.started = true + } + } + // Kick off the loops go es.eventLoop(startedState) if !es.apiManagedStream { @@ -633,6 +645,9 @@ func (es *eventStream) start(ctx context.Context, apiManagedCheckpoint *apitypes log.L(startedState.ctx).Errorf("Failed to start block listener: %s", err) return nil, err } + if l := es.listeners[*bl.ListenerID]; l != nil { + l.started = true // note we hold es.mux + } } return startedState, err @@ -698,6 +713,10 @@ func (es *eventStream) Stop(ctx context.Context) error { es.mux.Lock() es.currentState = nil defer es.mux.Unlock() + // The connector no longer has any of our listeners - they will be re-added on the next start + for _, l := range es.listeners { + l.started = false + } return es.checkSetStatus(ctx, apitypes.EventStreamStatusStopping, apitypes.EventStreamStatusStopped) } @@ -1093,8 +1112,13 @@ func (es *eventStream) generateCheckpoint(startedState *startedStreamState, batc } staleCheckpoints := make([]*listener, 0) for lID, l := range es.listeners { - cp.Listeners[lID], _ = json.Marshal(l.checkpoint) - if l.checkpoint == nil || l.lastCheckpoint == nil || time.Since(*l.lastCheckpoint.Time()) > es.checkpointInterval { + // Store all non-nil current checkpoints in the map + if l.checkpoint != nil { + cp.Listeners[lID], _ = json.Marshal(l.checkpoint) + } + // Add all started listeners with non-existent or stale checkpoints to the stale list, + // which we query below after we've dropped the lock + if l.started && (l.checkpoint == nil || l.lastCheckpoint == nil || time.Since(*l.lastCheckpoint.Time()) > es.checkpointInterval) { staleCheckpoints = append(staleCheckpoints, l) } } @@ -1102,8 +1126,9 @@ func (es *eventStream) generateCheckpoint(startedState *startedStreamState, batc // Ask the connector for any updated high watermark checkpoints - checking we don't have any in-flight confirmations for _, l := range staleCheckpoints { - cpb, _ := json.Marshal(es.checkUpdateHWMCheckpoint(startedState.ctx, l)) - cp.Listeners[*l.spec.ID] = cpb + if updatedCheckpoint := es.checkUpdateHWMCheckpoint(startedState.ctx, l); updatedCheckpoint != nil { + cp.Listeners[*l.spec.ID], _ = json.Marshal(updatedCheckpoint) + } } return cp } diff --git a/internal/events/eventstream_test.go b/internal/events/eventstream_test.go index 7c247d4b..1361fab8 100644 --- a/internal/events/eventstream_test.go +++ b/internal/events/eventstream_test.go @@ -2088,7 +2088,8 @@ func TestHWMCheckpointAfterInactivity(t *testing.T) { ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) li := &listener{ - spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + started: true, } mcm := &confirmationsmocks.Manager{} @@ -2133,7 +2134,8 @@ func TestHWMCheckpointInFlightSkip(t *testing.T) { ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) li := &listener{ - spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + started: true, } mcm := &confirmationsmocks.Manager{} @@ -2146,7 +2148,9 @@ func TestHWMCheckpointInFlightSkip(t *testing.T) { msp := es.checkpointsDB.(*persistencemocks.Persistence) msp.On("WriteCheckpoint", mock.Anything, mock.MatchedBy(func(cp *apitypes.EventStreamCheckpoint) bool { - return cp.StreamID.Equals(es.spec.ID) && string(cp.Listeners[*li.spec.ID]) == `null` + // The listener has no checkpoint, so no entry must be written for it (not a literal JSON "null") + _, hasEntry := cp.Listeners[*li.spec.ID] + return cp.StreamID.Equals(es.spec.ID) && !hasEntry })).Return(nil) es.checkpointInterval = 1 * time.Microsecond @@ -2170,7 +2174,8 @@ func TestHWMCheckpointFail(t *testing.T) { ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) li := &listener{ - spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + started: true, } mcm := &confirmationsmocks.Manager{} @@ -2188,7 +2193,9 @@ func TestHWMCheckpointFail(t *testing.T) { msp := es.checkpointsDB.(*persistencemocks.Persistence) msp.On("WriteCheckpoint", mock.Anything, mock.MatchedBy(func(cp *apitypes.EventStreamCheckpoint) bool { - return cp.StreamID.Equals(es.spec.ID) && string(cp.Listeners[*li.spec.ID]) == `null` + // The listener has no checkpoint, so no entry must be written for it (not a literal JSON "null") + _, hasEntry := cp.Listeners[*li.spec.ID] + return cp.StreamID.Equals(es.spec.ID) && !hasEntry })).Return(nil) es.checkpointInterval = 1 * time.Microsecond @@ -2200,6 +2207,177 @@ func TestHWMCheckpointFail(t *testing.T) { mcm.AssertExpectations(t) } +func TestHWMCheckpointPersistedDuringCatchup(t *testing.T) { + + es := newTestEventStream(t, `{ + "name": "ut_stream" + }`) + + ss := &startedStreamState{} + ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) + defer ss.cancelCtx() + + li := &listener{ + spec: &apitypes.Listener{ID: fftypes.NewUUID()}, + started: true, + } + + mcm := &confirmationsmocks.Manager{} + mcm.On("CheckInFlight", li.spec.ID).Return(false) + es.confirmations = mcm + es.confirmationsRequired = 1 + es.listeners[*li.spec.ID] = li + es.checkpointInterval = 1 * time.Microsecond + + // A listener catching up from a historical fromBlock reports its moving scan position as its + // HWM, with the informational catchup flag set. That scan position is the correct checkpoint - + // it must be persisted regularly, or a restart mid-catchup would rewind the listener to the + // beginning of a scan that can take hours + mfc := es.connector.(*ffcapimocks.API) + mfc.On("EventListenerHWM", mock.Anything, mock.MatchedBy(func(req *ffcapi.EventListenerHWMRequest) bool { + return req.StreamID.Equals(es.spec.ID) && req.ListenerID.Equals(li.spec.ID) + })).Return(&ffcapi.EventListenerHWMResponse{ + Checkpoint: &utCheckpointType{SomeSequenceNumber: 12345}, + Catchup: true, + }, ffcapi.ErrorReason(""), nil).Once() + + cp := es.generateCheckpoint(ss, nil) + assert.Equal(t, &utCheckpointType{SomeSequenceNumber: 12345}, li.checkpoint) + assert.JSONEq(t, `{"someSequenceNumber":12345}`, string(cp.Listeners[*li.spec.ID])) + + // And again as the scan position advances + li.lastCheckpoint = nil // force staleness rather than sleeping + mfc.On("EventListenerHWM", mock.Anything, mock.Anything).Return(&ffcapi.EventListenerHWMResponse{ + Checkpoint: &utCheckpointType{SomeSequenceNumber: 23456}, + Catchup: true, + }, ffcapi.ErrorReason(""), nil).Once() + + cp = es.generateCheckpoint(ss, nil) + assert.Equal(t, &utCheckpointType{SomeSequenceNumber: 23456}, li.checkpoint) + assert.JSONEq(t, `{"someSequenceNumber":23456}`, string(cp.Listeners[*li.spec.ID])) + + mfc.AssertExpectations(t) + mcm.AssertExpectations(t) +} + +func TestCheckpointRoundTripNeverCheckpointedListener(t *testing.T) { + + es := newTestEventStream(t, `{ + "name": "ut_stream" + }`) + + ss := &startedStreamState{} + ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) + defer ss.cancelCtx() + + li := &listener{ + es: es, + spec: &apitypes.Listener{ + ID: apitypes.NewULID(), + Name: strPtr("ut_listener"), + FromBlock: strPtr("0"), + }, + started: true, + } + + mcm := &confirmationsmocks.Manager{} + mcm.On("CheckInFlight", li.spec.ID).Return(false) + es.confirmations = mcm + es.confirmationsRequired = 1 + es.listeners[*li.spec.ID] = li + + // The connector cannot supply a HWM checkpoint + mfc := es.connector.(*ffcapimocks.API) + mfc.On("EventListenerHWM", mock.Anything, mock.Anything).Return(nil, ffcapi.ErrorReason(""), fmt.Errorf("pop")) + + // The generated checkpoint must carry no entry for the listener at all + cp := es.generateCheckpoint(ss, nil) + _, hasEntry := cp.Listeners[*li.spec.ID] + assert.False(t, hasEntry) + + // So on restart, no checkpoint is passed to the connector, and it initializes from fromBlock + req := li.buildAddRequest(context.Background(), cp) + assert.Nil(t, req.Checkpoint) + + // A legacy "null" entry written by an older version must be treated the same as no checkpoint, + // rather than being passed to the connector as a zero-valued checkpoint + cp.Listeners[*li.spec.ID] = json.RawMessage(`null`) + req = li.buildAddRequest(context.Background(), cp) + assert.Nil(t, req.Checkpoint) + + // Same for a block listener - where prior to this check a "null" entry unmarshalled to a + // zero-valued (non-nil) checkpoint at block zero + blSpec := &apitypes.Listener{ + ID: apitypes.NewULID(), + Name: strPtr("ut_block_listener"), + Type: &apitypes.ListenerTypeBlocks, + FromBlock: strPtr(ffcapi.FromBlockLatest), + } + bl := &listener{es: es, spec: blSpec} + blar := bl.buildBlockAddRequest(context.Background(), &apitypes.EventStreamCheckpoint{ + Listeners: apitypes.CheckpointListeners{ + *blSpec.ID: json.RawMessage(`null`), + }, + }) + assert.Nil(t, blar.Checkpoint) +} + +func TestNoHWMCheckpointForNotYetStartedListener(t *testing.T) { + + es := newTestEventStream(t, `{ + "name": "ut_stream" + }`) + + ss := &startedStreamState{} + ss.ctx, ss.cancelCtx = context.WithCancel(context.Background()) + defer ss.cancelCtx() + + // The listener is visible in the map, but the connector has not yet accepted it + // (EventListenerAdd has not returned) - as happens in AddOrUpdateListener between + // lockedListenerUpdate and l.start + li := &listener{ + es: es, + spec: &apitypes.Listener{ + ID: apitypes.NewULID(), + Name: strPtr("ut_listener"), + FromBlock: strPtr("0"), + }, + } + + mcm := &confirmationsmocks.Manager{} + es.confirmations = mcm + es.confirmationsRequired = 1 + es.listeners[*li.spec.ID] = li + + // Note no EventListenerHWM (or CheckInFlight) expectations - a checkpoint cycle must not + // query the connector for a listener it does not know about yet + mfc := es.connector.(*ffcapimocks.API) + cp := es.generateCheckpoint(ss, nil) + _, hasEntry := cp.Listeners[*li.spec.ID] + assert.False(t, hasEntry) + mfc.AssertNotCalled(t, "EventListenerHWM", mock.Anything, mock.Anything) + + // A successful EventListenerAdd (via l.start) marks it started + mfc.On("EventListenerAdd", mock.Anything, mock.Anything).Return(&ffcapi.EventListenerAddResponse{}, ffcapi.ErrorReason(""), nil).Once() + err := li.start(ss, nil) + assert.NoError(t, err) + assert.True(t, li.started) + + // And a listener stop marks it not started again + mfc.On("EventListenerRemove", mock.Anything, mock.Anything).Return(&ffcapi.EventListenerRemoveResponse{}, ffcapi.ErrorReason(""), nil).Once() + err = li.stop(ss) + assert.NoError(t, err) + assert.False(t, li.started) + + // A failed EventListenerAdd must not mark it started + mfc.On("EventListenerAdd", mock.Anything, mock.Anything).Return(nil, ffcapi.ErrorReason(""), fmt.Errorf("pop")).Once() + err = li.start(ss, nil) + assert.Error(t, err) + assert.False(t, li.started) + + mfc.AssertExpectations(t) +} + func TestCheckConfirmedEventForBatchIgnoreInvalid(t *testing.T) { es := newTestEventStream(t, `{"name": "ut_stream"}`) diff --git a/internal/events/listener.go b/internal/events/listener.go index a0a42e67..3d5e2414 100644 --- a/internal/events/listener.go +++ b/internal/events/listener.go @@ -31,6 +31,7 @@ type listener struct { spec *apitypes.Listener lastCheckpoint *fftypes.FFTime checkpoint ffcapi.EventListenerCheckpoint + started bool // protected by es.mux - set only once the connector has accepted the listener, so we never query the connector for a listener it does not know } type blockListenerAddRequest struct { @@ -49,7 +50,14 @@ func listenerSpecToOptions(spec *apitypes.Listener) ffcapi.EventListenerOptions } } +func (l *listener) markStarted(started bool) { + l.es.mux.Lock() + l.started = started + l.es.mux.Unlock() +} + func (l *listener) stop(startedState *startedStreamState) (err error) { + l.markStarted(false) if l.spec.Type != nil && *l.spec.Type == apitypes.ListenerTypeBlocks { err = l.es.confirmations.StopConfirmedBlockListener(startedState.ctx, l.spec.ID) } else { @@ -62,6 +70,10 @@ func (l *listener) stop(startedState *startedStreamState) (err error) { return } +func notNull(jsonCP json.RawMessage) bool { + return jsonCP != nil && string(jsonCP) != "null" +} + func (l *listener) buildAddRequest(ctx context.Context, cp *apitypes.EventStreamCheckpoint) *ffcapi.EventListenerAddRequest { req := &ffcapi.EventListenerAddRequest{ EventListenerOptions: listenerSpecToOptions(l.spec), @@ -71,7 +83,7 @@ func (l *listener) buildAddRequest(ctx context.Context, cp *apitypes.EventStream } if cp != nil { jsonCP := cp.Listeners[*l.spec.ID] - if jsonCP != nil { + if notNull(jsonCP) /* guard against previously persisted null values */ { listenerCheckpoint := l.es.connector.EventStreamNewCheckpointStruct() err := json.Unmarshal(jsonCP, &listenerCheckpoint) if err != nil { @@ -93,7 +105,7 @@ func (l *listener) buildBlockAddRequest(ctx context.Context, cp *apitypes.EventS } if cp != nil { jsonCP := cp.Listeners[*l.spec.ID] - if jsonCP != nil { + if notNull(jsonCP) { var listenerCheckpoint ffcapi.BlockListenerCheckpoint err := json.Unmarshal(jsonCP, &listenerCheckpoint) if err != nil { @@ -108,5 +120,8 @@ func (l *listener) buildBlockAddRequest(ctx context.Context, cp *apitypes.EventS func (l *listener) start(startedState *startedStreamState, cp *apitypes.EventStreamCheckpoint) error { _, _, err := l.es.connector.EventListenerAdd(startedState.ctx, l.buildAddRequest(startedState.ctx, cp)) + if err == nil { + l.markStarted(true) + } return err }