Enable hash based stellar consensus for latestledger

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:

chain_capabilities/stellar/actions/actions.go

@@ -4,6 +4,7 @@
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"strings"
 	"time"
 
@@ -23,10 +24,11 @@
 	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 @@
 	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.

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
+	}
+}

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

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
+}