feat(bench): add feeder-sim, a feeder gateway simulator for sync benchmarks

Makefile

@@ -47,6 +47,10 @@ corpus-gen: ## Build the corpus generator
 	@mkdir -p build
 	@go build -o build/corpus-gen ./bench/rpc/cmd/corpus-gen
 
+feeder-sim: ## Build the feeder gateway simulator
+	@mkdir -p build
+	@go build -o build/feeder-sim ./bench/sync/cmd/feeder-sim
+
 MINIMUM_RUST_VERSION = 1.94.1
 CURR_RUST_VERSION = $(shell rustc --version | grep -o '[0-9.]\+' | head -n1)
 check-rust: ## Ensure rust version is greater than minimum

bench/sync/README.md

@@ -0,0 +1,69 @@
+# Sync benchmark: feeder-sim
+
+Captures feeder gateway (FGW) responses for a block range, then serves them from memory behind a
+simulated chain tip. Removes FGW latency, rate limits and chain activity from sync benchmarks.
+
+## Use
+
+```sh
+make feeder-sim                                                                   # -> build/feeder-sim
+build/feeder-sim --data ./data/mainnet --network mainnet --from 1500000 --to 1501000 --listen off  # capture
+build/feeder-sim --data ./data/mainnet --from 1500000 --to 1501000                # serve, 2s cadence
+build/feeder-sim --data ./data/mainnet --from 1500000 --to 1501000 --speed 1      # serve, captured block times
+build/feeder-sim --data ./data/mainnet --from 1500000 --to 1501000 --tip 1501000  # serve everything at once
+```
+
+Capture is resumable. A 400 from the FGW usually means `--to` is above the tip. One `--data`
+directory per network, checked via `get_contract_addresses`. Without `--network` the sim is offline and fails on a missing file; with
+`--network` it fills gaps before serving.
+
+## Flags
+
+| Flag                | Meaning                                                                    | Default          |
+| ------------------- | -------------------------------------------------------------------------- | ---------------- |
+| `--data`            | dataset directory (required)                                               |                  |
+| `--from`, `--to`    | block range, inclusive (required)                                          |                  |
+| `--network`         | capture source: `mainnet`, `sepolia`, `sepolia-integration`                | offline          |
+| `--listen`          | `host:port`; `:7070` binds all interfaces; `off` = capture only            | `127.0.0.1:7070` |
+| `--tip`             | initial tip, in `[from, to]`                                               | `from`           |
+| `--interval`        | advance the tip by one block every interval                                | `2s`             |
+| `--speed`           | replay captured block timestamps at this multiplier; excludes `--interval` | unset            |
+| `--latency`         | fixed delay added to every response, no jitter                             | `0`              |
+| `--api-key`         | `X-Throttling-Bypass` header during capture                                |                  |
+| `--concurrency`     | parallel capture requests                                                  | `8`              |
+| `--capture-timeout` | per-request timeout during capture                                         | `30s`            |
+| `--capture-retries` | retries per request during capture                                         | `5`              |
+| `--log-level`       | `debug`, `info`, `warn`, `error`                                           | `info`           |
+
+`blockNumber=latest` resolves to the tip. Blocks above the tip get 400 (debug log). Classes are not tip-gated.
+
+## Juno
+
+```sh
+juno --db-path <copy of a DB synced to from-1> \
+  --cn-name mainnet --cn-feeder-url http://127.0.0.1:7070/feeder_gateway/ \
+  --cn-gateway-url http://127.0.0.1:7070/gateway/ \
+  --cn-l2-chain-id SN_MAIN --cn-l1-chain-id 1 \
+  --cn-core-contract-address 0xc662c410c0ecf747543f5ba90660f6abebd9c8c4 \
+  --cn-unverifiable-range 0,0 --preconfirmed-poll-interval 0 --metrics
+```
+
+- Restore the DB copy before every run. To prepare a DB at `X-1`, run against the sim with `--tip X-1`.
+- Keep the range at Starknet v0.13.2 or later; older blocks fail hash verification on custom networks.
+- Juno retries a 400 ten times with backoff, so wall clock between blocks measures the retry
+  schedule. Measure `sync_step_duration` instead, or use `--tip <to>` for throughput.
+
+## Dataset layout
+
+One gzipped file per FGW response:
+
+```
+<data>/
+  get_contract_addresses.json.gz
+  get_block/<N>.json.gz                           # headerOnly=true
+  get_state_update/<N>.json.gz                    # includeBlock=true&includeSignature=true
+  get_class_by_hash/<hash>.json.gz                # blockNumber=latest
+  get_compiled_class_by_class_hash/<hash>.json.gz # blockNumber=latest
+```
+
+Timestamps, class lists and completeness are derived from the state updates at startup.

bench/sync/cmd/feeder-sim/clock.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/NethermindEth/juno/utils/log"
+	"go.uber.org/zap"
+)
+
+type clock struct {
+	current  atomic.Uint64
+	from     uint64
+	to       uint64
+	interval time.Duration
+	speed    float64
+	blocks   []blockInfo
+	logger   *log.ZapLogger
+}
+
+func newClock(blocks []blockInfo, config *config, logger *log.ZapLogger) *clock {
+	clock := &clock{
+		from:     config.from,
+		to:       config.to,
+		interval: config.interval,
+		speed:    config.speed,
+		blocks:   blocks,
+		logger:   logger,
+	}
+	clock.current.Store(config.tip)
+	return clock
+}
+
+func (clock *clock) tip() uint64 {
+	return clock.current.Load()
+}
+
+func (clock *clock) run(ctx context.Context) {
+	if clock.speed > 0 {
+		clock.logger.Info("replaying captured block times", zap.Float64("speed", clock.speed))
+	} else {
+		clock.logger.Info("advancing tip on a fixed interval", zap.Duration("interval", clock.interval))
+	}
+
+	for clock.tip() < clock.to {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(clock.nextDelay()):
+			clock.logger.Info("tip advanced", zap.Uint64("tip", clock.current.Add(1)))
+		}
+	}
+
+	clock.logger.Info("tip reached --to; everything is served", zap.Uint64("tip", clock.to))
+}
+
+func (clock *clock) nextDelay() time.Duration {
+	if clock.speed == 0 {
+		return clock.interval
+	}
+	tip := clock.tip()
+	current := clock.blocks[tip-clock.from].timestamp
+	next := clock.blocks[tip+1-clock.from].timestamp
+
+	if next <= current {
+		return 0
+	}
+	return time.Duration(float64(next-current) * float64(time.Second) / clock.speed)
+}

bench/sync/cmd/feeder-sim/dataset.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+)
+
+const directoryMode = 0o750
+
+type dataset struct {
+	root string
+}
+
+func (dataset dataset) path(file string) string {
+	return filepath.Join(dataset.root, file)
+}
+
+func (dataset dataset) read(file string) ([]byte, error) {
+	return os.ReadFile(dataset.path(file))
+}
+
+func (dataset dataset) write(file string, gzipped []byte) error {
+	target := dataset.path(file)
+	if err := os.MkdirAll(filepath.Dir(target), directoryMode); err != nil {
+		return err
+	}
+
+	temporary, err := writeTemporary(target, gzipped)
+	if err != nil {
+		return err
+	}
+	return os.Rename(temporary, target)
+}
+
+func writeTemporary(target string, data []byte) (name string, err error) {
+	file, err := os.CreateTemp(filepath.Dir(target), filepath.Base(target)+".*.tmp")
+	if err != nil {
+		return "", err
+	}
+	defer func() {
+		err = errors.Join(err, file.Close())
+		if err != nil {
+			err = errors.Join(err, os.Remove(file.Name()))
+		}
+	}()
+
+	if err = writeAndSync(file, data); err != nil {
+		return "", err
+	}
+	return file.Name(), nil
+}
+
+func writeAndSync(file *os.File, data []byte) error {
+	if _, err := file.Write(data); err != nil {
+		return err
+	}
+	return file.Sync()
+}
+
+func gzipBytes(body []byte) ([]byte, error) {
+	var buffer bytes.Buffer
+	writer := gzip.NewWriter(&buffer)
+	if _, err := writer.Write(body); err != nil {
+		return nil, err
+	}
+	if err := writer.Close(); err != nil {
+		return nil, err
+	}
+	return buffer.Bytes(), nil
+}
+
+func gunzip(gzipped []byte) ([]byte, error) {
+	reader, err := gzip.NewReader(bytes.NewReader(gzipped))
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := io.ReadAll(reader)
+	if err != nil {
+		return nil, err
+	}
+	return body, reader.Close()
+}
+
+func unmarshalGzipped[T any](gzipped []byte) (T, error) {
+	var value T
+	body, err := gunzip(gzipped)
+	if err != nil {
+		return value, err
+	}
+	err = json.Unmarshal(body, &value)
+	return value, err
+}

bench/sync/cmd/feeder-sim/decode.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+
+	"github.com/NethermindEth/juno/l1/eth"
+	"github.com/NethermindEth/juno/starknet"
+)
+
+type contractAddressesResponse struct {
+	Starknet eth.Address `json:"Starknet"`
+}
+
+func coreContractAddress(gzipped []byte) (eth.Address, error) {
+	response, err := unmarshalGzipped[contractAddressesResponse](gzipped)
+	if err != nil {
+		return eth.Address{}, fmt.Errorf("%s: %w", contractAddresses.name, err)
+	}
+	return response.Starknet, nil
+}
+
+func hexAddress(address eth.Address) string {
+	return fmt.Sprintf("0x%x", address.Bytes())
+}
+
+type blockInfo struct {
+	number      uint64
+	timestamp   uint64
+	classHashes []string
+}
+
+func newBlockInfo(gzipped []byte, number uint64) (blockInfo, error) {
+	response, err := decodeStateUpdateResponse(gzipped)
+	if err != nil {
+		return blockInfo{}, fmt.Errorf("%s %d: %w", stateUpdate.name, number, err)
+	}
+	return blockInfo{
+		number:      response.Block.Number,
+		timestamp:   response.Block.Timestamp,
+		classHashes: classHashes(&response.StateUpdate.StateDiff),
+	}, nil
+}
+
+func decodeStateUpdateResponse(gzipped []byte) (*starknet.StateUpdateWithBlockAndSignature, error) {
+	response, err := unmarshalGzipped[starknet.StateUpdateWithBlockAndSignature](gzipped)
+	if err != nil {
+		return nil, err
+	}
+	if response.Block == nil || response.StateUpdate == nil {
+		return nil, errors.New("state update response lacks block or state_update")
+	}
+	return &response, nil
+}
+
+func classHashes(diff *starknet.StateDiff) []string {
+	size := len(diff.DeployedContracts) + len(diff.OldDeclaredContracts) + len(diff.DeclaredClasses)
+	hashes := make([]string, 0, size)
+
+	for _, deployed := range diff.DeployedContracts {
+		hashes = append(hashes, deployed.ClassHash.String())
+	}
+	for _, hash := range diff.OldDeclaredContracts {
+		hashes = append(hashes, hash.String())
+	}
+	for _, declared := range diff.DeclaredClasses {
+		hashes = append(hashes, declared.ClassHash.String())
+	}
+	return hashes
+}
+
+func uniqueClassHashes(blocks []blockInfo) []string {
+	set := make(map[string]struct{})
+	for _, info := range blocks {
+		for _, hash := range info.classHashes {
+			set[hash] = struct{}{}
+		}
+	}
+	return slices.Sorted(maps.Keys(set))
+}
+
+func isSierra(classGzipped []byte, hash string) (bool, error) {
+	class, err := unmarshalGzipped[starknet.ClassDefinition](classGzipped)
+	if err != nil {
+		return false, fmt.Errorf("%s %s: %w", classByHash.name, hash, err)
+	}
+	return class.Sierra != nil, nil
+}