feat: follower node sync from DA rebased to syncUpstream/active

cmd/geth/main.go

@@ -154,6 +154,12 @@ var (
 		utils.CircuitCapacityCheckWorkersFlag,
 		utils.RollupVerifyEnabledFlag,
 		utils.ShadowforkPeersFlag,
+		utils.DASyncEnabledFlag,
+		utils.DAModeFlag,
+		utils.DASnapshotFileFlag,
+		utils.DABlockNativeAPIEndpointFlag,
+		utils.DABlobScanAPIEndpointFlag,
+		utils.DABeaconNodeAPIEndpointFlag,
 	}, utils.NetworkFlags, utils.DatabaseFlags)
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -36,6 +36,9 @@ import (
 	"time"
 
 	pcsclite "github.com/gballet/go-libpcsclite"
+	gopsutil "github.com/shirou/gopsutil/mem"
+	"github.com/urfave/cli/v2"
+
 	"github.com/scroll-tech/go-ethereum/accounts"
 	"github.com/scroll-tech/go-ethereum/accounts/keystore"
 	"github.com/scroll-tech/go-ethereum/common"
@@ -73,13 +76,12 @@ 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"
 	"github.com/scroll-tech/go-ethereum/trie"
 	"github.com/scroll-tech/go-ethereum/trie/triedb/hashdb"
 	"github.com/scroll-tech/go-ethereum/trie/triedb/pathdb"
-	gopsutil "github.com/shirou/gopsutil/mem"
-	"github.com/urfave/cli/v2"
 )
 
 const (
@@ -1024,6 +1026,33 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		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",
+	}
+	DAModeFlag = &flags.TextMarshalerFlag{
+		Name:  "da.mode",
+		Usage: `DA sync mode ("l1rpc" or "snapshot")`,
+		Value: &ethconfig.Defaults.DA.FetcherMode,
+	}
+	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",
+	}
+	DABlockNativeAPIEndpointFlag = &cli.StringFlag{
+		Name:  "da.blob.blocknative",
+		Usage: "BlockNative blob API endpoint",
+	}
+	DABeaconNodeAPIEndpointFlag = &cli.StringFlag{
+		Name:  "da.blob.beaconnode",
+		Usage: "Beacon node API endpoint",
+	}
 )
 
 var (
@@ -1512,6 +1541,9 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	SetDataDir(ctx, cfg)
 	setSmartCard(ctx, cfg)
 	setL1(ctx, cfg)
+	if ctx.IsSet(DASyncEnabledFlag.Name) {
+		cfg.DaSyncingEnabled = ctx.Bool(DASyncEnabledFlag.Name)
+	}
 
 	if ctx.IsSet(JWTSecretFlag.Name) {
 		cfg.JWTSecret = ctx.String(JWTSecretFlag.Name)
@@ -1766,6 +1798,27 @@ func setEnableRollupVerify(ctx *cli.Context, cfg *ethconfig.Config) {
 	}
 }
 
+func setDA(ctx *cli.Context, cfg *ethconfig.Config) {
+	if ctx.IsSet(DASyncEnabledFlag.Name) {
+		cfg.EnableDASyncing = ctx.Bool(DASyncEnabledFlag.Name)
+		if ctx.IsSet(DAModeFlag.Name) {
+			cfg.DA.FetcherMode = *flags.GlobalTextMarshaler(ctx, DAModeFlag.Name).(*da_syncer.FetcherMode)
+		}
+		if ctx.IsSet(DASnapshotFileFlag.Name) {
+			cfg.DA.SnapshotFilePath = ctx.String(DASnapshotFileFlag.Name)
+		}
+		if ctx.IsSet(DABlobScanAPIEndpointFlag.Name) {
+			cfg.DA.BlobScanAPIEndpoint = ctx.String(DABlobScanAPIEndpointFlag.Name)
+		}
+		if ctx.IsSet(DABlockNativeAPIEndpointFlag.Name) {
+			cfg.DA.BlockNativeAPIEndpoint = ctx.String(DABlockNativeAPIEndpointFlag.Name)
+		}
+		if ctx.IsSet(DABeaconNodeAPIEndpointFlag.Name) {
+			cfg.DA.BeaconNodeAPIEndpoint = ctx.String(DABeaconNodeAPIEndpointFlag.Name)
+		}
+	}
+}
+
 func setMaxBlockRange(ctx *cli.Context, cfg *ethconfig.Config) {
 	if ctx.IsSet(MaxBlockRangeFlag.Name) {
 		cfg.MaxBlockRange = ctx.Int64(MaxBlockRangeFlag.Name)
@@ -1831,6 +1884,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.IsSet(ShadowforkPeersFlag.Name) {
 		cfg.ShadowForkPeerIDs = ctx.StringSlice(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")
+}