Skip to content

fix(sysgo): allow OS to assign ports for opreth, rbuilder, rollup-boost #18739

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.

Sign up for GitHub

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

Merged
bitwiseguy merged 3 commits into from
Jan 9, 2026
Merged

fix(sysgo): allow OS to assign ports for opreth, rbuilder, rollup-boost #18739

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0af3091
sysgo: allow OS to assign ports for opreth and rbuilder
bitwiseguy Jan 9, 2026
246a8aa
sysgo: allow OS to assign ports for rollup-boost
bitwiseguy Jan 9, 2026
19bd482
make lint-go
bitwiseguy Jan 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 3 additions & 6 deletions op-devstack/sysgo/l2_el_opreth.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 != "" {
Expand Down
152 changes: 92 additions & 60 deletions op-devstack/sysgo/op_rbuilder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand All @@ -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"
)

Expand Down Expand Up @@ -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.
Expand All @@ -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)
}
Expand All @@ -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 == "" {
Expand Down Expand Up @@ -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))
Expand All @@ -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)
}
}
Expand Down
77 changes: 49 additions & 28 deletions op-devstack/sysgo/rollup_boost.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package sysgo

import (
"net"
"net/http"
"strconv"
"strings"
Expand All @@ -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"
)

Expand Down Expand Up @@ -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())
Expand All @@ -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))
Expand All @@ -122,24 +142,23 @@ 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)
r.rpcProxy.SetUpstream(ProxyAddr(r.p.Require(), rpcUpstreamURL))
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)
}
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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...)

Expand Down
Loading