Fix: use hex for transaction digest in events (#257)

* update event indexer and txn indexer to insert txDigest as a hex string

* fix bug related to events cursor txDigest

* update events indexer test

* tidy

* block hash base58

* Misc fixes (#274)

* allow maxRetries instead of maxRetries - 1

* avoid overwriting the module name in event selector when querying for events

* update tests to skip if the DB url is not provided

* update txm confirmer test

* edit error logs

* Enhancement/allow override event offset (#295)

* firedrill contracts test

* add firedrill sanity check

* extend configs to allow overriding event offset

* add an offset override fallback for events

* remove unnecessary unlock

* ensure base58 instead of hex when parsing block and tx digest in transaction indexer

* cleanup logs

* search SourceChainConfigSet using strings instead of u64 for chain selectors

* add a database table to store transactions cursors when replaying failed transactions from transmitters

* disable firedrill sanity test

---------

Co-authored-by: Sishir Giri <[email protected]>
Co-authored-by: FelixFan1992 <[email protected]>

Author: 3 people

relayer/chainreader/config/config.go

@@ -58,6 +58,10 @@ type ChainReaderEvent struct {
 
 	// The expected event type (optional). When not provided, the event type is used as-is.
 	ExpectedEventType any
+
+	// A fallback for events selectors with no offset recorded in the DB and a starting point
+	// earlier than the pruning cutoff of the RPC
+	EventSelectorDefaultOffset *client.EventId
 }
 
 type EventsIndexerConfig struct {

relayer/chainreader/database/database.go

@@ -49,6 +49,11 @@ func (store *DBStore) EnsureSchema(ctx context.Context) error {
 		return fmt.Errorf("failed to create sui indexes: %w", err)
 	}
 
+	_, err = store.ds.ExecContext(ctx, CreateTransmitterCursorsTable)
+	if err != nil {
+		return fmt.Errorf("failed to create sui.transmitter_cursors table: %w", err)
+	}
+
 	return nil
 }
 
@@ -235,3 +240,23 @@ func operatorSQL(op primitives.ComparisonOperator) string {
 		return "="
 	}
 }
+
+func (store *DBStore) GetTransmitterCursor(ctx context.Context, transmitter models.SuiAddress) (string, error) {
+	var cursor string
+	err := store.ds.QueryRowxContext(ctx, GetTransmitterCursor, transmitter).Scan(&cursor)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return "", nil
+		}
+		return "", fmt.Errorf("failed to get transmitter cursor: %w", err)
+	}
+	return cursor, nil
+}
+
+func (store *DBStore) UpdateTransmitterCursor(ctx context.Context, transmitter models.SuiAddress, cursor string) error {
+	_, err := store.ds.ExecContext(ctx, UpdateTransmitterCursor, transmitter, cursor)
+	if err != nil {
+		return fmt.Errorf("failed to update transmitter cursor: %w", err)
+	}
+	return nil
+}

relayer/chainreader/database/database_queries.go

@@ -21,6 +21,13 @@ const (
 	);
     `
 
+	CreateTransmitterCursorsTable = `
+	CREATE TABLE IF NOT EXISTS sui.transmitter_cursors (
+		transmitter TEXT PRIMARY KEY,
+		cursor TEXT NOT NULL
+	);
+	`
+
 	CreateIndices = `
 	CREATE INDEX IF NOT EXISTS idx_events_account_handle_timestamp ON sui.events(event_account_address, event_handle, block_timestamp DESC);
 	CREATE INDEX IF NOT EXISTS idx_events_offset ON sui.events(event_account_address, event_handle, event_offset);
@@ -78,4 +85,15 @@ const (
 	FROM sui.events
 	WHERE id = $1
 	`
+
+	GetTransmitterCursor = `
+	SELECT cursor
+	FROM sui.transmitter_cursors
+	WHERE transmitter = $1
+	`
+
+	UpdateTransmitterCursor = `
+	INSERT INTO sui.transmitter_cursors (transmitter, cursor) VALUES ($1, $2)
+	ON CONFLICT (transmitter) DO UPDATE SET cursor = $2;
+	`
 )

relayer/chainreader/indexer/events_indexer.go

@@ -2,6 +2,7 @@ package indexer
 
 import (
 	"context"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"strconv"
@@ -10,6 +11,7 @@ import (
 	"time"
 
 	"github.com/block-vision/sui-go-sdk/models"
+	"github.com/mr-tron/base58"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
@@ -26,8 +28,9 @@ type EventsIndexer struct {
 	syncTimeout     time.Duration
 
 	// Protected by configMutex
-	eventConfigurations []*client.EventSelector
-	configMutex         sync.RWMutex
+	eventConfigurations  []*client.EventSelector
+	eventOffsetOverrides map[string]client.EventId
+	configMutex          sync.RWMutex
 
 	// Protected by cursorMutex
 	// a map of event handles to the last processed cursor
@@ -40,6 +43,7 @@ type EventsIndexerApi interface {
 	SyncAllEvents(ctx context.Context) error
 	SyncEvent(ctx context.Context, selector *client.EventSelector) error
 	AddEventSelector(ctx context.Context, selector *client.EventSelector) error
+	SetEventOffsetOverrides(ctx context.Context, offsetOverrides map[string]client.EventId) error
 	Ready() error
 	Close() error
 }
@@ -208,6 +212,7 @@ func (eIndexer *EventsIndexer) SyncEvent(ctx context.Context, selector *client.E
 		return fmt.Errorf("unspecified selector for SyncEvent call")
 	}
 
+	eventKey := fmt.Sprintf("%s::%s", selector.Module, selector.Event)
 	eventHandle := fmt.Sprintf("%s::%s::%s", selector.Package, selector.Module, selector.Event)
 
 	// check if the event selector is already tracked, if not add it to the list
@@ -227,16 +232,42 @@ func (eIndexer *EventsIndexer) SyncEvent(ctx context.Context, selector *client.E
 	cursor := eIndexer.lastProcessedCursors[eventHandle]
 
 	var totalCount uint64
-	var err error
+
 	if cursor == nil {
 		// attempt to get the latest event sync of the given type and use its data to construct a cursor
-		cursor, totalCount, err = eIndexer.db.GetLatestOffset(ctx, selector.Package, eventHandle)
-		if err != nil {
+		dbOffsetCursor, dbTotalCount, offsetErr := eIndexer.db.GetLatestOffset(ctx, selector.Package, eventHandle)
+		if offsetErr != nil {
 			eIndexer.cursorMutex.RUnlock()
-			return err
+			eIndexer.logger.Errorw("syncEvent: failed to get latest offset", "error", offsetErr)
+			return offsetErr
 		}
 
-		eIndexer.logger.Debugw("syncEvent: starting fresh sync", "handle", eventHandle, "cursor", cursor)
+		if dbOffsetCursor != nil {
+			txDigestBytes, err := hex.DecodeString(strings.TrimPrefix(dbOffsetCursor.TxDigest, "0x"))
+			if err != nil {
+				eIndexer.cursorMutex.RUnlock()
+				eIndexer.logger.Errorw("syncEvent: failed to decode tx digest", "error", err, "txDigest", dbOffsetCursor.TxDigest)
+				return err
+			}
+			// convert the db offset cursor digest from hex (the format stored in the DB) to base58 (the format expected by the client)
+			cursor = &models.EventId{
+				TxDigest: base58.Encode(txDigestBytes),
+				EventSeq: dbOffsetCursor.EventSeq,
+			}
+
+			totalCount = dbTotalCount
+		} else {
+			eIndexer.configMutex.RLock()
+			if override, ok := eIndexer.eventOffsetOverrides[eventKey]; ok {
+				cursor = &models.EventId{
+					TxDigest: override.TxDigest,
+					EventSeq: override.EventSeq,
+				}
+			} else {
+				eIndexer.logger.Debugw("syncEvent: starting fresh sync", "handle", eventHandle)
+			}
+			eIndexer.configMutex.RUnlock()
+		}
 	}
 
 	batchSize := uint(batchSizeRecords)
@@ -308,15 +339,29 @@ eventLoop:
 				// normalize the data, convert snake case to camel case
 				normalizedData := convertMapKeysToCamelCase(event.ParsedJson)
 
+				// Convert the txDigest to hex
+				txDigestHex := event.Id.TxDigest
+				if base58Bytes, err := base58.Decode(txDigestHex); err == nil {
+					hexTxId := hex.EncodeToString(base58Bytes)
+					txDigestHex = "0x" + hexTxId
+				}
+
+				blockHashBytes, err := base58.Decode(block.TxDigest)
+				if err != nil {
+					eIndexer.logger.Errorw("Failed to decode block hash", "error", err)
+					// fallback
+					blockHashBytes = []byte(block.TxDigest)
+				}
+
 				// Convert event to database record
 				record := database.EventRecord{
 					EventAccountAddress: selector.Package,
 					EventHandle:         eventHandle,
 					EventOffset:         offset,
-					TxDigest:            event.Id.TxDigest,
+					TxDigest:            txDigestHex,
 					BlockVersion:        0,
 					BlockHeight:         fmt.Sprintf("%d", block.Height),
-					BlockHash:           []byte(block.TxDigest),
+					BlockHash:           blockHashBytes,
 					// Sui returns block.Timestamp in ms; convert to seconds for consistency with CCIP readers.
 					BlockTimestamp: block.Timestamp / 1000,
 					Data:           normalizedData.(map[string]any),
@@ -409,6 +454,13 @@ func (eIndexer *EventsIndexer) AddEventSelector(ctx context.Context, selector *c
 	return nil
 }
 
+func (eIndexer *EventsIndexer) SetEventOffsetOverrides(ctx context.Context, offsetOverrides map[string]client.EventId) error {
+	eIndexer.configMutex.Lock()
+	defer eIndexer.configMutex.Unlock()
+	eIndexer.eventOffsetOverrides = offsetOverrides
+	return nil
+}
+
 // IsEventSelectorAdded checks if a specific event selector has already been included in the list of events to sync
 func (eIndexer *EventsIndexer) isEventSelectorAdded(eConfig client.EventSelector) bool {
 	eIndexer.configMutex.RLock()

relayer/chainreader/indexer/events_indexer_test.go

@@ -523,6 +523,36 @@ func TestEventsIndexer(t *testing.T) {
 		log.Infow("All concurrent access tests completed successfully")
 	})
 
+	t.Run("TestWithTimestamps", func(t *testing.T) {
+		log.Infow("Testing with timestamps")
+
+		// Trigger some events
+		for i := 1; i <= 3; i++ {
+			createEvent(i)
+		}
+
+		// Create a new event selector for timestamps
+		timestampEventSelector := &client.EventSelector{
+			Package: packageId,
+			Module:  "counter",
+			Event:   "CounterIncremented",
+		}
+
+		// Run sync to index events
+		err := indexer.SyncEvent(ctx, timestampEventSelector)
+		require.NoError(t, err)
+
+		// Wait for events to be indexed
+		events := waitForEventCount(3, 60*time.Second)
+		require.GreaterOrEqual(t, len(events), 3)
+
+		// Check that events are recorded with timestamps in seconds
+		for _, event := range events[:3] {
+			require.Greater(t, event.BlockTimestamp, uint64(0), "Event should have a timestamp")
+			require.Less(t, event.BlockTimestamp, uint64(time.Now().Unix()+1), "Event timestamp should be in the past")
+		}
+	})
+
 	t.Run("TestRaceDetection", func(t *testing.T) {
 		// Run with: go test -race -run TestEventsIndexer/TestRaceDetection
 		log.Infow("Starting race detection test")

relayer/chainreader/indexer/indexer.go

@@ -40,7 +40,7 @@ func NewIndexer(
 	transactionIndexer TransactionsIndexerApi,
 ) *Indexer {
 	return &Indexer{
-		log:                      logger.Named(l, "Indexers"),
+		log:                      logger.Named(l, "SuiIndexers"),
 		eventsIndexer:            eventsIndexer,
 		eventsIndexerCancel:      nil,
 		transactionIndexer:       transactionIndexer,

relayer/chainreader/indexer/transactions_indexer.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/block-vision/sui-go-sdk/models"
+	"github.com/mr-tron/base58"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/query"
@@ -65,12 +66,13 @@ func NewTransactionsIndexer(
 	syncTimeout time.Duration,
 	eventConfigs map[string]*config.ChainReaderEvent,
 ) TransactionsIndexerApi {
-	dataStore := database.NewDBStore(db, lggr)
+	logInstance := logger.Named(lggr, "SuiTransactionsIndexer")
+	dataStore := database.NewDBStore(db, logInstance)
 
 	return &TransactionsIndexer{
 		db:                      dataStore,
 		client:                  sdkClient,
-		logger:                  lggr,
+		logger:                  logInstance,
 		pollingInterval:         pollingInterval,
 		syncTimeout:             syncTimeout,
 		transmitters:            make(map[models.SuiAddress]string),
@@ -230,6 +232,8 @@ func (tIndexer *TransactionsIndexer) SyncAllTransmittersTransactions(ctx context
 		return nil
 	}
 
+	tIndexer.logger.Debugw("syncTransmittersTransactions start", "transmitters", transmitters)
+
 	var batchSize uint64 = 50
 	var totalProcessed int
 
@@ -269,6 +273,8 @@ func (tIndexer *TransactionsIndexer) syncTransmitterTransactions(ctx context.Con
 	cursor := tIndexer.transmitters[transmitter]
 	totalProcessed := 0
 
+	tIndexer.logger.Debugw("syncTransmitterTransactions start", "transmitter", transmitter, "cursor", cursor)
+
 	eventAccountAddress, latestOfframpPackageId, err := tIndexer.getEventPackageIdFromConfig()
 	if err != nil {
 		return 0, fmt.Errorf("failed to get ExecutionStateChanged event config: %w", err)
@@ -279,6 +285,21 @@ func (tIndexer *TransactionsIndexer) syncTransmitterTransactions(ctx context.Con
 	case <-ctx.Done():
 		return totalProcessed, ctx.Err()
 	default:
+		if cursor == "" {
+			// Get the cursor from the DB store
+			transmitterCursorFromDB, err := tIndexer.db.GetTransmitterCursor(ctx, transmitter)
+			if err != nil {
+				tIndexer.logger.Warnw("Failed to get transmitter cursor from DB store", "error", err)
+			}
+			// Attempt to check if a cursor exists in the DB store
+			if transmitterCursorFromDB != "" {
+				tIndexer.logger.Debugw("Found transmitter cursor in DB store", "transmitter", transmitter, "cursor", transmitterCursorFromDB)
+				cursor = transmitterCursorFromDB
+			} else {
+				tIndexer.logger.Debugw("No transmitter cursor found in DB store, starting fresh sync", "transmitter", transmitter)
+			}
+		}
+
 		queryResponse, err := tIndexer.client.QueryTransactions(ctx, string(transmitter), &cursor, &batchSize)
 		if err != nil {
 			return totalProcessed, fmt.Errorf("failed to fetch transactions for transmitter %s: %w", transmitter, err)
@@ -292,6 +313,12 @@ func (tIndexer *TransactionsIndexer) syncTransmitterTransactions(ctx context.Con
 		defer func() {
 			// Update the cursor to the last transaction digest regardless of the code path below
 			tIndexer.transmitters[transmitter] = lastDigest
+
+			// Update the cursor in the DB store
+			err := tIndexer.db.UpdateTransmitterCursor(ctx, transmitter, lastDigest)
+			if err != nil {
+				tIndexer.logger.Errorw("Failed to update transmitter cursor in DB store", "error", err)
+			}
 		}()
 
 		var records []database.EventRecord
@@ -358,6 +385,7 @@ func (tIndexer *TransactionsIndexer) syncTransmitterTransactions(ctx context.Con
 					"Expected PTB command (_::offramp::init_execute) not found in commands of failed PTB originating from known transmitter",
 					"transmitter", transmitter,
 					"digest", transactionRecord.Digest,
+					"transactionRecord", transactionRecord,
 				)
 				continue
 			}
@@ -473,13 +501,27 @@ func (tIndexer *TransactionsIndexer) syncTransmitterTransactions(ctx context.Con
 				continue
 			}
 
+			// Convert the txDigest to hex
+			txDigestHex := transactionRecord.Digest
+			if base64Bytes, err := base58.Decode(txDigestHex); err == nil {
+				hexTxId := hex.EncodeToString(base64Bytes)
+				txDigestHex = "0x" + hexTxId
+			}
+
+			blockHashBytes, err := base58.Decode(checkpointResponse.Digest)
+			if err != nil {
+				tIndexer.logger.Errorw("Failed to decode block hash", "error", err)
+				// fallback
+				blockHashBytes = []byte(checkpointResponse.Digest)
+			}
+
 			record := database.EventRecord{
 				EventAccountAddress: eventAccountAddress,
 				EventHandle:         eventHandle,
 				EventOffset:         0,
-				TxDigest:            transactionRecord.Digest,
+				TxDigest:            txDigestHex,
 				BlockHeight:         checkpointResponse.SequenceNumber,
-				BlockHash:           []byte(checkpointResponse.Digest),
+				BlockHash:           blockHashBytes,
 				BlockTimestamp:      blockTimestamp,
 				Data:                executionStateChanged,
 			}
@@ -596,7 +638,7 @@ func (tIndexer *TransactionsIndexer) getSourceChainConfig(ctx context.Context, s
 
 	filter := []query.Expression{
 		query.Comparator(selector,
-			primitives.ValueComparator{Value: sourceChainSelector, Operator: primitives.Eq},
+			primitives.ValueComparator{Value: strconv.FormatUint(sourceChainSelector, 10), Operator: primitives.Eq},
 		),
 	}