catchpoint: actualize limit constants and fix txtail table init

catchup/ledgerFetcher.go

@@ -39,8 +39,8 @@ import (
 var errNoLedgerForRound = errors.New("no ledger available for given round")
 
 const (
-	// maxCatchpointFileChunkSize is a rough estimate for the worst-case scenario we're going to have of all the accounts data per a single catchpoint file chunk.
-	maxCatchpointFileChunkSize = ledger.BalancesPerCatchpointFileChunk * basics.MaxEncodedAccountDataSize
+	// maxCatchpointFileChunkSize is a rough estimate for the worst-case scenario we're going to have of all the accounts data per a single catchpoint file chunk and one account with max resources.
+	maxCatchpointFileChunkSize = ledger.BalancesPerCatchpointFileChunk*(ledger.MaxEncodedBaseAccountDataSize+ledger.MaxEncodedKVDataSize) + ledger.ResourcesPerCatchpointFileChunk*ledger.MaxEncodedBaseResourceDataSize
 	// defaultMinCatchpointFileDownloadBytesPerSecond defines the worst-case scenario download speed we expect to get while downloading a catchpoint file
 	defaultMinCatchpointFileDownloadBytesPerSecond = 20 * 1024
 	// catchpointFileStreamReadSize defines the number of bytes we would attempt to read at each iteration from the incoming http data stream

data/basics/userBalance.go

@@ -42,20 +42,16 @@ const (
 	// These two accounts also have additional Algo transfer restrictions.
 	NotParticipating
 
-	// MaxEncodedAccountDataSize is a rough estimate for the worst-case scenario we're going to have of the account data and address serialized.
-	// this number is verified by the TestEncodedAccountDataSize function.
-	MaxEncodedAccountDataSize = 850000
-
 	// encodedMaxAssetsPerAccount is the decoder limit of number of assets stored per account.
 	// it's being verified by the unit test TestEncodedAccountAllocationBounds to align
 	// with config.Consensus[protocol.ConsensusCurrentVersion].MaxAssetsPerAccount; note that the decoded
 	// parameter is used only for protecting the decoder against malicious encoded account data stream.
-	// protocol-specific constains would be tested once the decoding is complete.
+	// protocol-specific contains would be tested once the decoding is complete.
 	encodedMaxAssetsPerAccount = 1024
 
 	// EncodedMaxAppLocalStates is the decoder limit for number of opted-in apps in a single account.
 	// It is verified in TestEncodedAccountAllocationBounds to align with
-	// config.Consensus[protocol.ConsensusCurrentVersion].MaxppsOptedIn
+	// config.Consensus[protocol.ConsensusCurrentVersion].MaxAppsOptedIn
 	EncodedMaxAppLocalStates = 64
 
 	// EncodedMaxAppParams is the decoder limit for number of created apps in a single account.

data/basics/userBalance_test.go

@@ -131,102 +131,6 @@ func getSampleAccountData() AccountData {
 	}
 }
 
-func TestEncodedAccountDataSize(t *testing.T) {
-	partitiontest.PartitionTest(t)
-
-	maxStateSchema := StateSchema{
-		NumUint:      0x1234123412341234,
-		NumByteSlice: 0x1234123412341234,
-	}
-	ad := getSampleAccountData()
-	ad.TotalAppSchema = maxStateSchema
-
-	// TODO after applications enabled: change back to protocol.ConsensusCurrentVersion
-	currentConsensusParams := config.Consensus[protocol.ConsensusFuture]
-
-	for assetCreatorAssets := 0; assetCreatorAssets < currentConsensusParams.MaxAssetsPerAccount; assetCreatorAssets++ {
-		ap := AssetParams{
-			Total:         0x1234123412341234,
-			Decimals:      0x12341234,
-			DefaultFrozen: true,
-			UnitName:      makeString(currentConsensusParams.MaxAssetUnitNameBytes),
-			AssetName:     makeString(currentConsensusParams.MaxAssetNameBytes),
-			URL:           makeString(currentConsensusParams.MaxAssetURLBytes),
-			Manager:       Address(crypto.Hash([]byte{1, byte(assetCreatorAssets)})),
-			Reserve:       Address(crypto.Hash([]byte{2, byte(assetCreatorAssets)})),
-			Freeze:        Address(crypto.Hash([]byte{3, byte(assetCreatorAssets)})),
-			Clawback:      Address(crypto.Hash([]byte{4, byte(assetCreatorAssets)})),
-		}
-		copy(ap.MetadataHash[:], makeString(32))
-		ad.AssetParams[AssetIndex(0x1234123412341234-assetCreatorAssets)] = ap
-	}
-
-	for assetHolderAssets := 0; assetHolderAssets < currentConsensusParams.MaxAssetsPerAccount; assetHolderAssets++ {
-		ah := AssetHolding{
-			Amount: 0x1234123412341234,
-			Frozen: true,
-		}
-		ad.Assets[AssetIndex(0x1234123412341234-assetHolderAssets)] = ah
-	}
-
-	maxProg := []byte(makeString(config.MaxAvailableAppProgramLen))
-	maxGlobalState := make(TealKeyValue, currentConsensusParams.MaxGlobalSchemaEntries)
-	maxLocalState := make(TealKeyValue, currentConsensusParams.MaxLocalSchemaEntries)
-
-	for globalKey := uint64(0); globalKey < currentConsensusParams.MaxGlobalSchemaEntries; globalKey++ {
-		prefix := fmt.Sprintf("%d|", globalKey)
-		padding := makeString(currentConsensusParams.MaxAppKeyLen - len(prefix))
-		maxKey := prefix + padding
-		maxValue := TealValue{
-			Type:  TealBytesType,
-			Bytes: makeString(currentConsensusParams.MaxAppSumKeyValueLens - len(maxKey)),
-		}
-		maxGlobalState[maxKey] = maxValue
-	}
-
-	for localKey := uint64(0); localKey < currentConsensusParams.MaxLocalSchemaEntries; localKey++ {
-		prefix := fmt.Sprintf("%d|", localKey)
-		padding := makeString(currentConsensusParams.MaxAppKeyLen - len(prefix))
-		maxKey := prefix + padding
-		maxValue := TealValue{
-			Type:  TealBytesType,
-			Bytes: makeString(currentConsensusParams.MaxAppSumKeyValueLens - len(maxKey)),
-		}
-		maxLocalState[maxKey] = maxValue
-	}
-	maxAppsCreate := currentConsensusParams.MaxAppsCreated
-	if maxAppsCreate == 0 {
-		maxAppsCreate = config.Consensus[protocol.ConsensusV30].MaxAppsCreated
-	}
-	for appCreatorApps := 0; appCreatorApps < maxAppsCreate; appCreatorApps++ {
-		ap := AppParams{
-			ApprovalProgram:   maxProg,
-			ClearStateProgram: maxProg,
-			GlobalState:       maxGlobalState,
-			StateSchemas: StateSchemas{
-				LocalStateSchema:  maxStateSchema,
-				GlobalStateSchema: maxStateSchema,
-			},
-		}
-		ad.AppParams[AppIndex(0x1234123412341234-appCreatorApps)] = ap
-	}
-
-	maxAppsOptedIn := currentConsensusParams.MaxAppsOptedIn
-	if maxAppsOptedIn == 0 {
-		maxAppsOptedIn = config.Consensus[protocol.ConsensusV30].MaxAppsOptedIn
-	}
-	for appHolderApps := 0; appHolderApps < maxAppsOptedIn; appHolderApps++ {
-		ls := AppLocalState{
-			KeyValue: maxLocalState,
-			Schema:   maxStateSchema,
-		}
-		ad.AppLocalStates[AppIndex(0x1234123412341234-appHolderApps)] = ls
-	}
-
-	encoded := ad.MarshalMsg(nil)
-	require.GreaterOrEqual(t, MaxEncodedAccountDataSize, len(encoded))
-}
-
 func TestEncodedAccountAllocationBounds(t *testing.T) {
 	partitiontest.PartitionTest(t)
 

ledger/accountdb.go

@@ -23,6 +23,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"math"
 	"strings"
 	"time"
 
@@ -372,6 +373,14 @@ const (
 	catchpointStateCatchpointLookback = catchpointState("catchpointLookback")
 )
 
+// MaxEncodedBaseAccountDataSize is a rough estimate for the worst-case scenario we're going to have of the base account data serialized.
+// this number is verified by the TestEncodedBaseAccountDataSize function.
+const MaxEncodedBaseAccountDataSize = 350
+
+// MaxEncodedBaseResourceDataSize is a rough estimate for the worst-case scenario we're going to have of the base resource data serialized.
+// this number is verified by the TestEncodedBaseResourceSize function.
+const MaxEncodedBaseResourceDataSize = 20000
+
 // normalizedAccountBalance is a staging area for a catchpoint file account information before it's being added to the catchpoint staging tables.
 type normalizedAccountBalance struct {
 	// The public key address to which the account belongs.
@@ -2250,13 +2259,24 @@ func performTxTailTableMigration(ctx context.Context, tx *sql.Tx, blockDb db.Acc
 			return fmt.Errorf("latest block header %d cannot be retrieved : %w", dbRound, err)
 		}
 
-		maxTxnLife := basics.Round(config.Consensus[latestHdr.CurrentProtocol].MaxTxnLife)
-		deeperBlockHistory := basics.Round(config.Consensus[latestHdr.CurrentProtocol].DeeperBlockHeaderHistory)
-		firstRound := (latestBlockRound + 1).SubSaturate(maxTxnLife + deeperBlockHistory)
+		proto := config.Consensus[latestHdr.CurrentProtocol]
+		maxTxnLife := basics.Round(proto.MaxTxnLife)
+		deeperBlockHistory := basics.Round(proto.DeeperBlockHeaderHistory)
+		// firstRound is either maxTxnLife + deeperBlockHistory back from the latest for regular init
+		// or maxTxnLife + deeperBlockHistory + CatchpointLookback back for catchpoint apply.
+		// Try to check the earliest available and start from there.
+		firstRound := (latestBlockRound + 1).SubSaturate(maxTxnLife + deeperBlockHistory + basics.Round(proto.CatchpointLookback))
 		// we don't need to have the txtail for round 0.
 		if firstRound == basics.Round(0) {
 			firstRound++
 		}
+		if _, err := blockGet(blockTx, firstRound); err != nil {
+			// looks like not catchpoint but a regular migration, start from maxTxnLife + deeperBlockHistory back
+			firstRound = (latestBlockRound + 1).SubSaturate(maxTxnLife + deeperBlockHistory)
+			if firstRound == basics.Round(0) {
+				firstRound++
+			}
+		}
 		tailRounds := make([][]byte, 0, maxTxnLife)
 		for rnd := firstRound; rnd <= dbRound; rnd++ {
 			blk, err := blockGet(blockTx, rnd)
@@ -4403,18 +4423,16 @@ type orderedAccountsIter struct {
 	pendingBaseRow     pendingBaseRow
 	pendingResourceRow pendingResourceRow
 	accountCount       int
-	resourceCount      int
 	insertStmt         *sql.Stmt
 }
 
 // makeOrderedAccountsIter creates an ordered account iterator. Note that due to implementation reasons,
 // only a single iterator can be active at a time.
-func makeOrderedAccountsIter(tx *sql.Tx, accountCount int, resourceCount int) *orderedAccountsIter {
+func makeOrderedAccountsIter(tx *sql.Tx, accountCount int) *orderedAccountsIter {
 	return &orderedAccountsIter{
-		tx:            tx,
-		accountCount:  accountCount,
-		resourceCount: resourceCount,
-		step:          oaiStepStartup,
+		tx:           tx,
+		accountCount: accountCount,
+		step:         oaiStepStartup,
 	}
 }
 
@@ -4812,7 +4830,7 @@ func (iterator *orderedAccountsIter) Next(ctx context.Context) (acct []accountAd
 		count, iterator.pendingBaseRow, iterator.pendingResourceRow, err = processAllBaseAccountRecords(
 			iterator.accountBaseRows, iterator.resourcesRows,
 			baseCb, resCb,
-			iterator.pendingBaseRow, iterator.pendingResourceRow, iterator.accountCount, iterator.resourceCount,
+			iterator.pendingBaseRow, iterator.pendingResourceRow, iterator.accountCount, math.MaxInt,
 		)
 		if err != nil {
 			iterator.Close(ctx)

ledger/accountdb_test.go

@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 	"math/rand"
 	"os"
 	"reflect"
@@ -4589,3 +4590,122 @@ func TestRemoveOfflineStateProofID(t *testing.T) {
 		}
 	}
 }
+
+func TestEncodedBaseAccountDataSize(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	t.Parallel()
+
+	vd := baseVotingData{
+		VoteFirstValid:  basics.Round(crypto.RandUint64()),
+		VoteLastValid:   basics.Round(crypto.RandUint64()),
+		VoteKeyDilution: crypto.RandUint64(),
+	}
+	crypto.RandBytes(vd.VoteID[:])
+	crypto.RandBytes(vd.StateProofID[:])
+	crypto.RandBytes(vd.SelectionID[:])
+
+	baseAD := baseAccountData{
+		Status:                     basics.Online,
+		MicroAlgos:                 basics.MicroAlgos{Raw: crypto.RandUint64()},
+		RewardsBase:                crypto.RandUint64(),
+		RewardedMicroAlgos:         basics.MicroAlgos{Raw: crypto.RandUint64()},
+		AuthAddr:                   ledgertesting.RandomAddress(),
+		TotalAppSchemaNumUint:      crypto.RandUint64(),
+		TotalAppSchemaNumByteSlice: crypto.RandUint64(),
+		TotalExtraAppPages:         uint32(crypto.RandUint63() % uint64(math.MaxUint32)),
+		TotalAssetParams:           crypto.RandUint64(),
+		TotalAssets:                crypto.RandUint64(),
+		TotalAppParams:             crypto.RandUint64(),
+		TotalAppLocalStates:        crypto.RandUint64(),
+		baseVotingData:             vd,
+		UpdateRound:                crypto.RandUint64(),
+	}
+
+	encoded := baseAD.MarshalMsg(nil)
+	require.GreaterOrEqual(t, MaxEncodedBaseAccountDataSize, len(encoded))
+}
+
+func TestEncodedBaseResourceSize(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	t.Parallel()
+
+	currentConsensusParams := config.Consensus[protocol.ConsensusCurrentVersion]
+
+	// resourcesData is suiteable for keeping asset params, holding, app params, app local state
+	// but only asset + holding or app + local state can appear there
+	rdAsset := resourcesData{
+		Total:         crypto.RandUint64(),
+		Decimals:      uint32(crypto.RandUint63() % uint64(math.MaxUint32)),
+		DefaultFrozen: true,
+		// MetadataHash
+		UnitName:  makeString(currentConsensusParams.MaxAssetUnitNameBytes),
+		AssetName: makeString(currentConsensusParams.MaxAssetNameBytes),
+		URL:       makeString(currentConsensusParams.MaxAssetURLBytes),
+		Manager:   ledgertesting.RandomAddress(),
+		Reserve:   ledgertesting.RandomAddress(),
+		Freeze:    ledgertesting.RandomAddress(),
+		Clawback:  ledgertesting.RandomAddress(),
+
+		Amount: crypto.RandUint64(),
+		Frozen: true,
+	}
+	crypto.RandBytes(rdAsset.MetadataHash[:])
+
+	rdApp := resourcesData{
+
+		SchemaNumUint:      crypto.RandUint64(),
+		SchemaNumByteSlice: crypto.RandUint64(),
+		// KeyValue
+
+		// ApprovalProgram
+		// ClearStateProgram
+		// GlobalState
+		LocalStateSchemaNumUint:       crypto.RandUint64(),
+		LocalStateSchemaNumByteSlice:  crypto.RandUint64(),
+		GlobalStateSchemaNumUint:      crypto.RandUint64(),
+		GlobalStateSchemaNumByteSlice: crypto.RandUint64(),
+		ExtraProgramPages:             uint32(crypto.RandUint63() % uint64(math.MaxUint32)),
+
+		ResourceFlags: 255,
+		UpdateRound:   crypto.RandUint64(),
+	}
+
+	// MaxAvailableAppProgramLen is conbined size of approval and clear state since it is bound by proto.MaxAppTotalProgramLen
+	rdApp.ApprovalProgram = make([]byte, config.MaxAvailableAppProgramLen/2)
+	crypto.RandBytes(rdApp.ApprovalProgram)
+	rdApp.ClearStateProgram = make([]byte, config.MaxAvailableAppProgramLen/2)
+	crypto.RandBytes(rdApp.ClearStateProgram)
+
+	maxGlobalState := make(basics.TealKeyValue, currentConsensusParams.MaxGlobalSchemaEntries)
+	for globalKey := uint64(0); globalKey < currentConsensusParams.MaxGlobalSchemaEntries; globalKey++ {
+		prefix := fmt.Sprintf("%d|", globalKey)
+		padding := makeString(currentConsensusParams.MaxAppKeyLen - len(prefix))
+		maxKey := prefix + padding
+		maxValue := basics.TealValue{
+			Type:  basics.TealBytesType,
+			Bytes: makeString(currentConsensusParams.MaxAppSumKeyValueLens - len(maxKey)),
+		}
+		maxGlobalState[maxKey] = maxValue
+	}
+
+	maxLocalState := make(basics.TealKeyValue, currentConsensusParams.MaxLocalSchemaEntries)
+	for localKey := uint64(0); localKey < currentConsensusParams.MaxLocalSchemaEntries; localKey++ {
+		prefix := fmt.Sprintf("%d|", localKey)
+		padding := makeString(currentConsensusParams.MaxAppKeyLen - len(prefix))
+		maxKey := prefix + padding
+		maxValue := basics.TealValue{
+			Type:  basics.TealBytesType,
+			Bytes: makeString(currentConsensusParams.MaxAppSumKeyValueLens - len(maxKey)),
+		}
+		maxLocalState[maxKey] = maxValue
+	}
+
+	rdApp.GlobalState = maxGlobalState
+	rdApp.KeyValue = maxLocalState
+
+	encodedAsset := rdAsset.MarshalMsg(nil)
+	encodedApp := rdApp.MarshalMsg(nil)
+
+	require.Less(t, len(encodedAsset), len(encodedApp))
+	require.GreaterOrEqual(t, MaxEncodedBaseResourceDataSize, len(encodedApp))
+}

ledger/catchpointtracker.go

@@ -1108,7 +1108,7 @@ func (ct *catchpointTracker) generateCatchpointData(ctx context.Context, account
 	start := time.Now()
 	ledgerGeneratecatchpointCount.Inc(nil)
 	err := ct.dbs.Rdb.Atomic(func(dbCtx context.Context, tx *sql.Tx) (err error) {
-		catchpointWriter, err = makeCatchpointWriter(ctx, catchpointDataFilePath, tx, DefaultMaxResourcesPerChunk)
+		catchpointWriter, err = makeCatchpointWriter(ctx, catchpointDataFilePath, tx, ResourcesPerCatchpointFileChunk)
 		if err != nil {
 			return
 		}
@@ -1527,7 +1527,7 @@ func (ct *catchpointTracker) accountsInitializeHashes(ctx context.Context, tx *s
 
 	if rootHash.IsZero() {
 		ct.log.Infof("accountsInitialize rebuilding merkle trie for round %d", rnd)
-		accountBuilderIt := makeOrderedAccountsIter(tx, trieRebuildAccountChunkSize, DefaultMaxResourcesPerChunk)
+		accountBuilderIt := makeOrderedAccountsIter(tx, trieRebuildAccountChunkSize)
 		defer accountBuilderIt.Close(ctx)
 		startTrieBuildTime := time.Now()
 		trieHashCount := 0
@@ -1552,7 +1552,7 @@ func (ct *catchpointTracker) accountsInitializeHashes(ctx context.Context, tx *s
 						return fmt.Errorf("accountsInitialize was unable to add changes to trie: %v", err)
 					}
 					if !added {
-						// we need to transalate the "addrid" into actual account address so that
+						// we need to translate the "addrid" into actual account address so that
 						// we can report the failure.
 						addr, err := lookupAccountAddressFromAddressID(ctx, tx, acct.addrid)
 						if err != nil {