[Cadence 1.0] Improve migrations

cmd/util/cmd/execution-state-extract/cmd.go

@@ -42,6 +42,7 @@ var (
 	flagInputPayloadFileName          string
 	flagOutputPayloadFileName         string
 	flagOutputPayloadByAddresses      string
+	flagMaxAccountSize                uint64
 )
 
 var Cmd = &cobra.Command{
@@ -132,6 +133,9 @@ func init() {
 		"",
 		"extract payloads of addresses (comma separated hex-encoded addresses) to file specified by output-payload-filename",
 	)
+
+	Cmd.Flags().Uint64Var(&flagMaxAccountSize, "max-account-size", 0,
+		"max account size")
 }
 
 func run(*cobra.Command, []string) {
@@ -346,6 +350,7 @@ func run(*cobra.Command, []string) {
 			exportedAddresses,
 			flagSortPayloads,
 			flagPrune,
+			flagMaxAccountSize,
 		)
 	} else {
 		err = extractExecutionState(
@@ -365,6 +370,7 @@ func run(*cobra.Command, []string) {
 			exportedAddresses,
 			flagSortPayloads,
 			flagPrune,
+			flagMaxAccountSize,
 		)
 	}
 

cmd/util/cmd/execution-state-extract/execution_state_extract.go

@@ -15,7 +15,6 @@ import (
 	"github.com/onflow/flow-go/cmd/util/ledger/reporters"
 	"github.com/onflow/flow-go/cmd/util/ledger/util"
 	"github.com/onflow/flow-go/ledger"
-	"github.com/onflow/flow-go/ledger/common/convert"
 	"github.com/onflow/flow-go/ledger/common/hash"
 	"github.com/onflow/flow-go/ledger/common/pathfinder"
 	"github.com/onflow/flow-go/ledger/complete"
@@ -48,6 +47,7 @@ func extractExecutionState(
 	exportPayloadsByAddresses []common.Address,
 	sortPayloads bool,
 	prune bool,
+	maxAccountSize uint64,
 ) error {
 
 	log.Info().Msg("init WAL")
@@ -119,6 +119,7 @@ func extractExecutionState(
 		burnerContractChange,
 		stagedContracts,
 		prune,
+		maxAccountSize,
 	)
 
 	newState := ledger.State(targetHash)
@@ -220,20 +221,6 @@ func writeStatusFile(fileName string, e error) error {
 	return err
 }
 
-func ByteCountIEC(b int64) string {
-	const unit = 1024
-	if b < unit {
-		return fmt.Sprintf("%d B", b)
-	}
-	div, exp := int64(unit), 0
-	for n := b / unit; n >= unit; n /= unit {
-		div *= unit
-		exp++
-	}
-	return fmt.Sprintf("%.1f %ciB",
-		float64(b)/float64(div), "KMGTPE"[exp])
-}
-
 func extractExecutionStateFromPayloads(
 	log zerolog.Logger,
 	dir string,
@@ -251,6 +238,7 @@ func extractExecutionStateFromPayloads(
 	exportPayloadsByAddresses []common.Address,
 	sortPayloads bool,
 	prune bool,
+	maxAccountSize uint64,
 ) error {
 
 	inputPayloadsFromPartialState, payloads, err := util.ReadPayloadFile(log, inputPayloadFile)
@@ -260,36 +248,6 @@ func extractExecutionStateFromPayloads(
 
 	log.Info().Msgf("read %d payloads", len(payloads))
 
-	if log.Debug().Enabled() {
-
-		type accountInfo struct {
-			count int
-			size  uint64
-		}
-		payloadCountByAddress := make(map[string]accountInfo)
-
-		for _, payload := range payloads {
-			registerID, payloadValue, err := convert.PayloadToRegister(payload)
-			if err != nil {
-				return fmt.Errorf("cannot convert payload to register: %w", err)
-			}
-			owner := registerID.Owner
-			accountInfo := payloadCountByAddress[owner]
-			accountInfo.count++
-			accountInfo.size += uint64(len(payloadValue))
-			payloadCountByAddress[owner] = accountInfo
-		}
-
-		for address, info := range payloadCountByAddress {
-			log.Debug().Msgf(
-				"address %x has %d payloads and a total size of %s",
-				address,
-				info.count,
-				ByteCountIEC(int64(info.size)),
-			)
-		}
-	}
-
 	migrations := newMigrations(
 		log,
 		dir,
@@ -302,6 +260,7 @@ func extractExecutionStateFromPayloads(
 		burnerContractChange,
 		stagedContracts,
 		prune,
+		maxAccountSize,
 	)
 
 	payloads, err = migratePayloads(log, payloads, migrations)
@@ -393,7 +352,7 @@ func migratePayloads(logger zerolog.Logger, payloads []*ledger.Payload, migratio
 
 	// migrate payloads
 	for i, migrate := range migrations {
-		logger.Info().Msgf("migration %d/%d is underway", i, len(migrations))
+		logger.Info().Msgf("migration %d/%d is underway", i+1, len(migrations))
 
 		start := time.Now()
 		payloads, err = migrate(payloads)
@@ -458,6 +417,7 @@ func newMigrations(
 	burnerContractChange migrators.BurnerContractChange,
 	stagedContracts []migrators.StagedContract,
 	prune bool,
+	maxAccountSize uint64,
 ) []ledger.Migration {
 	if !runMigrations {
 		return nil
@@ -478,6 +438,7 @@ func newMigrations(
 		burnerContractChange,
 		stagedContracts,
 		prune,
+		maxAccountSize,
 	)
 
 	migrations := make([]ledger.Migration, 0, len(namedMigrations))

cmd/util/ledger/migrations/account_based_migration.go

@@ -127,7 +127,7 @@ func withMigrations(
 				Type("migration", migrator).
 				Msg("closing migration")
 			if cerr := migrator.Close(); cerr != nil {
-				log.Error().Err(cerr).Msg("error closing migration")
+				log.Err(cerr).Msg("error closing migration")
 				if err == nil {
 					// only set the error if it's not already set
 					// so that we don't overwrite the original error
@@ -315,19 +315,7 @@ func MigrateGroupConcurrently(
 	return migrated, nil
 }
 
-var knownProblematicAccounts = map[common.Address]string{
-	// Testnet accounts with broken contracts
-	mustHexToAddress("434a1f199a7ae3ba"): "Broken contract FanTopPermission",
-	mustHexToAddress("454c9991c2b8d947"): "Broken contract Test",
-	mustHexToAddress("48602d8056ff9d93"): "Broken contract FanTopPermission",
-	mustHexToAddress("5d63c34d7f05e5a4"): "Broken contract FanTopPermission",
-	mustHexToAddress("5e3448b3cffb97f2"): "Broken contract FanTopPermission",
-	mustHexToAddress("7d8c7e050c694eaa"): "Broken contract Test",
-	mustHexToAddress("ba53f16ede01972d"): "Broken contract FanTopPermission",
-	mustHexToAddress("c843c1f5a4805c3a"): "Broken contract FanTopPermission",
-	mustHexToAddress("48d3be92e6e4a973"): "Broken contract FanTopPermission",
-	// Mainnet account
-}
+var knownProblematicAccounts = map[common.Address]string{}
 
 func mustHexToAddress(hex string) common.Address {
 	address, err := common.HexToAddress(hex)

cmd/util/ledger/migrations/account_size_filter_migration.go

@@ -0,0 +1,99 @@
+package migrations
+
+import (
+	"fmt"
+
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/convert"
+)
+
+func NewAccountSizeFilterMigration(
+	maxAccountSize uint64,
+	exceptions map[string]struct{},
+	log zerolog.Logger,
+) ledger.Migration {
+
+	if maxAccountSize == 0 {
+		return nil
+	}
+
+	return func(payloads []*ledger.Payload) ([]*ledger.Payload, error) {
+
+		type accountInfo struct {
+			count int
+			size  uint64
+		}
+		payloadCountByAddress := make(map[string]accountInfo)
+
+		for _, payload := range payloads {
+			registerID, payloadValue, err := convert.PayloadToRegister(payload)
+			if err != nil {
+				return nil, fmt.Errorf("cannot convert payload to register: %w", err)
+			}
+
+			owner := registerID.Owner
+			accountInfo := payloadCountByAddress[owner]
+			accountInfo.count++
+			accountInfo.size += uint64(len(payloadValue))
+			payloadCountByAddress[owner] = accountInfo
+		}
+
+		if log.Debug().Enabled() {
+			for address, info := range payloadCountByAddress {
+				log.Debug().Msgf(
+					"address %x has %d payloads and a total size of %s",
+					address,
+					info.count,
+					ByteCountIEC(int64(info.size)),
+				)
+
+				if _, ok := exceptions[address]; !ok && info.size > maxAccountSize {
+					log.Warn().Msgf(
+						"dropping payloads of account %x. size of payloads %s exceeds max size %s",
+						address,
+						ByteCountIEC(int64(info.size)),
+						ByteCountIEC(int64(maxAccountSize)),
+					)
+				}
+			}
+		}
+
+		filteredPayloads := make([]*ledger.Payload, 0, int(0.8*float32(len(payloads))))
+
+		for _, payload := range payloads {
+			registerID, _, err := convert.PayloadToRegister(payload)
+			if err != nil {
+				return nil, fmt.Errorf("cannot convert payload to register: %w", err)
+			}
+
+			owner := registerID.Owner
+
+			if _, ok := exceptions[owner]; !ok {
+				info := payloadCountByAddress[owner]
+				if info.size > maxAccountSize {
+					continue
+				}
+			}
+
+			filteredPayloads = append(filteredPayloads, payload)
+		}
+
+		return filteredPayloads, nil
+	}
+}
+
+func ByteCountIEC(b int64) string {
+	const unit = 1024
+	if b < unit {
+		return fmt.Sprintf("%d B", b)
+	}
+	div, exp := int64(unit), 0
+	for n := b / unit; n >= unit; n /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f %ciB",
+		float64(b)/float64(div), "KMGTPE"[exp])
+}

cmd/util/ledger/migrations/cadence.go

@@ -176,12 +176,29 @@ func NewCadence1ValueMigrations(
 
 	errorMessageHandler := &errorMessageHandler{}
 
+	// The value migrations are run as account-based migrations,
+	// i.e. the migrations are only given the payloads for the account to be migrated.
+	// However, the migrations need to be able to get the code for contracts of any account.
+	//
+	// To achieve this, the contracts are extracted from the payloads once,
+	// before the value migrations are run.
+
+	contracts := make(map[common.AddressLocation][]byte, 1000)
+
+	migrations = []NamedMigration{
+		{
+			Name:    "contracts",
+			Migrate: NewContractsExtractionMigration(contracts, log),
+		},
+	}
+
 	for _, accountBasedMigration := range []*CadenceBaseMigrator{
 		NewCadence1ValueMigrator(
 			rwf,
 			diffMigrations,
 			logVerboseDiff,
 			errorMessageHandler,
+			contracts,
 			NewCadence1CompositeStaticTypeConverter(chainID),
 			NewCadence1InterfaceStaticTypeConverter(chainID),
 		),
@@ -190,13 +207,15 @@ func NewCadence1ValueMigrations(
 			diffMigrations,
 			logVerboseDiff,
 			errorMessageHandler,
+			contracts,
 			capabilityMapping,
 		),
 		NewCadence1CapabilityValueMigrator(
 			rwf,
 			diffMigrations,
 			logVerboseDiff,
 			errorMessageHandler,
+			contracts,
 			capabilityMapping,
 		),
 	} {
@@ -288,10 +307,29 @@ func NewCadence1Migrations(
 	burnerContractChange BurnerContractChange,
 	stagedContracts []StagedContract,
 	prune bool,
+	maxAccountSize uint64,
 ) []NamedMigration {
 
 	var migrations []NamedMigration
 
+	if maxAccountSize > 0 {
+
+		maxSizeExceptions := map[string]struct{}{}
+
+		systemContracts := systemcontracts.SystemContractsForChain(chainID)
+		for _, systemContract := range systemContracts.All() {
+			maxSizeExceptions[string(systemContract.Address.Bytes())] = struct{}{}
+		}
+
+		migrations = append(
+			migrations,
+			NamedMigration{
+				Name:    "account-size-filter-migration",
+				Migrate: NewAccountSizeFilterMigration(maxAccountSize, maxSizeExceptions, log),
+			},
+		)
+	}
+
 	if prune {
 		migration := NewCadence1PruneMigration(chainID, log)
 		if migration != nil {