Remove redundant DataStreamer retry mechanism (#3957)

* Remove internal daprovider server for AnyTrust

This change is a backport of 0195abd.
There were significant changes in this area that prevent it from being
applied cleanly but the idea is the same.

The internal daprovider server was originally added in PR #2533 to unify
the code paths between internal AnyTrust and external DA providers by
making both use RPC clients. After working with it for a while it's
clear that the drawbacks outweigh any benefit from it: it added an
unnecessary HTTP/RPC layer for in-process communication, introduced
timeout configuration complexity, and made retry behavior difficult to
reason about.

The proper abstraction point is the daprovider.Writer and Reader
interfaces, not the transport layer. External DA providers need RPC
because they're remote, but internal AnyTrust components can communicate
directly. This change removes the internal server entirely and connects
the Nitro node directly to the AnyTrust aggregator writer and REST
aggregator reader.

* Remove redundant DataStreamer retry mechanism

The DataStreamer had its own retry logic (5 attempts with exponential
backoff) on top of the underlying RPC client's retry mechanism (4
attempts with configurable retry patterns). This created a nested retry
system where:

- RPC client retries transient errors (timeouts, connection failures)
- DataStreamer retried ALL errors indiscriminately, including permanent
  ones like "method does not exist"

This redundancy caused problems in production:

1. Blocked fallback logic: When a DAS backend doesn't support
   chunked streaming, it returns "method does not exist". The
   DataStreamer would retry this error 5 times over 30 seconds instead
   of failing fast, preventing the intended fallback to legacy store
   from executing before the overall request timeout.

2. Excessive retry attempts: Each RPC method (start/chunk/finalize)
   could be retried up to 20 times (5 DataStreamer × 4 RPC client),
   wasting time and resources on non-retryable errors.

3. Error masking complexity: To properly handle permanent errors
   while keeping DataStreamer retries, we would need error filtering at
   TWO levels:
   - RPC client level (already handles transient vs permanent)
   - DataStreamer level (would need to duplicate this logic)

   This duplication adds complexity with no benefit since the RPC client
   already provides appropriate retry behavior.

The RPC client's retry mechanism is sufficient:
- Retries on context.DeadlineExceeded
- Retries on connection errors (configurable regex pattern)
- Immediately fails on application errors (e.g., "method does not exist")
- Configurable timeout and retry count per deployment

Removed CLI flags:
- --*.data-stream.base-retry-delay
- --*.data-stream.max-retry-delay
- --*.data-stream.max-retry-attempts

Author: Tristan-Wilson

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...)
 }