Merge branch 'develop' into felipe/fix-consensus-shutdown

op-batcher/batcher/channel_builder.go

@@ -7,13 +7,13 @@ import (
 	"io"
 	"math"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
 var (
 	ErrInvalidChannelTimeout = errors.New("channel timeout is less than the safety margin")
-	ErrInputTargetReached    = errors.New("target amount of input data reached")
 	ErrMaxFrameIndex         = errors.New("max frame index reached (uint16)")
 	ErrMaxDurationReached    = errors.New("max channel duration reached")
 	ErrChannelTimeoutClose   = errors.New("close to channel timeout")
@@ -55,19 +55,9 @@ type ChannelConfig struct {
 	SubSafetyMargin uint64
 	// The maximum byte-size a frame can have.
 	MaxFrameSize uint64
-	// The target number of frames to create per channel. Note that if the
-	// realized compression ratio is worse than the approximate, more frames may
-	// actually be created. This also depends on how close TargetFrameSize is to
-	// MaxFrameSize.
-	TargetFrameSize uint64
-	// The target number of frames to create in this channel. If the realized
-	// compression ratio is worse than approxComprRatio, additional leftover
-	// frame(s) might get created.
-	TargetNumFrames int
-	// Approximated compression ratio to assume. Should be slightly smaller than
-	// average from experiments to avoid the chances of creating a small
-	// additional leftover frame.
-	ApproxComprRatio float64
+
+	// CompressorConfig contains the configuration for creating new compressors.
+	CompressorConfig compressor.Config
 }
 
 // Check validates the [ChannelConfig] parameters.
@@ -96,12 +86,6 @@ func (cc *ChannelConfig) Check() error {
 	return nil
 }
 
-// InputThreshold calculates the input data threshold in bytes from the given
-// parameters.
-func (c ChannelConfig) InputThreshold() uint64 {
-	return uint64(float64(c.TargetNumFrames) * float64(c.TargetFrameSize) / c.ApproxComprRatio)
-}
-
 type frameID struct {
 	chID        derive.ChannelID
 	frameNumber uint16
@@ -142,7 +126,11 @@ type channelBuilder struct {
 // newChannelBuilder creates a new channel builder or returns an error if the
 // channel out could not be created.
 func newChannelBuilder(cfg ChannelConfig) (*channelBuilder, error) {
-	co, err := derive.NewChannelOut()
+	c, err := cfg.CompressorConfig.NewCompressor()
+	if err != nil {
+		return nil, err
+	}
+	co, err := derive.NewChannelOut(c)
 	if err != nil {
 		return nil, err
 	}
@@ -209,7 +197,7 @@ func (c *channelBuilder) AddBlock(block *types.Block) (derive.L1BlockInfo, error
 		return l1info, fmt.Errorf("converting block to batch: %w", err)
 	}
 
-	if _, err = c.co.AddBatch(batch); errors.Is(err, derive.ErrTooManyRLPBytes) {
+	if _, err = c.co.AddBatch(batch); errors.Is(err, derive.ErrTooManyRLPBytes) || errors.Is(err, derive.CompressorFullErr) {
 		c.setFullErr(err)
 		return l1info, c.FullErr()
 	} else if err != nil {
@@ -218,8 +206,8 @@ func (c *channelBuilder) AddBlock(block *types.Block) (derive.L1BlockInfo, error
 	c.blocks = append(c.blocks, block)
 	c.updateSwTimeout(batch)
 
-	if c.inputTargetReached() {
-		c.setFullErr(ErrInputTargetReached)
+	if err = c.co.FullErr(); err != nil {
+		c.setFullErr(err)
 		// Adding this block still worked, so don't return error, just mark as full
 	}
 
@@ -293,12 +281,6 @@ func (c *channelBuilder) TimedOut(blockNum uint64) bool {
 	return c.timeout != 0 && blockNum >= c.timeout
 }
 
-// inputTargetReached says whether the target amount of input data has been
-// reached in this channel builder. No more blocks can be added afterwards.
-func (c *channelBuilder) inputTargetReached() bool {
-	return uint64(c.co.InputBytes()) >= c.cfg.InputThreshold()
-}
-
 // IsFull returns whether the channel is full.
 // FullErr returns the reason for the channel being full.
 func (c *channelBuilder) IsFull() bool {
@@ -310,7 +292,7 @@ func (c *channelBuilder) IsFull() bool {
 //
 // It returns a ChannelFullError wrapping one of the following possible reasons
 // for the channel being full:
-//   - ErrInputTargetReached if the target amount of input data has been reached,
+//   - derive.CompressorFullErr if the compressor target has been reached,
 //   - derive.MaxRLPBytesPerChannel if the general maximum amount of input data
 //     would have been exceeded by the latest AddBlock call,
 //   - ErrMaxFrameIndex if the maximum number of frames has been generated

op-batcher/batcher/channel_builder_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
@@ -28,9 +29,11 @@ var defaultTestChannelConfig = ChannelConfig{
 	MaxChannelDuration: 1,
 	SubSafetyMargin:    4,
 	MaxFrameSize:       120000,
-	TargetFrameSize:    100000,
-	TargetNumFrames:    1,
-	ApproxComprRatio:   0.4,
+	CompressorConfig: compressor.Config{
+		TargetFrameSize:  100000,
+		TargetNumFrames:  1,
+		ApproxComprRatio: 0.4,
+	},
 }
 
 // TestChannelConfig_Check tests the [ChannelConfig] [Check] function.
@@ -416,7 +419,9 @@ func TestChannelBuilder_OutputWrongFramePanic(t *testing.T) {
 
 	// Mock the internals of `channelBuilder.outputFrame`
 	// to construct a single frame
-	co, err := derive.NewChannelOut()
+	c, err := channelConfig.CompressorConfig.NewCompressor()
+	require.NoError(t, err)
+	co, err := derive.NewChannelOut(c)
 	require.NoError(t, err)
 	var buf bytes.Buffer
 	fn, err := co.OutputFrame(&buf, channelConfig.MaxFrameSize)
@@ -483,8 +488,8 @@ func TestChannelBuilder_MaxRLPBytesPerChannel(t *testing.T) {
 	t.Parallel()
 	channelConfig := defaultTestChannelConfig
 	channelConfig.MaxFrameSize = derive.MaxRLPBytesPerChannel * 2
-	channelConfig.TargetFrameSize = derive.MaxRLPBytesPerChannel * 2
-	channelConfig.ApproxComprRatio = 1
+	channelConfig.CompressorConfig.TargetFrameSize = derive.MaxRLPBytesPerChannel * 2
+	channelConfig.CompressorConfig.ApproxComprRatio = 1
 
 	// Construct the channel builder
 	cb, err := newChannelBuilder(channelConfig)
@@ -500,9 +505,9 @@ func TestChannelBuilder_MaxRLPBytesPerChannel(t *testing.T) {
 func TestChannelBuilder_OutputFramesMaxFrameIndex(t *testing.T) {
 	channelConfig := defaultTestChannelConfig
 	channelConfig.MaxFrameSize = 24
-	channelConfig.TargetNumFrames = math.MaxInt
-	channelConfig.TargetFrameSize = 24
-	channelConfig.ApproxComprRatio = 0
+	channelConfig.CompressorConfig.TargetNumFrames = math.MaxInt
+	channelConfig.CompressorConfig.TargetFrameSize = 24
+	channelConfig.CompressorConfig.ApproxComprRatio = 0
 
 	// Continuously add blocks until the max frame index is reached
 	// This should cause the [channelBuilder.OutputFrames] function
@@ -544,9 +549,9 @@ func TestChannelBuilder_AddBlock(t *testing.T) {
 	channelConfig.MaxFrameSize = 30
 
 	// Configure the Input Threshold params so we observe a full channel
-	channelConfig.TargetFrameSize = 30
-	channelConfig.TargetNumFrames = 2
-	channelConfig.ApproxComprRatio = 1
+	channelConfig.CompressorConfig.TargetFrameSize = 30
+	channelConfig.CompressorConfig.TargetNumFrames = 2
+	channelConfig.CompressorConfig.ApproxComprRatio = 1
 
 	// Construct the channel builder
 	cb, err := newChannelBuilder(channelConfig)
@@ -564,7 +569,7 @@ func TestChannelBuilder_AddBlock(t *testing.T) {
 
 	// Since the channel output is full, the next call to AddBlock
 	// should return the channel out full error
-	require.ErrorIs(t, addMiniBlock(cb), ErrInputTargetReached)
+	require.ErrorIs(t, addMiniBlock(cb), derive.CompressorFullErr)
 }
 
 // TestChannelBuilder_Reset tests the [Reset] function
@@ -701,10 +706,10 @@ func TestChannelBuilder_OutputBytes(t *testing.T) {
 	require := require.New(t)
 	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
 	cfg := defaultTestChannelConfig
-	cfg.TargetFrameSize = 1000
+	cfg.CompressorConfig.TargetFrameSize = 1000
 	cfg.MaxFrameSize = 1000
-	cfg.TargetNumFrames = 16
-	cfg.ApproxComprRatio = 1.0
+	cfg.CompressorConfig.TargetNumFrames = 16
+	cfg.CompressorConfig.ApproxComprRatio = 1.0
 	cb, err := newChannelBuilder(cfg)
 	require.NoError(err, "newChannelBuilder")
 
@@ -713,7 +718,7 @@ func TestChannelBuilder_OutputBytes(t *testing.T) {
 	for {
 		block, _ := dtest.RandomL2Block(rng, rng.Intn(32))
 		_, err := cb.AddBlock(block)
-		if errors.Is(err, ErrInputTargetReached) {
+		if errors.Is(err, derive.CompressorFullErr) {
 			break
 		}
 		require.NoError(err)

op-batcher/batcher/channel_manager_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
@@ -98,9 +99,11 @@ func TestChannelManagerReturnsErrReorgWhenDrained(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
-			MaxFrameSize:     120_000,
-			ApproxComprRatio: 1.0,
+			MaxFrameSize: 120_000,
+			CompressorConfig: compressor.Config{
+				TargetFrameSize:  0,
+				ApproxComprRatio: 1.0,
+			},
 		})
 
 	a := newMiniL2Block(0)
@@ -170,9 +173,12 @@ func TestChannelManager_Clear(t *testing.T) {
 		ChannelTimeout: 10,
 		// Have to set the max frame size here otherwise the channel builder would not
 		// be able to output any frames
-		MaxFrameSize:     24,
-		TargetFrameSize:  24,
-		ApproxComprRatio: 1.0,
+		MaxFrameSize: 24,
+		CompressorConfig: compressor.Config{
+			TargetFrameSize:  24,
+			TargetNumFrames:  1,
+			ApproxComprRatio: 1.0,
+		},
 	})
 
 	// Channel Manager state should be empty by default
@@ -331,9 +337,11 @@ func TestChannelManager_TxResend(t *testing.T) {
 	log := testlog.Logger(t, log.LvlError)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
-			MaxFrameSize:     120_000,
-			ApproxComprRatio: 1.0,
+			MaxFrameSize: 120_000,
+			CompressorConfig: compressor.Config{
+				TargetFrameSize:  0,
+				ApproxComprRatio: 1.0,
+			},
 		})
 
 	a, _ := derivetest.RandomL2Block(rng, 4)
@@ -372,10 +380,12 @@ func TestChannelManagerCloseBeforeFirstUse(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
-			MaxFrameSize:     100,
-			ApproxComprRatio: 1.0,
-			ChannelTimeout:   1000,
+			MaxFrameSize:   100,
+			ChannelTimeout: 1000,
+			CompressorConfig: compressor.Config{
+				TargetFrameSize:  0,
+				ApproxComprRatio: 1.0,
+			},
 		})
 
 	a, _ := derivetest.RandomL2Block(rng, 4)
@@ -397,10 +407,13 @@ func TestChannelManagerCloseNoPendingChannel(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
-			MaxFrameSize:     100,
-			ApproxComprRatio: 1.0,
-			ChannelTimeout:   1000,
+			MaxFrameSize:   100,
+			ChannelTimeout: 1000,
+			CompressorConfig: compressor.Config{
+				TargetFrameSize:  1,
+				TargetNumFrames:  1,
+				ApproxComprRatio: 1.0,
+			},
 		})
 	a := newMiniL2Block(0)
 	b := newMiniL2BlockWithNumberParent(0, big.NewInt(1), a.Hash())
@@ -433,11 +446,13 @@ func TestChannelManagerClosePendingChannel(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetNumFrames:  100,
-			TargetFrameSize:  1000,
-			MaxFrameSize:     1000,
-			ApproxComprRatio: 1.0,
-			ChannelTimeout:   1000,
+			MaxFrameSize:   1000,
+			ChannelTimeout: 1000,
+			CompressorConfig: compressor.Config{
+				TargetNumFrames:  100,
+				TargetFrameSize:  1000,
+				ApproxComprRatio: 1.0,
+			},
 		})
 
 	a := newMiniL2Block(50_000)
@@ -476,11 +491,13 @@ func TestChannelManagerCloseAllTxsFailed(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetNumFrames:  100,
-			TargetFrameSize:  1000,
-			MaxFrameSize:     1000,
-			ApproxComprRatio: 1.0,
-			ChannelTimeout:   1000,
+			MaxFrameSize:   1000,
+			ChannelTimeout: 1000,
+			CompressorConfig: compressor.Config{
+				TargetNumFrames:  100,
+				TargetFrameSize:  1000,
+				ApproxComprRatio: 1.0,
+			},
 		})
 
 	a := newMiniL2Block(50_000)

op-batcher/batcher/config.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-batcher/flags"
 	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
@@ -84,23 +85,14 @@ type CLIConfig struct {
 	// MaxL1TxSize is the maximum size of a batch tx submitted to L1.
 	MaxL1TxSize uint64
 
-	// TargetL1TxSize is the target size of a batch tx submitted to L1.
-	TargetL1TxSize uint64
-
-	// TargetNumFrames is the target number of frames per channel.
-	TargetNumFrames int
-
-	// ApproxComprRatio is the approximate compression ratio (<= 1.0) of the used
-	// compression algorithm.
-	ApproxComprRatio float64
-
 	Stopped bool
 
-	TxMgrConfig   txmgr.CLIConfig
-	RPCConfig     rpc.CLIConfig
-	LogConfig     oplog.CLIConfig
-	MetricsConfig opmetrics.CLIConfig
-	PprofConfig   oppprof.CLIConfig
+	TxMgrConfig      txmgr.CLIConfig
+	RPCConfig        rpc.CLIConfig
+	LogConfig        oplog.CLIConfig
+	MetricsConfig    opmetrics.CLIConfig
+	PprofConfig      oppprof.CLIConfig
+	CompressorConfig compressor.CLIConfig
 }
 
 func (c CLIConfig) Check() error {
@@ -136,14 +128,12 @@ func NewConfig(ctx *cli.Context) CLIConfig {
 		MaxPendingTransactions: ctx.GlobalUint64(flags.MaxPendingTransactionsFlag.Name),
 		MaxChannelDuration:     ctx.GlobalUint64(flags.MaxChannelDurationFlag.Name),
 		MaxL1TxSize:            ctx.GlobalUint64(flags.MaxL1TxSizeBytesFlag.Name),
-		TargetL1TxSize:         ctx.GlobalUint64(flags.TargetL1TxSizeBytesFlag.Name),
-		TargetNumFrames:        ctx.GlobalInt(flags.TargetNumFramesFlag.Name),
-		ApproxComprRatio:       ctx.GlobalFloat64(flags.ApproxComprRatioFlag.Name),
 		Stopped:                ctx.GlobalBool(flags.StoppedFlag.Name),
 		TxMgrConfig:            txmgr.ReadCLIConfig(ctx),
 		RPCConfig:              rpc.ReadCLIConfig(ctx),
 		LogConfig:              oplog.ReadCLIConfig(ctx),
 		MetricsConfig:          opmetrics.ReadCLIConfig(ctx),
 		PprofConfig:            oppprof.ReadCLIConfig(ctx),
+		CompressorConfig:       compressor.ReadCLIConfig(ctx),
 	}
 }

op-batcher/batcher/driver.go

@@ -89,10 +89,8 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metri
 			ChannelTimeout:     rcfg.ChannelTimeout,
 			MaxChannelDuration: cfg.MaxChannelDuration,
 			SubSafetyMargin:    cfg.SubSafetyMargin,
-			MaxFrameSize:       cfg.MaxL1TxSize - 1,    // subtract 1 byte for version
-			TargetFrameSize:    cfg.TargetL1TxSize - 1, // subtract 1 byte for version
-			TargetNumFrames:    cfg.TargetNumFrames,
-			ApproxComprRatio:   cfg.ApproxComprRatio,
+			MaxFrameSize:       cfg.MaxL1TxSize - 1, // subtract 1 byte for version
+			CompressorConfig:   cfg.CompressorConfig.Config(),
 		},
 	}
 

op-batcher/compressor/cli.go

+package compressor
+
+import (
+	"strings"
+
+	opservice "github.com/ethereum-optimism/optimism/op-service"
+	"github.com/urfave/cli"
+)
+
+const (
+	TargetL1TxSizeBytesFlagName = "target-l1-tx-size-bytes"
+	TargetNumFramesFlagName     = "target-num-frames"
+	ApproxComprRatioFlagName    = "approx-compr-ratio"
+	KindFlagName                = "compressor"
+)
+
+func CLIFlags(envPrefix string) []cli.Flag {
+	return []cli.Flag{
+		cli.Uint64Flag{
+			Name:   TargetL1TxSizeBytesFlagName,
+			Usage:  "The target size of a batch tx submitted to L1.",
+			Value:  100_000,
+			EnvVar: opservice.PrefixEnvVar(envPrefix, "TARGET_L1_TX_SIZE_BYTES"),
+		},
+		cli.IntFlag{
+			Name:   TargetNumFramesFlagName,
+			Usage:  "The target number of frames to create per channel",
+			Value:  1,
+			EnvVar: opservice.PrefixEnvVar(envPrefix, "TARGET_NUM_FRAMES"),
+		},
+		cli.Float64Flag{
+			Name:   ApproxComprRatioFlagName,
+			Usage:  "The approximate compression ratio (<= 1.0)",
+			Value:  0.4,
+			EnvVar: opservice.PrefixEnvVar(envPrefix, "APPROX_COMPR_RATIO"),
+		},
+		cli.StringFlag{
+			Name:   KindFlagName,
+			Usage:  "The type of compressor. Valid options: " + strings.Join(KindKeys, ", "),
+			EnvVar: opservice.PrefixEnvVar(envPrefix, "COMPRESSOR"),
+			Value:  RatioKind,
+		},
+	}
+}
+
+type CLIConfig struct {
+	// TargetL1TxSizeBytes to target when creating channel frames. Note that if the
+	// realized compression ratio is worse than the approximate, more frames may
+	// actually be created. This also depends on how close the target is to the
+	// max frame size.
+	TargetL1TxSizeBytes uint64
+	// TargetNumFrames to create in this channel. If the realized compression ratio
+	// is worse than approxComprRatio, additional leftover frame(s) might get created.
+	TargetNumFrames int
+	// ApproxComprRatio to assume. Should be slightly smaller than average from
+	// experiments to avoid the chances of creating a small additional leftover frame.
+	ApproxComprRatio float64
+	// Type of compressor to use. Must be one of KindKeys.
+	Kind string
+}
+
+func (c *CLIConfig) Config() Config {
+	return Config{
+		TargetFrameSize:  c.TargetL1TxSizeBytes - 1, // subtract 1 byte for version
+		TargetNumFrames:  c.TargetNumFrames,
+		ApproxComprRatio: c.ApproxComprRatio,
+		Kind:             c.Kind,
+	}
+}
+
+func ReadCLIConfig(ctx *cli.Context) CLIConfig {
+	return CLIConfig{
+		Kind:                ctx.GlobalString(KindFlagName),
+		TargetL1TxSizeBytes: ctx.GlobalUint64(TargetL1TxSizeBytesFlagName),
+		TargetNumFrames:     ctx.GlobalInt(TargetNumFramesFlagName),
+		ApproxComprRatio:    ctx.GlobalFloat64(ApproxComprRatioFlagName),
+	}
+}

op-batcher/compressor/compressors.go

+package compressor
+
+import (
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+)
+
+type FactoryFunc func(Config) (derive.Compressor, error)
+
+const RatioKind = "ratio"
+const ShadowKind = "shadow"
+
+var Kinds = map[string]FactoryFunc{
+	RatioKind:  NewRatioCompressor,
+	ShadowKind: NewShadowCompressor,
+}
+
+var KindKeys []string
+
+func init() {
+	for k := range Kinds {
+		KindKeys = append(KindKeys, k)
+	}
+}

op-batcher/compressor/config.go

+package compressor
+
+import (
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+)
+
+type Config struct {
+	// TargetFrameSize to target when creating channel frames. Note that if the
+	// realized compression ratio is worse than the approximate, more frames may
+	// actually be created. This also depends on how close the target is to the
+	// max frame size.
+	TargetFrameSize uint64
+	// TargetNumFrames to create in this channel. If the realized compression ratio
+	// is worse than approxComprRatio, additional leftover frame(s) might get created.
+	TargetNumFrames int
+	// ApproxComprRatio to assume. Should be slightly smaller than average from
+	// experiments to avoid the chances of creating a small additional leftover frame.
+	ApproxComprRatio float64
+	// Kind of compressor to use. Must
+	Kind string
+}
+
+func (c Config) NewCompressor() (derive.Compressor, error) {
+	if k, ok := Kinds[c.Kind]; ok {
+		return k(c)
+	}
+	// default to RatioCompressor
+	return Kinds[RatioKind](c)
+}

op-batcher/compressor/ratio_compressor.go

+package compressor
+
+import (
+	"bytes"
+	"compress/zlib"
+
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+)
+
+type RatioCompressor struct {
+	config Config
+
+	inputBytes int
+	buf        bytes.Buffer
+	compress   *zlib.Writer
+}
+
+// NewRatioCompressor creates a new derive.Compressor implementation that uses the target
+// size and a compression ratio parameter to determine how much data can be written to
+// the compressor before it's considered full. The full calculation is as follows:
+//
+//	full = uncompressedLength * approxCompRatio >= targetFrameSize * targetNumFrames
+func NewRatioCompressor(config Config) (derive.Compressor, error) {
+	c := &RatioCompressor{
+		config: config,
+	}
+
+	compress, err := zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+	c.compress = compress
+
+	return c, nil
+}
+
+func (t *RatioCompressor) Write(p []byte) (int, error) {
+	if err := t.FullErr(); err != nil {
+		return 0, err
+	}
+	t.inputBytes += len(p)
+	return t.compress.Write(p)
+}
+
+func (t *RatioCompressor) Close() error {
+	return t.compress.Close()
+}
+
+func (t *RatioCompressor) Read(p []byte) (int, error) {
+	return t.buf.Read(p)
+}
+
+func (t *RatioCompressor) Reset() {
+	t.buf.Reset()
+	t.compress.Reset(&t.buf)
+	t.inputBytes = 0
+}
+
+func (t *RatioCompressor) Len() int {
+	return t.buf.Len()
+}
+
+func (t *RatioCompressor) Flush() error {
+	return t.compress.Flush()
+}
+
+func (t *RatioCompressor) FullErr() error {
+	if t.inputTargetReached() {
+		return derive.CompressorFullErr
+	}
+	return nil
+}
+
+// InputThreshold calculates the input data threshold in bytes from the given
+// parameters.
+func (t *RatioCompressor) InputThreshold() uint64 {
+	return uint64(float64(t.config.TargetNumFrames) * float64(t.config.TargetFrameSize) / t.config.ApproxComprRatio)
+}
+
+// inputTargetReached says whether the target amount of input data has been
+// reached in this channel builder. No more blocks can be added afterwards.
+func (t *RatioCompressor) inputTargetReached() bool {
+	return uint64(t.inputBytes) >= t.InputThreshold()
+}

op-batcher/batcher/channel_config_test.go → op-batcher/compressor/ratio_compressor_test.go

-package batcher_test
+package compressor_test
 
 import (
 	"math"
 	"testing"
 
-	"github.com/ethereum-optimism/optimism/op-batcher/batcher"
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/stretchr/testify/require"
 )
 
@@ -118,12 +118,13 @@ func TestInputThreshold(t *testing.T) {
 
 	// Validate each test case
 	for _, tt := range tests {
-		config := batcher.ChannelConfig{
+		comp, err := compressor.NewRatioCompressor(compressor.Config{
 			TargetFrameSize:  tt.input.TargetFrameSize,
 			TargetNumFrames:  tt.input.TargetNumFrames,
 			ApproxComprRatio: tt.input.ApproxComprRatio,
-		}
-		got := config.InputThreshold()
+		})
+		require.NoError(t, err)
+		got := comp.(*compressor.RatioCompressor).InputThreshold()
 		tt.assertion(got)
 	}
 }

op-batcher/compressor/shadow_compressor.go

+package compressor
+
+import (
+	"bytes"
+	"compress/zlib"
+
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+)
+
+type ShadowCompressor struct {
+	config Config
+
+	buf      bytes.Buffer
+	compress *zlib.Writer
+
+	shadowBuf      bytes.Buffer
+	shadowCompress *zlib.Writer
+
+	fullErr error
+}
+
+// NewShadowCompressor creates a new derive.Compressor implementation that contains two
+// compression buffers: one used for size estimation, and one used for the final
+// compressed output. The first is flushed on every write, the second isn't, which means
+// the final compressed data is always slightly smaller than the target. There is one
+// exception to this rule: the first write to the buffer is not checked against the
+// target, which allows individual blocks larger than the target to be included (and will
+// be split across multiple channel frames).
+func NewShadowCompressor(config Config) (derive.Compressor, error) {
+	c := &ShadowCompressor{
+		config: config,
+	}
+
+	var err error
+	c.compress, err = zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+	c.shadowCompress, err = zlib.NewWriterLevel(&c.shadowBuf, zlib.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func (t *ShadowCompressor) Write(p []byte) (int, error) {
+	_, err := t.shadowCompress.Write(p)
+	if err != nil {
+		return 0, err
+	}
+	err = t.shadowCompress.Flush()
+	if err != nil {
+		return 0, err
+	}
+	if uint64(t.shadowBuf.Len()) > t.config.TargetFrameSize*uint64(t.config.TargetNumFrames) {
+		t.fullErr = derive.CompressorFullErr
+		if t.Len() > 0 {
+			// only return an error if we've already written data to this compressor before
+			// (otherwise individual blocks over the target would never be written)
+			return 0, t.fullErr
+		}
+	}
+	return t.compress.Write(p)
+}
+
+func (t *ShadowCompressor) Close() error {
+	return t.compress.Close()
+}
+
+func (t *ShadowCompressor) Read(p []byte) (int, error) {
+	return t.buf.Read(p)
+}
+
+func (t *ShadowCompressor) Reset() {
+	t.buf.Reset()
+	t.compress.Reset(&t.buf)
+	t.shadowBuf.Reset()
+	t.shadowCompress.Reset(&t.shadowBuf)
+	t.fullErr = nil
+}
+
+func (t *ShadowCompressor) Len() int {
+	return t.buf.Len()
+}
+
+func (t *ShadowCompressor) Flush() error {
+	return t.compress.Flush()
+}
+
+func (t *ShadowCompressor) FullErr() error {
+	return t.fullErr
+}

op-batcher/compressor/shadow_compressor_test.go

+package compressor_test
+
+import (
+	"bytes"
+	"compress/zlib"
+	"io"
+	"math/rand"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/stretchr/testify/require"
+)
+
+func randomBytes(t *testing.T, length int) []byte {
+	b := make([]byte, length)
+	_, err := rand.Read(b)
+	require.NoError(t, err)
+	return b
+}
+
+func TestShadowCompressor(t *testing.T) {
+	type test struct {
+		name            string
+		targetFrameSize uint64
+		targetNumFrames int
+		data            [][]byte
+		errs            []error
+		fullErr         error
+	}
+
+	tests := []test{{
+		name:            "no data",
+		targetFrameSize: 1,
+		targetNumFrames: 1,
+		data:            [][]byte{},
+		errs:            []error{},
+		fullErr:         nil,
+	}, {
+		name:            "large first block",
+		targetFrameSize: 1,
+		targetNumFrames: 1,
+		data:            [][]byte{bytes.Repeat([]byte{0}, 1024)},
+		errs:            []error{nil},
+		fullErr:         derive.CompressorFullErr,
+	}, {
+		name:            "large second block",
+		targetFrameSize: 1,
+		targetNumFrames: 1,
+		data:            [][]byte{bytes.Repeat([]byte{0}, 512), bytes.Repeat([]byte{0}, 1024)},
+		errs:            []error{nil, derive.CompressorFullErr},
+		fullErr:         derive.CompressorFullErr,
+	}, {
+		name:            "random data",
+		targetFrameSize: 1200,
+		targetNumFrames: 1,
+		data:            [][]byte{randomBytes(t, 512), randomBytes(t, 512), randomBytes(t, 512)},
+		errs:            []error{nil, nil, derive.CompressorFullErr},
+		fullErr:         derive.CompressorFullErr,
+	}}
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			require.Equal(t, len(test.errs), len(test.data), "invalid test case: len(data) != len(errs)")
+
+			sc, err := compressor.NewShadowCompressor(compressor.Config{
+				TargetFrameSize: test.targetFrameSize,
+				TargetNumFrames: test.targetNumFrames,
+			})
+			require.NoError(t, err)
+
+			for i, d := range test.data {
+				_, err = sc.Write(d)
+				if test.errs[i] != nil {
+					require.ErrorIs(t, err, test.errs[i])
+					require.Equal(t, i, len(test.data)-1)
+				} else {
+					require.NoError(t, err)
+				}
+			}
+
+			if test.fullErr != nil {
+				require.ErrorIs(t, sc.FullErr(), test.fullErr)
+			} else {
+				require.NoError(t, sc.FullErr())
+			}
+
+			err = sc.Close()
+			require.NoError(t, err)
+
+			buf, err := io.ReadAll(sc)
+			require.NoError(t, err)
+
+			r, err := zlib.NewReader(bytes.NewBuffer(buf))
+			require.NoError(t, err)
+
+			uncompressed, err := io.ReadAll(r)
+			require.NoError(t, err)
+
+			concat := make([]byte, 0)
+			for i, d := range test.data {
+				if test.errs[i] != nil {
+					break
+				}
+				concat = append(concat, d...)
+			}
+
+			require.Equal(t, concat, uncompressed)
+		})
+	}
+}

op-batcher/flags/flags.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/urfave/cli"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
 	opservice "github.com/ethereum-optimism/optimism/op-service"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
@@ -67,24 +68,6 @@ var (
 		Value:  120_000,
 		EnvVar: opservice.PrefixEnvVar(EnvVarPrefix, "MAX_L1_TX_SIZE_BYTES"),
 	}
-	TargetL1TxSizeBytesFlag = cli.Uint64Flag{
-		Name:   "target-l1-tx-size-bytes",
-		Usage:  "The target size of a batch tx submitted to L1.",
-		Value:  100_000,
-		EnvVar: opservice.PrefixEnvVar(EnvVarPrefix, "TARGET_L1_TX_SIZE_BYTES"),
-	}
-	TargetNumFramesFlag = cli.IntFlag{
-		Name:   "target-num-frames",
-		Usage:  "The target number of frames to create per channel",
-		Value:  1,
-		EnvVar: opservice.PrefixEnvVar(EnvVarPrefix, "TARGET_NUM_FRAMES"),
-	}
-	ApproxComprRatioFlag = cli.Float64Flag{
-		Name:   "approx-compr-ratio",
-		Usage:  "The approximate compression ratio (<= 1.0)",
-		Value:  0.4,
-		EnvVar: opservice.PrefixEnvVar(EnvVarPrefix, "APPROX_COMPR_RATIO"),
-	}
 	StoppedFlag = cli.BoolFlag{
 		Name:   "stopped",
 		Usage:  "Initialize the batcher in a stopped state. The batcher can be started using the admin_startBatcher RPC",
@@ -106,9 +89,6 @@ var optionalFlags = []cli.Flag{
 	MaxPendingTransactionsFlag,
 	MaxChannelDurationFlag,
 	MaxL1TxSizeBytesFlag,
-	TargetL1TxSizeBytesFlag,
-	TargetNumFramesFlag,
-	ApproxComprRatioFlag,
 	StoppedFlag,
 	SequencerHDPathFlag,
 }
@@ -120,6 +100,7 @@ func init() {
 	optionalFlags = append(optionalFlags, oppprof.CLIFlags(EnvVarPrefix)...)
 	optionalFlags = append(optionalFlags, rpc.CLIFlags(EnvVarPrefix)...)
 	optionalFlags = append(optionalFlags, txmgr.CLIFlags(EnvVarPrefix)...)
+	optionalFlags = append(optionalFlags, compressor.CLIFlags(EnvVarPrefix)...)
 
 	Flags = append(requiredFlags, optionalFlags...)
 }

op-chain-ops/cmd/op-migrate/main.go

@@ -202,6 +202,10 @@ func main() {
 
 			dryRun := ctx.Bool("dry-run")
 			noCheck := ctx.Bool("no-check")
+			if noCheck {
+				panic("must run with check on")
+			}
+
 			// Perform the migration
 			res, err := genesis.MigrateDB(ldb, config, block, &migrationData, !dryRun, noCheck)
 			if err != nil {

op-chain-ops/crossdomain/legacy_withdrawal.go

@@ -57,7 +57,9 @@ func (w *LegacyWithdrawal) Encode() ([]byte, error) {
 	return out, nil
 }
 
-// Decode will decode a serialized LegacyWithdrawal
+// Decode will decode a serialized LegacyWithdrawal. There is a known inconsistency
+// where the decoded `msg.sender` is not authenticated. A round trip of encoding and
+// decoding with a spoofed withdrawal will result in a different message being recovered.
 func (w *LegacyWithdrawal) Decode(data []byte) error {
 	if len(data) < len(predeploys.L2CrossDomainMessengerAddr)+4 {
 		return fmt.Errorf("withdrawal data too short: %d", len(data))
@@ -68,6 +70,8 @@ func (w *LegacyWithdrawal) Decode(data []byte) error {
 		return fmt.Errorf("invalid selector: 0x%x", data[0:4])
 	}
 
+	// This should be the L2CrossDomainMessenger address but is not guaranteed
+	// to be.
 	msgSender := data[len(data)-len(predeploys.L2CrossDomainMessengerAddr):]
 
 	raw := data[4 : len(data)-len(predeploys.L2CrossDomainMessengerAddr)]

op-e2e/actions/l2_batcher.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/stretchr/testify/require"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
@@ -133,7 +134,13 @@ func (s *L2Batcher) Buffer(t Testing) error {
 		if s.l2BatcherCfg.GarbageCfg != nil {
 			ch, err = NewGarbageChannelOut(s.l2BatcherCfg.GarbageCfg)
 		} else {
-			ch, err = derive.NewChannelOut()
+			c, e := compressor.NewRatioCompressor(compressor.Config{
+				TargetFrameSize:  s.l2BatcherCfg.MaxL1TxSize,
+				TargetNumFrames:  1,
+				ApproxComprRatio: 1,
+			})
+			require.NoError(t, e, "failed to create compressor")
+			ch, err = derive.NewChannelOut(c)
 		}
 		require.NoError(t, err, "failed to create channel")
 		s.l2ChannelOut = ch

op-e2e/migration_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	bss "github.com/ethereum-optimism/optimism/op-batcher/batcher"
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	batchermetrics "github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
@@ -335,12 +336,14 @@ func TestMigration(t *testing.T) {
 		RollupRpc:          rollupNode.HTTPEndpoint(),
 		MaxChannelDuration: 1,
 		MaxL1TxSize:        120_000,
-		TargetL1TxSize:     100_000,
-		TargetNumFrames:    1,
-		ApproxComprRatio:   0.4,
-		SubSafetyMargin:    4,
-		PollInterval:       50 * time.Millisecond,
-		TxMgrConfig:        newTxMgrConfig(forkedL1URL, secrets.Batcher),
+		CompressorConfig: compressor.CLIConfig{
+			TargetL1TxSizeBytes: 100_000,
+			TargetNumFrames:     1,
+			ApproxComprRatio:    0.4,
+		},
+		SubSafetyMargin: 4,
+		PollInterval:    50 * time.Millisecond,
+		TxMgrConfig:     newTxMgrConfig(forkedL1URL, secrets.Batcher),
 		LogConfig: oplog.CLIConfig{
 			Level:  "info",
 			Format: "text",

op-e2e/setup.go

@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	bss "github.com/ethereum-optimism/optimism/op-batcher/batcher"
+	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	batchermetrics "github.com/ethereum-optimism/optimism/op-batcher/metrics"
 	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
 	"github.com/ethereum-optimism/optimism/op-chain-ops/genesis"
@@ -599,12 +600,14 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 		MaxPendingTransactions: 1,
 		MaxChannelDuration:     1,
 		MaxL1TxSize:            120_000,
-		TargetL1TxSize:         100_000,
-		TargetNumFrames:        1,
-		ApproxComprRatio:       0.4,
-		SubSafetyMargin:        4,
-		PollInterval:           50 * time.Millisecond,
-		TxMgrConfig:            newTxMgrConfig(sys.Nodes["l1"].WSEndpoint(), cfg.Secrets.Batcher),
+		CompressorConfig: compressor.CLIConfig{
+			TargetL1TxSizeBytes: 100_000,
+			TargetNumFrames:     1,
+			ApproxComprRatio:    0.4,
+		},
+		SubSafetyMargin: 4,
+		PollInterval:    50 * time.Millisecond,
+		TxMgrConfig:     newTxMgrConfig(sys.Nodes["l1"].WSEndpoint(), cfg.Secrets.Batcher),
 		LogConfig: oplog.CLIConfig{
 			Level:  "info",
 			Format: "text",

op-node/node/server.go

@@ -37,7 +37,6 @@ func newRPCServer(ctx context.Context, rpcCfg *RPCConfig, rollupCfg *rollup.Conf
 		apis: []rpc.API{{
 			Namespace:     "optimism",
 			Service:       api,
-			Public:        true,
 			Authenticated: false,
 		}},
 		appVersion: appVersion,
@@ -51,7 +50,6 @@ func (s *rpcServer) EnableAdminAPI(api *adminAPI) {
 		Namespace:     "admin",
 		Version:       "",
 		Service:       api,
-		Public:        true, // TODO: this field is deprecated. Do we even need this anymore?
 		Authenticated: false,
 	})
 }
@@ -61,7 +59,6 @@ func (s *rpcServer) EnableP2P(backend *p2p.APIBackend) {
 		Namespace:     p2p.NamespaceRPC,
 		Version:       "",
 		Service:       backend,
-		Public:        true,
 		Authenticated: false,
 	})
 }

op-node/rollup/derive/channel_out.go

@@ -2,7 +2,6 @@ package derive
 
 import (
 	"bytes"
-	"compress/zlib"
 	"crypto/rand"
 	"errors"
 	"fmt"
@@ -25,6 +24,30 @@ var ErrTooManyRLPBytes = errors.New("batch would cause RLP bytes to go over limi
 // [Frame Format]: https://github.com/ethereum-optimism/optimism/blob/develop/specs/derivation.md#frame-format
 const FrameV0OverHeadSize = 23
 
+var CompressorFullErr = errors.New("compressor is full")
+
+type Compressor interface {
+	// Writer is used to write uncompressed data which will be compressed. Should return
+	// CompressorFullErr if the compressor is full and no more data should be written.
+	io.Writer
+	// Closer Close function should be called before reading any data.
+	io.Closer
+	// Reader is used to Read compressed data; should only be called after Close.
+	io.Reader
+	// Reset will reset all written data
+	Reset()
+	// Len returns an estimate of the current length of the compressed data; calling Flush will
+	// increase the accuracy at the expense of a poorer compression ratio.
+	Len() int
+	// Flush flushes any uncompressed data to the compression buffer. This will result in a
+	// non-optimal compression ratio.
+	Flush() error
+	// FullErr returns CompressorFullErr if the compressor is known to be full. Note that
+	// calls to Write will fail if an error is returned from this method, but calls to Write
+	// can still return CompressorFullErr even if this does not.
+	FullErr() error
+}
+
 type ChannelOut struct {
 	id ChannelID
 	// Frame ID of the next frame to emit. Increment after emitting
@@ -33,9 +56,7 @@ type ChannelOut struct {
 	rlpLength int
 
 	// Compressor stage. Write input data to it
-	compress *zlib.Writer
-	// post compression buffer
-	buf bytes.Buffer
+	compress Compressor
 
 	closed bool
 }
@@ -44,23 +65,18 @@ func (co *ChannelOut) ID() ChannelID {
 	return co.id
 }
 
-func NewChannelOut() (*ChannelOut, error) {
+func NewChannelOut(compress Compressor) (*ChannelOut, error) {
 	c := &ChannelOut{
 		id:        ChannelID{}, // TODO: use GUID here instead of fully random data
 		frame:     0,
 		rlpLength: 0,
+		compress:  compress,
 	}
 	_, err := rand.Read(c.id[:])
 	if err != nil {
 		return nil, err
 	}
 
-	compress, err := zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
-	if err != nil {
-		return nil, err
-	}
-	c.compress = compress
-
 	return c, nil
 }
 
@@ -68,8 +84,7 @@ func NewChannelOut() (*ChannelOut, error) {
 func (co *ChannelOut) Reset() error {
 	co.frame = 0
 	co.rlpLength = 0
-	co.buf.Reset()
-	co.compress.Reset(&co.buf)
+	co.compress.Reset()
 	co.closed = false
 	_, err := rand.Read(co.id[:])
 	return err
@@ -116,7 +131,8 @@ func (co *ChannelOut) AddBatch(batch *BatchData) (uint64, error) {
 	}
 	co.rlpLength += buf.Len()
 
-	written, err := io.Copy(co.compress, &buf)
+	// avoid using io.Copy here, because we need all or nothing
+	written, err := co.compress.Write(buf.Bytes())
 	return uint64(written), err
 }
 
@@ -129,7 +145,7 @@ func (co *ChannelOut) InputBytes() int {
 // Use `Flush` or `Close` to move data from the compression buffer into the ready buffer if more bytes
 // are needed. Add blocks may add to the ready buffer, but it is not guaranteed due to the compression stage.
 func (co *ChannelOut) ReadyBytes() int {
-	return co.buf.Len()
+	return co.compress.Len()
 }
 
 // Flush flushes the internal compression stage to the ready buffer. It enables pulling a larger & more
@@ -138,6 +154,10 @@ func (co *ChannelOut) Flush() error {
 	return co.compress.Flush()
 }
 
+func (co *ChannelOut) FullErr() error {
+	return co.compress.FullErr()
+}
+
 func (co *ChannelOut) Close() error {
 	if co.closed {
 		return errors.New("already closed")
@@ -166,8 +186,8 @@ func (co *ChannelOut) OutputFrame(w *bytes.Buffer, maxSize uint64) (uint16, erro
 
 	// Copy data from the local buffer into the frame data buffer
 	maxDataSize := maxSize - FrameV0OverHeadSize
-	if maxDataSize > uint64(co.buf.Len()) {
-		maxDataSize = uint64(co.buf.Len())
+	if maxDataSize > uint64(co.compress.Len()) {
+		maxDataSize = uint64(co.compress.Len())
 		// If we are closed & will not spill past the current frame
 		// mark it is the final frame of the channel.
 		if co.closed {
@@ -176,7 +196,7 @@ func (co *ChannelOut) OutputFrame(w *bytes.Buffer, maxSize uint64) (uint16, erro
 	}
 	f.Data = make([]byte, maxDataSize)
 
-	if _, err := io.ReadFull(&co.buf, f.Data); err != nil {
+	if _, err := io.ReadFull(co.compress, f.Data); err != nil {
 		return 0, err
 	}
 

op-node/rollup/derive/channel_out_test.go

@@ -11,8 +11,25 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// basic implementation of the Compressor interface that does no compression
+type nonCompressor struct {
+	bytes.Buffer
+}
+
+func (s *nonCompressor) Flush() error {
+	return nil
+}
+
+func (s *nonCompressor) Close() error {
+	return nil
+}
+
+func (s *nonCompressor) FullErr() error {
+	return nil
+}
+
 func TestChannelOutAddBlock(t *testing.T) {
-	cout, err := NewChannelOut()
+	cout, err := NewChannelOut(&nonCompressor{})
 	require.NoError(t, err)
 
 	t.Run("returns err if first tx is not an l1info tx", func(t *testing.T) {
@@ -33,7 +50,7 @@ func TestChannelOutAddBlock(t *testing.T) {
 // max size that is below the fixed frame size overhead of 23, will return
 // an error.
 func TestOutputFrameSmallMaxSize(t *testing.T) {
-	cout, err := NewChannelOut()
+	cout, err := NewChannelOut(&nonCompressor{})
 	require.NoError(t, err)
 
 	// Call OutputFrame with the range of small max size values that err

packages/contracts-bedrock/deploy-config/final-migration-rehearsal.json

-{
-  "finalSystemOwner": "0x62790eFcB3a5f3A5D398F95B47930A9Addd83807",
-  "portalGuardian": "0x62790eFcB3a5f3A5D398F95B47930A9Addd83807",
-  "controller": "0x2d30335B0b807bBa1682C487BaAFD2Ad6da5D675",
-
-  "l1StartingBlockTag": "0x4104895a540d87127ff11eef0d51d8f63ce00a6fc211db751a45a4b3a61a9c83",
-  "l1ChainID": 5,
-  "l2ChainID": 420,
-  "l2BlockTime": 2,
-
-  "maxSequencerDrift": 1200,
-  "sequencerWindowSize": 3600,
-  "channelTimeout": 120,
-
-  "p2pSequencerAddress": "0xCBABF46d40982B4530c0EAc9889f6e44e17f0383",
-  "batchInboxAddress": "0xff00000000000000000000000000000000000420",
-  "batchSenderAddress": "0x3a2baA0160275024A50C1be1FC677375E7DB4Bd7",
-  "l2OutputOracleSubmissionInterval": 20,
-  "l2OutputOracleStartingTimestamp": 1670625264,
-  "l2OutputOracleStartingBlockNumber": 3324764,
-  "l2OutputOracleProposer": "0x88BCa4Af3d950625752867f826E073E337076581",
-  "l2OutputOracleChallenger": "0x88BCa4Af3d950625752867f826E073E337076581",
-  "finalizationPeriodSeconds": 2,
-
-  "proxyAdminOwner": "0x62790eFcB3a5f3A5D398F95B47930A9Addd83807",
-
-  "governanceTokenName": "Optimism",
-  "governanceTokenSymbol": "OP",
-  "governanceTokenOwner": "0x038a8825A3C3B0c08d52Cc76E5E361953Cf6Dc76",
-
-  "l2GenesisBlockGasLimit": "0x1c9c380",
-  "l2GenesisBlockCoinbase": "0x4200000000000000000000000000000000000011",
-  "l2GenesisBlockBaseFeePerGas": "0x3b9aca00",
-
-
-  "gasPriceOracleOverhead": 2100,
-  "gasPriceOracleScalar": 1000000,
-  "eip1559Denominator": 50,
-  "eip1559Elasticity": 10
-}

packages/contracts-bedrock/deploy-config/final-migration-rehearsal.ts

-import { DeployConfig } from '../src/deploy-config'
-import config from './final-migration-rehearsal.json'
-
-export default config satisfies DeployConfig

packages/contracts-bedrock/deploy-config/mainnet.json

@@ -19,9 +19,9 @@
   "l2OutputOracleChallenger": "0x3C44CdDdB6a900fa2b585dd299e03d12FA4293BC",
   "finalizationPeriodSeconds": 2,
   "proxyAdminOwner": "0x90F79bf6EB2c4f870365E785982E1f101E93b906",
-  "baseFeeVaultRecipient": "0x90F79bf6EB2c4f870365E785982E1f101E93b906",
-  "l1FeeVaultRecipient": "0x90F79bf6EB2c4f870365E785982E1f101E93b906",
-  "sequencerFeeVaultRecipient": "0x90F79bf6EB2c4f870365E785982E1f101E93b906",
+  "baseFeeVaultRecipient": "0xa3d596EAfaB6B13Ab18D40FaE1A962700C84ADEa",
+  "l1FeeVaultRecipient": "0xa3d596EAfaB6B13Ab18D40FaE1A962700C84ADEa",
+  "sequencerFeeVaultRecipient": "0xa3d596EAfaB6B13Ab18D40FaE1A962700C84ADEa",
   "governanceTokenName": "Optimism",
   "governanceTokenSymbol": "OP",
   "governanceTokenOwner": "0x90F79bf6EB2c4f870365E785982E1f101E93b906",

packages/contracts-bedrock/hardhat.config.ts

@@ -93,12 +93,6 @@ const config: HardhatUserConfig = {
       accounts: [process.env.PRIVATE_KEY_DEPLOYER || ethers.constants.HashZero],
       live: true,
     },
-    'final-migration-rehearsal': {
-      chainId: 5,
-      url: process.env.L1_RPC || '',
-      accounts: [process.env.PRIVATE_KEY_DEPLOYER || ethers.constants.HashZero],
-      live: true,
-    },
     'internal-devnet': {
       chainId: 5,
       url: process.env.L1_RPC || '',
@@ -149,10 +143,6 @@ const config: HardhatUserConfig = {
         '../contracts/deployments/goerli',
         '../contracts-periphery/deployments/goerli',
       ],
-      'final-migration-rehearsal': [
-        '../contracts/deployments/goerli',
-        '../contracts-periphery/deployments/goerli',
-      ],
     },
   },
   solidity: {

packages/fault-detector/src/service.ts

@@ -46,7 +46,7 @@ type State = {
   fpw: number
   oo: OutputOracle<any>
   messenger: CrossChainMessenger
-  highestCheckedBatchIndex: number
+  currentBatchIndex: number
   diverged: boolean
 }
 
@@ -254,27 +254,22 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
       // but it happens often on testnets because the FPW is very short.
       if (firstUnfinalized === undefined) {
         this.logger.info(`no unfinalized batches found, starting from latest`)
-        this.state.highestCheckedBatchIndex = (
+        this.state.currentBatchIndex = (
           await this.state.oo.getTotalElements()
         ).toNumber()
       } else {
-        this.state.highestCheckedBatchIndex = firstUnfinalized
+        this.state.currentBatchIndex = firstUnfinalized
       }
     } else {
-      this.state.highestCheckedBatchIndex = this.options.startBatchIndex
+      this.state.currentBatchIndex = this.options.startBatchIndex
     }
 
-    this.logger.info(`starting height`, {
-      startBatchIndex: this.state.highestCheckedBatchIndex,
+    this.logger.info('starting height', {
+      startBatchIndex: this.state.currentBatchIndex,
     })
 
     // Set the initial metrics.
-    this.metrics.highestBatchIndex.set(
-      {
-        type: 'checked',
-      },
-      this.state.highestCheckedBatchIndex
-    )
+    this.metrics.isCurrentlyMismatched.set(0)
   }
 
   async routes(router: ExpressRouter): Promise<void> {
@@ -286,6 +281,8 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
   }
 
   async main(): Promise<void> {
+    const startMs = Date.now()
+
     let latestBatchIndex: number
     try {
       latestBatchIndex = (await this.state.oo.getTotalElements()).toNumber()
@@ -303,7 +300,7 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
       return
     }
 
-    if (this.state.highestCheckedBatchIndex >= latestBatchIndex) {
+    if (this.state.currentBatchIndex >= latestBatchIndex) {
       await sleep(15000)
       return
     } else {
@@ -316,7 +313,7 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
     }
 
     this.logger.info(`checking batch`, {
-      batchIndex: this.state.highestCheckedBatchIndex,
+      batchIndex: this.state.currentBatchIndex,
       latestIndex: latestBatchIndex,
     })
 
@@ -324,7 +321,7 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
     try {
       event = await findEventForStateBatch(
         this.state.oo,
-        this.state.highestCheckedBatchIndex
+        this.state.currentBatchIndex
       )
     } catch (err) {
       this.logger.error(`got error when connecting to node`, {
@@ -528,20 +525,24 @@ export class FaultDetector extends BaseServiceV2<Options, Metrics, State> {
       }
     }
 
-    this.logger.info(`checked batch ok`, {
-      batchIndex: this.state.highestCheckedBatchIndex,
-    })
+    const elapsedMs = Date.now() - startMs
 
-    this.state.highestCheckedBatchIndex++
+    // Mark the current batch index as checked
+    this.logger.info('checked batch ok', {
+      batchIndex: this.state.currentBatchIndex,
+      timeMs: elapsedMs,
+    })
     this.metrics.highestBatchIndex.set(
       {
         type: 'checked',
       },
-      this.state.highestCheckedBatchIndex
+      this.state.currentBatchIndex
     )
 
-    // If we got through the above without throwing an error, we should be fine to reset.
+    // If we got through the above without throwing an error, we should be
+    // fine to reset and move onto the next batch
     this.state.diverged = false
+    this.state.currentBatchIndex++
     this.metrics.isCurrentlyMismatched.set(0)
   }
 }

proxyd/cache.go

@@ -2,6 +2,7 @@ package proxyd
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -43,16 +44,24 @@ func (c *cache) Put(ctx context.Context, key string, value string) error {
 }
 
 type redisCache struct {
-	rdb *redis.Client
+	rdb    *redis.Client
+	prefix string
 }
 
-func newRedisCache(rdb *redis.Client) *redisCache {
-	return &redisCache{rdb}
+func newRedisCache(rdb *redis.Client, prefix string) *redisCache {
+	return &redisCache{rdb, prefix}
+}
+
+func (c *redisCache) namespaced(key string) string {
+	if c.prefix == "" {
+		return key
+	}
+	return strings.Join([]string{c.prefix, key}, ":")
 }
 
 func (c *redisCache) Get(ctx context.Context, key string) (string, error) {
 	start := time.Now()
-	val, err := c.rdb.Get(ctx, key).Result()
+	val, err := c.rdb.Get(ctx, c.namespaced(key)).Result()
 	redisCacheDurationSumm.WithLabelValues("GET").Observe(float64(time.Since(start).Milliseconds()))
 
 	if err == redis.Nil {
@@ -66,7 +75,7 @@ func (c *redisCache) Get(ctx context.Context, key string) (string, error) {
 
 func (c *redisCache) Put(ctx context.Context, key string, value string) error {
 	start := time.Now()
-	err := c.rdb.SetEX(ctx, key, value, redisTTL).Err()
+	err := c.rdb.SetEX(ctx, c.namespaced(key), value, redisTTL).Err()
 	redisCacheDurationSumm.WithLabelValues("SETEX").Observe(float64(time.Since(start).Milliseconds()))
 
 	if err != nil {

proxyd/config.go

@@ -32,7 +32,8 @@ type CacheConfig struct {
 }
 
 type RedisConfig struct {
-	URL string `toml:"url"`
+	URL       string `toml:"url"`
+	Namespace string `toml:"namespace"`
 }
 
 type MetricsConfig struct {

proxyd/consensus_poller.go

@@ -17,11 +17,14 @@ const (
 	PollerInterval = 1 * time.Second
 )
 
+type OnConsensusBroken func()
+
 // ConsensusPoller checks the consensus state for each member of a BackendGroup
 // resolves the highest common block for multiple nodes, and reconciles the consensus
 // in case of block hash divergence to minimize re-orgs
 type ConsensusPoller struct {
 	cancelFunc context.CancelFunc
+	listeners  []OnConsensusBroken
 
 	backendGroup      *BackendGroup
 	backendState      map[*Backend]*backendState
@@ -44,6 +47,7 @@ type backendState struct {
 	latestBlockNumber hexutil.Uint64
 	latestBlockHash   string
 	peerCount         uint64
+	inSync            bool
 
 	lastUpdate time.Time
 
@@ -149,6 +153,16 @@ func WithAsyncHandler(asyncHandler ConsensusAsyncHandler) ConsensusOpt {
 	}
 }
 
+func WithListener(listener OnConsensusBroken) ConsensusOpt {
+	return func(cp *ConsensusPoller) {
+		cp.AddListener(listener)
+	}
+}
+
+func (cp *ConsensusPoller) AddListener(listener OnConsensusBroken) {
+	cp.listeners = append(cp.listeners, listener)
+}
+
 func WithBanPeriod(banPeriod time.Duration) ConsensusOpt {
 	return func(cp *ConsensusPoller) {
 		cp.banPeriod = banPeriod
@@ -215,7 +229,13 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 	RecordConsensusBackendBanned(be, banned)
 
 	if banned {
-		log.Debug("skipping backend banned", "backend", be.Name)
+		log.Debug("skipping backend - banned", "backend", be.Name)
+		return
+	}
+
+	// if backend exhausted rate limit we'll skip it for now
+	if be.IsRateLimited() {
+		log.Debug("skipping backend - rate limited", "backend", be.Name)
 		return
 	}
 
@@ -228,24 +248,16 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 
 	// if backend it not in sync we'll check again after ban
 	inSync, err := cp.isInSync(ctx, be)
-	if err != nil || !inSync {
-		log.Warn("backend banned - not in sync", "backend", be.Name)
-		cp.Ban(be)
-		return
-	}
-	RecordConsensusBackendInSync(be, inSync)
-
-	// if backend exhausted rate limit we'll skip it for now
-	if be.IsRateLimited() {
-		return
+	RecordConsensusBackendInSync(be, err == nil && inSync)
+	if err != nil {
+		log.Warn("error updating backend sync state", "name", be.Name, "err", err)
 	}
 
 	var peerCount uint64
 	if !be.skipPeerCountCheck {
 		peerCount, err = cp.getPeerCount(ctx, be)
 		if err != nil {
-			log.Warn("error updating backend", "name", be.Name, "err", err)
-			return
+			log.Warn("error updating backend peer count", "name", be.Name, "err", err)
 		}
 		RecordConsensusBackendPeerCount(be, peerCount)
 	}
@@ -253,10 +265,9 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 	latestBlockNumber, latestBlockHash, err := cp.fetchBlock(ctx, be, "latest")
 	if err != nil {
 		log.Warn("error updating backend", "name", be.Name, "err", err)
-		return
 	}
 
-	changed, updateDelay := cp.setBackendState(be, peerCount, latestBlockNumber, latestBlockHash)
+	changed, updateDelay := cp.setBackendState(be, peerCount, inSync, latestBlockNumber, latestBlockHash)
 
 	if changed {
 		RecordBackendLatestBlock(be, latestBlockNumber)
@@ -264,6 +275,7 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 		log.Debug("backend state updated",
 			"name", be.Name,
 			"peerCount", peerCount,
+			"inSync", inSync,
 			"latestBlockNumber", latestBlockNumber,
 			"latestBlockHash", latestBlockHash,
 			"updateDelay", updateDelay)
@@ -280,11 +292,14 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 
 	// find the highest block, in order to use it defining the highest non-lagging ancestor block
 	for _, be := range cp.backendGroup.Backends {
-		peerCount, backendLatestBlockNumber, _, lastUpdate, _ := cp.getBackendState(be)
+		peerCount, inSync, backendLatestBlockNumber, _, lastUpdate, _ := cp.getBackendState(be)
 
 		if !be.skipPeerCountCheck && peerCount < cp.minPeerCount {
 			continue
 		}
+		if !inSync {
+			continue
+		}
 		if lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now()) {
 			continue
 		}
@@ -296,11 +311,14 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 
 	// find the highest common ancestor block
 	for _, be := range cp.backendGroup.Backends {
-		peerCount, backendLatestBlockNumber, backendLatestBlockHash, lastUpdate, _ := cp.getBackendState(be)
+		peerCount, inSync, backendLatestBlockNumber, backendLatestBlockHash, lastUpdate, _ := cp.getBackendState(be)
 
 		if !be.skipPeerCountCheck && peerCount < cp.minPeerCount {
 			continue
 		}
+		if !inSync {
+			continue
+		}
 		if lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now()) {
 			continue
 		}
@@ -349,14 +367,15 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 				- with minimum peer count
 				- updated recently
 				- not lagging
+				- in sync
 			*/
 
-			peerCount, latestBlockNumber, _, lastUpdate, bannedUntil := cp.getBackendState(be)
+			peerCount, inSync, latestBlockNumber, _, lastUpdate, bannedUntil := cp.getBackendState(be)
 			notUpdated := lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now())
 			isBanned := time.Now().Before(bannedUntil)
 			notEnoughPeers := !be.skipPeerCountCheck && peerCount < cp.minPeerCount
 			lagging := latestBlockNumber < proposedBlock
-			if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers || lagging {
+			if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers || lagging || !inSync {
 				filteredBackendsNames = append(filteredBackendsNames, be.Name)
 				continue
 			}
@@ -392,7 +411,9 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 	}
 
 	if broken {
-		// propagate event to other interested parts, such as cache invalidator
+		for _, l := range cp.listeners {
+			l()
+		}
 		log.Info("consensus broken", "currentConsensusBlockNumber", currentConsensusBlockNumber, "proposedBlock", proposedBlock, "proposedBlockHash", proposedBlockHash)
 	}
 
@@ -498,23 +519,25 @@ func (cp *ConsensusPoller) isInSync(ctx context.Context, be *Backend) (result bo
 	return res, nil
 }
 
-func (cp *ConsensusPoller) getBackendState(be *Backend) (peerCount uint64, blockNumber hexutil.Uint64, blockHash string, lastUpdate time.Time, bannedUntil time.Time) {
+func (cp *ConsensusPoller) getBackendState(be *Backend) (peerCount uint64, inSync bool, blockNumber hexutil.Uint64, blockHash string, lastUpdate time.Time, bannedUntil time.Time) {
 	bs := cp.backendState[be]
+	defer bs.backendStateMux.Unlock()
 	bs.backendStateMux.Lock()
 	peerCount = bs.peerCount
+	inSync = bs.inSync
 	blockNumber = bs.latestBlockNumber
 	blockHash = bs.latestBlockHash
 	lastUpdate = bs.lastUpdate
 	bannedUntil = bs.bannedUntil
-	bs.backendStateMux.Unlock()
 	return
 }
 
-func (cp *ConsensusPoller) setBackendState(be *Backend, peerCount uint64, blockNumber hexutil.Uint64, blockHash string) (changed bool, updateDelay time.Duration) {
+func (cp *ConsensusPoller) setBackendState(be *Backend, peerCount uint64, inSync bool, blockNumber hexutil.Uint64, blockHash string) (changed bool, updateDelay time.Duration) {
 	bs := cp.backendState[be]
 	bs.backendStateMux.Lock()
 	changed = bs.latestBlockHash != blockHash
 	bs.peerCount = peerCount
+	bs.inSync = inSync
 	bs.latestBlockNumber = blockNumber
 	bs.latestBlockHash = blockHash
 	updateDelay = time.Since(bs.lastUpdate)

proxyd/integration_tests/consensus_test.go

@@ -94,6 +94,7 @@ func TestConsensus(t *testing.T) {
 		consensusGroup := bg.Consensus.GetConsensusGroup()
 
 		require.NotContains(t, consensusGroup, be)
+		require.False(t, bg.Consensus.IsBanned(be))
 		require.Equal(t, 1, len(consensusGroup))
 	})
 
@@ -132,6 +133,7 @@ func TestConsensus(t *testing.T) {
 		be := backend(bg, "node1")
 		require.NotNil(t, be)
 		require.NotContains(t, consensusGroup, be)
+		require.False(t, bg.Consensus.IsBanned(be))
 		require.Equal(t, 1, len(consensusGroup))
 	})
 
@@ -232,6 +234,7 @@ func TestConsensus(t *testing.T) {
 		consensusGroup := bg.Consensus.GetConsensusGroup()
 
 		require.NotContains(t, consensusGroup, be)
+		require.False(t, bg.Consensus.IsBanned(be))
 		require.Equal(t, 1, len(consensusGroup))
 	})
 
@@ -286,6 +289,11 @@ func TestConsensus(t *testing.T) {
 		h2.ResetOverrides()
 		bg.Consensus.Unban()
 
+		listenerCalled := false
+		bg.Consensus.AddListener(func() {
+			listenerCalled = true
+		})
+
 		for _, be := range bg.Backends {
 			bg.Consensus.UpdateBackend(ctx, be)
 		}
@@ -331,7 +339,7 @@ func TestConsensus(t *testing.T) {
 		// should resolve to 0x1, since 0x2 is out of consensus at the moment
 		require.Equal(t, "0x1", bg.Consensus.GetConsensusBlockNumber().String())
 
-		// later, when impl events, listen to broken consensus event
+		require.True(t, listenerCalled)
 	})
 
 	t.Run("broken consensus with depth 2", func(t *testing.T) {

proxyd/integration_tests/testdata/caching.toml

@@ -6,6 +6,7 @@ response_timeout_seconds = 1
 
 [redis]
 url = "$REDIS_URL"
+namespace = "proxyd"
 
 [cache]
 enabled = true

proxyd/proxyd.go

@@ -236,7 +236,7 @@ func Start(config *Config) (*Server, func(), error) {
 			log.Warn("redis is not configured, using in-memory cache")
 			cache = newMemoryCache()
 		} else {
-			cache = newRedisCache(redisClient)
+			cache = newRedisCache(redisClient, config.Redis.Namespace)
 		}
 		// Ideally, the BlocKSyncRPCURL should be the sequencer or a HA replica that's not far behind
 		ethClient, err := ethclient.Dial(blockSyncRPCURL)