Merge pull request OffchainLabs#1787 from OffchainLabs/dataposter-for…

…-validator

Use dataposter from EoaValidatorWallet

arbnode/batch_poster.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/offchainlabs/nitro/arbnode/dataposter"
 	"github.com/offchainlabs/nitro/arbnode/dataposter/storage"
+	"github.com/offchainlabs/nitro/arbnode/redislock"
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbstate"
 	"github.com/offchainlabs/nitro/arbutil"
@@ -67,7 +68,7 @@ type BatchPoster struct {
 	building     *buildingBatch
 	daWriter     das.DataAvailabilityServiceWriter
 	dataPoster   *dataposter.DataPoster
-	redisLock    *SimpleRedisLock
+	redisLock    *redislock.Simple
 	firstAccErr  time.Time // first time a continuous missing accumulator occurred
 	backlog      uint64    // An estimate of the number of unposted batches
 
@@ -99,7 +100,7 @@ type BatchPosterConfig struct {
 	GasRefunderAddress                 string                      `koanf:"gas-refunder-address" reload:"hot"`
 	DataPoster                         dataposter.DataPosterConfig `koanf:"data-poster" reload:"hot"`
 	RedisUrl                           string                      `koanf:"redis-url"`
-	RedisLock                          SimpleRedisLockConfig       `koanf:"redis-lock" reload:"hot"`
+	RedisLock                          redislock.SimpleCfg         `koanf:"redis-lock" reload:"hot"`
 	ExtraBatchGas                      uint64                      `koanf:"extra-batch-gas" reload:"hot"`
 	L1Wallet                           genericconf.WalletConfig    `koanf:"parent-chain-wallet"`
 	L1BlockBound                       string                      `koanf:"l1-block-bound" reload:"hot"`
@@ -150,7 +151,7 @@ func BatchPosterConfigAddOptions(prefix string, f *pflag.FlagSet) {
 	f.String(prefix+".redis-url", DefaultBatchPosterConfig.RedisUrl, "if non-empty, the Redis URL to store queued transactions in")
 	f.String(prefix+".l1-block-bound", DefaultBatchPosterConfig.L1BlockBound, "only post messages to batches when they're within the max future block/timestamp as of this L1 block tag (\"safe\", \"finalized\", \"latest\", or \"ignore\" to ignore this check)")
 	f.Duration(prefix+".l1-block-bound-bypass", DefaultBatchPosterConfig.L1BlockBoundBypass, "post batches even if not within the layer 1 future bounds if we're within this margin of the max delay")
-	RedisLockConfigAddOptions(prefix+".redis-lock", f)
+	redislock.AddConfigOptions(prefix+".redis-lock", f)
 	dataposter.DataPosterConfigAddOptions(prefix+".data-poster", f)
 	genericconf.WalletConfigAddOptions(prefix+".parent-chain-wallet", f, DefaultBatchPosterConfig.L1Wallet.Pathname)
 }
@@ -218,10 +219,10 @@ func NewBatchPoster(dataPosterDB ethdb.Database, l1Reader *headerreader.HeaderRe
 	if err != nil {
 		return nil, err
 	}
-	redisLockConfigFetcher := func() *SimpleRedisLockConfig {
+	redisLockConfigFetcher := func() *redislock.SimpleCfg {
 		return &config().RedisLock
 	}
-	redisLock, err := NewSimpleRedisLock(redisClient, redisLockConfigFetcher, func() bool { return syncMonitor.Synced() })
+	redisLock, err := redislock.NewSimple(redisClient, redisLockConfigFetcher, func() bool { return syncMonitor.Synced() })
 	if err != nil {
 		return nil, err
 	}
@@ -871,8 +872,7 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context) (bool, error)
 	if err != nil {
 		return false, err
 	}
-	err = b.dataPoster.PostTransaction(ctx, firstMsgTime, nonce, newMeta, b.seqInboxAddr, data, gasLimit)
-	if err != nil {
+	if _, err := b.dataPoster.PostTransaction(ctx, firstMsgTime, nonce, newMeta, b.seqInboxAddr, data, gasLimit, new(big.Int)); err != nil {
 		return false, err
 	}
 	log.Info(

arbnode/dataposter/data_poster.go

@@ -75,7 +75,7 @@ func parseReplacementTimes(val string) ([]time.Duration, error) {
 	for _, s := range strings.Split(val, ",") {
 		t, err := time.ParseDuration(s)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("parsing durations: %w", err)
 		}
 		if t <= lastReplacementTime {
 			return nil, errors.New("replacement times must be increasing")
@@ -273,29 +273,29 @@ func (p *DataPoster) feeAndTipCaps(ctx context.Context, gasLimit uint64, lastFee
 	return newFeeCap, newTipCap, nil
 }
 
-func (p *DataPoster) PostTransaction(ctx context.Context, dataCreatedAt time.Time, nonce uint64, meta []byte, to common.Address, calldata []byte, gasLimit uint64) error {
+func (p *DataPoster) PostTransaction(ctx context.Context, dataCreatedAt time.Time, nonce uint64, meta []byte, to common.Address, calldata []byte, gasLimit uint64, value *big.Int) (*types.Transaction, error) {
 	p.mutex.Lock()
 	defer p.mutex.Unlock()
 	err := p.updateBalance(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to update data poster balance: %w", err)
+		return nil, fmt.Errorf("failed to update data poster balance: %w", err)
 	}
 	feeCap, tipCap, err := p.feeAndTipCaps(ctx, gasLimit, nil, nil, dataCreatedAt, 0)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	inner := types.DynamicFeeTx{
 		Nonce:     nonce,
 		GasTipCap: tipCap,
 		GasFeeCap: feeCap,
 		Gas:       gasLimit,
 		To:        &to,
-		Value:     new(big.Int),
+		Value:     value,
 		Data:      calldata,
 	}
 	fullTx, err := p.signer(p.sender, types.NewTx(&inner))
 	if err != nil {
-		return err
+		return nil, fmt.Errorf("signing transaction: %w", err)
 	}
 	queuedTx := storage.QueuedTransaction{
 		Data:            inner,
@@ -305,7 +305,7 @@ func (p *DataPoster) PostTransaction(ctx context.Context, dataCreatedAt time.Tim
 		Created:         dataCreatedAt,
 		NextReplacement: time.Now().Add(p.replacementTimes[0]),
 	}
-	return p.sendTx(ctx, nil, &queuedTx)
+	return fullTx, p.sendTx(ctx, nil, &queuedTx)
 }
 
 // the mutex must be held by the caller

arbnode/dataposter/storage/storage.go

@@ -10,7 +10,8 @@ import (
 var (
 	ErrStorageRace = errors.New("storage race error")
 
-	DataPosterPrefix string = "d" // the prefix for all data poster keys
+	BlockValidatorPrefix string = "v" // the prefix for all block validator keys
+	BatchPosterPrefix    string = "b" // the prefix for all batch poster keys
 	// TODO(anodar): move everything else from schema.go file to here once
 	// execution split is complete.
 )

arbnode/maintenance.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/offchainlabs/nitro/arbnode/redislock"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 	flag "github.com/spf13/pflag"
 )
@@ -27,12 +28,12 @@ type MaintenanceRunner struct {
 
 	// lock is used to ensures that at any given time, only single node is on
 	// maintenance mode.
-	lock *SimpleRedisLock
+	lock *redislock.Simple
 }
 
 type MaintenanceConfig struct {
-	TimeOfDay string                `koanf:"time-of-day" reload:"hot"`
-	Lock      SimpleRedisLockConfig `koanf:"lock" reload:"hot"`
+	TimeOfDay string              `koanf:"time-of-day" reload:"hot"`
+	Lock      redislock.SimpleCfg `koanf:"lock" reload:"hot"`
 
 	// Generated: the minutes since start of UTC day to compact at
 	minutesAfterMidnight int
@@ -70,7 +71,7 @@ func (c *MaintenanceConfig) Validate() error {
 
 func MaintenanceConfigAddOptions(prefix string, f *flag.FlagSet) {
 	f.String(prefix+".time-of-day", DefaultMaintenanceConfig.TimeOfDay, "UTC 24-hour time of day to run maintenance (currently only db compaction) at (e.g. 15:00)")
-	RedisLockConfigAddOptions(prefix+".lock", f)
+	redislock.AddConfigOptions(prefix+".lock", f)
 }
 
 var DefaultMaintenanceConfig = MaintenanceConfig{
@@ -94,9 +95,9 @@ func NewMaintenanceRunner(config MaintenanceConfigFetcher, seqCoordinator *SeqCo
 	}
 
 	if seqCoordinator != nil {
-		c := func() *SimpleRedisLockConfig { return &cfg.Lock }
+		c := func() *redislock.SimpleCfg { return &cfg.Lock }
 		r := func() bool { return true } // always ready to lock
-		rl, err := NewSimpleRedisLock(seqCoordinator.Client, c, r)
+		rl, err := redislock.NewSimple(seqCoordinator.Client, c, r)
 		if err != nil {
 			return nil, fmt.Errorf("creating new simple redis lock: %w", err)
 		}

arbnode/node.go

@@ -26,7 +26,10 @@ import (
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/offchainlabs/nitro/arbnode/dataposter"
+	"github.com/offchainlabs/nitro/arbnode/dataposter/storage"
 	"github.com/offchainlabs/nitro/arbnode/execution"
+	"github.com/offchainlabs/nitro/arbnode/redislock"
 	"github.com/offchainlabs/nitro/arbnode/resourcemanager"
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcastclient"
@@ -41,6 +44,7 @@ import (
 	"github.com/offchainlabs/nitro/staker"
 	"github.com/offchainlabs/nitro/util/contracts"
 	"github.com/offchainlabs/nitro/util/headerreader"
+	"github.com/offchainlabs/nitro/util/redisutil"
 	"github.com/offchainlabs/nitro/util/signature"
 	"github.com/offchainlabs/nitro/wsbroadcastserver"
 )
@@ -539,6 +543,29 @@ func checkArbDbSchemaVersion(arbDb ethdb.Database) error {
 	return nil
 }
 
+func ValidatorDataposter(db ethdb.Database, l1Reader *headerreader.HeaderReader,
+	transactOpts *bind.TransactOpts, cfgFetcher ConfigFetcher, syncMonitor *SyncMonitor) (*dataposter.DataPoster, error) {
+	cfg := cfgFetcher.Get()
+	mdRetriever := func(ctx context.Context, blockNum *big.Int) ([]byte, error) {
+		return nil, nil
+	}
+	redisC, err := redisutil.RedisClientFromURL(cfg.BlockValidator.RedisUrl)
+	if err != nil {
+		return nil, fmt.Errorf("creating redis client from url: %w", err)
+	}
+	lockCfgFetcher := func() *redislock.SimpleCfg {
+		return &cfg.BlockValidator.RedisLock
+	}
+	redisLock, err := redislock.NewSimple(redisC, lockCfgFetcher, func() bool { return syncMonitor.Synced() })
+	if err != nil {
+		return nil, err
+	}
+	dpCfg := func() *dataposter.DataPosterConfig {
+		return &cfg.BlockValidator.DataPoster
+	}
+	return dataposter.NewDataPoster(db, l1Reader, transactOpts, redisC, redisLock, dpCfg, mdRetriever)
+}
+
 func createNodeImpl(
 	ctx context.Context,
 	stack *node.Node,
@@ -741,7 +768,7 @@ func createNodeImpl(
 			inboxTracker,
 			txStreamer,
 			exec.Recorder,
-			rawdb.NewTable(arbDb, BlockValidatorPrefix),
+			rawdb.NewTable(arbDb, storage.BlockValidatorPrefix),
 			daReader,
 			func() *staker.BlockValidatorConfig { return &configFetcher.Get().BlockValidator },
 			stack,
@@ -775,6 +802,16 @@ func createNodeImpl(
 	var messagePruner *MessagePruner
 
 	if config.Staker.Enable {
+		dp, err := ValidatorDataposter(
+			rawdb.NewTable(arbDb, storage.BlockValidatorPrefix),
+			l1Reader,
+			txOptsValidator,
+			configFetcher,
+			syncMonitor,
+		)
+		if err != nil {
+			return nil, err
+		}
 		var wallet staker.ValidatorWalletInterface
 		if config.Staker.UseSmartContractWallet || txOptsValidator == nil {
 			var existingWalletAddress *common.Address
@@ -794,7 +831,7 @@ func createNodeImpl(
 			if len(config.Staker.ContractWalletAddress) > 0 {
 				return nil, errors.New("validator contract wallet specified but flag to use a smart contract wallet was not specified")
 			}
-			wallet, err = staker.NewEoaValidatorWallet(deployInfo.Rollup, l1client, txOptsValidator)
+			wallet, err = staker.NewEoaValidatorWallet(dp, deployInfo.Rollup, l1client, txOptsValidator)
 			if err != nil {
 				return nil, err
 			}
@@ -833,7 +870,7 @@ func createNodeImpl(
 		if txOptsBatchPoster == nil {
 			return nil, errors.New("batchposter, but no TxOpts")
 		}
-		batchPoster, err = NewBatchPoster(rawdb.NewTable(arbDb, BlockValidatorPrefix), l1Reader, inboxTracker, txStreamer, syncMonitor, func() *BatchPosterConfig { return &configFetcher.Get().BatchPoster }, deployInfo, txOptsBatchPoster, daWriter)
+		batchPoster, err = NewBatchPoster(rawdb.NewTable(arbDb, storage.BatchPosterPrefix), l1Reader, inboxTracker, txStreamer, syncMonitor, func() *BatchPosterConfig { return &configFetcher.Get().BatchPoster }, deployInfo, txOptsBatchPoster, daWriter)
 		if err != nil {
 			return nil, err
 		}