feat: implement flashblocks monitoring

cmd/serve.go

@@ -22,6 +22,8 @@ const (
 func CommandServe(cfg *config.Config) *cli.Command {
 	l1RpcFallback := &cli.StringSlice{}
 	l1WalletAddresses := &cli.StringSlice{}
+	l2FlashblocksPrivateStreams := &cli.StringSlice{}
+	l2FlashblocksPublicStreams := &cli.StringSlice{}
 	l2RpcFallback := &cli.StringSlice{}
 	l2WalletAddresses := &cli.StringSlice{}
 
@@ -195,6 +197,39 @@ func CommandServe(cfg *config.Config) *cli.Command {
 			Value:       "incrementFlashblockNumber()",
 		},
 
+		&cli.Int64Flag{
+			Category:    strings.ToUpper(categoryL2),
+			Destination: &cfg.L2.MonitorFlashblocksMaxWsMessageSizeKb,
+			EnvVars:     []string{envPrefix + strings.ToUpper(categoryL2) + "_MONITOR_FLASHBLOCKS_MAX_WS_MESSAGE_SIZE_KB"},
+			Name:        categoryL2 + "-monitor-flashblocks-max-ws-message-size-kb",
+			Usage:       "max size (in kb) of l2 builder flashblocks ws messages",
+			Value:       256,
+		},
+
+		&cli.StringFlag{ // --l2-monitor-flashblocks-main-public-stream-name
+			Category:    strings.ToUpper(categoryL2),
+			Destination: &cfg.L2.MonitorFlashblocksMainPublicStreamName,
+			EnvVars:     []string{envPrefix + strings.ToUpper(categoryL2) + "_MONITOR_FLASHBLOCKS_MAIN_PUBLIC_STREAM"},
+			Name:        categoryL2 + "-monitor-flashblocks-main-public-stream",
+			Usage:       "the name of the main public l2 flashblocks stream",
+		},
+
+		&cli.StringSliceFlag{ // --l2-monitor-flashblocks-private-stream
+			Category:    strings.ToUpper(categoryL2),
+			Destination: l2FlashblocksPrivateStreams,
+			EnvVars:     []string{envPrefix + strings.ToUpper(categoryL2) + "_MONITOR_FLASHBLOCKS_PRIVATE_STREAMS"},
+			Name:        categoryL2 + "-monitor-flashblocks-private-stream",
+			Usage:       "private websocket stream(s) of l2 flashblocks",
+		},
+
+		&cli.StringSliceFlag{ // --l2-monitor-flashblocks-public-stream
+			Category:    strings.ToUpper(categoryL2),
+			Destination: l2FlashblocksPublicStreams,
+			EnvVars:     []string{envPrefix + strings.ToUpper(categoryL2) + "_MONITOR_FLASHBLOCKS_PUBLIC_STREAMS"},
+			Name:        categoryL2 + "-monitor-flashblocks-public-stream",
+			Usage:       "public websocket stream(s) of l2 flashblocks",
+		},
+
 		&cli.BoolFlag{ // --l2-monitor-tx-receipts
 			Category:    strings.ToUpper(categoryL2),
 			Destination: &cfg.L2.MonitorTxReceipts,
@@ -354,17 +389,47 @@ func CommandServe(cfg *config.Config) *cli.Command {
 			{
 				cfg.L2.RpcFallback = l2RpcFallback.Value()
 
-				_walletAddresses := make(map[string]string, len(l2WalletAddresses.Value()))
-				for _, wa := range l2WalletAddresses.Value() {
-					parts := strings.Split(wa, "=")
-					if len(parts) != 2 {
-						return fmt.Errorf("invalid wallet address (mush be like `name=0xNNNN`): %s", wa)
+				{
+					_flashblocksPrivateStreams := make(map[string]string, len(l2FlashblocksPrivateStreams.Value()))
+					for _, wa := range l2FlashblocksPrivateStreams.Value() {
+						parts := strings.Split(wa, "=")
+						if len(parts) != 2 {
+							return fmt.Errorf("invalid private flashblocks stream (must be like `name=ws://f.q.d.n:1111`): %s", wa)
+						}
+						name := strings.TrimSpace(parts[0])
+						url := strings.TrimSpace(parts[1])
+						_flashblocksPrivateStreams[name] = url
 					}
-					name := strings.TrimSpace(parts[0])
-					addr := strings.TrimSpace(parts[1])
-					_walletAddresses[name] = addr
+					cfg.L2.MonitorFlashblocksPrivateStreams = _flashblocksPrivateStreams
+				}
+
+				{
+					_flashblocksPublicStreams := make(map[string]string, len(l2FlashblocksPublicStreams.Value()))
+					for _, wa := range l2FlashblocksPublicStreams.Value() {
+						parts := strings.Split(wa, "=")
+						if len(parts) != 2 {
+							return fmt.Errorf("invalid public flashblocks stream (must be like `name=ws://f.q.d.n:1111`): %s", wa)
+						}
+						name := strings.TrimSpace(parts[0])
+						url := strings.TrimSpace(parts[1])
+						_flashblocksPublicStreams[name] = url
+					}
+					cfg.L2.MonitorFlashblocksPublicStreams = _flashblocksPublicStreams
+				}
+
+				{
+					_walletAddresses := make(map[string]string, len(l2WalletAddresses.Value()))
+					for _, wa := range l2WalletAddresses.Value() {
+						parts := strings.Split(wa, "=")
+						if len(parts) != 2 {
+							return fmt.Errorf("invalid wallet address (mush be like `name=0xNNNN`): %s", wa)
+						}
+						name := strings.TrimSpace(parts[0])
+						addr := strings.TrimSpace(parts[1])
+						_walletAddresses[name] = addr
+					}
+					cfg.L2.MonitorWalletAddresses = _walletAddresses
 				}
-				cfg.L2.MonitorWalletAddresses = _walletAddresses
 			}
 
 			return cfg.Validate()

config/l2.go

@@ -29,6 +29,10 @@ type L2 struct {
 	MonitorBuilderPolicyContractFunctionSignature    string            `yaml:"monitor_builder_policy_contract_function_signature"`
 	MonitorFlashblockNumberContract                  string            `yaml:"monitor_builder_flashblock_number_contract"`
 	MonitorFlashblockNumberContractFunctionSignature string            `yaml:"monitor_builder_flashblock_number_contract_function_signature"`
+	MonitorFlashblocksMainPublicStreamName           string            `yaml:"monitor_flashblocks_main_public_stream_name"`
+	MonitorFlashblocksMaxWsMessageSizeKb             int64             `yaml:"monitor_flashblocks_max_ws_message_size_kb"`
+	MonitorFlashblocksPrivateStreams                 map[string]string `yaml:"monitor_flashblocks_private_streams"`
+	MonitorFlashblocksPublicStreams                  map[string]string `yaml:"monitor_flashblocks_public_streams"`
 	MonitorFlashtestationRegistryContract            string            `yaml:"monitor_flashtestation_registry_contract"`
 	MonitorFlashtestationRegistryEventSignature      string            `yaml:"monitor_flashtestation_registry_event_signature"`
 	MonitorFlashtestationRegistryFunctionSignature   string            `yaml:"monitor_flashtestation_registry_function_signature"`
@@ -46,6 +50,10 @@ var (
 	errL2InvalidBuilderAddress                 = errors.New("invalid l2 builder address")
 	errL2InvalidBuilderPolicyContact           = errors.New("invalid l2 builder policy contract address")
 	errL2InvalidFlashblockNumberContact        = errors.New("invalid l2 flashblocks number contract address")
+	errL2InvalidFlashblocksMaxWsMessageSizeKb  = errors.New("invalid l2 flashblocks max ws message size")
+	errL2InvalidFlashblocksPrivateStream       = errors.New("invalid l2 private flashblocks stream url")
+	errL2InvalidFlashblocksPublicStream        = errors.New("invalid l2 public flashblocks stream url")
+	errL2InvalidFlashblocksPublicStreamName    = errors.New("invalid l2 public flashblocks stream name")
 	errL2InvalidFlashtestationsRegistryContact = errors.New("invalid l2 flashtestations registry contract address")
 	errL2InvalidRpc                            = errors.New("invalid l2 rpc url")
 	errL2InvalidRpcFallback                    = errors.New("invalid l2 fallback rpc url")
@@ -168,6 +176,50 @@ func (cfg *L2) Validate() error {
 		}
 	}
 
+	{ // monitor_flashblocks_max_ws_message_size_kb
+		if cfg.MonitorFlashblocksMaxWsMessageSizeKb < 16 || cfg.MonitorFlashblocksMaxWsMessageSizeKb > 1024 {
+			errs = append(errs, fmt.Errorf("%w: must be within [16..1024] range: %d",
+				errL2InvalidFlashblocksMaxWsMessageSizeKb,
+				cfg.MonitorFlashblocksMaxWsMessageSizeKb,
+			))
+		}
+	}
+
+	{ // monitor_flashblocks_main_public_stream_name
+		if cfg.MonitorFlashblocksMainPublicStreamName != "" {
+			if _, exists := cfg.MonitorFlashblocksPublicStreams[cfg.MonitorFlashblocksMainPublicStreamName]; !exists {
+				errs = append(errs, fmt.Errorf("%w: public stream name is not configured: %s",
+					errL2InvalidFlashblocksPublicStreamName,
+					cfg.MonitorFlashblocksMainPublicStreamName,
+				))
+			}
+		}
+	}
+
+	{ // monitor_flashblocks_private_streams
+		for _, flashblocks := range cfg.MonitorFlashblocksPrivateStreams {
+			if _, err := url.Parse(flashblocks); err != nil {
+				errs = append(errs, fmt.Errorf("%w: %s: %w",
+					errL2InvalidFlashblocksPrivateStream,
+					flashblocks,
+					err,
+				))
+			}
+		}
+	}
+
+	{ // monitor_flashblocks_public_streams
+		for _, flashblocks := range cfg.MonitorFlashblocksPublicStreams {
+			if _, err := url.Parse(flashblocks); err != nil {
+				errs = append(errs, fmt.Errorf("%w: %s: %w",
+					errL2InvalidFlashblocksPublicStream,
+					flashblocks,
+					err,
+				))
+			}
+		}
+	}
+
 	{ // monitor_wallet_address
 		for _, wa := range cfg.MonitorWalletAddresses {
 			_addr, err := ethcommon.ParseHexOrString(wa)

go.mod

@@ -3,79 +3,81 @@ module github.com/flashbots/chain-monitor
 go 1.24.0
 
 require (
-	github.com/ethereum/go-ethereum v1.16.1
+	github.com/andybalholm/brotli v1.2.0
+	github.com/coder/websocket v1.8.14
+	github.com/ethereum/go-ethereum v1.16.7
 	github.com/google/uuid v1.6.0
-	github.com/prometheus/client_golang v1.22.0
-	github.com/stretchr/testify v1.10.0
+	github.com/prometheus/client_golang v1.23.2
+	github.com/stretchr/testify v1.11.1
 	github.com/urfave/cli/v2 v2.27.7
-	go.opentelemetry.io/otel v1.37.0
-	go.opentelemetry.io/otel/exporters/prometheus v0.59.0
-	go.opentelemetry.io/otel/metric v1.37.0
-	go.opentelemetry.io/otel/sdk v1.37.0
-	go.opentelemetry.io/otel/sdk/metric v1.37.0
-	go.uber.org/zap v1.27.0
-	golang.org/x/crypto v0.40.0
+	go.opentelemetry.io/otel v1.38.0
+	go.opentelemetry.io/otel/exporters/prometheus v0.60.0
+	go.opentelemetry.io/otel/metric v1.38.0
+	go.opentelemetry.io/otel/sdk v1.38.0
+	go.opentelemetry.io/otel/sdk/metric v1.38.0
+	go.uber.org/zap v1.27.1
+	golang.org/x/crypto v0.45.0
 )
 
-replace github.com/ethereum/go-ethereum => github.com/ethereum-optimism/op-geth v1.101511.1
+replace github.com/ethereum/go-ethereum => github.com/ethereum-optimism/op-geth v1.101603.5
 
-replace github.com/ethereum/go-ethereum/common => github.com/ethereum-optimism/op-geth/common v1.101511.1
+replace github.com/ethereum/go-ethereum/common => github.com/ethereum-optimism/op-geth/common v1.101603.5
 
-replace github.com/ethereum/go-ethereum/core/types => github.com/ethereum-optimism/op-geth/core/types v1.101511.1
+replace github.com/ethereum/go-ethereum/core/types => github.com/ethereum-optimism/op-geth/core/types v1.101603.5
 
 require (
 	github.com/BurntSushi/toml v1.5.0 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/bits-and-blooms/bitset v1.22.0 // indirect
+	github.com/bits-and-blooms/bitset v1.24.4 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/cockroachdb/pebble v1.1.5 // indirect
-	github.com/consensys/gnark-crypto v0.18.0 // indirect
+	github.com/consensys/gnark-crypto v0.19.2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
-	github.com/crate-crypto/go-eth-kzg v1.3.0 // indirect
+	github.com/crate-crypto/go-eth-kzg v1.4.0 // indirect
 	github.com/crate-crypto/go-ipa v0.0.0-20240724233137-53bbb0ceb27a // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/deckarep/golang-set/v2 v2.8.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
-	github.com/ethereum/c-kzg-4844/v2 v2.1.1 // indirect
+	github.com/ethereum/c-kzg-4844/v2 v2.1.5 // indirect
 	github.com/ethereum/go-verkle v0.2.2 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/hashicorp/go-bexpr v0.1.14 // indirect
+	github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect
+	github.com/hashicorp/go-bexpr v0.1.15 // indirect
 	github.com/holiman/uint256 v1.3.2 // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
-	github.com/mattn/go-runewidth v0.0.16 // indirect
+	github.com/klauspost/compress v1.18.1 // indirect
+	github.com/mattn/go-runewidth v0.0.19 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/naoina/go-stringutil v0.1.0 // indirect
 	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416 // indirect
 	github.com/pion/dtls/v2 v2.2.12 // indirect
 	github.com/pion/transport/v2 v2.2.10 // indirect
-	github.com/pion/transport/v3 v3.0.7 // indirect
+	github.com/pion/transport/v3 v3.1.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.65.0 // indirect
-	github.com/prometheus/procfs v0.17.0 // indirect
-	github.com/rivo/uniseg v0.4.7 // indirect
+	github.com/prometheus/common v0.67.4 // indirect
+	github.com/prometheus/otlptranslator v0.0.2 // indirect
+	github.com/prometheus/procfs v0.19.2 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
-	github.com/supranational/blst v0.3.15 // indirect
+	github.com/supranational/blst v0.3.16 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220614013038-64ee5596c38a // indirect
-	github.com/tklauser/go-sysconf v0.3.15 // indirect
-	github.com/tklauser/numcpus v0.10.0 // indirect
-	github.com/wlynxg/anet v0.0.5 // indirect
+	github.com/tklauser/go-sysconf v0.3.16 // indirect
+	github.com/tklauser/numcpus v0.11.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
-	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/otel/trace v1.37.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/exp v0.0.0-20250711185948-6ae5c78190dc // indirect
-	golang.org/x/sync v0.16.0 // indirect
-	golang.org/x/sys v0.34.0 // indirect
-	golang.org/x/time v0.12.0 // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
+	go.yaml.in/yaml/v2 v2.4.3 // indirect
+	golang.org/x/exp v0.0.0-20251125195548-87e1e737ad39 // indirect
+	golang.org/x/sync v0.18.0 // indirect
+	golang.org/x/sys v0.38.0 // indirect
+	golang.org/x/time v0.14.0 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )