feat(session): schema agreement functions and variables (#27)

Signed-off-by: Daniel Boll <danielboll.academico@gmail.com>

Author: Daniel-Boll

index.d.ts

@@ -15,6 +15,10 @@ export interface ClusterConfig {
   keyspace?: string
   auth?: Auth
   ssl?: Ssl
+  /** The driver automatically awaits schema agreement after a schema-altering query is executed. Waiting for schema agreement more than necessary is never a bug, but might slow down applications which do a lot of schema changes (e.g. a migration). For instance, in case where somebody wishes to create a keyspace and then a lot of tables in it, it makes sense only to wait after creating a keyspace and after creating all the tables rather than after every query. */
+  autoAwaitSchemaAgreement?: boolean
+  /** If the schema is not agreed upon, the driver sleeps for a duration in seconds before checking it again. The default value is 0.2 (200 milliseconds) */
+  schemaAgreementInterval?: number
 }
 export const enum Consistency {
   Any = 0,
@@ -134,6 +138,34 @@ export class ScyllaSession {
    * ```
    */
   useKeyspace(keyspaceName: string, caseSensitive?: boolean | undefined | null): Promise<void>
+  /**
+   * session.awaitSchemaAgreement returns a Promise that can be awaited as long as schema is not in an agreement.
+   * However, it won’t wait forever; ClusterConfig defines a timeout that limits the time of waiting. If the timeout elapses,
+   * the return value is an error, otherwise it is the schema_version.
+   *
+   * # Returns
+   *
+   * * `Promise<Uuid>` - schema_version
+   *
+   * # Errors
+   * * `GenericFailure` - if the timeout elapses
+   *
+   * # Example
+   * ```javascript
+   * import { Cluster } from ".";
+   *
+   * const cluster = new Cluster({ nodes: ["127.0.0.1:9042"] });
+   * const session = await cluster.connect();
+   *
+   * const schemaVersion = await session.awaitSchemaAgreement().catch(console.error);
+   * console.log(schemaVersion);
+   *
+   * const isAgreed = await session.checkSchemaAgreement().catch(console.error);
+   * console.log(isAgreed);
+   * ```
+   */
+  awaitSchemaAgreement(): Promise<Uuid>
+  checkSchemaAgreement(): Promise<boolean>
 }
 export class Uuid {
   /** Generates a random UUID v4. */

src/cluster/cluster_config/mod.rs

@@ -12,8 +12,12 @@ pub struct ClusterConfig {
   pub compression: Option<Compression>,
   pub default_execution_profile: Option<ExecutionProfile>,
 
-  // connection fields
   pub keyspace: Option<String>,
   pub auth: Option<Auth>,
   pub ssl: Option<Ssl>,
+
+  /// The driver automatically awaits schema agreement after a schema-altering query is executed. Waiting for schema agreement more than necessary is never a bug, but might slow down applications which do a lot of schema changes (e.g. a migration). For instance, in case where somebody wishes to create a keyspace and then a lot of tables in it, it makes sense only to wait after creating a keyspace and after creating all the tables rather than after every query.
+  pub auto_await_schema_agreement: Option<bool>,
+  /// If the schema is not agreed upon, the driver sleeps for a duration in seconds before checking it again. The default value is 0.2 (200 milliseconds)
+  pub schema_agreement_interval: Option<i32>,
 }

src/cluster/scylla_cluster.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use napi::Either;
 use openssl::ssl::SslContextBuilder;
 
@@ -14,6 +16,8 @@ struct ScyllaCluster {
   uri: String,
   compression: Option<Compression>,
   default_execution_profile: Option<ExecutionProfile>,
+  auto_await_schema_agreement: Option<bool>,
+  schema_agreement_interval: Option<Duration>,
 
   // connection fields
   connection: Option<ConnectionOptions>,
@@ -61,6 +65,8 @@ impl ScyllaCluster {
       keyspace,
       auth,
       ssl,
+      auto_await_schema_agreement,
+      schema_agreement_interval,
     } = cluster_config;
 
     let uri = nodes.first().expect("at least one node is required");
@@ -74,6 +80,8 @@ impl ScyllaCluster {
         auth,
         ssl,
       }),
+      auto_await_schema_agreement,
+      schema_agreement_interval: schema_agreement_interval.map(|d| Duration::from_secs(d as u64)),
     }
   }
 
@@ -152,7 +160,7 @@ impl ScyllaCluster {
         } else {
           Ok(options.ssl.clone())
         }
-      },
+      }
       (Some(Either::B(_)), Some(_)) => Err(napi::Error::new(
         napi::Status::InvalidArg,
         "Options cannot be provided twice",
@@ -163,14 +171,14 @@ impl ScyllaCluster {
         } else {
           Ok(options.ssl.clone())
         }
-      },
+      }
       (None, Some(options)) => {
         if options.ssl.is_none() {
           Ok(self.connection.as_ref().and_then(|conn| conn.ssl.clone()))
         } else {
           Ok(options.ssl.clone())
         }
-      },
+      }
       (None, None) => Ok(self.connection.as_ref().and_then(|conn| conn.ssl.clone())),
       (Some(Either::A(_)), None) => Ok(self.connection.as_ref().and_then(|conn| conn.ssl.clone())),
     };
@@ -212,6 +220,14 @@ impl ScyllaCluster {
         ));
       }
 
+      if let Some(auto_await_schema_agreement) = self.auto_await_schema_agreement {
+        builder = builder.auto_await_schema_agreement(auto_await_schema_agreement);
+      }
+
+      if let Some(schema_agreement_interval) = self.schema_agreement_interval {
+        builder = builder.schema_agreement_interval(schema_agreement_interval);
+      }
+
       builder = builder.ssl_context(Some(ssl_builder.build()));
     }
 

src/session/scylla_session.rs

@@ -146,4 +146,62 @@ impl ScyllaSession {
 
     Ok(())
   }
+
+  /// session.awaitSchemaAgreement returns a Promise that can be awaited as long as schema is not in an agreement.
+  /// However, it won’t wait forever; ClusterConfig defines a timeout that limits the time of waiting. If the timeout elapses,
+  /// the return value is an error, otherwise it is the schema_version.
+  ///
+  /// # Returns
+  ///
+  /// * `Promise<Uuid>` - schema_version
+  ///
+  /// # Errors
+  /// * `GenericFailure` - if the timeout elapses
+  ///
+  /// # Example
+  /// ```javascript
+  /// import { Cluster } from ".";
+  ///
+  /// const cluster = new Cluster({ nodes: ["127.0.0.1:9042"] });
+  /// const session = await cluster.connect();
+  ///
+  /// const schemaVersion = await session.awaitSchemaAgreement().catch(console.error);
+  /// console.log(schemaVersion);
+  ///
+  /// const isAgreed = await session.checkSchemaAgreement().catch(console.error);
+  /// console.log(isAgreed);
+  /// ```
+  #[napi]
+  pub async fn await_schema_agreement(&self) -> napi::Result<Uuid> {
+    Ok(
+      self
+        .session
+        .await_schema_agreement()
+        .await
+        .map_err(|e| {
+          napi::Error::new(
+            napi::Status::GenericFailure,
+            format!("Something went wrong with your schema agreement. - {e}"),
+          )
+        })?
+        .into(),
+    )
+  }
+
+  #[napi]
+  pub async fn check_schema_agreement(&self) -> napi::Result<bool> {
+    Ok(
+      self
+        .session
+        .check_schema_agreement()
+        .await
+        .map_err(|e| {
+          napi::Error::new(
+            napi::Status::GenericFailure,
+            format!("Something went wrong with your schema agreement. - {e}"),
+          )
+        })?
+        .is_some(),
+    )
+  }
 }