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/tx_hash_retriever.go

@@ -4,37 +4,77 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"strings"
+	"time"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 
 	capcommon "github.com/smartcontractkit/capabilities/chain_capabilities/common"
+	"github.com/smartcontractkit/capabilities/chain_capabilities/stellar/monitoring"
 )
 
 const (
-	reportProcessedTopicPrefix      = "forwarder_ReportProcessed"
-	defaultForwarderLookbackLedgers = int64(100)
-	failedToRetrieveTxHashErrorMsg  = "failed to retrieve tx hash for report"
+	failedToRetrieveTxHashErrorMsg = "failed to retrieve tx hash for report"
+)
+
+type TxHashRetrievalResult string
+
+const (
+	TxHashRetrievalResultFound             TxHashRetrievalResult = "Found"
+	TxHashRetrievalResultNotFound          TxHashRetrievalResult = "NotFound"
+	TxHashRetrievalResultFetchError        TxHashRetrievalResult = "FetchError"
+	TxHashRetrievalResultUnexpectedSuccess TxHashRetrievalResult = "UnexpectedSuccess"
+)
+
+type TxHashLookupType string
+
+const (
+	TxHashLookupTypeSuccessful TxHashLookupType = "SuccessfulTransmission"
+	TxHashLookupTypeFailed     TxHashLookupType = "FailedTransmission"
 )
 
 var ErrUnexpectedSuccessfulTransmission = errors.New("unexpected successful transmission")
 
 type TxHashRetriever struct {
-	forwarderClient CREForwarderClient
-	transmissionID  TransmissionID
-	lggr            logger.SugaredLogger
+	forwarderClient   CREForwarderClient
+	transmissionID    TransmissionID
+	lggr              logger.SugaredLogger
+	beholderProcessor beholder.ProtoProcessor
+	messageBuilder    *monitoring.MessageBuilder
+	telemetryContext  monitoring.TelemetryContext
+}
+
+type TxHashRetrieverOption func(*TxHashRetriever)
+
+func WithTxHashRetrieverMonitoring(
+	beholderProcessor beholder.ProtoProcessor,
+	messageBuilder *monitoring.MessageBuilder,
+	telemetryContext monitoring.TelemetryContext,
+) TxHashRetrieverOption {
+	return func(r *TxHashRetriever) {
+		r.beholderProcessor = beholderProcessor
+		r.messageBuilder = messageBuilder
+		r.telemetryContext = telemetryContext
+	}
 }
 
 func NewTxHashRetriever(
 	forwarderClient CREForwarderClient,
 	lggr logger.SugaredLogger,
 	transmissionID TransmissionID,
+	opts ...TxHashRetrieverOption,
 ) TxHashRetriever {
-	return TxHashRetriever{
+	r := TxHashRetriever{
 		forwarderClient: forwarderClient,
 		transmissionID:  transmissionID,
 		lggr:            lggr,
 	}
+	for _, opt := range opts {
+		opt(&r)
+	}
+	return r
 }
 
 type eventDetails struct {
@@ -65,16 +105,34 @@ func (l eventDetailsList) String() string {
 }
 
 func (r *TxHashRetriever) GetSuccessfulTransmissionHash(ctx context.Context) (string, error) {
-	details, err := r.fetchAndParseEvents(ctx)
+	details, phaseStart, err := r.fetchAndParseEvents(ctx, TxHashLookupTypeSuccessful)
 	if err != nil {
 		return "", err
 	}
 	for _, d := range details {
 		if d.isSuccess {
+			if r.monitoringEnabled() {
+				monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+					r.telemetryContext,
+					string(TxHashRetrievalResultFound),
+					int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+					d.txHash,
+					string(TxHashLookupTypeSuccessful),
+				))
+			}
 			return d.txHash, nil
 		}
 	}
 	r.lggr.Errorw("No successful transmission found", "txCount", len(details), "transactions", details.String())
+	if r.monitoringEnabled() {
+		monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+			r.telemetryContext,
+			string(TxHashRetrievalResultNotFound),
+			int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+			"",
+			string(TxHashLookupTypeSuccessful),
+		))
+	}
 	return "", fmt.Errorf("no successful transmission found. Found %d transactions (all failed): %s",
 		len(details), details)
 }
@@ -85,16 +143,37 @@ func (r *TxHashRetriever) GetFailedTransmissionHash(ctx context.Context) (string
 }
 
 func (r *TxHashRetriever) GetFailedTransmissionHashWithCount(ctx context.Context) (string, int, error) {
-	details, err := r.fetchAndParseEvents(ctx)
+	details, phaseStart, err := r.fetchAndParseEvents(ctx, TxHashLookupTypeFailed)
 	if err != nil {
 		return "", 0, err
 	}
 	for _, d := range details {
 		if d.isSuccess {
+			if r.monitoringEnabled() {
+				monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+					r.telemetryContext,
+					string(TxHashRetrievalResultUnexpectedSuccess),
+					int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+					d.txHash,
+					string(TxHashLookupTypeFailed),
+				))
+			}
 			return "", len(details), fmt.Errorf("%w, successful tx hash: %s",
 				ErrUnexpectedSuccessfulTransmission, d.txHash)
 		}
 	}
+	if len(details) == 0 {
+		if r.monitoringEnabled() {
+			monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+				r.telemetryContext,
+				string(TxHashRetrievalResultNotFound),
+				int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+				"",
+				string(TxHashLookupTypeFailed),
+			))
+		}
+		return "", 0, fmt.Errorf("no failed transmission found")
+	}
 
 	earliestIdx := 0
 	for i, d := range details {
@@ -103,17 +182,28 @@ 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
+	if r.monitoringEnabled() {
+		monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+			r.telemetryContext,
+			string(TxHashRetrievalResultFound),
+			int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+			selectedHash,
+			string(TxHashLookupTypeFailed),
+		))
+	}
+	return selectedHash, len(details), nil
 }
 
-func (r *TxHashRetriever) fetchAndParseEvents(ctx context.Context) (eventDetailsList, error) {
+func (r *TxHashRetriever) fetchAndParseEvents(ctx context.Context, lookupType TxHashLookupType) (eventDetailsList, time.Time, error) {
+	phaseStart := time.Now()
 	events, err := capcommon.WithPollingRetry(ctx, r.lggr, func(ctx context.Context) ([]ReportProcessedEvent, error) {
 		events, fetchErr := r.forwarderClient.GetReportProcessedEvents(ctx, r.transmissionID)
 		if fetchErr != nil {
@@ -125,10 +215,23 @@ func (r *TxHashRetriever) fetchAndParseEvents(ctx context.Context) (eventDetails
 		return events, nil
 	})
 	if err != nil {
-		return nil, fmt.Errorf("%s: %w", failedToRetrieveTxHashErrorMsg, err)
+		if r.monitoringEnabled() {
+			monitoring.EmitInitiated(ctx, r.lggr, r.beholderProcessor, r.messageBuilder.BuildWriteReportTxHashRetrievalPhase(
+				r.telemetryContext,
+				string(TxHashRetrievalResultFetchError),
+				int64(math.Max(float64(time.Since(phaseStart).Milliseconds()), 0)),
+				"",
+				string(lookupType),
+			))
+		}
+		return nil, phaseStart, fmt.Errorf("%s: %w", failedToRetrieveTxHashErrorMsg, err)
 	}
 
-	return buildEventDetails(events), nil
+	return buildEventDetails(events), phaseStart, nil
+}
+
+func (r *TxHashRetriever) monitoringEnabled() bool {
+	return r.beholderProcessor != nil && r.messageBuilder != nil
 }
 
 func buildEventDetails(events []ReportProcessedEvent) eventDetailsList {

chain_capabilities/stellar/actions/tx_hash_retriever_test.go

@@ -8,27 +8,45 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	stellartypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/stellar"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
+
+	"github.com/smartcontractkit/capabilities/chain_capabilities/stellar/monitoring"
 )
 
 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
 }
@@ -116,7 +134,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)
@@ -158,6 +176,52 @@ func TestTxHashRetriever_GetFailedTransmissionHash(t *testing.T) {
 		require.ErrorIs(t, err, ErrUnexpectedSuccessfulTransmission)
 	})
 
+	t.Run("emits single NotFound phase when all events failed", func(t *testing.T) {
+		t.Parallel()
+		processor := &recordingWriteReportProcessor{}
+		client := &stubForwarderClient{events: []ReportProcessedEvent{
+			{TxHash: "a", Ledger: 1, Success: false},
+			{TxHash: "b", Ledger: 2, Success: false},
+		}}
+		retriever := NewTxHashRetriever(
+			client,
+			lggr,
+			transmissionID,
+			WithTxHashRetrieverMonitoring(
+				processor,
+				monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""),
+				monitoring.TelemetryContext{},
+			),
+		)
+
+		_, err := retriever.GetSuccessfulTransmissionHash(t.Context())
+		require.Error(t, err)
+		require.Len(t, processor.messages, 1)
+		requireTxHashPhaseEvent(t, processor.messages, TxHashLookupTypeSuccessful, TxHashRetrievalResultNotFound, "")
+	})
+
+	t.Run("emits unexpected success phase telemetry", func(t *testing.T) {
+		t.Parallel()
+		processor := &recordingWriteReportProcessor{}
+		client := &stubForwarderClient{events: []ReportProcessedEvent{
+			{TxHash: testTxHash, Ledger: 100, Success: true},
+		}}
+		retriever := NewTxHashRetriever(
+			client,
+			lggr,
+			transmissionID,
+			WithTxHashRetrieverMonitoring(
+				processor,
+				monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""),
+				monitoring.TelemetryContext{},
+			),
+		)
+
+		_, err := retriever.GetFailedTransmissionHash(t.Context())
+		require.ErrorIs(t, err, ErrUnexpectedSuccessfulTransmission)
+		requireTxHashPhaseEvent(t, processor.messages, TxHashLookupTypeFailed, TxHashRetrievalResultUnexpectedSuccess, testTxHash)
+	})
+
 	t.Run("GetFailedTransmissionHashWithCount returns count", func(t *testing.T) {
 		t.Parallel()
 		client := &stubForwarderClient{events: []ReportProcessedEvent{