Merge branch 'master' into l1-rpc-logging

Author: PlasmaPower

arbnode/node.go

@@ -8,7 +8,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"math"
 	"math/big"
 	"os"
 	"path/filepath"
@@ -478,7 +477,7 @@ func ConfigAddOptions(prefix string, f *flag.FlagSet, feedInputEnable bool, feed
 	broadcastclient.FeedConfigAddOptions(prefix+".feed", f, feedInputEnable, feedOutputEnable)
 	staker.L1ValidatorConfigAddOptions(prefix+".validator", f)
 	SeqCoordinatorConfigAddOptions(prefix+".seq-coordinator", f)
-	das.DataAvailabilityConfigAddOptions(prefix+".data-availability", f)
+	das.DataAvailabilityConfigAddNodeOptions(prefix+".data-availability", f)
 	WasmConfigAddOptions(prefix+".wasm", f)
 	SyncMonitorConfigAddOptions(prefix+".sync-monitor", f)
 	DangerousConfigAddOptions(prefix+".dangerous", f)
@@ -949,7 +948,7 @@ func createNodeImpl(
 				return nil, err
 			}
 		} else {
-			daReader, dasLifecycleManager, err = SetUpDataAvailability(ctx, &config.DataAvailability, l1Reader, deployInfo)
+			daReader, dasLifecycleManager, err = das.CreateDAReaderForNode(ctx, &config.DataAvailability, l1Reader, &deployInfo.SequencerInbox)
 			if err != nil {
 				return nil, err
 			}
@@ -1129,225 +1128,6 @@ func (n *Node) OnConfigReload(_ *Config, _ *Config) error {
 	return nil
 }
 
-type L1ReaderCloser struct {
-	l1Reader *headerreader.HeaderReader
-}
-
-func (c *L1ReaderCloser) Close(_ context.Context) error {
-	c.l1Reader.StopOnly()
-	return nil
-}
-
-func (c *L1ReaderCloser) String() string {
-	return "l1 reader closer"
-}
-
-// SetUpDataAvailabilityWithoutNode sets up a das.DataAvailabilityService stack
-// without relying on any objects already created for setting up the Node.
-func SetUpDataAvailabilityWithoutNode(
-	ctx context.Context,
-	config *das.DataAvailabilityConfig,
-) (das.DataAvailabilityService, *das.LifecycleManager, error) {
-	var l1Reader *headerreader.HeaderReader
-	if config.L1NodeURL != "" && config.L1NodeURL != "none" {
-		l1Client, err := das.GetL1Client(ctx, config.L1ConnectionAttempts, config.L1NodeURL)
-		if err != nil {
-			return nil, nil, err
-		}
-		l1Reader = headerreader.New(l1Client, func() *headerreader.Config { return &headerreader.DefaultConfig }) // TODO: config
-	}
-	newDas, lifeCycle, err := SetUpDataAvailability(ctx, config, l1Reader, nil)
-	if err != nil {
-		return nil, nil, err
-	}
-	if l1Reader != nil {
-		l1Reader.Start(ctx)
-		lifeCycle.Register(&L1ReaderCloser{l1Reader})
-	}
-	return newDas, lifeCycle, err
-}
-
-// SetUpDataAvailability sets up a das.DataAvailabilityService stack allowing
-// some dependencies that were created for the Node to be injected.
-func SetUpDataAvailability(
-	ctx context.Context,
-	config *das.DataAvailabilityConfig,
-	l1Reader *headerreader.HeaderReader,
-	deployInfo *RollupAddresses,
-) (das.DataAvailabilityService, *das.LifecycleManager, error) {
-	if !config.Enable {
-		return nil, nil, nil
-	}
-	var syncFromStorageServices []*das.IterableStorageService
-	var syncToStorageServices []das.StorageService
-	var seqInbox *bridgegen.SequencerInbox
-	var err error
-	var seqInboxCaller *bridgegen.SequencerInboxCaller
-	var seqInboxAddress *common.Address
-
-	if l1Reader != nil && deployInfo != nil {
-		seqInboxAddress = &deployInfo.SequencerInbox
-		seqInbox, err = bridgegen.NewSequencerInbox(deployInfo.SequencerInbox, l1Reader.Client())
-		if err != nil {
-			return nil, nil, err
-		}
-		seqInboxCaller = &seqInbox.SequencerInboxCaller
-	} else if config.L1NodeURL == "none" && config.SequencerInboxAddress == "none" {
-		l1Reader = nil
-		seqInboxAddress = nil
-	} else if l1Reader != nil && len(config.SequencerInboxAddress) > 0 {
-		seqInboxAddress, err = das.OptionalAddressFromString(config.SequencerInboxAddress)
-		if err != nil {
-			return nil, nil, err
-		}
-		if seqInboxAddress == nil {
-			return nil, nil, errors.New("must provide data-availability.sequencer-inbox-address set to a valid contract address or 'none'")
-		}
-		seqInbox, err = bridgegen.NewSequencerInbox(*seqInboxAddress, l1Reader.Client())
-		if err != nil {
-			return nil, nil, err
-		}
-		seqInboxCaller = &seqInbox.SequencerInboxCaller
-	} else {
-		return nil, nil, errors.New("data-availabilty.l1-node-url and sequencer-inbox-address must be set to a valid L1 URL and contract address or 'none' if running daserver executable")
-	}
-
-	// This function builds up the DataAvailabilityService with the following topology, starting from the leaves.
-	/*
-			      ChainFetchDAS → Bigcache → Redis →
-				       SignAfterStoreDAS →
-				              FallbackDAS (if the REST client aggregator was specified)
-				              (primary) → RedundantStorage (if multiple persistent backing stores were specified)
-				                            → S3
-				                            → DiskStorage
-				                            → Database
-				         (fallback only)→ RESTful client aggregator
-
-		          → : X--delegates to-->Y
-	*/
-	topLevelStorageService, dasLifecycleManager, err := das.CreatePersistentStorageService(ctx, config, &syncFromStorageServices, &syncToStorageServices)
-	if err != nil {
-		return nil, nil, err
-	}
-	hasPersistentStorage := topLevelStorageService != nil
-
-	// Create the REST aggregator if one was requested. If other storage types were enabled above, then
-	// the REST aggregator is used as the fallback to them.
-	if config.RestfulClientAggregatorConfig.Enable {
-		restAgg, err := das.NewRestfulClientAggregator(ctx, &config.RestfulClientAggregatorConfig)
-		if err != nil {
-			return nil, nil, err
-		}
-		restAgg.Start(ctx)
-		dasLifecycleManager.Register(restAgg)
-
-		// Wrap the primary storage service with the fallback to the restful aggregator
-		if hasPersistentStorage {
-			syncConf := &config.RestfulClientAggregatorConfig.SyncToStorageConfig
-			var retentionPeriodSeconds uint64
-			if uint64(syncConf.RetentionPeriod) == math.MaxUint64 {
-				retentionPeriodSeconds = math.MaxUint64
-			} else {
-				retentionPeriodSeconds = uint64(syncConf.RetentionPeriod.Seconds())
-			}
-			if syncConf.Eager {
-				if l1Reader == nil || seqInboxAddress == nil {
-					return nil, nil, errors.New("l1-node-url and sequencer-inbox-address must be specified along with sync-to-storage.eager")
-				}
-				topLevelStorageService, err = das.NewSyncingFallbackStorageService(
-					ctx,
-					topLevelStorageService,
-					restAgg,
-					l1Reader,
-					*seqInboxAddress,
-					syncConf)
-				if err != nil {
-					return nil, nil, err
-				}
-			} else {
-				topLevelStorageService = das.NewFallbackStorageService(topLevelStorageService, restAgg,
-					retentionPeriodSeconds, syncConf.IgnoreWriteErrors, true)
-			}
-		} else {
-			topLevelStorageService = das.NewReadLimitedStorageService(restAgg)
-		}
-		dasLifecycleManager.Register(topLevelStorageService)
-	}
-
-	var topLevelDas das.DataAvailabilityService
-	if config.AggregatorConfig.Enable {
-		panic("Tried to make an aggregator using wrong factory method")
-	}
-	if hasPersistentStorage && (config.KeyConfig.KeyDir != "" || config.KeyConfig.PrivKey != "") {
-		_seqInboxCaller := seqInboxCaller
-		if config.DisableSignatureChecking {
-			_seqInboxCaller = nil
-		}
-
-		privKey, err := config.KeyConfig.BLSPrivKey()
-		if err != nil {
-			return nil, nil, err
-		}
-
-		// TODO rename StorageServiceDASAdapter
-		topLevelDas, err = das.NewSignAfterStoreDASWithSeqInboxCaller(
-			privKey,
-			_seqInboxCaller,
-			topLevelStorageService,
-			config.ExtraSignatureCheckingPublicKey,
-		)
-		if err != nil {
-			return nil, nil, err
-		}
-	} else {
-		topLevelDas = das.NewReadLimitedDataAvailabilityService(topLevelStorageService)
-	}
-
-	// Enable caches, Redis and (local) BigCache. Local is the outermost, so it will be tried first.
-	if config.RedisCacheConfig.Enable {
-		cache, err := das.NewRedisStorageService(config.RedisCacheConfig, das.NewEmptyStorageService())
-		dasLifecycleManager.Register(cache)
-		if err != nil {
-			return nil, nil, err
-		}
-		if config.RedisCacheConfig.SyncFromStorageServices {
-			iterableStorageService := das.NewIterableStorageService(das.ConvertStorageServiceToIterationCompatibleStorageService(cache))
-			syncFromStorageServices = append(syncFromStorageServices, iterableStorageService)
-			cache = iterableStorageService
-		}
-		if config.RedisCacheConfig.SyncToStorageServices {
-			syncToStorageServices = append(syncToStorageServices, cache)
-		}
-		topLevelDas = das.NewCacheStorageToDASAdapter(topLevelDas, cache)
-	}
-	if config.LocalCacheConfig.Enable {
-		cache, err := das.NewBigCacheStorageService(config.LocalCacheConfig, das.NewEmptyStorageService())
-		dasLifecycleManager.Register(cache)
-		if err != nil {
-			return nil, nil, err
-		}
-		topLevelDas = das.NewCacheStorageToDASAdapter(topLevelDas, cache)
-	}
-
-	if config.RegularSyncStorageConfig.Enable && len(syncFromStorageServices) != 0 && len(syncToStorageServices) != 0 {
-		regularlySyncStorage := das.NewRegularlySyncStorage(syncFromStorageServices, syncToStorageServices, config.RegularSyncStorageConfig)
-		regularlySyncStorage.Start(ctx)
-	}
-
-	if topLevelDas != nil && seqInbox != nil {
-		topLevelDas, err = das.NewChainFetchDASWithSeqInbox(topLevelDas, seqInbox)
-		if err != nil {
-			return nil, nil, err
-		}
-	}
-
-	if topLevelDas == nil {
-		return nil, nil, errors.New("data-availability.enable was specified but no Data Availability server types were enabled")
-	}
-
-	return topLevelDas, dasLifecycleManager, nil
-}
-
 func CreateNode(
 	ctx context.Context,
 	stack *node.Node,

arbstate/das_reader.go

@@ -21,7 +21,6 @@ import (
 
 type DataAvailabilityReader interface {
 	GetByHash(ctx context.Context, hash common.Hash) ([]byte, error)
-	HealthCheck(ctx context.Context) error
 	ExpirationPolicy(ctx context.Context) (ExpirationPolicy, error)
 }
 

cmd/daserver/daserver.go

@@ -16,14 +16,15 @@ import (
 	koanfjson "github.com/knadh/koanf/parsers/json"
 	flag "github.com/spf13/pflag"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/metrics/exp"
 
-	"github.com/offchainlabs/nitro/arbnode"
 	"github.com/offchainlabs/nitro/cmd/genericconf"
 	"github.com/offchainlabs/nitro/cmd/util/confighelpers"
 	"github.com/offchainlabs/nitro/das"
+	"github.com/offchainlabs/nitro/util/headerreader"
 )
 
 type DAServerConfig struct {
@@ -89,7 +90,7 @@ func parseDAServer(args []string) (*DAServerConfig, error) {
 	genericconf.MetricsServerAddOptions("metrics-server", f)
 
 	f.Int("log-level", int(log.LvlInfo), "log level; 1: ERROR, 2: WARN, 3: INFO, 4: DEBUG, 5: TRACE")
-	das.DataAvailabilityConfigAddOptions("data-availability", f)
+	das.DataAvailabilityConfigAddDaserverOptions("data-availability", f)
 	genericconf.ConfConfigAddOptions("conf", f)
 
 	k, err := confighelpers.BeginCommonParse(f, args)
@@ -121,6 +122,19 @@ func parseDAServer(args []string) (*DAServerConfig, error) {
 	return &serverConfig, nil
 }
 
+type L1ReaderCloser struct {
+	l1Reader *headerreader.HeaderReader
+}
+
+func (c *L1ReaderCloser) Close(_ context.Context) error {
+	c.l1Reader.StopOnly()
+	return nil
+}
+
+func (c *L1ReaderCloser) String() string {
+	return "l1 reader closer"
+}
+
 func startup() error {
 	// Some different defaults to DAS config in a node.
 	das.DefaultDataAvailabilityConfig.Enable = true
@@ -155,17 +169,46 @@ func startup() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	dasImpl, dasLifecycleManager, err := arbnode.SetUpDataAvailabilityWithoutNode(ctx, &serverConfig.DAConf)
+	var l1Reader *headerreader.HeaderReader
+	if serverConfig.DAConf.L1NodeURL != "" && serverConfig.DAConf.L1NodeURL != "none" {
+		l1Client, err := das.GetL1Client(ctx, serverConfig.DAConf.L1ConnectionAttempts, serverConfig.DAConf.L1NodeURL)
+		if err != nil {
+			return err
+		}
+		l1Reader = headerreader.New(l1Client, func() *headerreader.Config { return &headerreader.DefaultConfig }) // TODO: config
+	}
+
+	var seqInboxAddress *common.Address
+	if serverConfig.DAConf.SequencerInboxAddress == "none" {
+		seqInboxAddress = nil
+	} else if len(serverConfig.DAConf.SequencerInboxAddress) > 0 {
+		seqInboxAddress, err = das.OptionalAddressFromString(serverConfig.DAConf.SequencerInboxAddress)
+		if err != nil {
+			return err
+		}
+		if seqInboxAddress == nil {
+			return errors.New("must provide data-availability.sequencer-inbox-address set to a valid contract address or 'none'")
+		}
+	} else {
+		return errors.New("sequencer-inbox-address must be set to a valid L1 URL and contract address, or 'none'")
+	}
+
+	daReader, daWriter, daHealthChecker, dasLifecycleManager, err := das.CreateDAComponentsForDaserver(ctx, &serverConfig.DAConf, l1Reader, seqInboxAddress)
 	if err != nil {
 		return err
 	}
 
+	if l1Reader != nil {
+		l1Reader.Start(ctx)
+		dasLifecycleManager.Register(&L1ReaderCloser{l1Reader})
+	}
+
 	vcsRevision, vcsTime := confighelpers.GetVersion()
 	var rpcServer *http.Server
 	if serverConfig.EnableRPC {
 		log.Info("Starting HTTP-RPC server", "addr", serverConfig.RPCAddr, "port", serverConfig.RPCPort, "revision", vcsRevision, "vcs.time", vcsTime)
 
-		rpcServer, err = das.StartDASRPCServer(ctx, serverConfig.RPCAddr, serverConfig.RPCPort, serverConfig.RPCServerTimeouts, dasImpl)
+		rpcServer, err = das.StartDASRPCServer(ctx, serverConfig.RPCAddr, serverConfig.RPCPort, serverConfig.RPCServerTimeouts, daReader, daWriter, daHealthChecker)
 		if err != nil {
 			return err
 		}
@@ -175,7 +218,7 @@ func startup() error {
 	if serverConfig.EnableREST {
 		log.Info("Starting REST server", "addr", serverConfig.RESTAddr, "port", serverConfig.RESTPort, "revision", vcsRevision, "vcs.time", vcsTime)
 
-		restServer, err = das.NewRestfulDasServer(serverConfig.RESTAddr, serverConfig.RESTPort, serverConfig.RESTServerTimeouts, dasImpl)
+		restServer, err = das.NewRestfulDasServer(serverConfig.RESTAddr, serverConfig.RESTPort, serverConfig.RESTServerTimeouts, daReader, daHealthChecker)
 		if err != nil {
 			return err
 		}

cmd/dataavailability/data_availability_check.go

@@ -65,7 +65,6 @@ type DataAvailabilityCheck struct {
 	config         *DataAvailabilityCheckConfig
 	inboxAddr      *common.Address
 	inboxContract  *bridgegen.SequencerInbox
-	dataSource     arbstate.DataAvailabilityReader
 	urlToReaderMap map[string]arbstate.DataAvailabilityReader
 	checkInterval  time.Duration
 }
@@ -83,10 +82,6 @@ func newDataAvailabilityCheck(ctx context.Context, dataAvailabilityCheckConfig *
 	if err != nil {
 		return nil, err
 	}
-	dataSource, err := das.NewChainFetchReader(das.NewEmptyStorageService(), l1Client, *seqInboxAddress)
-	if err != nil {
-		return nil, err
-	}
 	onlineUrls, err := das.RestfulServerURLsFromList(ctx, dataAvailabilityCheckConfig.OnlineUrlList)
 	if err != nil {
 		return nil, err
@@ -104,7 +99,6 @@ func newDataAvailabilityCheck(ctx context.Context, dataAvailabilityCheckConfig *
 		config:         dataAvailabilityCheckConfig,
 		inboxAddr:      seqInboxAddress,
 		inboxContract:  inboxContract,
-		dataSource:     dataSource,
 		urlToReaderMap: urlToReaderMap,
 		checkInterval:  dataAvailabilityCheckConfig.CheckInterval,
 	}, nil