Update core configs plugin yaml + misc improvements

codec/decoder.go

@@ -13,7 +13,6 @@ import (
 
 	aptosBCS "github.com/aptos-labs/aptos-go-sdk/bcs"
 	"github.com/block-vision/sui-go-sdk/models"
-
 )
 
 const (

codec/encoder.go

@@ -9,7 +9,6 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
-
 )
 
 const (

relayer/chainreader/indexer/chain_poller.go

@@ -40,6 +40,7 @@ type SelectorProvider func() []*client.EventSelector
 // to the indexers.
 type ChainPoller struct {
 	client           client.SuiPTBClient
+	extendedClient   client.ExtendedPTBClient
 	logger           logger.Logger
 	config           config.ChainPollerConfig
 	eventsCh         chan CheckpointEventsBatch
@@ -67,6 +68,7 @@ func NewChainPoller(
 
 	return &ChainPoller{
 		client:           client,
+		extendedClient:   asExtendedPTBClient(client),
 		logger:           logger.Named(log, "ChainPoller"),
 		config:           cfg,
 		eventsCh:         make(chan CheckpointEventsBatch, bufferSize),
@@ -178,31 +180,65 @@ func (cp *ChainPoller) run(ctx context.Context) {
 
 // computeStartSequence calculates the starting checkpoint sequence number.
 func (cp *ChainPoller) computeStartSequence(ctx context.Context) (uint64, error) {
+	var startSeq uint64
+
 	// If StartCheckpointSequence is configured, use it directly
 	if cp.config.StartCheckpointSequence != nil {
-		return *cp.config.StartCheckpointSequence, nil
+		startSeq = *cp.config.StartCheckpointSequence
+	} else {
+		// Get the latest checkpoint
+		latestSeq, err := cp.getLatestCheckpointSequence(ctx)
+		if err != nil {
+			return 0, fmt.Errorf("failed to get latest checkpoint: %w", err)
+		}
+
+		cp.logger.Infow("Latest checkpoint fetched in chain poller", "sequence", latestSeq)
+
+		// If BackfillCheckpointCount is configured, start from latest - N
+		if cp.config.BackfillCheckpointCount != nil {
+			count := *cp.config.BackfillCheckpointCount
+			if latestSeq > count {
+				startSeq = latestSeq - count
+			}
+		} else {
+			// Default: start from latest (no backfill)
+			startSeq = latestSeq
+		}
 	}
 
-	// Get the latest checkpoint
-	latestSeq, err := cp.getLatestCheckpointSequence(ctx)
-	if err != nil {
-		return 0, fmt.Errorf("failed to get latest checkpoint: %w", err)
+	return cp.clampToProviderFloor(ctx, startSeq), nil
+}
+
+// clampToProviderFloor raises startSeq to the provider's lowest available checkpoint when history
+// has been pruned below the configured backfill/start point.
+func (cp *ChainPoller) clampToProviderFloor(ctx context.Context, startSeq uint64) uint64 {
+	if cp.extendedClient == nil {
+		cp.logger.Warnw("Failed to get provider checkpoint availability, using requested start sequence",
+			"startSequence", startSeq,
+			"error", errors.New("client does not implement ExtendedPTBClient"),
+		)
+		return startSeq
 	}
 
-	cp.logger.Infow("Latest checkpoint fetched in chain poller", "sequence", latestSeq)
+	info, err := cp.extendedClient.GetCheckpointAvailability(ctx)
+	if err != nil {
+		cp.logger.Warnw("Failed to get provider checkpoint availability, using requested start sequence",
+			"startSequence", startSeq,
+			"error", err,
+		)
+		return startSeq
+	}
 
-	// If BackfillCheckpointCount is configured, start from latest - N
-	if cp.config.BackfillCheckpointCount != nil {
-		count := *cp.config.BackfillCheckpointCount
-		if latestSeq > count {
-			return latestSeq - count, nil
-		}
-		// If latest < count, start from 0
-		return 0, nil
+	lowest := info.GetLowestAvailableCheckpoint()
+	if lowest > 0 && startSeq < lowest {
+		cp.logger.Warnw("Start sequence below provider history floor, clamping",
+			"requested", startSeq,
+			"lowestAvailable", lowest,
+		)
+		return lowest
 	}
 
-	// Default: start from latest (no backfill)
-	return latestSeq, nil
+	return startSeq
 }
 
 // getLatestCheckpointSequence fetches the latest checkpoint and returns its sequence number.
@@ -249,6 +285,11 @@ func (cp *ChainPoller) RescanRecent() {
 
 // catchUp processes checkpoints from startSeq to endSeq (inclusive).
 func (cp *ChainPoller) catchUp(ctx context.Context, startSeq, endSeq uint64) {
+	startSeq = cp.clampToProviderFloor(ctx, startSeq)
+	if startSeq > endSeq {
+		return
+	}
+
 	for seq := startSeq; seq <= endSeq; seq++ {
 		select {
 		case <-ctx.Done():
@@ -264,11 +305,22 @@ func (cp *ChainPoller) catchUp(ctx context.Context, startSeq, endSeq uint64) {
 						)
 						return
 					}
-					cp.logger.Warnw("Checkpoint not found during catch-up, skipping",
+
+					if lowest := cp.providerLowestAvailable(ctx); lowest > 0 && seq < lowest {
+						cp.logger.Errorw("Checkpoint below provider history floor during catch-up, jumping to lowest available",
+							"sequence", seq,
+							"lowestAvailable", lowest,
+							"error", err,
+						)
+						seq = lowest - 1
+						continue
+					}
+
+					cp.logger.Warnw("Checkpoint not found during catch-up, will retry on next poll",
 						"sequence", seq,
 						"error", err,
 					)
-					continue
+					return
 				}
 				cp.logger.Errorw("Failed to process checkpoint, will retry on next poll",
 					"sequence", seq,
@@ -281,6 +333,30 @@ func (cp *ChainPoller) catchUp(ctx context.Context, startSeq, endSeq uint64) {
 	}
 }
 
+func (cp *ChainPoller) providerLowestAvailable(ctx context.Context) uint64 {
+	if cp.extendedClient == nil {
+		cp.logger.Warnw("Failed to get provider checkpoint availability",
+			"error", errors.New("client does not implement ExtendedPTBClient"),
+		)
+		return 0
+	}
+
+	info, err := cp.extendedClient.GetCheckpointAvailability(ctx)
+	if err != nil {
+		cp.logger.Warnw("Failed to get provider checkpoint availability", "error", err)
+		return 0
+	}
+	return info.GetLowestAvailableCheckpoint()
+}
+
+func asExtendedPTBClient(suiClient client.SuiPTBClient) client.ExtendedPTBClient {
+	ext, ok := suiClient.(client.ExtendedPTBClient)
+	if !ok {
+		return nil
+	}
+	return ext
+}
+
 func isCheckpointNotFound(err error) bool {
 	for err != nil {
 		if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
@@ -418,6 +494,8 @@ func (cp *ChainPoller) filterEvents(meta CheckpointMeta, transactions []*suirpcv
 
 			// Check if event matches any selector
 			for _, sel := range selectors {
+				cp.logger.Debugw("Checking if event matches selector", "event", event.GetEventType())
+
 				if eventMatchesSelector(event, sel) {
 					item := CheckpointEventItem{
 						Event:      event,

relayer/chainreader/indexer/chain_poller_test.go

@@ -0,0 +1,114 @@
+package indexer
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	suirpcv2 "github.com/block-vision/sui-go-sdk/pb/sui/rpc/v2"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-sui/relayer/chainreader/config"
+	"github.com/smartcontractkit/chainlink-sui/relayer/client"
+	"github.com/smartcontractkit/chainlink-sui/relayer/testutils"
+)
+
+type checkpointTestClient struct {
+	testutils.FakeSuiPTBClient
+	lowestAvailable uint64
+	notFoundBelow   uint64
+	processed       []uint64
+}
+
+func (c *checkpointTestClient) GetCheckpointAvailability(ctx context.Context) (*suirpcv2.GetServiceInfoResponse, error) {
+	lowest := c.lowestAvailable
+	return &suirpcv2.GetServiceInfoResponse{
+		LowestAvailableCheckpoint: &lowest,
+	}, nil
+}
+
+func (c *checkpointTestClient) GetTransaction(ctx context.Context, digest string) (client.TransactionDetails, error) {
+	return client.TransactionDetails{}, nil
+}
+
+func (c *checkpointTestClient) GetCheckpointData(ctx context.Context, checkpointSequenceNumber uint64) (*client.CheckpointData, error) {
+	if checkpointSequenceNumber < c.notFoundBelow {
+		return nil, status.Error(codes.NotFound, "checkpoint not found")
+	}
+
+	c.processed = append(c.processed, checkpointSequenceNumber)
+	seq := checkpointSequenceNumber
+	return &client.CheckpointData{
+		Checkpoint: &suirpcv2.Checkpoint{
+			SequenceNumber: &seq,
+		},
+	}, nil
+}
+
+func TestClampToProviderFloor(t *testing.T) {
+	t.Parallel()
+
+	mockClient := &checkpointTestClient{
+		lowestAvailable: 500,
+	}
+	cp := &ChainPoller{
+		client:         mockClient,
+		extendedClient: asExtendedPTBClient(mockClient),
+		logger:         logger.Test(t),
+	}
+
+	require.Equal(t, uint64(500), cp.clampToProviderFloor(context.Background(), 100))
+	require.Equal(t, uint64(600), cp.clampToProviderFloor(context.Background(), 600))
+}
+
+func TestCatchUpJumpsOverPrunedCheckpoints(t *testing.T) {
+	t.Parallel()
+
+	mockClient := &checkpointTestClient{
+		lowestAvailable: 500,
+		notFoundBelow:   500,
+	}
+
+	cp := NewChainPoller(mockClient, logger.Test(t), config.ChainPollerConfig{
+		SyncTimeout: time.Minute,
+	}, func() []*client.EventSelector {
+		return nil
+	})
+
+	cp.catchUp(context.Background(), 400, 502)
+
+	require.Equal(t, []uint64{500, 501, 502}, mockClient.processed)
+	require.Equal(t, uint64(502), cp.lastProcessed)
+}
+
+func TestCatchUpRetriesInRangeNotFound(t *testing.T) {
+	t.Parallel()
+
+	mockClient := &checkpointTestClient{
+		lowestAvailable: 100,
+		notFoundBelow:   501,
+	}
+
+	cp := NewChainPoller(mockClient, logger.Test(t), config.ChainPollerConfig{
+		SyncTimeout: time.Minute,
+	}, func() []*client.EventSelector {
+		return nil
+	})
+
+	cp.catchUp(context.Background(), 500, 502)
+
+	require.Empty(t, mockClient.processed)
+	require.Equal(t, uint64(0), cp.lastProcessed)
+}
+
+func TestIsCheckpointNotFound(t *testing.T) {
+	t.Parallel()
+
+	require.True(t, isCheckpointNotFound(status.Error(codes.NotFound, "missing")))
+	require.True(t, isCheckpointNotFound(errors.New("wrapped: not found")))
+	require.False(t, isCheckpointNotFound(errors.New("timeout")))
+}

relayer/chainreader/indexer/events_indexer.go

@@ -205,14 +205,9 @@ func (eIndexer *EventsIndexer) processEventsForHandle(
 	}
 	packageID := parts[0]
 
-	totalCount, err := eIndexer.db.GetTotalCount(ctx, packageID, handle)
-	if err != nil {
-		return fmt.Errorf("failed to get total count: %w", err)
-	}
-
 	// Build event records
 	records := make([]database.EventRecord, 0, len(items))
-	for i, item := range items {
+	for _, item := range items {
 		// Convert event JSON data using SDK's AsInterface()
 		data := make(map[string]any)
 		if jsonVal := item.Event.GetJson(); jsonVal != nil {
@@ -238,9 +233,6 @@ func (eIndexer *EventsIndexer) processEventsForHandle(
 			}
 		}
 
-		// Calculate offset (totalCount + i + 1)
-		offset := totalCount + uint64(i) + 1
-
 		// Convert block hash to bytes
 		blockHashBytes := []byte(meta.Digest)
 		if decoded, err := base58.Decode(meta.Digest); err == nil {
@@ -250,7 +242,7 @@ func (eIndexer *EventsIndexer) processEventsForHandle(
 		record := database.EventRecord{
 			EventAccountAddress: packageID,
 			EventHandle:         handle,
-			EventOffset:         offset,
+			EventOffset:         uint64(item.EventIndex),
 			TxDigest:            txDigestHex,
 			BlockVersion:        0,
 			BlockHeight:         strconv.FormatUint(meta.SequenceNumber, 10),
@@ -260,6 +252,8 @@ func (eIndexer *EventsIndexer) processEventsForHandle(
 			IsSynthetic:         false,
 		}
 		records = append(records, record)
+
+		eIndexer.logger.Debugw("Prepared event record to insert", "event", record)
 	}
 
 	// Batch insert with fallback

relayer/chainreader/reader/chainreader.go

@@ -1039,6 +1039,12 @@ func (s *suiChainReader) executeFunction(ctx context.Context, parsed *readIdenti
 	if err != nil {
 		return []any{}, err
 	}
+	// Guard against an empty package ID: simulating against it would resolve to the zero package
+	// (0x0) and fail with a confusing "Dependent package not found on-chain" error. Fail loudly
+	// so the read is retried instead of silently targeting 0x0.
+	if latestPackageId == "" {
+		return []any{}, fmt.Errorf("resolved empty latest package id for contract %s (module %s)", parsed.contractName, common.GetModuleForContract(parsed.contractName))
+	}
 
 	// this is the upgraded pkgID
 	parsed.address = latestPackageId