Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 12 commits into from
Oct 16, 2024
Merged
6 changes: 6 additions & 0 deletions cmd/geth/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down
58 changes: 56 additions & 2 deletions cmd/utils/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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 (
Expand Down Expand Up @@ -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 (
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
51 changes: 51 additions & 0 deletions common/backoff/exponential.go
Original file line number Diff line number Diff line change
@@ -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
}
39 changes: 39 additions & 0 deletions common/backoff/exponential_test.go
Original file line number Diff line number Diff line change
@@ -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())
})
}
109 changes: 109 additions & 0 deletions common/heap.go
Original file line number Diff line number Diff line change
@@ -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
}
40 changes: 40 additions & 0 deletions common/heap_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
Loading
Loading