ingest/ledgerbackend: add trusted hash to captive core catchup

.github/workflows/horizon.yml

@@ -34,7 +34,7 @@ jobs:
       HORIZON_INTEGRATION_TESTS_CORE_MAX_SUPPORTED_PROTOCOL: ${{ matrix.protocol-version }}
       HORIZON_INTEGRATION_TESTS_CAPTIVE_CORE_USE_DB: true
       PROTOCOL_21_CORE_DEBIAN_PKG_VERSION: 21.0.0-1872.c6f474133.focal
-      PROTOCOL_21_CORE_DOCKER_IMG: stellar/stellar-core:21
+      PROTOCOL_21_CORE_DOCKER_IMG: stellar/stellar-core:21.2.1-1957.bbed247f6.focal
       PROTOCOL_21_SOROBAN_RPC_DOCKER_IMG: stellar/soroban-rpc:21.0.0-rc2-73
       PGHOST: localhost
       PGPORT: 5432

ingest/ledger_change_reader_test.go

@@ -23,7 +23,7 @@ const (
 
 func TestNewLedgerChangeReaderFails(t *testing.T) {
 	ctx := context.Background()
-	mock := &ledgerbackend.MockDatabaseBackend{}
+	mock := &ledgerbackend.MockLedgerBackend{}
 	seq := uint32(123)
 	mock.On("GetLedger", ctx, seq).Return(
 		xdr.LedgerCloseMeta{},
@@ -39,7 +39,7 @@ func TestNewLedgerChangeReaderFails(t *testing.T) {
 
 func TestNewLedgerChangeReaderSucceeds(t *testing.T) {
 	ctx := context.Background()
-	mock := &ledgerbackend.MockDatabaseBackend{}
+	mock := &ledgerbackend.MockLedgerBackend{}
 	seq := uint32(123)
 
 	header := xdr.LedgerHeaderHistoryEntry{
@@ -146,7 +146,7 @@ func assertChangesEqual(
 
 func TestLedgerChangeReaderOrder(t *testing.T) {
 	ctx := context.Background()
-	mock := &ledgerbackend.MockDatabaseBackend{}
+	mock := &ledgerbackend.MockLedgerBackend{}
 	seq := uint32(123)
 
 	src := xdr.MustAddress("GBXGQJWVLWOYHFLVTKWV5FGHA3LNYY2JQKM7OAJAUEQFU6LPCSEFVXON")
@@ -353,7 +353,7 @@ func TestLedgerChangeReaderOrder(t *testing.T) {
 
 func TestLedgerChangeLedgerCloseMetaV2(t *testing.T) {
 	ctx := context.Background()
-	mock := &ledgerbackend.MockDatabaseBackend{}
+	mock := &ledgerbackend.MockLedgerBackend{}
 	seq := uint32(123)
 
 	src := xdr.MustAddress("GBXGQJWVLWOYHFLVTKWV5FGHA3LNYY2JQKM7OAJAUEQFU6LPCSEFVXON")
@@ -600,7 +600,7 @@ func TestLedgerChangeLedgerCloseMetaV2(t *testing.T) {
 
 func TestLedgerChangeLedgerCloseMetaV2Empty(t *testing.T) {
 	ctx := context.Background()
-	mock := &ledgerbackend.MockDatabaseBackend{}
+	mock := &ledgerbackend.MockLedgerBackend{}
 	seq := uint32(123)
 
 	baseFee := xdr.Int64(100)

ingest/ledgerbackend/captive_core_backend.go

@@ -95,7 +95,8 @@ type CaptiveStellarCore struct {
 	stellarCoreLock sync.RWMutex
 
 	// For testing
-	stellarCoreRunnerFactory func() stellarCoreRunnerInterface
+	stellarCoreRunnerFactory  func() stellarCoreRunnerInterface
+	trustedHashBackendFactory func(config CaptiveCoreConfig) (LedgerBackend, error)
 
 	// cachedMeta keeps that ledger data of the last fetched ledger. Updated in GetLedger().
 	cachedMeta *xdr.LedgerCloseMeta
@@ -163,7 +164,7 @@ type CaptiveCoreConfig struct {
 }
 
 // NewCaptive returns a new CaptiveStellarCore instance.
-func NewCaptive(config CaptiveCoreConfig) (*CaptiveStellarCore, error) {
+func NewCaptive(config CaptiveCoreConfig) (LedgerBackend, error) {
 	// Here we set defaults in the config. Because config is not a pointer this code should
 	// not mutate the original CaptiveCoreConfig instance which was passed into NewCaptive()
 
@@ -233,6 +234,8 @@ func NewCaptive(config CaptiveCoreConfig) (*CaptiveStellarCore, error) {
 		return newStellarCoreRunner(config)
 	}
 
+	c.trustedHashBackendFactory = NewCaptive
+
 	if config.Toml != nil && config.Toml.HTTPPort != 0 {
 		c.stellarCoreClient = &stellarcore.Client{
 			HTTP: &http.Client{
@@ -327,6 +330,52 @@ func (c *CaptiveStellarCore) getLatestCheckpointSequence() (uint32, error) {
 	return has.CurrentLedger, nil
 }
 
+// Obtain hash for a ledger sequence from a live connection with network.
+// Will NOT use history archives for hash retrieval.
+// Returns the requested ledger sequence or the closest one to it that could be replayed
+// and its associated trusted hash.
+func (c *CaptiveStellarCore) getTrustedHashForLedger(sequence uint32) (uint32, string, error) {
+	// clone the current captive config but with unique context and hash store refs.
+	// create ad-hoc captive core instance to run unbounded range with requested sequence as start,
+	// this will trigger captive core online replay mode, obtaining the tx-meta and hash for the requested sequence
+	// first, direct from network which is considered trusted, after which the captive core instance is closed.
+	var trustedHashCancel context.CancelFunc
+	captiveConfigTrustedHash := c.config
+	captiveConfigTrustedHash.LedgerHashStore = nil
+	captiveConfigTrustedHash.Context, trustedHashCancel = context.WithCancel(c.config.Context)
+	captiveTrustedHash, err := c.trustedHashBackendFactory(captiveConfigTrustedHash)
+
+	if sequence <= 2 {
+		// The line below is to support minimum edge case for streaming ledgers from core in run 'from'
+		// earliest ledger it will emit on pipe will be 3.
+		sequence = 3
+	}
+
+	if err != nil {
+		trustedHashCancel()
+		return 0, "", errors.Wrapf(err, "error creating captive to get hash from network for Ledger %v", sequence)
+	}
+
+	defer func() {
+		trustedHashCancel()
+		if closeErr := captiveTrustedHash.Close(); closeErr != nil {
+			captiveConfigTrustedHash.Log.Error("error when closing captive core for network hash", closeErr)
+		}
+	}()
+
+	err = captiveTrustedHash.PrepareRange(captiveConfigTrustedHash.Context, UnboundedRange(sequence))
+	if err != nil {
+		return 0, "", errors.Wrapf(err, "error preparing to get network hash for Ledger %v", sequence)
+	}
+
+	networkLCM, err := captiveTrustedHash.GetLedger(captiveConfigTrustedHash.Context, sequence)
+	if err != nil {
+		return 0, "", errors.Wrapf(err, "error getting network hash for Ledger %v", sequence)
+	}
+
+	return sequence, networkLCM.LedgerHash().HexString(), nil
+}
+
 func (c *CaptiveStellarCore) openOfflineReplaySubprocess(from, to uint32) error {
 	latestCheckpointSequence, err := c.getLatestCheckpointSequence()
 	if err != nil {
@@ -349,8 +398,12 @@ func (c *CaptiveStellarCore) openOfflineReplaySubprocess(from, to uint32) error
 		)
 	}
 
+	toLedger, toLedgerHash, err := c.getTrustedHashForLedger(to)
+	if err != nil {
+		return err
+	}
 	stellarCoreRunner := c.stellarCoreRunnerFactory()
-	if err = stellarCoreRunner.catchup(from, to); err != nil {
+	if err = stellarCoreRunner.catchup(from, toLedger, toLedgerHash); err != nil {
 		return errors.Wrap(err, "error running stellar-core")
 	}
 	c.stellarCoreRunner = stellarCoreRunner
@@ -517,9 +570,6 @@ func (c *CaptiveStellarCore) startPreparingRange(ctx context.Context, ledgerRang
 //   - For BoundedRange it will start Stellar-Core in catchup mode.
 //   - For UnboundedRange it will first catchup to starting ledger and then run
 //     it normally (including connecting to the Stellar network).
-//
-// Please note that using a BoundedRange, currently, requires a full-trust on
-// history archive. This issue is being fixed in Stellar-Core.
 func (c *CaptiveStellarCore) PrepareRange(ctx context.Context, ledgerRange Range) error {
 	if alreadyPrepared, err := c.startPreparingRange(ctx, ledgerRange); err != nil {
 		return errors.Wrap(err, "error starting prepare range")