Merge pull request #520 from OffchainLabs/add-timeout-to-das-store

Add timeout parameter to DAS Store interface

Author: Tristan-Wilson

arbnode/batch_poster.go

@@ -47,6 +47,7 @@ type BatchPosterConfig struct {
 	BatchPollDelay       time.Duration `koanf:"poll-delay"`
 	PostingErrorDelay    time.Duration `koanf:"error-delay"`
 	CompressionLevel     int           `koanf:"compression-level"`
+	DASRetentionPeriod   time.Duration `koanf:"das-retention-period"`
 }
 
 func BatchPosterConfigAddOptions(prefix string, f *flag.FlagSet) {
@@ -56,6 +57,7 @@ func BatchPosterConfigAddOptions(prefix string, f *flag.FlagSet) {
 	f.Duration(prefix+".poll-delay", DefaultBatchPosterConfig.BatchPollDelay, "how long to delay after successfully posting batch")
 	f.Duration(prefix+".error-delay", DefaultBatchPosterConfig.PostingErrorDelay, "how long to delay after error posting batch")
 	f.Int(prefix+".compression-level", DefaultBatchPosterConfig.CompressionLevel, "batch compression level")
+	f.Duration(prefix+".das-retention-period", DefaultBatchPosterConfig.DASRetentionPeriod, "In AnyTrust mode, the period which DASes are requested to retain the stored batches.")
 }
 
 var DefaultBatchPosterConfig = BatchPosterConfig{
@@ -65,6 +67,7 @@ var DefaultBatchPosterConfig = BatchPosterConfig{
 	PostingErrorDelay:    time.Second * 10,
 	MaxBatchPostInterval: time.Hour,
 	CompressionLevel:     brotli.DefaultCompression,
+	DASRetentionPeriod:   time.Hour * 24 * 15,
 }
 
 var TestBatchPosterConfig = BatchPosterConfig{
@@ -74,6 +77,7 @@ var TestBatchPosterConfig = BatchPosterConfig{
 	PostingErrorDelay:    time.Millisecond * 10,
 	MaxBatchPostInterval: 0,
 	CompressionLevel:     2,
+	DASRetentionPeriod:   time.Hour * 24 * 15,
 }
 
 func NewBatchPoster(l1Reader *L1Reader, inbox *InboxTracker, streamer *TransactionStreamer, config *BatchPosterConfig, contractAddress common.Address, refunder common.Address, transactOpts *bind.TransactOpts, das das.DataAvailabilityService) (*BatchPoster, error) {
@@ -369,7 +373,7 @@ func (b *BatchPoster) maybePostSequencerBatch(ctx context.Context, timeSinceBatc
 	}
 
 	if b.das != nil {
-		cert, err := b.das.Store(ctx, sequencerMsg)
+		cert, err := b.das.Store(ctx, sequencerMsg, uint64(time.Now().Add(b.config.DASRetentionPeriod).Unix()))
 		if err != nil {
 			log.Warn("Unable to batch to DAS, falling back to storing data on chain", "err", err)
 		} else {

das/das.go

@@ -15,7 +15,8 @@ import (
 )
 
 type DataAvailabilityServiceWriter interface {
-	Store(ctx context.Context, message []byte) (*arbstate.DataAvailabilityCertificate, error)
+	// Requests that the message be stored until timeout (UTC time in unix epoch seconds).
+	Store(ctx context.Context, message []byte, timeout uint64) (*arbstate.DataAvailabilityCertificate, error)
 }
 
 type DataAvailabilityService interface {

das/das_test.go

@@ -6,9 +6,11 @@ package das
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/offchainlabs/nitro/util/testhelpers"
 )
@@ -23,9 +25,13 @@ func TestDASStoreRetrieveMultipleInstances(t *testing.T) {
 
 	ctx := context.Background()
 
+	timeout := uint64(time.Now().Add(time.Hour * 24).Unix())
 	messageSaved := []byte("hello world")
-	cert, err := das.Store(ctx, messageSaved)
+	cert, err := das.Store(ctx, messageSaved, timeout)
 	Require(t, err, "Error storing message")
+	if cert.Timeout != timeout {
+		Fail(t, fmt.Sprintf("Expected timeout of %d in cert, was %d", timeout, cert.Timeout))
+	}
 
 	certBytes := Serialize(*cert)
 
@@ -57,8 +63,12 @@ func TestDASMissingMessage(t *testing.T) {
 	ctx := context.Background()
 
 	messageSaved := []byte("hello world")
-	cert, err := das.Store(ctx, messageSaved)
+	timeout := uint64(time.Now().Add(time.Hour * 24).Unix())
+	cert, err := das.Store(ctx, messageSaved, timeout)
 	Require(t, err, "Error storing message")
+	if cert.Timeout != timeout {
+		Fail(t, fmt.Sprintf("Expected timeout of %d in cert, was %d", timeout, cert.Timeout))
+	}
 
 	// Change the hash to look up
 	cert.DataHash[0] += 1

das/dasrpc/dasRpcServer.go

@@ -44,7 +44,7 @@ func (serv *DASRPCServer) Stop() {
 }
 
 func (serv *DASRPCServer) Store(ctx context.Context, req *StoreRequest) (*StoreResponse, error) {
-	cert, err := serv.localDAS.Store(ctx, req.Message)
+	cert, err := serv.localDAS.Store(ctx, req.Message, req.Timeout)
 	if err != nil {
 		return nil, err
 	}

das/local_disk_das.go

@@ -9,7 +9,6 @@ import (
 	"errors"
 	"os"
 	"sync"
-	"time"
 
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
@@ -20,11 +19,10 @@ import (
 var dasMutex sync.Mutex
 
 type LocalDiskDataAvailabilityService struct {
-	dbPath          string
-	pubKey          *blsSignatures.PublicKey
-	privKey         blsSignatures.PrivateKey
-	retentionPeriod time.Duration
-	signerMask      uint64
+	dbPath     string
+	pubKey     *blsSignatures.PublicKey
+	privKey    blsSignatures.PrivateKey
+	signerMask uint64
 }
 
 func readKeysFromFile(dbPath string) (*blsSignatures.PublicKey, blsSignatures.PrivateKey, error) {
@@ -89,14 +87,14 @@ func NewLocalDiskDataAvailabilityService(dbPath string) (*LocalDiskDataAvailabil
 	}, nil
 }
 
-func (das *LocalDiskDataAvailabilityService) Store(ctx context.Context, message []byte) (c *arbstate.DataAvailabilityCertificate, err error) {
+func (das *LocalDiskDataAvailabilityService) Store(ctx context.Context, message []byte, timeout uint64) (c *arbstate.DataAvailabilityCertificate, err error) {
 	dasMutex.Lock()
 	defer dasMutex.Unlock()
 
 	c = &arbstate.DataAvailabilityCertificate{}
 	copy(c.DataHash[:], crypto.Keccak256(message))
 
-	c.Timeout = uint64(time.Now().Add(das.retentionPeriod).Unix())
+	c.Timeout = timeout
 	c.SignersMask = das.signerMask
 
 	fields := serializeSignableFields(*c)

das/wireFormat.proto

@@ -13,6 +13,7 @@ service DASServiceImpl {
 
 message StoreRequest {
   bytes message = 1;
+  uint64 timeout = 2;
 }
 
 message StoreResponse {