Repository: stellar/go
Pull request #5104 (2023-11-02 19:55:52)

services/horizon: Use COPY to speed up ClaimableBalanceChangeProcessor
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What
Use [FastBatchInsertBuilder](https://github.com/stellar/go/blob/9ce82de12cbade47de862ad5e9d0684029558564/support/db/fast_batch_insert_builder.go) for writing to claimable_balances and claimable_balance_claimants tables. FastBatchInsertBuilder uses ‘COPY' to insert into the db which is faster than using 'INSERT'.

### Why
https://github.com/stellar/go/issues/5086

### Known limitations
N/A


urvisavla: Since we'll be updating [all](https://github.com/stellar/go/issues/5098) change processors to use `FastBatchInsertBuilder`, I changed the change processor interface to accept the database session required for `FastBatchInsertBuilder` in both the `Commit` and `ProcessChange` (internally calls `Commit`) methods. This meant that I had to pass nil to the database session for the [StatsChangeProcessor](https://github.com/stellar/go/pull/5104/files#diff-bee14211e4439838b73cf0d9091990c8d02e3fe140e01348954666cc57ba97ebR81). Even though StatsChangeProcessor implements the change processor interface, it is different from other change processors in that It updates in-mem counters instead of storing it in the database. 

This solution works, but I am not sure about the design. Here are some other ideas I had:
1. Create a separate interface for the StatsChangeProcessor because it is different from other change processors and handle it separately in the processor runner [RunAllProcessorsOnLedger](https://github.com/stellar/go/blob/4670509a3e872359fbe11f6ee0383ecf883e8ace/services/horizon/internal/ingest/processor_runner.go#L394)  and [RunHistoryArchiveIngestion](https://github.com/stellar/go/blame/4670509a3e872359fbe11f6ee0383ecf883e8ace/services/horizon/internal/ingest/processor_runner.go#L233) 
2. Create a separate interface for the StatsChangeProcessor and pass an instance of StatsChangeProcessor to each of the change processors, so they can update the stats.
3. Pass the database session to the constructors of the change processors that need it instead of modifying `ProcessChange` and `Commit`. This would mean that each processor would be bound to a session throughout its lifetime, which might be okay in our case.

Patch:
diff --git a/services/horizon/internal/db2/history/claimable_balance_batch_insert_builder.go b/services/horizon/internal/db2/history/claimable_balance_batch_insert_builder.go
new file mode 100644
index 00000000..515cc164
--- /dev/null
+++ b/services/horizon/internal/db2/history/claimable_balance_batch_insert_builder.go
@@ -0,0 +1,40 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stellar/go/support/db"
+)
+
+// ClaimableBalanceBatchInsertBuilder is used to insert claimable balance into the
+// claimable_balances table
+type ClaimableBalanceBatchInsertBuilder interface {
+	Add(claimableBalance ClaimableBalance) error
+	Exec(ctx context.Context) error
+}
+
+// ClaimableBalanceBatchInsertBuilder is a simple wrapper around db.FastBatchInsertBuilder
+type claimableBalanceBatchInsertBuilder struct {
+	session db.SessionInterface
+	builder db.FastBatchInsertBuilder
+	table   string
+}
+
+// NewClaimableBalanceBatchInsertBuilder constructs a new ClaimableBalanceBatchInsertBuilder instance
+func (q *Q) NewClaimableBalanceBatchInsertBuilder() ClaimableBalanceBatchInsertBuilder {
+	return &claimableBalanceBatchInsertBuilder{
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "claimable_balances",
+	}
+}
+
+// Add adds a new claimable balance to the batch
+func (i *claimableBalanceBatchInsertBuilder) Add(claimableBalance ClaimableBalance) error {
+	return i.builder.RowStruct(claimableBalance)
+}
+
+// Exec writes the batch of claimable balances to the database.
+func (i *claimableBalanceBatchInsertBuilder) Exec(ctx context.Context) error {
+	return i.builder.Exec(ctx, i.session, i.table)
+}
diff --git a/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go b/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go
index a1a13d84..2f4dc673 100644
--- a/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go
+++ b/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go
@@ -4,39 +4,37 @@ import (
 	"context"
 
 	"github.com/stellar/go/support/db"
-	"github.com/stellar/go/xdr"
 )
 
-// ClaimableBalanceClaimantBatchInsertBuilder is used to insert transactions into the
-// history_transactions table
+// ClaimableBalanceClaimantBatchInsertBuilder is used to insert claimants into the
+// claimable_balance_claimants table
 type ClaimableBalanceClaimantBatchInsertBuilder interface {
-	Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error
+	Add(claimableBalanceClaimant ClaimableBalanceClaimant) error
 	Exec(ctx context.Context) error
 }
 
-// ClaimableBalanceClaimantBatchInsertBuilder is a simple wrapper around db.BatchInsertBuilder
+// ClaimableBalanceClaimantBatchInsertBuilder is a simple wrapper around db.FastBatchInsertBuilder
 type claimableBalanceClaimantBatchInsertBuilder struct {
-	encodingBuffer *xdr.EncodingBuffer
-	builder        db.BatchInsertBuilder
+	session db.SessionInterface
+	builder db.FastBatchInsertBuilder
+	table   string
 }
 
 // NewClaimableBalanceClaimantBatchInsertBuilder constructs a new ClaimableBalanceClaimantBatchInsertBuilder instance
-func (q *Q) NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder {
+func (q *Q) NewClaimableBalanceClaimantBatchInsertBuilder() ClaimableBalanceClaimantBatchInsertBuilder {
 	return &claimableBalanceClaimantBatchInsertBuilder{
-		encodingBuffer: xdr.NewEncodingBuffer(),
-		builder: db.BatchInsertBuilder{
-			Table:        q.GetTable("claimable_balance_claimants"),
-			MaxBatchSize: maxBatchSize,
-			Suffix:       "ON CONFLICT (id, destination) DO UPDATE SET last_modified_ledger=EXCLUDED.last_modified_ledger",
-		},
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "claimable_balance_claimants",
 	}
 }
 
-// Add adds a new transaction to the batch
-func (i *claimableBalanceClaimantBatchInsertBuilder) Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error {
-	return i.builder.RowStruct(ctx, claimableBalanceClaimant)
+// Add adds a new claimant to the batch
+func (i *claimableBalanceClaimantBatchInsertBuilder) Add(claimableBalanceClaimant ClaimableBalanceClaimant) error {
+	return i.builder.RowStruct(claimableBalanceClaimant)
 }
 
+// Exec writes the batch of claimants to the database.
 func (i *claimableBalanceClaimantBatchInsertBuilder) Exec(ctx context.Context) error {
-	return i.builder.Exec(ctx)
+	return i.builder.Exec(ctx, i.session, i.table)
 }
diff --git a/services/horizon/internal/db2/history/claimable_balances.go b/services/horizon/internal/db2/history/claimable_balances.go
index 48722a11..5490bef1 100644
--- a/services/horizon/internal/db2/history/claimable_balances.go
+++ b/services/horizon/internal/db2/history/claimable_balances.go
@@ -107,7 +107,11 @@ type ClaimableBalance struct {
 type Claimants []Claimant
 
 func (c Claimants) Value() (driver.Value, error) {
-	return json.Marshal(c)
+	// Convert the byte array into a string as a workaround to bypass buggy encoding in the pq driver
+	// (More info about this bug here https://github.com/stellar/go/issues/5086#issuecomment-1773215436).
+	// By doing so, the data will be written as a string rather than hex encoded bytes.
+	val, err := json.Marshal(c)
+	return string(val), err
 }
 
 func (c *Claimants) Scan(value interface{}) error {
@@ -126,12 +130,12 @@ type Claimant struct {
 
 // QClaimableBalances defines claimable-balance-related related queries.
 type QClaimableBalances interface {
-	UpsertClaimableBalances(ctx context.Context, cb []ClaimableBalance) error
 	RemoveClaimableBalances(ctx context.Context, ids []string) (int64, error)
 	RemoveClaimableBalanceClaimants(ctx context.Context, ids []string) (int64, error)
 	GetClaimableBalancesByID(ctx context.Context, ids []string) ([]ClaimableBalance, error)
 	CountClaimableBalances(ctx context.Context) (int, error)
-	NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder
+	NewClaimableBalanceClaimantBatchInsertBuilder() ClaimableBalanceClaimantBatchInsertBuilder
+	NewClaimableBalanceBatchInsertBuilder() ClaimableBalanceBatchInsertBuilder
 	GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (map[string][]ClaimableBalanceClaimant, error)
 }
 
@@ -171,36 +175,6 @@ func (q *Q) GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (
 	return claimantsMap, err
 }
 
-// UpsertClaimableBalances upserts a batch of claimable balances in the claimable_balances table.
-// There's currently no limit of the number of offers this method can
-// accept other than 2GB limit of the query string length what should be enough
-// for each ledger with the current limits.
-func (q *Q) UpsertClaimableBalances(ctx context.Context, cbs []ClaimableBalance) error {
-	var id, claimants, asset, amount, sponsor, lastModifiedLedger, flags []interface{}
-
-	for _, cb := range cbs {
-		id = append(id, cb.BalanceID)
-		claimants = append(claimants, cb.Claimants)
-		asset = append(asset, cb.Asset)
-		amount = append(amount, cb.Amount)
-		sponsor = append(sponsor, cb.Sponsor)
-		lastModifiedLedger = append(lastModifiedLedger, cb.LastModifiedLedger)
-		flags = append(flags, cb.Flags)
-	}
-
-	upsertFields := []upsertField{
-		{"id", "text", id},
-		{"claimants", "jsonb", claimants},
-		{"asset", "text", asset},
-		{"amount", "bigint", amount},
-		{"sponsor", "text", sponsor},
-		{"last_modified_ledger", "integer", lastModifiedLedger},
-		{"flags", "int", flags},
-	}
-
-	return q.upsertRows(ctx, "claimable_balances", "id", upsertFields)
-}
-
 // RemoveClaimableBalances deletes claimable balances table.
 // Returns number of rows affected and error.
 func (q *Q) RemoveClaimableBalances(ctx context.Context, ids []string) (int64, error) {
diff --git a/services/horizon/internal/db2/history/mock_claimable_balance_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_claimable_balance_batch_insert_builder.go
new file mode 100644
index 00000000..0c1a1e70
--- /dev/null
+++ b/services/horizon/internal/db2/history/mock_claimable_balance_batch_insert_builder.go
@@ -0,0 +1,21 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type MockClaimableBalanceBatchInsertBuilder struct {
+	mock.Mock
+}
+
+func (m *MockClaimableBalanceBatchInsertBuilder) Add(claimableBalance ClaimableBalance) error {
+	a := m.Called(claimableBalance)
+	return a.Error(0)
+}
+
+func (m *MockClaimableBalanceBatchInsertBuilder) Exec(ctx context.Context) error {
+	a := m.Called(ctx)
+	return a.Error(0)
+}
diff --git a/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go
index 63e2788e..3c8fb24d 100644
--- a/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go
+++ b/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go
@@ -2,7 +2,6 @@ package history
 
 import (
 	"context"
-
 	"github.com/stretchr/testify/mock"
 )
 
@@ -10,8 +9,8 @@ type MockClaimableBalanceClaimantBatchInsertBuilder struct {
 	mock.Mock
 }
 
-func (m *MockClaimableBalanceClaimantBatchInsertBuilder) Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error {
-	a := m.Called(ctx, claimableBalanceClaimant)
+func (m *MockClaimableBalanceClaimantBatchInsertBuilder) Add(claimableBalanceClaimant ClaimableBalanceClaimant) error {
+	a := m.Called(claimableBalanceClaimant)
 	return a.Error(0)
 }
 
diff --git a/services/horizon/internal/db2/history/mock_q_claimable_balances.go b/services/horizon/internal/db2/history/mock_q_claimable_balances.go
index 2493d9eb..64b65cf1 100644
--- a/services/horizon/internal/db2/history/mock_q_claimable_balances.go
+++ b/services/horizon/internal/db2/history/mock_q_claimable_balances.go
@@ -21,11 +21,6 @@ func (m *MockQClaimableBalances) GetClaimableBalancesByID(ctx context.Context, i
 	return a.Get(0).([]ClaimableBalance), a.Error(1)
 }
 
-func (m *MockQClaimableBalances) UpsertClaimableBalances(ctx context.Context, cbs []ClaimableBalance) error {
-	a := m.Called(ctx, cbs)
-	return a.Error(0)
-}
-
 func (m *MockQClaimableBalances) RemoveClaimableBalances(ctx context.Context, ids []string) (int64, error) {
 	a := m.Called(ctx, ids)
 	return a.Get(0).(int64), a.Error(1)
@@ -36,11 +31,16 @@ func (m *MockQClaimableBalances) RemoveClaimableBalanceClaimants(ctx context.Con
 	return a.Get(0).(int64), a.Error(1)
 }
 
-func (m *MockQClaimableBalances) NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder {
-	a := m.Called(maxBatchSize)
+func (m *MockQClaimableBalances) NewClaimableBalanceClaimantBatchInsertBuilder() ClaimableBalanceClaimantBatchInsertBuilder {
+	a := m.Called()
 	return a.Get(0).(ClaimableBalanceClaimantBatchInsertBuilder)
 }
 
+func (m *MockQClaimableBalances) NewClaimableBalanceBatchInsertBuilder() ClaimableBalanceBatchInsertBuilder {
+	a := m.Called()
+	return a.Get(0).(ClaimableBalanceBatchInsertBuilder)
+}
+
 func (m *MockQClaimableBalances) GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (map[string][]ClaimableBalanceClaimant, error) {
 	a := m.Called(ctx, ids)
 	return a.Get(0).(map[string][]ClaimableBalanceClaimant), a.Error(1)
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
index a9299273..5d985e69 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
@@ -2,6 +2,7 @@ package processors
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
@@ -10,10 +11,11 @@ import (
 )
 
 type ClaimableBalancesChangeProcessor struct {
-	encodingBuffer         *xdr.EncodingBuffer
-	qClaimableBalances     history.QClaimableBalances
-	cache                  *ingest.ChangeCompactor
-	claimantsInsertBuilder history.ClaimableBalanceClaimantBatchInsertBuilder
+	encodingBuffer                *xdr.EncodingBuffer
+	qClaimableBalances            history.QClaimableBalances
+	cache                         *ingest.ChangeCompactor
+	claimantsInsertBuilder        history.ClaimableBalanceClaimantBatchInsertBuilder
+	claimableBalanceInsertBuilder history.ClaimableBalanceBatchInsertBuilder
 }
 
 func NewClaimableBalancesChangeProcessor(Q history.QClaimableBalances) *ClaimableBalancesChangeProcessor {
@@ -27,7 +29,8 @@ func NewClaimableBalancesChangeProcessor(Q history.QClaimableBalances) *Claimabl
 
 func (p *ClaimableBalancesChangeProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
-	p.claimantsInsertBuilder = p.qClaimableBalances.NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize)
+	p.claimantsInsertBuilder = p.qClaimableBalances.NewClaimableBalanceClaimantBatchInsertBuilder()
+	p.claimableBalanceInsertBuilder = p.qClaimableBalances.NewClaimableBalanceBatchInsertBuilder()
 }
 
 func (p *ClaimableBalancesChangeProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
@@ -45,15 +48,14 @@ func (p *ClaimableBalancesChangeProcessor) ProcessChange(ctx context.Context, ch
 		if err != nil {
 			return errors.Wrap(err, "error in Commit")
 		}
-		p.reset()
 	}
 
 	return nil
 }
 
 func (p *ClaimableBalancesChangeProcessor) Commit(ctx context.Context) error {
+	defer p.reset()
 	var (
-		cbsToUpsert   []history.ClaimableBalance
 		cbIDsToDelete []string
 	)
 	changes := p.cache.GetChanges()
@@ -61,11 +63,27 @@ func (p *ClaimableBalancesChangeProcessor) Commit(ctx context.Context) error {
 		switch {
 		case change.Pre == nil && change.Post != nil:
 			// Created
-			row, err := p.ledgerEntryToRow(change.Post)
+			cb, err := p.ledgerEntryToRow(change.Post)
 			if err != nil {
 				return err
 			}
-			cbsToUpsert = append(cbsToUpsert, row)
+			// Add claimable balance
+			if err := p.claimableBalanceInsertBuilder.Add(cb); err != nil {
+				return errors.Wrap(err, "error adding to ClaimableBalanceBatchInsertBuilder")
+			}
+
+			// Add claimants
+			for _, claimant := range cb.Claimants {
+				claimant := history.ClaimableBalanceClaimant{
+					BalanceID:          cb.BalanceID,
+					Destination:        claimant.Destination,
+					LastModifiedLedger: cb.LastModifiedLedger,
+				}
+
+				if err := p.claimantsInsertBuilder.Add(claimant); err != nil {
+					return errors.Wrap(err, "error adding to ClaimableBalanceClaimantBatchInsertBuilder")
+				}
+			}
 		case change.Pre != nil && change.Post == nil:
 			// Removed
 			cBalance := change.Pre.Data.MustClaimableBalance()
@@ -75,38 +93,19 @@ func (p *ClaimableBalancesChangeProcessor) Commit(ctx context.Context) error {
 			}
 			cbIDsToDelete = append(cbIDsToDelete, id)
 		default:
-			// Updated
-			row, err := p.ledgerEntryToRow(change.Post)
-			if err != nil {
-				return err
-			}
-			cbsToUpsert = append(cbsToUpsert, row)
+			// claimable balance can only be created or removed
+			return fmt.Errorf("invalid change entry for a claimable balance was detected")
 		}
 	}
 
-	if len(cbsToUpsert) > 0 {
-		if err := p.qClaimableBalances.UpsertClaimableBalances(ctx, cbsToUpsert); err != nil {
-			return errors.Wrap(err, "error executing upsert")
-		}
-
-		// Add ClaimableBalanceClaimants
-		for _, cb := range cbsToUpsert {
-			for _, claimant := range cb.Claimants {
-				claimant := history.ClaimableBalanceClaimant{
-					BalanceID:          cb.BalanceID,
-					Destination:        claimant.Destination,
-					LastModifiedLedger: cb.LastModifiedLedger,
-				}
-				if err := p.claimantsInsertBuilder.Add(ctx, claimant); err != nil {
-					return errors.Wrap(err, "error adding to claimantsInsertBuilder")
-				}
-			}
-		}
+	err := p.claimantsInsertBuilder.Exec(ctx)
+	if err != nil {
+		return errors.Wrap(err, "error executing ClaimableBalanceClaimantBatchInsertBuilder")
+	}
 
-		err := p.claimantsInsertBuilder.Exec(ctx)
-		if err != nil {
-			return errors.Wrap(err, "error executing claimantsInsertBuilder")
-		}
+	err = p.claimableBalanceInsertBuilder.Exec(ctx)
+	if err != nil {
+		return errors.Wrap(err, "error executing ClaimableBalanceBatchInsertBuilder")
 	}
 
 	if len(cbIDsToDelete) > 0 {


Test patch:
diff --git a/services/horizon/internal/actions/claimable_balance_test.go b/services/horizon/internal/actions/claimable_balance_test.go
index ff59b961..fa922b93 100644
--- a/services/horizon/internal/actions/claimable_balance_test.go
+++ b/services/horizon/internal/actions/claimable_balance_test.go
@@ -1,6 +1,7 @@
 package actions
 
 import (
+	"database/sql"
 	"net/http/httptest"
 	"testing"
 
@@ -18,7 +19,12 @@ func TestGetClaimableBalanceByID(t *testing.T) {
 	tt := test.Start(t)
 	defer tt.Finish()
 	test.ResetHorizonDB(t, tt.HorizonDB)
-	q := &history.Q{tt.HorizonSession()}
+	q := &history.Q{SessionInterface: tt.HorizonSession()}
+
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
 
 	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	asset := xdr.MustNewCreditAsset("USD", accountID)
@@ -43,20 +49,22 @@ func TestGetClaimableBalanceByID(t *testing.T) {
 		LastModifiedLedger: 123,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []history.ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	balanceInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance))
 
-	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder()
 	for _, claimant := range cBalance.Claimants {
 		claimant := history.ClaimableBalanceClaimant{
 			BalanceID:          cBalance.BalanceID,
 			Destination:        claimant.Destination,
 			LastModifiedLedger: cBalance.LastModifiedLedger,
 		}
-		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-		tt.Assert.NoError(err)
+		tt.Assert.NoError(claimantsInsertBuilder.Add(claimant))
 	}
 
+	tt.Assert.NoError(balanceInsertBuilder.Exec(tt.Ctx))
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
+
 	handler := GetClaimableBalanceByIDHandler{}
 	response, err := handler.GetResource(httptest.NewRecorder(), makeRequest(
 		t,
@@ -148,6 +156,11 @@ func TestGetClaimableBalances(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &history.Q{tt.HorizonSession()}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
+
 	entriesMeta := []struct {
 		id        xdr.Hash
 		accountID string
@@ -187,25 +200,25 @@ func TestGetClaimableBalances(t *testing.T) {
 		hCBs = append(hCBs, cb)
 	}
 
-	err := q.UpsertClaimableBalances(tt.Ctx, hCBs)
-	tt.Assert.NoError(err)
+	balanceInsertbuilder := q.NewClaimableBalanceBatchInsertBuilder()
 
-	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder()
 
 	for _, cBalance := range hCBs {
+		tt.Assert.NoError(balanceInsertbuilder.Add(cBalance))
+
 		for _, claimant := range cBalance.Claimants {
 			claimant := history.ClaimableBalanceClaimant{
 				BalanceID:          cBalance.BalanceID,
 				Destination:        claimant.Destination,
 				LastModifiedLedger: cBalance.LastModifiedLedger,
 			}
-			err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-			tt.Assert.NoError(err)
+			tt.Assert.NoError(claimantsInsertBuilder.Add(claimant))
 		}
 	}
 
-	err = claimantsInsertBuilder.Exec(tt.Ctx)
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(balanceInsertbuilder.Exec(tt.Ctx))
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
 
 	handler := GetClaimableBalancesHandler{}
 	response, err := handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
@@ -284,11 +297,9 @@ func TestGetClaimableBalances(t *testing.T) {
 	tt.Assert.NoError(err)
 	tt.Assert.Len(response, 0)
 
-	// new claimable balances are ingest and one of them updated, they should appear in the next pages
-	cbToBeUpdated := hCBs[3]
-	cbToBeUpdated.Sponsor = null.StringFrom("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML")
-	cbToBeUpdated.LastModifiedLedger = 1238
-	q.UpsertClaimableBalances(tt.Ctx, []history.ClaimableBalance{cbToBeUpdated})
+	// new claimable balances are ingested, they should appear in the next pages
+	balanceInsertbuilder = q.NewClaimableBalanceBatchInsertBuilder()
+	claimantsInsertBuilder = q.NewClaimableBalanceClaimantBatchInsertBuilder()
 
 	entriesMeta = []struct {
 		id        xdr.Hash
@@ -310,29 +321,26 @@ func TestGetClaimableBalances(t *testing.T) {
 		},
 	}
 
-	hCBs = nil
 	for _, e := range entriesMeta {
 		entry := buildClaimableBalance(tt, e.id, e.accountID, e.ledger, e.asset)
 		hCBs = append(hCBs, entry)
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, hCBs)
-	tt.Assert.NoError(err)
+	for _, cBalance := range hCBs[4:] {
+		tt.Assert.NoError(balanceInsertbuilder.Add(cBalance))
 
-	for _, cBalance := range hCBs {
 		for _, claimant := range cBalance.Claimants {
 			claimant := history.ClaimableBalanceClaimant{
 				BalanceID:          cBalance.BalanceID,
 				Destination:        claimant.Destination,
 				LastModifiedLedger: cBalance.LastModifiedLedger,
 			}
-			err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-			tt.Assert.NoError(err)
+			tt.Assert.NoError(claimantsInsertBuilder.Add(claimant))
 		}
 	}
 
-	err = claimantsInsertBuilder.Exec(tt.Ctx)
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(balanceInsertbuilder.Exec(tt.Ctx))
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
 
 	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
 		t,
@@ -358,7 +366,7 @@ func TestGetClaimableBalances(t *testing.T) {
 	tt.Assert.Len(response, 2)
 
 	// response should be the first 2 elements of entries
-	for i, entry := range hCBs {
+	for i, entry := range hCBs[4:] {
 		tt.Assert.Equal(entry.BalanceID, response[i].(protocol.ClaimableBalance).BalanceID)
 	}
 
@@ -372,21 +380,6 @@ func TestGetClaimableBalances(t *testing.T) {
 		q,
 	))
 
-	tt.Assert.NoError(err)
-	tt.Assert.Len(response, 1)
-
-	tt.Assert.Equal(cbToBeUpdated.BalanceID, response[0].(protocol.ClaimableBalance).BalanceID)
-
-	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
-		t,
-		map[string]string{
-			"limit":  "2",
-			"cursor": response[0].(protocol.ClaimableBalance).PagingToken(),
-		},
-		map[string]string{},
-		q,
-	))
-
 	tt.Assert.NoError(err)
 	tt.Assert.Len(response, 0)
 
@@ -404,9 +397,9 @@ func TestGetClaimableBalances(t *testing.T) {
 	tt.Assert.NoError(err)
 	tt.Assert.Len(response, 2)
 
-	tt.Assert.Equal(cbToBeUpdated.BalanceID, response[0].(protocol.ClaimableBalance).BalanceID)
+	tt.Assert.Equal(hCBs[5].BalanceID, response[0].(protocol.ClaimableBalance).BalanceID)
 
-	tt.Assert.Equal(hCBs[1].BalanceID, response[1].(protocol.ClaimableBalance).BalanceID)
+	tt.Assert.Equal(hCBs[4].BalanceID, response[1].(protocol.ClaimableBalance).BalanceID)
 
 	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
 		t,
@@ -492,7 +485,7 @@ func TestGetClaimableBalances(t *testing.T) {
 	))
 
 	tt.Assert.NoError(err)
-	tt.Assert.Len(response, 3)
+	tt.Assert.Len(response, 2)
 	for _, resource := range response {
 		tt.Assert.Equal(
 			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
diff --git a/services/horizon/internal/db2/history/claimable_balances_test.go b/services/horizon/internal/db2/history/claimable_balances_test.go
index 238b03da..ca32975c 100644
--- a/services/horizon/internal/db2/history/claimable_balances_test.go
+++ b/services/horizon/internal/db2/history/claimable_balances_test.go
@@ -1,6 +1,7 @@
 package history
 
 import (
+	"database/sql"
 	"fmt"
 	"testing"
 
@@ -15,6 +16,10 @@ func TestRemoveClaimableBalance(t *testing.T) {
 	defer tt.Finish()
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
 
 	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	asset := xdr.MustNewCreditAsset("USD", accountID)
@@ -39,8 +44,9 @@ func TestRemoveClaimableBalance(t *testing.T) {
 		Amount:             10,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	balanceBatchInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	tt.Assert.NoError(balanceBatchInsertBuilder.Add(cBalance))
+	tt.Assert.NoError(balanceBatchInsertBuilder.Exec(tt.Ctx))
 
 	r, err := q.FindClaimableBalanceByID(tt.Ctx, id)
 	tt.Assert.NoError(err)
@@ -63,6 +69,10 @@ func TestRemoveClaimableBalanceClaimants(t *testing.T) {
 	defer tt.Finish()
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
 
 	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	asset := xdr.MustNewCreditAsset("USD", accountID)
@@ -87,20 +97,9 @@ func TestRemoveClaimableBalanceClaimants(t *testing.T) {
 		Amount:             10,
 	}
 
-	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
-
-	for _, claimant := range cBalance.Claimants {
-		claimant := ClaimableBalanceClaimant{
-			BalanceID:          cBalance.BalanceID,
-			Destination:        claimant.Destination,
-			LastModifiedLedger: cBalance.LastModifiedLedger,
-		}
-		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-		tt.Assert.NoError(err)
-	}
-
-	err = claimantsInsertBuilder.Exec(tt.Ctx)
-	tt.Assert.NoError(err)
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder()
+	tt.Assert.NoError(insertClaimants(claimantsInsertBuilder, cBalance))
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
 
 	removed, err := q.RemoveClaimableBalanceClaimants(tt.Ctx, []string{id})
 	tt.Assert.NoError(err)
@@ -113,6 +112,11 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
+
 	dest1 := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	dest2 := "GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H"
 
@@ -138,19 +142,11 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 		Amount:             10,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	balanceInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder()
 
-	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
-	for _, claimant := range cBalance.Claimants {
-		claimant := ClaimableBalanceClaimant{
-			BalanceID:          cBalance.BalanceID,
-			Destination:        claimant.Destination,
-			LastModifiedLedger: cBalance.LastModifiedLedger,
-		}
-		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-		tt.Assert.NoError(err)
-	}
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance))
+	tt.Assert.NoError(insertClaimants(claimantsInsertBuilder, cBalance))
 
 	balanceID = xdr.ClaimableBalanceId{
 		Type: xdr.ClaimableBalanceIdTypeClaimableBalanceIdTypeV0,
@@ -179,21 +175,11 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 		Amount:             10,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance))
+	tt.Assert.NoError(insertClaimants(claimantsInsertBuilder, cBalance))
 
-	for _, claimant := range cBalance.Claimants {
-		claimant := ClaimableBalanceClaimant{
-			BalanceID:          cBalance.BalanceID,
-			Destination:        claimant.Destination,
-			LastModifiedLedger: cBalance.LastModifiedLedger,
-		}
-		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
-		tt.Assert.NoError(err)
-	}
-
-	err = claimantsInsertBuilder.Exec(tt.Ctx)
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
+	tt.Assert.NoError(balanceInsertBuilder.Exec(tt.Ctx))
 
 	query := ClaimableBalancesQuery{
 		PageQuery: db2.MustPageQuery("", false, "", 10),
@@ -233,20 +219,19 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 	tt.Assert.Len(cbs, 1)
 }
 
-func insertClaimants(q *Q, tt *test.T, cBalance ClaimableBalance) error {
-	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+func insertClaimants(claimantsInsertBuilder ClaimableBalanceClaimantBatchInsertBuilder, cBalance ClaimableBalance) error {
 	for _, claimant := range cBalance.Claimants {
 		claimant := ClaimableBalanceClaimant{
 			BalanceID:          cBalance.BalanceID,
 			Destination:        claimant.Destination,
 			LastModifiedLedger: cBalance.LastModifiedLedger,
 		}
-		err := claimantsInsertBuilder.Add(tt.Ctx, claimant)
+		err := claimantsInsertBuilder.Add(claimant)
 		if err != nil {
 			return err
 		}
 	}
-	return claimantsInsertBuilder.Exec(tt.Ctx)
+	return nil
 }
 
 type claimableBalanceQueryResult struct {
@@ -279,6 +264,11 @@ func TestFindClaimableBalancesByDestinationWithLimit(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
+
 	assetIssuer := "GA25GQLHJU3LPEJXEIAXK23AWEA5GWDUGRSHTQHDFT6HXHVMRULSQJUJ"
 	asset1 := xdr.MustNewCreditAsset("ASSET1", assetIssuer)
 	asset2 := xdr.MustNewCreditAsset("ASSET2", assetIssuer)
@@ -318,8 +308,11 @@ func TestFindClaimableBalancesByDestinationWithLimit(t *testing.T) {
 		LastModifiedLedger: 123,
 		Amount:             10,
 	}
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance1})
-	tt.Assert.NoError(err)
+	balanceInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder()
+
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance1))
+	tt.Assert.NoError(insertClaimants(claimantsInsertBuilder, cBalance1))
 
 	claimants2 := []Claimant{
 		{
@@ -345,14 +338,12 @@ func TestFindClaimableBalancesByDestinationWithLimit(t *testing.T) {
 		LastModifiedLedger: 456,
 		Amount:             10,
 	}
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance2})
-	tt.Assert.NoError(err)
 
-	err = insertClaimants(q, tt, cBalance1)
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance2))
+	tt.Assert.NoError(insertClaimants(claimantsInsertBuilder, cBalance2))
 
-	err = insertClaimants(q, tt, cBalance2)
-	tt.Assert.NoError(err)
+	tt.Assert.NoError(claimantsInsertBuilder.Exec(tt.Ctx))
+	tt.Assert.NoError(balanceInsertBuilder.Exec(tt.Ctx))
 
 	pageQuery := db2.MustPageQuery("", false, "", 1)
 
@@ -414,73 +405,17 @@ func TestFindClaimableBalancesByDestinationWithLimit(t *testing.T) {
 	})
 }
 
-func TestUpdateClaimableBalance(t *testing.T) {
-	tt := test.Start(t)
-	defer tt.Finish()
-	test.ResetHorizonDB(t, tt.HorizonDB)
-	q := &Q{tt.HorizonSession()}
-
-	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
-	lastModifiedLedgerSeq := xdr.Uint32(123)
-	asset := xdr.MustNewCreditAsset("USD", accountID)
-	balanceID := xdr.ClaimableBalanceId{
-		Type: xdr.ClaimableBalanceIdTypeClaimableBalanceIdTypeV0,
-		V0:   &xdr.Hash{1, 2, 3},
-	}
-	id, err := xdr.MarshalHex(balanceID)
-	tt.Assert.NoError(err)
-	cBalance := ClaimableBalance{
-		BalanceID: id,
-		Claimants: []Claimant{
-			{
-				Destination: accountID,
-				Predicate: xdr.ClaimPredicate{
-					Type: xdr.ClaimPredicateTypeClaimPredicateUnconditional,
-				},
-			},
-		},
-		Asset:              asset,
-		LastModifiedLedger: 123,
-		Amount:             10,
-	}
-
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
-
-	// add sponsor
-	cBalance2 := ClaimableBalance{
-		BalanceID: id,
-		Claimants: []Claimant{
-			{
-				Destination: accountID,
-				Predicate: xdr.ClaimPredicate{
-					Type: xdr.ClaimPredicateTypeClaimPredicateUnconditional,
-				},
-			},
-		},
-		Asset:              asset,
-		LastModifiedLedger: 123 + 1,
-		Amount:             10,
-		Sponsor:            null.StringFrom("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-	}
-
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance2})
-	tt.Assert.NoError(err)
-
-	cbs := []ClaimableBalance{}
-	err = q.Select(tt.Ctx, &cbs, selectClaimableBalances)
-	tt.Assert.NoError(err)
-	tt.Assert.Len(cbs, 1)
-	tt.Assert.Equal("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML", cbs[0].Sponsor.String)
-	tt.Assert.Equal(uint32(lastModifiedLedgerSeq+1), cbs[0].LastModifiedLedger)
-}
-
 func TestFindClaimableBalance(t *testing.T) {
 	tt := test.Start(t)
 	defer tt.Finish()
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
+
 	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	asset := xdr.MustNewCreditAsset("USD", accountID)
 	balanceID := xdr.ClaimableBalanceId{
@@ -504,11 +439,11 @@ func TestFindClaimableBalance(t *testing.T) {
 		Amount:             10,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	balanceInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance))
+	tt.Assert.NoError(balanceInsertBuilder.Exec(tt.Ctx))
 
 	cb, err := q.FindClaimableBalanceByID(tt.Ctx, id)
-	tt.Assert.NoError(err)
 
 	tt.Assert.Equal(cBalance.BalanceID, cb.BalanceID)
 	tt.Assert.Equal(cBalance.Asset, cb.Asset)
@@ -525,6 +460,11 @@ func TestGetClaimableBalancesByID(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &Q{tt.HorizonSession()}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+	defer func() {
+		_ = q.Rollback()
+	}()
+
 	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
 	asset := xdr.MustNewCreditAsset("USD", accountID)
 	balanceID := xdr.ClaimableBalanceId{
@@ -548,8 +488,9 @@ func TestGetClaimableBalancesByID(t *testing.T) {
 		Amount:             10,
 	}
 
-	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
-	tt.Assert.NoError(err)
+	balanceInsertBuilder := q.NewClaimableBalanceBatchInsertBuilder()
+	tt.Assert.NoError(balanceInsertBuilder.Add(cBalance))
+	tt.Assert.NoError(balanceInsertBuilder.Exec(tt.Ctx))
 
 	r, err := q.GetClaimableBalancesByID(tt.Ctx, []string{id})
 	tt.Assert.NoError(err)
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index 4af1a5be..63b8caa1 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -24,6 +24,8 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 	ctx := context.Background()
 	maxBatchSize := 100000
 
+	mockSession := &db.MockSession{}
+
 	q := &mockDBQ{}
 
 	q.MockQAccounts.On("UpsertAccounts", ctx, []history.AccountEntry{
@@ -49,8 +51,16 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
 
-	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder).Twice()
+
+	mockClaimantsBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimantsBatchInsertBuilder).Twice()
+
+	mockClaimantsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	mockClaimableBalanceBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
@@ -62,6 +72,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 		},
 		historyQ: q,
 		filters:  &MockFilters{},
+		session:  mockSession,
 	}
 
 	_, err := runner.RunGenesisStateIngestion()
@@ -76,6 +87,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 		NetworkPassphrase: network.PublicNetworkPassphrase,
 	}
 
+	mockSession := &db.MockSession{}
 	q := &mockDBQ{}
 	defer mock.AssertExpectationsForObjects(t, q)
 	historyAdapter := &mockHistoryArchiveAdapter{}
@@ -116,8 +128,17 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 	mockAccountSignersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
-	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder).Twice()
+
+	mockClaimantsBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimantsBatchInsertBuilder).Twice()
+
+	mockClaimantsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	mockClaimableBalanceBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
@@ -128,6 +149,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 		historyQ:       q,
 		historyAdapter: historyAdapter,
 		filters:        &MockFilters{},
+		session:        mockSession,
 	}
 
 	_, err := runner.RunHistoryArchiveIngestion(63, false, MaxSupportedProtocolVersion, bucketListHash)
@@ -142,6 +164,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 		NetworkPassphrase: network.PublicNetworkPassphrase,
 	}
 
+	mockSession := &db.MockSession{}
 	q := &mockDBQ{}
 	defer mock.AssertExpectationsForObjects(t, q)
 	historyAdapter := &mockHistoryArchiveAdapter{}
@@ -153,8 +176,17 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
-	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder).Once()
+
+	mockClaimantsBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimantsBatchInsertBuilder).Once()
+
+	mockClaimantsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	mockClaimableBalanceBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
@@ -165,6 +197,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 		historyQ:       q,
 		historyAdapter: historyAdapter,
 		filters:        &MockFilters{},
+		session:        mockSession,
 	}
 
 	_, err := runner.RunHistoryArchiveIngestion(100, false, 200, xdr.Hash{})
@@ -186,8 +219,15 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	// Twice = checking ledgerSource and historyArchiveSource
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(&history.MockAccountSignersBatchInsertBuilder{}).Twice()
-	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Twice()
+
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder).Twice()
+
+	mockClaimantsBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimantsBatchInsertBuilder).Twice()
+
 	runner := ProcessorRunner{
 		ctx:      ctx,
 		historyQ: q,
@@ -313,6 +353,17 @@ func TestProcessorRunnerWithFilterEnabled(t *testing.T) {
 	q.MockQTransactions.On("NewTransactionFilteredTmpBatchInsertBuilder").
 		Return(mockTransactionsFilteredTmpBatchInsertBuilder)
 
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder).Once()
+
+	mockClaimantsBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimantsBatchInsertBuilder).Once()
+
+	mockClaimantsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	mockClaimableBalanceBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+
 	q.On("DeleteTransactionsFilteredTmpOlderThan", ctx, mock.AnythingOfType("uint64")).
 		Return(int64(0), nil)
 
@@ -493,8 +544,15 @@ func mockBatchBuilders(q *mockDBQ, mockSession *db.MockSession, ctx context.Cont
 	q.MockQHistoryLiquidityPools.On("NewOperationLiquidityPoolBatchInsertBuilder").
 		Return(mockOperationLiquidityPoolBatchInsertBuilder)
 
-	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+	mockClaimableBalanceClaimantBatchInsertBuilder := &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	mockClaimableBalanceClaimantBatchInsertBuilder.On("Exec", ctx).Return(nil)
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(mockClaimableBalanceClaimantBatchInsertBuilder)
+
+	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
+	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
+		Return(mockClaimableBalanceBatchInsertBuilder)
+	mockClaimableBalanceBatchInsertBuilder.On("Exec", ctx).Return(nil)
 
 	q.On("NewTradeBatchInsertBuilder").Return(&history.MockTradeBatchInsertBuilder{})
 
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go b/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
index 70cd43bb..524de095 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
@@ -7,11 +7,11 @@ import (
 	"testing"
 
 	"github.com/guregu/null"
-	"github.com/stretchr/testify/suite"
 
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
 	"github.com/stellar/go/xdr"
+	"github.com/stretchr/testify/suite"
 )
 
 func TestClaimableBalancesChangeProcessorTestSuiteState(t *testing.T) {
@@ -20,19 +20,28 @@ func TestClaimableBalancesChangeProcessorTestSuiteState(t *testing.T) {
 
 type ClaimableBalancesChangeProcessorTestSuiteState struct {
 	suite.Suite
-	ctx                    context.Context
-	processor              *ClaimableBalancesChangeProcessor
-	mockQ                  *history.MockQClaimableBalances
-	mockBatchInsertBuilder *history.MockClaimableBalanceClaimantBatchInsertBuilder
+	ctx                                    context.Context
+	processor                              *ClaimableBalancesChangeProcessor
+	mockQ                                  *history.MockQClaimableBalances
+	mockClaimantsBatchInsertBuilder        *history.MockClaimableBalanceClaimantBatchInsertBuilder
+	mockClaimableBalanceBatchInsertBuilder *history.MockClaimableBalanceBatchInsertBuilder
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteState) SetupTest() {
 	s.ctx = context.Background()
-	s.mockBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	s.mockClaimantsBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	s.mockClaimableBalanceBatchInsertBuilder = &history.MockClaimableBalanceBatchInsertBuilder{}
+
 	s.mockQ = &history.MockQClaimableBalances{}
 	s.mockQ.
-		On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(s.mockBatchInsertBuilder).Once()
+		On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(s.mockClaimantsBatchInsertBuilder)
+	s.mockQ.
+		On("NewClaimableBalanceBatchInsertBuilder").
+		Return(s.mockClaimableBalanceBatchInsertBuilder)
+
+	s.mockClaimantsBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
+	s.mockClaimableBalanceBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
 
 	s.processor = NewClaimableBalancesChangeProcessor(s.mockQ)
 }
@@ -67,27 +76,27 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteState) TestCreatesClaimableBal
 	}
 	id, err := xdr.MarshalHex(balanceID)
 	s.Assert().NoError(err)
-	s.mockQ.On("UpsertClaimableBalances", s.ctx, []history.ClaimableBalance{
-		{
-			BalanceID: id,
-			Claimants: []history.Claimant{
-				{
-					Destination: "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-				},
+	s.mockClaimableBalanceBatchInsertBuilder.On("Add", history.ClaimableBalance{
+		BalanceID: id,
+		Claimants: []history.Claimant{
+			{
+				Destination: "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 			},
-			Asset:              cBalance.Asset,
-			Amount:             cBalance.Amount,
-			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 		},
+		Asset:              cBalance.Asset,
+		Amount:             cBalance.Amount,
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 	}).Return(nil).Once()
 
-	s.mockBatchInsertBuilder.On("Add", s.ctx, history.ClaimableBalanceClaimant{
+	s.mockClaimableBalanceBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
+	s.mockClaimantsBatchInsertBuilder.On("Add", history.ClaimableBalanceClaimant{
 		BalanceID:          id,
 		Destination:        "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 	}).Return(nil).Once()
 
-	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+	s.mockClaimantsBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
 
 	err = s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeClaimableBalance,
@@ -109,25 +118,34 @@ func TestClaimableBalancesChangeProcessorTestSuiteLedger(t *testing.T) {
 
 type ClaimableBalancesChangeProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx                    context.Context
-	processor              *ClaimableBalancesChangeProcessor
-	mockQ                  *history.MockQClaimableBalances
-	mockBatchInsertBuilder *history.MockClaimableBalanceClaimantBatchInsertBuilder
+	ctx                                    context.Context
+	processor                              *ClaimableBalancesChangeProcessor
+	mockQ                                  *history.MockQClaimableBalances
+	mockClaimantsBatchInsertBuilder        *history.MockClaimableBalanceClaimantBatchInsertBuilder
+	mockClaimableBalanceBatchInsertBuilder *history.MockClaimableBalanceBatchInsertBuilder
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
-	s.mockBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	s.mockClaimantsBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
+	s.mockClaimableBalanceBatchInsertBuilder = &history.MockClaimableBalanceBatchInsertBuilder{}
 	s.mockQ = &history.MockQClaimableBalances{}
 	s.mockQ.
-		On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
-		Return(s.mockBatchInsertBuilder).Once()
+		On("NewClaimableBalanceClaimantBatchInsertBuilder").
+		Return(s.mockClaimantsBatchInsertBuilder)
+	s.mockQ.
+		On("NewClaimableBalanceBatchInsertBuilder").
+		Return(s.mockClaimableBalanceBatchInsertBuilder)
+
+	s.mockClaimantsBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
+	s.mockClaimableBalanceBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
 
 	s.processor = NewClaimableBalancesChangeProcessor(s.mockQ)
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TearDownTest() {
 	s.Assert().NoError(s.processor.Commit(s.ctx))
+	s.processor.reset()
 	s.mockQ.AssertExpectations(s.T())
 }
 
@@ -148,29 +166,6 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestNewClaimableBalanc
 		Amount:    10,
 	}
 	entry := xdr.LedgerEntry{
-		Data: xdr.LedgerEntryData{
-			Type:             xdr.LedgerEntryTypeClaimableBalance,
-			ClaimableBalance: &cBalance,
-		},
-		LastModifiedLedgerSeq: lastModifiedLedgerSeq,
-		Ext: xdr.LedgerEntryExt{
-			V: 1,
-			V1: &xdr.LedgerEntryExtensionV1{
-				SponsoringId: nil,
-			},
-		},
-	}
-	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
-
-	err := s.processor.ProcessChange(s.ctx, ingest.Change{
-		Type: xdr.LedgerEntryTypeClaimableBalance,
-		Pre:  nil,
-		Post: &entry,
-	})
-	s.Assert().NoError(err)
-
-	// add sponsor
-	updated := xdr.LedgerEntry{
 		Data: xdr.LedgerEntryData{
 			Type:             xdr.LedgerEntryTypeClaimableBalance,
 			ClaimableBalance: &cBalance,
@@ -184,99 +179,28 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestNewClaimableBalanc
 		},
 	}
 
-	entry.LastModifiedLedgerSeq = entry.LastModifiedLedgerSeq - 1
-	err = s.processor.ProcessChange(s.ctx, ingest.Change{
-		Type: xdr.LedgerEntryTypeClaimableBalance,
-		Pre:  &entry,
-		Post: &updated,
-	})
-	s.Assert().NoError(err)
-
 	id, err := xdr.MarshalHex(balanceID)
 	s.Assert().NoError(err)
+
 	// We use LedgerEntryChangesCache so all changes are squashed
-	s.mockQ.On(
-		"UpsertClaimableBalances",
-		s.ctx,
-		[]history.ClaimableBalance{
-			{
-				BalanceID:          id,
-				Claimants:          []history.Claimant{},
-				Asset:              cBalance.Asset,
-				Amount:             cBalance.Amount,
-				LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				Sponsor:            null.StringFrom("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-			},
+	s.mockClaimableBalanceBatchInsertBuilder.On(
+		"Add",
+		history.ClaimableBalance{
+			BalanceID:          id,
+			Claimants:          []history.Claimant{},
+			Asset:              cBalance.Asset,
+			Amount:             cBalance.Amount,
+			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+			Sponsor:            null.StringFrom("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
 		},
 	).Return(nil).Once()
-}
 
-func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestUpdateClaimableBalance() {
-	balanceID := xdr.ClaimableBalanceId{
-		Type: xdr.ClaimableBalanceIdTypeClaimableBalanceIdTypeV0,
-		V0:   &xdr.Hash{1, 2, 3},
-	}
-	cBalance := xdr.ClaimableBalanceEntry{
-		BalanceId: balanceID,
-		Claimants: []xdr.Claimant{},
-		Asset:     xdr.MustNewCreditAsset("USD", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-		Amount:    10,
-	}
-	lastModifiedLedgerSeq := xdr.Uint32(123)
-
-	pre := xdr.LedgerEntry{
-		Data: xdr.LedgerEntryData{
-			Type:             xdr.LedgerEntryTypeClaimableBalance,
-			ClaimableBalance: &cBalance,
-		},
-		LastModifiedLedgerSeq: lastModifiedLedgerSeq - 1,
-		Ext: xdr.LedgerEntryExt{
-			V: 1,
-			V1: &xdr.LedgerEntryExtensionV1{
-				SponsoringId: nil,
-			},
-		},
-	}
-
-	// add sponsor
-	updated := xdr.LedgerEntry{
-		Data: xdr.LedgerEntryData{
-			Type:             xdr.LedgerEntryTypeClaimableBalance,
-			ClaimableBalance: &cBalance,
-		},
-		LastModifiedLedgerSeq: lastModifiedLedgerSeq,
-		Ext: xdr.LedgerEntryExt{
-			V: 1,
-			V1: &xdr.LedgerEntryExtensionV1{
-				SponsoringId: xdr.MustAddressPtr("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-			},
-		},
-	}
-	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
-
-	err := s.processor.ProcessChange(s.ctx, ingest.Change{
+	err = s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeClaimableBalance,
-		Pre:  &pre,
-		Post: &updated,
+		Pre:  nil,
+		Post: &entry,
 	})
 	s.Assert().NoError(err)
-
-	id, err := xdr.MarshalHex(balanceID)
-	s.Assert().NoError(err)
-	s.mockQ.On(
-		"UpsertClaimableBalances",
-		s.ctx,
-		[]history.ClaimableBalance{
-			{
-				BalanceID:          id,
-				Claimants:          []history.Claimant{},
-				Asset:              cBalance.Asset,
-				Amount:             cBalance.Amount,
-				LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				Sponsor:            null.StringFrom("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-			},
-		},
-	).Return(nil).Once()
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestRemoveClaimableBalance() {
diff --git a/services/horizon/internal/ingest/verify_test.go b/services/horizon/internal/ingest/verify_test.go
index 47d29cb6..1fbb6b8f 100644
--- a/services/horizon/internal/ingest/verify_test.go
+++ b/services/horizon/internal/ingest/verify_test.go
@@ -271,6 +271,8 @@ func TestStateVerifierLockBusy(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &history.Q{&db.Session{DB: tt.HorizonDB}}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+
 	checkpointLedger := uint32(63)
 	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "")
 
@@ -291,6 +293,8 @@ func TestStateVerifierLockBusy(t *testing.T) {
 	}
 	tt.Assert.NoError(changeProcessor.Commit(tt.Ctx))
 
+	tt.Assert.NoError(q.Commit())
+
 	q.UpdateLastLedgerIngest(tt.Ctx, checkpointLedger)
 
 	mockHistoryAdapter := &mockHistoryArchiveAdapter{}
@@ -324,6 +328,8 @@ func TestStateVerifier(t *testing.T) {
 	test.ResetHorizonDB(t, tt.HorizonDB)
 	q := &history.Q{&db.Session{DB: tt.HorizonDB}}
 
+	tt.Assert.NoError(q.BeginTx(tt.Ctx, &sql.TxOptions{}))
+
 	checkpointLedger := uint32(63)
 	changeProcessor := buildChangeProcessor(q, &ingest.StatsChangeProcessor{}, ledgerSource, checkpointLedger, "")
 	mockChangeReader := &ingest.MockChangeReader{}
@@ -354,6 +360,8 @@ func TestStateVerifier(t *testing.T) {
 	tt.Assert.NoError(changeProcessor.Commit(tt.Ctx))
 	tt.Assert.Equal(len(xdr.LedgerEntryTypeMap), len(coverage))
 
+	tt.Assert.NoError(q.Commit())
+
 	q.UpdateLastLedgerIngest(tt.Ctx, checkpointLedger)
 
 	mockChangeReader.On("Read").Return(ingest.Change{}, io.EOF).Twice()