Ensure checkpoint capture excludes non-started listeners, and null checkpoint data

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,17 +1112,23 @@ 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)
 		}
 	}
 	es.mux.Unlock()
 
 	// 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
 }

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"}`)

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
 }