Non-E2E Contract Reader Bug Fixes

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/prometheus/client_golang v1.20.5
 	github.com/smartcontractkit/chainlink-ccip v0.0.0-20250203132120-f0d42463e405
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3
-	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb
+	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad
 	github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250211162441-3d6cea220efb
 	github.com/smartcontractkit/libocr v0.0.0-20241223215956-e5b78d8e3919
 	github.com/stretchr/testify v1.10.0

go.sum

@@ -584,8 +584,8 @@ github.com/smartcontractkit/chainlink-ccip v0.0.0-20250203132120-f0d42463e405 h1
 github.com/smartcontractkit/chainlink-ccip v0.0.0-20250203132120-f0d42463e405/go.mod h1:UEnHaxkUsfreeA7rR45LMmua1Uen95tOFUR8/AI9BAo=
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3 h1:f4F/7OCuMybsPKKXXvLQz+Q1hGq07I1cfoWy5EA9iRg=
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3/go.mod h1:Bmwq4lNb5tE47sydN0TKetcLEGbgl+VxHEWp4S0LI60=
-github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb h1:1VC/hN1ojPiEWCsjxhvcw4p1Zveo90O38VQhktvo3Ag=
-github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb/go.mod h1:Z2e1ynSJ4pg83b4Qldbmryc5lmnrI3ojOdg1FUloa68=
+github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad h1:n7vQtigsgTKIhrN0FxfO9PpeqOVHNn9GBkQijwjPFBU=
+github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad/go.mod h1:Z2e1ynSJ4pg83b4Qldbmryc5lmnrI3ojOdg1FUloa68=
 github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250211162441-3d6cea220efb h1:LWijSyJ2lhppkFLN19EGsLHZXQ5wen2DEk1cyR0tV+o=
 github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250211162441-3d6cea220efb/go.mod h1:4JqpgFy01LaqG1yM2iFTzwX3ZgcAvW9WdstBZQgPHzU=
 github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7 h1:12ijqMM9tvYVEm+nR826WsrNi6zCKpwBhuApq127wHs=

integration-tests/go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/rs/zerolog v1.33.0
 	github.com/smartcontractkit/chainlink-ccip v0.0.0-20250212131315-e9b53b05b02a
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3
-	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb
+	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad
 	github.com/smartcontractkit/chainlink-solana v1.1.2-0.20250213035259-e727e73f6181
 	github.com/smartcontractkit/chainlink-testing-framework/lib v1.51.0
 	github.com/smartcontractkit/chainlink-testing-framework/seth v1.50.10

integration-tests/go.sum

@@ -1104,8 +1104,8 @@ github.com/smartcontractkit/chainlink-ccip v0.0.0-20250212131315-e9b53b05b02a h1
 github.com/smartcontractkit/chainlink-ccip v0.0.0-20250212131315-e9b53b05b02a/go.mod h1:Hht/OJq/PxC+gnBCIPyzHt4Otsw6mYwUVsmtOqIvlxo=
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3 h1:f4F/7OCuMybsPKKXXvLQz+Q1hGq07I1cfoWy5EA9iRg=
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250206215114-fb6c3c35e8e3/go.mod h1:Bmwq4lNb5tE47sydN0TKetcLEGbgl+VxHEWp4S0LI60=
-github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb h1:1VC/hN1ojPiEWCsjxhvcw4p1Zveo90O38VQhktvo3Ag=
-github.com/smartcontractkit/chainlink-common v0.4.2-0.20250205141137-8f50d72601bb/go.mod h1:Z2e1ynSJ4pg83b4Qldbmryc5lmnrI3ojOdg1FUloa68=
+github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad h1:n7vQtigsgTKIhrN0FxfO9PpeqOVHNn9GBkQijwjPFBU=
+github.com/smartcontractkit/chainlink-common v0.4.2-0.20250221044241-800d6b475fad/go.mod h1:Z2e1ynSJ4pg83b4Qldbmryc5lmnrI3ojOdg1FUloa68=
 github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250128203428-08031923fbe5 h1:CvDfgWoLoYPapOumE/UZCplfCu5oNmy9BuH+6V6+fJ8=
 github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250128203428-08031923fbe5/go.mod h1:pDZagSGjs9U+l4YIFhveDznMHqxuuz+5vRxvVgpbdr8=
 github.com/smartcontractkit/chainlink-feeds v0.1.1 h1:JzvUOM/OgGQA1sOqTXXl52R6AnNt+Wg64sVG+XSA49c=

pkg/solana/chainreader/batch.go

@@ -81,6 +81,8 @@ func doMethodBatchCall(ctx context.Context, lggr logger.Logger, client MultipleA
 
 	// map batch call index to key index (some calls are event reads and will be handled by a different binding)
 	dataMap := make(map[int]int)
+	eventMap := make(map[int]struct{})
+
 	for idx, batchCall := range batch {
 		rBinding, err := bdRegistry.GetReader(batchCall.Namespace, batchCall.ReadName)
 		if err != nil {
@@ -102,6 +104,7 @@ func doMethodBatchCall(ctx context.Context, lggr logger.Logger, client MultipleA
 			}
 
 			results[idx].err = eBinding.GetLatestValue(ctx, batchCall.Params, results[idx].returnVal)
+			eventMap[idx] = struct{}{}
 
 			continue
 		}
@@ -120,6 +123,10 @@ func doMethodBatchCall(ctx context.Context, lggr logger.Logger, client MultipleA
 
 	// decode batch call results
 	for idx, batchCall := range batch {
+		if _, ok := eventMap[idx]; ok {
+			continue
+		}
+
 		dataIdx, ok := dataMap[idx]
 		if !ok {
 			return nil, fmt.Errorf("%w: unexpected data index state", types.ErrInternal)

pkg/solana/chainreader/chain_reader.go

@@ -215,6 +215,10 @@ func (s *ContractReaderService) GetLatestValue(ctx context.Context, readIdentifi
 	}
 
 	if results[0].err != nil {
+		if errors.Is(results[0].err, types.ErrNotFound) {
+			return types.ErrNotFound
+		}
+
 		return fmt.Errorf("%w: %s", types.ErrInternal, results[0].err)
 	}
 

pkg/solana/chainreader/event_read_binding.go

@@ -300,7 +300,7 @@ func (b *eventReadBinding) extractFilterSubkeys(offChainParams any) ([]query.Exp
 
 		fieldVal, err := valueForPath(reflect.ValueOf(offChainParams), offChainKey)
 		if err != nil {
-			return nil, fmt.Errorf("%w: no value for path %s", types.ErrInternal, b.genericName+"."+offChainKey)
+			return nil, fmt.Errorf("%w: no value for path %s; err: %w", types.ErrInternal, b.genericName+"."+offChainKey, err)
 		}
 
 		onChainValue, err := b.modifier.TransformToOnChain(fieldVal, itemType)
@@ -353,7 +353,7 @@ func (b *eventReadBinding) encodeComparator(comparator *primitives.Comparator) (
 		return query.Expression{}, fmt.Errorf("%w: unknown indexed subkey mapping %s", types.ErrInvalidConfig, comparator.Name)
 	}
 
-	itemType := strings.Join([]string{b.namespace, b.genericName, comparator.Name}, ".")
+	itemType := codec.WrapItemType(true, b.namespace, b.genericName+"."+comparator.Name)
 
 	for idx, comp := range comparator.ValueComparators {
 		// need to do a transform and then extract the value for the subkey
@@ -362,7 +362,7 @@ func (b *eventReadBinding) encodeComparator(comparator *primitives.Comparator) (
 			return query.Expression{}, err
 		}
 
-		comparator.ValueComparators[idx].Value = newValue
+		comparator.ValueComparators[idx].Value = reflect.Indirect(reflect.ValueOf(newValue)).Interface()
 	}
 
 	return logpoller.NewEventBySubKeyFilter(subKeyIndex, comparator.ValueComparators)
@@ -521,6 +521,10 @@ func valueForPath(from reflect.Value, itemType string) (any, error) {
 
 	switch from.Kind() {
 	case reflect.Pointer:
+		if from.IsNil() {
+			from = reflect.New(from.Type().Elem())
+		}
+
 		elem, err := valueForPath(from.Elem(), itemType)
 		if err != nil {
 			return nil, err

pkg/solana/codec/byte_string_modifier.go

@@ -27,9 +27,9 @@ func (s SolanaAddressModifier) DecodeAddress(str string) ([]byte, error) {
 		return nil, fmt.Errorf("%w: failed to decode Base58 address: %s", commontypes.ErrInvalidType, err)
 	}
 
-	if !pubkey.IsOnCurve() {
-		return nil, fmt.Errorf("%w: address %q with length of %d is not on the ed25519 curve", commontypes.ErrInvalidType, str, len(str))
-	}
+	// PDAs are used extensively and do not exist on the ed25519 curve
+	// for this reason we ignore the IsOnCurve check and rely on the user to verify that the public key
+	// is on the curve or not if intended.
 
 	return pubkey.Bytes(), nil
 }

pkg/solana/codec/byte_string_modifier_test.go

@@ -17,7 +17,9 @@ func TestSolanaAddressModifier(t *testing.T) {
 
 	// Valid Solana address (32 bytes, Base58 encoded)
 	validAddressStr := "9nQhQ7iCyY5SgAX2Zm4DtxNh9Ubc4vbiLkiYbX43SDXY"
+	addressNotOnCurveStr := "8opHzTAnfzRpPEx21XtnrVTX28YQuCpAjcn1PczScKh"
 	validAddressBytes := solana.MustPublicKeyFromBase58(validAddressStr).Bytes()
+	addressNotOnCurveBytes := solana.MustPublicKeyFromBase58(addressNotOnCurveStr).Bytes()
 	invalidLengthAddressStr := "abc123"
 
 	t.Run("EncodeAddress encodes valid Solana address bytes", func(t *testing.T) {
@@ -58,10 +60,11 @@ func TestSolanaAddressModifier(t *testing.T) {
 		assert.Contains(t, err.Error(), commontypes.ErrInvalidType.Error())
 	})
 
-	t.Run("DecodeAddress returns error for valid length address not on the ed25519 curve", func(t *testing.T) {
-		_, err := modifier.DecodeAddress("AddressLookupTab1e11111111111111111111111111")
-		assert.Error(t, err)
-		assert.Contains(t, err.Error(), commontypes.ErrInvalidType.Error())
+	t.Run("DecodeAddress decodes address not on ed25519 curve", func(t *testing.T) {
+		decodedBytes, err := modifier.DecodeAddress(addressNotOnCurveStr)
+
+		require.NoError(t, err)
+		assert.Equal(t, addressNotOnCurveBytes, decodedBytes)
 	})
 
 	t.Run("Length returns 32 for Solana addresses", func(t *testing.T) {

pkg/solana/logpoller/log_poller.go

@@ -468,7 +468,7 @@ func (lp *Service) replayComplete(from, to int64) bool {
 	return true
 }
 
-func appendBuffered(ch <-chan Block, max int, blocks []Block) []Block {
+func appendBuffered(ch <-chan Block, maxNum int, blocks []Block) []Block {
 	for {
 		select {
 		case block, ok := <-ch:
@@ -477,7 +477,7 @@ func appendBuffered(ch <-chan Block, max int, blocks []Block) []Block {
 			}
 
 			blocks = append(blocks, block)
-			if len(blocks) >= max {
+			if len(blocks) >= maxNum {
 				return blocks
 			}
 		default:

pkg/solana/logpoller/parser.go

@@ -35,9 +35,8 @@ var (
 	ErrInvalidSortDir      = errors.New("invalid sort direction")
 	ErrInvalidSortType     = errors.New("invalid sort by type")
 
-	logsFields = [...]string{"id", "filter_id", "chain_id", "log_index", "block_hash", "block_number",
-		"block_timestamp", "address", "event_sig", "subkey_values", "tx_hash", "data", "created_at",
-		"expires_at", "sequence_num"}
+	logsFields = [...]string{"chain_id", "log_index", "block_hash", "block_number", "block_timestamp", "address",
+		"event_sig", "tx_hash", "data"}
 
 	filterFields = [...]string{"id", "name", "address", "event_name", "event_sig", "starting_block",
 		"event_idl", "subkey_paths", "retention", "max_logs_kept", "is_deleted", "is_backfilled"}
@@ -231,7 +230,7 @@ func (v *pgDSLParser) whereClause(expressions []query.Expression, limiter query.
 			return "", ErrInvalidCursorDir
 		}
 
-		block, logIdx, _, err := valuesFromCursor(limiter.Limit.Cursor)
+		block, logIdx, err := valuesFromCursor(limiter.Limit.Cursor)
 		if err != nil {
 			return "", err
 		}
@@ -374,30 +373,25 @@ func cmpOpToString(op primitives.ComparisonOperator) (string, error) {
 }
 
 // ensure valuesFromCursor remains consistent with the function above that creates a cursor
-func valuesFromCursor(cursor string) (int64, int, []byte, error) {
+func valuesFromCursor(cursor string) (int64, int64, error) {
 	partCount := 3
 
 	parts := strings.Split(cursor, "-")
 	if len(parts) != partCount {
-		return 0, 0, nil, fmt.Errorf("%w: must be composed as block-logindex-txHash", ErrInvalidCursorFormat)
+		return 0, 0, fmt.Errorf("%w: must be composed as block-logindex-txHash", ErrInvalidCursorFormat)
 	}
 
 	block, err := strconv.ParseInt(parts[0], 10, 64)
 	if err != nil {
-		return 0, 0, nil, fmt.Errorf("%w: block number not parsable as int64", ErrInvalidCursorFormat)
+		return 0, 0, fmt.Errorf("%w: block number not parsable as int64", ErrInvalidCursorFormat)
 	}
 
-	logIdx, err := strconv.ParseInt(parts[1], 10, 32)
+	logIdx, err := strconv.ParseInt(parts[1], 10, 64)
 	if err != nil {
-		return 0, 0, nil, fmt.Errorf("%w: log index not parsable as int", ErrInvalidCursorFormat)
+		return 0, 0, fmt.Errorf("%w: log index not parsable as int64", ErrInvalidCursorFormat)
 	}
 
-	txHash, err := solana.PublicKeyFromBase58(parts[2])
-	if err != nil {
-		return 0, 0, nil, fmt.Errorf("%w: invalid transaction hash: %s", ErrInvalidCursorFormat, err.Error())
-	}
-
-	return block, int(logIdx), txHash.Bytes(), nil
+	return block, logIdx, nil
 }
 
 func orderToString(dir query.SortDirection) (string, error) {
@@ -480,7 +474,7 @@ func (f *eventBySubKeyFilter) Accept(visitor primitives.Visitor) {
 
 // FormatContractReaderCursor is exported to ensure cursor structure remains consistent.
 func FormatContractReaderCursor(log Log) string {
-	return fmt.Sprintf("%d-%d-%s", log.BlockNumber, log.LogIndex, log.TxHash)
+	return fmt.Sprintf("%d-%d-%s", log.BlockNumber, log.LogIndex, log.TxHash.ToSolana().String())
 }
 
 func makeComp(comp IndexedValueComparator, args *queryArgs, field, subfield, pattern string) (string, error) {

pkg/solana/logpoller/query.go

@@ -125,7 +125,8 @@ func (q *queryArgs) withIsBackfilled(isBackfilled bool) *queryArgs {
 }
 
 func logsQuery(clause string) string {
-	return fmt.Sprintf(`SELECT %s FROM solana.logs %s`, strings.Join(logsFields[:], ", "), clause)
+	// TODO: using DISTINCT in a query is less efficient but required because of duplicate logs coming from multipler filters
+	return fmt.Sprintf(`SELECT DISTINCT %s FROM solana.logs %s`, strings.Join(logsFields[:], ", "), clause)
 }
 
 func filtersQuery(clause string) string {