Skip to content

Commit

Permalink
feat: follower node sync from DA rebased to syncUpstream/active (#1013
Browse files Browse the repository at this point in the history
)

* feat: follower node sync from DA #631

 Conflicts:
	cmd/geth/main.go
	core/state_processor_test.go
	core/txpool/legacypool/legacypool.go
	eth/backend.go
	eth/ethconfig/config.go
	eth/gasprice/gasprice_test.go
	eth/handler.go
	eth/protocols/eth/broadcast.go
	eth/protocols/eth/handlers.go
	go.mod
	go.sum
	miner/miner.go
	miner/miner_test.go
	miner/scroll_worker.go
	miner/scroll_worker_test.go
	params/config.go
	params/version.go
	rollup/rollup_sync_service/rollup_sync_service_test.go

* fixes after merge

* fix: panic after startup due to misconfigured api

* fix bug of not calculating block state root

* minor adjustment of fix

* address review comments

* upgrade da-codec to scroll-tech/da-codec@41c6486

* fix go.mod after merge

* fix some tests

---------

Co-authored-by: Nazarii Denha <dengaaa2002@gmail.com>
Co-authored-by: 0xmountaintop <37070449+0xmountaintop@users.noreply.github.com>
  • Loading branch information
3 people authored Oct 16, 2024
1 parent f14ecc4 commit 8454491
Show file tree
Hide file tree
Showing 47 changed files with 2,600 additions and 184 deletions.
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

0 comments on commit 8454491

Please sign in to comment.