Merge pull request #140 from Layr-Labs/kira--rebase-v3.8.0

 Merges upstream v3.8.0 (from v3.8.0-rc.16)

Author: shyam-patel-kira

arbnode/inbox_tracker.go

@@ -318,8 +318,7 @@ func (t *InboxTracker) PopulateFeedBacklog(broadcastServer *broadcaster.Broadcas
 		}
 		feedMessages = append(feedMessages, feedMessage)
 	}
-	broadcastServer.BroadcastFeedMessages(feedMessages)
-	return nil
+	return broadcastServer.PopulateFeedBacklog(feedMessages)
 }
 
 func (t *InboxTracker) legacyGetDelayedMessageAndAccumulator(ctx context.Context, seqNum uint64) (*arbostypes.L1IncomingMessage, common.Hash, error) {

arbnode/node.go

@@ -9,9 +9,6 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
-	"os"
-	"path"
-	"path/filepath"
 	"strings"
 
 	"github.com/spf13/pflag"
@@ -36,12 +33,11 @@ import (
 	"github.com/offchainlabs/nitro/broadcastclients"
 	"github.com/offchainlabs/nitro/broadcaster"
 	"github.com/offchainlabs/nitro/cmd/chaininfo"
-	"github.com/offchainlabs/nitro/cmd/genericconf"
 	"github.com/offchainlabs/nitro/daprovider"
 	"github.com/offchainlabs/nitro/daprovider/daclient"
 	"github.com/offchainlabs/nitro/daprovider/das"
 	"github.com/offchainlabs/nitro/daprovider/data_streaming"
-	dapserver "github.com/offchainlabs/nitro/daprovider/server"
+	"github.com/offchainlabs/nitro/daprovider/factory"
 	"github.com/offchainlabs/nitro/eigenda"
 	"github.com/offchainlabs/nitro/execution"
 	"github.com/offchainlabs/nitro/execution/gethexec"
@@ -593,6 +589,8 @@ func getDAS(
 
 	var err error
 	var daClient *daclient.Client
+	var anytrustWriter daprovider.Writer
+	var anytrustReader daprovider.Reader
 	var withDAWriter bool
 	var dasServerCloseFn func()
 	if config.DAProvider.Enable {
@@ -603,43 +601,49 @@ func getDAS(
 		// Only allow dawriter if batchposter is enabled
 		withDAWriter = config.DAProvider.WithWriter && config.BatchPoster.Enable
 	} else if config.DataAvailability.Enable {
-		jwtPath := path.Join(filepath.Dir(stack.InstanceDir()), "dasserver-jwtsecret")
-		if err := genericconf.TryCreatingJWTSecret(jwtPath); err != nil {
-			return nil, nil, nil, nil, fmt.Errorf("error writing ephemeral jwtsecret of dasserver to file: %w", err)
+		// Create AnyTrust factory
+		daFactory, err := factory.NewDAProviderFactory(
+			factory.ModeAnyTrust,
+			&config.DataAvailability,
+			nil, // referencedaCfg
+			dataSigner,
+			l1client,
+			l1Reader,
+			deployInfo.SequencerInbox,
+			config.BatchPoster.Enable,
+		)
+		if err != nil {
+			return nil, nil, nil, nil, err
 		}
-		log.Info("Generated ephemeral JWT secret for dasserver", "jwtPath", jwtPath)
-		// JWTSecret is no longer needed, cleanup when returning
-		defer func() {
-			if err := os.Remove(jwtPath); err != nil {
-				log.Error("error deleting generated ephemeral JWT secret of dasserver", "jwtPath", jwtPath)
-			}
-		}()
 
-		serverConfig := dapserver.DefaultDASServerConfig
-		serverConfig.Port = 0 // Initializes server at a random available port
-		serverConfig.DataAvailability = config.DataAvailability
-		serverConfig.EnableDAWriter = config.BatchPoster.Enable
-		serverConfig.JWTSecret = jwtPath
-		withDAWriter = config.BatchPoster.Enable
-		dasServer, closeFn, err := dapserver.NewServerForDAS(ctx, &serverConfig, dataSigner, l1client, l1Reader, deployInfo.SequencerInbox)
-		if err != nil {
+		if err := daFactory.ValidateConfig(); err != nil {
 			return nil, nil, nil, nil, err
 		}
-		rpcClientConfig := rpcclient.DefaultClientConfig
-		rpcClientConfig.URL = dasServer.Addr
-		rpcClientConfig.JWTSecret = jwtPath
 
-		daClientConfig := config.DAProvider
-		daClientConfig.RPC = rpcClientConfig
+		// Create writer if batch poster is enabled
+		var writerCleanup func()
+		if config.BatchPoster.Enable {
+			anytrustWriter, writerCleanup, err = daFactory.CreateWriter(ctx)
+			if err != nil {
+				return nil, nil, nil, nil, err
+			}
+			withDAWriter = true
+		}
 
-		daClient, err = daclient.NewClient(ctx, &daClientConfig, data_streaming.PayloadCommiter())
+		// Create reader
+		var readerCleanup func()
+		anytrustReader, readerCleanup, err = daFactory.CreateReader(ctx)
 		if err != nil {
 			return nil, nil, nil, nil, err
 		}
+
+		// Set up cleanup function
 		dasServerCloseFn = func() {
-			_ = dasServer.Shutdown(ctx)
-			if closeFn != nil {
-				closeFn()
+			if writerCleanup != nil {
+				writerCleanup()
+			}
+			if readerCleanup != nil {
+				readerCleanup()
 			}
 		}
 	} else if l2Config.ArbitrumChainParams.DataAvailabilityCommittee {
@@ -651,7 +655,7 @@ func getDAS(
 	}
 
 	// We support a nil txStreamer for the pruning code
-	if txStreamer != nil && txStreamer.chainConfig.ArbitrumChainParams.DataAvailabilityCommittee && daClient == nil {
+	if txStreamer != nil && txStreamer.chainConfig.ArbitrumChainParams.DataAvailabilityCommittee && daClient == nil && anytrustReader == nil {
 		return nil, nil, nil, nil, errors.New("data availability service required but unconfigured")
 	}
 
@@ -679,15 +683,26 @@ func getDAS(
 			return nil, nil, nil, nil, fmt.Errorf("failed to register DA client: %w", err)
 		}
 	}
+	if anytrustReader != nil {
+		headerBytes := []byte{
+			daprovider.DASMessageHeaderFlag,
+			daprovider.DASMessageHeaderFlag | daprovider.TreeDASMessageHeaderFlag,
+		}
+		if err := dapReaders.RegisterAll(headerBytes, anytrustReader); err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("failed to register AnyTrust reader: %w", err)
+		}
+	}
 	if blobReader != nil {
 		if err := dapReaders.SetupBlobReader(daprovider.NewReaderForBlobReader(blobReader)); err != nil {
 			return nil, nil, nil, nil, fmt.Errorf("failed to register blob reader: %w", err)
 		}
 	}
-	// AnyTrust now always uses the daClient, which is already registered,
-	// so we don't need to register it separately here.
 
 	if withDAWriter {
+		// Return anytrustWriter if it exists, otherwise daClient
+		if anytrustWriter != nil {
+			return anytrustWriter, dasServerCloseFn, dapReaders, eigenDAWriter, nil
+		}
 		return daClient, dasServerCloseFn, dapReaders, eigenDAWriter, nil
 	}
 	return nil, dasServerCloseFn, dapReaders, eigenDAWriter, nil

broadcaster/broadcaster.go

@@ -128,6 +128,14 @@ func (b *Broadcaster) BroadcastFeedMessages(messages []*m.BroadcastFeedMessage)
 	b.server.Broadcast(bm)
 }
 
+func (s *Broadcaster) PopulateFeedBacklog(messages []*m.BroadcastFeedMessage) error {
+	bm := &m.BroadcastMessage{
+		Version:  1,
+		Messages: messages,
+	}
+	return s.server.PopulateFeedBacklog(bm)
+}
+
 func (b *Broadcaster) Confirm(msgIdx arbutil.MessageIndex) {
 	log.Debug("confirming msgIdx", "msgIdx", msgIdx)
 	b.server.Broadcast(&m.BroadcastMessage{

cmd/chaininfo/arbitrum_chain_info.json

@@ -254,7 +254,7 @@
       "bridge": "0x38f918D0E9F1b721EDaA41302E399fa1B79333a9",
       "inbox": "0xaAe29B0366299461418F5324a79Afc425BE5ae21",
       "sequencer-inbox": "0x6c97864CE4bEf387dE0b3310A44230f7E3F1be0D",
-      "rollup": "0xd80810638dbDF9081b72C1B33c65375e807281C8",
+      "rollup": "0x042B2E6C5E99d4c521bd49beeD5E99651D9B0Cf4",
       "validator-utils": "0x1f6860C3cac255fFFa72B7410b1183c3a0D261e0",
       "validator-wallet-creator": "0x894fC71fA0A666352824EC954B401573C861D664",
       "stake-token": "0xefb383126640fe4a760010c6e59c397d2b6c7141",

daprovider/data_streaming/protocol_test.go

@@ -5,7 +5,6 @@ package data_streaming
 
 import (
 	"context"
-	"errors"
 	"math/rand"
 	"net"
 	"net/http"
@@ -169,23 +168,6 @@ func TestDataStreaming_ProtocolSucceedsEvenWithDelays(t *testing.T) {
 	require.Equal(t, message, ([]byte)(result.Message), "protocol resulted in an incorrect message")
 }
 
-func TestDataStreaming_ClientRetriesWhenThereAreConnectionProblems(t *testing.T) {
-	// Server 'goes offline' for a moment just before reading the second chunk
-	var alreadyWentOffline = false
-	ctx, streamer := prepareTestEnv(t, func(i uint64) error {
-		if i == 1 && !alreadyWentOffline {
-			alreadyWentOffline = true
-			return errors.New("service unavailable")
-		}
-		return nil
-
-	})
-	message, _ := getLongRandomMessage(streamer.chunkSize)
-	result, err := streamer.StreamData(ctx, message, timeout)
-	testhelpers.RequireImpl(t, err)
-	require.Equal(t, message, ([]byte)(result.Message), "protocol resulted in an incorrect message")
-}
-
 func TestDataStreaming_ServerDeniesTooOldAndFutureRequests(t *testing.T) {
 	ctx, streamer := prepareTestEnv(t, nil)
 	message, _ := getLongRandomMessage(streamer.chunkSize)

daprovider/data_streaming/sender.go

@@ -19,60 +19,42 @@ import (
 )
 
 const (
-	DefaultHttpBodyLimit    = 5 * 1024 * 1024 // Taken from go-ethereum http.defaultBodyLimit
-	TestHttpBodyLimit       = 1024
-	DefaultBaseRetryDelay   = 2 * time.Second
-	DefaultMaxRetryDelay    = 1 * time.Minute
-	DefaultMaxRetryAttempts = 5
+	DefaultHttpBodyLimit = 5 * 1024 * 1024 // Taken from go-ethereum http.defaultBodyLimit
+	TestHttpBodyLimit    = 1024
 )
 
 // lint:require-exhaustive-initialization
 type DataStreamerConfig struct {
 	MaxStoreChunkBodySize int                     `koanf:"max-store-chunk-body-size"`
 	RpcMethods            DataStreamingRPCMethods `koanf:"rpc-methods"`
-
-	// Retry policy for RPC calls
-	BaseRetryDelay   time.Duration `koanf:"base-retry-delay"`
-	MaxRetryDelay    time.Duration `koanf:"max-retry-delay"`
-	MaxRetryAttempts int           `koanf:"max-retry-attempts"`
 }
 
 func DefaultDataStreamerConfig(rpcMethods DataStreamingRPCMethods) DataStreamerConfig {
 	return DataStreamerConfig{
 		MaxStoreChunkBodySize: DefaultHttpBodyLimit,
 		RpcMethods:            rpcMethods,
-		BaseRetryDelay:        DefaultBaseRetryDelay,
-		MaxRetryDelay:         DefaultMaxRetryDelay,
-		MaxRetryAttempts:      DefaultMaxRetryAttempts,
 	}
 }
 
 func TestDataStreamerConfig(rpcMethods DataStreamingRPCMethods) DataStreamerConfig {
 	return DataStreamerConfig{
 		MaxStoreChunkBodySize: TestHttpBodyLimit,
 		RpcMethods:            rpcMethods,
-		BaseRetryDelay:        100 * time.Millisecond,
-		MaxRetryDelay:         100 * time.Millisecond,
-		MaxRetryAttempts:      3,
 	}
 }
 
 func DataStreamerConfigAddOptions(prefix string, f *pflag.FlagSet, defaultRpcMethods DataStreamingRPCMethods) {
 	f.Int(prefix+".max-store-chunk-body-size", DefaultHttpBodyLimit, "maximum HTTP body size for chunked store requests")
-	f.Duration(prefix+".base-retry-delay", DefaultBaseRetryDelay, "base delay for retrying failed RPC calls")
-	f.Duration(prefix+".max-retry-delay", DefaultMaxRetryDelay, "maximum delay for retrying failed RPC calls")
-	f.Int(prefix+".max-retry-attempts", DefaultMaxRetryAttempts, "maximum number of attempts for retrying failed RPC calls")
 	DataStreamingRPCMethodsAddOptions(prefix+".rpc-methods", f, defaultRpcMethods)
 }
 
 // DataStreamer allows sending arbitrarily big payloads with JSON RPC. It follows a simple chunk-based protocol.
 // lint:require-exhaustive-initialization
 type DataStreamer[Result any] struct {
-	rpcClient        *rpcclient.RpcClient
-	chunkSize        uint64
-	dataSigner       *PayloadSigner
-	rpcMethods       DataStreamingRPCMethods
-	retryDelayPolicy *expDelayPolicy
+	rpcClient  *rpcclient.RpcClient
+	chunkSize  uint64
+	dataSigner *PayloadSigner
+	rpcMethods DataStreamingRPCMethods
 }
 
 // DataStreamingRPCMethods configuration specifies names of the protocol's RPC methods on the server side.
@@ -104,11 +86,6 @@ func NewDataStreamer[T any](config DataStreamerConfig, dataSigner *PayloadSigner
 		chunkSize:  chunkSize,
 		dataSigner: dataSigner,
 		rpcMethods: config.RpcMethods,
-		retryDelayPolicy: &expDelayPolicy{
-			baseDelay:   config.BaseRetryDelay,
-			maxDelay:    config.MaxRetryDelay,
-			maxAttempts: config.MaxRetryAttempts,
-		},
 	}, nil
 }
 
@@ -144,7 +121,7 @@ func (ds *DataStreamer[Result]) startStream(ctx context.Context, params streamPa
 	}
 
 	var result StartStreamingResult
-	err = ds.call(
+	err = ds.rpcClient.CallContext(
 		ctx,
 		&result,
 		ds.rpcMethods.StartStream,
@@ -172,51 +149,18 @@ func (ds *DataStreamer[Result]) sendChunk(ctx context.Context, messageId Message
 	if err != nil {
 		return err
 	}
-	return ds.call(ctx, nil, ds.rpcMethods.StreamChunk, hexutil.Uint64(messageId), hexutil.Uint64(chunkId), hexutil.Bytes(chunkData), hexutil.Bytes(payloadSignature))
+	return ds.rpcClient.CallContext(ctx, nil, ds.rpcMethods.StreamChunk, hexutil.Uint64(messageId), hexutil.Uint64(chunkId), hexutil.Bytes(chunkData), hexutil.Bytes(payloadSignature))
 }
 
 func (ds *DataStreamer[Result]) finalizeStream(ctx context.Context, messageId MessageId) (result *Result, err error) {
 	payloadSignature, err := ds.sign(nil, uint64(messageId))
 	if err != nil {
 		return nil, err
 	}
-	err = ds.call(ctx, &result, ds.rpcMethods.FinalizeStream, hexutil.Uint64(messageId), hexutil.Bytes(payloadSignature))
+	err = ds.rpcClient.CallContext(ctx, &result, ds.rpcMethods.FinalizeStream, hexutil.Uint64(messageId), hexutil.Bytes(payloadSignature))
 	return
 }
 
-type expDelayPolicy struct {
-	baseDelay, maxDelay time.Duration
-	maxAttempts         int
-}
-
-func (e *expDelayPolicy) NextDelay(attempt int) (time.Duration, bool) {
-	if attempt >= e.maxAttempts {
-		return 0, false
-	}
-	if attempt <= 0 {
-		return time.Duration(0), true
-	}
-
-	delay := e.baseDelay * time.Duration(1<<uint(attempt-1)) // nolint:gosec
-	if delay > e.maxDelay {
-		delay = e.maxDelay
-	}
-	return delay, true
-}
-
-func (ds *DataStreamer[Result]) call(ctx context.Context, result interface{}, method string, args ...interface{}) (err error) {
-	for attempt := 1; ; attempt++ {
-		if err = ds.rpcClient.CallContext(ctx, result, method, args...); err == nil {
-			return nil
-		}
-		delay, proceed := ds.retryDelayPolicy.NextDelay(attempt)
-		if !proceed {
-			return fmt.Errorf("failed after %d attempts: %w", attempt, err)
-		}
-		time.Sleep(delay)
-	}
-}
-
 func (ds *DataStreamer[Result]) sign(bytes []byte, extras ...uint64) ([]byte, error) {
 	return ds.dataSigner.signPayload(bytes, extras...)
 }

execution/gethexec/sync_monitor.go

@@ -278,9 +278,6 @@ func (s *SyncMonitor) SetFinalityData(
 	finalizedFinalityData *arbutil.FinalityData,
 	validatedFinalityData *arbutil.FinalityData,
 ) error {
-	s.exec.createBlocksMutex.Lock()
-	defer s.exec.createBlocksMutex.Unlock()
-
 	finalizedBlockHeader, err := s.getFinalityBlockHeader(
 		s.config.FinalizedBlockWaitForBlockValidator,
 		validatedFinalityData,

wsbroadcastserver/clientmanager.go

@@ -151,6 +151,31 @@ func (cm *ClientManager) Broadcast(bm *m.BroadcastMessage) {
 	cm.broadcastChan <- bm
 }
 
+// populateFeedBacklog adds the given BroadcastMessage to backlog, and also broadcasts it to feed if the ClientManager is started
+func (cm *ClientManager) populateFeedBacklog(bm *m.BroadcastMessage) error {
+	if cm.Started() {
+		cm.Broadcast(bm)
+		return nil
+	}
+	if len(bm.Messages) == 0 {
+		return cm.backlog.Append(bm)
+	}
+	for i, msg := range bm.Messages {
+		m := &m.BroadcastMessage{
+			Version:  bm.Version,
+			Messages: []*m.BroadcastFeedMessage{msg},
+		}
+		// This ensures that only one message is added to backlog with the confirmed sequence number
+		if i == 0 {
+			m.ConfirmedSequenceNumberMessage = bm.ConfirmedSequenceNumberMessage
+		}
+		if err := cm.backlog.Append(m); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (cm *ClientManager) doBroadcast(bm *m.BroadcastMessage) ([]*ClientConnection, error) {
 	if err := cm.backlog.Append(bm); err != nil {
 		return nil, err