Merge pull request #152 from cspr-rad/e2e-test-da

end-to-end test the data availability + nixos module updates + fixes

kairos-cli/Cargo.toml

@@ -14,7 +14,7 @@ license.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-default = ["demo"]
+default = ["demo", "database"]
 all-tests = ["cctl-tests", "database"]
 cctl-tests = []
 demo = ["dep:kairos-test-utils", "dep:tokio", "dep:dotenvy"]

kairos-server/Cargo.toml

@@ -12,7 +12,7 @@ name = "kairos-server"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-default = ["cctl-tests", "deposit-mock"]
+default = ["cctl-tests", "deposit-mock", "database"]
 all-tests = ["cctl-tests", "deposit-mock", "database"]
 cctl-tests = []
 deposit-mock = []

kairos-server/src/l1_sync/event_manager.rs

@@ -7,6 +7,8 @@ use casper_event_toolkit::rpc::client::CasperClient;
 
 use crate::state::ServerStateInner;
 use kairos_circuit_logic::transactions::{KairosTransaction, L1Deposit};
+#[cfg(feature = "database")]
+use kairos_data::transaction as db;
 
 use super::error::L1SyncError;
 
@@ -69,6 +71,16 @@ impl EventManager {
                     let recipient: Vec<u8> = deposit.recipient;
                     let txn = KairosTransaction::Deposit(L1Deposit { amount, recipient });
 
+                    #[cfg(feature = "database")]
+                    db::insert(&self.server_state.pool, txn.clone())
+                        .await
+                        .map_err(|e| {
+                            L1SyncError::UnexpectedError(format!(
+                                "Failed to add to database: {}",
+                                e
+                            ))
+                        })?;
+
                     // Push deposit to trie.
                     self.server_state
                         .batch_state_manager

kairos-server/src/lib.rs

@@ -26,25 +26,22 @@ use kairos_data::new as new_pool;
 pub type PublicKey = Vec<u8>;
 type Signature = Vec<u8>;
 
-#[cfg(not(feature = "deposit-mock"))]
 pub fn app_router(state: ServerState) -> Router {
-    Router::new()
+    let mut router = Router::new()
         .typed_post(routes::deposit_handler)
         .typed_post(routes::withdraw_handler)
         .typed_post(routes::transfer_handler)
-        .with_state(state)
-}
-
-#[cfg(feature = "deposit-mock")]
-pub fn app_router(state: ServerState) -> Router {
-    Router::new()
-        .typed_post(routes::deposit_handler)
-        .typed_post(routes::withdraw_handler)
-        .typed_post(routes::transfer_handler)
-        .typed_post(routes::deposit_mock_handler)
         .typed_post(routes::get_nonce_handler)
-        .typed_get(routes::contract_hash_handler)
-        .with_state(state)
+        .typed_get(routes::contract_hash_handler);
+    #[cfg(feature = "deposit-mock")]
+    {
+        router = router.typed_post(routes::deposit_mock_handler)
+    }
+    #[cfg(feature = "database")]
+    {
+        router = router.typed_post(routes::query_transactions_handler);
+    }
+    router.with_state(state)
 }
 
 pub async fn run_l1_sync(server_state: Arc<ServerStateInner>) {

kairos-server/src/routes/mod.rs

@@ -12,6 +12,8 @@ pub use contract_hash::contract_hash_handler;
 pub use deposit::deposit_handler;
 #[cfg(feature = "deposit-mock")]
 pub use deposit_mock::deposit_mock_handler;
+#[cfg(feature = "database")]
+pub use fetch::query_transactions_handler;
 pub use get_nonce::get_nonce_handler;
 pub use transfer::transfer_handler;
 pub use withdraw::withdraw_handler;

kairos-test-utils/Cargo.toml

@@ -14,6 +14,7 @@ bench = false
 [features]
 # FIXME enable cctl-tests once this crate is factored out in a separate repository
 #all-tests = ["cctl-tests"]
+default = ["database"]
 all-tests = ["database"]
 cctl-tests = []
 database = ["kairos-server/database"]

kairos-test-utils/src/kairos.rs

@@ -122,20 +122,22 @@ impl Drop for Kairos {
 #[cfg(test)]
 mod tests {
     use super::*;
+    #[cfg(feature = "database")]
+    use crate::postgres::PostgresDB;
     #[tokio::test]
     async fn test_kairos_starts_and_terminates() {
         let dummy_rpc = Url::parse("http://127.0.0.1:11101/rpc").unwrap();
         let dummy_sse = Url::parse("http://127.0.0.1:18101/events/main").unwrap();
         #[cfg(feature = "database")]
-        let dummy_postgres = Url::parse("postgres://kairos:kairos@localhost/kairos").unwrap();
+        let postgres = PostgresDB::run(None).unwrap();
 
         let _kairos = Kairos::run(
             &dummy_rpc,
             &dummy_sse,
             None,
             None,
             #[cfg(feature = "database")]
-            &dummy_postgres,
+            &postgres.connection.clone().into(),
         )
         .await
         .unwrap();

nixos/modules/kairos.nix

@@ -1,4 +1,4 @@
-{ lib, pkgs, config, ... }:
+{ lib, config, ... }:
 let
   inherit (lib)
     types
@@ -121,7 +121,6 @@ in
       };
     };
 
-
     logLevel = mkOption {
       type = types.enum [
         "error"
@@ -135,6 +134,44 @@ in
         The log-level that should be used.
       '';
     };
+
+    database =
+      {
+        host = mkOption {
+          type = types.str;
+          default = "/run/postgresql";
+          example = "/run/postgresql";
+          description = ''
+            Host of the PostgreSQL server
+          '';
+        };
+
+        port = mkOption {
+          type = types.port;
+          default = config.services.postgresql.settings.port;
+          description = ''
+            Port of the PostgreSQL server
+          '';
+        };
+
+        databaseName = mkOption {
+          type = types.str;
+          default = "kairos";
+          example = "kairos";
+          description = ''
+            Name of the PostgreSQL database
+          '';
+        };
+
+        userName = mkOption {
+          type = types.str;
+          default = "kairos";
+          example = "kairos";
+          description = ''
+            Username for the PostgreSQL connection
+          '';
+        };
+      };
   };
 
   config = mkIf cfg.enable {
@@ -144,8 +181,8 @@ in
         description = "kairos";
         documentation = [ "" ];
         wantedBy = [ "multi-user.target" ];
-        after = [ "network-online.target" "kairos-prover.service" ];
-        requires = [ "network-online.target" "kairos-prover.service" ];
+        after = [ "network-online.target" "kairos-prover.service" "postgresql.service" ];
+        requires = [ "network-online.target" "kairos-prover.service" "postgresql.service" ];
         environment = {
           RUST_LOG = cfg.logLevel;
           KAIROS_SERVER_SOCKET_ADDR = "${cfg.bindAddress}:${builtins.toString cfg.port}";
@@ -154,6 +191,7 @@ in
           KAIROS_SERVER_CASPER_SYNC_INTERVAL = builtins.toString cfg.casperSyncInterval;
           KAIROS_SERVER_DEMO_CONTRACT_HASH = cfg.demoContractHash;
           KAIROS_PROVER_SERVER_URL = "${cfg.prover.protocol}://${cfg.prover.bindAddress}:${builtins.toString cfg.prover.port}";
+          KAIROS_SERVER_DB_ADDR = "postgresql://${cfg.database.userName}@localhost:${builtins.toString cfg.database.port}/${cfg.database.databaseName}?host=${cfg.database.host}";
         } // optionalAttrs (!builtins.isNull cfg.prover.maxBatchSize) {
           KAIROS_SERVER_MAX_BATCH_SIZE = cfg.maxBatchSize;
         } // optionalAttrs (!builtins.isNull cfg.prover.maxBatchDuration) {
@@ -172,5 +210,16 @@ in
       enable = true;
       inherit (cfg.prover) bindAddress port;
     };
+
+    services.postgresql = {
+      enable = true;
+      ensureDatabases = [ cfg.database.databaseName ];
+      ensureUsers = [
+        {
+          name = cfg.database.userName;
+          ensureDBOwnership = true;
+        }
+      ];
+    };
   };
 }

nixos/tests/end-to-end.nix

@@ -83,7 +83,6 @@ nixosTest {
   testScript = ''
     import json
     import backoff
-    import time
 
     # Utils
     def verify_deploy_success(json_data):
@@ -104,6 +103,14 @@ nixosTest {
       if not verify_deploy_success(get_deploy_result):
         raise Exception("Success key not found in JSON")
 
+    @backoff.on_exception(backoff.expo, Exception, max_tries=5, jitter=backoff.full_jitter)
+    def wait_for_deposit(depositor, amount):
+      transactions_query = { "sender": depositor }
+      transactions_result = client.succeed("curl --fail-with-body -X POST http://kairos/api/v1/transactions -H 'Content-Type: application/json' -d '{}'".format(json.dumps(transactions_query)))
+      transactions = json.loads(transactions_result)
+      if not any(transaction.get("public_key") == depositor and transaction.get("amount") == str(amount) for transaction in transactions):
+        raise Exception("Couldn't find deposit for depositor {} with amount {} in transactions\n:{}".format(depositor, amount, transactions))
+
     # Test
     start_all()
 
@@ -132,14 +139,19 @@ nixosTest {
 
     wait_for_successful_deploy(deposit_deploy_hash)
 
-    # wait for l2 to sync with l1 every 5 seconds
-    time.sleep(${builtins.toString (casperSyncInterval * 2)})
+    wait_for_deposit(depositor, 3000000000)
 
     # transfer
     beneficiary = client.succeed("cat ${clientUsersDirectory}/user-3/public_key_hex")
     transfer_output = client.succeed("kairos-cli --kairos-server-address http://kairos transfer --amount 1000 --recipient {} --private-key {}".format(beneficiary, depositor_private_key))
     assert "Transfer successfully sent to L2\n" in transfer_output, "The transfer command was not successful: {}".format(transfer_output)
 
+    # data availability
+    transactions_query = { "recipient": beneficiary }
+    transactions_result = client.succeed("curl --fail-with-body -X POST http://kairos/api/v1/transactions -H 'Content-Type: application/json' -d '{}'".format(json.dumps(transactions_query)))
+    transactions = json.loads(transactions_result)
+    assert any(transaction.get("recipient") == beneficiary and transaction.get("amount") == str(1000) for transaction in transactions), "Couldn't find the transfer in the L2's DA: {}".format(transactions)
+
     # TODO test withdraw
 
     # TODO cctl does not provide any secp256k1 keys