services/horizon/internal/ingest/processors: Refactor liquidity pools, trades, and claimable balances processors to support new ingestion data flow

services/horizon/internal/db2/history/account_loader.go

@@ -28,7 +28,7 @@ const loaderLookupBatchSize = 50000
 
 // Value implements the database/sql/driver Valuer interface.
 func (a FutureAccountID) Value() (driver.Value, error) {
-	return a.loader.getNow(a.address), nil
+	return a.loader.GetNow(a.address), nil
 }
 
 // AccountLoader will map account addresses to their history
@@ -67,11 +67,11 @@ func (a *AccountLoader) GetFuture(address string) FutureAccountID {
 	}
 }
 
-// getNow returns the history account id for the given address.
-// getNow should only be called on values which were registered by
-// GetFuture() calls. Also, Exec() must be called before any getNow
+// GetNow returns the history account id for the given address.
+// GetNow should only be called on values which were registered by
+// GetFuture() calls. Also, Exec() must be called before any GetNow
 // call can succeed.
-func (a *AccountLoader) getNow(address string) int64 {
+func (a *AccountLoader) GetNow(address string) int64 {
 	if id, ok := a.ids[address]; !ok {
 		panic(fmt.Errorf("address %v not present", address))
 	} else {
@@ -190,3 +190,20 @@ func bulkInsert(ctx context.Context, q *Q, table string, conflictFields []string
 	)
 	return err
 }
+
+// AccountLoaderStub is a stub wrapper around AccountLoader which allows
+// you to manually configure the mapping of addresses to history account ids
+type AccountLoaderStub struct {
+	Loader *AccountLoader
+}
+
+// NewAccountLoaderStub returns a new AccountLoaderStub instance
+func NewAccountLoaderStub() AccountLoaderStub {
+	return AccountLoaderStub{Loader: NewAccountLoader()}
+}
+
+// Insert updates the wrapped AccountLoader so that the given account
+// address is mapped to the provided history account id
+func (a AccountLoaderStub) Insert(address string, id int64) {
+	a.Loader.ids[address] = id
+}

services/horizon/internal/db2/history/account_loader_test.go

@@ -27,7 +27,7 @@ func TestAccountLoader(t *testing.T) {
 		future := loader.GetFuture(address)
 		futures = append(futures, future)
 		assert.Panics(t, func() {
-			loader.getNow(address)
+			loader.GetNow(address)
 		})
 		assert.Panics(t, func() {
 			future.Value()
@@ -44,7 +44,7 @@ func TestAccountLoader(t *testing.T) {
 	q := &Q{session}
 	for i, address := range addresses {
 		future := futures[i]
-		id := loader.getNow(address)
+		id := loader.GetNow(address)
 		val, err := future.Value()
 		assert.NoError(t, err)
 		assert.Equal(t, id, val)

services/horizon/internal/db2/history/asset_loader.go

@@ -31,7 +31,7 @@ type FutureAssetID struct {
 
 // Value implements the database/sql/driver Valuer interface.
 func (a FutureAssetID) Value() (driver.Value, error) {
-	return a.loader.getNow(a.asset), nil
+	return a.loader.GetNow(a.asset), nil
 }
 
 // AssetLoader will map assets to their history
@@ -67,11 +67,11 @@ func (a *AssetLoader) GetFuture(asset AssetKey) FutureAssetID {
 	}
 }
 
-// getNow returns the history asset id for the given asset.
-// getNow should only be called on values which were registered by
-// GetFuture() calls. Also, Exec() must be called before any getNow
+// GetNow returns the history asset id for the given asset.
+// GetNow should only be called on values which were registered by
+// GetFuture() calls. Also, Exec() must be called before any GetNow
 // call can succeed.
-func (a *AssetLoader) getNow(asset AssetKey) int64 {
+func (a *AssetLoader) GetNow(asset AssetKey) int64 {
 	if id, ok := a.ids[asset]; !ok {
 		panic(fmt.Errorf("asset %v not present", asset))
 	} else {
@@ -186,3 +186,20 @@ func (a *AssetLoader) Exec(ctx context.Context, session db.SessionInterface) err
 
 	return a.lookupKeys(ctx, q, keys)
 }
+
+// AssetLoaderStub is a stub wrapper around AssetLoader which allows
+// you to manually configure the mapping of assets to history asset ids
+type AssetLoaderStub struct {
+	Loader *AssetLoader
+}
+
+// NewAssetLoaderStub returns a new AssetLoaderStub instance
+func NewAssetLoaderStub() AssetLoaderStub {
+	return AssetLoaderStub{Loader: NewAssetLoader()}
+}
+
+// Insert updates the wrapped AssetLoaderStub so that the given asset
+// address is mapped to the provided history asset id
+func (a AssetLoaderStub) Insert(asset AssetKey, id int64) {
+	a.Loader.ids[asset] = id
+}

services/horizon/internal/db2/history/asset_loader_test.go

@@ -41,7 +41,7 @@ func TestAssetLoader(t *testing.T) {
 		future := loader.GetFuture(key)
 		futures = append(futures, future)
 		assert.Panics(t, func() {
-			loader.getNow(key)
+			loader.GetNow(key)
 		})
 		assert.Panics(t, func() {
 			future.Value()
@@ -58,7 +58,7 @@ func TestAssetLoader(t *testing.T) {
 	q := &Q{session}
 	for i, key := range keys {
 		future := futures[i]
-		internalID := loader.getNow(key)
+		internalID := loader.GetNow(key)
 		val, err := future.Value()
 		assert.NoError(t, err)
 		assert.Equal(t, internalID, val)

services/horizon/internal/db2/history/liquidity_pool_loader.go

@@ -23,7 +23,7 @@ type FutureLiquidityPoolID struct {
 
 // Value implements the database/sql/driver Valuer interface.
 func (a FutureLiquidityPoolID) Value() (driver.Value, error) {
-	return a.loader.getNow(a.id), nil
+	return a.loader.GetNow(a.id), nil
 }
 
 // LiquidityPoolLoader will map liquidity pools to their internal
@@ -60,11 +60,11 @@ func (a *LiquidityPoolLoader) GetFuture(id string) FutureLiquidityPoolID {
 	}
 }
 
-// getNow returns the internal history id for the given liquidity pool.
-// getNow should only be called on values which were registered by
-// GetFuture() calls. Also, Exec() must be called before any getNow
+// GetNow returns the internal history id for the given liquidity pool.
+// GetNow should only be called on values which were registered by
+// GetFuture() calls. Also, Exec() must be called before any GetNow
 // call can succeed.
-func (a *LiquidityPoolLoader) getNow(id string) int64 {
+func (a *LiquidityPoolLoader) GetNow(id string) int64 {
 	if id, ok := a.ids[id]; !ok {
 		panic(fmt.Errorf("id %v not present", id))
 	} else {
@@ -141,3 +141,20 @@ func (a *LiquidityPoolLoader) Exec(ctx context.Context, session db.SessionInterf
 
 	return a.lookupKeys(ctx, q, ids)
 }
+
+// LiquidityPoolLoaderStub is a stub wrapper around LiquidityPoolLoader which allows
+// you to manually configure the mapping of liquidity pools to history liquidity ppol ids
+type LiquidityPoolLoaderStub struct {
+	Loader *LiquidityPoolLoader
+}
+
+// NewLiquidityPoolLoaderStub returns a new LiquidityPoolLoader instance
+func NewLiquidityPoolLoaderStub() LiquidityPoolLoaderStub {
+	return LiquidityPoolLoaderStub{Loader: NewLiquidityPoolLoader()}
+}
+
+// Insert updates the wrapped LiquidityPoolLoader so that the given liquidity pool
+// is mapped to the provided history liquidity pool id
+func (a LiquidityPoolLoaderStub) Insert(lp string, id int64) {
+	a.Loader.ids[lp] = id
+}

services/horizon/internal/db2/history/liquidity_pool_loader_test.go

@@ -30,7 +30,7 @@ func TestLiquidityPoolLoader(t *testing.T) {
 		future := loader.GetFuture(id)
 		futures = append(futures, future)
 		assert.Panics(t, func() {
-			loader.getNow(id)
+			loader.GetNow(id)
 		})
 		assert.Panics(t, func() {
 			future.Value()
@@ -47,7 +47,7 @@ func TestLiquidityPoolLoader(t *testing.T) {
 	q := &Q{session}
 	for i, id := range ids {
 		future := futures[i]
-		internalID := loader.getNow(id)
+		internalID := loader.GetNow(id)
 		val, err := future.Value()
 		assert.NoError(t, err)
 		assert.Equal(t, internalID, val)

services/horizon/internal/db2/history/operation_participant_batch_insert_builder_test.go

@@ -43,6 +43,6 @@ func TestAddOperationParticipants(t *testing.T) {
 
 		op := ops[0]
 		tt.Assert.Equal(int64(240518172673), op.OperationID)
-		tt.Assert.Equal(accountLoader.getNow(address), op.AccountID)
+		tt.Assert.Equal(accountLoader.GetNow(address), op.AccountID)
 	}
 }

services/horizon/internal/db2/history/participants_test.go

@@ -63,7 +63,7 @@ func TestTransactionParticipantsBatch(t *testing.T) {
 		{TransactionID: 2},
 	}
 	for i := range expected {
-		expected[i].AccountID = accountLoader.getNow(addresses[i])
+		expected[i].AccountID = accountLoader.GetNow(addresses[i])
 	}
 	tt.Assert.ElementsMatch(expected, participants)
 }

services/horizon/internal/ingest/processors/claimable_balances_transaction_processor.go

@@ -138,7 +138,7 @@ func (p *ClaimableBalancesTransactionProcessor) addOperationClaimableBalances(
 	return nil
 }
 
-func (p *ClaimableBalancesTransactionProcessor) Commit(ctx context.Context, session db.SessionInterface) error {
+func (p *ClaimableBalancesTransactionProcessor) Flush(ctx context.Context, session db.SessionInterface) error {
 	err := p.txBatch.Exec(ctx, session)
 	if err != nil {
 		return err

services/horizon/internal/ingest/processors/claimable_balances_transaction_processor_test.go

@@ -62,7 +62,7 @@ func (s *ClaimableBalancesTransactionProcessorTestSuiteLedger) TestEmptyClaimabl
 	s.mockTransactionBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 	s.mockOperationBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 
-	s.Assert().NoError(s.processor.Commit(s.ctx, s.mockSession))
+	s.Assert().NoError(s.processor.Flush(s.ctx, s.mockSession))
 }
 
 func (s *ClaimableBalancesTransactionProcessorTestSuiteLedger) testOperationInserts(balanceID xdr.ClaimableBalanceId, body xdr.OperationBody, change xdr.LedgerEntryChange) {
@@ -138,7 +138,7 @@ func (s *ClaimableBalancesTransactionProcessorTestSuiteLedger) testOperationInse
 	// Process the transaction
 	err = s.processor.ProcessTransaction(s.lcm, txn)
 	s.Assert().NoError(err)
-	err = s.processor.Commit(s.ctx, s.mockSession)
+	err = s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().NoError(err)
 }
 

services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor.go

@@ -139,7 +139,7 @@ func (p *LiquidityPoolsTransactionProcessor) addOperationLiquidityPools(sequence
 	return nil
 }
 
-func (p *LiquidityPoolsTransactionProcessor) Commit(ctx context.Context, session db.SessionInterface) error {
+func (p *LiquidityPoolsTransactionProcessor) Flush(ctx context.Context, session db.SessionInterface) error {
 	if err := p.txBatch.Exec(ctx, session); err != nil {
 		return errors.Wrap(err, "Could not flush transaction liquidity pools to db")
 	}

services/horizon/internal/ingest/processors/liquidity_pools_transaction_processor_test.go

@@ -62,7 +62,7 @@ func (s *LiquidityPoolsTransactionProcessorTestSuiteLedger) TestEmptyLiquidityPo
 	s.mockTransactionBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 	s.mockOperationBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 
-	err := s.processor.Commit(context.Background(), s.mockSession)
+	err := s.processor.Flush(context.Background(), s.mockSession)
 	s.Assert().NoError(err)
 }
 
@@ -138,7 +138,7 @@ func (s *LiquidityPoolsTransactionProcessorTestSuiteLedger) testOperationInserts
 	// Process the transaction
 	err := s.processor.ProcessTransaction(s.lcm, txn)
 	s.Assert().NoError(err)
-	err = s.processor.Commit(s.ctx, s.mockSession)
+	err = s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().NoError(err)
 }
 

services/horizon/internal/ingest/processors/participants_processor.go

@@ -155,7 +155,7 @@ func (p *ParticipantsProcessor) ProcessTransaction(lcm xdr.LedgerCloseMeta, tran
 	return nil
 }
 
-func (p *ParticipantsProcessor) Commit(ctx context.Context, session db.SessionInterface) error {
+func (p *ParticipantsProcessor) Flush(ctx context.Context, session db.SessionInterface) error {
 	if err := p.txBatch.Exec(ctx, session); err != nil {
 		return errors.Wrap(err, "Could not flush transaction participants to db")
 	}

services/horizon/internal/ingest/processors/participants_processor_test.go

@@ -145,7 +145,7 @@ func (s *ParticipantsProcessorTestSuiteLedger) TestEmptyParticipants() {
 	s.mockBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 	s.mockOperationsBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 
-	err := s.processor.Commit(s.ctx, s.mockSession)
+	err := s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().NoError(err)
 }
 
@@ -190,7 +190,7 @@ func (s *ParticipantsProcessorTestSuiteLedger) TestFeeBumptransaction() {
 	s.mockOperationsBatchInsertBuilder.On("Exec", s.ctx, s.mockSession).Return(nil).Once()
 
 	s.Assert().NoError(s.processor.ProcessTransaction(s.lcm, feeBumpTx))
-	s.Assert().NoError(s.processor.Commit(s.ctx, s.mockSession))
+	s.Assert().NoError(s.processor.Flush(s.ctx, s.mockSession))
 }
 
 func (s *ParticipantsProcessorTestSuiteLedger) TestIngestParticipantsSucceeds() {
@@ -204,7 +204,7 @@ func (s *ParticipantsProcessorTestSuiteLedger) TestIngestParticipantsSucceeds()
 		err := s.processor.ProcessTransaction(s.lcm, tx)
 		s.Assert().NoError(err)
 	}
-	err := s.processor.Commit(s.ctx, s.mockSession)
+	err := s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().NoError(err)
 }
 
@@ -240,7 +240,7 @@ func (s *ParticipantsProcessorTestSuiteLedger) TestBatchAddExecFails() {
 		err := s.processor.ProcessTransaction(s.lcm, tx)
 		s.Assert().NoError(err)
 	}
-	err := s.processor.Commit(s.ctx, s.mockSession)
+	err := s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().EqualError(err, "Could not flush transaction participants to db: transient error")
 }
 
@@ -255,6 +255,6 @@ func (s *ParticipantsProcessorTestSuiteLedger) TestOperationBatchAddExecFails()
 		err := s.processor.ProcessTransaction(s.lcm, tx)
 		s.Assert().NoError(err)
 	}
-	err := s.processor.Commit(s.ctx, s.mockSession)
+	err := s.processor.Flush(s.ctx, s.mockSession)
 	s.Assert().EqualError(err, "Could not flush operation participants to db: transient error")
 }