Add Stellar WriteReport Beholder telemetry

chain_capabilities/stellar/actions/actions.go

@@ -113,7 +113,6 @@ func (s *Stellar) GetLatestLedger(ctx context.Context, metadata capabilities.Req
 	if err != nil {
 		s.lggr.Errorw("Failed to GetLatestLedger", "error", err)
 		return nil, capcommon.GetError(fmt.Errorf("failed to GetLatestLedger: %w", err), false)
-
 	}
 	return responseAndMetadata, nil
 }
@@ -220,6 +219,3 @@ func isUserError(err error) bool {
 func isStellarNodeInfraError(err error) bool {
 	return errors.Is(err, multinode.ErrNodeError) || strings.Contains(err.Error(), multinode.ErrNodeError.Error())
 }
-
-var GetError = capcommon.GetError
-var NewUserError = caperrors.NewPublicUserError

chain_capabilities/stellar/actions/actions_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors"
 	stellarcap "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/stellar"
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/stellar/scval"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
@@ -271,6 +272,17 @@ func TestConvertReadContractRequestFromProto(t *testing.T) {
 		require.Error(t, err)
 		require.Contains(t, err.Error(), "function is required")
 	})
+
+	t.Run("invalid arg conversion", func(t *testing.T) {
+		t.Parallel()
+		_, err := convertReadContractRequestFromProto(&stellarcap.ReadContractRequest{
+			ContractId: testForwarderAddress,
+			Function:   "balance",
+			Args:       []*scval.ScVal{nil},
+		})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "args[0]")
+	})
 }
 
 func TestIsStellarNodeInfraError_MessageSubstring(t *testing.T) {

chain_capabilities/stellar/actions/cre_forwarder_codec.go

@@ -12,6 +12,8 @@ import (
 	capcommon "github.com/smartcontractkit/capabilities/chain_capabilities/common"
 )
 
+const reportProcessedTopicPrefix = "forwarder_ReportProcessed"
+
 // CREForwarderCodec encodes and decodes Stellar CRE forwarder contract calls.
 type CREForwarderCodec interface {
 	EncodeReport(transmitter, receiver string, report *sdk.ReportResponse) ([]stellartypes.ScVal, error)

chain_capabilities/stellar/actions/forwarder_client.go

@@ -21,6 +21,8 @@ const (
 	forwarderReportFunction              = "report"
 	forwarderGetTransmissionInfoFunction = "get_transmission_info"
 	defaultLedgerBoundsOffset            = uint32(20)
+	// DefaultForwarderLookbackLedgers is how many ledgers back to search for ReportProcessed events.
+	DefaultForwarderLookbackLedgers = int64(100)
 )
 
 type TransmissionState uint32
@@ -111,7 +113,7 @@ type ReportProcessedEvent struct {
 
 func newForwarderClient(service types.StellarService, lggr logger.Logger, forwarderAddress string, forwarderLookbackLedgers int64) CREForwarderClient {
 	if forwarderLookbackLedgers <= 0 {
-		forwarderLookbackLedgers = defaultForwarderLookbackLedgers
+		forwarderLookbackLedgers = DefaultForwarderLookbackLedgers
 	}
 	return &forwarderClient{
 		StellarService:           service,

chain_capabilities/stellar/actions/forwarder_client_test.go

@@ -21,6 +21,33 @@ func TestForwarderClient_ForwarderAddress(t *testing.T) {
 	require.Equal(t, testForwarderAddress, client.ForwarderAddress())
 }
 
+func TestForwarderClient_DefaultForwarderLookbackLedgers(t *testing.T) {
+	t.Parallel()
+	lggr := logger.Test(t)
+	_, reqMeta, req := newWRReportFixture(t)
+	transmissionID, err := getTransmissionID(reqMeta.WorkflowExecutionID, req)
+	require.NoError(t, err)
+
+	svc := mocks.NewStellarService(t)
+	success := true
+	svc.EXPECT().GetLatestLedger(mock.Anything).
+		Return(stellartypes.GetLatestLedgerResponse{Sequence: 200}, nil).Once()
+	svc.EXPECT().GetEvents(mock.Anything, mock.MatchedBy(func(req stellartypes.GetEventsRequest) bool {
+		return req.StartLedger == 100
+	})).Return(stellartypes.GetEventsResponse{
+		Events: []stellartypes.EventInfo{{
+			TransactionHash: testTxHash,
+			Ledger:          150,
+			Value:           stellartypes.ScVal{Type: stellartypes.ScValTypeBool, Bool: &success},
+		}},
+	}, nil).Once()
+
+	client := newForwarderClient(svc, lggr, testForwarderAddress, 0)
+	events, err := client.GetReportProcessedEvents(t.Context(), transmissionID)
+	require.NoError(t, err)
+	require.Len(t, events, 1)
+}
+
 func TestForwarderClient_InvokeOnReport(t *testing.T) {
 	t.Parallel()
 	lggr := logger.Test(t)

chain_capabilities/stellar/actions/tx_hash_retriever.go

@@ -12,9 +12,7 @@ import (
 )
 
 const (
-	reportProcessedTopicPrefix      = "forwarder_ReportProcessed"
-	defaultForwarderLookbackLedgers = int64(100)
-	failedToRetrieveTxHashErrorMsg  = "failed to retrieve tx hash for report"
+	failedToRetrieveTxHashErrorMsg = "failed to retrieve tx hash for report"
 )
 
 var ErrUnexpectedSuccessfulTransmission = errors.New("unexpected successful transmission")
@@ -95,6 +93,9 @@ func (r *TxHashRetriever) GetFailedTransmissionHashWithCount(ctx context.Context
 				ErrUnexpectedSuccessfulTransmission, d.txHash)
 		}
 	}
+	if len(details) == 0 {
+		return "", 0, fmt.Errorf("no failed transmission found")
+	}
 
 	earliestIdx := 0
 	for i, d := range details {
@@ -103,14 +104,15 @@ func (r *TxHashRetriever) GetFailedTransmissionHashWithCount(ctx context.Context
 		}
 	}
 
+	selectedHash := details[earliestIdx].txHash
 	r.lggr.Debugw("Returning earliest failed transmission",
 		append([]any{
 			"txCount", len(details),
-			"selectedTxHash", details[earliestIdx].txHash,
+			"selectedTxHash", selectedHash,
 		}, r.transmissionID.LogAttrs()...)...,
 	)
 
-	return details[earliestIdx].txHash, len(details), nil
+	return selectedHash, len(details), nil
 }
 
 func (r *TxHashRetriever) fetchAndParseEvents(ctx context.Context) (eventDetailsList, error) {

chain_capabilities/stellar/actions/tx_hash_retriever_test.go

@@ -14,21 +14,35 @@ import (
 )
 
 type stubForwarderClient struct {
-	events []ReportProcessedEvent
-	err    error
+	events             []ReportProcessedEvent
+	eventsErr          error
+	transmissionInfoFn func(call int) (TransmissionInfo, error)
+	invokeOnReportResp *stellartypes.SubmitTransactionResponse
+	invokeOnReportErr  error
+	transmissionCalls  int
 }
 
 func (s *stubForwarderClient) InvokeOnReport(context.Context, string, *sdk.ReportResponse) (*stellartypes.SubmitTransactionResponse, error) {
-	panic("not implemented")
+	if s.invokeOnReportErr != nil {
+		return nil, s.invokeOnReportErr
+	}
+	if s.invokeOnReportResp != nil {
+		return s.invokeOnReportResp, nil
+	}
+	panic("stubForwarderClient.InvokeOnReport not configured")
 }
 
 func (s *stubForwarderClient) GetTransmissionInfo(context.Context, TransmissionID) (TransmissionInfo, error) {
-	panic("not implemented")
+	s.transmissionCalls++
+	if s.transmissionInfoFn != nil {
+		return s.transmissionInfoFn(s.transmissionCalls)
+	}
+	panic("stubForwarderClient.GetTransmissionInfo not configured")
 }
 
 func (s *stubForwarderClient) GetReportProcessedEvents(context.Context, TransmissionID) ([]ReportProcessedEvent, error) {
-	if s.err != nil {
-		return nil, s.err
+	if s.eventsErr != nil {
+		return nil, s.eventsErr
 	}
 	return s.events, nil
 }
@@ -60,6 +74,7 @@ func TestEventDetailsList_String(t *testing.T) {
 	t.Parallel()
 
 	require.Equal(t, "[]", eventDetailsList(nil).String())
+	require.Equal(t, "[]", eventDetailsList{}.String())
 	require.Equal(t, "[hash=a ledger=1 result=success, hash=b ledger=2 result=failed]",
 		eventDetailsList{
 			{txHash: "a", ledger: 1, isSuccess: true},
@@ -116,7 +131,7 @@ func TestTxHashRetriever_GetSuccessfulTransmissionHash(t *testing.T) {
 
 	t.Run("returns error when event fetch fails", func(t *testing.T) {
 		t.Parallel()
-		client := &stubForwarderClient{err: errors.New("rpc down")}
+		client := &stubForwarderClient{eventsErr: errors.New("rpc down")}
 		retriever := NewTxHashRetriever(client, lggr, transmissionID)
 
 		ctx, cancel := context.WithTimeout(t.Context(), 300*time.Millisecond)