Skip to content

Commit 16503c2

Browse files
peterarguepeterargue
authored
Merge branch 'master' into petera/make-peer-cache-config
2 parents 7b53766 + 8fbbd56 commit 16503c2

File tree

54 files changed

+636
-563
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

54 files changed

+636
-563
lines changed

‎cmd/access/node_builder/access_node_builder.go‎

Lines changed: 32 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,10 @@ import (
117117
"github.com/onflow/flow-go/state/protocol/blocktimer"
118118
"github.com/onflow/flow-go/storage"
119119
bstorage "github.com/onflow/flow-go/storage/badger"
120+
"github.com/onflow/flow-go/storage/operation/badgerimpl"
121+
"github.com/onflow/flow-go/storage/operation/pebbleimpl"
120122
pstorage "github.com/onflow/flow-go/storage/pebble"
123+
"github.com/onflow/flow-go/storage/store"
121124
"github.com/onflow/flow-go/utils/grpcutils"
122125
)
123126

@@ -552,8 +555,8 @@ func (builder *FlowAccessNodeBuilder) BuildConsensusFollower() *FlowAccessNodeBu
552555
func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccessNodeBuilder {
553556
var ds datastore.Batching
554557
var bs network.BlobService
555-
var processedBlockHeight storage.ConsumerProgress
556-
var processedNotifications storage.ConsumerProgress
558+
var processedBlockHeight storage.ConsumerProgressInitializer
559+
var processedNotifications storage.ConsumerProgressInitializer
557560
var bsDependable *module.ProxiedReadyDoneAware
558561
var execDataDistributor *edrequester.ExecutionDataDistributor
559562
var execDataCacheBackend *herocache.BlockExecutionData
@@ -607,21 +610,30 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
607610
Module("processed block height consumer progress", func(node *cmd.NodeConfig) error {
608611
// Note: progress is stored in the datastore's DB since that is where the jobqueue
609612
// writes execution data to.
610-
if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
611-
processedBlockHeight = bstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*badger.DB), module.ConsumeProgressExecutionDataRequesterBlockHeight)
613+
var db storage.DB
614+
edmdb := builder.ExecutionDatastoreManager.DB()
615+
616+
if bdb, ok := edmdb.(*badger.DB); ok {
617+
db = badgerimpl.ToDB(bdb)
618+
} else if pdb, ok := edmdb.(*pebble.DB); ok {
619+
db = pebbleimpl.ToDB(pdb)
612620
} else {
613-
processedBlockHeight = pstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*pebble.DB), module.ConsumeProgressExecutionDataRequesterBlockHeight)
621+
return fmt.Errorf("unsupported execution data DB type: %T", edmdb)
614622
}
623+
624+
processedBlockHeight = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterBlockHeight)
615625
return nil
616626
}).
617627
Module("processed notifications consumer progress", func(node *cmd.NodeConfig) error {
618628
// Note: progress is stored in the datastore's DB since that is where the jobqueue
619629
// writes execution data to.
630+
var db storage.DB
620631
if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
621-
processedNotifications = bstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*badger.DB), module.ConsumeProgressExecutionDataRequesterNotification)
632+
db = badgerimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*badger.DB))
622633
} else {
623-
processedNotifications = pstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*pebble.DB), module.ConsumeProgressExecutionDataRequesterNotification)
634+
db = pebbleimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*pebble.DB))
624635
}
636+
processedNotifications = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterNotification)
625637
return nil
626638
}).
627639
Module("blobservice peer manager dependencies", func(node *cmd.NodeConfig) error {
@@ -848,15 +860,15 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
848860
}
849861

850862
if builder.executionDataIndexingEnabled {
851-
var indexedBlockHeight storage.ConsumerProgress
863+
var indexedBlockHeight storage.ConsumerProgressInitializer
852864

853865
builder.
854866
AdminCommand("execute-script", func(config *cmd.NodeConfig) commands.AdminCommand {
855867
return stateSyncCommands.NewExecuteScriptCommand(builder.ScriptExecutor)
856868
}).
857869
Module("indexed block height consumer progress", func(node *cmd.NodeConfig) error {
858870
// Note: progress is stored in the MAIN db since that is where indexed execution data is stored.
859-
indexedBlockHeight = bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressExecutionDataIndexerBlockHeight)
871+
indexedBlockHeight = store.NewConsumerProgress(badgerimpl.ToDB(builder.DB), module.ConsumeProgressExecutionDataIndexerBlockHeight)
860872
return nil
861873
}).
862874
Module("transaction results storage", func(node *cmd.NodeConfig) error {
@@ -1633,8 +1645,8 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
16331645
}
16341646

16351647
func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
1636-
var processedFinalizedBlockHeight storage.ConsumerProgress
1637-
var processedTxErrorMessagesBlockHeight storage.ConsumerProgress
1648+
var processedFinalizedBlockHeight storage.ConsumerProgressInitializer
1649+
var processedTxErrorMessagesBlockHeight storage.ConsumerProgressInitializer
16381650

16391651
if builder.executionDataSyncEnabled {
16401652
builder.BuildExecutionSyncComponents()
@@ -1838,17 +1850,18 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
18381850
return nil
18391851
}).
18401852
Module("processed finalized block height consumer progress", func(node *cmd.NodeConfig) error {
1841-
processedFinalizedBlockHeight = bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressIngestionEngineBlockHeight)
1853+
processedFinalizedBlockHeight = store.NewConsumerProgress(badgerimpl.ToDB(builder.DB), module.ConsumeProgressIngestionEngineBlockHeight)
18421854
return nil
18431855
}).
18441856
Module("processed last full block height monotonic consumer progress", func(node *cmd.NodeConfig) error {
18451857
rootBlockHeight := node.State.Params().FinalizedRoot().Height
18461858

1847-
var err error
1848-
lastFullBlockHeight, err = counters.NewPersistentStrictMonotonicCounter(
1849-
bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressLastFullBlockHeight),
1850-
rootBlockHeight,
1851-
)
1859+
progress, err := store.NewConsumerProgress(badgerimpl.ToDB(builder.DB), module.ConsumeProgressLastFullBlockHeight).Initialize(rootBlockHeight)
1860+
if err != nil {
1861+
return err
1862+
}
1863+
1864+
lastFullBlockHeight, err = counters.NewPersistentStrictMonotonicCounter(progress)
18521865
if err != nil {
18531866
return fmt.Errorf("failed to initialize monotonic consumer progress: %w", err)
18541867
}
@@ -2149,8 +2162,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
21492162

21502163
if builder.storeTxResultErrorMessages {
21512164
builder.Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
2152-
processedTxErrorMessagesBlockHeight = bstorage.NewConsumerProgress(
2153-
builder.DB,
2165+
processedTxErrorMessagesBlockHeight = store.NewConsumerProgress(
2166+
badgerimpl.ToDB(builder.DB),
21542167
module.ConsumeProgressEngineTxErrorMessagesBlockHeight,
21552168
)
21562169
return nil

‎cmd/observer/node_builder/observer_builder.go‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,10 @@ import (
108108
"github.com/onflow/flow-go/state/protocol/events/gadgets"
109109
"github.com/onflow/flow-go/storage"
110110
bstorage "github.com/onflow/flow-go/storage/badger"
111+
"github.com/onflow/flow-go/storage/operation/badgerimpl"
112+
"github.com/onflow/flow-go/storage/operation/pebbleimpl"
111113
pstorage "github.com/onflow/flow-go/storage/pebble"
114+
"github.com/onflow/flow-go/storage/store"
112115
"github.com/onflow/flow-go/utils/grpcutils"
113116
"github.com/onflow/flow-go/utils/io"
114117
)
@@ -1057,8 +1060,8 @@ func (builder *ObserverServiceBuilder) Build() (cmd.Node, error) {
10571060
func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverServiceBuilder {
10581061
var ds datastore.Batching
10591062
var bs network.BlobService
1060-
var processedBlockHeight storage.ConsumerProgress
1061-
var processedNotifications storage.ConsumerProgress
1063+
var processedBlockHeight storage.ConsumerProgressInitializer
1064+
var processedNotifications storage.ConsumerProgressInitializer
10621065
var publicBsDependable *module.ProxiedReadyDoneAware
10631066
var execDataDistributor *edrequester.ExecutionDataDistributor
10641067
var execDataCacheBackend *herocache.BlockExecutionData
@@ -1112,21 +1115,26 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
11121115
Module("processed block height consumer progress", func(node *cmd.NodeConfig) error {
11131116
// Note: progress is stored in the datastore's DB since that is where the jobqueue
11141117
// writes execution data to.
1118+
var db storage.DB
11151119
if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
1116-
processedBlockHeight = bstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*badger.DB), module.ConsumeProgressExecutionDataRequesterBlockHeight)
1120+
db = badgerimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*badger.DB))
11171121
} else {
1118-
processedBlockHeight = pstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*pebble.DB), module.ConsumeProgressExecutionDataRequesterBlockHeight)
1122+
db = pebbleimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*pebble.DB))
11191123
}
1124+
1125+
processedBlockHeight = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterBlockHeight)
11201126
return nil
11211127
}).
11221128
Module("processed notifications consumer progress", func(node *cmd.NodeConfig) error {
11231129
// Note: progress is stored in the datastore's DB since that is where the jobqueue
11241130
// writes execution data to.
1131+
var db storage.DB
11251132
if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
1126-
processedNotifications = bstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*badger.DB), module.ConsumeProgressExecutionDataRequesterNotification)
1133+
db = badgerimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*badger.DB))
11271134
} else {
1128-
processedNotifications = pstorage.NewConsumerProgress(builder.ExecutionDatastoreManager.DB().(*pebble.DB), module.ConsumeProgressExecutionDataRequesterNotification)
1135+
db = pebbleimpl.ToDB(builder.ExecutionDatastoreManager.DB().(*pebble.DB))
11291136
}
1137+
processedNotifications = store.NewConsumerProgress(db, module.ConsumeProgressExecutionDataRequesterNotification)
11301138
return nil
11311139
}).
11321140
Module("blobservice peer manager dependencies", func(node *cmd.NodeConfig) error {
@@ -1311,11 +1319,11 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
13111319
return builder.ExecutionDataPruner, nil
13121320
})
13131321
if builder.executionDataIndexingEnabled {
1314-
var indexedBlockHeight storage.ConsumerProgress
1322+
var indexedBlockHeight storage.ConsumerProgressInitializer
13151323

13161324
builder.Module("indexed block height consumer progress", func(node *cmd.NodeConfig) error {
13171325
// Note: progress is stored in the MAIN db since that is where indexed execution data is stored.
1318-
indexedBlockHeight = bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressExecutionDataIndexerBlockHeight)
1326+
indexedBlockHeight = store.NewConsumerProgress(badgerimpl.ToDB(builder.DB), module.ConsumeProgressExecutionDataIndexerBlockHeight)
13191327
return nil
13201328
}).Module("transaction results storage", func(node *cmd.NodeConfig) error {
13211329
builder.Storage.LightTransactionResults = bstorage.NewLightTransactionResults(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)

‎cmd/util/cmd/diff-states/cmd.go‎

Lines changed: 69 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111

1212
"github.com/dustin/go-humanize/english"
1313
"github.com/onflow/cadence/common"
14+
"github.com/onflow/cadence/interpreter"
1415
"github.com/rs/zerolog/log"
1516
"github.com/spf13/cobra"
1617
"golang.org/x/sync/errgroup"
@@ -26,17 +27,18 @@ import (
2627
)
2728

2829
var (
29-
flagOutputDirectory string
30-
flagPayloads1 string
31-
flagPayloads2 string
32-
flagState1 string
33-
flagState2 string
34-
flagStateCommitment1 string
35-
flagStateCommitment2 string
36-
flagMode string
37-
flagAlwaysDiffValues bool
38-
flagNWorker int
39-
flagChain string
30+
flagOutputDirectory string
31+
flagPayloads1 string
32+
flagPayloads2 string
33+
flagState1 string
34+
flagState2 string
35+
flagStateCommitment1 string
36+
flagStateCommitment2 string
37+
flagMode string
38+
flagAlwaysDiffValues bool
39+
flagExcludeRandomBeaconHistory bool
40+
flagNWorker int
41+
flagChain string
4042
)
4143

4244
var Cmd = &cobra.Command{
@@ -139,8 +141,20 @@ func init() {
139141
"Chain name",
140142
)
141143
_ = Cmd.MarkFlagRequired("chain")
144+
145+
Cmd.Flags().BoolVar(
146+
&flagExcludeRandomBeaconHistory,
147+
"exclude-randombeaconhistory",
148+
false,
149+
"exclude random beacon history",
150+
)
142151
}
143152

153+
const (
154+
randomBeaconHistoryDomain = common.StorageDomainContract
155+
randomBeaconHistoryDomainKey = interpreter.StringStorageMapKey("RandomBeaconHistory")
156+
)
157+
144158
type mode uint8
145159

146160
const (
@@ -191,6 +205,10 @@ func run(*cobra.Command, []string) {
191205
)
192206
}
193207

208+
if flagExcludeRandomBeaconHistory {
209+
log.Info().Msg("--exclude-randombeaconhistory is set to exclude random beacon history")
210+
}
211+
194212
var acctsToSkipForCadenceValueDiff []string
195213

196214
// Skip EVM storage account when diffing Cadence values.
@@ -336,6 +354,7 @@ func diffAccount(
336354
rw reporters.ReportWriter,
337355
mode mode,
338356
acctsToSkip []string,
357+
serviceAccountAddress common.Address,
339358
) (err error) {
340359

341360
diffValues := flagAlwaysDiffValues
@@ -398,6 +417,15 @@ func diffAccount(
398417
accountRegisters1,
399418
accountRegisters2,
400419
common.AllStorageDomains,
420+
func(address common.Address, domain common.StorageDomain, key any) bool {
421+
if flagExcludeRandomBeaconHistory {
422+
if isRandomBeaconHistory(serviceAccountAddress, address, domain, key) {
423+
log.Info().Msgf("excluding random beacon history in account %s, domain %s, key %v", address, domain.Identifier(), key)
424+
return false
425+
}
426+
}
427+
return true
428+
},
401429
)
402430
}
403431

@@ -415,6 +443,8 @@ func diff(
415443
) error {
416444
log.Info().Msgf("Diffing %d accounts", registers1.AccountCount())
417445

446+
serviceAccountAddress := serviceAccountAddressForChain(chainID)
447+
418448
if registers1.AccountCount() < nWorkers {
419449
nWorkers = registers1.AccountCount()
420450
}
@@ -454,6 +484,7 @@ func diff(
454484
rw,
455485
mode,
456486
acctsToSkip,
487+
common.Address(serviceAccountAddress),
457488
)
458489
if err != nil {
459490
log.Warn().Err(err).Msgf("failed to diff account %x", []byte(owner))
@@ -509,6 +540,7 @@ func diff(
509540
rw,
510541
mode,
511542
acctsToSkip,
543+
common.Address(serviceAccountAddress),
512544
)
513545

514546
select {
@@ -669,3 +701,29 @@ func (e countDiff) MarshalJSON() ([]byte, error) {
669701
State2: e.State2,
670702
})
671703
}
704+
705+
func isRandomBeaconHistory(serviceAccountAddress, address common.Address, domain common.StorageDomain, key any) bool {
706+
if serviceAccountAddress.Compare(address) != 0 {
707+
return false
708+
}
709+
710+
if domain != randomBeaconHistoryDomain {
711+
return false
712+
}
713+
714+
switch key := key.(type) {
715+
case interpreter.StringAtreeValue:
716+
return interpreter.StringStorageMapKey(key) == randomBeaconHistoryDomainKey
717+
718+
case interpreter.StringStorageMapKey:
719+
return key == randomBeaconHistoryDomainKey
720+
721+
default:
722+
return false
723+
}
724+
}
725+
726+
func serviceAccountAddressForChain(chainID flow.ChainID) flow.Address {
727+
sc := systemcontracts.SystemContractsForChain(chainID)
728+
return sc.FlowServiceAccount.Address
729+
}

0 commit comments

Comments
 (0)