Unify DAS interfaces

cmd/replay/main.go

@@ -115,6 +115,10 @@ func (i WavmInbox) ReadDelayedInbox(seqNum uint64) (*arbostypes.L1IncomingMessag
 type PreimageDASReader struct {
 }
 
+func (*PreimageDASReader) String() string {
+	return "PreimageDASReader"
+}
+
 func (dasReader *PreimageDASReader) GetByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
 	oracle := func(hash common.Hash) ([]byte, error) {
 		return wavmio.ResolveTypedPreimage(arbutil.Keccak256PreimageType, hash)

daprovider/das/aggregator.go

@@ -33,7 +33,7 @@ var (
 	// there was a Store that had no backend failures.
 	anyErrorGauge = metrics.GetOrRegisterGauge(metricBase+"/error/gauge", nil)
 
-// Other aggregator metrics are generated dynamically in the Store function.
+	// Other aggregator metrics are generated dynamically in the Store function.
 )
 
 type AggregatorConfig struct {
@@ -74,7 +74,7 @@ type Aggregator struct {
 }
 
 type ServiceDetails struct {
-	service     DataAvailabilityServiceWriter
+	service     dasutil.DASWriter
 	pubKey      blsSignatures.PublicKey
 	signersMask uint64
 	metricName  string
@@ -84,7 +84,7 @@ func (s *ServiceDetails) String() string {
 	return fmt.Sprintf("ServiceDetails{service: %v, signersMask %d}", s.service, s.signersMask)
 }
 
-func NewServiceDetails(service DataAvailabilityServiceWriter, pubKey blsSignatures.PublicKey, signersMask uint64, metricName string) (*ServiceDetails, error) {
+func NewServiceDetails(service dasutil.DASWriter, pubKey blsSignatures.PublicKey, signersMask uint64, metricName string) (*ServiceDetails, error) {
 	if bits.OnesCount64(signersMask) != 1 {
 		return nil, fmt.Errorf("tried to configure backend DAS %v with invalid signersMask %X", service, signersMask)
 	}

daprovider/das/aggregator_test.go

@@ -121,20 +121,20 @@ func (b *randomBagOfFailures) shouldFail() failureType {
 type WrapStore struct {
 	t        *testing.T
 	injector failureInjector
-	DataAvailabilityServiceWriter
+	dasutil.DASWriter
 }
 
 func (w *WrapStore) Store(ctx context.Context, message []byte, timeout uint64) (*dasutil.DataAvailabilityCertificate, error) {
 	switch w.injector.shouldFail() {
 	case success:
-		return w.DataAvailabilityServiceWriter.Store(ctx, message, timeout)
+		return w.DASWriter.Store(ctx, message, timeout)
 	case immediateError:
 		return nil, errors.New("expected Store failure")
 	case tooSlow:
 		<-ctx.Done()
 		return nil, ctx.Err()
 	case dataCorruption:
-		cert, err := w.DataAvailabilityServiceWriter.Store(ctx, message, timeout)
+		cert, err := w.DASWriter.Store(ctx, message, timeout)
 		if err != nil {
 			return nil, err
 		}

daprovider/das/das.go

@@ -15,21 +15,8 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
-
-	"github.com/offchainlabs/nitro/daprovider/das/dasutil"
 )
 
-type DataAvailabilityServiceWriter interface {
-	// Store requests that the message be stored until timeout (UTC time in unix epoch seconds).
-	Store(ctx context.Context, message []byte, timeout uint64) (*dasutil.DataAvailabilityCertificate, error)
-	fmt.Stringer
-}
-
-type DataAvailabilityServiceReader interface {
-	dasutil.DASReader
-	fmt.Stringer
-}
-
 type DataAvailabilityServiceHealthChecker interface {
 	HealthCheck(ctx context.Context) error
 }

daprovider/das/dasRpcServer.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/offchainlabs/nitro/blsSignatures"
 	"github.com/offchainlabs/nitro/cmd/genericconf"
+	"github.com/offchainlabs/nitro/daprovider/das/dasutil"
 	"github.com/offchainlabs/nitro/util/pretty"
 )
 
@@ -36,24 +37,24 @@ var (
 )
 
 type DASRPCServer struct {
-	daReader        DataAvailabilityServiceReader
-	daWriter        DataAvailabilityServiceWriter
+	daReader        dasutil.DASReader
+	daWriter        dasutil.DASWriter
 	daHealthChecker DataAvailabilityServiceHealthChecker
 
 	signatureVerifier *SignatureVerifier
 
 	batches *batchBuilder
 }
 
-func StartDASRPCServer(ctx context.Context, addr string, portNum uint64, rpcServerTimeouts genericconf.HTTPServerTimeoutConfig, rpcServerBodyLimit int, daReader DataAvailabilityServiceReader, daWriter DataAvailabilityServiceWriter, daHealthChecker DataAvailabilityServiceHealthChecker, signatureVerifier *SignatureVerifier) (*http.Server, error) {
+func StartDASRPCServer(ctx context.Context, addr string, portNum uint64, rpcServerTimeouts genericconf.HTTPServerTimeoutConfig, rpcServerBodyLimit int, daReader dasutil.DASReader, daWriter dasutil.DASWriter, daHealthChecker DataAvailabilityServiceHealthChecker, signatureVerifier *SignatureVerifier) (*http.Server, error) {
 	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", addr, portNum))
 	if err != nil {
 		return nil, err
 	}
 	return StartDASRPCServerOnListener(ctx, listener, rpcServerTimeouts, rpcServerBodyLimit, daReader, daWriter, daHealthChecker, signatureVerifier)
 }
 
-func StartDASRPCServerOnListener(ctx context.Context, listener net.Listener, rpcServerTimeouts genericconf.HTTPServerTimeoutConfig, rpcServerBodyLimit int, daReader DataAvailabilityServiceReader, daWriter DataAvailabilityServiceWriter, daHealthChecker DataAvailabilityServiceHealthChecker, signatureVerifier *SignatureVerifier) (*http.Server, error) {
+func StartDASRPCServerOnListener(ctx context.Context, listener net.Listener, rpcServerTimeouts genericconf.HTTPServerTimeoutConfig, rpcServerBodyLimit int, daReader dasutil.DASReader, daWriter dasutil.DASWriter, daHealthChecker DataAvailabilityServiceHealthChecker, signatureVerifier *SignatureVerifier) (*http.Server, error) {
 	if daWriter == nil {
 		return nil, errors.New("No writer backend was configured for DAS RPC server. Has the BLS signing key been set up (--data-availability.key.key-dir or --data-availability.key.priv-key options)?")
 	}

daprovider/das/das_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/offchainlabs/nitro/daprovider/das/dasutil"
 	"github.com/offchainlabs/nitro/util/testhelpers"
 )
 
@@ -53,7 +54,7 @@ func testDASStoreRetrieveMultipleInstances(t *testing.T, storageType string) {
 	defer lifecycleManager.StopAndWaitUntil(time.Second)
 	daWriter, err := NewSignAfterStoreDASWriter(firstCtx, config, storageService)
 	Require(t, err, "no das")
-	var daReader DataAvailabilityServiceReader = storageService
+	var daReader dasutil.DASReader = storageService
 
 	// #nosec G115
 	timeout := uint64(time.Now().Add(time.Hour * 24).Unix())
@@ -80,7 +81,7 @@ func testDASStoreRetrieveMultipleInstances(t *testing.T, storageType string) {
 	storageService2, lifecycleManager, err := CreatePersistentStorageService(secondCtx, &config)
 	Require(t, err)
 	defer lifecycleManager.StopAndWaitUntil(time.Second)
-	var daReader2 DataAvailabilityServiceReader = storageService2
+	var daReader2 dasutil.DASReader = storageService2
 
 	messageRetrieved2, err := daReader2.GetByHash(secondCtx, cert.DataHash)
 	Require(t, err, "Failed to retrieve message")
@@ -144,7 +145,7 @@ func testDASMissingMessage(t *testing.T, storageType string) {
 	defer lifecycleManager.StopAndWaitUntil(time.Second)
 	daWriter, err := NewSignAfterStoreDASWriter(ctx, config, storageService)
 	Require(t, err, "no das")
-	var daReader DataAvailabilityServiceReader = storageService
+	var daReader dasutil.DASReader = storageService
 
 	messageSaved := []byte("hello world")
 	// #nosec G115

daprovider/das/dasserver/dasserver.go

@@ -79,8 +79,8 @@ func fetchJWTSecret(fileName string) ([]byte, error) {
 
 func NewServer(ctx context.Context, config *ServerConfig, dataSigner signature.DataSignerFunc, l1Client *ethclient.Client, l1Reader *headerreader.HeaderReader, sequencerInboxAddr common.Address) (*http.Server, func(), error) {
 	var err error
-	var daWriter das.DataAvailabilityServiceWriter
-	var daReader das.DataAvailabilityServiceReader
+	var daWriter dasutil.DASWriter
+	var daReader dasutil.DASReader
 	var dasKeysetFetcher *das.KeysetFetcher
 	var dasLifecycleManager *das.LifecycleManager
 	if config.EnableDAWriter {

daprovider/das/dasutil/dasutil.go

@@ -25,6 +25,7 @@ import (
 type DASReader interface {
 	GetByHash(ctx context.Context, hash common.Hash) ([]byte, error)
 	ExpirationPolicy(ctx context.Context) (ExpirationPolicy, error)
+	fmt.Stringer
 }
 
 type DASWriter interface {

daprovider/das/factory.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 
+	"github.com/offchainlabs/nitro/daprovider/das/dasutil"
 	"github.com/offchainlabs/nitro/solgen/go/bridgegen"
 	"github.com/offchainlabs/nitro/util/headerreader"
 	"github.com/offchainlabs/nitro/util/signature"
@@ -123,7 +124,7 @@ func CreateDAReaderAndWriter(
 	dataSigner signature.DataSignerFunc,
 	l1Reader *ethclient.Client,
 	sequencerInboxAddr common.Address,
-) (DataAvailabilityServiceWriter, DataAvailabilityServiceReader, *KeysetFetcher, *LifecycleManager, error) {
+) (dasutil.DASWriter, dasutil.DASReader, *KeysetFetcher, *LifecycleManager, error) {
 	if !config.Enable {
 		return nil, nil, nil, nil, nil
 	}
@@ -134,7 +135,7 @@ func CreateDAReaderAndWriter(
 	}
 	// Done checking config requirements
 
-	var daWriter DataAvailabilityServiceWriter
+	var daWriter dasutil.DASWriter
 	daWriter, err := NewRPCAggregator(ctx, *config, dataSigner)
 	if err != nil {
 		return nil, nil, nil, nil, err
@@ -147,7 +148,7 @@ func CreateDAReaderAndWriter(
 	restAgg.Start(ctx)
 	var lifecycleManager LifecycleManager
 	lifecycleManager.Register(restAgg)
-	var daReader DataAvailabilityServiceReader = restAgg
+	var daReader dasutil.DASReader = restAgg
 	keysetFetcher, err := NewKeysetFetcher(l1Reader, sequencerInboxAddr)
 	if err != nil {
 		return nil, nil, nil, nil, err
@@ -161,7 +162,7 @@ func CreateDAComponentsForDaserver(
 	config *DataAvailabilityConfig,
 	l1Reader *headerreader.HeaderReader,
 	seqInboxAddress *common.Address,
-) (DataAvailabilityServiceReader, DataAvailabilityServiceWriter, *SignatureVerifier, DataAvailabilityServiceHealthChecker, *LifecycleManager, error) {
+) (dasutil.DASReader, dasutil.DASWriter, *SignatureVerifier, DataAvailabilityServiceHealthChecker, *LifecycleManager, error) {
 	if !config.Enable {
 		return nil, nil, nil, nil, nil, nil
 	}
@@ -221,8 +222,8 @@ func CreateDAComponentsForDaserver(
 
 	}
 
-	var daWriter DataAvailabilityServiceWriter
-	var daReader DataAvailabilityServiceReader = storageService
+	var daWriter dasutil.DASWriter
+	var daReader dasutil.DASReader = storageService
 	var daHealthChecker DataAvailabilityServiceHealthChecker = storageService
 	var signatureVerifier *SignatureVerifier
 
@@ -262,7 +263,7 @@ func CreateDAReader(
 	config *DataAvailabilityConfig,
 	l1Reader *headerreader.HeaderReader,
 	seqInboxAddress *common.Address,
-) (DataAvailabilityServiceReader, *KeysetFetcher, *LifecycleManager, error) {
+) (dasutil.DASReader, *KeysetFetcher, *LifecycleManager, error) {
 	if !config.Enable {
 		return nil, nil, nil, nil
 	}
@@ -278,7 +279,7 @@ func CreateDAReader(
 	// Done checking config requirements
 
 	var lifecycleManager LifecycleManager
-	var daReader DataAvailabilityServiceReader
+	var daReader dasutil.DASReader
 	if config.RestAggregator.Enable {
 		var restAgg *SimpleDASReaderAggregator
 		restAgg, err := NewRestfulClientAggregator(ctx, &config.RestAggregator)

daprovider/das/panic_wrapper.go

@@ -15,41 +15,39 @@ import (
 )
 
 type WriterPanicWrapper struct {
-	DataAvailabilityServiceWriter
+	dasutil.DASWriter
 }
 
-func NewWriterPanicWrapper(dataAvailabilityService DataAvailabilityServiceWriter) DataAvailabilityServiceWriter {
-	return &WriterPanicWrapper{
-		DataAvailabilityServiceWriter: dataAvailabilityService,
-	}
+func NewWriterPanicWrapper(dataAvailabilityService dasutil.DASWriter) dasutil.DASWriter {
+	return &WriterPanicWrapper{DASWriter: dataAvailabilityService}
 }
 func (w *WriterPanicWrapper) String() string {
-	return fmt.Sprintf("WriterPanicWrapper{%v}", w.DataAvailabilityServiceWriter)
+	return fmt.Sprintf("WriterPanicWrapper{%v}", w.DASWriter)
 }
 
 func (w *WriterPanicWrapper) Store(ctx context.Context, message []byte, timeout uint64) (*dasutil.DataAvailabilityCertificate, error) {
-	cert, err := w.DataAvailabilityServiceWriter.Store(ctx, message, timeout)
+	cert, err := w.DASWriter.Store(ctx, message, timeout)
 	if err != nil {
 		panic(fmt.Sprintf("panic wrapper Store: %v", err))
 	}
 	return cert, nil
 }
 
 type ReaderPanicWrapper struct {
-	DataAvailabilityServiceReader
+	dasutil.DASReader
 }
 
-func NewReaderPanicWrapper(dataAvailabilityService DataAvailabilityServiceReader) DataAvailabilityServiceReader {
+func NewReaderPanicWrapper(dataAvailabilityService dasutil.DASReader) dasutil.DASReader {
 	return &ReaderPanicWrapper{
-		DataAvailabilityServiceReader: dataAvailabilityService,
+		DASReader: dataAvailabilityService,
 	}
 }
 func (w *ReaderPanicWrapper) String() string {
-	return fmt.Sprintf("ReaderPanicWrapper{%v}", w.DataAvailabilityServiceReader)
+	return fmt.Sprintf("ReaderPanicWrapper{%v}", w.DASReader)
 }
 
 func (w *ReaderPanicWrapper) GetByHash(ctx context.Context, hash common.Hash) ([]byte, error) {
-	data, err := w.DataAvailabilityServiceReader.GetByHash(ctx, hash)
+	data, err := w.DASReader.GetByHash(ctx, hash)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
 			log.Error("DAS hash lookup failed from cancelled context")

daprovider/das/reader_aggregator_strategies_test.go

@@ -19,6 +19,10 @@ type dummyReader struct {
 	int
 }
 
+func (*dummyReader) String() string {
+	return "dummyReader"
+}
+
 func (*dummyReader) GetByHash(context.Context, common.Hash) ([]byte, error) {
 	return nil, errors.New("not implemented")
 }

daprovider/das/restful_client.go

@@ -24,6 +24,10 @@ type RestfulDasClient struct {
 	url string
 }
 
+func (c *RestfulDasClient) String() string {
+	return fmt.Sprintf("Restful DAS client for %s", c.url)
+}
+
 func NewRestfulDasClient(protocol string, host string, port int) *RestfulDasClient {
 	return &RestfulDasClient{
 		url: fmt.Sprintf("%s://%s:%d", protocol, host, port),

daprovider/das/timeout_wrapper.go

@@ -9,21 +9,23 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/offchainlabs/nitro/daprovider/das/dasutil"
 )
 
 type ReaderTimeoutWrapper struct {
 	t time.Duration
-	DataAvailabilityServiceReader
+	dasutil.DASReader
 }
 
 type TimeoutWrapper struct {
 	ReaderTimeoutWrapper
 }
 
-func NewReaderTimeoutWrapper(dataAvailabilityServiceReader DataAvailabilityServiceReader, t time.Duration) DataAvailabilityServiceReader {
+func NewReaderTimeoutWrapper(dataAvailabilityServiceReader dasutil.DASReader, t time.Duration) dasutil.DASReader {
 	return &ReaderTimeoutWrapper{
-		t:                             t,
-		DataAvailabilityServiceReader: dataAvailabilityServiceReader,
+		t:         t,
+		DASReader: dataAvailabilityServiceReader,
 	}
 }
 
@@ -32,9 +34,9 @@ func (w *ReaderTimeoutWrapper) GetByHash(ctx context.Context, hash common.Hash)
 	// For GetByHash we want fast cancellation of all goroutines started by
 	// the aggregator as soon as one returns.
 	defer cancel()
-	return w.DataAvailabilityServiceReader.GetByHash(deadlineCtx, hash)
+	return w.DASReader.GetByHash(deadlineCtx, hash)
 }
 
 func (w *ReaderTimeoutWrapper) String() string {
-	return fmt.Sprintf("ReaderTimeoutWrapper{%v}", w.DataAvailabilityServiceReader)
+	return fmt.Sprintf("ReaderTimeoutWrapper{%v}", w.DASReader)
 }

system_tests/common_test.go

@@ -1880,8 +1880,8 @@ func setupConfigWithDAS(
 
 	l1NodeConfigA.DataAvailability = das.DefaultDataAvailabilityConfig
 	var lifecycleManager *das.LifecycleManager
-	var daReader das.DataAvailabilityServiceReader
-	var daWriter das.DataAvailabilityServiceWriter
+	var daReader dasutil.DASReader
+	var daWriter dasutil.DASWriter
 	var daHealthChecker das.DataAvailabilityServiceHealthChecker
 	var signatureVerifier *das.SignatureVerifier
 	if dasModeString != "onchain" {