diff --git a/chain_capabilities/stellar/.mockery.yaml b/chain_capabilities/stellar/.mockery.yaml index 86e862182..48ab605ad 100644 --- a/chain_capabilities/stellar/.mockery.yaml +++ b/chain_capabilities/stellar/.mockery.yaml @@ -15,3 +15,12 @@ packages: filename: "test_stellar_service_mock.go" interfaces: StellarService: + github.com/smartcontractkit/capabilities/chain_capabilities/stellar/height: + config: + dir: "{{ .InterfaceDir }}/mocks" + mockname: "{{ .InterfaceName }}" + outpkg: mocks + inpackage: false + filename: "{{ .InterfaceName | snakecase }}.go" + interfaces: + LedgerProvider: diff --git a/chain_capabilities/stellar/actions/actions.go b/chain_capabilities/stellar/actions/actions.go index 82bcab648..9ce648c7e 100644 --- a/chain_capabilities/stellar/actions/actions.go +++ b/chain_capabilities/stellar/actions/actions.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "math" "strings" "time" @@ -23,10 +24,11 @@ import ( capcommon "github.com/smartcontractkit/capabilities/chain_capabilities/common" commonmon "github.com/smartcontractkit/capabilities/chain_capabilities/common/monitoring" ts "github.com/smartcontractkit/capabilities/chain_capabilities/common/transmission_schedule" - "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/metering" - "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/monitoring" "github.com/smartcontractkit/capabilities/libs/chainconsensus" ctypes "github.com/smartcontractkit/capabilities/libs/chainconsensus/types" + + "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/metering" + "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/monitoring" ) // Stellar implements the CRE capability actions for the Stellar chain. @@ -82,16 +84,38 @@ func (s *Stellar) Close() error { return services.CloseAll(s.reportSizeLimit) } -func (s *Stellar) GetLatestLedger(ctx context.Context, _ capabilities.RequestMetadata, _ *stellarcap.GetLatestLedgerRequest) (*capabilities.ResponseAndMetadata[*stellarcap.GetLatestLedgerResponse], caperrors.Error) { - resp, err := s.StellarService.GetLatestLedger(ctx) - if err != nil { - return nil, GetError(err, false) +// GetLatestLedger performs a consensus read of the current ledger. Each node +// observes its latest ledger and the DON aggregates via the OCR consensus handler +// (same path as ReadContract), keyed by the observed ledger sequence. +func (s *Stellar) GetLatestLedger(ctx context.Context, metadata capabilities.RequestMetadata, _ *stellarcap.GetLatestLedgerRequest) (*capabilities.ResponseAndMetadata[*stellarcap.GetLatestLedgerResponse], caperrors.Error) { + s.lggr.Debug("Received GetLatestLedger request") + + observe := func(_ context.Context, height *ctypes.ChainHeight) (*stellarcap.GetLatestLedgerResponse, error) { + if height == nil || height.Latest <= 0 { + return nil, fmt.Errorf("no agreed chain height available for GetLatestLedger consensus") + } + // stellar ledger sequences are uint32 on-chain + if height.Latest > math.MaxUint32 { + return nil, fmt.Errorf("agreed ledger sequence %d exceeds uint32", height.Latest) + } + // TODO PLEX-3243 implement a by sequence ledger fetch to populate block metadata + return &stellarcap.GetLatestLedgerResponse{Sequence: uint32(height.Latest)}, nil } - protoResp, err := stellarcap.ConvertGetLatestLedgerResponseToProto(resp) + + request := ctypes.NewLockableToBlockHashableRequest( + metadata.WorkflowExecutionID, + metadata.ReferenceID, + metering.GetResponseMetadata(metering.GetLatestLedger), + observe, + ) + + responseAndMetadata, err := chainconsensus.ReadHashableRequestReport(ctx, s.handler, request) if err != nil { - return nil, GetError(err, false) + s.lggr.Errorw("Failed to GetLatestLedger", "error", err) + return nil, capcommon.GetError(fmt.Errorf("failed to GetLatestLedger: %w", err), false) + } - return &capabilities.ResponseAndMetadata[*stellarcap.GetLatestLedgerResponse]{Response: protoResp}, nil + return responseAndMetadata, nil } // ReadContract performs a consensus read of a read-only Soroban contract call. diff --git a/chain_capabilities/stellar/actions/actions_test.go b/chain_capabilities/stellar/actions/actions_test.go index 71e726bd3..f0313c444 100644 --- a/chain_capabilities/stellar/actions/actions_test.go +++ b/chain_capabilities/stellar/actions/actions_test.go @@ -117,28 +117,18 @@ func TestGetLatestLedger(t *testing.T) { t.Run("happy path", func(t *testing.T) { t.Parallel() helper := newMockedStellar(t) - helper.stellarService.EXPECT(). - GetLatestLedger(mock.Anything). - Return(stellartypes.GetLatestLedgerResponse{ - Sequence: 123, - LedgerCloseTime: 1_700_000_000, - }, nil). - Once() + helper.stellar.handler = testConsensusHandler{handle: runLockableToBlockHandle(&ctypes.ChainHeight{Latest: 123})} resp, err := helper.stellar.GetLatestLedger(t.Context(), capabilities.RequestMetadata{}, &stellarcap.GetLatestLedgerRequest{}) require.NoError(t, err) require.NotNil(t, resp) require.Equal(t, uint32(123), resp.Response.GetSequence()) - require.Equal(t, int64(1_700_000_000), resp.Response.GetLedgerCloseTime()) }) - t.Run("service error", func(t *testing.T) { + t.Run("no agreed height", func(t *testing.T) { t.Parallel() helper := newMockedStellar(t) - helper.stellarService.EXPECT(). - GetLatestLedger(mock.Anything). - Return(stellartypes.GetLatestLedgerResponse{}, errors.New("node unavailable")). - Once() + helper.stellar.handler = testConsensusHandler{handle: runLockableToBlockHandle(&ctypes.ChainHeight{Latest: 0})} _, err := helper.stellar.GetLatestLedger(t.Context(), capabilities.RequestMetadata{}, &stellarcap.GetLatestLedgerRequest{}) require.Error(t, err) @@ -346,3 +336,49 @@ func runVolatileHashableHandle(ctx context.Context, req ctypes.Request) (<-chan ch <- reply return ch, nil } + +// runLockableToBlockHandle locks the request to the given agreed ChainHeight (as the +// real handler does after height agreement), then observes the resulting hashable request. +func runLockableToBlockHandle(chainHeight *ctypes.ChainHeight) func(context.Context, ctypes.Request) (<-chan ctypes.Reply, error) { + return func(ctx context.Context, req ctypes.Request) (<-chan ctypes.Reply, error) { + if lockable, ok := req.(interface { + LockToABlock(*ctypes.ChainHeight) ctypes.Request + }); ok { + req = lockable.LockToABlock(chainHeight) + } + + 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: + if len(tObs.Hashable) != ctypes.HashLength { + return nil, fmt.Errorf("unexpected hashable length: got %d, want %d", len(tObs.Hashable), ctypes.HashLength) + } + var rd ctypes.Hash + copy(rd[:], tObs.Hashable) + reply = ctypes.Reply{Value: ctypes.NewHashableRequestReport(ocrtypes.ConfigDigest{}, 0, rd, 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 + } +} diff --git a/chain_capabilities/stellar/block_provider.go b/chain_capabilities/stellar/block_provider.go deleted file mode 100644 index 3b412bbe7..000000000 --- a/chain_capabilities/stellar/block_provider.go +++ /dev/null @@ -1,10 +0,0 @@ -package main - -// noopBlocksProvider returns 0 for all blocks. Stellar consensus reads are volatile (keyed by -// ledger sequence inside the observation) and are never locked to a block height, so the OCR -// reporting plugin does not need a real chain-height provider. -type noopBlocksProvider struct{} - -func (n noopBlocksProvider) GetLatest() int64 { return 0 } -func (n noopBlocksProvider) GetSafe() int64 { return 0 } -func (n noopBlocksProvider) GetFinalized() int64 { return 0 } diff --git a/chain_capabilities/stellar/go.mod b/chain_capabilities/stellar/go.mod index b56025c35..38cd9df44 100644 --- a/chain_capabilities/stellar/go.mod +++ b/chain_capabilities/stellar/go.mod @@ -114,7 +114,7 @@ require ( go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/goleak v1.3.0 // indirect go.uber.org/multierr v1.11.0 // indirect - go.uber.org/zap v1.27.1 // indirect + go.uber.org/zap v1.27.1 go.yaml.in/yaml/v2 v2.4.4 // indirect golang.org/x/crypto v0.52.0 // indirect golang.org/x/exp v0.0.0-20260508232706-74f9aab9d74a // indirect diff --git a/chain_capabilities/stellar/height/mocks/ledger_provider.go b/chain_capabilities/stellar/height/mocks/ledger_provider.go new file mode 100644 index 000000000..03c5b3000 --- /dev/null +++ b/chain_capabilities/stellar/height/mocks/ledger_provider.go @@ -0,0 +1,93 @@ +// Code generated by mockery v2.53.5. DO NOT EDIT. + +package mocks + +import ( + context "context" + + stellar "github.com/smartcontractkit/chainlink-common/pkg/types/chains/stellar" + mock "github.com/stretchr/testify/mock" +) + +// LedgerProvider is an autogenerated mock type for the LedgerProvider type +type LedgerProvider struct { + mock.Mock +} + +type LedgerProvider_Expecter struct { + mock *mock.Mock +} + +func (_m *LedgerProvider) EXPECT() *LedgerProvider_Expecter { + return &LedgerProvider_Expecter{mock: &_m.Mock} +} + +// GetLatestLedger provides a mock function with given fields: ctx +func (_m *LedgerProvider) GetLatestLedger(ctx context.Context) (stellar.GetLatestLedgerResponse, error) { + ret := _m.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for GetLatestLedger") + } + + var r0 stellar.GetLatestLedgerResponse + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) (stellar.GetLatestLedgerResponse, error)); ok { + return rf(ctx) + } + if rf, ok := ret.Get(0).(func(context.Context) stellar.GetLatestLedgerResponse); ok { + r0 = rf(ctx) + } else { + r0 = ret.Get(0).(stellar.GetLatestLedgerResponse) + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(ctx) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// LedgerProvider_GetLatestLedger_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetLatestLedger' +type LedgerProvider_GetLatestLedger_Call struct { + *mock.Call +} + +// GetLatestLedger is a helper method to define mock.On call +// - ctx context.Context +func (_e *LedgerProvider_Expecter) GetLatestLedger(ctx interface{}) *LedgerProvider_GetLatestLedger_Call { + return &LedgerProvider_GetLatestLedger_Call{Call: _e.mock.On("GetLatestLedger", ctx)} +} + +func (_c *LedgerProvider_GetLatestLedger_Call) Run(run func(ctx context.Context)) *LedgerProvider_GetLatestLedger_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context)) + }) + return _c +} + +func (_c *LedgerProvider_GetLatestLedger_Call) Return(_a0 stellar.GetLatestLedgerResponse, _a1 error) *LedgerProvider_GetLatestLedger_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *LedgerProvider_GetLatestLedger_Call) RunAndReturn(run func(context.Context) (stellar.GetLatestLedgerResponse, error)) *LedgerProvider_GetLatestLedger_Call { + _c.Call.Return(run) + return _c +} + +// NewLedgerProvider creates a new instance of LedgerProvider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewLedgerProvider(t interface { + mock.TestingT + Cleanup(func()) +}) *LedgerProvider { + mock := &LedgerProvider{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/chain_capabilities/stellar/height/provider.go b/chain_capabilities/stellar/height/provider.go new file mode 100644 index 000000000..d59ba577c --- /dev/null +++ b/chain_capabilities/stellar/height/provider.go @@ -0,0 +1,104 @@ +package height + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" + stellartypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/stellar" +) + +// LedgerProvider is the subset of the Stellar chain service used to poll the latest closed ledger sequence. +type LedgerProvider interface { + GetLatestLedger(ctx context.Context) (stellartypes.GetLatestLedgerResponse, error) +} + +// Provider is a service that polls the latest closed Stellar ledger sequence. +type Provider struct { + services.Service + engine *services.Engine + + lggr logger.SugaredLogger + pollPeriod time.Duration + service LedgerProvider + + mutex sync.RWMutex + sequence int64 +} + +// NewProvider builds a Stellar height provider +func NewProvider(lggr logger.Logger, pollPeriod time.Duration, service LedgerProvider) (*Provider, error) { + if pollPeriod <= 0 { + return nil, fmt.Errorf("height provider poll period must be positive, got %s", pollPeriod) + } + if service == nil { + return nil, fmt.Errorf("height provider requires a non-nil ledger service") + } + b := &Provider{ + pollPeriod: pollPeriod, + service: service, + } + b.Service, b.engine = services.Config{ + Name: "StellarHeightProvider", + Start: b.start, + Close: b.close, + }.NewServiceEngine(lggr) + b.lggr = b.engine.SugaredLogger + return b, nil +} + +func (b *Provider) start(_ context.Context) error { + b.engine.Go(b.poll) + return nil +} + +func (b *Provider) close() error { return nil } + +func (b *Provider) poll(ctx context.Context) { + b.pollLedger(ctx) + ticker := time.NewTicker(b.pollPeriod) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + b.pollLedger(ctx) + } + } +} + +func (b *Provider) pollLedger(ctx context.Context) { + resp, err := b.service.GetLatestLedger(ctx) + if err != nil { + b.lggr.Errorw("failed to poll latest ledger", "error", err) + return + } + b.mutex.Lock() + defer b.mutex.Unlock() + // Ledger sequence is monotonically increasing; retain the max so a lagging RPC + // can't lower the agreed height across rounds. + if seq := int64(resp.Sequence); seq > b.sequence { + b.sequence = seq + } +} + +// GetLatest returns the latest observed ledger sequence. +func (b *Provider) GetLatest() int64 { return b.get() } + +// GetSafe returns the safe ledger sequence. Stellar ledgers are final on close, so +// this equals GetLatest. +func (b *Provider) GetSafe() int64 { return b.get() } + +// GetFinalized returns the finalized ledger sequence. Stellar ledgers are final on +// close, so this equals GetLatest. +func (b *Provider) GetFinalized() int64 { return b.get() } + +func (b *Provider) get() int64 { + b.mutex.RLock() + defer b.mutex.RUnlock() + return b.sequence +} diff --git a/chain_capabilities/stellar/height/provider_test.go b/chain_capabilities/stellar/height/provider_test.go new file mode 100644 index 000000000..bd4455975 --- /dev/null +++ b/chain_capabilities/stellar/height/provider_test.go @@ -0,0 +1,92 @@ +package height + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + stellartypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/stellar" + "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" + + "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/height/mocks" +) + +func ledgerResp(seq uint32) stellartypes.GetLatestLedgerResponse { + return stellartypes.GetLatestLedgerResponse{Sequence: seq} +} + +func TestProvider(t *testing.T) { + lggr, _ := logger.TestObserved(t, zapcore.DebugLevel) + + t.Run("reports latest ledger sequence for all tags", func(t *testing.T) { + svc := mocks.NewLedgerProvider(t) + svc.EXPECT().GetLatestLedger(mock.Anything).Return(ledgerResp(42), nil) + + p, err := NewProvider(lggr, 50*time.Millisecond, svc) + require.NoError(t, err) + require.NoError(t, p.Start(t.Context())) + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + // Stellar ledgers are final on close, so latest == safe == finalized. + tests.AssertEventually(t, func() bool { return p.GetLatest() == 42 }) + require.Equal(t, int64(42), p.GetSafe()) + require.Equal(t, int64(42), p.GetFinalized()) + }) + + t.Run("retains max sequence (monotonic, ignores a lagging RPC)", func(t *testing.T) { + svc := mocks.NewLedgerProvider(t) + var calls int + svc.EXPECT().GetLatestLedger(mock.Anything). + RunAndReturn(func(context.Context) (stellartypes.GetLatestLedgerResponse, error) { + calls++ + if calls == 1 { + return ledgerResp(100), nil + } + return ledgerResp(90), nil // lagging RPC reports a lower sequence + }) + + p, err := NewProvider(lggr, time.Hour, svc) + require.NoError(t, err) + + // Drive the poll deterministically (no ticker/timing) via the in-package method. + p.pollLedger(t.Context()) + require.Equal(t, int64(100), p.GetLatest()) + p.pollLedger(t.Context()) + require.Equal(t, int64(100), p.GetLatest(), "must not drop below the max observed sequence") + }) + + t.Run("service error leaves height unchanged", func(t *testing.T) { + svc := mocks.NewLedgerProvider(t) + svc.EXPECT().GetLatestLedger(mock.Anything). + Return(stellartypes.GetLatestLedgerResponse{}, errors.New("node unavailable")) + + p, err := NewProvider(lggr, time.Hour, svc) + require.NoError(t, err) + + p.pollLedger(t.Context()) + require.Equal(t, int64(0), p.GetLatest()) + }) +} + +func TestNewProvider_Validation(t *testing.T) { + lggr, _ := logger.TestObserved(t, zapcore.DebugLevel) + + t.Run("rejects non-positive poll period", func(t *testing.T) { + svc := mocks.NewLedgerProvider(t) + _, err := NewProvider(lggr, 0, svc) + require.Error(t, err) + require.Contains(t, err.Error(), "poll period must be positive") + }) + + t.Run("rejects nil service", func(t *testing.T) { + _, err := NewProvider(lggr, time.Second, nil) + require.Error(t, err) + require.Contains(t, err.Error(), "non-nil ledger service") + }) +} diff --git a/chain_capabilities/stellar/main.go b/chain_capabilities/stellar/main.go index c8e4ea78f..ec45dfb03 100644 --- a/chain_capabilities/stellar/main.go +++ b/chain_capabilities/stellar/main.go @@ -18,8 +18,10 @@ import ( "github.com/smartcontractkit/capabilities/libs/loopserver" ts "github.com/smartcontractkit/capabilities/chain_capabilities/common/transmission_schedule" + "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/actions" "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/config" + "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/height" "github.com/smartcontractkit/capabilities/chain_capabilities/stellar/monitoring" "github.com/smartcontractkit/chainlink-common/pkg/capabilities" @@ -57,6 +59,7 @@ type capability struct { requestPoller *poller.Poller consensusHandler chainconsensus.Handler oracle core.Oracle + blocksProvider *height.Provider id string } @@ -101,6 +104,9 @@ func (c *capabilityGRPCService) Close() error { if c.consensusHandler != nil { closers = append(closers, c.consensusHandler) } + if c.blocksProvider != nil { + closers = append(closers, c.blocksProvider) + } if c.Stellar != nil { closers = append(closers, c.Stellar) } @@ -177,6 +183,10 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor } c.requestPoller = poller.NewPoller(c.lggr, consensusMetrics, cfg.ObservationPollerWorkersCount, cfg.ObservationPollPeriod) c.consensusHandler = chainconsensus.NewHandler(c.lggr, c.requestPoller, consensusMetrics, cfg.UnknownRequestsTTL) + c.blocksProvider, err = height.NewProvider(c.lggr, cfg.ObservationPollPeriod, stellarService) + if err != nil { + return fmt.Errorf("failed to create stellar height provider: %w", err) + } c.oracle, err = dependencies.OracleFactory.NewOracle(ctx, core.OracleArgs{ LocalConfig: ocrtypes.LocalConfig{ BlockchainTimeout: time.Second * 20, @@ -187,7 +197,7 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor ContractConfigLoadTimeout: time.Second * 10, DefaultMaxDurationInitialization: time.Second * 10, }, - ReportingPluginFactoryService: oracle.NewReportingPluginFactory(logger.Sugared(c.lggr), c.consensusHandler, noopBlocksProvider{}, consensusMetrics), + ReportingPluginFactoryService: oracle.NewReportingPluginFactory(logger.Sugared(c.lggr), c.consensusHandler, c.blocksProvider, consensusMetrics), ContractTransmitter: oracle.NewContractTransmitter(c.lggr, c.consensusHandler), }) if err != nil { @@ -227,7 +237,7 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor return err } - for _, service := range []interface{ Start(context.Context) error }{c.requestPoller, c.consensusHandler, c.oracle} { + for _, service := range []interface{ Start(context.Context) error }{c.requestPoller, c.consensusHandler, c.blocksProvider, c.oracle} { if err = service.Start(ctx); err != nil { return err } diff --git a/chain_capabilities/stellar/metering/metering.go b/chain_capabilities/stellar/metering/metering.go index eae9dcd51..0aac1ee19 100644 --- a/chain_capabilities/stellar/metering/metering.go +++ b/chain_capabilities/stellar/metering/metering.go @@ -15,6 +15,10 @@ const ( // TODO: PLEX-3022 - replace with actual values. ReadContract SpendValueCredits = "1" + // GetLatestLedger is the placeholder spend value for a GetLatestLedger consensus read. + // TODO: PLEX-3022 - replace with actual values. + GetLatestLedger SpendValueCredits = "1" + // WriteReportSpendUnitFormat is the spend unit for write operations, parameterised by chain selector. WriteReportSpendUnitFormat = "STROOP.%d" )