feat(cat-gateway): Add autogenerated cassandra schema version and syn…

…c status tables. (#889) * fix(docs): Fix up docs issues * fix(backend): Huge refactor to prep for scylladb config management * fix(backend): Clean up logging a little, and add build info logs as required for production. * Refactor and setup cassandra config/session * feat(backend): Index DB schema setup seems to work * WIP * fix(rust): Format fixes * fix(rust): Build fixes * fix(rust): Adjust index DB so we can index without querying, and can optimize on first detected spend. * fix(rust): add more docs * fix(rust): basic new follower integration * fix(rust): wip * fix(ci): Bump rust compiler version to match CI * ci(backend): Bump rust version to match CI * fix(backend): Fix code format and lints * feat(backend): simple new block indexer just to test the logic works * feat(gateway): Simple indexing with cassandra seems to work * refactor(backend): Remove lazy and once_cell in favor of new standard library replacements * fix(backend): WIP indexing for stake addresses and unstaked ada * fix(backend): indexing WIP * fix(backend): Add support for log control with env vars, default to mainnet, adjust `justfile` to properly select preprod and also refresh git dependencies. * feat(backend): Make local test scylla db run with 4 nodes, not 1 * fix(backend-lib): Add stop for cassandra db cluster * refactor(backend-lib): Remove c509-certificate because its moved to catalyst-libs * fix(backend): Remove dependencies from Workspace, and move into project * fix(backend): Use temporary cat-ci branch for rust builders * fix(backend): Remove obsolete common crates subdirectory * fix(backend): Don't use pre-packaged mithril snapshots in integration tests * fix(backend): Fix code so it builds with latest chain follower code. Also eliminates redundant logic now incorporated into chain follower. * fix(backend): Fix broken reference to catalyst libs * ci(ci): Bump all earthfiles to latest WIP cat-ci branch * fix(frontend-pkg): Ignore .dart_tool directory in frontend files checking markdown * fix(ci): Fix spelling * fix(spelling): Add more project words and properly sort list * fix(backend): Sync rust configs and add target to make it easier in future * fix(backend): Enable all features of Scylla for now. * fix(frontend-pkg): Fix markdown table having too many columns * ci(spelling): Fix spelling issues * fix(docs): Bump docs to latest WIP cat-ci version * feat(gateway): Add low resource scylla db instance for local testing * feat(gateway): Add and update developer convenience functions for backend * fix(backend): Fix code format * fix(backend): Fix spelling issues in CQL files * fix(spelling): Remove duplicates from the project words dictionary * fix(backend): Get the backend building properly with earthly. * feat(backend): remove obsoleted postgres logic for chain indexing * revert(event-db): Revert extension changes to sql files after fixing sqlfluff version * fix(frontend): Regenerate the dart api interface file, and add doing that to the pre-push just command * fix(backend): temporarily disable API tests * fix(backend): Also temporarily stop workflow consuming test reports that are disabled * fix(ci): Try and stop coveralls running for api-tests * ci(general): Replace temp CI branch with tagged release * feat: Add Handler for Permissionless Auth (#825) * docs(cips): Add Formal Defintion of auth token * fix(docs): Fix comments in cddl file * fix(docs): sig size * fix(docs): Rename CDDL for the auth token * docs(docs): Add auth-header documentation * docs(docs): Fix markdown line length error * docs(general): Fix spelling * fix(backend-lib): Bump to catalyst-libs tagged version * fix(backend): stub out obsolete code (to be removed in follow up PR). * fix(backend-lib): code format * fix(backend): remove unused crate dependencies * feat: auth token (#723) * feat(auth token encode and decode): permissionless auth * feat(auth token encode and decode): permissionless auth * feat(auth token encode and decode): permissionless auth * feat(auth token encode and decode): permissionless auth * feat(auth token encode and decode): permissionless auth * iron out tests * iron out tests * refactor(auth token encode and decode): ed25519 Signature cbor fields Sig over the preceding two fields - sig(cbor(kid), cbor(ulid)) * refactor(auth token encode and decode): ed25519 Signature cbor fields Sig over the preceding two fields - sig(cbor(kid), cbor(ulid)) * feat(cat security scheme): open api * feat(cat security scheme): open api * feat(mock cert state): given kid from bearer return pub key * feat(auth token): cache TTL * feat(auth token): cache TTL * feat(auth token): cache TT * ci(spell check): fix * ci(spell check): fix * ci(spell check): fix * refactor(clippy): housekeeping tidy * refactor(clippy): housekeeping tidy * refactor(clippy): housekeeping tidy * refactor(clippy): housekeeping tidy * fix(backend): Re-enable dependent crates used by this code * fix(backend): clippy lints * fix(backend): spelling --------- Co-authored-by: Steven Johnson <[email protected]> Co-authored-by: Steven Johnson <[email protected]> * feat: Update GET staked_ada endpoint to fetch from ScyllaDB (#728) * feat: get staked ada from scylladb * chore: revert justfile changes * chore: filter TXOs in rust instead of filtering in ScyllaDB query * fix(backend): spelling * fix(backend): Eliminate lint errors from Derived function * fix(backend): code format * fix(backend): Udate autogenerated dart code * chore(cat-voices): fix tests --------- Co-authored-by: Steven Johnson <[email protected]> Co-authored-by: Steven Johnson <[email protected]> Co-authored-by: Dominik Toton <[email protected]> * feat: DB Indexing for CIP-36 registrations (#788) * feat: add schema for cip-36 registration tables * feat: index cip-36 by stake address * feat: index cip-36 registrations by vote key * fix: use TxiInserParams::new when adding txi data * fix: remove unused cfg attributes * fix: refactor Cip36RegistrationInsertQuery::new * fix(backend): Refactor queries and add multiple tables for cip36 registration indexes * fix(backend): Cip36 Primary key is stake key. Stake Key N->1 Vote Key * fix(backend): code format --------- Co-authored-by: Steven Johnson <[email protected]> Co-authored-by: Steven Johnson <[email protected]> * docs(general): Cleanup project dictionary * docs(spelling): Fix spelling * fix(backend): remove obsolete clippy lint cfg * docs(backend): Improve field documentation so its not ambiguous. * docs(backend): Fix comment * docs(backend): Improve comment * fix(backend): Vote Key index logic, and update comments * fix(backend): Earthfile needs to be executed from root of repo, to properly pick up secrets * fix(backend): make generic saturating value converter and use it instead of type specific ones * test(cat-gateway): Add tests for float conversion and better docs about functions limitations. * fix(cat-gateway): Developer lints in release mode, and also refer to correct local release binary * fix(cat-gateway): CIP36 index schema error * fix(cat-gateway): Cip36 indexing working, improve bad cassandra query reporting. * refactor(cat-gateway): Make settings a sub-module * refactor(cat-gateway): Break up Envvar handling into multiple files * refactor(cat-gateway): Fix code format * feat(cat-gateway): Add chain-sync downloader options to cat-gateway env vars * test(cat-gateway): Make debug logs visible in local testing * fix(cat-gateway): Minimum timeout for mithril downloads * test(cat-gateway): Silence gratuitous debug log in cassandra queries * refactor(cat-gateway): Decrease verboseness of the string env var logging code * fix(general): Bump alpine version and pin to fix missing upstream containers * ci(general): Bump cat-ci to v3.2.07 * ci(general): fix docker in docker container image * feat(cat-gateway): Autogenerate the cassandra schema version and add schema sync progress table * fix(cat-gateway): Bump rust library dependency versions * fix(cat-gateway): Make cassandra namespace compliant. * refactor(cat-gateway): rename utilities to utils * refactor(cat-gateway): use ensure! instead of if/bail * ci(general): Bump cat-ci version to latest * fix(flutter): Again, try and see if caching is working ok for flutter builds with cat-ci * ci(general): Bump CI to latest version to try and fix docs build issues * docs(cat-gateway): Better document the generate_cql_schema_version function * fix(cat-gateway): Use a string array reference rather than a vec of strings for strings to uuid generation * refactor(cat-gateway): fix code format --------- Co-authored-by: cong-or <[email protected]> Co-authored-by: Felipe Rosa <[email protected]> Co-authored-by: Dominik Toton <[email protected]> Co-authored-by: Joaquín Rosales <[email protected]>
input-output-hk · Sep 28, 2024 · ff7a0d7 · ff7a0d7
1 parent b76bd6d
commit ff7a0d7
Show file tree

Hide file tree

Showing 21 changed files with 365 additions and 99 deletions.
diff --git a/.config/dictionaries/project.dic b/.config/dictionaries/project.dic
@@ -189,6 +189,7 @@ Precertificate
 preprod
 projectcatalyst
 Prokhorenko
+proptest
 psql
 Ptarget
 pubkey

diff --git a/Earthfile b/Earthfile
@@ -1,8 +1,8 @@
 VERSION 0.8
 
-IMPORT github.com/input-output-hk/catalyst-ci/earthly/mdlint:v3.2.07 AS mdlint-ci
-IMPORT github.com/input-output-hk/catalyst-ci/earthly/cspell:v3.2.07 AS cspell-ci
-IMPORT github.com/input-output-hk/catalyst-ci/earthly/postgresql:v3.2.07 AS postgresql-ci
+IMPORT github.com/input-output-hk/catalyst-ci/earthly/mdlint:v3.2.10 AS mdlint-ci
+IMPORT github.com/input-output-hk/catalyst-ci/earthly/cspell:v3.2.10 AS cspell-ci
+IMPORT github.com/input-output-hk/catalyst-ci/earthly/postgresql:v3.2.10 AS postgresql-ci
 
 FROM debian:stable-slim
 

diff --git a/catalyst-gateway/Earthfile b/catalyst-gateway/Earthfile
@@ -1,6 +1,6 @@
 VERSION 0.8
 
-IMPORT github.com/input-output-hk/catalyst-ci/earthly/rust:v3.2.07 AS rust-ci
+IMPORT github.com/input-output-hk/catalyst-ci/earthly/rust:v3.2.10 AS rust-ci
 
 #cspell: words rustfmt toolsets USERARCH stdcfgs
 

diff --git a/catalyst-gateway/bin/Cargo.toml b/catalyst-gateway/bin/Cargo.toml
@@ -21,7 +21,7 @@ pallas = { version = "0.30.1", git = "https://github.com/input-output-hk/catalys
 pallas-traverse = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
 #pallas-crypto = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
 
-clap = { version = "4.5.17", features = ["derive", "env"] }
+clap = { version = "4.5.18", features = ["derive", "env"] }
 tracing = { version = "0.1.40", features = ["log"] }
 tracing-subscriber = { version = "0.3.18", features = [
     "fmt",
@@ -31,25 +31,25 @@ tracing-subscriber = { version = "0.3.18", features = [
     "time",
     "env-filter",
 ] }
-serde = { version = "1.0.204", features = ["derive"] }
+serde = { version = "1.0.210", features = ["derive"] }
 serde_json = "1.0.128"
-thiserror = "1.0.63"
+thiserror = "1.0.64"
 chrono = "0.4.38"
 # async-trait = "0.1.82"
 bb8 = "0.8.5"
 bb8-postgres = "0.8.1"
-tokio-postgres = { version = "0.7.11", features = [
+tokio-postgres = { version = "0.7.12", features = [
     "with-chrono-0_4",
     "with-serde_json-1",
     "with-time-0_3",
 ] }
-tokio = { version = "1.39.2", features = ["rt", "macros", "rt-multi-thread"] }
+tokio = { version = "1.40.0", features = ["rt", "macros", "rt-multi-thread"] }
 dotenvy = "0.15.7"
-local-ip-address = "0.6.2"
+local-ip-address = "0.6.3"
 gethostname = "0.5.0"
 hex = "0.4.3"
-handlebars = "6.0.0"
-anyhow = "1.0.86"
+handlebars = "6.1.0"
+anyhow = "1.0.89"
 #cddl = "0.9.4"
 #ciborium = "0.2.2"
 # stringzilla = "3.9.3"
@@ -69,8 +69,8 @@ rust_decimal = { version = "1.36.0", features = [
     "serde-with-float",
     "db-tokio-postgres",
 ] }
-poem = { version = "3.0.4", features = ["embed", "prometheus", "compression"] }
-poem-openapi = { version = "5.0.3", features = [
+poem = { version = "3.1.0", features = ["embed", "prometheus", "compression"] }
+poem-openapi = { version = "5.1.1", features = [
     "openapi-explorer",
     "rapidoc",
     "redoc",
@@ -81,15 +81,18 @@ poem-openapi = { version = "5.0.3", features = [
 ] }
 uuid = { version = "1.10.0", features = ["v4", "serde"] }
 ulid = { version = "1.1.3", features = ["serde", "uuid"] }
-cryptoxide = "0.4.4" # TODO: For blake2b replace with blake2b_simd.
+blake2b_simd = "1.0.2"
 url = "2.5.2"
 panic-message = "0.3.0"
 cpu-time = "1.0.0"
 prometheus = "0.13.4"
 rust-embed = "8.5.0"
 num-traits = "0.2.19"
 base64 = "0.22.1"
-dashmap = "6.0.1"
+dashmap = "6.1.0"
+
+[dev-dependencies]
+proptest = "1.5.0"
 
 [build-dependencies]
 build-info-build = "0.0.38"
diff --git a/catalyst-gateway/bin/src/cardano/util.rs b/catalyst-gateway/bin/src/cardano/util.rs
@@ -1,5 +1,4 @@
 //! Block stream parsing and filtering utils
-use cryptoxide::{blake2b::Blake2b, digest::Digest};
 use pallas::ledger::{
     primitives::conway::StakeCredential,
     traverse::{Era, MultiEraAsset, MultiEraCert, MultiEraPolicyAssets},
@@ -18,21 +17,6 @@ pub type StakeCredentialHash = String;
 /// Correct stake credential key in hex
 pub type StakeCredentialKey = String;
 
-/// Hash size
-#[allow(dead_code)]
-pub(crate) const BLAKE_2B_256_HASH_SIZE: usize = 256 / 8;
-
-/// Helper function to generate the `blake2b_256` hash of a byte slice
-#[allow(dead_code)]
-pub(crate) fn hash(bytes: &[u8]) -> [u8; BLAKE_2B_256_HASH_SIZE] {
-    let mut digest = [0u8; BLAKE_2B_256_HASH_SIZE];
-    let mut context = Blake2b::new(BLAKE_2B_256_HASH_SIZE);
-    context.input(bytes);
-    context.result(&mut digest);
-
-    digest
-}
-
 #[derive(Default, Debug, Serialize)]
 /// Assets
 pub struct Asset {

diff --git a/catalyst-gateway/bin/src/db/index/schema/cql/sync_status.cql b/catalyst-gateway/bin/src/db/index/schema/cql/sync_status.cql
@@ -0,0 +1,11 @@
+-- Most recent completed 
+-- Can also be used to convert a known stake key hash back to a full stake address.
+CREATE TABLE IF NOT EXISTS sync_status (
+    -- Primary Key Data
+    end_slot        varint,    -- The slot that has been indexed up-to (inclusive).
+    start_slot      varint,    -- The slot the sync block started at (inclusive).
+    sync_time       timestamp, -- The time we finished the sync.
+    node_id         uuid,      -- The node that synced this data.
+
+    PRIMARY KEY (end_slot, start_slot, sync_time, node_id)
+);
diff --git a/catalyst-gateway/bin/src/db/index/schema/mod.rs b/catalyst-gateway/bin/src/db/index/schema/mod.rs
@@ -8,17 +8,27 @@ use scylla::Session;
 use serde_json::json;
 use tracing::error;
 
-use crate::settings::cassandra_db;
+use crate::{settings::cassandra_db, utils::blake2b_hash::generate_uuid_string_from_data};
+
+/// The version of the Index DB Schema we SHOULD BE using.
+/// DO NOT change this unless you are intentionally changing the Schema.
+///
+/// This constant is ONLY used by Unit tests to identify when the schema version will
+/// change accidentally, and is NOT to be used directly to set the schema version of the
+/// table namespaces.
+#[allow(dead_code)]
+const SCHEMA_VERSION: &str = "a0e54866-1f30-8ad2-9ac7-df1cfaf9c634";
 
 /// Keyspace Create (Templated)
 const CREATE_NAMESPACE_CQL: &str = include_str!("./cql/namespace.cql");
 
-/// The version of the Schema we are using.
-/// Must be incremented if there is a breaking change in any schema tables below.
-pub(crate) const SCHEMA_VERSION: u64 = 1;
-
 /// All Schema Creation Statements
 const SCHEMAS: &[(&str, &str)] = &[
+    (
+        // Sync Status Table Schema
+        include_str!("./cql/sync_status.cql"),
+        "Create Sync Status Table",
+    ),
     (
         // TXO by Stake Address Table Schema
         include_str!("./cql/txo_by_stake_table.cql"),
@@ -66,10 +76,83 @@ const SCHEMAS: &[(&str, &str)] = &[
     ),
 ];
 
+/// Removes all comments from each line in the input query text and joins the remaining
+/// lines into a single string, reducing consecutive whitespace characters to a single
+/// space. Comments are defined as any text following `--` on a line.
+///
+/// # Arguments
+///
+/// * `text`: A string slice that holds the query to be cleaned.
+///
+/// # Returns
+///
+/// A new string with comments removed and whitespace reduced, where each remaining line
+/// from the original text is separated by a newline character.
+fn remove_comments_and_join_query_lines(text: &str) -> String {
+    // Split the input text into lines, removing any trailing empty lines
+    let raw_lines: Vec<&str> = text.lines().collect();
+    let mut clean_lines: Vec<String> = Vec::new();
+
+    // Filter out comments from each line
+    for line in raw_lines {
+        let mut clean_line = line.to_string();
+        if let Some(no_comment) = line.split_once("--") {
+            clean_line = no_comment.0.to_string();
+        }
+        clean_line = clean_line
+            .split_whitespace()
+            .collect::<Vec<&str>>()
+            .join(" ")
+            .trim()
+            .to_string();
+        if !clean_line.is_empty() {
+            clean_lines.push(clean_line);
+        }
+    }
+    clean_lines.join("\n")
+}
+
+/// Generates a unique schema version identifier based on the content of all CQL schemas.
+///
+/// This function processes each CQL schema, removes comments from its lines and joins
+/// them into a single string. It then sorts these processed strings to ensure consistency
+/// in schema versions regardless of their order in the list. Finally, it generates a UUID
+/// from a 127 bit hash of this sorted collection of schema contents, which serves as a
+/// unique identifier for the current version of all schemas.
+///
+/// # Returns
+///
+/// A string representing the UUID derived from the concatenated and cleaned CQL
+/// schema contents.
+fn generate_cql_schema_version() -> String {
+    // Where we will actually store the bytes we derive the UUID from.
+    let mut clean_schemas: Vec<String> = Vec::new();
+
+    // Iterate through each CQL schema and add it to the list of clean schemas documents.
+    for (schema, _) in SCHEMAS {
+        let schema = remove_comments_and_join_query_lines(schema);
+        if !schema.is_empty() {
+            clean_schemas.push(schema);
+        }
+    }
+
+    // make sure any re-ordering of the schemas in the list does not effect the generated
+    // schema version
+    clean_schemas.sort();
+
+    // Generate a unique hash of the clean schemas,
+    // and use it to form a UUID to identify the schema version.
+    generate_uuid_string_from_data("Catalyst-Gateway Index Database Schema", &clean_schemas)
+}
+
 /// Get the namespace for a particular db configuration
 pub(crate) fn namespace(cfg: &cassandra_db::EnvVars) -> String {
     // Build and set the Keyspace to use.
-    format!("{}_V{}", cfg.namespace.as_str(), SCHEMA_VERSION)
+    format!(
+        "{}_{}",
+        cfg.namespace.as_str(),
+        generate_cql_schema_version().replace('-', "_")
+    )
 }
 
 /// Create the namespace we will use for this session
@@ -83,11 +166,19 @@ async fn create_namespace(
     // disable default `html_escape` function
     // which transforms `<`, `>` symbols to `&lt`, `&gt`
     reg.register_escape_fn(|s| s.into());
-    let query = reg.render_template(CREATE_NAMESPACE_CQL, &json!({"keyspace": keyspace}))?;
+    let query = reg
+        .render_template(CREATE_NAMESPACE_CQL, &json!({"keyspace": keyspace}))
+        .context(format!("Keyspace: {keyspace}"))?;
 
     // Create the Keyspace if it doesn't exist already.
-    let stmt = session.prepare(query).await?;
-    session.execute_unpaged(&stmt, ()).await?;
+    let stmt = session
+        .prepare(query)
+        .await
+        .context(format!("Keyspace: {keyspace}"))?;
+    session
+        .execute_unpaged(&stmt, ())
+        .await
+        .context(format!("Keyspace: {keyspace}"))?;
 
     // Wait for the Schema to be ready.
     session.await_schema_agreement().await?;
@@ -104,22 +195,88 @@ async fn create_namespace(
 pub(crate) async fn create_schema(
     session: &mut Arc<Session>, cfg: &cassandra_db::EnvVars,
 ) -> anyhow::Result<()> {
-    create_namespace(session, cfg).await?;
-
-    for schema in SCHEMAS {
-        let stmt = session
-            .prepare(schema.0)
-            .await
-            .context(format!("{} : Prepared", schema.1))?;
-
-        session
-            .execute_unpaged(&stmt, ())
-            .await
-            .context(format!("{} : Executed", schema.1))?;
+    create_namespace(session, cfg)
+        .await
+        .context("Creating Namespace")?;
+
+    let mut failed = false;
+
+    for (schema, schema_name) in SCHEMAS {
+        match session.prepare(*schema).await {
+            Ok(stmt) => {
+                if let Err(err) = session.execute_unpaged(&stmt, ()).await {
+                    failed = true;
+                    error!(schema=schema_name, error=%err, "Failed to Execute Create Schema Query");
+                };
+            },
+            Err(err) => {
+                failed = true;
+                error!(schema=schema_name, error=%err, "Failed to Prepare Create Schema Query");
+            },
+        }
     }
 
+    anyhow::ensure!(!failed, "Failed to Create Schema");
+
     // Wait for the Schema to be ready.
     session.await_schema_agreement().await?;
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    /// This test is designed to fail if the schema version has changed.
+    /// It is used to help detect inadvertent schema version changes.
+    /// If you did NOT intend to change the index db schema and this test fails,
+    /// then revert or fix your changes to the schema files.
+    fn check_schema_version_has_not_changed() {
+        let calculated_version = generate_cql_schema_version();
+        assert_eq!(SCHEMA_VERSION, calculated_version);
+    }
+
+    #[test]
+    fn test_no_comments() {
+        let input = "SELECT * FROM table1;";
+        let expected_output = "SELECT * FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_single_line_comment() {
+        let input = "SELECT -- some comment * FROM table1;";
+        let expected_output = "SELECT";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_multi_line_comment() {
+        let input = "SELECT -- some comment\n* FROM table1;";
+        let expected_output = "SELECT\n* FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_multiple_lines() {
+        let input = "SELECT * FROM table1;\n-- another comment\nSELECT * FROM table2;";
+        let expected_output = "SELECT * FROM table1;\nSELECT * FROM table2;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_empty_lines() {
+        let input = "\n\nSELECT * FROM table1;\n-- comment here\n\n";
+        let expected_output = "SELECT * FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_whitespace_only() {
+        let input = "   \n  -- comment here\n   ";
+        let expected_output = "";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+}
diff --git a/catalyst-gateway/bin/src/main.rs b/catalyst-gateway/bin/src/main.rs
@@ -8,6 +8,7 @@ mod db;
 mod logger;
 mod service;
 mod settings;
+mod utils;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {