Fjord: Add Brotli channel compression support

go.mod

@@ -4,6 +4,7 @@ go 1.21
 
 require (
 	github.com/BurntSushi/toml v1.3.2
+	github.com/andybalholm/brotli v1.1.0
 	github.com/btcsuite/btcd v0.24.0
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0
 	github.com/cockroachdb/pebble v0.0.0-20231018212520-f6cde3fc2fa4
@@ -64,7 +65,6 @@ require (
 	github.com/Shopify/goreferrer v0.0.0-20220729165902-8cddb4f5de06 // indirect
 	github.com/VictoriaMetrics/fastcache v1.12.1 // indirect
 	github.com/allegro/bigcache v1.2.1 // indirect
-	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/aymerick/douceur v0.2.0 // indirect
 	github.com/benbjohnson/clock v1.3.5 // indirect

op-batcher/batcher/channel_builder.go

@@ -86,7 +86,7 @@ func NewChannelBuilder(cfg ChannelConfig, rollupCfg rollup.Config, latestL1Origi
 	}
 	var co derive.ChannelOut
 	if cfg.BatchType == derive.SpanBatchType {
-		co, err = derive.NewSpanChannelOut(rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID, cfg.CompressorConfig.TargetOutputSize)
+		co, err = derive.NewSpanChannelOut(rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID, cfg.CompressorConfig.TargetOutputSize, cfg.CompressorConfig.CompressionAlgo, cfg.CompressorConfig.CompressLevel)
 	} else {
 		co, err = derive.NewSingularChannelOut(c)
 	}

op-batcher/batcher/channel_builder_test.go

@@ -393,7 +393,7 @@ func TestChannelBuilder_OutputFrames(t *testing.T) {
 	channelConfig := defaultTestChannelConfig()
 	channelConfig.MaxFrameSize = derive.FrameV0OverHeadSize + 1
 	channelConfig.TargetNumFrames = 1000
-	channelConfig.InitNoneCompressor()
+	channelConfig.InitNoneCompressor("brotli", 10)
 
 	// Construct the channel builder
 	cb, err := NewChannelBuilder(channelConfig, defaultTestRollupConfig, latestL1BlockOrigin)
@@ -430,54 +430,72 @@ func TestChannelBuilder_OutputFrames(t *testing.T) {
 }
 
 func TestChannelBuilder_OutputFrames_SpanBatch(t *testing.T) {
-	channelConfig := defaultTestChannelConfig()
-	channelConfig.MaxFrameSize = 20 + derive.FrameV0OverHeadSize
-	channelConfig.TargetNumFrames = 5
-	channelConfig.BatchType = derive.SpanBatchType
-	channelConfig.InitRatioCompressor(1)
-
-	// Construct the channel builder
-	cb, err := NewChannelBuilder(channelConfig, defaultTestRollupConfig, latestL1BlockOrigin)
-	require.NoError(t, err)
-	require.False(t, cb.IsFull())
-	require.Equal(t, 0, cb.PendingFrames())
-
-	// Calling OutputFrames without having called [AddBlock]
-	// should return no error
-	require.NoError(t, cb.OutputFrames())
-
-	// There should be no ready bytes yet
-	require.Equal(t, 0, cb.co.ReadyBytes())
-
-	// fill up
-	for {
-		err = addMiniBlock(cb)
-		if err == nil {
-			require.False(t, cb.IsFull())
-			// There should be no ready bytes until the channel is full
-			require.Equal(t, cb.co.ReadyBytes(), 0)
-		} else {
-			require.ErrorIs(t, err, derive.ErrCompressorFull)
-			break
-		}
+	testCases := []struct {
+		testName        string
+		algo            string
+		compressLevel   int
+		targetNumFrames int
+	}{
+		{testName: "Span Batch output frames with zlib", algo: "zlib", compressLevel: 9, targetNumFrames: 5},
+		{testName: "Span Batch output frames with brotli", algo: "brotli", compressLevel: 10, targetNumFrames: 1}, // to fill faster
 	}
 
-	require.True(t, cb.IsFull())
-	// Check how many ready bytes
-	require.GreaterOrEqual(t,
-		cb.co.ReadyBytes()+derive.FrameV0OverHeadSize,
-		int(channelConfig.MaxFrameSize))
-	require.Equal(t, 0, cb.PendingFrames())
-
-	// We should be able to output the frames
-	require.NoError(t, cb.OutputFrames())
-
-	// There should be several frames in the channel builder now
-	require.Greater(t, cb.PendingFrames(), 1)
-	for i := 0; i < cb.numFrames-1; i++ {
-		require.Len(t, cb.frames[i].data, int(channelConfig.MaxFrameSize))
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			channelConfig := defaultTestChannelConfig()
+			channelConfig.MaxFrameSize = 20 + derive.FrameV0OverHeadSize
+			channelConfig.TargetNumFrames = tc.targetNumFrames
+			channelConfig.BatchType = derive.SpanBatchType
+			channelConfig.InitRatioCompressor(1, tc.algo, tc.compressLevel)
+
+			// Construct the channel builder
+			cb, err := NewChannelBuilder(channelConfig, defaultTestRollupConfig, latestL1BlockOrigin)
+			require.NoError(t, err)
+			require.False(t, cb.IsFull())
+			require.Equal(t, 0, cb.PendingFrames())
+
+			// Calling OutputFrames without having called [AddBlock]
+			// should return no error
+			require.NoError(t, cb.OutputFrames())
+
+			// There should be no ready bytes yet
+			require.Equal(t, 0, cb.co.ReadyBytes())
+
+			// fill up
+			for {
+				err = addMiniBlock(cb)
+				if err == nil {
+					if cb.IsFull() {
+						// this happens when the data exactly fills the channel
+						break
+					}
+					require.False(t, cb.IsFull())
+					// There should be no ready bytes until the channel is full
+					require.Equal(t, cb.co.ReadyBytes(), 0)
+				} else {
+					require.ErrorIs(t, err, derive.ErrCompressorFull)
+					break
+				}
+			}
+
+			require.True(t, cb.IsFull())
+			// Check how many ready bytes
+			require.GreaterOrEqual(t,
+				cb.co.ReadyBytes()+derive.FrameV0OverHeadSize,
+				int(channelConfig.MaxFrameSize))
+			require.Equal(t, 0, cb.PendingFrames())
+
+			// We should be able to output the frames
+			require.NoError(t, cb.OutputFrames())
+
+			// There should be several frames in the channel builder now
+			require.Greater(t, cb.PendingFrames(), 1)
+			for i := 0; i < cb.numFrames-1; i++ {
+				require.Len(t, cb.frames[i].data, int(channelConfig.MaxFrameSize))
+			}
+			require.LessOrEqual(t, len(cb.frames[len(cb.frames)-1].data), int(channelConfig.MaxFrameSize))
+		})
 	}
-	require.LessOrEqual(t, len(cb.frames[len(cb.frames)-1].data), int(channelConfig.MaxFrameSize))
 }
 
 // ChannelBuilder_MaxRLPBytesPerChannel tests the [ChannelBuilder.OutputFrames]
@@ -486,7 +504,7 @@ func ChannelBuilder_MaxRLPBytesPerChannel(t *testing.T, batchType uint) {
 	t.Parallel()
 	channelConfig := defaultTestChannelConfig()
 	channelConfig.MaxFrameSize = derive.MaxRLPBytesPerChannel * 2
-	channelConfig.InitNoneCompressor()
+	channelConfig.InitNoneCompressor("brotli", 10)
 	channelConfig.BatchType = batchType
 
 	// Construct the channel builder
@@ -504,7 +522,7 @@ func ChannelBuilder_OutputFramesMaxFrameIndex(t *testing.T, batchType uint) {
 	channelConfig := defaultTestChannelConfig()
 	channelConfig.MaxFrameSize = derive.FrameV0OverHeadSize + 1
 	channelConfig.TargetNumFrames = math.MaxUint16 + 1
-	channelConfig.InitRatioCompressor(.1)
+	channelConfig.InitRatioCompressor(.1, "brotli", 10)
 	channelConfig.BatchType = batchType
 
 	rng := rand.New(rand.NewSource(123))
@@ -546,27 +564,43 @@ func TestChannelBuilder_FullShadowCompressor(t *testing.T) {
 		TargetNumFrames: 1,
 		BatchType:       derive.SpanBatchType,
 	}
-	cfg.InitShadowCompressor()
 
-	cb, err := NewChannelBuilder(cfg, defaultTestRollupConfig, latestL1BlockOrigin)
-	require.NoError(err)
+	testCases := []struct {
+		testName        string
+		algo            string
+		compressLevel   int
+		targetNumFrames int
+	}{
+		{testName: "Full shadow compressor with zlib", algo: "zlib", compressLevel: 9},
+		{testName: "Full shadow compressor with brotli", algo: "brotli", compressLevel: 10},
+	}
 
-	rng := rand.New(rand.NewSource(420))
-	a := dtest.RandomL2BlockWithChainId(rng, 1, defaultTestRollupConfig.L2ChainID)
-	_, err = cb.AddBlock(a)
-	require.NoError(err)
-	_, err = cb.AddBlock(a)
-	require.ErrorIs(err, derive.ErrCompressorFull)
-	// without fix, adding the second block would succeed and then adding a
-	// third block would fail with full error and the compressor would be full.
+	for _, tc := range testCases {
+		t.Run(tc.testName, func(t *testing.T) {
+			cfg.InitShadowCompressor(tc.algo, tc.compressLevel)
 
-	require.NoError(cb.OutputFrames())
+			cb, err := NewChannelBuilder(cfg, defaultTestRollupConfig, latestL1BlockOrigin)
+			require.NoError(err)
+
+			rng := rand.New(rand.NewSource(420))
+			a := dtest.RandomL2BlockWithChainId(rng, 1, defaultTestRollupConfig.L2ChainID)
+
+			for {
+				_, err = cb.AddBlock(a)
+				if err != nil {
+					require.ErrorIs(err, derive.ErrCompressorFull)
+					break
+				}
+			}
 
-	require.True(cb.HasFrame())
-	f := cb.NextFrame()
-	require.Less(len(f.data), int(cfg.MaxFrameSize)) // would fail without fix, full frame
+			require.NoError(cb.OutputFrames())
+			require.True(cb.HasFrame())
+			f := cb.NextFrame()
+			require.Less(len(f.data), int(cfg.MaxFrameSize)) // would fail without fix, full frame
 
-	require.False(cb.HasFrame(), "no leftover frame expected") // would fail without fix
+			require.False(cb.HasFrame(), "no leftover frame expected") // would fail without fix
+		})
+	}
 }
 
 func ChannelBuilder_AddBlock(t *testing.T, batchType uint) {
@@ -577,7 +611,7 @@ func ChannelBuilder_AddBlock(t *testing.T, batchType uint) {
 	channelConfig.MaxFrameSize = 20 + derive.FrameV0OverHeadSize
 	channelConfig.TargetNumFrames = 2
 	// Configure the Input Threshold params so we observe a full channel
-	channelConfig.InitRatioCompressor(1)
+	channelConfig.InitRatioCompressor(1, "brotli", 10)
 
 	// Construct the channel builder
 	cb, err := NewChannelBuilder(channelConfig, defaultTestRollupConfig, latestL1BlockOrigin)
@@ -700,7 +734,7 @@ func ChannelBuilder_PendingFrames_TotalFrames(t *testing.T, batchType uint) {
 	cfg.MaxFrameSize = 1000
 	cfg.TargetNumFrames = tnf
 	cfg.BatchType = batchType
-	cfg.InitShadowCompressor()
+	cfg.InitShadowCompressor("brotli", 10)
 	cb, err := NewChannelBuilder(cfg, defaultTestRollupConfig, latestL1BlockOrigin)
 	require.NoError(err)
 
@@ -782,7 +816,7 @@ func ChannelBuilder_OutputBytes(t *testing.T, batchType uint) {
 	cfg.MaxFrameSize = 1000
 	cfg.TargetNumFrames = 16
 	cfg.BatchType = batchType
-	cfg.InitRatioCompressor(1.0)
+	cfg.InitRatioCompressor(1.0, "brotli", 10)
 	cb, err := NewChannelBuilder(cfg, defaultTestRollupConfig, latestL1BlockOrigin)
 	require.NoError(err, "NewChannelBuilder")
 

op-batcher/batcher/channel_config.go

@@ -53,25 +53,27 @@ type ChannelConfig struct {
 // value consistent with cc.TargetNumFrames and cc.MaxFrameSize.
 // comprKind can be the empty string, in which case the default compressor will
 // be used.
-func (cc *ChannelConfig) InitCompressorConfig(approxComprRatio float64, comprKind string) {
+func (cc *ChannelConfig) InitCompressorConfig(approxComprRatio float64, comprKind string, compressionAlgo string, compressLevel int) {
 	cc.CompressorConfig = compressor.Config{
 		// Compressor output size needs to account for frame encoding overhead
 		TargetOutputSize: MaxDataSize(cc.TargetNumFrames, cc.MaxFrameSize),
 		ApproxComprRatio: approxComprRatio,
 		Kind:             comprKind,
+		CompressionAlgo:  compressionAlgo,
+		CompressLevel:    compressLevel,
 	}
 }
 
-func (cc *ChannelConfig) InitRatioCompressor(approxComprRatio float64) {
-	cc.InitCompressorConfig(approxComprRatio, compressor.RatioKind)
+func (cc *ChannelConfig) InitRatioCompressor(approxComprRatio float64, compressionAlgo string, compressLevel int) {
+	cc.InitCompressorConfig(approxComprRatio, compressor.RatioKind, compressionAlgo, compressLevel)
 }
 
-func (cc *ChannelConfig) InitShadowCompressor() {
-	cc.InitCompressorConfig(0, compressor.ShadowKind)
+func (cc *ChannelConfig) InitShadowCompressor(compressionAlgo string, compressLevel int) {
+	cc.InitCompressorConfig(0, compressor.ShadowKind, compressionAlgo, compressLevel)
 }
 
-func (cc *ChannelConfig) InitNoneCompressor() {
-	cc.InitCompressorConfig(0, compressor.NoneKind)
+func (cc *ChannelConfig) InitNoneCompressor(compressionAlgo string, compressLevel int) {
+	cc.InitCompressorConfig(0, compressor.NoneKind, compressionAlgo, compressLevel)
 }
 
 func (cc *ChannelConfig) MaxFramesPerTx() int {

op-batcher/batcher/channel_config_test.go

@@ -20,7 +20,7 @@ func defaultTestChannelConfig() ChannelConfig {
 		TargetNumFrames:    1,
 		BatchType:          derive.SingularBatchType,
 	}
-	c.InitRatioCompressor(0.4)
+	c.InitRatioCompressor(0.4, "brotli", 10)
 	return c
 }
 

op-batcher/batcher/channel_manager_test.go

@@ -25,7 +25,7 @@ func channelManagerTestConfig(maxFrameSize uint64, batchType uint) ChannelConfig
 		TargetNumFrames: 1,
 		BatchType:       batchType,
 	}
-	cfg.InitRatioCompressor(1)
+	cfg.InitRatioCompressor(1, "brotli", 10)
 	return cfg
 }
 
@@ -123,7 +123,7 @@ func ChannelManager_Clear(t *testing.T, batchType uint) {
 	// channels on confirmation. This would result in [TxConfirmed]
 	// clearing confirmed transactions, and resetting the pendingChannels map
 	cfg.ChannelTimeout = 10
-	cfg.InitRatioCompressor(1)
+	cfg.InitRatioCompressor(1, "brotli", 10)
 	m := NewChannelManager(log, metrics.NoopMetrics, cfg, &defaultTestRollupConfig)
 
 	// Channel Manager state should be empty by default
@@ -345,7 +345,7 @@ func TestChannelManager_Close_PartiallyPendingChannel(t *testing.T) {
 		ChannelTimeout:  1000,
 		TargetNumFrames: 100,
 	}
-	cfg.InitNoneCompressor()
+	cfg.InitNoneCompressor("brotli", 10)
 	m := NewChannelManager(log, metrics.NoopMetrics, cfg, &defaultTestRollupConfig)
 	m.Clear(eth.BlockID{})
 
@@ -397,7 +397,7 @@ func ChannelManagerCloseAllTxsFailed(t *testing.T, batchType uint) {
 	log := testlog.Logger(t, log.LevelCrit)
 	cfg := channelManagerTestConfig(100, batchType)
 	cfg.TargetNumFrames = 1000
-	cfg.InitNoneCompressor()
+	cfg.InitNoneCompressor("brotli", 10)
 	m := NewChannelManager(log, metrics.NoopMetrics, cfg, &defaultTestRollupConfig)
 	m.Clear(eth.BlockID{})
 
@@ -445,7 +445,7 @@ func TestChannelManager_ChannelCreation(t *testing.T) {
 	const maxChannelDuration = 15
 	cfg := channelManagerTestConfig(1000, derive.SpanBatchType)
 	cfg.MaxChannelDuration = maxChannelDuration
-	cfg.InitNoneCompressor()
+	cfg.InitNoneCompressor("brotli", 10)
 
 	for _, tt := range []struct {
 		name                   string

op-batcher/batcher/config.go

@@ -67,6 +67,12 @@ type CLIConfig struct {
 	// Type of compressor to use. Must be one of [compressor.KindKeys].
 	Compressor string
 
+	// Type of compression algorithm to use. Must be one of [zlib, brotli]
+	CompressionAlgo string
+
+	// Levels of compression to use. E.g. 1-11 for brotli, 0-9 for zlib
+	CompressLevel int
+
 	// If Stopped is true, the batcher starts stopped and won't start batching right away.
 	// Batching needs to be started via an admin RPC.
 	Stopped bool
@@ -124,6 +130,12 @@ func (c *CLIConfig) Check() error {
 	if c.Compressor == compressor.RatioKind && (c.ApproxComprRatio <= 0 || c.ApproxComprRatio > 1) {
 		return fmt.Errorf("invalid ApproxComprRatio %v for ratio compressor", c.ApproxComprRatio)
 	}
+	if c.CompressionAlgo == "" {
+		return errors.New("must set compression algo")
+	}
+	if (c.CompressionAlgo == "zlib" && (c.CompressLevel < 0 || c.CompressLevel > 9)) || (c.CompressionAlgo == "brotli" && (c.CompressLevel < 1 || c.CompressLevel > 11)) {
+		return fmt.Errorf("invalid compression level %v for %v", c.CompressLevel, c.CompressionAlgo)
+	}
 	if c.BatchType > 1 {
 		return fmt.Errorf("unknown batch type: %v", c.BatchType)
 	}
@@ -168,6 +180,8 @@ func NewConfig(ctx *cli.Context) *CLIConfig {
 		TargetNumFrames:              ctx.Int(flags.TargetNumFramesFlag.Name),
 		ApproxComprRatio:             ctx.Float64(flags.ApproxComprRatioFlag.Name),
 		Compressor:                   ctx.String(flags.CompressorFlag.Name),
+		CompressionAlgo:              ctx.String(flags.CompressionAlgoFlag.Name),
+		CompressLevel:                ctx.Int(flags.CompressLevelFlag.Name),
 		Stopped:                      ctx.Bool(flags.StoppedFlag.Name),
 		WaitNodeSync:                 ctx.Bool(flags.WaitNodeSyncFlag.Name),
 		CheckRecentTxsDepth:          ctx.Int(flags.CheckRecentTxsDepthFlag.Name),

op-batcher/batcher/config_test.go

@@ -35,7 +35,9 @@ func validBatcherConfig() batcher.CLIConfig {
 		MetricsConfig:          metrics.DefaultCLIConfig(),
 		PprofConfig:            oppprof.DefaultCLIConfig(),
 		// The compressor config is not checked in config.Check()
-		RPC: rpc.DefaultCLIConfig(),
+		RPC:             rpc.DefaultCLIConfig(),
+		CompressionAlgo: "brotli",
+		CompressLevel:   9,
 	}
 }