Local Ledger Deployment

README.md

@@ -195,7 +195,7 @@ Settings can be provided from the command line, a configuration file, or an envi
 | default-balances-limit        |         | default-balances-limit        | INDEXER_DEFAULT_BALANCES_LIMIT        |
 | max-applications-limit        |         | max-applications-limit        | INDEXER_MAX_APPLICATIONS_LIMIT        |
 | default-applications-limit    |         | default-applications-limit    | INDEXER_DEFAULT_APPLICATIONS_LIMIT    |
-
+| data-dir                      | i       | data-dir                      | INDEXER_DATA_DIR                      |
 ## Command line
 
 The command line arguments always take priority over the config file and environment variables.

api/handlers_e2e_test.go

@@ -13,15 +13,17 @@ import (
 	"time"
 
 	"github.com/algorand/go-algorand/crypto"
+	"github.com/algorand/go-algorand/crypto/merklesignature"
+	"github.com/algorand/go-algorand/data/basics"
+	"github.com/algorand/go-algorand/data/bookkeeping"
+	"github.com/algorand/go-algorand/rpcs"
+	"github.com/algorand/indexer/processor/blockprocessor"
 	"github.com/labstack/echo/v4"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/algorand/go-algorand-sdk/encoding/json"
-	"github.com/algorand/go-algorand/crypto/merklesignature"
-	"github.com/algorand/go-algorand/data/basics"
-	"github.com/algorand/go-algorand/data/bookkeeping"
 	"github.com/algorand/go-algorand/data/transactions"
 
 	"github.com/algorand/indexer/api/generated/v2"
@@ -70,9 +72,6 @@ func setupIdb(t *testing.T, genesis bookkeeping.Genesis, genesisBlock bookkeepin
 	err = db.LoadGenesis(genesis)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&genesisBlock)
-	require.NoError(t, err)
-
 	return db, newShutdownFunc
 }
 
@@ -112,7 +111,12 @@ func TestApplicationHandlers(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &txn, &optInTxnA, &optInTxnB)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	//////////
@@ -238,7 +242,12 @@ func TestAccountExcludeParameters(t *testing.T) {
 		&appOptInTxnA, &appOptInTxnB, &assetOptInTxnA, &assetOptInTxnB)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	//////////
@@ -443,7 +452,12 @@ func TestAccountMaxResultsLimit(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, ptxns...)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	//////////
@@ -845,7 +859,12 @@ func TestInnerTxn(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &appCall)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	for _, tc := range testcases {
@@ -897,7 +916,12 @@ func TestPagingRootTxnDeduplication(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &appCall)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	testcases := []struct {
@@ -1044,7 +1068,12 @@ func TestKeyregTransactionWithStateProofKeys(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &txn)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	e := echo.New()
@@ -1147,7 +1176,12 @@ func TestAccountClearsNonUTF8(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &createAsset)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	verify := func(params generated.AssetParams) {
@@ -1268,7 +1302,12 @@ func TestLookupInnerLogs(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &appCall)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	for _, tc := range testcases {
@@ -1366,7 +1405,12 @@ func TestLookupMultiInnerLogs(t *testing.T) {
 	block, err := test.MakeBlockForTxns(test.MakeGenesisBlock().BlockHeader, &appCall)
 	require.NoError(t, err)
 
-	err = db.AddBlock(&block)
+	l := test.MakeTestLedger("ledger")
+	defer l.Close()
+	proc, err := blockprocessor.MakeProcessor(l, db.AddBlock)
+	require.NoError(t, err, "failed to open ledger")
+	blockCert := rpcs.EncodedBlockCert{Block: block}
+	err = proc.Process(&blockCert)
 	require.NoError(t, err, "failed to commit")
 
 	for _, tc := range testcases {

cmd/algorand-indexer/daemon.go

@@ -3,24 +3,36 @@ package main
 import (
 	"context"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"os"
 	"os/signal"
+	"path"
+	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
 	"time"
 
-	"github.com/spf13/cobra"
-	"github.com/spf13/viper"
-
+	"github.com/algorand/go-algorand-sdk/client/v2/algod"
+	algodConfig "github.com/algorand/go-algorand/config"
+	"github.com/algorand/go-algorand/data/bookkeeping"
+	"github.com/algorand/go-algorand/ledger"
+	"github.com/algorand/go-algorand/logging"
+	"github.com/algorand/go-algorand/protocol"
 	"github.com/algorand/go-algorand/rpcs"
 	"github.com/algorand/indexer/api"
 	"github.com/algorand/indexer/api/generated/v2"
 	"github.com/algorand/indexer/config"
 	"github.com/algorand/indexer/fetcher"
 	"github.com/algorand/indexer/idb"
 	"github.com/algorand/indexer/importer"
+	"github.com/algorand/indexer/processor"
+	"github.com/algorand/indexer/processor/blockprocessor"
+	"github.com/algorand/indexer/util"
 	"github.com/algorand/indexer/util/metrics"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
 )
 
 var (
@@ -118,11 +130,19 @@ var daemonCmd = &cobra.Command{
 		}
 
 		opts.MaxConn = maxConn
+		opts.IndexerDatadir = indexerDataDir
+		opts.AlgodDataDir = algodDataDir
+		opts.AlgodToken = algodToken
+		opts.AlgodAddr = algodAddr
 
 		db, availableCh := indexerDbFromFlags(opts)
 		defer db.Close()
 		var wg sync.WaitGroup
 		if bot != nil {
+			if indexerDataDir == "" {
+				fmt.Fprint(os.Stderr, "missing indexer data directory")
+				os.Exit(1)
+			}
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
@@ -135,13 +155,27 @@ var daemonCmd = &cobra.Command{
 				_, err := importer.EnsureInitialImport(db, genesisReader, logger)
 				maybeFail(err, "importer.EnsureInitialImport() error")
 				logger.Info("Initializing block import handler.")
-
-				nextRound, err := db.GetNextRoundToAccount()
-				maybeFail(err, "failed to get next round, %v", err)
-				bot.SetNextRound(nextRound)
-
 				imp := importer.NewImporter(db)
-				handler := blockHandler(imp, 1*time.Second)
+
+				genesisReader = importer.GetGenesisFile(genesisJSONPath, bot.Algod(), logger)
+				genesis, err := readGenesis(genesisReader)
+				maybeFail(err, "Error reading genesis file")
+				genesisBlock, err := getGenesisBlock(bot.Algod())
+				maybeFail(err, "Error getting genesis block")
+				initState, err := util.CreateInitState(&genesis, &genesisBlock)
+				maybeFail(err, "Error creating init state")
+
+				logger.Info("Initializing local ledger.")
+				localLedger, err := ledger.OpenLedger(logging.NewLogger(), filepath.Join(path.Dir(indexerDataDir), "ledger"), false, initState, algodConfig.GetDefaultLocal())
+				maybeFail(err, "failed to open ledger %v", err)
+				defer localLedger.Close()
+				bot.SetNextRound(uint64(localLedger.Latest()) + 1)
+
+				proc, err := blockprocessor.MakeProcessor(localLedger, imp.ImportBlock)
+				if err != nil {
+					maybeFail(err, "blockprocessor.MakeProcessor() err %v", err)
+				}
+				handler := blockHandler(proc, 1*time.Second)
 				bot.SetBlockHandler(handler)
 
 				logger.Info("Starting block importer.")
@@ -265,10 +299,10 @@ func makeOptions() (options api.ExtraOptions) {
 
 // blockHandler creates a handler complying to the fetcher block handler interface. In case of a failure it keeps
 // attempting to add the block until the fetcher shuts down.
-func blockHandler(imp importer.Importer, retryDelay time.Duration) func(context.Context, *rpcs.EncodedBlockCert) error {
+func blockHandler(proc processor.Processor, retryDelay time.Duration) func(context.Context, *rpcs.EncodedBlockCert) error {
 	return func(ctx context.Context, block *rpcs.EncodedBlockCert) error {
 		for {
-			err := handleBlock(block, imp)
+			err := handleBlock(block, proc)
 			if err == nil {
 				// return on success.
 				return nil
@@ -285,12 +319,12 @@ func blockHandler(imp importer.Importer, retryDelay time.Duration) func(context.
 	}
 }
 
-func handleBlock(block *rpcs.EncodedBlockCert, imp importer.Importer) error {
+func handleBlock(block *rpcs.EncodedBlockCert, proc processor.Processor) error {
 	start := time.Now()
-	err := imp.ImportBlock(block)
+	err := proc.Process(block)
 	if err != nil {
 		logger.WithError(err).Errorf(
-			"adding block %d to database failed", block.Block.Round())
+			"block %d import failed", block.Block.Round())
 		return fmt.Errorf("handleBlock() err: %w", err)
 	}
 	dt := time.Since(start)
@@ -313,3 +347,34 @@ func handleBlock(block *rpcs.EncodedBlockCert, imp importer.Importer) error {
 
 	return nil
 }
+
+func readGenesis(reader io.Reader) (bookkeeping.Genesis, error) {
+	var genesis bookkeeping.Genesis
+	if reader == nil {
+		return bookkeeping.Genesis{}, fmt.Errorf("readGenesis() err: reader is nil")
+	}
+	gbytes, err := ioutil.ReadAll(reader)
+	if err != nil {
+		return bookkeeping.Genesis{}, fmt.Errorf("readGenesis() err: %w", err)
+	}
+	err = protocol.DecodeJSON(gbytes, &genesis)
+	if err != nil {
+		return bookkeeping.Genesis{}, fmt.Errorf("readGenesis() err: %w", err)
+	}
+	return genesis, nil
+}
+
+func getGenesisBlock(client *algod.Client) (bookkeeping.Block, error) {
+	data, err := client.BlockRaw(0).Do(context.Background())
+	if err != nil {
+		return bookkeeping.Block{}, fmt.Errorf("getGenesisBlock() client err: %w", err)
+	}
+
+	var block rpcs.EncodedBlockCert
+	err = protocol.Decode(data, &block)
+	if err != nil {
+		return bookkeeping.Block{}, fmt.Errorf("getGenesisBlock() decode err: %w", err)
+	}
+
+	return block.Block, nil
+}