Repository: algorand/indexer
Pull request #280 (2021-01-15 20:58:23)

Remove idb from accounting.State
## Summary

Move db.CommitRoundAccounting out of the accounting package. This will make it easier to use the AddTransaction function in different ways. For example the RoundAccounting object can be extracted and filtered such that it only applies updates for a single account.


Patch:
diff --git a/accounting/accounting.go b/accounting/accounting.go
index b8a19d9..df74ce5 100644
--- a/accounting/accounting.go
+++ b/accounting/accounting.go
@@ -2,6 +2,7 @@ package accounting
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"math/big"
 
@@ -14,12 +15,9 @@ import (
 
 // State is used to record accounting changes as a result of processing transactions.
 type State struct {
-	db idb.IndexerDb
-
 	defaultFrozen map[uint64]bool
 
 	currentRound uint64
-	dirty        bool
 
 	idb.RoundUpdates
 
@@ -36,52 +34,20 @@ type State struct {
 }
 
 // New creates a new State object.
-func New(db idb.IndexerDb) *State {
-	return &State{db: db, defaultFrozen: make(map[uint64]bool)}
-}
-
-func (accounting *State) getTxnCounter(round uint64) (txnCounter uint64, err error) {
-	if round != accounting.txnCounterRound {
-		block, err := accounting.db.GetBlock(round)
-		if err != nil {
-			return 0, err
-		}
-		accounting.txnCounter = block.TxnCounter
-		accounting.txnCounterRound = round
-	}
-	return accounting.txnCounter, nil
+func New() *State {
+	return &State{defaultFrozen: make(map[uint64]bool)}
 }
 
-func (accounting *State) initRound(round uint64) error {
-	block, err := accounting.db.GetBlock(round)
-	if err != nil {
-		return err
-	}
+// InitRound should be called before each round to initialize the accounting state.
+func (accounting *State) InitRound(block types.Block) error {
+	accounting.RoundUpdates.Clear()
 	accounting.feeAddr = block.FeeSink
 	accounting.rewardAddr = block.RewardsPool
 	accounting.rewardsLevel = block.RewardsLevel
-	accounting.currentRound = round
-	return nil
-}
-
-func (accounting *State) commitRound() error {
-	if !accounting.dirty {
-		return nil
-	}
-	err := accounting.db.CommitRoundAccounting(accounting.RoundUpdates, accounting.currentRound, accounting.rewardsLevel)
-	if err != nil {
-		return err
-	}
-	accounting.RoundUpdates.Clear()
-	accounting.dirty = false
+	accounting.currentRound = uint64(block.Round)
 	return nil
 }
 
-// Close is part of the Closer interface.
-func (accounting *State) Close() error {
-	return accounting.commitRound()
-}
-
 var zeroAddr = [32]byte{}
 
 func addrIsZero(a types.Address) bool {
@@ -246,6 +212,10 @@ func blankLsig(lsig atypes.LogicSig) bool {
 	return len(lsig.Logic) == 0
 }
 
+// ErrWrongRound is returned when adding a transaction which belongs to a different round
+// than what the State is configured for.
+var ErrWrongRound error = errors.New("wrong round")
+
 // AddTransaction updates the State with the provided idb.TxnRow data.
 func (accounting *State) AddTransaction(txnr *idb.TxnRow) (err error) {
 	round := txnr.Round
@@ -257,16 +227,8 @@ func (accounting *State) AddTransaction(txnr *idb.TxnRow) (err error) {
 		return fmt.Errorf("txn r=%d i=%d failed decode, %v", round, intra, err)
 	}
 	if accounting.currentRound != round {
-		err = accounting.commitRound()
-		if err != nil {
-			return fmt.Errorf("add tx commit round %d, %v", accounting.currentRound, err)
-		}
-		err = accounting.initRound(round)
-		if err != nil {
-			return fmt.Errorf("add tx init round %d, %v", round, err)
-		}
+		return ErrWrongRound
 	}
-	accounting.dirty = true
 
 	var ktype string
 	if !zeroSig(stxn.Sig) {
diff --git a/importer/helper.go b/importer/helper.go
index 5f8ea59..754e293 100644
--- a/importer/helper.go
+++ b/importer/helper.go
@@ -252,17 +252,33 @@ func updateAccounting(db idb.IndexerDb, genesisJSONPath string, numRoundsLimit i
 	}
 
 	lastlog := time.Now()
-	act := accounting.New(db)
+	act := accounting.New()
 	txns := db.YieldTxns(context.Background(), state.AccountRound)
 	currentRound := uint64(0)
 	roundsSeen := 0
 	lastRoundsSeen := roundsSeen
+	txnForRound := 0
+	var blockPtr *types.Block = nil
 	for txn := range txns {
 		maybeFail(txn.Error, l, "updateAccounting txn fetch, %v", txn.Error)
 		if txn.Round != currentRound {
+			// TODO: commit rounds with no transactions to avoid a special case to update the db metastate.
+			if blockPtr != nil && txnForRound > 0 {
+				err = db.CommitRoundAccounting(act.RoundUpdates, currentRound, blockPtr.RewardsLevel)
+				maybeFail(err, l, "failed to commit round accounting")
+			}
+
+			// initialize accounting for next round
+			txnForRound = 0
 			prevRound := currentRound
 			roundsSeen++
 			currentRound = txn.Round
+			block, err := db.GetBlock(currentRound)
+			maybeFail(err, l, "problem fetching next round (%d)", currentRound)
+			blockPtr = &block
+			act.InitRound(block)
+
+			// Log progress
 			if (numRoundsLimit != 0) && (roundsSeen > numRoundsLimit) {
 				l.Infof("hit rounds limit %d > %d", roundsSeen, numRoundsLimit)
 				break
@@ -279,9 +295,16 @@ func updateAccounting(db idb.IndexerDb, genesisJSONPath string, numRoundsLimit i
 		err = act.AddTransaction(&txn)
 		maybeFail(err, l, "txn accounting r=%d i=%d, %v", txn.Round, txn.Intra, err)
 		txnCount++
+		txnForRound++
 	}
-	err = act.Close()
-	maybeFail(err, l, "accounting close %v", err)
+
+	// Commit the final round
+	// TODO: commit rounds with empty paysets to avoid a special case to update the db metastate.
+	if blockPtr != nil && txnForRound > 0 {
+		err = db.CommitRoundAccounting(act.RoundUpdates, currentRound, blockPtr.RewardsLevel)
+		maybeFail(err, l, "failed to commit round accounting")
+	}
+
 	rounds += roundsSeen
 	if rounds > 0 {
 		l.Infof("accounting updated through round %d", currentRound)


Test patch:
diff --git a/accounting/accounting_test.go b/accounting/accounting_test.go
index 181ab74..8727eb2 100644
--- a/accounting/accounting_test.go
+++ b/accounting/accounting_test.go
@@ -9,7 +9,6 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/algorand/indexer/idb"
-	"github.com/algorand/indexer/idb/mocks"
 )
 
 var mainAcct = decodeAddressOrPanic("GJR76Q6OXNZ2CYIVCFCDTJRBAAR6TYEJJENEII3G2U3JH546SPBQA62IFY")
@@ -75,8 +74,7 @@ func assertUpdates(t *testing.T, update *idb.AlgoUpdate, closed bool, balance, r
 }
 
 func getAccounting() *State {
-	mockIndexer := &mocks.IndexerDb{}
-	accountingState := New(mockIndexer)
+	accountingState := New()
 	accountingState.feeAddr = feeAddr
 	accountingState.rewardAddr = rewardAddr
 	accountingState.currentRound = closeMainToBC.Round
diff --git a/test/common.sh b/test/common.sh
index 49c268c..fc54d89 100755
--- a/test/common.sh
+++ b/test/common.sh
@@ -199,8 +199,10 @@ function start_indexer_with_blocks() {
   local TEMPDIR=$(mktemp -d -t ci-XXXXXXX)
   tar -xf "$2" -C $TEMPDIR
 
-  echo "Start args 'import -P \"${CONNECTION_STRING/DB_NAME_HERE/$1}\" --genesis \"$TEMPDIR/algod/genesis.json\" $TEMPDIR/blocktars/*"
-  #sleep infinity
+  if [ ! -z $3 ]; then
+    echo "Start args 'import -P \"${CONNECTION_STRING/DB_NAME_HERE/$1}\" --genesis \"$TEMPDIR/algod/genesis.json\" $TEMPDIR/blocktars/*"
+    sleep infinity
+  fi
   ALGORAND_DATA= ../cmd/algorand-indexer/algorand-indexer import \
     -P "${CONNECTION_STRING/DB_NAME_HERE/$1}" \
     --genesis "$TEMPDIR/algod/genesis.json" \
diff --git a/test/postgres_integration_test.sh b/test/postgres_integration_test.sh
index e156876..bbbeba0 100755
--- a/test/postgres_integration_test.sh
+++ b/test/postgres_integration_test.sh
@@ -18,6 +18,7 @@ start_postgres
 ## RUN TESTS ##
 ###############
 # Test 1
+print_alert "Integration Test 1"
 kill_indexer
 start_indexer_with_blocks createdestroy blockdata/create_destroy.tar.bz2
 wait_for_migrated
diff --git a/test/postgres_migration_test.sh b/test/postgres_migration_test.sh
index 3978bb7..436f3ba 100755
--- a/test/postgres_migration_test.sh
+++ b/test/postgres_migration_test.sh
@@ -34,6 +34,7 @@ start_postgres
 ###############
 
 # Rewards tests
+print_alert "Migration Rewards Tests"
 kill_indexer
 initialize_db test1 migrations/cumulative_rewards_dump.txt
 start_indexer test1
@@ -41,6 +42,7 @@ wait_for_migrated
 cumulative_rewards_tests
 
 # Create Delete Tests
+print_alert "Create Delete Tests"
 kill_indexer
 initialize_db test2 migrations/create_delete.2.2.1.txt
 start_indexer test2