ledger: Clear Merkle Trie on Commit Error

daemon/algod/server.go

@@ -231,6 +231,10 @@
 		return fmt.Errorf("couldn't initialize the node: %s", err)
 	}
 	s.node = serverNode
+
+	// When a caller to logging uses Fatal, we want to stop the node before os.Exit is called.
+	logging.RegisterExitHandler(s.Stop)
+
 	return nil
 }
 

ledger/acctonline.go

@@ -349,7 +349,11 @@
 	return offset
 }
 
-func (ao *onlineAccounts) handleUnorderedCommitOrError(dcc *deferredCommitContext) {
+func (ao *onlineAccounts) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (ao *onlineAccounts) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (ao *onlineAccounts) handleCommitError(dcc *deferredCommitContext) {
 }
 
 func (ao *onlineAccounts) maxBalLookback() uint64 {

ledger/acctupdates.go

@@ -1414,7 +1414,11 @@
 	return off, nil
 }
 
-func (au *accountUpdates) handleUnorderedCommitOrError(dcc *deferredCommitContext) {
+func (au *accountUpdates) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (au *accountUpdates) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (au *accountUpdates) handleCommitError(dcc *deferredCommitContext) {
 }
 
 // prepareCommit prepares data to write to the database a "chunk" of rounds, and update the cached dbRound accordingly.

ledger/bulletin.go

@@ -123,7 +123,11 @@
 func (b *bulletin) postCommitUnlocked(ctx context.Context, dcc *deferredCommitContext) {
 }
 
-func (b *bulletin) handleUnorderedCommitOrError(*deferredCommitContext) {
+func (b *bulletin) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (b *bulletin) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (b *bulletin) handleCommitError(dcc *deferredCommitContext) {
 }
 
 func (b *bulletin) produceCommittingTask(committedRound basics.Round, dbRound basics.Round, dcr *deferredCommitRange) *deferredCommitRange {

ledger/catchpointtracker.go

@@ -953,10 +953,27 @@ func (ct *catchpointTracker) postCommitUnlocked(ctx context.Context, dcc *deferr
 	}
 }
 
-// handleUnorderedCommitOrError is a special method for handling deferred commits that are out of order.
-// Tracker might update own state in this case. For example, account catchpoint tracker cancels
-// scheduled catchpoint writing that deferred commit.
-func (ct *catchpointTracker) handleUnorderedCommitOrError(dcc *deferredCommitContext) {
+// when the deferred commit is found to be out of order, cancel writing
+func (ct *catchpointTracker) handleUnorderedCommit(dcc *deferredCommitContext) {
+	ct.cancelWrite(dcc)
+}
+
+// if an error is encountered during commit preparation, cancel writing
+func (ct *catchpointTracker) handlePrepareCommitError(dcc *deferredCommitContext) {
+	ct.cancelWrite(dcc)
+}
+
+// if an error is encountered during commit, cancel writing and clear the balances trie
+func (ct *catchpointTracker) handleCommitError(dcc *deferredCommitContext) {
+	// in cases where the commitRound fails, it is not certain that the merkle trie is in a clean state, and should be cleared.
+	// Specifically, modifications to the trie happen through accountsUpdateBalances,
+	// which happens before commit to disk. Errors in this tracker, subsequent trackers, or the commit to disk may cause the trie cache to be incorrect,
+	// affecting the perceived root on subsequent rounds
+	ct.balancesTrie = nil
+	ct.cancelWrite(dcc)
+}
+
+func (ct *catchpointTracker) cancelWrite(dcc *deferredCommitContext) {
 	// if the node is configured to generate catchpoint files, we might need to update the catchpointWriting variable.
 	if ct.enableGeneratingCatchpointFiles {
 		// determine if this was a catchpoint round

ledger/catchpointtracker_test.go

@@ -342,6 +342,88 @@ func createCatchpoint(t *testing.T, ct *catchpointTracker, accountsRound basics.
 	require.NoError(t, err)
 }
 
+// TestCatchpointCommitErrorHandling exists to confirm that when an error occurs during catchpoint generation,
+// the catchpoint tracker will clear the appropriate state - specifically, the balancesTrie will be cleared,
+// and the balancesTrie will remain functional if loaded from disk, or if lazily loaded during commitRound
+func TestCatchpointCommitErrorHandling(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	temporaryDirectory := t.TempDir()
+
+	accts := []map[basics.Address]basics.AccountData{ledgertesting.RandomAccounts(20, true)}
+	ml := makeMockLedgerForTracker(t, true, 10, protocol.ConsensusCurrentVersion, accts)
+	defer ml.Close()
+
+	ct := &catchpointTracker{}
+	conf := config.GetDefaultLocal()
+
+	conf.Archival = true
+	ct.initialize(conf, ".")
+	defer ct.close()
+	ct.dbDirectory = temporaryDirectory
+
+	_, err := trackerDBInitialize(ml, true, ct.dbDirectory)
+	require.NoError(t, err)
+
+	err = ct.loadFromDisk(ml, ml.Latest())
+	require.NoError(t, err)
+
+	txn, err := ml.dbs.BeginTransaction(context.Background())
+	require.NoError(t, err)
+	dcc := deferredCommitContext{
+		compactKvDeltas: map[string]modifiedKvValue{"key": {data: []byte("value")}},
+	}
+
+	// before commitRound is called, record the trie RootHash
+	require.NotNil(t, ct.balancesTrie)
+	root1, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+
+	ct.commitRound(context.Background(), txn, &dcc)
+
+	txn.Commit()
+
+	// after commitRound is called, confirm the RootHash has changed
+	root2, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+	require.NotEqual(t, root1, root2)
+
+	// demonstrate that handleUnordered does not restore the trie
+	ct.handleUnorderedCommit(&dcc)
+	root2a, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+	require.Equal(t, root2, root2a)
+
+	// demonstrate that handlePrepareCommitError does not restore the trie
+	ct.handlePrepareCommitError(&dcc)
+	root2b, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+	require.Equal(t, root2, root2b)
+
+	// now have the ct handle a commit error
+	ct.handleCommitError(&dcc)
+	// after error handling, the trie should be nil
+	require.Nil(t, ct.balancesTrie)
+
+	// after reloading from disk, the trie should be equal to root1
+	err = ct.loadFromDisk(ml, ml.Latest())
+	require.NoError(t, err)
+	root3, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+	require.Equal(t, root1, root3)
+
+	// also demonstrate that lazy initialization allows a nil trie to go back to root2 immediately after error if the same delta is applied
+	txn, err = ml.dbs.BeginTransaction(context.Background())
+	require.NoError(t, err)
+	ct.handleCommitError(&dcc) // clear trie
+	require.Nil(t, ct.balancesTrie)
+	ct.commitRound(context.Background(), txn, &dcc)
+	txn.Commit()
+	root4, err := ct.balancesTrie.RootHash()
+	require.NoError(t, err)
+	require.Equal(t, root2, root4)
+}
+
 // TestCatchpointFileWithLargeSpVerification makes sure that CatchpointFirstStageInfo.BiggestChunkLen is calculated based on state proof verification contexts
 // as well as other chunks in the catchpoint files.
 func TestCatchpointFileWithLargeSpVerification(t *testing.T) {
@@ -702,8 +784,12 @@ func (bt *blockingTracker) postCommitUnlocked(ctx context.Context, dcc *deferred
 	}
 }
 
-// handleUnorderedCommitOrError is not used by the blockingTracker
-func (bt *blockingTracker) handleUnorderedCommitOrError(*deferredCommitContext) {
+// control functions are not used by the blockingTracker
+func (bt *blockingTracker) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (bt *blockingTracker) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (bt *blockingTracker) handleCommitError(dcc *deferredCommitContext) {
 }
 
 // close is not used by the blockingTracker

ledger/metrics.go

@@ -87,7 +87,11 @@
 func (mt *metricsTracker) postCommitUnlocked(ctx context.Context, dcc *deferredCommitContext) {
 }
 
-func (mt *metricsTracker) handleUnorderedCommitOrError(*deferredCommitContext) {
+func (mt *metricsTracker) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (mt *metricsTracker) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (mt *metricsTracker) handleCommitError(dcc *deferredCommitContext) {
 }
 
 func (mt *metricsTracker) produceCommittingTask(committedRound basics.Round, dbRound basics.Round, dcr *deferredCommitRange) *deferredCommitRange {

ledger/notifier.go

@@ -123,7 +123,11 @@
 func (bn *blockNotifier) postCommitUnlocked(ctx context.Context, dcc *deferredCommitContext) {
 }
 
-func (bn *blockNotifier) handleUnorderedCommitOrError(*deferredCommitContext) {
+func (bn *blockNotifier) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (bn *blockNotifier) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (bn *blockNotifier) handleCommitError(dcc *deferredCommitContext) {
 }
 
 func (bn *blockNotifier) produceCommittingTask(committedRound basics.Round, dbRound basics.Round, dcr *deferredCommitRange) *deferredCommitRange {

ledger/spverificationtracker.go

@@ -160,7 +160,11 @@
 func (spt *spVerificationTracker) postCommitUnlocked(context.Context, *deferredCommitContext) {
 }
 
-func (spt *spVerificationTracker) handleUnorderedCommitOrError(*deferredCommitContext) {
+func (spt *spVerificationTracker) handleUnorderedCommit(dcc *deferredCommitContext) {
+}
+func (spt *spVerificationTracker) handlePrepareCommitError(dcc *deferredCommitContext) {
+}
+func (spt *spVerificationTracker) handleCommitError(dcc *deferredCommitContext) {
 }
 
 func (spt *spVerificationTracker) close() {

ledger/store/trackerdb/interface.go

@@ -19,6 +19,7 @@
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/algorand/go-algorand/config"
 	"github.com/algorand/go-algorand/crypto"
@@ -30,6 +31,19 @@
 // ErrNotFound is returned when a record is not found.
 var ErrNotFound = errors.New("trackerdb: not found")
 
+// ErrIoErr is returned when a Disk/IO error is encountered
+type ErrIoErr struct {
+	InnerError error
+}
+
+func (e *ErrIoErr) Error() string {
+	return fmt.Sprintf("trackerdb: io error: %v", e.InnerError)
+}
+
+func (e *ErrIoErr) Unwrap() error {
+	return e.InnerError
+}
+
 // AccountRef is an opaque ref to an account in the db.
 type AccountRef interface {
 	AccountRefMarker()

ledger/store/trackerdb/sqlitedriver/sql_test.go

@@ -19,12 +19,15 @@ package sqlitedriver
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"testing"
 
 	"github.com/algorand/go-algorand/data/basics"
 	storetesting "github.com/algorand/go-algorand/ledger/store/testing"
+	"github.com/algorand/go-algorand/ledger/store/trackerdb"
 	"github.com/algorand/go-algorand/protocol"
 	"github.com/algorand/go-algorand/test/partitiontest"
+	"github.com/mattn/go-sqlite3"
 	"github.com/stretchr/testify/require"
 )
 
@@ -81,3 +84,28 @@ func TestAccountsDbQueriesCreateClose(t *testing.T) {
 	qs.Close()
 	require.Nil(t, qs.lookupAccountStmt)
 }
+
+// TestWrapIOError ensures that SQL ErrIOErr is converted to trackerdb.ErrIoErr
+// github.com/mattn/go-sqlite3/blob/master/error.go
+// github.com/mattn/go-sqlite3/blob/master/sqlite3.go#L830
+func TestWrapIOError(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	t.Parallel()
+
+	// This structure is how sqlite3 returns Errors
+	err := sqlite3.Error{Code: sqlite3.ErrIoErr}
+	var trackerIOErr *trackerdb.ErrIoErr
+	require.ErrorAs(t, wrapIOError(err), &trackerIOErr)
+
+	// ErrNo10 is a sqlite3 error code for ErrIoErr
+	err = sqlite3.Error{Code: sqlite3.ErrNo(10)}
+	require.ErrorAs(t, wrapIOError(err), &trackerIOErr)
+
+	err = sqlite3.Error{Code: sqlite3.ErrSchema}
+	require.False(t, errors.As(wrapIOError(err), &trackerIOErr))
+
+	// confirm that double wrapping only applies once
+	err = sqlite3.Error{Code: sqlite3.ErrIoErr}
+	require.Equal(t, wrapIOError(err), wrapIOError(wrapIOError(err)))
+
+}