exp/ingest: Fix fee and transaction meta processing

exp/ingest/io/ledger_transaction.go

@@ -146,12 +146,16 @@ func (c *Change) AccountSignersChanged() bool {
 	return false
 }
 
+// GetFeeChanges returns a developer friendly representation of LedgerEntryChanges
+// connected to fees.
+func (t *LedgerTransaction) GetFeeChanges() []Change {
+	return getChangesFromLedgerEntryChanges(t.FeeChanges)
+}
+
 // GetChanges returns a developer friendly representation of LedgerEntryChanges.
-// It contains fee changes, transaction changes and operation changes in that
-// order.
+// It contains transaction changes and operation changes in that order.
 func (t *LedgerTransaction) GetChanges() []Change {
-	// Fee meta
-	changes := getChangesFromLedgerEntryChanges(t.FeeChanges)
+	var changes []Change
 
 	// Transaction meta
 	switch t.Meta.V {

exp/ingest/io/ledger_transaction_test.go

@@ -17,6 +17,78 @@ func TestChangeAccountChangedExceptSignersInvalidType(t *testing.T) {
 	})
 }
 
+func TestFeeAndMetaChangesSeparate(t *testing.T) {
+	tx := LedgerTransaction{
+		FeeChanges: xdr.LedgerEntryChanges{
+			xdr.LedgerEntryChange{
+				Type: xdr.LedgerEntryChangeTypeLedgerEntryState,
+				State: &xdr.LedgerEntry{
+					Data: xdr.LedgerEntryData{
+						Type: xdr.LedgerEntryTypeAccount,
+						Account: &xdr.AccountEntry{
+							AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+							Balance:   100,
+						},
+					},
+				},
+			},
+			xdr.LedgerEntryChange{
+				Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated,
+				Updated: &xdr.LedgerEntry{
+					Data: xdr.LedgerEntryData{
+						Type: xdr.LedgerEntryTypeAccount,
+						Account: &xdr.AccountEntry{
+							AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+							Balance:   200,
+						},
+					},
+				},
+			},
+		},
+		Meta: xdr.TransactionMeta{
+			Operations: &[]xdr.OperationMeta{
+				{
+					Changes: xdr.LedgerEntryChanges{
+						xdr.LedgerEntryChange{
+							Type: xdr.LedgerEntryChangeTypeLedgerEntryState,
+							State: &xdr.LedgerEntry{
+								Data: xdr.LedgerEntryData{
+									Type: xdr.LedgerEntryTypeAccount,
+									Account: &xdr.AccountEntry{
+										AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+										Balance:   300,
+									},
+								},
+							},
+						},
+						xdr.LedgerEntryChange{
+							Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated,
+							Updated: &xdr.LedgerEntry{
+								Data: xdr.LedgerEntryData{
+									Type: xdr.LedgerEntryTypeAccount,
+									Account: &xdr.AccountEntry{
+										AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+										Balance:   400,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}}
+
+	feeChanges := tx.GetFeeChanges()
+	assert.Len(t, feeChanges, 1)
+	assert.Equal(t, feeChanges[0].Pre.Data.MustAccount().Balance, xdr.Int64(100))
+	assert.Equal(t, feeChanges[0].Post.Data.MustAccount().Balance, xdr.Int64(200))
+
+	metaChanges := tx.GetChanges()
+	assert.Len(t, metaChanges, 1)
+	assert.Equal(t, metaChanges[0].Pre.Data.MustAccount().Balance, xdr.Int64(300))
+	assert.Equal(t, metaChanges[0].Post.Data.MustAccount().Balance, xdr.Int64(400))
+}
+
 func TestChangeAccountChangedExceptSignersLastModifiedLedgerSeq(t *testing.T) {
 	change := Change{
 		Type: xdr.LedgerEntryTypeAccount,

services/horizon/internal/expingest/main.go

@@ -39,7 +39,9 @@ const (
 	//      when preauth tx is failed.
 	// - 9: Fixes a bug in asset stats processor that counted unauthorized
 	//      trustlines.
-	CurrentVersion = 9
+	// - 9: Fixes a bug in meta processing (fees are now processed before
+	//      everything else).
+	CurrentVersion = 10
 )
 
 var log = logpkg.DefaultLogger.WithField("service", "expingest")

services/horizon/internal/expingest/processors/accounts_processor_test.go

@@ -410,3 +410,93 @@ func (s *AccountsProcessorTestSuiteLedger) TestProcessUpgradeChange() {
 
 	s.Assert().NoError(err)
 }
+
+func (s *AccountsProcessorTestSuiteLedger) TestFeeProcessedBeforeEverythingElse() {
+	s.mockLedgerReader.On("Read").
+		Return(io.LedgerTransaction{
+			Meta: createTransactionMeta([]xdr.OperationMeta{
+				xdr.OperationMeta{
+					Changes: []xdr.LedgerEntryChange{
+						xdr.LedgerEntryChange{
+							Type: xdr.LedgerEntryChangeTypeLedgerEntryState,
+							State: &xdr.LedgerEntry{
+								Data: xdr.LedgerEntryData{
+									Type: xdr.LedgerEntryTypeAccount,
+									Account: &xdr.AccountEntry{
+										AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+										Balance:   100,
+									},
+								},
+							},
+						},
+						xdr.LedgerEntryChange{
+							Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated,
+							Updated: &xdr.LedgerEntry{
+								Data: xdr.LedgerEntryData{
+									Type: xdr.LedgerEntryTypeAccount,
+									Account: &xdr.AccountEntry{
+										AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+										Balance:   300,
+									},
+								},
+							},
+						},
+					},
+				},
+			}),
+		}, nil).Once()
+
+	s.mockLedgerReader.On("Read").
+		Return(io.LedgerTransaction{
+			FeeChanges: []xdr.LedgerEntryChange{
+				xdr.LedgerEntryChange{
+					Type: xdr.LedgerEntryChangeTypeLedgerEntryState,
+					State: &xdr.LedgerEntry{
+						Data: xdr.LedgerEntryData{
+							Type: xdr.LedgerEntryTypeAccount,
+							Account: &xdr.AccountEntry{
+								AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+								Balance:   200,
+							},
+						},
+					},
+				},
+				xdr.LedgerEntryChange{
+					Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated,
+					Updated: &xdr.LedgerEntry{
+						Data: xdr.LedgerEntryData{
+							Type: xdr.LedgerEntryTypeAccount,
+							Account: &xdr.AccountEntry{
+								AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+								Balance:   100,
+							},
+						},
+					},
+				},
+			},
+			Meta: xdr.TransactionMeta{
+				Operations: &[]xdr.OperationMeta{},
+			},
+		}, nil).Once()
+
+	expectedAccount := xdr.AccountEntry{
+		AccountId: xdr.MustAddress("GAHK7EEG2WWHVKDNT4CEQFZGKF2LGDSW2IVM4S5DP42RBW3K6BTODB4A"),
+		// If fee meta wasn't procesed before everything else, this would be 100
+		Balance: 300,
+	}
+
+	s.mockQ.On("UpdateAccount", expectedAccount, xdr.Uint32(0)).Return(int64(1), nil).Once()
+
+	s.mockLedgerReader.
+		On("Read").
+		Return(io.LedgerTransaction{}, stdio.EOF).Once()
+
+	err := s.processor.ProcessLedger(
+		context.Background(),
+		&supportPipeline.Store{},
+		s.mockLedgerReader,
+		s.mockLedgerWriter,
+	)
+
+	s.Assert().NoError(err)
+}

services/horizon/internal/expingest/processors/database_processor.go

@@ -214,7 +214,8 @@ func (p *DatabaseProcessor) ProcessLedger(ctx context.Context, store *pipeline.S
 
 	var successTxCount, failedTxCount, opCount int
 
-	// Process transaction meta
+	// Get all transactions
+	var transactions []io.LedgerTransaction
 	for {
 		transaction, err := r.Read()
 		if err != nil {
@@ -232,27 +233,49 @@ func (p *DatabaseProcessor) ProcessLedger(ctx context.Context, store *pipeline.S
 			failedTxCount++
 		}
 
+		transactions = append(transactions, transaction)
+	}
+
+	if p.Action != Ledgers {
 		// Remember that it's possible that transaction can remove a preauth
-		// tx signer even when it's a failed transaction.
-		if p.Action != Ledgers {
-			for _, change := range transaction.GetChanges() {
+		// tx signer even when it's a failed transaction so we need to check
+		// failed transactions too.
+
+		// Fees are processed before everything else.
+		for _, transaction := range transactions {
+			for _, change := range transaction.GetFeeChanges() {
 				err := ledgerCache.AddChange(change)
 				if err != nil {
-					return errors.Wrap(err, "error addint to ledgerCache")
+					return errors.Wrap(err, "error adding to ledgerCache")
 				}
 			}
+
+			select {
+			case <-ctx.Done():
+				return nil
+			default:
+				continue
+			}
 		}
 
-		select {
-		case <-ctx.Done():
-			return nil
-		default:
-			continue
+		// Tx meta
+		for _, transaction := range transactions {
+			for _, change := range transaction.GetChanges() {
+				err := ledgerCache.AddChange(change)
+				if err != nil {
+					return errors.Wrap(err, "error adding to ledgerCache")
+				}
+			}
+
+			select {
+			case <-ctx.Done():
+				return nil
+			default:
+				continue
+			}
 		}
-	}
 
-	// Process upgrades meta
-	if p.Action != Ledgers {
+		// Process upgrades meta
 		for {
 			change, err := r.ReadUpgradeChange()
 			if err != nil {

services/horizon/internal/expingest/verify.go

@@ -28,7 +28,7 @@ const assetStatsBatchSize = 500
 // check them.
 // There is a test that checks it, to fix it: update the actual `verifyState`
 // method instead of just updating this value!
-const stateVerifierExpectedIngestionVersion = 9
+const stateVerifierExpectedIngestionVersion = 10
 
 // verifyState is called as a go routine from pipeline post hook every 64
 // ledgers. It checks if the state is correct. If another go routine is already