feat: Report discarded transactions to external service

gradle/common-dependencies.gradle

@@ -34,4 +34,6 @@ dependencies {
 
   testImplementation 'org.mockito:mockito-core'
   testImplementation 'org.mockito:mockito-junit-jupiter'
+
+  testImplementation "org.wiremock:wiremock"
 }

gradle/dependency-management.gradle

@@ -104,6 +104,8 @@ dependencyManagement {
       entry 'picocli-codegen'
     }
 
+    dependency 'com.squareup.okhttp3:okhttp:4.12.0'
+
     dependencySet(group: 'io.tmio', version: '2.4.2') {
       entry 'tuweni-bytes'
       entry 'tuweni-net'
@@ -157,5 +159,7 @@ dependencyManagement {
       entry 'core'
       entry 'crypto'
     }
+
+    dependency "org.wiremock:wiremock:3.9.1"
   }
 }

sequencer/build.gradle

@@ -47,6 +47,8 @@ dependencies {
 
   compileOnly 'io.vertx:vertx-core'
 
+  compileOnly 'com.squareup.okhttp3:okhttp'
+
   implementation project(":native:compress")
 
   implementation 'com.fasterxml.jackson.core:jackson-databind'
@@ -68,6 +70,7 @@ dependencies {
   testImplementation "${besuArtifactGroup}.internal:core"
   testImplementation "${besuArtifactGroup}.internal:rlp"
   testImplementation "${besuArtifactGroup}.internal:core"
+  testImplementation "${besuArtifactGroup}:plugin-api"
 
   // workaround for bug https://github.com/dnsjava/dnsjava/issues/329, remove when upgraded upstream
   testImplementation 'dnsjava:dnsjava:3.6.1'

sequencer/src/main/java/net/consensys/linea/config/LineaTransactionSelectorCliOptions.java

@@ -15,6 +15,8 @@
 
 package net.consensys.linea.config;
 
+import java.net.URI;
+
 import com.google.common.base.MoreObjects;
 import jakarta.validation.constraints.Positive;
 import net.consensys.linea.plugins.LineaCliOptions;
@@ -39,6 +41,8 @@ public class LineaTransactionSelectorCliOptions implements LineaCliOptions {
   public static final String UNPROFITABLE_RETRY_LIMIT = "--plugin-linea-unprofitable-retry-limit";
   public static final int DEFAULT_UNPROFITABLE_RETRY_LIMIT = 10;
 
+  public static final String REJECTED_TX_ENDPOINT = "--plugin-linea-rejected-tx-endpoint";
+
   @Positive
   @CommandLine.Option(
       names = {MAX_BLOCK_CALLDATA_SIZE},
@@ -82,6 +86,13 @@ public class LineaTransactionSelectorCliOptions implements LineaCliOptions {
           "Max number of unprofitable transactions we retry on each block creation (default: ${DEFAULT-VALUE})")
   private int unprofitableRetryLimit = DEFAULT_UNPROFITABLE_RETRY_LIMIT;
 
+  @CommandLine.Option(
+      names = {REJECTED_TX_ENDPOINT},
+      hidden = true,
+      paramLabel = "<URI>",
+      description = "Endpoint URI for reporting rejected transactions  (default: ${DEFAULT-VALUE})")
+  private URI rejectedTxEndpoint = null;
+
   private LineaTransactionSelectorCliOptions() {}
 
   /**
@@ -107,6 +118,7 @@ public static LineaTransactionSelectorCliOptions fromConfig(
     options.maxGasPerBlock = config.maxGasPerBlock();
     options.unprofitableCacheSize = config.unprofitableCacheSize();
     options.unprofitableRetryLimit = config.unprofitableRetryLimit();
+    options.rejectedTxEndpoint = config.rejectedTxEndpoint();
     return options;
   }
 
@@ -123,6 +135,7 @@ public LineaTransactionSelectorConfiguration toDomainObject() {
         .maxGasPerBlock(maxGasPerBlock)
         .unprofitableCacheSize(unprofitableCacheSize)
         .unprofitableRetryLimit(unprofitableRetryLimit)
+        .rejectedTxEndpoint(rejectedTxEndpoint)
         .build();
   }
 
@@ -134,6 +147,7 @@ public String toString() {
         .add(MAX_GAS_PER_BLOCK, maxGasPerBlock)
         .add(UNPROFITABLE_CACHE_SIZE, unprofitableCacheSize)
         .add(UNPROFITABLE_RETRY_LIMIT, unprofitableRetryLimit)
+        .add(REJECTED_TX_ENDPOINT, rejectedTxEndpoint)
         .toString();
   }
 }

sequencer/src/main/java/net/consensys/linea/config/LineaTransactionSelectorConfiguration.java

@@ -15,6 +15,8 @@
 
 package net.consensys.linea.config;
 
+import java.net.URI;
+
 import lombok.Builder;
 import net.consensys.linea.plugins.LineaOptionsConfiguration;
 
@@ -25,5 +27,6 @@ public record LineaTransactionSelectorConfiguration(
     int overLinesLimitCacheSize,
     long maxGasPerBlock,
     int unprofitableCacheSize,
-    int unprofitableRetryLimit)
+    int unprofitableRetryLimit,
+    URI rejectedTxEndpoint)
     implements LineaOptionsConfiguration {}

sequencer/src/main/java/net/consensys/linea/jsonrpc/JsonRpcManager.java

@@ -0,0 +1,233 @@
+/*
+ * Copyright Consensys Software Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package net.consensys.linea.jsonrpc;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.file.DirectoryStream;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+
+/** This class is responsible for managing JSON-RPC requests for reporting rejected transactions */
+@Slf4j
+public class JsonRpcManager {
+  private static final long INITIAL_RETRY_DELAY = 1000L;
+  private static final int MAX_THREADS =
+      Math.min(32, Runtime.getRuntime().availableProcessors() * 2);
+  private static final long MAX_RETRY_DURATION = TimeUnit.HOURS.toMillis(2);
+  private static final MediaType JSON = MediaType.get("application/json; charset=utf-8");
+
+  private final OkHttpClient client = new OkHttpClient();
+  private final ObjectMapper objectMapper = new ObjectMapper();
+  private final Map<Path, Long> fileStartTimes = new ConcurrentHashMap<>();
+
+  private final Path rejTxRpcDirectory;
+  private final URI rejectedTxEndpoint;
+  private final ExecutorService executorService;
+  private final ScheduledExecutorService retrySchedulerService;
+
+  /**
+   * Creates a new JSON-RPC manager.
+   *
+   * @param besuDataDir Path to Besu data directory. The json-rpc files will be stored here under
+   *     rej-tx-rpc subdirectory.
+   * @param rejectedTxEndpoint The endpoint to send rejected transactions to
+   */
+  public JsonRpcManager(final Path besuDataDir, final URI rejectedTxEndpoint) {
+    this.rejTxRpcDirectory = besuDataDir.resolve("rej_tx_rpc");
+    this.rejectedTxEndpoint = rejectedTxEndpoint;
+    this.executorService = Executors.newFixedThreadPool(MAX_THREADS);
+    this.retrySchedulerService = Executors.newScheduledThreadPool(1);
+  }
+
+  /** Load existing JSON-RPC and submit them. */
+  public JsonRpcManager start() {
+    try {
+      // Create the rej-tx-rpc directory if it doesn't exist
+      Files.createDirectories(rejTxRpcDirectory);
+
+      // Load existing JSON files
+      loadExistingJsonFiles();
+      return this;
+    } catch (IOException e) {
+      log.error("Failed to create or access rej-tx-rpc directory", e);
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /** Shuts down the executor service and scheduler service. */
+  public void shutdown() {
+    executorService.shutdown();
+    retrySchedulerService.shutdown();
+  }
+
+  /**
+   * Submits a new JSON-RPC call.
+   *
+   * @param jsonContent The JSON content to submit
+   */
+  public void submitNewJsonRpcCall(final String jsonContent) {
+    try {
+      final Path jsonFile = saveJsonToFile(jsonContent);
+      fileStartTimes.put(jsonFile, System.currentTimeMillis());
+      submitJsonRpcCall(jsonFile, INITIAL_RETRY_DELAY);
+    } catch (final IOException e) {
+      log.error("Failed to save JSON content", e);
+    }
+  }
+
+  private void loadExistingJsonFiles() {
+    try {
+      final TreeSet<Path> sortedFiles = new TreeSet<>(Comparator.comparing(Path::getFileName));
+
+      try (DirectoryStream<Path> stream =
+          Files.newDirectoryStream(rejTxRpcDirectory, "rpc_*.json")) {
+        for (Path path : stream) {
+          sortedFiles.add(path);
+        }
+      }
+
+      for (Path path : sortedFiles) {
+        fileStartTimes.put(path, System.currentTimeMillis());
+        submitJsonRpcCall(path, INITIAL_RETRY_DELAY);
+      }
+
+      log.info("Loaded {} existing JSON files for rej-tx reporting", sortedFiles.size());
+    } catch (final IOException e) {
+      log.error("Failed to load existing JSON files", e);
+    }
+  }
+
+  private void submitJsonRpcCall(final Path jsonFile, final long nextDelay) {
+    executorService.submit(
+        () -> {
+          if (!Files.exists(jsonFile)) {
+            log.debug("json-rpc file no longer exists, skipping processing: {}", jsonFile);
+            fileStartTimes.remove(jsonFile);
+            return;
+          }
+          try {
+            final String jsonContent = new String(Files.readAllBytes(jsonFile));
+            final boolean success = sendJsonRpcCall(jsonContent);
+            if (success) {
+              Files.deleteIfExists(jsonFile);
+              fileStartTimes.remove(jsonFile);
+            } else {
+              log.error(
+                  "Failed to send JSON-RPC call to {}, retrying: {}", rejectedTxEndpoint, jsonFile);
+              scheduleRetry(jsonFile, nextDelay);
+            }
+          } catch (final Exception e) {
+            log.error("Failed to process json-rpc file: {}", jsonFile, e);
+            scheduleRetry(jsonFile, nextDelay);
+          }
+        });
+  }
+
+  private void scheduleRetry(final Path jsonFile, final long currentDelay) {
+    final Long startTime = fileStartTimes.get(jsonFile);
+    if (startTime == null) {
+      log.debug("No start time found for file: {}. Skipping retry.", jsonFile);
+      return;
+    }
+
+    // check if we're still within the maximum retry duration
+    if (System.currentTimeMillis() - startTime < MAX_RETRY_DURATION) {
+      // schedule a retry with exponential backoff
+      long nextDelay = Math.min(currentDelay * 2, TimeUnit.MINUTES.toMillis(1)); // Cap at 1 minute
+      retrySchedulerService.schedule(
+          () -> submitJsonRpcCall(jsonFile, nextDelay), currentDelay, TimeUnit.MILLISECONDS);
+    } else {
+      log.error("Exceeded maximum retry duration for rej-tx json-rpc file: {}", jsonFile);
+      fileStartTimes.remove(jsonFile);
+    }
+  }
+
+  private boolean sendJsonRpcCall(final String jsonContent) {
+    final RequestBody body = RequestBody.create(jsonContent, JSON);
+    final Request request =
+        new Request.Builder().url(rejectedTxEndpoint.toString()).post(body).build();
+    try (final Response response = client.newCall(request).execute()) {
+      if (!response.isSuccessful()) {
+        log.error("Unexpected response code from rejected-tx endpoint: {}", response.code());
+        return false;
+      }
+
+      // process the response body here ...
+      final String responseBody = response.body() != null ? response.body().string() : null;
+      if (responseBody == null) {
+        log.error("Unexpected empty response body from rejected-tx endpoint");
+        return false;
+      }
+
+      final JsonNode jsonNode = objectMapper.readTree(responseBody);
+      if (jsonNode == null) {
+        log.error("Failed to parse JSON response from rejected-tx endpoint: {}", responseBody);
+        return false;
+      }
+      if (jsonNode.has("error")) {
+        log.error("Error response from rejected-tx endpoint: {}", jsonNode.get("error"));
+        return false;
+      }
+      // Check for result
+      if (jsonNode.has("result")) {
+        String status = jsonNode.get("result").get("status").asText();
+        log.debug("Rejected-tx JSON-RPC call successful. Status: {}", status);
+        return true;
+      }
+
+      log.warn("Unexpected rejected-tx JSON-RPC response format: {}", responseBody);
+      return false;
+    } catch (final IOException e) {
+      log.error("Failed to send JSON-RPC call to rejected-tx endpoint {}", rejectedTxEndpoint, e);
+      return false;
+    }
+  }
+
+  private Path saveJsonToFile(final String jsonContent) throws IOException {
+    long timestamp = System.currentTimeMillis();
+    for (int attempt = 0; attempt < 100; attempt++) {
+      final String fileName = String.format("rpc_%d_%d.json", timestamp, attempt);
+      final Path filePath = rejTxRpcDirectory.resolve(fileName);
+      try {
+        return Files.writeString(filePath, jsonContent, StandardOpenOption.CREATE_NEW);
+      } catch (final FileAlreadyExistsException e) {
+        log.trace("File already exists {}, retrying.", filePath);
+      }
+    }
+    throw new IOException("Failed to save JSON content after 100 attempts");
+  }
+}