Merge pull request stellar#5101 from stellar/ingestion-next

horizon/services/ingest: merge ingestion-next to master

services/horizon/docker/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3'
 services:
   horizon-postgres:
     platform: linux/amd64
-    image: postgres:12-bullseye
+    image: postgres:postgres:12-bullseye
     restart: on-failure
     environment:
       - POSTGRES_HOST_AUTH_METHOD=trust

services/horizon/docker/verify-range/Dockerfile

@@ -6,7 +6,7 @@ ENV STELLAR_CORE_VERSION=${STELLAR_CORE_VERSION:-*}
 ENV DEBIAN_FRONTEND=noninteractive
 
 ADD dependencies /
-RUN ["chmod", "+x", "dependencies"]
+RUN ["chmod", "+x", "/dependencies"]
 RUN /dependencies
 
 ADD stellar-core.cfg /

services/horizon/internal/action_offers_test.go

@@ -24,7 +24,9 @@ func TestOfferActions_Show(t *testing.T) {
 	ht.Assert.NoError(err)
 
 	ledgerCloseTime := time.Now().Unix()
-	_, err = q.InsertLedger(ctx, xdr.LedgerHeaderHistoryEntry{
+	ht.Assert.NoError(q.Begin(ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 100,
 			ScpValue: xdr.StellarValue{
@@ -33,6 +35,8 @@ func TestOfferActions_Show(t *testing.T) {
 		},
 	}, 0, 0, 0, 0, 0)
 	ht.Assert.NoError(err)
+	ht.Assert.NoError(ledgerBatch.Exec(ht.Ctx, q))
+	ht.Assert.NoError(q.Commit())
 
 	issuer := xdr.MustAddress("GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H")
 	nativeAsset := xdr.MustNewNativeAsset()

services/horizon/internal/actions/account_test.go

@@ -228,7 +228,9 @@ func TestAccountInfo(t *testing.T) {
 	assert.NoError(t, err)
 
 	ledgerFourCloseTime := time.Now().Unix()
-	_, err = q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
+	assert.NoError(t, q.Begin(tt.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 4,
 			ScpValue: xdr.StellarValue{
@@ -237,6 +239,8 @@ func TestAccountInfo(t *testing.T) {
 		},
 	}, 0, 0, 0, 0, 0)
 	assert.NoError(t, err)
+	assert.NoError(t, ledgerBatch.Exec(tt.Ctx, q))
+	assert.NoError(t, q.Commit())
 
 	account, err := AccountInfo(tt.Ctx, &history.Q{tt.HorizonSession()}, accountID)
 	tt.Assert.NoError(err)
@@ -408,7 +412,9 @@ func TestGetAccountsHandlerPageResultsByAsset(t *testing.T) {
 	err := q.UpsertAccounts(tt.Ctx, []history.AccountEntry{account1, account2})
 	assert.NoError(t, err)
 	ledgerCloseTime := time.Now().Unix()
-	_, err = q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
+	assert.NoError(t, q.Begin(tt.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 1234,
 			ScpValue: xdr.StellarValue{
@@ -417,6 +423,8 @@ func TestGetAccountsHandlerPageResultsByAsset(t *testing.T) {
 		},
 	}, 0, 0, 0, 0, 0)
 	assert.NoError(t, err)
+	assert.NoError(t, ledgerBatch.Exec(tt.Ctx, q))
+	assert.NoError(t, q.Commit())
 
 	for _, row := range accountSigners {
 		_, err = q.CreateAccountSigner(tt.Ctx, row.Account, row.Signer, row.Weight, nil)
@@ -511,7 +519,9 @@ func TestGetAccountsHandlerPageResultsByLiquidityPool(t *testing.T) {
 	assert.NoError(t, err)
 
 	ledgerCloseTime := time.Now().Unix()
-	_, err = q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
+	assert.NoError(t, q.Begin(tt.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 1234,
 			ScpValue: xdr.StellarValue{
@@ -520,6 +530,9 @@ func TestGetAccountsHandlerPageResultsByLiquidityPool(t *testing.T) {
 		},
 	}, 0, 0, 0, 0, 0)
 	assert.NoError(t, err)
+	assert.NoError(t, ledgerBatch.Exec(tt.Ctx, q))
+	assert.NoError(t, q.Commit())
+
 	var assetType, code, issuer string
 	usd.MustExtract(&assetType, &code, &issuer)
 	params := map[string]string{

services/horizon/internal/actions/offer_test.go

@@ -79,15 +79,19 @@ func TestGetOfferByIDHandler(t *testing.T) {
 	handler := GetOfferByID{}
 
 	ledgerCloseTime := time.Now().Unix()
-	_, err := q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
+	assert.NoError(t, q.Begin(tt.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err := ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 3,
 			ScpValue: xdr.StellarValue{
 				CloseTime: xdr.TimePoint(ledgerCloseTime),
 			},
 		},
 	}, 0, 0, 0, 0, 0)
-	tt.Assert.NoError(err)
+	assert.NoError(t, err)
+	assert.NoError(t, ledgerBatch.Exec(tt.Ctx, q))
+	assert.NoError(t, q.Commit())
 
 	err = q.UpsertOffers(tt.Ctx, []history.Offer{eurOffer, usdOffer})
 	tt.Assert.NoError(err)
@@ -186,15 +190,19 @@ func TestGetOffersHandler(t *testing.T) {
 	handler := GetOffersHandler{}
 
 	ledgerCloseTime := time.Now().Unix()
-	_, err := q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
+	assert.NoError(t, q.Begin(tt.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err := ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 3,
 			ScpValue: xdr.StellarValue{
 				CloseTime: xdr.TimePoint(ledgerCloseTime),
 			},
 		},
 	}, 0, 0, 0, 0, 0)
-	tt.Assert.NoError(err)
+	assert.NoError(t, err)
+	assert.NoError(t, ledgerBatch.Exec(tt.Ctx, q))
+	assert.NoError(t, q.Commit())
 
 	err = q.UpsertOffers(tt.Ctx, []history.Offer{eurOffer, twoEurOffer, usdOffer})
 	tt.Assert.NoError(err)

services/horizon/internal/actions/operation_test.go

@@ -36,17 +36,21 @@ func TestInvokeHostFnDetailsInPaymentOperations(t *testing.T) {
 	opID1 := toid.New(sequence, txIndex, 1).ToInt64()
 
 	ledgerCloseTime := time.Now().Unix()
-	_, err := q.InsertLedger(tt.Ctx, xdr.LedgerHeaderHistoryEntry{
-		Header: xdr.LedgerHeader{
-			LedgerSeq: xdr.Uint32(sequence),
-			ScpValue: xdr.StellarValue{
-				CloseTime: xdr.TimePoint(ledgerCloseTime),
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err := ledgerBatch.Add(
+		xdr.LedgerHeaderHistoryEntry{
+			Header: xdr.LedgerHeader{
+				LedgerSeq: xdr.Uint32(sequence),
+				ScpValue: xdr.StellarValue{
+					CloseTime: xdr.TimePoint(ledgerCloseTime),
+				},
 			},
-		},
-	}, 1, 0, 1, 0, 0)
+		}, 1, 0, 1, 0, 0)
 	tt.Assert.NoError(err)
+	tt.Assert.NoError(q.Begin(tt.Ctx))
+	tt.Assert.NoError(ledgerBatch.Exec(tt.Ctx, q))
 
-	transactionBuilder := q.NewTransactionBatchInsertBuilder(1)
+	transactionBuilder := q.NewTransactionBatchInsertBuilder()
 	firstTransaction := buildLedgerTransaction(tt.T, testTransaction{
 		index:         uint32(txIndex),
 		envelopeXDR:   "AAAAACiSTRmpH6bHC6Ekna5e82oiGY5vKDEEUgkq9CB//t+rAAAAyAEXUhsAADDRAAAAAAAAAAAAAAABAAAAAAAAAAsBF1IbAABX4QAAAAAAAAAA",
@@ -55,11 +59,13 @@ func TestInvokeHostFnDetailsInPaymentOperations(t *testing.T) {
 		metaXDR:       "AAAAAQAAAAAAAAAA",
 		hash:          "19aaa18db88605aedec04659fb45e06f240b022eb2d429e05133e4d53cd945ba",
 	})
-	err = transactionBuilder.Add(tt.Ctx, firstTransaction, uint32(sequence))
+	err = transactionBuilder.Add(firstTransaction, uint32(sequence))
 	tt.Assert.NoError(err)
+	tt.Assert.NoError(transactionBuilder.Exec(tt.Ctx, q))
+
+	operationBuilder := q.NewOperationBatchInsertBuilder()
 
-	operationBuilder := q.NewOperationBatchInsertBuilder(1)
-	err = operationBuilder.Add(tt.Ctx,
+	err = operationBuilder.Add(
 		opID1,
 		txID,
 		1,
@@ -118,6 +124,8 @@ func TestInvokeHostFnDetailsInPaymentOperations(t *testing.T) {
 		null.String{},
 		true)
 	tt.Assert.NoError(err)
+	tt.Assert.NoError(operationBuilder.Exec(tt.Ctx, q))
+	tt.Assert.NoError(q.Commit())
 
 	records, err := handler.GetResourcePage(
 		httptest.NewRecorder(),

services/horizon/internal/actions/transaction_test.go

@@ -149,6 +149,7 @@ func checkOuterHashResponse(
 }
 
 func TestFeeBumpTransactionPage(t *testing.T) {
+
 	tt := test.Start(t)
 	defer tt.Finish()
 	test.ResetHorizonDB(t, tt.HorizonDB)

services/horizon/internal/actions_account_test.go

@@ -18,12 +18,17 @@ func TestAccountActions_InvalidID(t *testing.T) {
 	ht.Assert.NoError(err)
 	err = q.UpdateIngestVersion(ht.Ctx, ingest.CurrentVersion)
 	ht.Assert.NoError(err)
-	_, err = q.InsertLedger(ht.Ctx, xdr.LedgerHeaderHistoryEntry{
+
+	ht.Assert.NoError(q.Begin(ht.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 100,
 		},
 	}, 0, 0, 0, 0, 0)
 	ht.Assert.NoError(err)
+	ht.Assert.NoError(ledgerBatch.Exec(ht.Ctx, q))
+	ht.Assert.NoError(q.Commit())
 
 	// existing account
 	w := ht.Get(

services/horizon/internal/actions_data_test.go

@@ -44,12 +44,16 @@ func TestDataActions_Show(t *testing.T) {
 	ht.Assert.NoError(err)
 	err = q.UpdateIngestVersion(ht.Ctx, ingest.CurrentVersion)
 	ht.Assert.NoError(err)
-	_, err = q.InsertLedger(ht.Ctx, xdr.LedgerHeaderHistoryEntry{
+	ht.Assert.NoError(q.Begin(ht.Ctx))
+	ledgerBatch := q.NewLedgerBatchInsertBuilder()
+	err = ledgerBatch.Add(xdr.LedgerHeaderHistoryEntry{
 		Header: xdr.LedgerHeader{
 			LedgerSeq: 100,
 		},
 	}, 0, 0, 0, 0, 0)
 	ht.Assert.NoError(err)
+	ht.Assert.NoError(ledgerBatch.Exec(ht.Ctx, q))
+	ht.Assert.NoError(q.Commit())
 
 	err = q.UpsertAccountData(ht.Ctx, []history.Data{data1, data2})
 	assert.NoError(t, err)

services/horizon/internal/actions_trade_test.go

@@ -820,8 +820,11 @@ func IngestTestTrade(
 		return err
 	}
 
-	batch := q.NewTradeBatchInsertBuilder(0)
-	batch.Add(ctx, history.InsertTrade{
+	if err = q.Begin(ctx); err != nil {
+		return err
+	}
+	batch := q.NewTradeBatchInsertBuilder()
+	err = batch.Add(history.InsertTrade{
 		HistoryOperationID: opCounter,
 		Order:              0,
 		CounterAssetID:     assets[assetBought.String()].ID,
@@ -839,7 +842,10 @@ func IngestTestTrade(
 
 		Type: history.OrderbookTradeType,
 	})
-	err = batch.Exec(ctx)
+	if err != nil {
+		return err
+	}
+	err = batch.Exec(ctx, q)
 	if err != nil {
 		return err
 	}
@@ -849,6 +855,10 @@ func IngestTestTrade(
 		return err
 	}
 
+	if err := q.Commit(); err != nil {
+		return err
+	}
+
 	return nil
 }