Update connector contract to protect delivery-ack window with certainty

[peterbroadhurst]

This fixes a scenario where an event stream listener could silently drop events.
The scenario has been observed in practice when there are multiple listeners on a single event stream, and those listeners are catching up to the head while redelivery is occurring delaying acknowledgment of events detected during catchup by one of those listeners.

Each listener's checkpoint within the overall event stream advances two ways:

1. From an event delivered and acked by the application
2. From FFTM poling the collector with EventListenerHWM to check the position the connector's scan has reached

The second can be ahead of the first, as the connector pushes events into a channel in FFTM and continues to scan forwards to find more events while FFTM assembles those events into a batch, and delivers that batch awaiting ack.

The guards against moving the checkpoint forwards while those events were pending acknowledgment were not strong enough in FFTM.

Fix architecture

A minimal additive change is made to the FFCAPI on the return value to EventListenerHWMResponse(), so that it not only returns the scan position, but also the checkpoint position of the highest event passed to the FFTM channel.

When a connector implements that simple contract (separate PR to EVMConnect), the changes in this FFTM PR provide an absolute assurance that the checkpoint is never moved forward by the scan interval, while there are uncommitted deliveries that have been pushed from the connector to FFTM.

transaction-manager/pkg/ffcapi/event_listener_hwm.go

Lines 28 to 58 in 5e887b9

	// EventListenerHWM return value.
	//
	// Important:
	//
	// The connector SHOULD implement LastDetected as well as Checkpoint on the response returned.
	//
	// Checkpoint is how far the connector has scanned.
	// LastDetected is the checkpoint of the last event it pushed to FFTM.
	//
	// These differ, because the connector can scan past events that are still sat in the
	// channel unread - FFTM cannot see those for itself, so it must be told.
	//
	// FFTM applies Checkpoint only once everything up to LastDetected has been delivered and acked.
	//
	// The connector needs to hold in memory a LastDetected value (to return when asked) that
	// is updated BEFORE updating the value it returns for the scan position.
	// When building the response for EventListenerHWM() read the scan position first, then
	// the last detected (or guard them both under a single mutex).
	//
	// Following these simple rules ensures FFTM never moves the checkpoint past an event
	// that is still in flight.
	//
	// Setting nil LastDetected is correct when nothing has been pushed for this listener.
	//
	// A connector that never implements LastDetected (always nil) is functional, but accepts
	// a risk of a checkpoint being persisted past events FFTM has not yet acknowledged.
	type EventListenerHWMResponse struct {
	Checkpoint EventListenerCheckpoint `json:"checkpoint"` // how far the connector has scanned
	Catchup bool `json:"catchup,omitempty"` // informational only - informs an operator that the stream is catching up
	LastDetected EventListenerCheckpoint `json:"lastDetected,omitempty"` // the checkpoint of the last event it pushed to FFTM
	}

Also included:

• Block listeners no longer take the HWM route. They're handled inside FFTM's own confirmation manager, so we were querying the connector with listener IDs it has never heard of.
• Never silently drop An event arriving behind an HWM-derived checkpoint is now delivered with a warning. We can't distinguish it from a genuine re-detection, and a duplicate is recoverable
  • Note FireFly core de-duplicates on protocol ID, so this does not mean application redelivery in practice

[Chengxuan]

The incremental changes look great to me. A few suggestions on the doc and naming of the variable.