DDLs and validations for recovery process

ddls/create-table/h2-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, HASH LONGVARBINARY NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY LONGVARBINARY NOT NULL, ENCRYPTED_PAYLOAD LONGVARBINARY NOT NULL, NONCE LONGVARBINARY NOT NULL, SENDER LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, HASH LONGVARBINARY NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID LONGVARBINARY NOT NULL, LOOKUP_ID LONGVARBINARY NOT NULL, DATA LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID BIGINT NOT NULL PRIMARY KEY, HASH VARCHAR NOT NULL, PAYLOAD LONGVARBINARY, PRIVACY_MODE INTEGER, TIMESTAMP BIGINT, VALIDATION_STAGE BIGINT);
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID BIGINT NOT NULL PRIMARY KEY, AFFECTED_HASH VARCHAR NOT NULL, TXN_ID BIGINT NOT NULL, CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID FOREIGN KEY (TXN_ID) REFERENCES ST_TRANSACTION (ID));
+CREATE INDEX IF NOT EXISTS ST_TRANSACTION_VALSTG ON ST_TRANSACTION(VALIDATION_STAGE);

ddls/create-table/hsql-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, HASH LONGVARBINARY NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY LONGVARBINARY NOT NULL, ENCRYPTED_PAYLOAD LONGVARBINARY NOT NULL, NONCE LONGVARBINARY NOT NULL, SENDER LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, HASH LONGVARBINARY NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID LONGVARBINARY NOT NULL, LOOKUP_ID LONGVARBINARY NOT NULL, DATA LONGVARBINARY NOT NULL, TIMESTAMP BIGINT, PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID BIGINT NOT NULL PRIMARY KEY, HASH VARCHAR(100) NOT NULL, PAYLOAD LONGVARBINARY, PRIVACY_MODE INTEGER, TIMESTAMP BIGINT, VALIDATION_STAGE BIGINT);
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID BIGINT NOT NULL PRIMARY KEY, AFFECTED_HASH VARCHAR(100) NOT NULL, TXN_ID BIGINT NOT NULL, CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID FOREIGN KEY (TXN_ID) REFERENCES ST_TRANSACTION (ID));
+CREATE INDEX IF NOT EXISTS ST_TRANSACTION_VALSTG ON ST_TRANSACTION(VALIDATION_STAGE);

ddls/create-table/mysql-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD BLOB NOT NULL, HASH VARBINARY(100) NOT NULL, TIMESTAMP BIGINT, PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY BLOB NOT NULL, ENCRYPTED_PAYLOAD BLOB NOT NULL, NONCE BLOB NOT NULL, SENDER BLOB NOT NULL, TIMESTAMP BIGINT, HASH VARBINARY(100) NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID VARBINARY(100) NOT NULL, LOOKUP_ID BLOB NOT NULL, DATA BLOB NOT NULL, TIMESTAMP BIGINT, PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID BIGINT(19) NOT NULL, HASH VARCHAR(100) NOT NULL, PAYLOAD BLOB, PRIVACY_MODE BIGINT(10), TIMESTAMP BIGINT(19), VALIDATION_STAGE BIGINT(19), PRIMARY KEY (ID));
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID BIGINT(19) NOT NULL, AFFECTED_HASH VARCHAR(100) NOT NULL, TXN_ID BIGINT(19) NOT NULL, CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID FOREIGN KEY (TXN_ID) REFERENCES ST_TRANSACTION(ID), PRIMARY KEY (ID));
+CREATE INDEX IF NOT EXISTS ST_TRANSACTION_VALSTG ON ST_TRANSACTION(VALIDATION_STAGE);

ddls/create-table/oracle-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD BLOB NOT NULL, HASH RAW(100) NOT NULL, TIMESTAMP NUMBER(19), PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY BLOB NOT NULL, ENCRYPTED_PAYLOAD BLOB NOT NULL, NONCE BLOB NOT NULL, SENDER BLOB NOT NULL, TIMESTAMP NUMBER(19), HASH RAW(100) NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID RAW(100) NOT NULL, LOOKUP_ID BLOB NOT NULL, DATA BLOB NOT NULL, TIMESTAMP NUMBER(19), PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID NUMBER(19) NOT NULL PRIMARY KEY, HASH VARCHAR(100) NOT NULL, PAYLOAD BLOB, PRIVACY_MODE INTEGER, TIMESTAMP NUMBER(19), VALIDATION_STAGE NUMBER(19));
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID NUMBER(19) NOT NULL PRIMARY KEY, AFFECTED_HASH VARCHAR(100) NOT NULL, TXN_ID NUMBER(19) NOT NULL, CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID FOREIGN KEY (TXN_ID) REFERENCES ST_TRANSACTION (ID));
+CREATE INDEX ST_TRANSACTION_VALSTG ON ST_TRANSACTION(VALIDATION_STAGE);

ddls/create-table/postgresql-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD BYTEA NOT NULL, HASH BYTEA NOT NULL, TIMESTAMP DECIMAL(19), PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY BYTEA NOT NULL, ENCRYPTED_PAYLOAD BYTEA NOT NULL, NONCE BYTEA NOT NULL, SENDER BYTEA NOT NULL, TIMESTAMP DECIMAL(19), HASH BYTEA NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID BYTEA NOT NULL, LOOKUP_ID BYTEA NOT NULL, DATA BYTEA NOT NULL, TIMESTAMP DECIMAL(19), PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID BIGINT NOT NULL CONSTRAINT ST_TRANSACTION_PKEY PRIMARY KEY, HASH VARCHAR(255) NOT NULL, PAYLOAD BYTEA, PRIVACY_MODE INTEGER, TIMESTAMP BIGINT,VALIDATION_STAGE BIGINT);
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID BIGINT NOT NULL CONSTRAINT ST_AFFECTED_TRANSACTION_PKEY PRIMARY KEY, AFFECTED_HASH VARCHAR(255) NOT NULL, TXN_ID BIGINT NOT NULL CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID REFERENCES ST_TRANSACTION);
+CREATE INDEX IF NOT EXISTS ST_TRANSACTION_VALSTG ON ST_TRANSACTION (VALIDATION_STAGE);

ddls/create-table/sqlite-ddl.sql

@@ -1,3 +1,6 @@
 CREATE TABLE ENCRYPTED_TRANSACTION (ENCODED_PAYLOAD BLOB NOT NULL, TIMESTAMP NUMBER(19), HASH BLOB NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE ENCRYPTED_RAW_TRANSACTION (ENCRYPTED_KEY BLOB NOT NULL, ENCRYPTED_PAYLOAD BLOB NOT NULL, NONCE BLOB NOT NULL, SENDER BLOB NOT NULL, TIMESTAMP NUMBER(19), HASH BLOB NOT NULL, PRIMARY KEY (HASH));
 CREATE TABLE PRIVACY_GROUP(ID BLOB NOT NULL, LOOKUP_ID BLOB NOT NULL, DATA BLOB NOT NULL, TIMESTAMP NUMBER(19), PRIMARY KEY (ID));
+CREATE TABLE ST_TRANSACTION(ID NUMBER(19) NOT NULL PRIMARY KEY, HASH VARCHAR NOT NULL, PAYLOAD BLOB, PRIVACY_MODE NUMBER(10), TIMESTAMP NUMBER(19), VALIDATION_STAGE NUMBER(19));
+CREATE TABLE ST_AFFECTED_TRANSACTION(ID NUMBER(19) NOT NULL PRIMARY KEY, AFFECTED_HASH VARCHAR NOT NULL, TXN_ID NUMBER(19) NOT NULL, CONSTRAINT FK_ST_AFFECTED_TRANSACTION_TXN_ID FOREIGN KEY (TXN_ID) REFERENCES ST_TRANSACTION (ID));
+CREATE INDEX IF NOT EXISTS ST_TRANSACTION_VALSTG ON ST_TRANSACTION(VALIDATION_STAGE);

tessera-data/src/main/java/com/quorum/tessera/data/staging/StagingAffectedTransaction.java

@@ -4,6 +4,9 @@
 import jakarta.validation.constraints.NotNull;
 import java.util.Objects;
 
+@NamedQuery(
+    name = "StagingAffectedTransaction.countAll",
+    query = "select count(sat) from StagingAffectedTransaction sat")
 @Entity
 @Table(name = "ST_AFFECTED_TRANSACTION")
 public class StagingAffectedTransaction {

tessera-data/src/main/java/com/quorum/tessera/data/staging/StagingEntityDAO.java

@@ -59,6 +59,13 @@ public interface StagingEntityDAO {
    */
   int updateStageForBatch(int batchSize, long validationStage);
 
+  /**
+   * counts all records in staging affected transactions
+   *
+   * @return the count
+   */
+  long countAllAffected();
+
   static StagingEntityDAO create() {
     return ServiceLoader.load(StagingEntityDAO.class).findFirst().get();
   }

tessera-data/src/main/java/com/quorum/tessera/data/staging/internal/StagingEntityDAOImpl.java

@@ -107,4 +107,12 @@ public int updateStageForBatch(int batchSize, long validationStage) {
           return resultList.size();
         });
   }
+
+  @Override
+  public long countAllAffected() {
+    return entityManagerTemplate.execute(
+        em ->
+            em.createNamedQuery("StagingAffectedTransaction.countAll", Long.class)
+                .getSingleResult());
+  }
 }

tessera-data/src/main/java/com/quorum/tessera/data/staging/internal/StagingEntityDAOProvider.java

@@ -34,11 +34,10 @@ public static StagingEntityDAO provider() {
 
     properties.put(
         "jakarta.persistence.schema-generation.database.action",
-        config.getJdbcConfig().isAutoCreateTables() ? "create" : "none");
+        config.getJdbcConfig().isAutoCreateTables() ? "drop-and-create" : "none");
 
     properties.put(
         "eclipselink.session.customizer", "com.quorum.tessera.eclipselink.AtomicLongSequence");
-    properties.put("jakarta.persistence.schema-generation.database.action", "drop-and-create");
 
     LOGGER.debug("Creating EntityManagerFactory from {}", properties);
     final EntityManagerFactory entityManagerFactory =

tessera-data/src/main/java/module-info.java

@@ -8,6 +8,7 @@
   requires tessera.encryption.api;
   requires tessera.shared;
   requires java.sql;
+  requires java.sql.rowset;
   requires com.zaxxer.hikari;
   requires jakarta.validation;
   requires tessera.eclipselink.utils;

tessera-data/src/test/java/com/quorum/tessera/data/staging/internal/StagingEntityDAOTest.java

@@ -148,6 +148,8 @@ public void testStagingQuery() {
         .isEqualTo(1);
     assertThat(stagingEntityDAO.countAll()).isEqualTo(7);
     assertThat(stagingEntityDAO.countStaged()).isEqualTo(6);
+
+    assertThat(stagingEntityDAO.countAllAffected()).isEqualTo(7);
   }
 
   @Test
@@ -197,6 +199,8 @@ public void paginationCanCauseDifferentStagingValueButOrderShouldBeMaintained()
         .isEqualTo(1);
     assertThat(stagingEntityDAO.countAll()).isEqualTo(7);
     assertThat(stagingEntityDAO.countStaged()).isEqualTo(6);
+
+    assertThat(stagingEntityDAO.countAllAffected()).isEqualTo(7);
   }
 
   @Test

tessera-recover/src/main/java/com/quorum/tessera/recovery/internal/RecoveryImpl.java

@@ -155,6 +155,18 @@ public RecoveryResult sync() {
   @Override
   public int recover() {
 
+    try {
+      if (stagingEntityDAO.countAll() != 0 || stagingEntityDAO.countAllAffected() != 0) {
+        LOGGER.error(
+            "Staging tables are not empty. Please ensure database has been setup correctly for recovery process");
+        return RecoveryResult.FAILURE.getCode();
+      }
+    } catch (Exception ex) {
+      LOGGER.error(
+          "Attempt to query failed. Please ensure database has been setup correctly for recovery process");
+      return RecoveryResult.FAILURE.getCode();
+    }
+
     final long startTime = System.nanoTime();
 
     LOGGER.debug("Requesting transactions from other nodes");

tessera-recover/src/test/java/com/quorum/tessera/recovery/internal/RecoveryImplTest.java

@@ -16,6 +16,7 @@
 import com.quorum.tessera.recovery.resend.BatchTransactionRequester;
 import com.quorum.tessera.transaction.TransactionManager;
 import com.quorum.tessera.transaction.exception.PrivacyViolationException;
+import jakarta.persistence.PersistenceException;
 import java.util.List;
 import org.junit.After;
 import org.junit.Before;
@@ -331,6 +332,9 @@ public void testRecoverSuccess() {
     verify(spy).request();
     verify(spy).stage();
     verify(spy).sync();
+
+    verify(stagingEntityDAO).countAll();
+    verify(stagingEntityDAO).countAllAffected();
   }
 
   @Test
@@ -347,6 +351,9 @@ public void testRecoverPartialSuccess() {
     verify(spy).request();
     verify(spy).stage();
     verify(spy).sync();
+
+    verify(stagingEntityDAO).countAll();
+    verify(stagingEntityDAO).countAllAffected();
   }
 
   @Test
@@ -363,5 +370,45 @@ public void testRecoverFailed() {
     verify(spy).request();
     verify(spy).stage();
     verify(spy).sync();
+
+    verify(stagingEntityDAO).countAll();
+    verify(stagingEntityDAO).countAllAffected();
+  }
+
+  @Test
+  public void testDBStagingTxNotEmpty() {
+
+    final Recovery spy = spy(recovery);
+
+    when(stagingEntityDAO.countAll()).thenReturn(1L);
+
+    assertThat(spy.recover()).isEqualTo(2);
+
+    verify(stagingEntityDAO).countAll();
+  }
+
+  @Test
+  public void testDBStagingAffectedTxNotEmpty() {
+
+    final Recovery spy = spy(recovery);
+
+    when(stagingEntityDAO.countAllAffected()).thenReturn(1L);
+
+    assertThat(spy.recover()).isEqualTo(2);
+
+    verify(stagingEntityDAO).countAll();
+    verify(stagingEntityDAO).countAllAffected();
+  }
+
+  @Test
+  public void testDBTableNotExisted() {
+
+    final Recovery spy = spy(recovery);
+
+    when(stagingEntityDAO.countAll()).thenThrow(new PersistenceException("OUCH"));
+
+    assertThat(spy.recover()).isEqualTo(2);
+
+    verify(stagingEntityDAO).countAll();
   }
 }

tests/acceptance-test/src/test/java/com/quorum/tessera/test/rest/RecoverIT.java

@@ -32,7 +32,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -59,7 +58,12 @@ public class RecoverIT {
 
   private DBType dbType;
 
+  private boolean autoCreateTables;
+
+  private DatabaseServer dbServer;
+
   public RecoverIT(TestConfig testConfig) {
+    this.autoCreateTables = testConfig.autoCreateTables;
     this.dbType = testConfig.dbType;
   }
 
@@ -74,6 +78,7 @@ public void startNetwork() throws Exception {
             .with(EncryptorType.NACL)
             .with(ClientMode.TESSERA)
             .prefix(RecoverIT.class.getSimpleName().toLowerCase())
+            .withAutoCreateTables(autoCreateTables)
             .createAndSetupContext();
 
     partyHelper = PartyHelper.create();
@@ -85,11 +90,14 @@ public void startNetwork() throws Exception {
             .collect(Collectors.toList());
 
     String nodeId = NodeId.generate(executionContext);
-    DatabaseServer databaseServer = executionContext.getDbType().createDatabaseServer(nodeId);
-    databaseServer.start();
+    dbServer = executionContext.getDbType().createDatabaseServer(nodeId);
+    dbServer.start();
 
     setupDatabase = new SetupDatabase(executionContext);
-    setupDatabase.setUp();
+
+    if (!autoCreateTables) {
+      setupDatabase.setUp();
+    }
 
     this.executors =
         executionContext.getConfigs().stream()
@@ -119,6 +127,7 @@ public void stopNetwork() throws Exception {
     setupDatabase.dropAll();
     ExecutionContext.destroyContext();
     executors.values().forEach(ExecManager::stop);
+    dbServer.stop();
   }
 
   void sendTransactions() {
@@ -163,24 +172,69 @@ public void recoverNodes() throws Exception {
     Collections.shuffle(aliases);
 
     for (NodeAlias nodeAlias : aliases) {
+
+      // Stop node
+      ExecManager execManager = executors.get(nodeAlias);
+      execManager.stop();
+      // Drop database
+      setupDatabase.drop(nodeAlias);
+
+      if (!autoCreateTables) {
+        recoverNodeShouldFail(nodeAlias); // as staging tables not existed
+
+        setupDatabase.setUp(nodeAlias);
+        insertBogusData(nodeAlias);
+
+        recoverNodeShouldFail(nodeAlias); // as staging tables contain data
+
+        setupDatabase.drop(nodeAlias);
+
+        // Let's setup correctly
+        setupDatabase.setUp(nodeAlias);
+      }
+
+      // Should recover successfully
       recoverNode(nodeAlias);
     }
   }
 
-  private void recoverNode(NodeAlias nodeAlias) throws Exception {
+  private void insertBogusData(NodeAlias nodeAlias) {
 
-    ExecManager execManager = executors.get(nodeAlias);
-    execManager.stop();
-    setupDatabase.drop(nodeAlias);
-    setupDatabase.setUp(nodeAlias);
+    Connection connection = partyHelper.findByAlias(nodeAlias).getDatabaseConnection();
 
-    assertThat(doCount(nodeAlias)).isZero();
+    try (connection) {
 
+      PreparedStatement preparedStatement =
+          connection.prepareStatement(
+              "INSERT INTO ST_TRANSACTION(ID, HASH, PAYLOAD) VALUES (?,?,?)");
+      preparedStatement.setInt(1, 1);
+      preparedStatement.setString(2, Base64.getEncoder().encodeToString("hash".getBytes()));
+      preparedStatement.setBytes(3, "payload".getBytes());
+      try (preparedStatement) {
+        preparedStatement.execute();
+      }
+    } catch (SQLException ex) {
+      throw new UncheckedSQLException(ex);
+    }
+  }
+
+  private void recoverNodeShouldFail(NodeAlias nodeAlias) throws InterruptedException {
+    ExecManager execManager = executors.get(nodeAlias);
     RecoveryExecManager recoveryExecManager =
         new RecoveryExecManager(execManager.getConfigDescriptor());
-
     Process process = recoveryExecManager.start();
+    int exitCode = process.waitFor();
+    assertThat(exitCode).isEqualTo(2);
+    recoveryExecManager.stop();
+  }
+
+  private void recoverNode(NodeAlias nodeAlias) throws Exception {
+
+    ExecManager execManager = executors.get(nodeAlias);
 
+    RecoveryExecManager recoveryExecManager =
+        new RecoveryExecManager(execManager.getConfigDescriptor());
+    Process process = recoveryExecManager.start();
     int exitCode = process.waitFor();
 
     assertThat(exitCode).describedAs("Exit code should be zero. %s", nodeAlias.name()).isZero();
@@ -285,19 +339,27 @@ private void partyInfoSync() throws InterruptedException {
 
   @Parameterized.Parameters(name = "{0}")
   public static List<TestConfig> configs() {
-    return Stream.of(DBType.SQLITE).map(TestConfig::new).collect(Collectors.toUnmodifiableList());
+    return List.of(
+        new TestConfig(DBType.H2, true),
+        new TestConfig(DBType.H2, false),
+        //        new TestConfig(DBType.HSQL, true),
+        //        new TestConfig(DBType.HSQL, false),
+        new TestConfig(DBType.SQLITE, true),
+        new TestConfig(DBType.SQLITE, false));
   }
 
   static class TestConfig {
     DBType dbType;
+    boolean autoCreateTables;
 
-    TestConfig(DBType dbType) {
-      this.dbType = Objects.requireNonNull(dbType);
+    TestConfig(DBType dbType, boolean autoCreateTables) {
+      this.dbType = dbType;
+      this.autoCreateTables = autoCreateTables;
     }
 
     @Override
     public String toString() {
-      return "TestConfig{" + "dbType=" + dbType + '}';
+      return "TestConfig{" + "dbType=" + dbType + ", autoCreateTables=" + autoCreateTables + '}';
     }
   }
 }

tests/acceptance-test/src/test/java/config/ConfigBuilder.java

@@ -151,13 +151,10 @@ public Config build() {
     config.setEncryptor(encryptorConfig);
     JdbcConfig jdbcConfig = new JdbcConfig();
 
-    String nodeName =
-        executionContext.getPrefix().map(s -> s.concat("-").concat(nodeId)).orElse(nodeId);
-
-    jdbcConfig.setUrl(executionContext.getDbType().createUrl(nodeName, nodeNumber));
+    jdbcConfig.setUrl(executionContext.getDbType().createUrl(nodeId, nodeNumber));
     jdbcConfig.setUsername("sa");
-    jdbcConfig.setPassword("");
-    jdbcConfig.setAutoCreateTables(true);
+    jdbcConfig.setPassword("password");
+    jdbcConfig.setAutoCreateTables(executionContext.isAutoCreateTables());
     config.setJdbcConfig(jdbcConfig);
 
     ServerConfig q2tServerConfig = new ServerConfig();