Merge branch 'master' into bold_staker_balance_metrics

Author: amsanghi

arbnode/node.go

@@ -880,7 +880,7 @@ func getStatelessBlockValidator(
 		err = errors.New("no validator url specified")
 	}
 	if err != nil {
-		if config.ValidatorRequired() || config.Staker.Enable {
+		if config.ValidatorRequired() {
 			return nil, fmt.Errorf("%w: failed to init block validator", err)
 		}
 		log.Warn("validation not supported", "err", err)

arbnode/seq_coordinator.go

@@ -69,6 +69,7 @@ type SeqCoordinatorConfig struct {
 	ChosenHealthcheckAddr string        `koanf:"chosen-healthcheck-addr"`
 	RedisUrl              string        `koanf:"redis-url"`
 	NewRedisUrl           string        `koanf:"new-redis-url"`
+	RedisQuorumSize       uint64        `koanf:"redis-quorum-size"`
 	LockoutDuration       time.Duration `koanf:"lockout-duration"`
 	LockoutSpare          time.Duration `koanf:"lockout-spare"`
 	SeqNumDuration        time.Duration `koanf:"seq-num-duration"`
@@ -96,6 +97,7 @@ func SeqCoordinatorConfigAddOptions(prefix string, f *flag.FlagSet) {
 	f.Bool(prefix+".enable", DefaultSeqCoordinatorConfig.Enable, "enable sequence coordinator")
 	f.String(prefix+".redis-url", DefaultSeqCoordinatorConfig.RedisUrl, "the Redis URL to coordinate via")
 	f.String(prefix+".new-redis-url", DefaultSeqCoordinatorConfig.NewRedisUrl, "switch to the new Redis URL to coordinate via")
+	f.Uint64(prefix+".redis-quorum-size", DefaultSeqCoordinatorConfig.RedisQuorumSize, "the quorum size needed to qualify a redis GET as valid")
 	f.String(prefix+".chosen-healthcheck-addr", DefaultSeqCoordinatorConfig.ChosenHealthcheckAddr, "if non-empty, launch an HTTP service binding to this address that returns status code 200 when chosen and 503 otherwise")
 	f.Duration(prefix+".lockout-duration", DefaultSeqCoordinatorConfig.LockoutDuration, "duration to hold the sequencer lockout after acquiring it")
 	f.Duration(prefix+".lockout-spare", DefaultSeqCoordinatorConfig.LockoutSpare, "time to subtract from lockout duration to ensure timely renewal")
@@ -117,6 +119,7 @@ var DefaultSeqCoordinatorConfig = SeqCoordinatorConfig{
 	ChosenHealthcheckAddr: "",
 	RedisUrl:              "",
 	NewRedisUrl:           "",
+	RedisQuorumSize:       1,
 	LockoutDuration:       time.Minute,
 	LockoutSpare:          30 * time.Second,
 	SeqNumDuration:        10 * 24 * time.Hour,
@@ -136,6 +139,7 @@ var TestSeqCoordinatorConfig = SeqCoordinatorConfig{
 	Enable:                false,
 	RedisUrl:              "",
 	NewRedisUrl:           "",
+	RedisQuorumSize:       1,
 	LockoutDuration:       time.Second * 2,
 	LockoutSpare:          time.Millisecond * 10,
 	SeqNumDuration:        time.Minute * 10,
@@ -159,7 +163,7 @@ func NewSeqCoordinator(
 	sync *SyncMonitor,
 	config SeqCoordinatorConfig,
 ) (*SeqCoordinator, error) {
-	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl)
+	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl, config.RedisQuorumSize)
 	if err != nil {
 		return nil, err
 	}
@@ -610,7 +614,7 @@ func (c *SeqCoordinator) deleteFinalizedMsgsFromRedis(ctx context.Context, final
 }
 
 func (c *SeqCoordinator) blockMetadataAt(ctx context.Context, pos arbutil.MessageIndex) (common.BlockMetadata, error) {
-	blockMetadataStr, err := c.RedisCoordinator().Client.Get(ctx, redisutil.BlockMetadataKeyFor(pos)).Result()
+	blockMetadataStr, err := c.RedisCoordinator().GetIfInQuorum(ctx, redisutil.BlockMetadataKeyFor(pos))
 	if err != nil {
 		if errors.Is(err, redis.Nil) {
 			return nil, nil
@@ -673,12 +677,12 @@ func (c *SeqCoordinator) update(ctx context.Context) (time.Duration, error) {
 		return c.retryAfterRedisError(), nil
 	}
 	readUntil := min(localMsgCount+c.config.MsgPerPoll, remoteMsgCount)
-	client := c.RedisCoordinator().Client
+	redisCoordinator := c.RedisCoordinator()
 	// If we have a previous redis coordinator,
 	// we can read from it until the local message count catches up to the prev coordinator's message count
 	if c.prevRedisMessageCount > localMsgCount {
 		readUntil = min(readUntil, c.prevRedisMessageCount)
-		client = c.prevRedisCoordinator.Client
+		redisCoordinator = c.prevRedisCoordinator
 	}
 	if c.prevRedisMessageCount != 0 && localMsgCount >= c.prevRedisMessageCount {
 		log.Info("coordinator caught up to prev redis coordinator", "msgcount", localMsgCount, "prevMsgCount", c.prevRedisMessageCount)
@@ -689,7 +693,7 @@ func (c *SeqCoordinator) update(ctx context.Context) (time.Duration, error) {
 	var msgReadErr error
 	for msgToRead < readUntil && localMsgCount >= remoteFinalizedMsgCount {
 		var resString string
-		resString, msgReadErr = client.Get(ctx, redisutil.MessageKeyFor(msgToRead)).Result()
+		resString, msgReadErr = redisCoordinator.GetIfInQuorum(ctx, redisutil.MessageKeyFor(msgToRead))
 		if msgReadErr != nil && c.sequencer.Synced(ctx) {
 			log.Warn("coordinator failed reading message", "pos", msgToRead, "err", msgReadErr)
 			break
@@ -698,7 +702,7 @@ func (c *SeqCoordinator) update(ctx context.Context) (time.Duration, error) {
 		var sigString string
 		var sigBytes []byte
 		sigSeparateKey := true
-		sigString, msgReadErr = client.Get(ctx, redisutil.MessageSigKeyFor(msgToRead)).Result()
+		sigString, msgReadErr = redisCoordinator.GetIfInQuorum(ctx, redisutil.MessageSigKeyFor(msgToRead))
 		if errors.Is(msgReadErr, redis.Nil) {
 			// no separate signature. Try reading old-style sig
 			if len(rsBytes) < 32 {
@@ -900,7 +904,7 @@ func (c *SeqCoordinator) Start(ctxIn context.Context) {
 	var newRedisCoordinator *redisutil.RedisCoordinator
 	if c.config.NewRedisUrl != "" {
 		var err error
-		newRedisCoordinator, err = redisutil.NewRedisCoordinator(c.config.NewRedisUrl)
+		newRedisCoordinator, err = redisutil.NewRedisCoordinator(c.config.NewRedisUrl, c.config.RedisQuorumSize)
 		if err != nil {
 			log.Warn("failed to create new redis coordinator", "err",
 				err, "newRedisUrl", c.config.NewRedisUrl)

arbnode/seq_coordinator_test.go

@@ -125,7 +125,7 @@ func TestRedisSeqCoordinatorAtomic(t *testing.T) {
 	for i := 0; i < NumOfThreads; i++ {
 		config := coordConfig
 		config.MyUrl = fmt.Sprint(i)
-		redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl)
+		redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl, config.RedisQuorumSize)
 		Require(t, err)
 		coordinator := &SeqCoordinator{
 			redisCoordinator: *redisCoordinator,
@@ -181,7 +181,7 @@ func TestSeqCoordinatorDeletesFinalizedMessages(t *testing.T) {
 
 	config := coordConfig
 	config.MyUrl = "test"
-	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl)
+	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl, config.RedisQuorumSize)
 	Require(t, err)
 	coordinator := &SeqCoordinator{
 		redisCoordinator: *redisCoordinator,
@@ -272,7 +272,7 @@ func TestSeqCoordinatorAddsBlockMetadata(t *testing.T) {
 
 	config := coordConfig
 	config.MyUrl = "test"
-	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl)
+	redisCoordinator, err := redisutil.NewRedisCoordinator(config.RedisUrl, config.RedisQuorumSize)
 	Require(t, err)
 	coordinator := &SeqCoordinator{
 		redisCoordinator: *redisCoordinator,

cmd/nitro/nitro.go

@@ -224,7 +224,7 @@ func mainImpl() int {
 	var l1TransactionOptsBatchPoster *bind.TransactOpts
 	// If sequencer and signing is enabled or batchposter is enabled without
 	// external signing sequencer will need a key.
-	sequencerNeedsKey := (nodeConfig.Node.Sequencer && !nodeConfig.Node.Feed.Output.DisableSigning) ||
+	sequencerNeedsKey := (nodeConfig.Node.Sequencer && nodeConfig.Node.Feed.Output.Signed) ||
 		(nodeConfig.Node.BatchPoster.Enable && (nodeConfig.Node.BatchPoster.DataPoster.ExternalSigner.URL == "" || nodeConfig.Node.DataAvailability.Enable))
 	validatorNeedsKey := nodeConfig.Node.Staker.OnlyCreateWalletContract ||
 		(nodeConfig.Node.Staker.Enable && !strings.EqualFold(nodeConfig.Node.Staker.Strategy, "watchtower") && nodeConfig.Node.Staker.DataPoster.ExternalSigner.URL == "")
@@ -531,9 +531,13 @@ func mainImpl() int {
 		log.Error("failed to create execution node", "err", err)
 		return 1
 	}
-	locator, err := server_common.NewMachineLocator(liveNodeConfig.Get().Validation.Wasm.RootPath)
-	if err != nil {
-		log.Error("failed to create machine locator: %w", err)
+	var wasmModuleRoot common.Hash
+	if liveNodeConfig.Get().Node.ValidatorRequired() {
+		locator, err := server_common.NewMachineLocator(liveNodeConfig.Get().Validation.Wasm.RootPath)
+		if err != nil {
+			log.Error("failed to create machine locator: %w", err)
+		}
+		wasmModuleRoot = locator.LatestWasmModuleRoot()
 	}
 
 	currentNode, err := arbnode.CreateNodeFullExecutionClient(
@@ -554,7 +558,7 @@ func mainImpl() int {
 		fatalErrChan,
 		new(big.Int).SetUint64(nodeConfig.ParentChain.ID),
 		blobReader,
-		locator.LatestWasmModuleRoot(),
+		wasmModuleRoot,
 	)
 	if err != nil {
 		log.Error("failed to create node", "err", err)

cmd/seq-coordinator-manager/seq-coordinator-manager.go

@@ -49,7 +49,7 @@ func main() {
 		os.Exit(1)
 	}
 	redisURL := args[0]
-	redisutilCoordinator, err := redisutil.NewRedisCoordinator(redisURL)
+	redisutilCoordinator, err := redisutil.NewRedisCoordinator(redisURL, 1)
 	if err != nil {
 		panic(err)
 	}

execution/gethexec/blockchain.go

@@ -44,6 +44,7 @@ type CachingConfig struct {
 	DisableStylusCacheMetricsCollection bool          `koanf:"disable-stylus-cache-metrics-collection"`
 	StateScheme                         string        `koanf:"state-scheme"`
 	StateHistory                        uint64        `koanf:"state-history"`
+	EnablePreimages                     bool          `koanf:"enable-preimages"`
 }
 
 func CachingConfigAddOptions(prefix string, f *flag.FlagSet) {
@@ -65,6 +66,7 @@ func CachingConfigAddOptions(prefix string, f *flag.FlagSet) {
 	f.Bool(prefix+".disable-stylus-cache-metrics-collection", DefaultCachingConfig.DisableStylusCacheMetricsCollection, "disable metrics collection for the stylus cache")
 	f.String(prefix+".state-scheme", DefaultCachingConfig.StateScheme, "scheme to use for state trie storage (hash, path)")
 	f.Uint64(prefix+".state-history", DefaultCachingConfig.StateHistory, "number of recent blocks to retain state history for (path state-scheme only)")
+	f.Bool(prefix+".enable-preimages", DefaultCachingConfig.EnablePreimages, "enable recording of preimages")
 }
 
 func getStateHistory(maxBlockSpeed time.Duration) uint64 {
@@ -109,7 +111,7 @@ func DefaultCacheConfigFor(stack *node.Node, cachingConfig *CachingConfig) *core
 		TriesInMemory:                      cachingConfig.BlockCount,
 		TrieRetention:                      cachingConfig.BlockAge,
 		SnapshotLimit:                      cachingConfig.SnapshotCache,
-		Preimages:                          baseConf.Preimages,
+		Preimages:                          baseConf.Preimages || cachingConfig.EnablePreimages,
 		SnapshotRestoreMaxGas:              cachingConfig.SnapshotRestoreGasLimit,
 		HeadRewindBlocksLimit:              cachingConfig.HeadRewindBlocksLimit,
 		MaxNumberOfBlocksToSkipStateSaving: cachingConfig.MaxNumberOfBlocksToSkipStateSaving,

execution/gethexec/forwarder.go

@@ -376,7 +376,7 @@ func (f *RedisTxForwarder) CheckHealth(ctx context.Context) error {
 // not thread safe vs update and itself
 func (f *RedisTxForwarder) Initialize(ctx context.Context) error {
 	var err error
-	f.redisCoordinator, err = redisutil.NewRedisCoordinator(f.config.RedisUrl)
+	f.redisCoordinator, err = redisutil.NewRedisCoordinator(f.config.RedisUrl, 1)
 	if err != nil {
 		return fmt.Errorf("unable to create redis coordinator: %w", err)
 	}

system_tests/timeboost_test.go

@@ -323,7 +323,7 @@ func testTxsHandlingDuringSequencerSwap(t *testing.T, dueToCrash bool) {
 	Require(t, err)
 
 	// Set reader and writer coordinators for redis
-	redisCoordinatorGetter, err := redisutil.NewRedisCoordinator(builderSeq.nodeConfig.SeqCoordinator.RedisUrl)
+	redisCoordinatorGetter, err := redisutil.NewRedisCoordinator(builderSeq.nodeConfig.SeqCoordinator.RedisUrl, builderSeq.nodeConfig.SeqCoordinator.RedisQuorumSize)
 	Require(t, err)
 	currentChosen, err := redisCoordinatorGetter.CurrentChosenSequencer(ctx)
 	Require(t, err)