feat: follower node sync from DA

cmd/geth/main.go

@@ -170,6 +170,11 @@ var (
 		utils.CircuitCapacityCheckEnabledFlag,
 		utils.RollupVerifyEnabledFlag,
 		utils.ShadowforkPeersFlag,
+		utils.DASyncEnabledFlag,
+		utils.DAModeFlag,
+		utils.DASnapshotFileFlag,
+		utils.DABlockNativeAPIEndpointFlag,
+		utils.DABlobScanAPIEndpointFlag,
 	}
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -73,6 +73,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/p2p/nat"
 	"github.com/scroll-tech/go-ethereum/p2p/netutil"
 	"github.com/scroll-tech/go-ethereum/params"
+	"github.com/scroll-tech/go-ethereum/rollup/da_syncer"
 	"github.com/scroll-tech/go-ethereum/rollup/tracing"
 	"github.com/scroll-tech/go-ethereum/rpc"
 )
@@ -864,6 +865,31 @@ var (
 		Name:  "net.shadowforkpeers",
 		Usage: "peer ids of shadow fork peers",
 	}
+
+	// DA syncing settings
+	DASyncEnabledFlag = cli.BoolFlag{
+		Name:  "da.sync",
+		Usage: "Enable node syncing from DA",
+	}
+	defaultDA  = ethconfig.Defaults.DA.FetcherMode
+	DAModeFlag = TextMarshalerFlag{
+		Name:  "da.mode",
+		Usage: `DA sync mode ("l1rpc" or "snapshot")`,
+		Value: &defaultDA,
+	}
+	DASnapshotFileFlag = cli.StringFlag{
+		Name:  "da.snapshot.file",
+		Usage: "Snapshot file to sync from da",
+	}
+	DABlobScanAPIEndpointFlag = cli.StringFlag{
+		Name:  "da.blob.blobscan",
+		Usage: "BlobScan blob api endpoint",
+		Value: ethconfig.Defaults.DA.BlobScanAPIEndpoint,
+	}
+	DABlockNativeAPIEndpointFlag = cli.StringFlag{
+		Name:  "da.blob.blocknative",
+		Usage: "BlockNative blob api endpoint",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1307,6 +1333,9 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	setDataDir(ctx, cfg)
 	setSmartCard(ctx, cfg)
 	setL1(ctx, cfg)
+	if ctx.GlobalIsSet(DASyncEnabledFlag.Name) {
+		cfg.DaSyncingEnabled = ctx.GlobalBool(DASyncEnabledFlag.Name)
+	}
 
 	if ctx.GlobalIsSet(ExternalSignerFlag.Name) {
 		cfg.ExternalSigner = ctx.GlobalString(ExternalSignerFlag.Name)
@@ -1581,6 +1610,24 @@ func setEnableRollupVerify(ctx *cli.Context, cfg *ethconfig.Config) {
 	}
 }
 
+func setDA(ctx *cli.Context, cfg *ethconfig.Config) {
+	if ctx.GlobalIsSet(DASyncEnabledFlag.Name) {
+		cfg.EnableDASyncing = ctx.GlobalBool(DASyncEnabledFlag.Name)
+		if ctx.GlobalIsSet(DAModeFlag.Name) {
+			cfg.DA.FetcherMode = *GlobalTextMarshaler(ctx, DAModeFlag.Name).(*da_syncer.FetcherMode)
+		}
+		if ctx.GlobalIsSet(DASnapshotFileFlag.Name) {
+			cfg.DA.SnapshotFilePath = ctx.GlobalString(DASnapshotFileFlag.Name)
+		}
+		if ctx.GlobalIsSet(DABlobScanAPIEndpointFlag.Name) {
+			cfg.DA.BlobScanAPIEndpoint = ctx.GlobalString(DABlobScanAPIEndpointFlag.Name)
+		}
+		if ctx.GlobalIsSet(DABlockNativeAPIEndpointFlag.Name) {
+			cfg.DA.BlockNativeAPIEndpoint = ctx.GlobalString(DABlockNativeAPIEndpointFlag.Name)
+		}
+	}
+}
+
 func setMaxBlockRange(ctx *cli.Context, cfg *ethconfig.Config) {
 	if ctx.GlobalIsSet(MaxBlockRangeFlag.Name) {
 		cfg.MaxBlockRange = ctx.GlobalInt64(MaxBlockRangeFlag.Name)
@@ -1656,6 +1703,7 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 	setLes(ctx, cfg)
 	setCircuitCapacityCheck(ctx, cfg)
 	setEnableRollupVerify(ctx, cfg)
+	setDA(ctx, cfg)
 	setMaxBlockRange(ctx, cfg)
 	if ctx.GlobalIsSet(ShadowforkPeersFlag.Name) {
 		cfg.ShadowForkPeerIDs = ctx.GlobalStringSlice(ShadowforkPeersFlag.Name)

common/backoff/exponential.go

@@ -0,0 +1,51 @@
+package backoff
+
+import (
+	"math"
+	"math/rand"
+	"time"
+)
+
+// Exponential is a backoff strategy that increases the delay between retries exponentially.
+type Exponential struct {
+	attempt int
+
+	maxJitter time.Duration
+
+	min time.Duration
+	max time.Duration
+}
+
+func NewExponential(minimum, maximum, maxJitter time.Duration) *Exponential {
+	return &Exponential{
+		min:       minimum,
+		max:       maximum,
+		maxJitter: maxJitter,
+	}
+}
+
+func (e *Exponential) NextDuration() time.Duration {
+	var jitter time.Duration
+	if e.maxJitter > 0 {
+		jitter = time.Duration(rand.Int63n(e.maxJitter.Nanoseconds()))
+	}
+
+	minFloat := float64(e.min)
+	duration := math.Pow(2, float64(e.attempt)) * minFloat
+
+	// limit at configured maximum
+	if duration > float64(e.max) {
+		duration = float64(e.max)
+	}
+
+	e.attempt++
+	return time.Duration(duration) + jitter
+}
+
+func (e *Exponential) Reset() {
+	e.attempt = 0
+}
+
+func (e *Exponential) Attempt() int {
+	return e.attempt
+}

common/backoff/exponential_test.go

@@ -0,0 +1,39 @@
+package backoff
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestExponentialBackoff(t *testing.T) {
+	t.Run("Multiple attempts", func(t *testing.T) {
+		e := NewExponential(100*time.Millisecond, 10*time.Second, 0)
+		expectedDurations := []time.Duration{
+			100 * time.Millisecond,
+			200 * time.Millisecond,
+			400 * time.Millisecond,
+			800 * time.Millisecond,
+			1600 * time.Millisecond,
+			3200 * time.Millisecond,
+			6400 * time.Millisecond,
+			10 * time.Second, // capped at max
+		}
+		for i, expected := range expectedDurations {
+			require.Equal(t, expected, e.NextDuration(), "attempt %d", i)
+		}
+	})
+
+	t.Run("Jitter added", func(t *testing.T) {
+		e := NewExponential(1*time.Second, 10*time.Second, 1*time.Second)
+		duration := e.NextDuration()
+		require.GreaterOrEqual(t, duration, 1*time.Second)
+		require.Less(t, duration, 2*time.Second)
+	})
+
+	t.Run("Edge case: min > max", func(t *testing.T) {
+		e := NewExponential(10*time.Second, 5*time.Second, 0)
+		require.Equal(t, 5*time.Second, e.NextDuration())
+	})
+}

common/heap.go

@@ -0,0 +1,109 @@
+package common
+
+import (
+	"container/heap"
+)
+
+// Heap is a generic min-heap (or max-heap, depending on Comparable behavior) implementation.
+type Heap[T Comparable[T]] struct {
+	heap innerHeap[T]
+}
+
+func NewHeap[T Comparable[T]]() *Heap[T] {
+	return &Heap[T]{
+		heap: make(innerHeap[T], 0),
+	}
+}
+
+func (h *Heap[T]) Len() int {
+	return len(h.heap)
+}
+
+func (h *Heap[T]) Push(element T) *HeapElement[T] {
+	heapElement := NewHeapElement(element)
+	heap.Push(&h.heap, heapElement)
+
+	return heapElement
+}
+
+func (h *Heap[T]) Pop() *HeapElement[T] {
+	return heap.Pop(&h.heap).(*HeapElement[T])
+}
+
+func (h *Heap[T]) Peek() *HeapElement[T] {
+	if h.Len() == 0 {
+		return nil
+	}
+
+	return h.heap[0]
+}
+
+func (h *Heap[T]) Remove(element *HeapElement[T]) {
+	heap.Remove(&h.heap, element.index)
+}
+
+func (h *Heap[T]) Clear() {
+	h.heap = make(innerHeap[T], 0)
+}
+
+type innerHeap[T Comparable[T]] []*HeapElement[T]
+
+func (h innerHeap[T]) Len() int {
+	return len(h)
+}
+
+func (h innerHeap[T]) Less(i, j int) bool {
+	return h[i].Value().CompareTo(h[j].Value()) < 0
+}
+
+func (h innerHeap[T]) Swap(i, j int) {
+	h[i], h[j] = h[j], h[i]
+	h[i].index, h[j].index = i, j
+}
+
+func (h *innerHeap[T]) Push(x interface{}) {
+	data := x.(*HeapElement[T])
+	*h = append(*h, data)
+	data.index = len(*h) - 1
+}
+
+func (h *innerHeap[T]) Pop() interface{} {
+	n := len(*h)
+	element := (*h)[n-1]
+	(*h)[n-1] = nil // avoid memory leak
+	*h = (*h)[:n-1]
+	element.index = -1
+
+	return element
+}
+
+// Comparable is an interface for types that can be compared.
+type Comparable[T any] interface {
+	// CompareTo compares x with other.
+	// To create a min heap, return:
+	//	-1 if x <  other
+	//	 0 if x == other
+	//	+1 if x >  other
+	// To create a max heap, return the opposite.
+	CompareTo(other T) int
+}
+
+// HeapElement is a wrapper around the value stored in the heap.
+type HeapElement[T Comparable[T]] struct {
+	value T
+	index int
+}
+
+func NewHeapElement[T Comparable[T]](value T) *HeapElement[T] {
+	return &HeapElement[T]{
+		value: value,
+	}
+}
+
+func (h *HeapElement[T]) Value() T {
+	return h.value
+}
+
+func (h *HeapElement[T]) Index() int {
+	return h.index
+}

common/heap_test.go

@@ -0,0 +1,40 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type Int int
+
+func (i Int) CompareTo(other Int) int {
+	if i < other {
+		return -1
+	} else if i > other {
+		return 1
+	} else {
+		return 0
+	}
+}
+
+func TestHeap(t *testing.T) {
+	h := NewHeap[Int]()
+
+	require.Equal(t, 0, h.Len(), "Heap should be empty initially")
+
+	h.Push(Int(3))
+	h.Push(Int(1))
+	h.Push(Int(2))
+
+	require.Equal(t, 3, h.Len(), "Heap should have three elements after pushing")
+
+	require.EqualValues(t, 1, h.Pop(), "Pop should return the smallest element")
+	require.Equal(t, 2, h.Len(), "Heap should have two elements after popping")
+
+	require.EqualValues(t, 2, h.Pop(), "Pop should return the next smallest element")
+	require.Equal(t, 1, h.Len(), "Heap should have one element after popping")
+
+	require.EqualValues(t, 3, h.Pop(), "Pop should return the last element")
+	require.Equal(t, 0, h.Len(), "Heap should be empty after popping all elements")
+}