Merge pull request #265 from rsksmart/avoid-signing-same-pegout-integ…

…ration

Avoid Pegnatories From Signing Multiple Times The Same Peg-out

src/main/java/co/rsk/federate/btcreleaseclient/BtcReleaseClient.java

@@ -12,6 +12,8 @@
 import co.rsk.crypto.Keccak256;
 import co.rsk.federate.FederatorSupport;
 import co.rsk.federate.adapter.ThinConverter;
+import co.rsk.federate.btcreleaseclient.cache.PegoutSignedCache;
+import co.rsk.federate.btcreleaseclient.cache.PegoutSignedCacheImpl;
 import co.rsk.federate.config.FedNodeSystemProperties;
 import co.rsk.federate.signing.ECDSASigner;
 import co.rsk.federate.signing.FederationCantSignException;
@@ -35,6 +37,7 @@
 import co.rsk.peg.federation.Federation;
 import co.rsk.peg.federation.ErpFederation;
 import co.rsk.peg.StateForFederator;
+import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -84,11 +87,11 @@ public class BtcReleaseClient {
 
     private final Ethereum ethereum;
     private final FederatorSupport federatorSupport;
-    private final FedNodeSystemProperties systemProperties;
     private final Set<Federation> observedFederations;
     private final NodeBlockProcessor nodeBlockProcessor;
     private final BridgeConstants bridgeConstants;
     private final boolean isPegoutEnabled;
+    private final PegoutSignedCache pegoutSignedCache;
 
     private ECDSASigner signer;
     private BtcReleaseEthereumListener blockListener;
@@ -108,12 +111,13 @@ public BtcReleaseClient(
     ) {
         this.ethereum = ethereum;
         this.federatorSupport = federatorSupport;
-        this.systemProperties = systemProperties;
         this.observedFederations = new HashSet<>();
         this.blockListener = new BtcReleaseEthereumListener();
-        this.bridgeConstants = this.systemProperties.getNetworkConstants().getBridgeConstants();
-        this.isPegoutEnabled = this.systemProperties.isPegoutEnabled();
+        this.bridgeConstants = systemProperties.getNetworkConstants().getBridgeConstants();
+        this.isPegoutEnabled = systemProperties.isPegoutEnabled();
         this.nodeBlockProcessor = nodeBlockProcessor;
+        this.pegoutSignedCache = new PegoutSignedCacheImpl(
+            systemProperties.getPegoutSignedCacheTtl(), Clock.systemUTC());
     }
 
     public void setup(
@@ -143,7 +147,7 @@ public void setup(
         }
         peerGroup.start();
 
-        blockListener = new BtcReleaseEthereumListener();
+        this.blockListener = new BtcReleaseEthereumListener();
         this.signerMessageBuilderFactory = signerMessageBuilderFactory;
         this.releaseCreationInformationGetter = pegoutCreationInformationGetter;
         this.releaseRequirementsEnforcer = releaseRequirementsEnforcer;
@@ -283,10 +287,15 @@ protected Optional<ReleaseCreationInformation> tryGetReleaseInformation(
         BtcTransaction pegoutBtcTx
     ) {
         try {
+            // Discard pegout tx if processed in a previous round of execution
+            logger.trace(
+                "[tryGetReleaseInformation] Checking if pegoutCreationTxHash {} has already been signed",
+                pegoutCreationRskTxHash);
+            validateTxIsNotCached(pegoutCreationRskTxHash);
+
             // Discard pegout btc tx this fed already signed or cannot be signed by the observed federations
             logger.trace("[tryGetReleaseInformation] Validating if pegoutBtcTxHash {} can be signed by observed federations and " +
                              "that it is not already signed by current fed", pegoutBtcTx.getHash());
-
             validateTxCanBeSigned(pegoutBtcTx);
 
             // IMPORTANT: As per the current behaviour of the bridge, no pegout should have inputs to be signed
@@ -326,6 +335,15 @@ protected Optional<ReleaseCreationInformation> tryGetReleaseInformation(
         return Optional.empty();
     }
 
+    void validateTxIsNotCached(Keccak256 pegoutCreationRskTxHash) throws FederatorAlreadySignedException {
+        if (pegoutSignedCache.hasAlreadyBeenSigned(pegoutCreationRskTxHash)) {
+            String message = String.format(
+                "Rsk pegout creation tx hash %s was found in the pegouts signed cache",
+                pegoutCreationRskTxHash);
+            throw new FederatorAlreadySignedException(message);
+        }
+    }
+
     protected void validateTxCanBeSigned(BtcTransaction pegoutBtcTx) throws FederatorAlreadySignedException, FederationCantSignException {
         try {
             BtcECKey federatorPublicKey = signer.getPublicKey(BTC_KEY_ID.getKeyId()).toBtcKey();
@@ -400,6 +418,11 @@ protected void signRelease(int signerVersion, ReleaseCreationInformation pegoutC
 
             logger.info("[signRelease] Signed pegout created in rsk transaction {}", pegoutCreationInformation.getPegoutConfirmationRskTxHash());
             federatorSupport.addSignature(signatures, pegoutCreationInformation.getPegoutConfirmationRskTxHash().getBytes());
+
+            logger.trace("[signRelease] Put pegoutCreationRskTxHash {} in the pegouts signed cache",
+                pegoutCreationInformation.getPegoutCreationRskTxHash());
+            pegoutSignedCache.putIfAbsent(
+                pegoutCreationInformation.getPegoutCreationRskTxHash());
         } catch (SignerException e) {
             String message = String.format("Error signing pegout created in rsk transaction %s", pegoutCreationInformation.getPegoutCreationRskTxHash());
             logger.error(message, e);

src/main/java/co/rsk/federate/btcreleaseclient/cache/PegoutSignedCache.java

@@ -0,0 +1,26 @@
+package co.rsk.federate.btcreleaseclient.cache;
+
+import co.rsk.crypto.Keccak256;
+
+public interface PegoutSignedCache {
+
+  /**
+   * Checks if the specified RSK transaction hash for pegout creation has already
+   * been signed.
+   * 
+   * @param pegoutCreationRskTxHash The Keccak256 hash of the RSK transaction for
+   *                                pegout creation.
+   * @return {@code true} if the hash of the transaction has already been signed,
+   *         {@code false} otherwise.
+   */
+  boolean hasAlreadyBeenSigned(Keccak256 pegoutCreationRskTxHash);
+
+  /**
+   * Stores the specified RSK transaction hash for pegout creation along with its
+   * timestamp in the cache if absent.
+   * 
+   * @param pegoutCreationRskTxHash The Keccak256 hash of the RSK transaction for
+   *                                pegout creation to be stored.
+   */
+  void putIfAbsent(Keccak256 pegoutCreationRskTxHash);
+}

src/main/java/co/rsk/federate/btcreleaseclient/cache/PegoutSignedCacheImpl.java

@@ -0,0 +1,86 @@
+package co.rsk.federate.btcreleaseclient.cache;
+
+import co.rsk.crypto.Keccak256;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PegoutSignedCacheImpl implements PegoutSignedCache {
+
+  private static final Logger logger = LoggerFactory.getLogger(PegoutSignedCacheImpl.class);
+  private static final Integer CLEANUP_INTERVAL_IN_HOURS = 1;
+
+  private final Map<Keccak256, Instant> cache = new ConcurrentHashMap<>();
+  private final ScheduledExecutorService cleanupScheduler = Executors.newSingleThreadScheduledExecutor();
+  private final Duration ttl;
+  private final Clock clock;
+
+  public PegoutSignedCacheImpl(Duration ttl, Clock clock) {
+    validateTtl(ttl);
+
+    this.ttl = ttl;
+    this.clock = clock;
+
+    // Start a background thread for periodic cleanup
+    cleanupScheduler.scheduleAtFixedRate(
+        this::performCleanup,
+        CLEANUP_INTERVAL_IN_HOURS, // initial delay
+        CLEANUP_INTERVAL_IN_HOURS, // period
+        TimeUnit.HOURS
+    );
+  }
+
+  @Override
+  public boolean hasAlreadyBeenSigned(Keccak256 pegoutCreationRskTxHash) {
+    return Optional.ofNullable(pegoutCreationRskTxHash)
+        .map(cache::get)
+        .map(this::isValidTimestamp)
+        .orElse(false);
+  }
+
+  @Override
+  public void putIfAbsent(Keccak256 pegoutCreationRskTxHash) {
+    if (pegoutCreationRskTxHash == null) {
+      throw new IllegalArgumentException(
+          "The pegoutCreationRskTxHash argument must not be null");
+    }
+
+    Optional.of(pegoutCreationRskTxHash)
+        .ifPresent(rskTxHash -> cache.putIfAbsent(rskTxHash, clock.instant()));
+  }
+
+  void performCleanup() {
+    logger.trace(
+        "[performCleanup] Pegouts signed cache before cleanup: {}", cache.keySet());
+    cache.entrySet().removeIf(
+        entry -> !isValidTimestamp(entry.getValue()));
+    logger.trace(
+        "[performCleanup] Pegouts signed cache after cleanup: {}", cache.keySet());
+  }
+
+  private boolean isValidTimestamp(Instant timestampInCache) {
+    return Optional.ofNullable(timestampInCache)
+        .map(timestamp -> clock.instant().toEpochMilli() - timestamp.toEpochMilli())
+        .map(timeCachedInMillis -> timeCachedInMillis <= ttl.toMillis())
+        .orElse(false);
+  }
+
+  private static void validateTtl(Duration ttl) {
+    if (ttl == null || ttl.isNegative() || ttl.isZero()) {
+      Long ttlInMinutes = ttl != null ? ttl.toMinutes() : null;
+      String message = String.format(
+          "Invalid pegouts signed cache TTL value in minutes supplied: %d", ttlInMinutes);
+      logger.error("[validateTtl] {}", message);
+
+      throw new IllegalArgumentException(message);
+    }
+  }
+}

src/main/java/co/rsk/federate/config/FedNodeSystemProperties.java

@@ -3,6 +3,7 @@
 import co.rsk.config.ConfigLoader;
 import co.rsk.config.RskSystemProperties;
 import com.typesafe.config.Config;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -83,4 +84,17 @@ public int getBtcReleaseClientInitializationMaxDepth() {
             configFromFiles.getInt("federator.pegoutStorageInitializationDepth") :
             6_000;
     }
+
+    /**
+     * Retrieves the time to live (TTL) duration for the pegout signed cache.
+     * The TTL duration specifies the validity period for cache entries.
+     * If the TTL value is not configured, a default value of 30 minutes is used.
+     * 
+     * @return The time to live (TTL) duration for the pegout signed cache,
+     *         or a default value of 30 minutes if not configured.
+     */
+    public Duration getPegoutSignedCacheTtl() {
+        return Duration.ofMinutes(
+            getInt("federator.pegoutSignedCacheTtlInMinutes", 30));
+    }
 }

src/main/resources/config/fed-sample.conf

@@ -46,4 +46,7 @@ federator {
 
     # Gas price to use for federate node transactions
     gasPrice = 1000
+
+    # Pegout signed cache ttl value to avoid signing the same pegout btc transaction
+    pegoutSignedCacheTtlInMinutes = 30
 }

src/test/java/co/rsk/federate/FedNodeSystemPropertiesTest.java

@@ -1,10 +1,12 @@
 package co.rsk.federate;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.time.Duration;
 import co.rsk.config.ConfigLoader;
 import co.rsk.federate.config.FedNodeSystemProperties;
 import com.typesafe.config.Config;
@@ -143,4 +145,36 @@ void updateBridgeTimer_enabled_mainnet() {
 
         assertTrue(fedNodeSystemProperties.isUpdateBridgeTimerEnabled());
     }
+
+    @Test
+    void getPegoutSignedCacheTtl_whenConfigurationPathExists_shouldUseCustomTtl() {
+        int customTtlInMinutes = 10;
+        when(configLoader.getConfig()).thenReturn(config);
+        when(config.hasPath(
+            "federator.pegoutSignedCacheTtlInMinutes")).thenReturn(true);
+        when(config.getInt(
+            "federator.pegoutSignedCacheTtlInMinutes")).thenReturn(customTtlInMinutes);
+        when(config.root()).thenReturn(configObject);
+
+        FedNodeSystemProperties fedNodeSystemProperties = new FedNodeSystemProperties(configLoader);
+
+        assertEquals(
+            Duration.ofMinutes(customTtlInMinutes),
+            fedNodeSystemProperties.getPegoutSignedCacheTtl());
+    }
+
+    @Test
+    void getPegoutSignedCacheTtl_whenConfigurationPathDoesNotExist_shouldUseDefaultTtl() {
+        int defaultTtlInMinutes = 30;
+        when(configLoader.getConfig()).thenReturn(config);
+        when(config.hasPath(
+            "federator.pegoutSignedCacheTtlInMinutes")).thenReturn(false);
+        when(config.root()).thenReturn(configObject);
+
+        FedNodeSystemProperties fedNodeSystemProperties = new FedNodeSystemProperties(configLoader);
+
+        assertEquals(
+            Duration.ofMinutes(defaultTtlInMinutes),
+            fedNodeSystemProperties.getPegoutSignedCacheTtl());
+    }
 }