From 0af30915b4f8c90c7f13652a78edd3c00394cec0 Mon Sep 17 00:00:00 2001 From: Samuel Stokes Date: Fri, 9 Jan 2026 13:00:57 -0500 Subject: [PATCH 1/3] sysgo: allow OS to assign ports for opreth and rbuilder --- op-devstack/sysgo/l2_el_opreth.go | 9 +- op-devstack/sysgo/op_rbuilder.go | 152 ++++++++++++++++++------------ op-devstack/sysgo/util.go | 19 ++++ 3 files changed, 114 insertions(+), 66 deletions(-) diff --git a/op-devstack/sysgo/l2_el_opreth.go b/op-devstack/sysgo/l2_el_opreth.go index 3377ba09209..7a74d62f37a 100644 --- a/op-devstack/sysgo/l2_el_opreth.go +++ b/op-devstack/sysgo/l2_el_opreth.go @@ -265,12 +265,9 @@ func WithOpReth(id stack.L2ELNodeID, opts ...L2ELOption) stack.Option[*Orchestra } if areMetricsEnabled() { - // NB: Instead of getAvailableLocalPort, we should pass "0" so the OS picks its - // own port, but that is not currently logged properly so we cannot parse it. - // See: https://github.com/op-rs/op-reth/issues/333 - metricsPort, err := getAvailableLocalPort() - p.Require().NoError(err, "WithOpReth: getting metrics port") - args = append(args, "--metrics="+metricsPort) + // Use port 0 to let the OS assign a port atomically at bind time. + // The actual port will be discovered by parsing the process logs. + args = append(args, "--metrics=127.0.0.1:0") } if supervisorRPC != "" { diff --git a/op-devstack/sysgo/op_rbuilder.go b/op-devstack/sysgo/op_rbuilder.go index 93d4235ffa9..677567aa473 100644 --- a/op-devstack/sysgo/op_rbuilder.go +++ b/op-devstack/sysgo/op_rbuilder.go @@ -4,13 +4,11 @@ package sysgo import ( "encoding/hex" "encoding/json" - "net" "os" "path/filepath" "strconv" "strings" "sync" - "time" "github.com/ethereum/go-ethereum/log" @@ -20,6 +18,7 @@ import ( "github.com/ethereum-optimism/optimism/op-node/rollup" "github.com/ethereum-optimism/optimism/op-service/client" "github.com/ethereum-optimism/optimism/op-service/logpipe" + "github.com/ethereum-optimism/optimism/op-service/tasks" "github.com/ethereum-optimism/optimism/op-service/testutils/tcpproxy" ) @@ -142,16 +141,16 @@ func (cfg *OPRBuilderNodeConfig) LaunchSpec(p devtest.P) (args []string, env []s if cfg.FlashblocksAddr == "" { cfg.FlashblocksAddr = "127.0.0.1" } - if cfg.FlashblocksPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate flashblocks port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse flashblocks port") - cfg.FlashblocksPort = portVal - } - fbPortStr := strconv.Itoa(cfg.FlashblocksPort) args = append(args, "--flashblocks.enabled") - args = append(args, "--flashblocks.addr="+cfg.FlashblocksAddr, "--flashblocks.port="+fbPortStr) + args = append(args, "--flashblocks.addr="+cfg.FlashblocksAddr) + if cfg.FlashblocksPort > 0 { + // Use explicitly configured port + args = append(args, "--flashblocks.port="+strconv.Itoa(cfg.FlashblocksPort)) + } else { + // Use port 0 to let the OS assign a port atomically at bind time. + // The actual port will be discovered by parsing the process logs. + args = append(args, "--flashblocks.port=0") + } } // P2P configuration: enforce deterministic identity and static peering to the sequencer EL. @@ -176,30 +175,28 @@ func (cfg *OPRBuilderNodeConfig) LaunchSpec(p devtest.P) (args []string, env []s if cfg.EnableRPC { args = append(args, "--http") args = append(args, "--http.addr="+cfg.RPCAddr) - if cfg.RPCPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate rpc port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse rpc port") - cfg.RPCPort = portVal + if cfg.RPCPort > 0 { + // Use explicitly configured port + args = append(args, "--http.port="+strconv.Itoa(cfg.RPCPort)) + } else { + // Use port 0 to let the OS assign a port atomically at bind time. + // The actual port will be discovered by parsing the process logs. + args = append(args, "--http.port=0") } - rpcPortStr := strconv.Itoa(cfg.RPCPort) - args = append(args, "--http.port="+rpcPortStr) args = append(args, "--http.api="+cfg.RPCAPI) - } if cfg.AuthRPCAddr != "" { args = append(args, "--authrpc.addr="+cfg.AuthRPCAddr) } - if cfg.AuthRPCPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate auth rpc port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse auth rpc port") - cfg.AuthRPCPort = portVal + if cfg.AuthRPCPort > 0 { + // Use explicitly configured port + args = append(args, "--authrpc.port="+strconv.Itoa(cfg.AuthRPCPort)) + } else { + // Use port 0 to let the OS assign a port atomically at bind time. + // The actual port will be discovered by parsing the process logs. + args = append(args, "--authrpc.port=0") } - args = append(args, "--authrpc.port="+strconv.Itoa(cfg.AuthRPCPort)) if cfg.AuthRPCJWTPath != "" { args = append(args, "--authrpc.jwtsecret="+cfg.AuthRPCJWTPath) } @@ -217,22 +214,16 @@ func (cfg *OPRBuilderNodeConfig) LaunchSpec(p devtest.P) (args []string, env []s if cfg.Chain != "" { args = append(args, "--chain="+cfg.Chain) } - if cfg.WithUnusedPorts { - args = append(args, "--with-unused-ports") - } if cfg.DisableDiscovery { args = append(args, "--disable-discovery") } - if !cfg.WithUnusedPorts { - if cfg.P2PPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate p2p port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse p2p port") - cfg.P2PPort = portVal - } + if cfg.P2PPort > 0 { + // Use explicitly configured P2P port args = append(args, "--port="+strconv.Itoa(cfg.P2PPort)) + } else { + // Use --with-unused-ports to let reth assign P2P port atomically at bind time. + args = append(args, "--with-unused-ports") } if cfg.DataDir == "" { @@ -365,12 +356,60 @@ func (b *OPRBuilderNode) Start() { args, env := cfg.LaunchSpec(b.p) - // Forward structured logs to Go logger + // Create channels for discovering ports from process logs. + // When using port 0, the OS assigns ports at bind time and the process logs them. + flashblocksWSChan := make(chan string, 1) + httpRPCChan := make(chan string, 1) + authRPCChan := make(chan string, 1) + defer close(flashblocksWSChan) + defer close(httpRPCChan) + defer close(authRPCChan) + + // Forward structured logs to Go logger and parse for port discovery logOut := logpipe.ToLogger(b.logger.New("component", "op-OPRbuilderNode", "src", "stdout")) logErr := logpipe.ToLogger(b.logger.New("component", "op-OPRbuilderNode", "src", "stderr")) + // Log parsing callback to extract bound addresses from process output + onLogEntry := func(e logpipe.LogEntry) { + msg := e.LogMessage() + // Flashblocks WS - custom log message from wspub.rs + if strings.HasPrefix(msg, "Flashblocks WebSocketPublisher listening on ") { + addr := strings.TrimPrefix(msg, "Flashblocks WebSocketPublisher listening on ") + if validURL := parseAndValidateAddr(addr, "ws"); validURL != "" { + select { + case flashblocksWSChan <- validURL: + default: + } + } + } + // HTTP RPC - standard reth log message + if msg == "RPC HTTP server started" { + if addr, ok := e.FieldValue("url").(string); ok { + if validURL := parseAndValidateAddr(addr, "http"); validURL != "" { + select { + case httpRPCChan <- validURL: + default: + } + } + } + } + // Auth RPC - standard reth log message + if msg == "RPC auth server started" { + if addr, ok := e.FieldValue("url").(string); ok { + if validURL := parseAndValidateAddr(addr, "http"); validURL != "" { + select { + case authRPCChan <- validURL: + default: + } + } + } + } + } + stdOut := logpipe.LogCallback(func(line []byte) { - logOut(logpipe.ParseRustStructuredLogs(line)) + e := logpipe.ParseRustStructuredLogs(line) + logOut(e) + onLogEntry(e) }) stdErr := logpipe.LogCallback(func(line []byte) { logErr(logpipe.ParseRustStructuredLogs(line)) @@ -389,33 +428,26 @@ func (b *OPRBuilderNode) Start() { err = b.sub.Start(execPath, args, env) b.p.Require().NoError(err, "start OPRBuilderNode") - const readinessTimeout = 15 * time.Second - + // Wait for ports to be discovered from logs, then configure proxies if cfg.EnableRPC { - rpcUpstreamHostport := net.JoinHostPort(cfg.RPCAddr, strconv.Itoa(cfg.RPCPort)) - rpcUpstreamURL := "http://" + rpcUpstreamHostport - waitTCPReady(b.p, rpcUpstreamURL, readinessTimeout) - b.logger.Info("OPRBuilderNode upstream RPC ready", "rpc", rpcUpstreamURL) - b.rpcProxy.SetUpstream(ProxyAddr(b.p.Require(), rpcUpstreamURL)) - waitTCPReady(b.p, b.rpcProxyURL, readinessTimeout) + var httpRPCAddr string + b.p.Require().NoError(tasks.Await(b.p.Ctx(), httpRPCChan, &httpRPCAddr), "need HTTP RPC address from logs") + b.logger.Info("OPRBuilderNode upstream RPC ready", "rpc", httpRPCAddr) + b.rpcProxy.SetUpstream(ProxyAddr(b.p.Require(), httpRPCAddr)) b.logger.Info("OPRBuilderNode proxy RPC ready", "proxy_rpc", b.rpcProxyURL) - authUpstreamHostport := net.JoinHostPort(cfg.RPCAddr, strconv.Itoa(cfg.AuthRPCPort)) - authUpstreamURL := "http://" + authUpstreamHostport - waitTCPReady(b.p, authUpstreamURL, readinessTimeout) - b.logger.Info("OPRBuilderNode upstream auth RPC ready", "auth_rpc", authUpstreamURL) - b.authProxy.SetUpstream(ProxyAddr(b.p.Require(), authUpstreamURL)) - waitTCPReady(b.p, b.authProxyURL, readinessTimeout) + var authRPCAddr string + b.p.Require().NoError(tasks.Await(b.p.Ctx(), authRPCChan, &authRPCAddr), "need Auth RPC address from logs") + b.logger.Info("OPRBuilderNode upstream auth RPC ready", "auth_rpc", authRPCAddr) + b.authProxy.SetUpstream(ProxyAddr(b.p.Require(), authRPCAddr)) b.logger.Info("OPRBuilderNode proxy auth RPC ready", "proxy_auth_rpc", b.authProxyURL) } if cfg.EnableFlashblocks { - wsUpstreamHostport := net.JoinHostPort(cfg.FlashblocksAddr, strconv.Itoa(cfg.FlashblocksPort)) - wsUpstreamURL := "ws://" + wsUpstreamHostport - waitWSReady(b.p, wsUpstreamURL, readinessTimeout) - b.logger.Info("OPRBuilderNode upstream WS ready", "ws", wsUpstreamURL) - b.wsProxy.SetUpstream(ProxyAddr(b.p.Require(), wsUpstreamURL)) - waitWSReady(b.p, b.wsProxyURL, readinessTimeout) + var flashblocksAddr string + b.p.Require().NoError(tasks.Await(b.p.Ctx(), flashblocksWSChan, &flashblocksAddr), "need Flashblocks WS address from logs") + b.logger.Info("OPRBuilderNode upstream WS ready", "ws", flashblocksAddr) + b.wsProxy.SetUpstream(ProxyAddr(b.p.Require(), flashblocksAddr)) b.logger.Info("OPRBuilderNode proxy WS ready", "proxy_ws", b.wsProxyURL) } } diff --git a/op-devstack/sysgo/util.go b/op-devstack/sysgo/util.go index b3903c26d69..b26d422a85e 100644 --- a/op-devstack/sysgo/util.go +++ b/op-devstack/sysgo/util.go @@ -8,6 +8,7 @@ import ( "net/url" "os" "strconv" + "strings" "sync" "time" @@ -105,3 +106,21 @@ func waitWSReady(p devtest.P, rawURL string, timeout time.Duration) { assert.NoError(c, err, "WebSocket handshake to %s should succeed", rawURL) }, timeout, 100*time.Millisecond, waitWSMsg) } + +// parseAndValidateAddr ensures the address has a scheme and is a valid URL. +// Returns the validated URL string or empty string if invalid. +// This is used to parse addresses from process (e.g. op-rbuilder) log output. +func parseAndValidateAddr(addr, defaultScheme string) string { + if addr == "" { + return "" + } + // Add scheme if not present + if !strings.Contains(addr, "://") { + addr = defaultScheme + "://" + addr + } + u, err := url.Parse(addr) + if err != nil || u.Host == "" || u.Hostname() == "" { + return "" + } + return u.String() +} From 246a8aa8557f8e3269f019f94865cdd657a33c14 Mon Sep 17 00:00:00 2001 From: Samuel Stokes Date: Fri, 9 Jan 2026 14:31:15 -0500 Subject: [PATCH 2/3] sysgo: allow OS to assign ports for rollup-boost --- op-devstack/sysgo/rollup_boost.go | 77 ++++++++++++++++++++----------- 1 file changed, 49 insertions(+), 28 deletions(-) diff --git a/op-devstack/sysgo/rollup_boost.go b/op-devstack/sysgo/rollup_boost.go index f7bffc40273..bbbee9a1f4c 100644 --- a/op-devstack/sysgo/rollup_boost.go +++ b/op-devstack/sysgo/rollup_boost.go @@ -1,7 +1,6 @@ package sysgo import ( - "net" "net/http" "strconv" "strings" @@ -15,6 +14,7 @@ import ( "github.com/ethereum-optimism/optimism/op-devstack/stack" "github.com/ethereum-optimism/optimism/op-service/client" "github.com/ethereum-optimism/optimism/op-service/logpipe" + "github.com/ethereum-optimism/optimism/op-service/tasks" "github.com/ethereum-optimism/optimism/op-service/testutils/tcpproxy" ) @@ -82,8 +82,6 @@ func (r *RollupBoostNode) Start() { cfg := r.cfg r.p.Require().NotNil(cfg, "rollup-boost config not initialized") - args, env := cfg.LaunchSpec(r.p) - if r.wsProxy == nil { r.wsProxy = tcpproxy.New(r.p.Logger()) r.p.Require().NoError(r.wsProxy.Start()) @@ -98,12 +96,34 @@ func (r *RollupBoostNode) Start() { r.p.Cleanup(func() { r.rpcProxy.Close() }) } + args, env := cfg.LaunchSpec(r.p) + + // Create channel for discovering flashblocks WS port from process logs. + // When using port 0, the OS assigns the port at bind time and the process logs it. + flashblocksWSChan := make(chan string, 1) + defer close(flashblocksWSChan) + // Parse Rust-structured logs and forward into Go logger with attributes logOut := logpipe.ToLogger(r.logger.New("stream", "stdout")) logErr := logpipe.ToLogger(r.logger.New("stream", "stderr")) + // Log parsing callback to extract bound addresses from process output + onLogEntry := func(e logpipe.LogEntry) { + msg := e.LogMessage() + // Flashblocks WS - custom log message from outbound.rs + if strings.HasPrefix(msg, "Flashblocks WebSocketPublisher listening on ") { + addr := strings.TrimPrefix(msg, "Flashblocks WebSocketPublisher listening on ") + select { + case flashblocksWSChan <- "ws://" + addr: + default: + } + } + } + stdOut := logpipe.LogCallback(func(line []byte) { - logOut(logpipe.ParseRustStructuredLogs(line)) + e := logpipe.ParseRustStructuredLogs(line) + logOut(e) + onLogEntry(e) }) stdErr := logpipe.LogCallback(func(line []byte) { logErr(logpipe.ParseRustStructuredLogs(line)) @@ -122,6 +142,9 @@ func (r *RollupBoostNode) Start() { err = r.sub.Start(execPath, args, env) r.p.Require().NoError(err, "start rollup-boost") + // RPC port: still uses pre-allocation because rollup-boost doesn't log the actual + // bound RPC address when using port 0. This requires a Rust change to fix. + // TODO: Update rollup-boost to log "RPC server listening on {addr}" and parse it here. rpcUpstreamURL := "http://" + cfg.RPCHost + ":" + strconv.Itoa(int(cfg.RPCPort)) waitTCPReady(r.p, rpcUpstreamURL, 5*time.Second) r.logger.Info("rollup-boost upstream RPC ready", "rpc", rpcUpstreamURL) @@ -129,17 +152,13 @@ func (r *RollupBoostNode) Start() { waitTCPReady(r.p, r.rpcProxyURL, 10*time.Second) r.logger.Info("rollup-boost proxy RPC ready", "proxy_rpc", r.rpcProxyURL) - // WS: wait for upstream first, then configure and test proxy + // Flashblocks WS: discover port from logs, then configure proxy if cfg.EnableFlashblocks { - wsUpstreamHostport := net.JoinHostPort(cfg.FlashblocksHost, strconv.Itoa(cfg.FlashblocksPort)) - wsUpstreamURL := "ws://" + wsUpstreamHostport + var flashblocksAddr string + r.p.Require().NoError(tasks.Await(r.p.Ctx(), flashblocksWSChan, &flashblocksAddr), "need Flashblocks WS address from logs") + r.logger.Info("rollup-boost upstream WS ready", "upstream_ws", flashblocksAddr) - // Wait for upstream WS TCP endpoint - waitTCPReady(r.p, wsUpstreamURL, 5*time.Second) - r.logger.Info("rollup-boost upstream WS ready", "upstream_ws", wsUpstreamURL) - - r.wsProxy.SetUpstream(ProxyAddr(r.p.Require(), wsUpstreamURL)) - waitWSReady(r.p, r.wsProxyURL, 10*time.Second) + r.wsProxy.SetUpstream(ProxyAddr(r.p.Require(), flashblocksAddr)) r.logger.Info("rollup-boost proxy WS ready", "proxy_ws", r.wsProxyURL) } } @@ -272,15 +291,15 @@ func (cfg *RollupBoostConfig) LaunchSpec(p devtest.P) (args []string, env []stri if cfg.FlashblocksHost == "" { cfg.FlashblocksHost = "127.0.0.1" } - if cfg.FlashblocksPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate flashblocks port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse flashblocks port") - cfg.FlashblocksPort = portVal + args = append(args, "--flashblocks", "--flashblocks-host="+cfg.FlashblocksHost) + if cfg.FlashblocksPort > 0 { + // Use explicitly configured port + args = append(args, "--flashblocks-port="+strconv.Itoa(cfg.FlashblocksPort)) + } else { + // Use port 0 to let the OS assign a port atomically at bind time. + // The actual port will be discovered by parsing the process logs. + args = append(args, "--flashblocks-port=0") } - fbPortStr := strconv.Itoa(cfg.FlashblocksPort) - args = append(args, "--flashblocks", "--flashblocks-host="+cfg.FlashblocksHost, "--flashblocks-port="+fbPortStr) if cfg.FlashblocksBuilderURL != "" { args = append(args, "--flashblocks-builder-url="+cfg.FlashblocksBuilderURL) } @@ -319,14 +338,16 @@ func (cfg *RollupBoostConfig) LaunchSpec(p devtest.P) (args []string, env []stri if cfg.DebugHost == "" { cfg.DebugHost = "127.0.0.1" } - if cfg.DebugPort <= 0 { - portStr, err := getAvailableLocalPort() - p.Require().NoError(err, "allocate rollup-boost debug port") - portVal, err := strconv.Atoi(portStr) - p.Require().NoError(err, "parse rollup-boost debug port") - cfg.DebugPort = portVal + args = append(args, "--debug-host="+cfg.DebugHost) + if cfg.DebugPort > 0 { + // Use explicitly configured port + args = append(args, "--debug-server-port="+strconv.Itoa(cfg.DebugPort)) + } else { + // Use port 0 to let the OS assign a port atomically at bind time. + // The debug server logs its bound address, but we don't need to parse it + // since the debug port is only used for manual debugging. + args = append(args, "--debug-server-port=0") } - args = append(args, "--debug-host="+cfg.DebugHost, "--debug-server-port="+strconv.Itoa(cfg.DebugPort)) args = append(args, cfg.ExtraArgs...) From 19bd482dea33a5ecdf36a9360baf40e8872a8d29 Mon Sep 17 00:00:00 2001 From: Samuel Stokes Date: Fri, 9 Jan 2026 14:51:06 -0500 Subject: [PATCH 3/3] make lint-go --- op-devstack/sysgo/util.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/op-devstack/sysgo/util.go b/op-devstack/sysgo/util.go index b26d422a85e..f323543b4e2 100644 --- a/op-devstack/sysgo/util.go +++ b/op-devstack/sysgo/util.go @@ -1,7 +1,6 @@ package sysgo import ( - "context" "errors" "fmt" "net" @@ -13,7 +12,6 @@ import ( "time" "github.com/ethereum-optimism/optimism/op-devstack/devtest" - opclient "github.com/ethereum-optimism/optimism/op-service/client" "github.com/stretchr/testify/assert" ) @@ -95,18 +93,6 @@ func waitTCPReady(p devtest.P, rawURL string, timeout time.Duration) { }, timeout, 100*time.Millisecond, waitMsg) } -// waitWSReady attempts an actual WebSocket handshake to confirm readiness using EventuallyWithT. -func waitWSReady(p devtest.P, rawURL string, timeout time.Duration) { - p.Helper() - waitWSMsg := fmt.Sprintf("WebSocket endpoint %s not ready within %v", rawURL, timeout) - p.Require().EventuallyWithT(func(c *assert.CollectT) { - ctx, cancel := context.WithTimeout(context.Background(), 750*time.Millisecond) - err := opclient.ProbeWS(ctx, rawURL) - cancel() - assert.NoError(c, err, "WebSocket handshake to %s should succeed", rawURL) - }, timeout, 100*time.Millisecond, waitWSMsg) -} - // parseAndValidateAddr ensures the address has a scheme and is a valid URL. // Returns the validated URL string or empty string if invalid. // This is used to parse addresses from process (e.g. op-rbuilder) log output.