Merge pull request #2110 from onetechnical/onetechnical/relstable2.5.6

go-algorand 2.5.6-stable

Author: algojohnlee

buildnumber.dat

@@ -1 +1 @@
-5
+6

config/consensus.go

@@ -348,6 +348,9 @@ type ConsensusParams struct {
 
 	// update the initial rewards rate calculation to take the reward pool minimum balance into account
 	InitialRewardsRateCalculation bool
+
+	// NoEmptyLocalDeltas updates how ApplyDelta.EvalDelta.LocalDeltas are stored
+	NoEmptyLocalDeltas bool
 }
 
 // PaysetCommitType enumerates possible ways for the block header to commit to
@@ -872,9 +875,24 @@ func initConsensusProtocols() {
 	v25.ApprovedUpgrades[protocol.ConsensusV26] = 140000
 	v24.ApprovedUpgrades[protocol.ConsensusV26] = 140000
 
+	// v27 updates ApplyDelta.EvalDelta.LocalDeltas format
+	v27 := v26
+	v27.ApprovedUpgrades = map[protocol.ConsensusVersion]uint64{}
+
+	// Enable the ApplyDelta.EvalDelta.LocalDeltas fix
+	v27.NoEmptyLocalDeltas = true
+
+	Consensus[protocol.ConsensusV27] = v27
+
+	// v26 can be upgraded to v27, with an update delay of 3 days
+	// 60279 = (3 * 24 * 60 * 60 / 4.3)
+	// for the sake of future manual calculations, we'll round that down
+	// a bit :
+	v26.ApprovedUpgrades[protocol.ConsensusV27] = 60000
+
 	// ConsensusFuture is used to test features that are implemented
 	// but not yet released in a production protocol version.
-	vFuture := v26
+	vFuture := v27
 	vFuture.ApprovedUpgrades = map[protocol.ConsensusVersion]uint64{}
 
 	// FilterTimeout for period 0 should take a new optimized, configured value, need to revisit this later
@@ -895,7 +913,7 @@ func initConsensusProtocols() {
 	Consensus[protocol.ConsensusFuture] = vFuture
 }
 
-// Global defines global Algorand protocol parameters which should not be overriden.
+// Global defines global Algorand protocol parameters which should not be overridden.
 type Global struct {
 	SmallLambda time.Duration // min amount of time to wait for leader's credential (i.e., time to propagate one credential)
 	BigLambda   time.Duration // max amount of time to wait for leader's proposal (i.e., time to propagate one block)

crypto/merklearray/worker.go

@@ -25,6 +25,11 @@ import (
 // workerState describes a group of goroutines processing a sequential list
 // of maxidx elements starting from 0.
 type workerState struct {
+	// maxidx is the total number of elements to process, and nextidx
+	// is the next element that a worker should process.
+	maxidx  uint64
+	nextidx uint64
+
 	// nworkers is the number of workers that can be started.
 	// This field gets decremented once workers are launched,
 	// and represents the number of remaining workers that can
@@ -43,11 +48,6 @@ type workerState struct {
 	// wg tracks outstanding workers, to determine when all workers
 	// have finished their processing.
 	wg sync.WaitGroup
-
-	// maxidx is the total number of elements to process, and nextidx
-	// is the next element that a worker should process.
-	maxidx  uint64
-	nextidx uint64
 }
 
 func newWorkerState(max uint64) *workerState {

data/transactions/logic/eval.go

@@ -134,9 +134,9 @@ type LedgerForLogic interface {
 	CreatorAddress() basics.Address
 	OptedIn(addr basics.Address, appIdx basics.AppIndex) (bool, error)
 
-	GetLocal(addr basics.Address, appIdx basics.AppIndex, key string) (value basics.TealValue, exists bool, err error)
-	SetLocal(addr basics.Address, key string, value basics.TealValue) error
-	DelLocal(addr basics.Address, key string) error
+	GetLocal(addr basics.Address, appIdx basics.AppIndex, key string, accountIdx uint64) (value basics.TealValue, exists bool, err error)
+	SetLocal(addr basics.Address, key string, value basics.TealValue, accountIdx uint64) error
+	DelLocal(addr basics.Address, key string, accountIdx uint64) error
 
 	GetGlobal(appIdx basics.AppIndex, key string) (value basics.TealValue, exists bool, err error)
 	SetGlobal(key string, value basics.TealValue) error
@@ -2059,7 +2059,7 @@ func (cx *evalContext) appReadLocalKey(appIdx uint64, accountIdx uint64, key str
 	if err != nil {
 		return basics.TealValue{}, false, err
 	}
-	return cx.Ledger.GetLocal(addr, basics.AppIndex(appIdx), key)
+	return cx.Ledger.GetLocal(addr, basics.AppIndex(appIdx), key, accountIdx)
 }
 
 // appWriteLocalKey writes value to local key/value cow
@@ -2069,7 +2069,7 @@ func (cx *evalContext) appWriteLocalKey(accountIdx uint64, key string, tv basics
 	if err != nil {
 		return err
 	}
-	return cx.Ledger.SetLocal(addr, key, tv)
+	return cx.Ledger.SetLocal(addr, key, tv, accountIdx)
 }
 
 // appDeleteLocalKey deletes a value from the key/value cow
@@ -2079,7 +2079,7 @@ func (cx *evalContext) appDeleteLocalKey(accountIdx uint64, key string) error {
 	if err != nil {
 		return err
 	}
-	return cx.Ledger.DelLocal(addr, key)
+	return cx.Ledger.DelLocal(addr, key, accountIdx)
 }
 
 func (cx *evalContext) appReadGlobalKey(foreignAppsIndex uint64, key string) (basics.TealValue, bool, error) {

data/transactions/logic/evalStateful_test.go

@@ -258,7 +258,7 @@ func (l *testLedger) DelGlobal(key string) error {
 	return nil
 }
 
-func (l *testLedger) GetLocal(addr basics.Address, appIdx basics.AppIndex, key string) (basics.TealValue, bool, error) {
+func (l *testLedger) GetLocal(addr basics.Address, appIdx basics.AppIndex, key string, accountIdx uint64) (basics.TealValue, bool, error) {
 	if appIdx == 0 {
 		appIdx = l.appID
 	}
@@ -285,7 +285,7 @@ func (l *testLedger) GetLocal(addr basics.Address, appIdx basics.AppIndex, key s
 	return val, ok, nil
 }
 
-func (l *testLedger) SetLocal(addr basics.Address, key string, value basics.TealValue) error {
+func (l *testLedger) SetLocal(addr basics.Address, key string, value basics.TealValue, accountIdx uint64) error {
 	appIdx := l.appID
 
 	br, ok := l.balances[addr]
@@ -313,7 +313,7 @@ func (l *testLedger) SetLocal(addr basics.Address, key string, value basics.Teal
 	return nil
 }
 
-func (l *testLedger) DelLocal(addr basics.Address, key string) error {
+func (l *testLedger) DelLocal(addr basics.Address, key string, accountIdx uint64) error {
 	appIdx := l.appID
 
 	br, ok := l.balances[addr]

ledger/accountdb.go

@@ -114,7 +114,7 @@ var accountsResetExprs = []string{
 // accountDBVersion is the database version that this binary would know how to support and how to upgrade to.
 // details about the content of each of the versions can be found in the upgrade functions upgradeDatabaseSchemaXXXX
 // and their descriptions.
-var accountDBVersion = int32(4)
+var accountDBVersion = int32(5)
 
 // persistedAccountData is used for representing a single account stored on the disk. In addition to the
 // basics.AccountData, it also stores complete referencing information used to maintain the base accounts
@@ -631,6 +631,50 @@ func accountsAddNormalizedBalance(tx *sql.Tx, proto config.ConsensusParams) erro
 	return rows.Err()
 }
 
+// removeEmptyAccountData removes empty AccountData msgp-encoded entries from accountbase table
+// and optionally returns list of addresses that were eliminated
+func removeEmptyAccountData(tx *sql.Tx, queryAddresses bool) (num int64, addresses []basics.Address, err error) {
+	if queryAddresses {
+		rows, err := tx.Query("SELECT address FROM accountbase where length(data) = 1 and data = x'80'") // empty AccountData is 0x80
+		if err != nil {
+			return 0, nil, err
+		}
+		defer rows.Close()
+
+		for rows.Next() {
+			var addrbuf []byte
+			err = rows.Scan(&addrbuf)
+			if err != nil {
+				return 0, nil, err
+			}
+			var addr basics.Address
+			if len(addrbuf) != len(addr) {
+				err = fmt.Errorf("Account DB address length mismatch: %d != %d", len(addrbuf), len(addr))
+				return 0, nil, err
+			}
+			copy(addr[:], addrbuf)
+			addresses = append(addresses, addr)
+		}
+
+		// if the above loop was abrupted by an error, test it now.
+		if err = rows.Err(); err != nil {
+			return 0, nil, err
+		}
+	}
+
+	result, err := tx.Exec("DELETE from accountbase where length(data) = 1 and data = x'80'")
+	if err != nil {
+		return 0, nil, err
+	}
+	num, err = result.RowsAffected()
+	if err != nil {
+		// something wrong on getting rows count but data deleted, ignore the error
+		num = int64(len(addresses))
+		err = nil
+	}
+	return num, addresses, err
+}
+
 // accountDataToOnline returns the part of the AccountData that matters
 // for online accounts (to answer top-N queries).  We store a subset of
 // the full AccountData because we need to store a large number of these

ledger/acctupdates.go

@@ -1069,6 +1069,12 @@ func (au *accountUpdates) accountsInitialize(ctx context.Context, tx *sql.Tx) (b
 					au.log.Warnf("accountsInitialize failed to upgrade accounts database (ledger.tracker.sqlite) from schema 3 : %v", err)
 					return 0, err
 				}
+			case 4:
+				dbVersion, err = au.upgradeDatabaseSchema4(ctx, tx)
+				if err != nil {
+					au.log.Warnf("accountsInitialize failed to upgrade accounts database (ledger.tracker.sqlite) from schema 4 : %v", err)
+					return 0, err
+				}
 			default:
 				return 0, fmt.Errorf("accountsInitialize unable to upgrade database from schema version %d", dbVersion)
 			}
@@ -1324,6 +1330,63 @@ func (au *accountUpdates) upgradeDatabaseSchema3(ctx context.Context, tx *sql.Tx
 	return 4, nil
 }
 
+// upgradeDatabaseSchema4 does not change the schema but migrates data:
+// remove empty AccountData entries from accountbase table
+func (au *accountUpdates) upgradeDatabaseSchema4(ctx context.Context, tx *sql.Tx) (updatedDBVersion int32, err error) {
+
+	queryAddresses := au.catchpointInterval != 0
+	numDeleted, addresses, err := removeEmptyAccountData(tx, queryAddresses)
+	if err != nil {
+		return 0, err
+	}
+
+	if queryAddresses && len(addresses) > 0 {
+		mc, err := makeMerkleCommitter(tx, false)
+		if err != nil {
+			// at this point record deleted and DB is pruned for account data
+			// if hash deletion fails just log it and do not about startup
+			au.log.Errorf("upgradeDatabaseSchema4: failed to create merkle committer: %v", err)
+			goto done
+		}
+		trie, err := merkletrie.MakeTrie(mc, trieMemoryConfig)
+		if err != nil {
+			au.log.Errorf("upgradeDatabaseSchema4: failed to create merkle trie: %v", err)
+			goto done
+		}
+
+		var totalHashesDeleted int
+		for _, addr := range addresses {
+			hash := accountHashBuilder(addr, basics.AccountData{}, []byte{0x80})
+			deleted, err := trie.Delete(hash)
+			if err != nil {
+				au.log.Errorf("upgradeDatabaseSchema4: failed to delete hash '%s' from merkle trie for account %v: %v", hex.EncodeToString(hash), addr, err)
+			} else {
+				if !deleted {
+					au.log.Warnf("upgradeDatabaseSchema4: failed to delete hash '%s' from merkle trie for account %v", hex.EncodeToString(hash), addr)
+				} else {
+					totalHashesDeleted++
+				}
+			}
+		}
+
+		if _, err = trie.Commit(); err != nil {
+			au.log.Errorf("upgradeDatabaseSchema4: failed to commit changes to merkle trie: %v", err)
+		}
+
+		au.log.Infof("upgradeDatabaseSchema4: deleted %d hashes", totalHashesDeleted)
+	}
+
+done:
+	au.log.Infof("upgradeDatabaseSchema4: deleted %d rows", numDeleted)
+
+	// update version
+	_, err = db.SetUserVersion(ctx, tx, 5)
+	if err != nil {
+		return 0, fmt.Errorf("accountsInitialize unable to update database schema version from 4 to 5: %v", err)
+	}
+	return 5, nil
+}
+
 // deleteStoredCatchpoints iterates over the storedcatchpoints table and deletes all the files stored on disk.
 // once all the files have been deleted, it would go ahead and remove the entries from the table.
 func (au *accountUpdates) deleteStoredCatchpoints(ctx context.Context, dbQueries *accountsDbQueries) (err error) {