diff --git a/chain_capabilities/solana/actions/actions.go b/chain_capabilities/solana/actions/actions.go index d5b9a0036..81ad0668c 100644 --- a/chain_capabilities/solana/actions/actions.go +++ b/chain_capabilities/solana/actions/actions.go @@ -1,26 +1,36 @@ package actions import ( + "bytes" "context" "errors" "fmt" + "math/big" + "slices" "strings" "time" + "google.golang.org/protobuf/proto" + "github.com/smartcontractkit/capabilities/chain_capabilities/solana/metering" "github.com/smartcontractkit/capabilities/libs/chainconsensus" ctypes "github.com/smartcontractkit/capabilities/libs/chainconsensus/types" + commonMon "github.com/smartcontractkit/capabilities/libs/monitoring" "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/capabilities" caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" solcap "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/solana" + capmon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/monitoring" commoncfg "github.com/smartcontractkit/chainlink-common/pkg/config" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink-common/pkg/types" + soltypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/solana" "github.com/smartcontractkit/chainlink-framework/multinode" + valuespb "github.com/smartcontractkit/chainlink-protos/cre/go/values/pb" capcommon "github.com/smartcontractkit/capabilities/chain_capabilities/common" ts "github.com/smartcontractkit/capabilities/chain_capabilities/common/transmission_schedule" @@ -37,6 +47,8 @@ type Solana struct { chainSelector uint64 txComputeLimit limits.BoundLimiter[uint32] reportSizeLimit limits.BoundLimiter[commoncfg.Size] + readPayloadSizeLimiter limits.BoundLimiter[commoncfg.Size] + batchItemLimit limits.BoundLimiter[int] beholderProcessor beholder.ProtoProcessor messageBuilder *monitoring.MessageBuilder transmissionScheduler ts.TransmissionScheduler @@ -76,15 +88,13 @@ func (s *Solana) GetAccountInfoWithOpts( if !s.readsEnabled { return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) } - // TODO: implement metrics on higher level PLEX-2918 // TODO: implement billing once generalized PLEX-3022 request, err := solcap.ConvertGetAccountInfoRequestFromProto(input) if err != nil { return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) } - + request.IsExternal = true lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) - lggr.Debugw("Received GetAccountInfoWithOpts request") cReq := ctypes.NewVolatileRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetAccountInfoWithOptsReply, uint64, error) { rawResponse, err := s.SolanaService.GetAccountInfoWithOpts(ctx, request) if err != nil { @@ -96,7 +106,9 @@ func (s *Solana) GetAccountInfoWithOpts( return nil, 0, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) } - // TODO PLEX-3061: limit response size + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, 0, NewUserError(err) + } return response, rawResponse.Slot, nil }, lggr) responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetAccountInfoWithOptsReply](ctx, s.handler, cReq) @@ -104,89 +116,321 @@ func (s *Solana) GetAccountInfoWithOpts( return nil, getReadError(lggr, fmt.Errorf("failed to GetAccountInfoWithOpts: %w", err)) } - lggr.Debugw("Successfully handled GetAccountInfoWithOpts") return responseAndMetadata, nil } -func getReadError(lggr logger.SugaredLogger, err error) caperrors.Error { - if err == nil { - return nil +func (s *Solana) GetBalance( + ctx context.Context, + metadata capabilities.RequestMetadata, + input *solcap.GetBalanceRequest) (*capabilities.ResponseAndMetadata[*solcap.GetBalanceReply], caperrors.Error) { + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) } - - isUserErr := isUserError(err) - capErr := GetError(err, isUserErr) - - // TODO: logging of init, success and error should be move to a higher level - lggr = lggr.With("error", err) - const msg = "Read operation failed" - if isUserErr { - lggr.Debug(msg) - } else { - lggr.Error(msg) + request, err := solcap.ConvertGetBalanceRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) } - return capErr -} + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewVolatileRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetBalanceReply, uint64, error) { + rawResponse, err := s.SolanaService.GetBalance(ctx, request) + if err != nil { + return nil, 0, err + } -func isUserError(err error) bool { - return !errors.Is(err, context.DeadlineExceeded) && !isNodeInfraError(err) -} + response, err := solcap.ConvertGetBalanceReplyToProto(rawResponse) + if err != nil { + return nil, 0, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } -func isNodeInfraError(err error) bool { - return errors.Is(err, multinode.ErrNodeError) || - strings.Contains(err.Error(), multinode.ErrNodeError.Error()) -} + return response, 0, nil + }, lggr) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetBalanceReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetBalance: %w", err)) + } -func (s *Solana) GetBalance( - ctx context.Context, - metadata capabilities.RequestMetadata, - input *solcap.GetBalanceRequest) (*capabilities.ResponseAndMetadata[*solcap.GetBalanceReply], caperrors.Error) { - // TODO - return nil, GetError(errors.New("unimplemented"), false) + return responseAndMetadata, nil } func (s *Solana) GetBlock( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetBlockRequest) (*capabilities.ResponseAndMetadata[*solcap.GetBlockReply], caperrors.Error) { - // TODO - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetBlockRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewECHashableRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetBlockReply, error) { + rawResponse, err := s.SolanaService.GetBlock(ctx, *request) + if err != nil { + return nil, err + } + + response, err := solcap.ConvertGetBlockReplyToProto(rawResponse) + if err != nil { + return nil, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } + + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, NewUserError(err) + } + return response, nil + }) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetBlockReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetBlock: %w", err)) + } + + return responseAndMetadata, nil } func (s *Solana) GetFeeForMessage( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetFeeForMessageRequest) (*capabilities.ResponseAndMetadata[*solcap.GetFeeForMessageReply], caperrors.Error) { - // TODO - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetFeeForMessageRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewAggregatableRequest(commonMon.RequestID(metadata.WorkflowExecutionID, metadata.ReferenceID), func(ctx context.Context) (*ctypes.AggregatableObservation, error) { + rawResponse, err := s.SolanaService.GetFeeForMessage(ctx, *request) + if err != nil { + return nil, err + } + + return &ctypes.AggregatableObservation{ + Method: ctypes.AggregationMethodFPlusOneHighest, + Value: &valuespb.Decimal{ + Coefficient: valuespb.NewBigIntFromInt(new(big.Int).SetUint64(rawResponse.Fee)), + Exponent: 0, + }, + }, nil + }) + aggregatedFee, err := chainconsensus.ReadDecimal(ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetFeeForMessage: %w", err)) + } + + return &capabilities.ResponseAndMetadata[*solcap.GetFeeForMessageReply]{ + Response: &solcap.GetFeeForMessageReply{Fee: aggregatedFee.BigInt().Uint64()}, + ResponseMetadata: metering.GetResponseMetadata(metering.GetAccountInfo), + }, nil } func (s *Solana) GetMultipleAccountsWithOpts( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetMultipleAccountsWithOptsRequest) (*capabilities.ResponseAndMetadata[*solcap.GetMultipleAccountsWithOptsReply], caperrors.Error) { - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetMultipleAccountsRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + if err := s.validateBatchItemCount(ctx, len(input.GetAccounts())); err != nil { + return nil, NewUserError(err) + } + request.IsExternal = true + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewVolatileRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetMultipleAccountsWithOptsReply, uint64, error) { + rawResponse, err := s.SolanaService.GetMultipleAccountsWithOpts(ctx, *request) + if err != nil { + return nil, 0, err + } + + response, err := solcap.ConvertGetMultipleAccountsReplyToProto(rawResponse) + if err != nil { + return nil, 0, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } + + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, 0, NewUserError(err) + } + return response, rawResponse.Slot, nil + }, lggr) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetMultipleAccountsWithOptsReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetMultipleAccountsWithOpts: %w", err)) + } + + return responseAndMetadata, nil } func (s *Solana) GetSignatureStatuses( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetSignatureStatusesRequest) (*capabilities.ResponseAndMetadata[*solcap.GetSignatureStatusesReply], caperrors.Error) { - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetSignatureStatusesRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + if err := s.validateBatchItemCount(ctx, len(input.GetSigs())); err != nil { + return nil, NewUserError(err) + } + + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewECHashableRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetSignatureStatusesReply, error) { + rawResponse, err := s.SolanaService.GetSignatureStatuses(ctx, *request) + if err != nil { + return nil, err + } + + response, err := solcap.ConvertGetSignatureStatusesReplyToProto(rawResponse) + if err != nil { + return nil, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } + + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, NewUserError(err) + } + return response, nil + }) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetSignatureStatusesReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetSignatureStatuses: %w", err)) + } + + return responseAndMetadata, nil } func (s *Solana) GetSlotHeight( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetSlotHeightRequest) (*capabilities.ResponseAndMetadata[*solcap.GetSlotHeightReply], caperrors.Error) { - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetSlotHeightRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewAggregatableRequest(commonMon.RequestID(metadata.WorkflowExecutionID, metadata.ReferenceID), func(ctx context.Context) (*ctypes.AggregatableObservation, error) { + rawResponse, err := s.SolanaService.GetSlotHeight(ctx, request) + if err != nil { + return nil, err + } + + return &ctypes.AggregatableObservation{ + Method: ctypes.AggregationMethodFPlusOneHighest, + Value: &valuespb.Decimal{ + Coefficient: valuespb.NewBigIntFromInt(new(big.Int).SetUint64(rawResponse.Height)), + Exponent: 0, + }, + }, nil + }) + aggregatedHeight, err := chainconsensus.ReadDecimal(ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetSlotHeight: %w", err)) + } + + return &capabilities.ResponseAndMetadata[*solcap.GetSlotHeightReply]{ + Response: &solcap.GetSlotHeightReply{Height: aggregatedHeight.BigInt().Uint64()}, + ResponseMetadata: metering.GetResponseMetadata(metering.GetAccountInfo), + }, nil } func (s *Solana) GetTransaction( ctx context.Context, metadata capabilities.RequestMetadata, input *solcap.GetTransactionRequest) (*capabilities.ResponseAndMetadata[*solcap.GetTransactionReply], caperrors.Error) { - return nil, GetError(errors.New("unimplemented"), false) + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetTransactionRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + request.IsExternal = true + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewECHashableRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetTransactionReply, error) { + rawResponse, err := s.SolanaService.GetTransaction(ctx, request) + if err != nil { + return nil, err + } + + response, err := solcap.ConvertGetTransactionReplyToProto(rawResponse) + if err != nil { + return nil, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } + + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, NewUserError(err) + } + return response, nil + }) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetTransactionReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetTransaction: %w", err)) + } + + return responseAndMetadata, nil +} + +func (s *Solana) GetProgramAccounts( + ctx context.Context, + metadata capabilities.RequestMetadata, + input *solcap.GetProgramAccountsRequest) (*capabilities.ResponseAndMetadata[*solcap.GetProgramAccountsReply], caperrors.Error) { + if !s.readsEnabled { + return nil, caperrors.NewPublicSystemError(errors.New("reads are not available"), caperrors.Internal) + } + request, err := solcap.ConvertGetProgramAccountsRequestFromProto(input) + if err != nil { + return nil, NewUserError(fmt.Errorf("invalid request: %w", err)) + } + request.IsExternal = true + lggr := s.messageBuilder.RequestLggr(s.lggr, monitoring.TelemetryContext{TsStart: time.Now().UnixMilli(), RequestMetadata: metadata}).With("request", request) + cReq := ctypes.NewVolatileRequest(metadata.WorkflowExecutionID, metadata.ReferenceID, metering.GetResponseMetadata(metering.GetAccountInfo), func(ctx context.Context) (*solcap.GetProgramAccountsReply, uint64, error) { + rawResponse, err := s.SolanaService.GetProgramAccounts(ctx, *request) + if err != nil { + return nil, 0, err + } + + // getProgramAccounts does not guarantee ordering across RPC nodes. + // Sort by pubkey so all nodes produce an identical hash. + slices.SortFunc(rawResponse.Value, func(a, b *soltypes.KeyedAccount) int { + return bytes.Compare(a.Pubkey[:], b.Pubkey[:]) + }) + + response, err := solcap.ConvertGetProgramAccountsReplyToProto(rawResponse) + if err != nil { + return nil, 0, caperrors.NewPublicSystemError(fmt.Errorf("failed to convert response to proto: %w", err), caperrors.Internal) + } + + if err = s.checkReadPayloadSize(ctx, response); err != nil { + return nil, 0, NewUserError(err) + } + return response, 0, nil + }, lggr) + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport[*solcap.GetProgramAccountsReply](ctx, s.handler, cReq) + if err != nil { + return nil, getReadError(lggr, fmt.Errorf("failed to GetProgramAccounts: %w", err)) + } + + return responseAndMetadata, nil +} + +func (s *Solana) MonitoringContext() capmon.MonitoringContext { + return capmon.MonitoringContext{ + Logger: s.lggr, + MetricsAttributes: s.messageBuilder.CapabilityMetricsAttributes, + } } func (s *Solana) initLimiters(limitsFactory limits.Factory) (err error) { @@ -200,8 +444,60 @@ func (s *Solana) initLimiters(limitsFactory limits.Factory) (err error) { if err != nil { return } + + s.readPayloadSizeLimiter, err = limits.MakeUpperBoundLimiter(limitsFactory, cresettings.Default.PerWorkflow.ChainRead.Solana.PayloadSizeLimit) + if err != nil { + return + } + + s.batchItemLimit, err = limits.MakeUpperBoundLimiter(limitsFactory, cresettings.Default.PerWorkflow.ChainRead.Solana.BatchItemLimit) return } +func (s *Solana) Close() error { + return services.CloseAll(s.reportSizeLimit, s.txComputeLimit, s.readPayloadSizeLimiter, s.batchItemLimit) +} + +func (s *Solana) checkReadPayloadSize(ctx context.Context, msg proto.Message) error { + b, err := proto.MarshalOptions{Deterministic: true}.Marshal(msg) + if err != nil { + return err + } + return s.readPayloadSizeLimiter.Check(ctx, commoncfg.SizeOf(b)) +} + +func (s *Solana) validateBatchItemCount(ctx context.Context, count int) error { + return s.batchItemLimit.Check(ctx, count) +} + +func getReadError(lggr logger.SugaredLogger, err error) caperrors.Error { + if err == nil { + return nil + } + + isUserErr := isUserError(err) + capErr := GetError(err, isUserErr) + + // TODO: logging of init, success and error should be move to a higher level + lggr = lggr.With("error", err) + const msg = "Read operation failed" + if isUserErr { + lggr.Debug(msg) + } else { + lggr.Error(msg) + } + + return capErr +} + +func isUserError(err error) bool { + return !errors.Is(err, context.DeadlineExceeded) && !isNodeInfraError(err) +} + +func isNodeInfraError(err error) bool { + return errors.Is(err, multinode.ErrNodeError) || + strings.Contains(err.Error(), multinode.ErrNodeError.Error()) +} + var GetError = capcommon.GetError var NewUserError = capcommon.NewUserError diff --git a/chain_capabilities/solana/actions/actions_test.go b/chain_capabilities/solana/actions/actions_test.go index 388e97e2d..e82433714 100644 --- a/chain_capabilities/solana/actions/actions_test.go +++ b/chain_capabilities/solana/actions/actions_test.go @@ -1,6 +1,7 @@ package actions import ( + "bytes" "context" "errors" "fmt" @@ -32,6 +33,20 @@ type mockedSolana struct { solana *Solana } +func testRequestMetadata(workflowExecutionID, referenceID string) capabilities.RequestMetadata { + return capabilities.RequestMetadata{ + WorkflowID: "wf-id", + WorkflowExecutionID: workflowExecutionID, + ReferenceID: referenceID, + } +} + +func testReadCall(t *testing.T, workflowExecutionID, referenceID string) (context.Context, capabilities.RequestMetadata) { + t.Helper() + meta := testRequestMetadata(workflowExecutionID, referenceID) + return meta.ContextWithCRE(t.Context()), meta +} + func TestGetAccountInfoWithOpts(t *testing.T) { t.Parallel() @@ -53,10 +68,8 @@ func TestGetAccountInfoWithOpts(t *testing.T) { t.Parallel() helper := newMockedSolana(t, true) - _, err := helper.solana.GetAccountInfoWithOpts(t.Context(), capabilities.RequestMetadata{ - WorkflowExecutionID: "weid", - ReferenceID: "step-id", - }, &solcap.GetAccountInfoWithOptsRequest{ + ctx, meta := testReadCall(t, "weid", "step-id") + _, err := helper.solana.GetAccountInfoWithOpts(ctx, meta, &solcap.GetAccountInfoWithOptsRequest{ Account: []byte{1, 2, 3}, }) require.Error(t, err) @@ -94,10 +107,8 @@ func TestGetAccountInfoWithOpts(t *testing.T) { Return(serviceReply, nil). Once() - resp, err := helper.solana.GetAccountInfoWithOpts(t.Context(), capabilities.RequestMetadata{ - WorkflowExecutionID: "weid", - ReferenceID: "step-id", - }, &solcap.GetAccountInfoWithOptsRequest{ + ctx, meta := testReadCall(t, "weid", "step-id") + resp, err := helper.solana.GetAccountInfoWithOpts(ctx, meta, &solcap.GetAccountInfoWithOptsRequest{ Account: accountKey.PublicKey().Bytes(), }) require.NoError(t, err) @@ -122,10 +133,8 @@ func TestGetAccountInfoWithOpts(t *testing.T) { Return((*soltypes.GetAccountInfoReply)(nil), expectedErr). Once() - _, err = helper.solana.GetAccountInfoWithOpts(t.Context(), capabilities.RequestMetadata{ - WorkflowExecutionID: "weid", - ReferenceID: "step-id", - }, &solcap.GetAccountInfoWithOptsRequest{ + ctx, meta := testReadCall(t, "weid", "step-id") + _, err = helper.solana.GetAccountInfoWithOpts(ctx, meta, &solcap.GetAccountInfoWithOptsRequest{ Account: accountKey.PublicKey().Bytes(), }) require.Error(t, err) @@ -167,7 +176,7 @@ func newMockedSolana(t *testing.T, readsEnabled bool) *mockedSolana { lggr := logger.Test(t) service := &Solana{ readsEnabled: readsEnabled, - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), beholderProcessor: NopBeholderProcessor{}, messageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), chainSelector: 1, @@ -239,3 +248,764 @@ func runVolatileHashableHandle(ctx context.Context, req ctypes.Request) (<-chan ch <- reply return ch, nil } + +// runECHashableHandle simulates OCR consensus for ECHashableRequest (EventuallyConsistent). +func runECHashableHandle(ctx context.Context, req ctypes.Request) (<-chan ctypes.Reply, error) { + observableRequest, ok := req.(ctypes.ObservableRequest) + if !ok { + return nil, fmt.Errorf("request is not an ObservableRequest") + } + _ = observableRequest.CaptureObservation(ctx) + observation, err := observableRequest.GetOCRObservation() + if err != nil { + return nil, fmt.Errorf("failed to get OCR observation: %w", err) + } + if observation == nil { + ch := make(chan ctypes.Reply, 1) + ch <- ctypes.Reply{Err: fmt.Errorf("no observation captured")} + return ch, nil + } + + var reply ctypes.Reply + switch tObs := observation.Observation.(type) { + case *ctypes.RequestObservation_Hashable: + hashBytes := tObs.Hashable + if len(hashBytes) != ctypes.HashLength { + return nil, fmt.Errorf("unexpected hash length: got %d, want %d", len(hashBytes), ctypes.HashLength) + } + var reportData ctypes.Hash + copy(reportData[:], hashBytes) + reply = ctypes.Reply{Value: ctypes.NewHashableRequestReport(ocrtypes.ConfigDigest{}, 0, reportData, nil)} + case *ctypes.RequestObservation_Error: + reply = ctypes.Reply{Err: ctypes.ObservationError(tObs.Error).Err()} + default: + return nil, fmt.Errorf("unexpected observation type: %T", observation.Observation) + } + + ch := make(chan ctypes.Reply, 1) + ch <- reply + return ch, nil +} + +// runAggregatableHandle simulates OCR consensus for AggregatableRequest (Volatile Aggregatable). +// It passes the observed Decimal value straight through as the "aggregated" result. +func runAggregatableHandle(ctx context.Context, req ctypes.Request) (<-chan ctypes.Reply, error) { + observableRequest, ok := req.(ctypes.ObservableRequest) + if !ok { + return nil, fmt.Errorf("request is not an ObservableRequest") + } + _ = observableRequest.CaptureObservation(ctx) + observation, err := observableRequest.GetOCRObservation() + if err != nil { + return nil, fmt.Errorf("failed to get OCR observation: %w", err) + } + if observation == nil { + ch := make(chan ctypes.Reply, 1) + ch <- ctypes.Reply{Err: fmt.Errorf("no observation captured")} + return ch, nil + } + + var reply ctypes.Reply + switch tObs := observation.Observation.(type) { + case *ctypes.RequestObservation_Aggregatable: + if tObs.Aggregatable == nil { + return nil, fmt.Errorf("nil aggregatable observation") + } + reply = ctypes.Reply{Value: tObs.Aggregatable.Value} + case *ctypes.RequestObservation_Error: + reply = ctypes.Reply{Err: ctypes.ObservationError(tObs.Error).Err()} + default: + return nil, fmt.Errorf("unexpected observation type: %T", observation.Observation) + } + + ch := make(chan ctypes.Reply, 1) + ch <- reply + return ch, nil +} + +// validSig returns a valid 64-byte signature slice. +func validSig() []byte { + sig := make([]byte, soltypes.SignatureLength) + for i := range sig { + sig[i] = byte(i + 1) + } + return sig +} + +// ---- GetBalance ---- + +func TestGetBalance(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + _, err := helper.solana.GetBalance(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetBalanceRequest{Addr: solana.NewWallet().PublicKey().Bytes()}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid addr", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + _, err := helper.solana.GetBalance(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetBalanceRequest{Addr: []byte{1, 2, 3}}) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + require.Contains(t, err.Error(), "invalid request") + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + helper.solanaService.EXPECT(). + GetBalance(mock.Anything, mock.MatchedBy(func(req soltypes.GetBalanceRequest) bool { + return req.Addr == soltypes.PublicKey(key.PublicKey()) + })). + Return(&soltypes.GetBalanceReply{Value: 1_000_000}, nil).Once() + + resp, err := helper.solana.GetBalance(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetBalanceRequest{Addr: key.PublicKey().Bytes()}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, uint64(1_000_000), resp.Response.Value) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + svcErr := errors.New("rpc down") + + helper.solanaService.EXPECT(). + GetBalance(mock.Anything, mock.Anything). + Return((*soltypes.GetBalanceReply)(nil), svcErr).Once() + + _, err = helper.solana.GetBalance(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetBalanceRequest{Addr: key.PublicKey().Bytes()}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetMultipleAccountsWithOpts ---- + +func TestGetMultipleAccountsWithOpts(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + _, err := helper.solana.GetMultipleAccountsWithOpts(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetMultipleAccountsWithOptsRequest{ + Accounts: [][]byte{solana.NewWallet().PublicKey().Bytes()}, + }) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid account key", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetMultipleAccountsWithOpts(ctx, meta, &solcap.GetMultipleAccountsWithOptsRequest{ + Accounts: [][]byte{{1, 2, 3}}, + }) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + serviceReply := &soltypes.GetMultipleAccountsReply{ + RPCContext: soltypes.RPCContext{Slot: 77}, + Value: []*soltypes.Account{{Lamports: 500}}, + } + helper.solanaService.EXPECT(). + GetMultipleAccountsWithOpts(mock.Anything, mock.Anything). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetMultipleAccountsWithOpts(ctx, meta, &solcap.GetMultipleAccountsWithOptsRequest{ + Accounts: [][]byte{key.PublicKey().Bytes()}, + }) + require.NoError(t, err) + require.NotNil(t, resp) + require.Len(t, resp.Response.Value, 1) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + svcErr := errors.New("rpc timeout") + + helper.solanaService.EXPECT(). + GetMultipleAccountsWithOpts(mock.Anything, mock.Anything). + Return((*soltypes.GetMultipleAccountsReply)(nil), svcErr).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err = helper.solana.GetMultipleAccountsWithOpts(ctx, meta, &solcap.GetMultipleAccountsWithOptsRequest{ + Accounts: [][]byte{key.PublicKey().Bytes()}, + }) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetProgramAccounts ---- + +func TestGetProgramAccounts(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + _, err := helper.solana.GetProgramAccounts(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetProgramAccountsRequest{Program: solana.NewWallet().PublicKey().Bytes()}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid program key", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetProgramAccounts(ctx, meta, &solcap.GetProgramAccountsRequest{Program: []byte{1, 2, 3}}) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + require.Contains(t, err.Error(), "invalid request") + }) + + t.Run("nil filter entry", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + ctx, meta := testReadCall(t, "weid", "ref") + _, err = helper.solana.GetProgramAccounts(ctx, meta, &solcap.GetProgramAccountsRequest{ + Program: key.PublicKey().Bytes(), + Opts: &solcap.GetProgramAccountsOpts{Filters: []*solcap.RPCFilter{nil}}, + }) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + serviceReply := &soltypes.GetProgramAccountsReply{ + Value: []*soltypes.KeyedAccount{ + {Pubkey: soltypes.PublicKey(key.PublicKey())}, + }, + } + helper.solanaService.EXPECT(). + GetProgramAccounts(mock.Anything, mock.MatchedBy(func(req soltypes.GetProgramAccountsRequest) bool { + return req.Program == soltypes.PublicKey(key.PublicKey()) + })). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetProgramAccounts(ctx, meta, &solcap.GetProgramAccountsRequest{Program: key.PublicKey().Bytes()}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Len(t, resp.Response.Value, 1) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("accounts sorted by pubkey for deterministic hashing", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + program, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + // Build three keys whose raw bytes are deliberately out of order. + highKey, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + lowKey, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + midKey, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + // Force a deterministic ordering: low < mid < high by assigning first byte. + var low, mid, high soltypes.PublicKey + lowPK, midPK, highPK := lowKey.PublicKey(), midKey.PublicKey(), highKey.PublicKey() + low[0], mid[0], high[0] = 0x10, 0x50, 0xFF + copy(low[1:], lowPK[1:]) + copy(mid[1:], midPK[1:]) + copy(high[1:], highPK[1:]) + + // RPC returns accounts in reverse order (high → mid → low). + serviceReply := &soltypes.GetProgramAccountsReply{ + Value: []*soltypes.KeyedAccount{ + {Pubkey: high}, + {Pubkey: mid}, + {Pubkey: low}, + }, + } + helper.solanaService.EXPECT(). + GetProgramAccounts(mock.Anything, mock.Anything). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetProgramAccounts(ctx, meta, &solcap.GetProgramAccountsRequest{Program: program.PublicKey().Bytes()}) + require.NoError(t, err) + require.Len(t, resp.Response.Value, 3) + + // After sorting the response must be low → mid → high. + require.Equal(t, low[:], resp.Response.Value[0].Pubkey) + require.Equal(t, mid[:], resp.Response.Value[1].Pubkey) + require.Equal(t, high[:], resp.Response.Value[2].Pubkey) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + svcErr := errors.New("rpc unavailable") + + helper.solanaService.EXPECT(). + GetProgramAccounts(mock.Anything, mock.Anything). + Return((*soltypes.GetProgramAccountsReply)(nil), svcErr).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err = helper.solana.GetProgramAccounts(ctx, meta, &solcap.GetProgramAccountsRequest{Program: key.PublicKey().Bytes()}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetTransaction ---- + +func TestGetTransaction(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + _, err := helper.solana.GetTransaction(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetTransactionRequest{Signature: validSig()}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid signature", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetTransaction(ctx, meta, &solcap.GetTransactionRequest{Signature: []byte{1, 2, 3}}) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + + serviceReply := &soltypes.GetTransactionReply{Slot: 99} + helper.solanaService.EXPECT(). + GetTransaction(mock.Anything, mock.Anything). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetTransaction(ctx, meta, &solcap.GetTransactionRequest{Signature: validSig()}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, uint64(99), resp.Response.Slot) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + svcErr := errors.New("tx not found") + + helper.solanaService.EXPECT(). + GetTransaction(mock.Anything, mock.Anything). + Return((*soltypes.GetTransactionReply)(nil), svcErr).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetTransaction(ctx, meta, &solcap.GetTransactionRequest{Signature: validSig()}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetSignatureStatuses ---- + +func TestGetSignatureStatuses(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + _, err := helper.solana.GetSignatureStatuses(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetSignatureStatusesRequest{Sigs: [][]byte{validSig()}}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid signature", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetSignatureStatuses(ctx, meta, &solcap.GetSignatureStatusesRequest{Sigs: [][]byte{{1, 2, 3}}}) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + + conf := uint64(10) + serviceReply := &soltypes.GetSignatureStatusesReply{ + Results: []soltypes.GetSignatureStatusesResult{ + {Slot: 42, Confirmations: &conf, ConfirmationStatus: soltypes.ConfirmationStatusConfirmed}, + }, + } + helper.solanaService.EXPECT(). + GetSignatureStatuses(mock.Anything, mock.Anything). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetSignatureStatuses(ctx, meta, &solcap.GetSignatureStatusesRequest{Sigs: [][]byte{validSig()}}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Len(t, resp.Response.Results, 1) + require.Equal(t, uint64(42), resp.Response.Results[0].Slot) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + svcErr := errors.New("sig lookup failed") + + helper.solanaService.EXPECT(). + GetSignatureStatuses(mock.Anything, mock.Anything). + Return((*soltypes.GetSignatureStatusesReply)(nil), svcErr).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetSignatureStatuses(ctx, meta, &solcap.GetSignatureStatusesRequest{Sigs: [][]byte{validSig()}}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetBlock ---- + +func TestGetBlock(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + _, err := helper.solana.GetBlock(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetBlockRequest{Slot: 1}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid commitment", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetBlock(ctx, meta, &solcap.GetBlockRequest{ + Slot: 1, + Opts: &solcap.GetBlockOpts{Commitment: solcap.CommitmentType(999)}, + }) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + + serviceReply := &soltypes.GetBlockReply{ParentSlot: 41} + helper.solanaService.EXPECT(). + GetBlock(mock.Anything, mock.MatchedBy(func(req soltypes.GetBlockRequest) bool { + return req.Slot == 42 + })). + Return(serviceReply, nil).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + resp, err := helper.solana.GetBlock(ctx, meta, &solcap.GetBlockRequest{Slot: 42}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, uint64(41), resp.Response.ParentSlot) + require.NotNil(t, resp.OCRAttestation) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runECHashableHandle} + svcErr := errors.New("block not available") + + helper.solanaService.EXPECT(). + GetBlock(mock.Anything, mock.Anything). + Return((*soltypes.GetBlockReply)(nil), svcErr).Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetBlock(ctx, meta, &solcap.GetBlockRequest{Slot: 42}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetSlotHeight ---- + +func TestGetSlotHeight(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + _, err := helper.solana.GetSlotHeight(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetSlotHeightRequest{}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid commitment", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + _, err := helper.solana.GetSlotHeight(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetSlotHeightRequest{Commitment: solcap.CommitmentType(999)}) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + + const height uint64 = 500_000 + helper.solanaService.EXPECT(). + GetSlotHeight(mock.Anything, mock.Anything). + Return(&soltypes.GetSlotHeightReply{Height: height}, nil).Once() + + resp, err := helper.solana.GetSlotHeight(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetSlotHeightRequest{Commitment: solcap.CommitmentType_COMMITMENT_TYPE_CONFIRMED}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, height, resp.Response.Height) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + svcErr := errors.New("slot unavailable") + + helper.solanaService.EXPECT(). + GetSlotHeight(mock.Anything, mock.Anything). + Return((*soltypes.GetSlotHeightReply)(nil), svcErr).Once() + + _, err := helper.solana.GetSlotHeight(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetSlotHeightRequest{}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +// ---- GetFeeForMessage ---- + +func TestGetFeeForMessage(t *testing.T) { + t.Parallel() + + t.Run("reads disabled", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, false) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + _, err := helper.solana.GetFeeForMessage(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetFeeForMessageRequest{Message: "someMsg"}) + require.Error(t, err) + require.Contains(t, err.Error(), "reads are not available") + }) + + t.Run("invalid commitment", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + _, err := helper.solana.GetFeeForMessage(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetFeeForMessageRequest{ + Message: "someMsg", + Commitment: solcap.CommitmentType(999), + }) + require.Error(t, err) + var capErr caperrors.Error + require.True(t, errors.As(err, &capErr)) + require.Equal(t, caperrors.OriginUser, capErr.Origin()) + }) + + t.Run("happy path", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + + const fee uint64 = 5000 + helper.solanaService.EXPECT(). + GetFeeForMessage(mock.Anything, mock.MatchedBy(func(req soltypes.GetFeeForMessageRequest) bool { + return req.Message == "someMsg" + })). + Return(&soltypes.GetFeeForMessageReply{Fee: fee}, nil).Once() + + resp, err := helper.solana.GetFeeForMessage(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetFeeForMessageRequest{Message: "someMsg"}) + require.NoError(t, err) + require.NotNil(t, resp) + require.Equal(t, fee, resp.Response.Fee) + }) + + t.Run("service error", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + helper.solana.handler = testConsensusHandler{handle: runAggregatableHandle} + svcErr := errors.New("fee estimation failed") + + helper.solanaService.EXPECT(). + GetFeeForMessage(mock.Anything, mock.Anything). + Return((*soltypes.GetFeeForMessageReply)(nil), svcErr).Once() + + _, err := helper.solana.GetFeeForMessage(t.Context(), capabilities.RequestMetadata{ + WorkflowExecutionID: "weid", ReferenceID: "ref", + }, &solcap.GetFeeForMessageRequest{Message: "someMsg"}) + require.Error(t, err) + require.ErrorContains(t, err, svcErr.Error()) + }) +} + +func TestReadLimits(t *testing.T) { + t.Parallel() + + t.Run("GetMultipleAccountsWithOpts batch item limit exceeded", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + + accounts := make([][]byte, 101) + for i := range accounts { + key, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + accounts[i] = key.PublicKey().Bytes() + } + + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetMultipleAccountsWithOpts(ctx, meta, &solcap.GetMultipleAccountsWithOptsRequest{ + Accounts: accounts, + }) + require.Error(t, err) + require.ErrorContains(t, err, "BatchItemLimit") + }) + + t.Run("GetSignatureStatuses batch item limit exceeded", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + + sigs := make([][]byte, 101) + for i := range sigs { + sigs[i] = validSig() + } + + ctx, meta := testReadCall(t, "weid", "ref") + _, err := helper.solana.GetSignatureStatuses(ctx, meta, &solcap.GetSignatureStatusesRequest{ + Sigs: sigs, + }) + require.Error(t, err) + require.ErrorContains(t, err, "BatchItemLimit") + }) + + t.Run("GetAccountInfoWithOpts response payload limit exceeded", func(t *testing.T) { + t.Parallel() + helper := newMockedSolana(t, true) + + accountKey, err := solana.NewRandomPrivateKey() + require.NoError(t, err) + + serviceReply := &soltypes.GetAccountInfoReply{ + RPCContext: soltypes.RPCContext{Slot: 42}, + Value: &soltypes.Account{ + Lamports: 1, + Data: &soltypes.DataBytesOrJSON{ + RawDataEncoding: soltypes.EncodingBase64, + AsDecodedBinary: bytes.Repeat([]byte{'z'}, 6000), + }, + }, + } + helper.solanaService.EXPECT(). + GetAccountInfoWithOpts(mock.Anything, mock.Anything). + Return(serviceReply, nil). + Once() + + ctx, meta := testReadCall(t, "weid", "ref") + _, err = helper.solana.GetAccountInfoWithOpts(ctx, meta, &solcap.GetAccountInfoWithOptsRequest{ + Account: accountKey.PublicKey().Bytes(), + }) + require.Error(t, err) + require.ErrorContains(t, err, "PayloadSizeLimit") + }) +} diff --git a/chain_capabilities/solana/actions/write_report_test.go b/chain_capabilities/solana/actions/write_report_test.go index 597120d55..3899fb7d4 100644 --- a/chain_capabilities/solana/actions/write_report_test.go +++ b/chain_capabilities/solana/actions/write_report_test.go @@ -324,7 +324,7 @@ func createMocksAndCapability(t *testing.T, lggr logger.Logger) *testHelper { mockTrInfo := NewTransmissionInfoProvider_mock(t) mockClient := NewCREForwarderClient_mock(t) service := &Solana{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), forwarderClient: mockClient, transmissionInfoProvider: mockTrInfo, beholderProcessor: NopBeholderProcessor{}, @@ -736,7 +736,7 @@ func TestGetFee(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) wr := &WriteReport{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), lggr: testLogger, } @@ -756,7 +756,7 @@ func TestGetFee(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) wr := &WriteReport{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), lggr: testLogger, } @@ -776,7 +776,7 @@ func TestGetFee(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) wr := &WriteReport{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), lggr: testLogger, } diff --git a/chain_capabilities/solana/go.mod b/chain_capabilities/solana/go.mod index decd378d7..01492f5e8 100644 --- a/chain_capabilities/solana/go.mod +++ b/chain_capabilities/solana/go.mod @@ -1,6 +1,6 @@ module github.com/smartcontractkit/capabilities/chain_capabilities/solana -go 1.26.2 +go 1.26.3 require ( github.com/gagliardetto/solana-go v1.14.0 @@ -9,10 +9,10 @@ require ( github.com/smartcontractkit/capabilities/chain_capabilities/common v0.0.0-20260615195421-fb87220e503f github.com/smartcontractkit/capabilities/libs v0.0.0-20260609124022-2749e4a32bfb github.com/smartcontractkit/chain-selectors v1.0.100 - github.com/smartcontractkit/chainlink-common v0.11.2-0.20260602135221-cc7a5b50532a - github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260326180413-c69f27e37a13 - github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260526195338-adcf8013a1b7 - github.com/smartcontractkit/chainlink-solana v1.1.2-0.20260420191419-ea62f88cbdb4 + github.com/smartcontractkit/chainlink-common v0.11.2-0.20260713160348-3e9ebef19093 + github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805-26d78d5e1243 + github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b + github.com/smartcontractkit/chainlink-solana v1.3.1-0.20260710051325-0fdf1bb70f47 github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260421131224-c46cbfe7bc6c github.com/smartcontractkit/libocr v0.0.0-20260403184524-b6409238958d github.com/stretchr/testify v1.11.1 @@ -131,10 +131,10 @@ require ( github.com/smartcontractkit/chainlink-ccip v0.1.1-solana.0.20260317185256-d5f7db87ae70 // indirect github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260310183131-8d0f0e383288 // indirect github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260310183131-8d0f0e383288 // indirect - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0 // indirect - github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20251210101658-1c5c8e4c4f15 // indirect + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 // indirect + github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20260521164805-26d78d5e1243 // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b // indirect - github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260319180422-b5808c964785 // indirect + github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect github.com/smartcontractkit/freeport v0.1.3-0.20250828155247-add56fa28aad // indirect github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7 // indirect github.com/streamingfast/logging v0.0.0-20230608130331-f22c91403091 // indirect diff --git a/chain_capabilities/solana/go.sum b/chain_capabilities/solana/go.sum index 70052a53a..045f695b0 100644 --- a/chain_capabilities/solana/go.sum +++ b/chain_capabilities/solana/go.sum @@ -629,28 +629,28 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260310183131-8 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260310183131-8d0f0e383288/go.mod h1:jPHlo/IN2YAArI001JJixmm6ZHQwgnAVJXY8VBFiFTc= github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260310183131-8d0f0e383288 h1:eEjTgIQn4RW0ZPRepUDYTdgGwaRCMawMwgXkHItUc9U= github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260310183131-8d0f0e383288/go.mod h1:67YbnoglYD61Pz/jTVCgav9wFq7S35OU8UyQSvPllRw= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260602135221-cc7a5b50532a h1:DHdGiDkMe+WS2e+Cy6ooNxGVX4gwiRRcgperFWmR4R4= -github.com/smartcontractkit/chainlink-common v0.11.2-0.20260602135221-cc7a5b50532a/go.mod h1:6jgqiFXFJHqjkvFFmuf8gvoUFa6Ygx/D1tKnIL+CCF8= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0 h1:NExKM/D0HneOq/N5LGTbkV4VOa0UHCvfTNEb4GqYpto= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260713160348-3e9ebef19093 h1:/sSmhE9SVrZ5e1xjrGkg92h8tsfBoi5/NirMkbtSkVY= +github.com/smartcontractkit/chainlink-common v0.11.2-0.20260713160348-3e9ebef19093/go.mod h1:snfVBRRQTpC2x5O3bQHZe9SvJX5yv/SbG8oHkJTKLtE= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 h1:o7vfwNQjQbMKQ9YsZFQOxvU7RMXD/wKnZsX5N9sDS3w= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62/go.mod h1:HmUyH2oD9m+GRpKq7q3vuRnm1F2Uczf/Nd1v3ipMSK8= github.com/smartcontractkit/chainlink-common/pkg/monitoring v0.0.0-20251215152504-b1e41f508340 h1:PsjEI+5jZIz9AS4eOsLS5VpSWJINf38clXV3wryPyMk= github.com/smartcontractkit/chainlink-common/pkg/monitoring v0.0.0-20251215152504-b1e41f508340/go.mod h1:P/0OSXUlFaxxD4B/P6HWbxYtIRmmWGDJAvanq19879c= -github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20251210101658-1c5c8e4c4f15 h1:IXF7+k8I1YY/yvXC1wnS3FAAggtCy6ByEQ9hv/F2FvQ= -github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20251210101658-1c5c8e4c4f15/go.mod h1:HG/aei0MgBOpsyRLexdKGtOUO8yjSJO3iUu0Uu8KBm4= -github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260326180413-c69f27e37a13 h1:3KLLkTCIAy9CvT35Ey0k6pcWX/u+qsm3Y/58TI5VSAg= -github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260326180413-c69f27e37a13/go.mod h1:Y7h84PqCe/Vimf2h1Nc6tMiOJStDbtM33fEUeaaF5xk= +github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20260521164805-26d78d5e1243 h1:vaFBupfFfImQgqOeuC7Muk2GflbYP6Gpi0Y/TLroFU8= +github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20260521164805-26d78d5e1243/go.mod h1:HG/aei0MgBOpsyRLexdKGtOUO8yjSJO3iUu0Uu8KBm4= +github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805-26d78d5e1243 h1:71PGTkjdFZ0JrloEC2Fs8eHl1b1gmUuH+bq7q23usKk= +github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260521164805-26d78d5e1243/go.mod h1:7ketk4ischPQW/JQgmyHz6zdzLUJv1VC29SiSgosydQ= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4 h1:GCzrxDWn3b7jFfEA+WiYRi8CKoegsayiDoJBCjYkneE= github.com/smartcontractkit/chainlink-protos/billing/go v0.0.0-20251024234028-0988426d98f4/go.mod h1:HHGeDUpAsPa0pmOx7wrByCitjQ0mbUxf0R9v+g67uCA= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260526195338-adcf8013a1b7 h1:iljEJss3WOwcsMkWy72Yn2zvjw7Gyxc+RXL7r8YKM6g= -github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260526195338-adcf8013a1b7/go.mod h1:vTFHTCbLui4Vn8fTmAadfE3rdnvfrDwOmMujmW857D0= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b h1:VDgJWDipihV9f7M5+d21d1RzSsg5rEv+iI12oN1VQbo= +github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b/go.mod h1:vTFHTCbLui4Vn8fTmAadfE3rdnvfrDwOmMujmW857D0= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b h1:QuI6SmQFK/zyUlVWEf0GMkiUYBPY4lssn26nKSd/bOM= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b/go.mod h1:qSTSwX3cBP3FKQwQacdjArqv0g6QnukjV4XuzO6UyoY= -github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260319180422-b5808c964785 h1:oli+2uLU6jcrJGCuYFqk3475hiwL17SWlITWLv+tx/w= -github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260319180422-b5808c964785/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= +github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 h1:/vkKPJoweLkRd56V4YHGRAtTG4+/JAlgklGEfvH6l4c= +github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16/go.mod h1:dkR2uYg9XYJuT1JASkPzWE51jjFkVb86P7a/yXe5/GM= github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9 h1:LQy2j2+TdKLSWsUTUYuqmQPn8kjqCLjGI3ZJYGtDc08= github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9/go.mod h1:GTpDgyK0OObf7jpch6p8N281KxN92wbB8serZhU9yRc= -github.com/smartcontractkit/chainlink-solana v1.1.2-0.20260420191419-ea62f88cbdb4 h1:WaQzRyMCmi7gddZraPtHSqCjZ5dnVYFYXqEtq2DsQjw= -github.com/smartcontractkit/chainlink-solana v1.1.2-0.20260420191419-ea62f88cbdb4/go.mod h1:sUsEwLtVPBlz0wPcysaolS+HVj9cOAt4jYhwE6J8dXg= +github.com/smartcontractkit/chainlink-solana v1.3.1-0.20260710051325-0fdf1bb70f47 h1:V7XUufui82J5/b4AukUT0qco3r2pk9B4KHq5k0Ln1pg= +github.com/smartcontractkit/chainlink-solana v1.3.1-0.20260710051325-0fdf1bb70f47/go.mod h1:SIPlLXeil004BhzBQgiZRE0cAAKCwmHZfd3o5ts91pE= github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260421131224-c46cbfe7bc6c h1:Hn/80PyYFrQhRlNSaq9HY4cjc/7AuP9zyWLle22t34A= github.com/smartcontractkit/chainlink-solana/contracts v0.0.0-20260421131224-c46cbfe7bc6c/go.mod h1:C5pZsbYX3qkhZTYWr1aYJi9QMfonFAun+Jl1npQ7UJA= github.com/smartcontractkit/freeport v0.1.3-0.20250828155247-add56fa28aad h1:lgHxTHuzJIF3Vj6LSMOnjhqKgRqYW+0MV2SExtCYL1Q= diff --git a/chain_capabilities/solana/main.go b/chain_capabilities/solana/main.go index 515f90239..e4d3c0421 100644 --- a/chain_capabilities/solana/main.go +++ b/chain_capabilities/solana/main.go @@ -35,6 +35,7 @@ import ( caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/solana" solcapserver "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/solana/server" + capmon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/monitoring" ) const ( @@ -67,7 +68,7 @@ var _ solcapserver.ClientCapability = &capabilityGRPCService{} func main() { loopserver.ServeNew(CapabilityName, func(s *loop.Server) loop.StandardCapabilities { return solcapserver.NewClientServer(&capabilityGRPCService{lggr: s.Logger.Named(CapabilityName), limitsFactory: s.LimitsFactory}) - }) + }, loop.WithOtelViews(append(consMetrics.MetricViews(), capmon.MetricViews()...))) } func (c *capabilityGRPCService) ChainSelector() uint64 { @@ -103,6 +104,9 @@ func (c *capabilityGRPCService) Close() error { if c.triggerService != nil { closers = append(closers, c.triggerService) } + if c.Solana != nil { + closers = append(closers, c.Solana) + } return services.CloseAll(closers...) } @@ -125,6 +129,7 @@ func (c *capabilityGRPCService) Description() string { func (c *capabilityGRPCService) Ready() error { return nil } + func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies core.StandardCapabilitiesDependencies) error { c.lggr.Infof("Initialising %s", CapabilityName) @@ -150,10 +155,17 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor } c.id = "solana" + ":ChainSelector:" + strconv.FormatUint(c.chainSelector, 10) + "@1.0.0" + c.CapabilityInfo = capabilities.CapabilityInfo{ + ID: c.id, + CapabilityType: capabilities.CapabilityTypeCombined, + Description: c.Description(), + IsLocal: cfg.IsLocal, + } var chainInfo types.ChainInfo - // protection for e2e tests when we run against local validator - if !cfg.IsLocal { + if cfg.IsLocal { + chainInfo = localChainInfo(cfg, c.chainSelector) + } else { chainInfo, err = relayer.GetChainInfo(ctx) if err != nil { return fmt.Errorf("failed to fetch chain info for chainID %s from relayer: %w", cfg.ChainID, err) @@ -270,6 +282,29 @@ func (s *capabilityGRPCService) setSelector(cfg *config.Config) error { return nil } +// localChainInfo builds monitoring labels for local CRE runs where the relayer +// cannot resolve chain metadata from a fixed genesis hash. +func localChainInfo(cfg *config.Config, chainSelector uint64) types.ChainInfo { + chainID := cfg.ChainID + if chainID == "" { + chainID = strconv.FormatUint(chainSelector, 10) + } + + networkName := cfg.Network + if networkName == "" { + networkName = "local" + } + + networkNameFull := networkName + "-local" + + return types.ChainInfo{ + FamilyName: "solana", + ChainID: chainID, + NetworkName: networkName, + NetworkNameFull: networkNameFull, + } +} + func (c *capabilityGRPCService) unmarshalConfig(configStr string) (*config.Config, error) { var cfg config.Config if err := json.Unmarshal([]byte(configStr), &cfg); err != nil { diff --git a/chain_capabilities/solana/monitoring/messages.go b/chain_capabilities/solana/monitoring/messages.go index fe1fb1b3a..e6e4c965f 100644 --- a/chain_capabilities/solana/monitoring/messages.go +++ b/chain_capabilities/solana/monitoring/messages.go @@ -13,6 +13,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/capabilities" solanacappb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/solana" + capmon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/monitoring" "github.com/smartcontractkit/chainlink-common/pkg/types" "github.com/smartcontractkit/chainlink-common/pkg/types/chains/solana" @@ -36,6 +37,18 @@ func NewMessageBuilder(chainInfo types.ChainInfo, capInfo capabilities.Capabilit } } +// CapabilityMetricsAttributes returns capability-scoped OTel labels for v2 action metrics. +func (m *MessageBuilder) CapabilityMetricsAttributes() []attribute.KeyValue { + return []attribute.KeyValue{ + attribute.String(capmon.LabelChainFamilyName, capmon.ValOrUnknown(m.ChainInfo.FamilyName)), + attribute.String(capmon.LabelChainID, capmon.ValOrUnknown(m.ChainInfo.ChainID)), + attribute.String(capmon.LabelNetworkName, capmon.ValOrUnknown(m.ChainInfo.NetworkName)), + attribute.String(capmon.LabelNetworkNameFull, capmon.ValOrUnknown(m.ChainInfo.NetworkNameFull)), + attribute.String(capmon.LabelCapabilityType, capmon.ValOrUnknown(string(m.CapInfo.CapabilityType))), + attribute.String(capmon.LabelCapabilityID, capmon.ValOrUnknown(m.CapInfo.ID)), + } +} + func (m *MessageBuilder) BuildWriteReportTxFeeCalculationError(tc TelemetryContext, req *solcap.WriteReportRequest, signature solgo.Signature, cause string) ErrorMessage { summary := "Failed to calculate transaction fee" if !signature.IsZero() { diff --git a/chain_capabilities/solana/trigger/trigger_test.go b/chain_capabilities/solana/trigger/trigger_test.go index 14ae54a85..178c307de 100644 --- a/chain_capabilities/solana/trigger/trigger_test.go +++ b/chain_capabilities/solana/trigger/trigger_test.go @@ -17,7 +17,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/capabilities" solanacappb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/solana" "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/settings" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink-common/pkg/types" "github.com/smartcontractkit/chainlink-common/pkg/types/chains/solana" @@ -34,18 +33,10 @@ const ( testWorkflowID = "test-workflow-1" testWorkflowOwner = "test-owner-1" testChannelBufferSize = 100 - testCapabilityID = "test-cap" testAddress = "11111111111111111111111111111112" testEventName = "TestEvent" ) -func testLimitsFactory(tb testing.TB) limits.Factory { - tb.Helper() - g, err := settings.NewJSONGetter([]byte(`{}`)) - require.NoError(tb, err) - return limits.Factory{Settings: g, Logger: logger.Test(tb)} -} - var ( testPublicKey = createTestPublicKey(testAddress) testEventSig = createTestEventSignature("TestEvent(string,uint256)") @@ -94,17 +85,11 @@ func createTestTelemetryContext() monitoring.TelemetryContext { func waitForTriggerRegistered(t *testing.T, service *SolanaLogTriggerService, triggerID string) { t.Helper() tests.AssertEventually(t, func() bool { - trigger, ok := service.triggers.Read(triggerID) - return ok && trigger.stopPolling != nil + _, ok := service.triggers.Read(triggerID) + return ok }) } -func startBaseTrigger(t *testing.T, service *SolanaLogTriggerService) { - t.Helper() - require.NoError(t, service.baseTrigger.Start(t.Context())) - t.Cleanup(func() { service.baseTrigger.Stop() }) -} - func startPollingAsync( ctx context.Context, t *testing.T, @@ -116,8 +101,6 @@ func startPollingAsync( logCh chan capabilities.TriggerAndId[*solanacappb.Log], ) { t.Helper() - startBaseTrigger(t, service) - service.baseTrigger.RegisterTrigger(triggerID, logCh) var wg sync.WaitGroup wg.Add(1) go func() { @@ -135,7 +118,7 @@ func setupTest(t *testing.T) (*SolanaLogTriggerService, *mocks.SolanaService) { lggr := logger.Test(t) opts := LogTriggerServiceOpts{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), Logger: lggr, BeholderProcessor: NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), @@ -144,9 +127,7 @@ func setupTest(t *testing.T) (*SolanaLogTriggerService, *mocks.SolanaService) { LogTriggerSendChannelBufferSize: testChannelBufferSize, Retention: time.Hour * 24, MaxLogsKept: 10000, - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, + LimitsFactory: limits.Factory{Logger: lggr}, } service, err := NewLogTriggerService(opts) @@ -643,7 +624,7 @@ func TestStartPolling(t *testing.T) { store := NewSolanaLogTriggerStore() opts := LogTriggerServiceOpts{ - SolanaService: mockSolanaService, + SolanaService: mocks.WrapSolanaService(mockSolanaService), Logger: logger.Nop(), BeholderProcessor: NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), @@ -652,9 +633,7 @@ func TestStartPolling(t *testing.T) { LogTriggerSendChannelBufferSize: testChannelBufferSize, Retention: time.Hour * 24, MaxLogsKept: 10000, - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, + LimitsFactory: limits.Factory{Logger: logger.Nop()}, } service, err := NewLogTriggerService(opts) @@ -748,15 +727,14 @@ func TestStartPolling(t *testing.T) { <-polled }) - t.Run("handles full inbox without panic", func(t *testing.T) { + t.Run("drops events when channel is full", func(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) store := NewSolanaLogTriggerStore() // Create service with very small buffer - lggr := logger.Test(t) opts := LogTriggerServiceOpts{ - SolanaService: mockSolanaService, - Logger: lggr, + SolanaService: mocks.WrapSolanaService(mockSolanaService), + Logger: logger.Test(t), Triggers: store, LogTriggerPollInterval: 1 * time.Millisecond, LogTriggerSendChannelBufferSize: 1, // Very small buffer @@ -764,9 +742,6 @@ func TestStartPolling(t *testing.T) { MaxLogsKept: 10000, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, } service, err := NewLogTriggerService(opts) @@ -802,18 +777,17 @@ func TestStartPolling(t *testing.T) { // Don't read from channel to force it to fill up <-ctx.Done() - // Test passes if no panic occurs - events are persisted and retried via baseTrigger + // Test passes if no panic occurs - drops are handled gracefully mockSolanaService.AssertExpectations(t) }) - t.Run("updates lastProcessedBlock when inbox is full", func(t *testing.T) { + t.Run("updates lastProcessedBlock even when logs are dropped", func(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) store := NewSolanaLogTriggerStore() - lggr := logger.Test(t) opts := LogTriggerServiceOpts{ - SolanaService: mockSolanaService, - Logger: lggr, + SolanaService: mocks.WrapSolanaService(mockSolanaService), + Logger: logger.Test(t), Triggers: store, LogTriggerPollInterval: 10 * time.Millisecond, LogTriggerSendChannelBufferSize: 1, @@ -821,9 +795,6 @@ func TestStartPolling(t *testing.T) { MaxLogsKept: 10000, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, } service, err := NewLogTriggerService(opts) @@ -872,10 +843,9 @@ func TestStartPolling(t *testing.T) { mockSolanaService := mocks.NewSolanaService(t) store := NewSolanaLogTriggerStore() - lggr := logger.Test(t) opts := LogTriggerServiceOpts{ - SolanaService: mockSolanaService, - Logger: lggr, + SolanaService: mocks.WrapSolanaService(mockSolanaService), + Logger: logger.Test(t), Triggers: store, LogTriggerPollInterval: 5 * time.Millisecond, LogTriggerSendChannelBufferSize: testChannelBufferSize, @@ -883,9 +853,6 @@ func TestStartPolling(t *testing.T) { MaxLogsKept: 10000, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, } service, err := NewLogTriggerService(opts) @@ -1107,16 +1074,12 @@ func BenchmarkSolanaLogTriggerService_BuildQueryExpressions(b *testing.B) { } func BenchmarkSolanaLogTriggerService_ToLogPollerFilter(b *testing.B) { - lggr := logger.Test(&testing.T{}) opts := LogTriggerServiceOpts{ - Logger: lggr, + Logger: logger.Test(&testing.T{}), BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), Retention: time.Hour * 24, MaxLogsKept: 10000, - LimitsFactory: testLimitsFactory(b), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, } service, err := NewLogTriggerService(opts) @@ -1150,9 +1113,6 @@ func TestSolanaLogTriggerService_NewLogTriggerService(t *testing.T) { Logger: lggr, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, }) require.NoError(t, err) require.NotNil(t, service) @@ -1165,11 +1125,12 @@ func TestSolanaLogTriggerService_NewLogTriggerService(t *testing.T) { t.Run("respects provided values", func(t *testing.T) { mockService := mocks.NewSolanaService(t) + wrappedService := mocks.WrapSolanaService(mockService) store := NewSolanaLogTriggerStore() lggr := logger.Test(t) opts := LogTriggerServiceOpts{ - SolanaService: mockService, + SolanaService: wrappedService, Logger: lggr, Triggers: store, LogTriggerPollInterval: 5 * time.Second, @@ -1178,59 +1139,19 @@ func TestSolanaLogTriggerService_NewLogTriggerService(t *testing.T) { MaxLogsKept: 20000, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, } service, err := NewLogTriggerService(opts) require.NoError(t, err) require.NotNil(t, service) - assert.Equal(t, mockService, service.SolanaService) + assert.Equal(t, wrappedService, service.SolanaService) assert.Equal(t, store, service.triggers) assert.Equal(t, 5*time.Second, service.logTriggerPollInterval) assert.Equal(t, uint64(2000), service.logTriggerSendChannelBufferSize) assert.Equal(t, 48*time.Hour, service.retention) assert.Equal(t, int64(20000), service.maxLogsKept) }) - - t.Run("requires trigger event store", func(t *testing.T) { - lggr := logger.Test(t) - _, err := NewLogTriggerService(LogTriggerServiceOpts{ - Logger: lggr, - BeholderProcessor: test.NopBeholderProcessor{}, - MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - CapabilityID: testCapabilityID, - }) - require.Error(t, err) - assert.Contains(t, err.Error(), "no trigger event store provided") - }) - - t.Run("requires capability ID", func(t *testing.T) { - lggr := logger.Test(t) - _, err := NewLogTriggerService(LogTriggerServiceOpts{ - Logger: lggr, - BeholderProcessor: test.NopBeholderProcessor{}, - MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - }) - require.Error(t, err) - assert.Contains(t, err.Error(), "capabilityID must be non-empty") - }) -} - -func TestAckEvent(t *testing.T) { - service, _ := setupTest(t) - startBaseTrigger(t, service) - - triggerID := "test-trigger" - logCh := make(chan capabilities.TriggerAndId[*solanacappb.Log], 1) - service.baseTrigger.RegisterTrigger(triggerID, logCh) - - require.Nil(t, service.AckEvent(t.Context(), triggerID, "event-1")) } func TestSolanaLogTriggerStore(t *testing.T) { @@ -1303,9 +1224,6 @@ func TestSolanaLogTriggerService_EdgeCases(t *testing.T) { Logger: lggr, BeholderProcessor: test.NopBeholderProcessor{}, MessageBuilder: monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""), - LimitsFactory: testLimitsFactory(t), - TriggerEventStore: capabilities.NewMemEventStore(), - CapabilityID: testCapabilityID, }) require.NoError(t, err) require.NotNil(t, service) @@ -1411,31 +1329,6 @@ func TestCleanUpStaleFilters(t *testing.T) { service.cleanUpStaleFilters(t.Context()) // No panic, no error - just silently skips }) - - t.Run("preserves filter when trigger is registered", func(t *testing.T) { - service, mockSolana := setupTest(t) - filterName := testTriggerID + SuffixLogTriggerFilterID - - mockSolana.EXPECT().GetFiltersNames(mock.Anything).Return([]string{filterName}, nil).Once() - - service.triggers.Write(testTriggerID, solanaLogTriggerState{ - stopPolling: func() {}, - filter: createTestRequest(), - }) - - service.cleanUpStaleFilters(t.Context()) - mockSolana.AssertNotCalled(t, "UnregisterLogTracking", mock.Anything, mock.Anything) - }) - - t.Run("removes orphan filter when trigger is not registered", func(t *testing.T) { - service, mockSolana := setupTest(t) - filterName := testTriggerID + SuffixLogTriggerFilterID - - mockSolana.EXPECT().GetFiltersNames(mock.Anything).Return([]string{filterName}, nil).Once() - mockSolana.EXPECT().UnregisterLogTracking(mock.Anything, filterName).Return(nil).Once() - - service.cleanUpStaleFilters(t.Context()) - }) } func TestRegisterLogTrigger_InputValidation(t *testing.T) {