feat(relayer): persist eth scan progress (#344)

* feat(relayer): persist Ethereum scan progress via checkpoints

The relayer only persisted its Ethereum scan offset as a side effect of
processing a deposit, so block ranges with no deposits never advanced the
persisted offset — on restart the poller re-scanned from the last block that
happened to contain a deposit.

The poller now emits a scan-progress checkpoint after every fully scanned
slice (including empty ones), carried on the existing handler as a
DepositEvent{Checkpoint: true}. The Ethereum source turns it into a checkpoint
relayer.Event that rides the same in-order channel behind that slice's
deposits; the processor persists the offset from it and does no transfer work.

Because the checkpoint is delivered in order after the slice's deposits — and
each deposit is durably recorded via CreateTransfer (idempotent) before submit —
the offset never advances past an unprocessed deposit, so nothing is missed or
double-processed. Carrying the signal on the existing handler keeps the
EthereumBridgeClient interface (and its mock) unchanged.

* fix(relayer): advance scan block only after checkpoint is accepted

Move currentBlock/setLastScannedBlock past the checkpoint handler call so a
failed checkpoint re-scans and re-checkpoints the same range on the next tick
instead of skipping its watermark, matching the documented intent.

* fix(e2e): retry P2 transfers on CONTRACT_NOT_FOUND contention

Author: sadiq1971

pkg/ethereum/client.go

@@ -234,7 +234,16 @@ func chunkRange(start, end, maxRange uint64) []blockRange {
 // config.MaxBlockRange blocks so requests stay under the provider's per-call cap.
 // On a slice failure, currentBlock advances only through the last successful
 // slice and the failing range is retried on the next tick.
-func (c *Client) WatchDepositEvents(ctx context.Context, fromBlock uint64, handler func(*DepositEvent) error) error {
+// After each fully scanned slice — including slices with no deposits — the handler
+// is also invoked with a checkpoint event (DepositEvent{Checkpoint: true}) carrying
+// the last scanned block, so the caller can persist scan progress and avoid
+// re-scanning from the start on restart. The checkpoint rides the same in-order
+// handler path as deposits, so it is only delivered after that slice's deposits.
+func (c *Client) WatchDepositEvents(
+	ctx context.Context,
+	fromBlock uint64,
+	handler func(*DepositEvent) error,
+) error {
 	c.logger.Info("Starting deposit event poller",
 		zap.Uint64("from_block", fromBlock),
 		zap.Uint64("max_block_range", c.config.MaxBlockRange))
@@ -284,6 +293,15 @@ func (c *Client) WatchDepositEvents(ctx context.Context, fromBlock uint64, handl
 					if err := c.scanDepositRange(ctx, r, handler); err != nil {
 						return
 					}
+					// Emit a scan-progress checkpoint after this slice's deposits were
+					// handed to the handler (in order), so a restart resumes near here
+					// instead of re-scanning from the start. Advance currentBlock only
+					// once the checkpoint is accepted, so on failure the same range is
+					// re-scanned and re-checkpointed on the next tick rather than having
+					// its watermark silently skipped.
+					if err := handler(&DepositEvent{BlockNumber: r.end, Checkpoint: true}); err != nil {
+						return
+					}
 					currentBlock = r.end
 					c.setLastScannedBlock(currentBlock)
 				}

pkg/ethereum/types.go

@@ -21,6 +21,13 @@ type DepositEvent struct {
 	BlockNumber     uint64
 	TxHash          common.Hash
 	LogIndex        uint
+
+	// Checkpoint marks a scan-progress signal rather than a real deposit: the poller
+	// emits one (with only BlockNumber set) after each fully scanned block range —
+	// including ranges with no deposits — so the consumer can persist scan progress.
+	// It rides the same in-order handler path as deposits, so it is only observed
+	// after every deposit in the range it covers.
+	Checkpoint bool
 }
 
 // WithdrawalEvent represents a withdrawal from Canton event on Ethereum

pkg/relayer/engine/processor.go

@@ -104,6 +104,13 @@ func (p *Processor) Start(ctx context.Context, startOffset string) error {
 			if !ok {
 				return nil
 			}
+			if event.Checkpoint {
+				// Scan-progress watermark, not a transfer: advance the persisted
+				// offset only. It arrives in order after the events it covers, so
+				// everything up to it has already been processed by this loop.
+				p.persistOffset(ctx, event)
+				continue
+			}
 			if err := p.processEvent(ctx, event); err != nil {
 				p.logger.Error("Failed to process event",
 					zap.String("event_id", event.ID),

pkg/relayer/engine/processor_test.go

@@ -80,6 +80,47 @@ func TestProcessor_Start_ProcessesEventSuccess(t *testing.T) {
 	}
 }
 
+func TestProcessor_Start_CheckpointPersistsOffsetWithoutProcessing(t *testing.T) {
+	ctx := context.Background()
+	source := relayermocks.NewSource(t)
+	destination := relayermocks.NewDestination(t)
+	store := relayermocks.NewBridgeStore(t)
+
+	eventCh := make(chan *relayer.Event, 1)
+	errCh := make(chan error)
+
+	// A scan-progress checkpoint (no transfer payload).
+	eventCh <- &relayer.Event{
+		SourceChain:       relayer.ChainEthereum,
+		SourceBlockNumber: 150,
+		Checkpoint:        true,
+	}
+	close(eventCh)
+
+	source.EXPECT().GetChainID().Return(relayer.ChainEthereum).Maybe()
+	destination.EXPECT().GetChainID().Return(relayer.ChainCanton).Maybe()
+	source.EXPECT().StreamEvents(ctx, "100").Return((<-chan *relayer.Event)(eventCh), (<-chan error)(errCh)).Once()
+	source.EXPECT().ExtractOffset(mock.AnythingOfType("*relayer.Event")).Return("150").Once()
+	// No CreateTransfer / SubmitTransfer / UpdateTransferStatus expected — a
+	// checkpoint must only advance the offset. The strict mocks fail the test if any
+	// transfer-processing call is made.
+
+	var persistedChain, persistedOffset string
+	processor := engine.NewProcessor(source, destination, store, engine.NewNopMetrics(), zap.NewNop(), "processor_test", relayer.DirectionEthereumToCanton).
+		WithOffsetUpdate(func(_ context.Context, chainID string, offset string) error {
+			persistedChain = chainID
+			persistedOffset = offset
+			return nil
+		})
+
+	if err := processor.Start(ctx, "100"); err != nil {
+		t.Fatalf("Start() failed: %v", err)
+	}
+	if persistedChain != relayer.ChainEthereum || persistedOffset != "150" {
+		t.Fatalf("expected checkpoint to persist ethereum offset 150, got chain=%s offset=%s", persistedChain, persistedOffset)
+	}
+}
+
 func TestProcessor_Start_DuplicateTransferPersistsOffsetOnce(t *testing.T) {
 	ctx := context.Background()
 	source := relayermocks.NewSource(t)

pkg/relayer/engine/source.go

@@ -127,10 +127,30 @@ func (s *ethereumSource) StreamEvents(ctx context.Context, offset string) (<-cha
 			fromBlock = n
 		}
 
+		emit := func(ev *relayer.Event) error {
+			select {
+			case outCh <- ev:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+
 		err := s.client.WatchDepositEvents(ctx, fromBlock, func(event *ethereum.DepositEvent) error {
+			// A checkpoint carries scan progress, not a deposit. It rides the same
+			// ordered channel behind that slice's deposits, so the processor persists
+			// the block only once those deposits are processed.
+			if event.Checkpoint {
+				return emit(&relayer.Event{
+					SourceChain:       relayer.ChainEthereum,
+					SourceBlockNumber: event.BlockNumber,
+					Checkpoint:        true,
+				})
+			}
+
 			s.metrics.IncEventsDetected(relayer.ChainEthereum, EventTypeDeposit)
 
-			relayerEvent := &relayer.Event{
+			return emit(&relayer.Event{
 				ID:                fmt.Sprintf("%s-%d", event.TxHash.Hex(), event.LogIndex),
 				TransactionID:     event.TxHash.Hex(),
 				SourceChain:       relayer.ChainEthereum,
@@ -142,14 +162,7 @@ func (s *ethereumSource) StreamEvents(ctx context.Context, offset string) (<-cha
 				Recipient:         fmt.Sprintf("%x", event.CantonRecipient),
 				Nonce:             event.Nonce.Int64(),
 				SourceBlockNumber: event.BlockNumber,
-			}
-
-			select {
-			case outCh <- relayerEvent:
-				return nil
-			case <-ctx.Done():
-				return ctx.Err()
-			}
+			})
 		})
 
 		if err != nil && ctx.Err() == nil {

pkg/relayer/engine/source_test.go

@@ -175,3 +175,30 @@ func TestEthereumSource_StreamEvents_MapsDepositEvent(t *testing.T) {
 		t.Fatal("timed out waiting for mapped ethereum event")
 	}
 }
+
+func TestEthereumSource_StreamEvents_EmitsScanCheckpoint(t *testing.T) {
+	ctx := context.Background()
+	ethClient := relayermocks.NewEthereumBridgeClient(t)
+
+	// The poller reports scan progress via a checkpoint event even with no deposits.
+	ethClient.EXPECT().
+		WatchDepositEvents(ctx, uint64(12), mock.Anything).
+		RunAndReturn(func(_ context.Context, _ uint64, handler func(*ethereum.DepositEvent) error) error {
+			return handler(&ethereum.DepositEvent{BlockNumber: 200, Checkpoint: true})
+		})
+
+	source := engine.NewEthereumSource(ethClient, relayer.ChainEthereum, engine.NewNopMetrics())
+	eventCh, _ := source.StreamEvents(ctx, "12")
+
+	select {
+	case event, ok := <-eventCh:
+		if !ok {
+			t.Fatal("expected a checkpoint event, channel closed")
+		}
+		if !event.Checkpoint || event.SourceBlockNumber != 200 {
+			t.Fatalf("expected checkpoint at block 200, got %+v", event)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for scan checkpoint event")
+	}
+}

pkg/relayer/types.go

@@ -76,4 +76,10 @@ type Event struct {
 	Recipient         string
 	Nonce             int64
 	SourceBlockNumber uint64
+
+	// Checkpoint marks a progress watermark rather than a real transfer: the source
+	// has scanned through SourceBlockNumber and emitted every event up to it. The
+	// processor persists the offset and skips transfer processing. Emitted in-order
+	// after a slice's events so persisting it cannot skip an unprocessed event.
+	Checkpoint bool
 }

tests/e2e/devstack/shim/canton2.go

@@ -243,9 +243,10 @@ func (c *Canton2Shim) GetHoldings(ctx context.Context, ownerParty, tokenSymbol s
 //
 // Parallel e2e tests share the same USDCx issuer party and contend on its
 // Holding set. Canton may reject with LOCAL_VERDICT_LOCKED_CONTRACTS (another
-// transfer holds the lock right now) or LOCAL_VERDICT_INACTIVE_CONTRACTS (the
-// Holding we selected was archived between GetHoldings and submit). Both are
-// transient — refetch holdings on the next attempt and try again.
+// transfer holds the lock right now), LOCAL_VERDICT_INACTIVE_CONTRACTS, or
+// CONTRACT_NOT_FOUND (the Holding we selected was archived between GetHoldings
+// and submit). All are transient — refetch holdings on the next attempt and
+// try again.
 func (c *Canton2Shim) TransferToken(ctx context.Context, senderParty, recipientParty, tokenSymbol, amount string) error {
 	const maxAttempts = 5
 	var lastErr error
@@ -270,7 +271,8 @@ func (c *Canton2Shim) TransferToken(ctx context.Context, senderParty, recipientP
 func isHoldingContention(err error) bool {
 	msg := err.Error()
 	return strings.Contains(msg, "LOCAL_VERDICT_LOCKED_CONTRACTS") ||
-		strings.Contains(msg, "LOCAL_VERDICT_INACTIVE_CONTRACTS")
+		strings.Contains(msg, "LOCAL_VERDICT_INACTIVE_CONTRACTS") ||
+		strings.Contains(msg, "CONTRACT_NOT_FOUND")
 }
 
 // GetFingerprintMapping is not supported on Participant 2.