op-conductor,op-node: SSZ-binary commit-unsafe-payload endpoint

op-conductor/conductor/service.go

@@ -190,6 +190,7 @@ func (c *OpConductor) initConsensus(ctx context.Context) error {
 		TrailingLogs:       c.cfg.RaftTrailingLogs,
 		HeartbeatTimeout:   c.cfg.RaftHeartbeatTimeout,
 		LeaderLeaseTimeout: c.cfg.RaftLeaderLeaseTimeout,
+		Metrics:            c.metrics,
 	}
 	cons, err := consensus.NewRaftConsensus(c.log, raftConsensusConfig)
 	if err != nil {
@@ -293,6 +294,13 @@ func (oc *OpConductor) initRPCServer(ctx context.Context) error {
 		Service:   api,
 	})
 
+	// Binary SSZ commit endpoint. Sized to comfortably fit a 10MB SSZ block;
+	// raise alongside the JSON-RPC body limit if larger blocks are needed.
+	server.AddHandler(
+		conductorrpc.CommitUnsafePayloadPath,
+		conductorrpc.BinaryCommitHandler(oc.log, oc, 16*1024*1024, oc.metrics),
+	)
+
 	if oc.cfg.RPCEnableProxy {
 		execClient, err := dial.DialEthClientWithTimeout(ctx, 1*time.Minute, oc.log, oc.cfg.ExecutionRPC)
 		if err != nil {
@@ -637,6 +645,12 @@ func (oc *OpConductor) CommitUnsafePayload(_ context.Context, payload *eth.Execu
 	return oc.cons.CommitUnsafePayload(payload)
 }
 
+// CommitUnsafePayloadSSZ commits a pre-SSZ-encoded unsafe payload to the cluster FSM.
+// Used by the binary HTTP endpoint to avoid the JSON-decode -> SSZ-marshal round trip.
+func (oc *OpConductor) CommitUnsafePayloadSSZ(_ context.Context, ssz []byte) error {
+	return oc.cons.CommitUnsafePayloadSSZ(ssz)
+}
+
 // SequencerHealthy returns true if sequencer is healthy.
 func (oc *OpConductor) SequencerHealthy(_ context.Context) bool {
 	return oc.healthy.Load()

op-conductor/consensus/iface.go

@@ -4,6 +4,16 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
+// ConsensusMetrics defines metrics for consensus commit operations.
+// This is intentionally minimal so that the consensus layer does not
+// depend on the full metrics.Metricer interface.
+type ConsensusMetrics interface {
+	RecordCommitDuration(marshalSec, raftApplySec float64)
+	RecordCommitPayloadSize(payloadBytes float64)
+	RecordFSMApplyDuration(seconds float64)
+	RecordLogStoreDuration(seconds float64)
+}
+
 // ServerSuffrage determines whether a Server in a Configuration gets a vote.
 type ServerSuffrage int
 
@@ -72,6 +82,10 @@ type Consensus interface {
 
 	// CommitUnsafePayload commits latest unsafe payload to the FSM in a strongly consistent fashion.
 	CommitUnsafePayload(payload *eth.ExecutionPayloadEnvelope) error
+	// CommitUnsafePayloadSSZ commits a pre-SSZ-encoded unsafe payload to the FSM,
+	// skipping the marshal step. The bytes are handed directly to raft.Apply and
+	// validated by the FSM on receive. Used by the binary HTTP endpoint.
+	CommitUnsafePayloadSSZ(ssz []byte) error
 	// LatestUnsafePayload returns the latest unsafe payload from FSM in a strongly consistent fashion.
 	LatestUnsafePayload() (*eth.ExecutionPayloadEnvelope, error)
 

op-conductor/consensus/raft.go

@@ -24,6 +24,8 @@ var _ Consensus = (*RaftConsensus)(nil)
 type RaftConsensus struct {
 	log log.Logger
 
+	metrics ConsensusMetrics
+
 	serverID raft.ServerID
 	r        *raft.Raft
 
@@ -58,6 +60,10 @@ type RaftConsensusConfig struct {
 	TrailingLogs       uint64
 	HeartbeatTimeout   time.Duration
 	LeaderLeaseTimeout time.Duration
+
+	// Metrics collects sub-operation timing data for the commit path.
+	// If nil, no metrics are recorded.
+	Metrics ConsensusMetrics
 }
 
 // checkTCPPortOpen attempts to connect to the specified address and returns an error if the connection fails.
@@ -89,10 +95,11 @@ func NewRaftConsensus(log log.Logger, cfg *RaftConsensusConfig) (*RaftConsensus,
 
 	var err error
 	logStorePath := filepath.Join(baseDir, "raft-log.db")
-	logStore, err := boltdb.NewBoltStore(logStorePath)
+	boltLogStore, err := boltdb.NewBoltStore(logStorePath)
 	if err != nil {
 		return nil, fmt.Errorf(`boltdb.NewBoltStore(%q): %w`, logStorePath, err)
 	}
+	logStore := wrapInstrumentedLogStore(boltLogStore, cfg.Metrics)
 
 	stableStorePath := filepath.Join(baseDir, "raft-stable.db")
 	stableStore, err := boltdb.NewBoltStore(stableStorePath)
@@ -131,7 +138,7 @@ func NewRaftConsensus(log log.Logger, cfg *RaftConsensusConfig) (*RaftConsensus,
 	}
 	log.Info("Raft server network transport is up", "addr", transport.LocalAddr())
 
-	fsm := NewUnsafeHeadTracker(log)
+	fsm := NewUnsafeHeadTracker(log, cfg.Metrics)
 
 	r, err := raft.NewRaft(rc, fsm, logStore, stableStore, snapshotStore, transport)
 	if err != nil {
@@ -173,13 +180,47 @@ func NewRaftConsensus(log log.Logger, cfg *RaftConsensusConfig) (*RaftConsensus,
 
 	return &RaftConsensus{
 		log:           log,
+		metrics:       cfg.Metrics,
 		r:             r,
 		serverID:      raft.ServerID(cfg.ServerID),
 		unsafeTracker: fsm,
 		transport:     transport,
 	}, err
 }
 
+func wrapInstrumentedLogStore(store raft.LogStore, metrics ConsensusMetrics) raft.LogStore {
+	if metrics == nil {
+		return store
+	}
+	return &instrumentedLogStore{
+		LogStore: store,
+		metrics:  metrics,
+	}
+}
+
+type instrumentedLogStore struct {
+	raft.LogStore
+	metrics ConsensusMetrics
+}
+
+func (s *instrumentedLogStore) StoreLog(logEntry *raft.Log) error {
+	start := time.Now()
+	err := s.LogStore.StoreLog(logEntry)
+	if s.metrics != nil {
+		s.metrics.RecordLogStoreDuration(time.Since(start).Seconds())
+	}
+	return err
+}
+
+func (s *instrumentedLogStore) StoreLogs(logEntries []*raft.Log) error {
+	start := time.Now()
+	err := s.LogStore.StoreLogs(logEntries)
+	if s.metrics != nil {
+		s.metrics.RecordLogStoreDuration(time.Since(start).Seconds())
+	}
+	return err
+}
+
 // Addr returns the address to contact this raft consensus server.
 // If no explicit address to advertise was configured,
 // the local network address that the raft-consensus server is listening on will be used.
@@ -296,19 +337,44 @@ func (rc *RaftConsensus) CommitUnsafePayload(payload *eth.ExecutionPayloadEnvelo
 	rc.log.Debug("committing unsafe payload", "number", uint64(payload.ExecutionPayload.BlockNumber), "hash", payload.ExecutionPayload.BlockHash.Hex())
 
 	var buf bytes.Buffer
+	marshalStart := time.Now()
 	if _, err := payload.MarshalSSZ(&buf); err != nil {
 		return errors.Wrap(err, "failed to marshal payload envelope")
 	}
+	marshalDur := time.Since(marshalStart)
 
+	applyStart := time.Now()
 	f := rc.r.Apply(buf.Bytes(), defaultTimeout)
 	if err := f.Error(); err != nil {
 		return errors.Wrap(err, "failed to apply payload envelope")
 	}
+	applyDur := time.Since(applyStart)
+
+	if rc.metrics != nil {
+		rc.metrics.RecordCommitDuration(marshalDur.Seconds(), applyDur.Seconds())
+		rc.metrics.RecordCommitPayloadSize(float64(buf.Len()))
+	}
 	rc.log.Debug("unsafe payload committed", "number", uint64(payload.ExecutionPayload.BlockNumber), "hash", payload.ExecutionPayload.BlockHash.Hex())
 
 	return nil
 }
 
+// CommitUnsafePayloadSSZ implements Consensus. The bytes are passed directly to
+// raft.Apply; the FSM validates by attempting UnmarshalSSZ on receive. This
+// avoids the SSZ marshal step (and, when callers send SSZ over the wire, the
+// JSON-decode-then-SSZ-marshal round trip the typed entrypoint requires).
+func (rc *RaftConsensus) CommitUnsafePayloadSSZ(ssz []byte) error {
+	if len(ssz) == 0 {
+		return errors.New("empty payload")
+	}
+
+	f := rc.r.Apply(ssz, defaultTimeout)
+	if err := f.Error(); err != nil {
+		return errors.Wrap(err, "failed to apply payload envelope")
+	}
+	return nil
+}
+
 // LatestUnsafePayload implements Consensus, it returns the latest unsafe payload from FSM in a strongly consistent fashion.
 func (rc *RaftConsensus) LatestUnsafePayload() (*eth.ExecutionPayloadEnvelope, error) {
 	if err := rc.r.Barrier(defaultTimeout).Error(); err != nil {

op-conductor/consensus/raft_fsm.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/hashicorp/raft"
@@ -17,18 +18,27 @@ var _ raft.FSM = (*unsafeHeadTracker)(nil)
 // unsafeHeadTracker implements raft.FSM for storing unsafe head payload into raft consensus layer.
 type unsafeHeadTracker struct {
 	log        log.Logger
+	metrics    ConsensusMetrics
 	mtx        sync.RWMutex
 	unsafeHead *eth.ExecutionPayloadEnvelope
 }
 
-func NewUnsafeHeadTracker(log log.Logger) *unsafeHeadTracker {
+func NewUnsafeHeadTracker(log log.Logger, metrics ConsensusMetrics) *unsafeHeadTracker {
 	return &unsafeHeadTracker{
-		log: log,
+		log:     log,
+		metrics: metrics,
 	}
 }
 
 // Apply implements raft.FSM, it applies the latest change (latest unsafe head payload) to FSM.
 func (t *unsafeHeadTracker) Apply(l *raft.Log) interface{} {
+	start := time.Now()
+	defer func() {
+		if t.metrics != nil {
+			t.metrics.RecordFSMApplyDuration(time.Since(start).Seconds())
+		}
+	}()
+
 	if len(l.Data) == 0 {
 		return fmt.Errorf("log data is nil or empty")
 	}

op-conductor/metrics/metrics.go

@@ -3,6 +3,7 @@ package metrics
 import (
 	"strconv"
 
+	"github.com/ethereum-optimism/optimism/op-conductor/consensus"
 	"github.com/ethereum-optimism/optimism/op-service/httputil"
 	opmetrics "github.com/ethereum-optimism/optimism/op-service/metrics"
 	"github.com/prometheus/client_golang/prometheus"
@@ -21,11 +22,17 @@ type Metricer interface {
 	RecordLoopExecutionTime(duration float64)
 	RecordRollupBoostConnectionAttempts(success bool, source string)
 	RecordWebSocketClientCount(count int)
+	// RecordBinaryCommitDuration records end-to-end handler duration for
+	// POST /commit-unsafe-payload requests. The equivalent metric for the
+	// JSON-RPC path is rpc_server_request_duration_seconds{method=conductor_commitUnsafePayload}.
+	RecordBinaryCommitDuration(seconds float64, success bool)
 	opmetrics.RPCMetricer
+	consensus.ConsensusMetrics
 }
 
 // Metrics implementation must implement RegistryMetricer to allow the metrics server to work.
 var _ opmetrics.RegistryMetricer = (*Metrics)(nil)
+var _ consensus.ConsensusMetrics = (*Metrics)(nil)
 
 type Metrics struct {
 	ns       string
@@ -46,6 +53,14 @@ type Metrics struct {
 
 	loopExecutionTime prometheus.Histogram
 	webSocketClients  prometheus.Gauge
+
+	binaryCommitRequestDuration *prometheus.HistogramVec
+
+	commitMarshalDuration   prometheus.Histogram
+	commitRaftApplyDuration prometheus.Histogram
+	commitPayloadSize       prometheus.Histogram
+	fsmApplyDuration        prometheus.Histogram
+	logStoreDuration        prometheus.Histogram
 }
 
 func (m *Metrics) Registry() *prometheus.Registry {
@@ -122,6 +137,44 @@ func NewMetrics() *Metrics {
 			Name:      "websocket_clients_connected",
 			Help:      "Number of WebSocket clients currently connected to the hub",
 		}),
+		binaryCommitRequestDuration: factory.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "binary_commit_request_duration_seconds",
+			Help: "End-to-end handler duration for POST /commit-unsafe-payload requests. " +
+				"Directly comparable to rpc_server_request_duration_seconds{method=conductor_commitunsafepayload} " +
+				"on the JSON-RPC path.",
+			Buckets: []float64{.001, .0025, .005, .01, .025, .05, .1, .25, .5},
+		}, []string{"success"}),
+		commitMarshalDuration: factory.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "commit_marshal_duration_seconds",
+			Help:      "Time (in seconds) to SSZ-marshal the payload in CommitUnsafePayload",
+			Buckets:   []float64{.0001, .00025, .0005, .001, .0025, .005, .01, .025},
+		}),
+		commitRaftApplyDuration: factory.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "commit_raft_apply_duration_seconds",
+			Help:      "Time (in seconds) for raft Apply (replication, storage, and FSM apply) in CommitUnsafePayload",
+			Buckets:   []float64{.001, .0025, .005, .01, .025, .05, .1, .25, .5},
+		}),
+		commitPayloadSize: factory.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "commit_payload_size_bytes",
+			Help:      "SSZ-encoded payload size in bytes",
+			Buckets:   []float64{1000, 10000, 50000, 100000, 500000, 1000000, 1500000, 2000000, 2500000},
+		}),
+		fsmApplyDuration: factory.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "fsm_apply_duration_seconds",
+			Help:      "Time (in seconds) spent in FSM Apply (SSZ decode plus unsafe-head update)",
+			Buckets:   []float64{.0001, .00025, .0005, .001, .0025, .005, .01, .025},
+		}),
+		logStoreDuration: factory.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Name:      "raft_log_store_duration_seconds",
+			Help:      "Time (in seconds) spent writing raft log entries to the underlying log store",
+			Buckets:   []float64{.0001, .00025, .0005, .001, .0025, .005, .01, .025, .05},
+		}),
 	}
 }
 
@@ -183,3 +236,24 @@ func (m *Metrics) RecordRollupBoostConnectionAttempts(success bool, source strin
 func (m *Metrics) RecordWebSocketClientCount(count int) {
 	m.webSocketClients.Set(float64(count))
 }
+
+func (m *Metrics) RecordBinaryCommitDuration(seconds float64, success bool) {
+	m.binaryCommitRequestDuration.WithLabelValues(strconv.FormatBool(success)).Observe(seconds)
+}
+
+func (m *Metrics) RecordCommitDuration(marshalSec, raftApplySec float64) {
+	m.commitMarshalDuration.Observe(marshalSec)
+	m.commitRaftApplyDuration.Observe(raftApplySec)
+}
+
+func (m *Metrics) RecordCommitPayloadSize(payloadBytes float64) {
+	m.commitPayloadSize.Observe(payloadBytes)
+}
+
+func (m *Metrics) RecordFSMApplyDuration(seconds float64) {
+	m.fsmApplyDuration.Observe(seconds)
+}
+
+func (m *Metrics) RecordLogStoreDuration(seconds float64) {
+	m.logStoreDuration.Observe(seconds)
+}