Repository: stellar/go
Pull request #5105 (2023-11-08 22:07:39)

Update SignersProcessor to use FastBatchInsertBuilder
<!-- 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

Optimize the `SignersProcessor` to use `FastBatchInsertBuilder` which uses `COPY` to bulk insert into the database table `account_signers`.

* Using the `FastBatchInsertBuilder` when inserting into the `account_signers` table during building state (i.e. `useLedgerEntryCache` is false).
* Introducing another optimization in the insertion process during live ingestion (i.e., `useLedgerEntryCache` is true) to also use bulk insert via `FastBatchInsertBuilder`.
* Refactor the code to reduce code duplication and improve overall readability.


### Why

see #5098

### Known limitations

Previously in SignersProcessor, during live ingestion, each row was inserted individually and a `ingest.StateError` was thrown in case of failure. With the bulk write, this individual check is no longer possible and so the system now throws an error for the entire batch if a failure occurs.



Patch:
diff --git a/services/horizon/internal/db2/history/account_signers.go b/services/horizon/internal/db2/history/account_signers.go
index 32c6cdc5..d523c96f 100644
--- a/services/horizon/internal/db2/history/account_signers.go
+++ b/services/horizon/internal/db2/history/account_signers.go
@@ -53,6 +53,9 @@ func (q *Q) AccountsForSigner(ctx context.Context, signer string, page db2.PageQ
 // CreateAccountSigner creates a row in the accounts_signers table.
 // Returns number of rows affected and error.
 func (q *Q) CreateAccountSigner(ctx context.Context, account, signer string, weight int32, sponsor *string) (int64, error) {
+	// This function is not used in the ingestion code path. SignersProcessor now uses bulk insertion with
+	// FastBatchInsertBuilder, but this function is used in unit tests when only a single row needs to be inserted.
+
 	sql := sq.Insert("accounts_signers").
 		Columns("account_id", "signer", "weight", "sponsor").
 		Values(account, signer, weight, sponsor)
diff --git a/services/horizon/internal/db2/history/account_signers_batch_insert_builder.go b/services/horizon/internal/db2/history/account_signers_batch_insert_builder.go
index fd1cab29..d347fb3e 100644
--- a/services/horizon/internal/db2/history/account_signers_batch_insert_builder.go
+++ b/services/horizon/internal/db2/history/account_signers_batch_insert_builder.go
@@ -4,8 +4,8 @@ import (
 	"context"
 )
 
-func (i *accountSignersBatchInsertBuilder) Add(ctx context.Context, signer AccountSigner) error {
-	return i.builder.Row(ctx, map[string]interface{}{
+func (i *accountSignersBatchInsertBuilder) Add(signer AccountSigner) error {
+	return i.builder.Row(map[string]interface{}{
 		"account_id": signer.Account,
 		"signer":     signer.Signer,
 		"weight":     signer.Weight,
@@ -14,5 +14,5 @@ func (i *accountSignersBatchInsertBuilder) Add(ctx context.Context, signer Accou
 }
 
 func (i *accountSignersBatchInsertBuilder) 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/main.go b/services/horizon/internal/db2/history/main.go
index 4210fe10..850b8bef 100644
--- a/services/horizon/internal/db2/history/main.go
+++ b/services/horizon/internal/db2/history/main.go
@@ -323,13 +323,15 @@ type AccountSigner struct {
 }
 
 type AccountSignersBatchInsertBuilder interface {
-	Add(ctx context.Context, signer AccountSigner) error
+	Add(signer AccountSigner) error
 	Exec(ctx context.Context) error
 }
 
 // accountSignersBatchInsertBuilder is a simple wrapper around db.BatchInsertBuilder
 type accountSignersBatchInsertBuilder struct {
-	builder db.BatchInsertBuilder
+	builder db.FastBatchInsertBuilder
+	session db.SessionInterface
+	table   string
 }
 
 // Data is a row of data from the `account_data` table
@@ -742,7 +744,7 @@ type QSigners interface {
 	GetLastLedgerIngest(ctx context.Context) (uint32, error)
 	UpdateLastLedgerIngest(ctx context.Context, ledgerSequence uint32) error
 	AccountsForSigner(ctx context.Context, signer string, page db2.PageQuery) ([]AccountSigner, error)
-	NewAccountSignersBatchInsertBuilder(maxBatchSize int) AccountSignersBatchInsertBuilder
+	NewAccountSignersBatchInsertBuilder() AccountSignersBatchInsertBuilder
 	CreateAccountSigner(ctx context.Context, account, signer string, weight int32, sponsor *string) (int64, error)
 	RemoveAccountSigner(ctx context.Context, account, signer string) (int64, error)
 	SignersForAccounts(ctx context.Context, accounts []string) ([]AccountSigner, error)
@@ -846,12 +848,11 @@ type QTrustLines interface {
 	RemoveTrustLines(ctx context.Context, ledgerKeys []string) (int64, error)
 }
 
-func (q *Q) NewAccountSignersBatchInsertBuilder(maxBatchSize int) AccountSignersBatchInsertBuilder {
+func (q *Q) NewAccountSignersBatchInsertBuilder() AccountSignersBatchInsertBuilder {
 	return &accountSignersBatchInsertBuilder{
-		builder: db.BatchInsertBuilder{
-			Table:        q.GetTable("accounts_signers"),
-			MaxBatchSize: maxBatchSize,
-		},
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "accounts_signers",
 	}
 }
 
diff --git a/services/horizon/internal/db2/history/mock_account_signers_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_account_signers_batch_insert_builder.go
index 3f786f65..9ce4fd7d 100644
--- a/services/horizon/internal/db2/history/mock_account_signers_batch_insert_builder.go
+++ b/services/horizon/internal/db2/history/mock_account_signers_batch_insert_builder.go
@@ -9,8 +9,8 @@ type MockAccountSignersBatchInsertBuilder struct {
 	mock.Mock
 }
 
-func (m *MockAccountSignersBatchInsertBuilder) Add(ctx context.Context, signer AccountSigner) error {
-	a := m.Called(ctx, signer)
+func (m *MockAccountSignersBatchInsertBuilder) Add(signer AccountSigner) error {
+	a := m.Called(signer)
 	return a.Error(0)
 }
 
diff --git a/services/horizon/internal/db2/history/mock_q_signers.go b/services/horizon/internal/db2/history/mock_q_signers.go
index 81f56d9f..1b2a4864 100644
--- a/services/horizon/internal/db2/history/mock_q_signers.go
+++ b/services/horizon/internal/db2/history/mock_q_signers.go
@@ -32,8 +32,8 @@ func (m *MockQSigners) AccountsForSigner(ctx context.Context, signer string, pag
 	return a.Get(0).([]AccountSigner), a.Error(1)
 }
 
-func (m *MockQSigners) NewAccountSignersBatchInsertBuilder(maxBatchSize int) AccountSignersBatchInsertBuilder {
-	a := m.Called(maxBatchSize)
+func (m *MockQSigners) NewAccountSignersBatchInsertBuilder() AccountSignersBatchInsertBuilder {
+	a := m.Called()
 	return a.Get(0).(AccountSignersBatchInsertBuilder)
 }
 
diff --git a/services/horizon/internal/ingest/processors/signers_processor.go b/services/horizon/internal/ingest/processors/signers_processor.go
index d165110b..b7286752 100644
--- a/services/horizon/internal/ingest/processors/signers_processor.go
+++ b/services/horizon/internal/ingest/processors/signers_processor.go
@@ -14,8 +14,8 @@ import (
 type SignersProcessor struct {
 	signersQ history.QSigners
 
-	cache *ingest.ChangeCompactor
-	batch history.AccountSignersBatchInsertBuilder
+	cache              *ingest.ChangeCompactor
+	batchInsertBuilder history.AccountSignersBatchInsertBuilder
 	// insertOnlyMode is a mode in which we don't use ledger cache and we just
 	// add signers to a batch, then we Exec all signers in one insert query.
 	// This is done to make history buckets processing faster (batch inserting).
@@ -31,7 +31,7 @@ func NewSignersProcessor(
 }
 
 func (p *SignersProcessor) reset() {
-	p.batch = p.signersQ.NewAccountSignersBatchInsertBuilder(maxBatchSize)
+	p.batchInsertBuilder = p.signersQ.NewAccountSignersBatchInsertBuilder()
 	p.cache = ingest.NewChangeCompactor()
 }
 
@@ -51,107 +51,95 @@ func (p *SignersProcessor) ProcessChange(ctx context.Context, change ingest.Chan
 			if err != nil {
 				return errors.Wrap(err, "error in Commit")
 			}
-			p.reset()
 		}
 
 		return nil
 	}
 
-	if !(change.Pre == nil && change.Post != nil) {
-		return errors.New("AssetStatsProSignersProcessorcessor is in insert only mode")
-	}
-
-	accountEntry := change.Post.Data.MustAccount()
-	account := accountEntry.AccountId.Address()
-
-	sponsors := accountEntry.SponsorPerSigner()
-	for signer, weight := range accountEntry.SignerSummary() {
-		var sponsor null.String
-		if sponsorDesc, isSponsored := sponsors[signer]; isSponsored {
-			sponsor = null.StringFrom(sponsorDesc.Address())
-		}
-
-		err := p.batch.Add(ctx, history.AccountSigner{
-			Account: account,
-			Signer:  signer,
-			Weight:  weight,
-			Sponsor: sponsor,
-		})
-		if err != nil {
-			return errors.Wrap(err, "Error adding row to accountSignerBatch")
+	if change.Pre == nil && change.Post != nil {
+		postAccountEntry := change.Post.Data.MustAccount()
+		if err := p.addAccountSigners(postAccountEntry); err != nil {
+			return err
 		}
+	} else {
+		return errors.New("SignersProcessor is in insert only mode")
 	}
 
 	return nil
 }
 
 func (p *SignersProcessor) Commit(ctx context.Context) error {
-	if !p.useLedgerEntryCache {
-		return p.batch.Exec(ctx)
-	}
+	defer p.reset()
 
-	changes := p.cache.GetChanges()
-	for _, change := range changes {
-		if !change.AccountSignersChanged() {
-			continue
-		}
+	if p.useLedgerEntryCache {
+		changes := p.cache.GetChanges()
+		for _, change := range changes {
+			if !change.AccountSignersChanged() {
+				continue
+			}
 
-		// The code below removes all Pre signers adds Post signers but
-		// can be improved by finding a diff (check performance first).
-		if change.Pre != nil {
-			preAccountEntry := change.Pre.Data.MustAccount()
-			for signer := range preAccountEntry.SignerSummary() {
-				rowsAffected, err := p.signersQ.RemoveAccountSigner(ctx, preAccountEntry.AccountId.Address(), signer)
-				if err != nil {
-					return errors.Wrap(err, "Error removing a signer")
+			// The code below removes all Pre signers adds Post signers but
+			// can be improved by finding a diff (check performance first).
+			if change.Pre != nil {
+				if err := p.removeAccountSigners(ctx, change.Pre.Data.MustAccount()); err != nil {
+					return err
 				}
+			}
 
-				if rowsAffected != 1 {
-					return ingest.NewStateError(errors.Errorf(
-						"Expected account=%s signer=%s in database but not found when removing (rows affected = %d)",
-						preAccountEntry.AccountId.Address(),
-						signer,
-						rowsAffected,
-					))
+			if change.Post != nil {
+				if err := p.addAccountSigners(change.Post.Data.MustAccount()); err != nil {
+					return err
 				}
 			}
 		}
+	}
 
-		if change.Post != nil {
-			postAccountEntry := change.Post.Data.MustAccount()
-			sponsorsPerSigner := postAccountEntry.SponsorPerSigner()
-			for signer, weight := range postAccountEntry.SignerSummary() {
-
-				// Ignore master key
-				var sponsor *string
-				if signer != postAccountEntry.AccountId.Address() {
-					if s, ok := sponsorsPerSigner[signer]; ok {
-						a := s.Address()
-						sponsor = &a
-					}
-				}
+	err := p.batchInsertBuilder.Exec(ctx)
+	if err != nil {
+		return errors.Wrap(err, "error executing AccountSignersBatchInsertBuilder")
+	}
 
-				rowsAffected, err := p.signersQ.CreateAccountSigner(ctx,
-					postAccountEntry.AccountId.Address(),
-					signer,
-					weight,
-					sponsor,
-				)
-				if err != nil {
-					return errors.Wrapf(err, "Error inserting a signer (%s)", signer)
-				}
+	return nil
+}
 
-				if rowsAffected != 1 {
-					return ingest.NewStateError(errors.Errorf(
-						"%d rows affected when inserting account=%s signer=%s to database",
-						rowsAffected,
-						postAccountEntry.AccountId.Address(),
-						signer,
-					))
-				}
-			}
+func (p *SignersProcessor) removeAccountSigners(ctx context.Context, accountEntry xdr.AccountEntry) error {
+	for signer := range accountEntry.SignerSummary() {
+		rowsAffected, err := p.signersQ.RemoveAccountSigner(ctx, accountEntry.AccountId.Address(), signer)
+		if err != nil {
+			return errors.Wrap(err, "Error removing a signer")
+		}
+
+		if rowsAffected != 1 {
+			return ingest.NewStateError(errors.Errorf(
+				"Expected account=%s signer=%s in database but not found when removing (rows affected = %d)",
+				accountEntry.AccountId.Address(),
+				signer,
+				rowsAffected,
+			))
 		}
 	}
+	return nil
+}
 
+func (p *SignersProcessor) addAccountSigners(accountEntry xdr.AccountEntry) error {
+	sponsorsPerSigner := accountEntry.SponsorPerSigner()
+	for signer, weight := range accountEntry.SignerSummary() {
+		// Ignore master key
+		var sponsor null.String
+		if signer != accountEntry.AccountId.Address() {
+			if sponsorDesc, isSponsored := sponsorsPerSigner[signer]; isSponsored {
+				sponsor = null.StringFrom(sponsorDesc.Address())
+			}
+		}
+
+		if err := p.batchInsertBuilder.Add(history.AccountSigner{
+			Account: accountEntry.AccountId.Address(),
+			Signer:  signer,
+			Weight:  weight,
+			Sponsor: sponsor,
+		}); err != nil {
+			return errors.Wrapf(err, "Error adding signer (%s) to AccountSignersBatchInsertBuilder", signer)
+		}
+	}
 	return nil
 }


Test patch:
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index 63b8caa1..f1ac3145 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -22,7 +22,6 @@ import (
 
 func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 	ctx := context.Background()
-	maxBatchSize := 100000
 
 	mockSession := &db.MockSession{}
 
@@ -41,15 +40,15 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 
 	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
 	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
-	mockAccountSignersBatchInsertBuilder.On("Add", ctx, history.AccountSigner{
+	mockAccountSignersBatchInsertBuilder.On("Add", history.AccountSigner{
 		Account: "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN7",
 		Signer:  "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN7",
 		Weight:  1,
 		Sponsor: null.String{},
 	}).Return(nil).Once()
 	mockAccountSignersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(mockAccountSignersBatchInsertBuilder).Once()
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
+		Return(mockAccountSignersBatchInsertBuilder).Twice()
 
 	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
 	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
@@ -81,7 +80,6 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 
 func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 	ctx := context.Background()
-	maxBatchSize := 100000
 
 	config := Config{
 		NetworkPassphrase: network.PublicNetworkPassphrase,
@@ -120,14 +118,14 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 
 	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
 	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
-	mockAccountSignersBatchInsertBuilder.On("Add", ctx, history.AccountSigner{
+	mockAccountSignersBatchInsertBuilder.On("Add", history.AccountSigner{
 		Account: "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN7",
 		Signer:  "GAAZI4TCR3TY5OJHCTJC2A4QSY6CJWJH5IAJTGKIN2ER7LBNVKOCCWN7",
 		Weight:  1,
 	}).Return(nil).Once()
 	mockAccountSignersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(mockAccountSignersBatchInsertBuilder).Once()
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
+		Return(mockAccountSignersBatchInsertBuilder).Twice()
 
 	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
 	q.MockQClaimableBalances.On("NewClaimableBalanceBatchInsertBuilder").
@@ -158,7 +156,6 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 
 func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t *testing.T) {
 	ctx := context.Background()
-	maxBatchSize := 100000
 
 	config := Config{
 		NetworkPassphrase: network.PublicNetworkPassphrase,
@@ -174,7 +171,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 
 	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
 	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
 		Return(mockAccountSignersBatchInsertBuilder).Once()
 
 	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
@@ -211,13 +208,12 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 
 func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	ctx := context.Background()
-	maxBatchSize := 100000
 
 	q := &mockDBQ{}
 	defer mock.AssertExpectationsForObjects(t, q)
 
 	// Twice = checking ledgerSource and historyArchiveSource
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
 		Return(&history.MockAccountSignersBatchInsertBuilder{}).Twice()
 
 	mockClaimableBalanceBatchInsertBuilder := &history.MockClaimableBalanceBatchInsertBuilder{}
@@ -467,8 +463,8 @@ func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupported(t *t
 		Return(mockTransactionsBatchInsertBuilder).Twice()
 
 	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(mockAccountSignersBatchInsertBuilder).Once()
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
+		Return(mockAccountSignersBatchInsertBuilder).Twice()
 
 	mockOperationsBatchInsertBuilder := &history.MockOperationsBatchInsertBuilder{}
 	q.MockQOperations.On("NewOperationBatchInsertBuilder").
@@ -501,8 +497,9 @@ func mockBatchBuilders(q *mockDBQ, mockSession *db.MockSession, ctx context.Cont
 		Return(mockTransactionsBatchInsertBuilder)
 
 	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
-	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(mockAccountSignersBatchInsertBuilder).Once()
+	mockAccountSignersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder").
+		Return(mockAccountSignersBatchInsertBuilder).Twice()
 
 	mockOperationsBatchInsertBuilder := &history.MockOperationsBatchInsertBuilder{}
 	mockOperationsBatchInsertBuilder.On("Exec", ctx, mockSession).Return(nil).Once()
diff --git a/services/horizon/internal/ingest/processors/signer_processor_test.go b/services/horizon/internal/ingest/processors/signer_processor_test.go
index de5c70bd..5b4d147f 100644
--- a/services/horizon/internal/ingest/processors/signer_processor_test.go
+++ b/services/horizon/internal/ingest/processors/signer_processor_test.go
@@ -33,14 +33,14 @@ func (s *AccountsSignerProcessorTestSuiteState) SetupTest() {
 	s.mockBatchInsertBuilder = &history.MockAccountSignersBatchInsertBuilder{}
 
 	s.mockQ.
-		On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(s.mockBatchInsertBuilder).Once()
+		On("NewAccountSignersBatchInsertBuilder").
+		Return(s.mockBatchInsertBuilder).Twice()
+	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
 
 	s.processor = NewSignersProcessor(s.mockQ, false)
 }
 
 func (s *AccountsSignerProcessorTestSuiteState) TearDownTest() {
-	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
 	s.Assert().NoError(s.processor.Commit(s.ctx))
 
 	s.mockQ.AssertExpectations(s.T())
@@ -53,7 +53,7 @@ func (s *AccountsSignerProcessorTestSuiteState) TestNoEntries() {
 
 func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSigners() {
 	s.mockBatchInsertBuilder.
-		On("Add", s.ctx, history.AccountSigner{
+		On("Add", history.AccountSigner{
 			Account: "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 			Signer:  "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 			Weight:  int32(1),
@@ -75,7 +75,7 @@ func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSigners() {
 	s.Assert().NoError(err)
 
 	s.mockBatchInsertBuilder.
-		On("Add", s.ctx, history.AccountSigner{
+		On("Add", history.AccountSigner{
 			Account: "GCCCU34WDY2RATQTOOQKY6SZWU6J5DONY42SWGW2CIXGW4LICAGNRZKX",
 			Signer:  "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 			Weight:  int32(10),
@@ -105,7 +105,7 @@ func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSigners() {
 
 func (s *AccountsSignerProcessorTestSuiteState) TestCreatesSignerWithSponsor() {
 	s.mockBatchInsertBuilder.
-		On("Add", s.ctx, history.AccountSigner{
+		On("Add", history.AccountSigner{
 			Account: "GCCCU34WDY2RATQTOOQKY6SZWU6J5DONY42SWGW2CIXGW4LICAGNRZKX",
 			Signer:  "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
 			Weight:  int32(10),
@@ -156,17 +156,20 @@ func TestAccountsSignerProcessorTestSuiteLedger(t *testing.T) {
 
 type AccountsSignerProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx       context.Context
-	processor *SignersProcessor
-	mockQ     *history.MockQSigners
+	ctx                                  context.Context
+	processor                            *SignersProcessor
+	mockQ                                *history.MockQSigners
+	mockAccountSignersBatchInsertBuilder *history.MockAccountSignersBatchInsertBuilder
 }
 
 func (s *AccountsSignerProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQSigners{}
+	s.mockAccountSignersBatchInsertBuilder = &history.MockAccountSignersBatchInsertBuilder{}
 	s.mockQ.
-		On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
-		Return(&history.MockAccountSignersBatchInsertBuilder{}).Once()
+		On("NewAccountSignersBatchInsertBuilder").
+		Return(s.mockAccountSignersBatchInsertBuilder).Twice()
+	s.mockAccountSignersBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
 
 	s.processor = NewSignersProcessor(s.mockQ, true)
 }
@@ -181,16 +184,17 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestNoTransactions() {
 }
 
 func (s *AccountsSignerProcessorTestSuiteLedger) TestNewAccount() {
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			int32(1),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				int32(1),
+				null.String{},
+			},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeAccount,
@@ -247,27 +251,28 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestNewSigner() {
 		Return(int64(1), nil).Once()
 
 	// Create new and old signer
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
-			int32(10),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
+				int32(10),
+				null.String{},
+			},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
-			int32(15),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
+				int32(15),
+				null.StringFromPtr(nil)},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeAccount,
@@ -329,16 +334,17 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestSignerRemoved() {
 		Return(int64(1), nil).Once()
 
 	// Create new signer
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
-			int32(15),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
+				int32(15),
+				null.String{},
+			},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeAccount,
@@ -402,16 +408,17 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestSignerPreAuthTxRemovedTxFai
 		Return(int64(1), nil).Once()
 
 	// Create new signer
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
-			int32(10),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
+				int32(10),
+				null.String{},
+			},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeAccount,
@@ -479,44 +486,6 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestRemoveAccount() {
 	s.Assert().NoError(s.processor.Commit(s.ctx))
 }
 
-func (s *AccountsSignerProcessorTestSuiteLedger) TestNewAccountNoRowsAffected() {
-	s.mockQ.
-		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			int32(1),
-			(*string)(nil),
-		).
-		Return(int64(0), nil).Once()
-
-	err := s.processor.ProcessChange(s.ctx, ingest.Change{
-		Type: xdr.LedgerEntryTypeAccount,
-		Pre:  nil,
-		Post: &xdr.LedgerEntry{
-			Data: xdr.LedgerEntryData{
-				Type: xdr.LedgerEntryTypeAccount,
-				Account: &xdr.AccountEntry{
-					AccountId:  xdr.MustAddress("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-					Thresholds: [4]byte{1, 1, 1, 1},
-				},
-			},
-		},
-	})
-	s.Assert().NoError(err)
-
-	err = s.processor.Commit(s.ctx)
-	s.Assert().Error(err)
-	s.Assert().IsType(ingest.StateError{}, errors.Cause(err))
-	s.Assert().EqualError(
-		err,
-		"0 rows affected when inserting "+
-			"account=GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML "+
-			"signer=GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML to database",
-	)
-}
-
 func (s *AccountsSignerProcessorTestSuiteLedger) TestRemoveAccountNoRowsAffected() {
 	s.mockQ.
 		On(
@@ -566,27 +535,28 @@ func (s *AccountsSignerProcessorTestSuiteLedger) TestProcessUpgradeChange() {
 		Return(int64(1), nil).Once()
 
 	// Create new and old (updated) signer
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
-			int32(12),
-			(*string)(nil),
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCBBDQLCTNASZJ3MTKAOYEOWRGSHDFAJVI7VPZUOP7KXNHYR3HP2BUKV",
+				int32(12),
+				null.String{},
+			},
 		).
-		Return(int64(1), nil).Once()
+		Return(nil).Once()
 
-	s.mockQ.
+	s.mockAccountSignersBatchInsertBuilder.
 		On(
-			"CreateAccountSigner",
-			s.ctx,
-			"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
-			int32(15),
-			(*string)(nil),
-		).
-		Return(int64(1), nil).Once()
+			"Add",
+			history.AccountSigner{
+				"GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				"GCAHY6JSXQFKWKP6R7U5JPXDVNV4DJWOWRFLY3Y6YPBF64QRL4BPFDNS",
+				int32(15),
+				null.String{},
+			},
+		).Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeAccount,