[Besu-Plugin] Use blob compressor selector by timestamp

besu-plugins/linea-sequencer/acceptance-tests/build.gradle

@@ -80,8 +80,9 @@ dependencies {
   kapt("org.apache.logging.log4j:log4j-core:${libs.versions.log4j.get()}")
   annotationProcessor "org.apache.logging.log4j:log4j-core:${libs.versions.log4j.get()}"
   implementation "org.apache.logging.log4j:log4j-core:${libs.versions.log4j.get()}"
-
   implementation project("${lineaSequencerProjectPath}:sequencer")
+  implementation "org.jetbrains.kotlin:kotlin-stdlib:${libs.versions.kotlin.get()}"
+
   testImplementation "build.linea:blob-compressor:${libs.versions.blobCompressor.get()}"
   testImplementation project(":tracer:arithmetization")
   testImplementation project(":jvm-libs:generic:serialization:jackson")

besu-plugins/linea-sequencer/acceptance-tests/src/test/kotlin/linea/plugin/acc/test/rpc/linea/EstimateGasTest.kt

@@ -17,8 +17,8 @@ import com.fasterxml.jackson.databind.JsonDeserializer
 import com.fasterxml.jackson.databind.ObjectReader
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import linea.blob.BlobCompressorSelectorByTimestamp
 import linea.blob.BlobCompressorVersion
-import linea.blob.GoBackedBlobCompressor
 import linea.plugin.acc.test.LineaPluginPoSTestBase
 import linea.plugin.acc.test.TestCommandLineOptionsBuilder
 import net.consensys.linea.bl.TransactionProfitabilityCalculator
@@ -47,6 +47,7 @@ import java.net.http.HttpClient
 import java.net.http.HttpRequest
 import java.net.http.HttpResponse
 import java.nio.charset.StandardCharsets
+import kotlin.time.Instant
 
 open class EstimateGasTest : LineaPluginPoSTestBase() {
   protected lateinit var profitabilityCalculator: TransactionProfitabilityCalculator
@@ -79,7 +80,12 @@ open class EstimateGasTest : LineaPluginPoSTestBase() {
         .build()
     profitabilityCalculator = TransactionProfitabilityCalculator(
       profitabilityConf,
-      CachingTransactionCompressor(GoBackedBlobCompressor.getInstance(BlobCompressorVersion.V2, 128 * 1024)),
+      CachingTransactionCompressor(
+        BlobCompressorSelectorByTimestamp(
+          mapOf(BlobCompressorVersion.V2 to Instant.DISTANT_PAST),
+          128 * 1024,
+        ),
+      ),
     )
   }
 

besu-plugins/linea-sequencer/gradle/dist.gradle

@@ -33,4 +33,12 @@ jar {
       'Implementation-Version': calculateVersion()
     )
   }
+
+  // Explicitly include Kotlin stdlib in the JAR to ensure kotlin.time.Instant and other
+  // Kotlin standard library classes are available at runtime when the plugin is loaded by Besu
+  from {
+    configurations.runtimeClasspath
+      .filter { it.name.startsWith('kotlin-stdlib') }
+      .collect { zipTree(it) }
+  }
 }

besu-plugins/linea-sequencer/sequencer/build.gradle

@@ -20,10 +20,25 @@ apply from: lineaSequencerProject.file("gradle/java.gradle")
 apply from: lineaSequencerProject.file("gradle/dependency-management.gradle")
 apply from: lineaSequencerProject.file("gradle/lint.gradle")
 
+// Force a single version of kotlin-stdlib across all transitive dependencies.
+// Multiple dependencies (blob-compressor, tuweni, dagger-compiler, etc.) each bring in different
+// versions of kotlin-stdlib (1.9.10, 2.0.21, 2.1.0, 2.3.0, etc.), causing version conflicts.
+// This resolution strategy ensures we use the canonical Kotlin version (libs.versions.kotlin)
+// throughout the entire dependency tree, preventing "Cannot find a version that satisfies
+// all constraints" errors and ensuring consistent Kotlin behavior at runtime.
+configurations.all {
+  resolutionStrategy.eachDependency { details ->
+    if (details.requested.group == 'org.jetbrains.kotlin' && details.requested.name.startsWith('kotlin-stdlib')) {
+      details.useVersion "${libs.versions.kotlin.get()}"
+    }
+  }
+}
+
 dependencies {
   implementation project(":tracer:arithmetization")
   implementation "build.linea:blob-compressor:${libs.versions.blobCompressor.get()}"
   implementation "build.linea.internal:kotlin:${libs.versions.lineaKotlin.get()}"
+  implementation "org.jetbrains.kotlin:kotlin-stdlib:${libs.versions.kotlin.get()}"
 
   testImplementation 'org.assertj:assertj-core'
 

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/AbstractLineaSharedPrivateOptionsPlugin.java

@@ -12,9 +12,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
-import linea.blob.BlobCompressor;
-import linea.blob.BlobCompressorVersion;
-import linea.blob.GoBackedBlobCompressor;
+import linea.blob.BlobCompressorSelectorByTimestamp;
 import lombok.extern.slf4j.Slf4j;
 import net.consensys.linea.bl.TransactionProfitabilityCalculator;
 import net.consensys.linea.bundles.BundlePoolService;
@@ -84,7 +82,7 @@ public abstract class AbstractLineaSharedPrivateOptionsPlugin
   protected static MetricCategoryRegistry metricCategoryRegistry;
   protected static RpcEndpointService rpcEndpointService;
   protected static InvalidTransactionByLineCountCache invalidTransactionByLineCountCache;
-  protected static BlobCompressor blobCompressor;
+  protected static BlobCompressorSelectorByTimestamp blobCompressorSelectorByTimestamp;
   protected static TransactionCompressor transactionCompressor;
   protected static TransactionProfitabilityCalculator transactionProfitabilityCalculator;
 
@@ -300,13 +298,15 @@ private void performSharedStartTasksOnce(final ServiceManager serviceManager) {
         transactionSelectorConfiguration().blobSizeLimit() != null
             ? transactionSelectorConfiguration().blobSizeLimit()
             : DEFAULT_COMPRESSED_SIZE_LIMIT;
-    blobCompressor =
-        GoBackedBlobCompressor.getInstance(BlobCompressorVersion.V2, effectiveBlobLimit);
+    blobCompressorSelectorByTimestamp =
+        new BlobCompressorSelectorByTimestamp(
+            transactionSelectorConfiguration().blobCompressorVersionActivationTimes(),
+            effectiveBlobLimit);
 
     final LineaProfitabilityConfiguration profitabilityConfiguration = profitabilityConfiguration();
     transactionCompressor =
         new CachingTransactionCompressor(
-            profitabilityConfiguration.compressedTxCacheSize(), blobCompressor);
+            profitabilityConfiguration.compressedTxCacheSize(), blobCompressorSelectorByTimestamp);
     transactionProfitabilityCalculator =
         new TransactionProfitabilityCalculator(profitabilityConfiguration, transactionCompressor);
   }
@@ -318,6 +318,6 @@ public void stop() {
     sharedStartTasksDone.set(false);
     blockchainService = null;
     metricsSystem = null;
-    blobCompressor = null;
+    blobCompressorSelectorByTimestamp = null;
   }
 }

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/config/LineaTransactionSelectorCliOptions.java

@@ -9,18 +9,24 @@
 
 package net.consensys.linea.config;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import jakarta.validation.constraints.Positive;
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Stream;
+import kotlin.time.Instant;
+import linea.blob.BlobCompressorVersion;
 import net.consensys.linea.plugins.LineaCliOptions;
 import net.consensys.linea.sequencer.txselection.selectors.TransactionEventFilter;
 import org.apache.tuweni.bytes.Bytes32;
@@ -52,6 +58,10 @@ public class LineaTransactionSelectorCliOptions implements LineaCliOptions {
   public static final String EVENTS_DENY_LIST_PATH = "--plugin-linea-events-deny-list-path";
   public static final String EVENTS_BUNDLE_DENY_LIST_PATH =
       "--plugin-linea-events-bundle-deny-list-path";
+  public static final String BLOB_COMPRESSOR_VERSION_TIMESTAMPS =
+      "--plugin-linea-blob-compressor-version-timestamps";
+  public static final String BLOB_COMPRESSOR_VERSION_TIMESTAMPS_DEFAULT =
+      String.format("%s=%s", BlobCompressorVersion.V2.name(), Instant.Companion.getDISTANT_PAST());
 
   @Positive
   @CommandLine.Option(
@@ -132,7 +142,24 @@ public class LineaTransactionSelectorCliOptions implements LineaCliOptions {
       description = "Path to the file containing the events deny list for bundles")
   private String eventsBundleDenyListPath;
 
-  private LineaTransactionSelectorCliOptions() {}
+  private static final class BlobCompressorVersionCandidates implements Iterable<String> {
+    @Override
+    public Iterator<String> iterator() {
+      return Arrays.stream(BlobCompressorVersion.values()).map(Enum::name).iterator();
+    }
+  }
+
+  @CommandLine.Option(
+      names = {BLOB_COMPRESSOR_VERSION_TIMESTAMPS},
+      hidden = true,
+      paramLabel = "<MAP>",
+      completionCandidates = BlobCompressorVersionCandidates.class,
+      description =
+          "Comma-separated map of BlobCompressorVersion to Instant, "
+              + "e.g. V1_2=2025-01-01T00:00:00Z,V2=2026-01-01T00:00:00Z ."
+              + "Available versions: ${COMPLETION-CANDIDATES} . "
+              + "(default: ${DEFAULT-VALUE})")
+  private String blobCompressorVersionTimestampsRaw = BLOB_COMPRESSOR_VERSION_TIMESTAMPS_DEFAULT;
 
   /**
    * Create Linea cli options.
@@ -181,6 +208,7 @@ public LineaTransactionSelectorConfiguration toDomainObject() {
         .eventsDenyList(parseTransactionEventDenyList(eventsDenyListPath))
         .eventsBundleDenyListPath(eventsBundleDenyListPath)
         .eventsBundleDenyList(parseTransactionEventDenyList(eventsBundleDenyListPath))
+        .blobCompressorVersionActivationTimes(getBlobCompressorVersionTimestamps())
         .build();
   }
 
@@ -196,6 +224,7 @@ public String toString() {
         .add(MAX_BUNDLE_POOL_SIZE_BYTES, maxBundlePoolSizeBytes)
         .add(EVENTS_DENY_LIST_PATH, eventsDenyListPath)
         .add(EVENTS_BUNDLE_DENY_LIST_PATH, eventsBundleDenyListPath)
+        .add(BLOB_COMPRESSOR_VERSION_TIMESTAMPS, blobCompressorVersionTimestampsRaw)
         .toString();
   }
 
@@ -233,4 +262,24 @@ public Map<Address, Set<TransactionEventFilter>> parseTransactionEventDenyList(
       throw new RuntimeException(e);
     }
   }
+
+  @VisibleForTesting
+  Map<BlobCompressorVersion, Instant> parseBlobCompressorVersionTimestamps(String input) {
+    Map<BlobCompressorVersion, Instant> result = new HashMap<>();
+    String[] pairs = input.split(",");
+    for (String pair : pairs) {
+      String[] kv = pair.split("=");
+      if (kv.length != 2) {
+        throw new IllegalArgumentException("Invalid BlobCompressorVersion=Instant pair: " + pair);
+      }
+      BlobCompressorVersion version = BlobCompressorVersion.valueOf(kv[0]);
+      Instant instant = Instant.Companion.parse(kv[1]);
+      result.put(version, instant);
+    }
+    return result;
+  }
+
+  public Map<BlobCompressorVersion, Instant> getBlobCompressorVersionTimestamps() {
+    return parseBlobCompressorVersionTimestamps(blobCompressorVersionTimestampsRaw);
+  }
 }

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/config/LineaTransactionSelectorConfiguration.java

@@ -11,6 +11,8 @@
 
 import java.util.Map;
 import java.util.Set;
+import kotlin.time.Instant;
+import linea.blob.BlobCompressorVersion;
 import lombok.Builder;
 import net.consensys.linea.plugins.LineaOptionsConfiguration;
 import net.consensys.linea.sequencer.txselection.selectors.TransactionEventFilter;
@@ -29,5 +31,6 @@ public record LineaTransactionSelectorConfiguration(
     String eventsDenyListPath,
     Map<Address, Set<TransactionEventFilter>> eventsDenyList,
     String eventsBundleDenyListPath,
-    Map<Address, Set<TransactionEventFilter>> eventsBundleDenyList)
+    Map<Address, Set<TransactionEventFilter>> eventsBundleDenyList,
+    Map<BlobCompressorVersion, Instant> blobCompressorVersionActivationTimes)
     implements LineaOptionsConfiguration {}

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/sequencer/txselection/LineaTransactionSelectorFactory.java

@@ -21,7 +21,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-import linea.blob.BlobCompressor;
+import linea.blob.BlobCompressorSelectorByTimestamp;
 import lombok.extern.slf4j.Slf4j;
 import net.consensys.linea.bl.TransactionProfitabilityCalculator;
 import net.consensys.linea.bundles.BundlePoolService;
@@ -73,7 +73,7 @@ public class LineaTransactionSelectorFactory implements PluginTransactionSelecto
   private final AtomicBoolean isSelectionInterrupted = new AtomicBoolean(false);
   private final TransactionProfitabilityCalculator transactionProfitabilityCalculator;
   private final TransactionCompressor transactionCompressor;
-  private final BlobCompressor blobCompressor;
+  private final BlobCompressorSelectorByTimestamp blobCompressorSelectorByTimestamp;
 
   public LineaTransactionSelectorFactory(
       final BlockchainService blockchainService,
@@ -92,7 +92,7 @@ public LineaTransactionSelectorFactory(
       final AtomicReference<Set<Address>> deniedAddresses,
       final TransactionProfitabilityCalculator transactionProfitabilityCalculator,
       final TransactionCompressor transactionCompressor,
-      final BlobCompressor blobCompressor) {
+      final BlobCompressorSelectorByTimestamp blobCompressorSelectorByTimestamp) {
     this.blockchainService = blockchainService;
     this.txSelectorConfiguration = txSelectorConfiguration;
     this.l1L2BridgeConfiguration = l1L2BridgeConfiguration;
@@ -109,7 +109,7 @@ public LineaTransactionSelectorFactory(
     this.deniedAddresses = deniedAddresses;
     this.transactionProfitabilityCalculator = transactionProfitabilityCalculator;
     this.transactionCompressor = transactionCompressor;
-    this.blobCompressor = blobCompressor;
+    this.blobCompressorSelectorByTimestamp = blobCompressorSelectorByTimestamp;
 
     if (txSelectorConfiguration.maxBlockCallDataSize() != null) {
       log.warn(
@@ -137,7 +137,7 @@ public PluginTransactionSelector create(final SelectorsStateManager selectorsSta
             deniedAddresses,
             transactionProfitabilityCalculator,
             transactionCompressor,
-            blobCompressor);
+            blobCompressorSelectorByTimestamp);
     currSelector.set(selector);
     return selector;
   }

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/sequencer/txselection/LineaTransactionSelectorPlugin.java

@@ -21,7 +21,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
-import linea.blob.BlobCompressor;
+import linea.blob.BlobCompressorSelectorByTimestamp;
 import lombok.extern.slf4j.Slf4j;
 import net.consensys.linea.AbstractLineaRequiredPlugin;
 import net.consensys.linea.config.LineaRejectedTxReportingConfiguration;
@@ -130,8 +130,8 @@ public void doStart() {
     // blobCompressor is initialised in AbstractLineaSharedPrivateOptionsPlugin with the effective
     // limit. Only pass it to the factory when a blob size limit is explicitly configured so that
     // CompressionAwareTransactionSelector is only active when intentionally enabled.
-    final BlobCompressor selectorBlobCompressor =
-        txSelectorConfiguration.blobSizeLimit() != null ? blobCompressor : null;
+    final BlobCompressorSelectorByTimestamp blobCompressorSelector =
+        txSelectorConfiguration.blobSizeLimit() != null ? blobCompressorSelectorByTimestamp : null;
 
     transactionSelectionService.registerPluginTransactionSelectorFactory(
         new LineaTransactionSelectorFactory(
@@ -151,7 +151,7 @@ public void doStart() {
             deniedAddresses,
             transactionProfitabilityCalculator,
             transactionCompressor,
-            selectorBlobCompressor));
+            blobCompressorSelector));
   }
 
   @Override

besu-plugins/linea-sequencer/sequencer/src/main/java/net/consensys/linea/sequencer/txselection/selectors/CompressionAwareTransactionSelector.java

@@ -19,7 +19,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import kotlin.time.Instant;
 import linea.blob.BlobCompressor;
+import linea.blob.BlobCompressorSelectorByTimestamp;
 import lombok.extern.slf4j.Slf4j;
 import net.consensys.linea.utils.TransactionCompressor;
 import org.apache.tuweni.bytes.Bytes;
@@ -105,7 +107,7 @@ public class CompressionAwareTransactionSelector
 
   private final long fastExecutionPathLimit;
   private final TransactionCompressor transactionCompressor;
-  private final BlobCompressor blobCompressor;
+  private final BlobCompressorSelectorByTimestamp blobCompressorSelectorByTimestamp;
 
   /**
    * Shared single-threaded executor for slow-execution-path compression. Declared static so that
@@ -143,7 +145,7 @@ public CompressionAwareTransactionSelector(
       final int blobSizeLimit,
       final int compressedBlockHeaderOverhead,
       final TransactionCompressor transactionCompressor,
-      final BlobCompressor blobCompressor) {
+      final BlobCompressorSelectorByTimestamp blobCompressorSelectorByTimestamp) {
     super(
         selectorsStateManager,
         new CompressionState(0L, new ArrayList<>()),
@@ -154,7 +156,7 @@ public CompressionAwareTransactionSelector(
           "fastExecutionPathLimit must be positive, got " + fastExecutionPathLimit);
     }
     this.transactionCompressor = transactionCompressor;
-    this.blobCompressor = blobCompressor;
+    this.blobCompressorSelectorByTimestamp = blobCompressorSelectorByTimestamp;
   }
 
   @Override
@@ -163,8 +165,11 @@ public TransactionSelectionResult evaluateTransactionPreProcessing(
     final Transaction transaction =
         (Transaction) evaluationContext.getPendingTransaction().getTransaction();
     final CompressionState state = getWorkingState();
-
-    final int txCompressedSize = transactionCompressor.getCompressedSize(transaction);
+    final ProcessableBlockHeader pendingHeader = evaluationContext.getPendingBlockHeader();
+    final Instant timestamp = Instant.Companion.fromEpochSeconds(pendingHeader.getTimestamp(), 0L);
+    final BlobCompressor blobCompressor =
+        blobCompressorSelectorByTimestamp.getBlobCompressor(timestamp);
+    final int txCompressedSize = transactionCompressor.getCompressedSize(transaction, timestamp);
     long newConservativeCumulative = state.cumulativeCompressedSize() + txCompressedSize;
 
     // Fast execution path: sum of per-tx compressed sizes is at or below the effective limit (blob
@@ -187,7 +192,6 @@ public TransactionSelectionResult evaluateTransactionPreProcessing(
     // Slow execution path: conservative estimate exceeded the limit.
     // Build the block RLP on this thread (cheap, CPU-only), then submit the native
     // reset+appendBlock to the background executor so it can overlap with EVM execution.
-    final ProcessableBlockHeader pendingHeader = evaluationContext.getPendingBlockHeader();
     final List<Transaction> tentativeTxs = new ArrayList<>(state.selectedTransactions());
     tentativeTxs.add(transaction);
     final byte[] blockRlp = buildBlockRlp(pendingHeader, tentativeTxs);