init

Author: THardy98

crates/sdk-core-c-bridge/Cargo.toml

@@ -55,4 +55,5 @@ thiserror = { workspace = true }
 cbindgen = { version = "0.29", default-features = false }
 
 [features]
+antithesis_assertions = ["temporalio-sdk-core/antithesis_assertions"]
 xz2-static = ["xz2/static"]

crates/sdk-core/Cargo.toml

@@ -27,9 +27,11 @@ tokio-console = ["console-subscriber"]
 ephemeral-server = ["dep:flate2", "dep:reqwest", "dep:tar", "dep:zip"]
 debug-plugin = ["dep:reqwest"]
 test-utilities = ["dep:assert_matches", "dep:bimap"]
+antithesis_assertions = ["dep:antithesis_sdk"]
 
 [dependencies]
 anyhow = "1.0"
+antithesis_sdk = { version = "0.2.1", optional = true, default-features = false, features = ["full"] }
 assert_matches = { version = "1.5", optional = true }
 bimap = { version = "0.6.3", optional = true }
 async-trait = "0.1"

crates/sdk-core/src/abstractions.rs

@@ -422,11 +422,30 @@ impl<SK: SlotKind> OwnedMeteredSemPermit<SK> {
 #[derive(Debug)]
 pub(crate) struct UsedMeteredSemPermit<SK: SlotKind>(#[allow(dead_code)] OwnedMeteredSemPermit<SK>);
 
+#[cfg(feature = "antithesis_assertions")]
 macro_rules! dbg_panic {
-  ($($arg:tt)*) => {
-      error!($($arg)*);
-      debug_assert!(false, $($arg)*);
-  };
+    ($($arg:tt)*) => {{
+        let message = format!($($arg)*);
+        error!("{}", message);
+        crate::antithesis::assert_always_failure(
+            "dbg_panic invariant triggered",
+            ::serde_json::json!({
+                "message": message.clone(),
+                "file": file!(),
+                "line": line!(),
+                "module": module_path!(),
+            }),
+        );
+        debug_assert!(false, "{}", message);
+    }};
+}
+
+#[cfg(not(feature = "antithesis_assertions"))]
+macro_rules! dbg_panic {
+    ($($arg:tt)*) => {{
+        error!($($arg)*);
+        debug_assert!(false, $($arg)*);
+    }};
 }
 pub(crate) use dbg_panic;
 

crates/sdk-core/src/antithesis.rs

@@ -0,0 +1,25 @@
+#[cfg(feature = "antithesis_assertions")]
+use std::sync::OnceLock;
+
+#[cfg(feature = "antithesis_assertions")]
+use serde_json::Value;
+
+/// Ensure Antithesis is initialized exactly once.
+#[cfg(feature = "antithesis_assertions")]
+pub(crate) fn ensure_init() {
+    static INIT: OnceLock<()> = OnceLock::new();
+    INIT.get_or_init(|| {
+        ::antithesis_sdk::antithesis_init();
+    });
+}
+
+#[cfg(not(feature = "antithesis_assertions"))]
+#[inline]
+pub(crate) fn ensure_init() {}
+
+/// Record an always-true assertion failure to Antithesis when an invariant is
+/// violated.
+#[cfg(feature = "antithesis_assertions")]
+pub(crate) fn assert_always_failure(name: &str, metadata: Value) {
+    ::antithesis_sdk::assert_always!(false, name, &metadata);
+}

crates/sdk-core/src/lib.rs

@@ -2,7 +2,12 @@
 #![allow(clippy::upper_case_acronyms)]
 
 //! This crate provides a basis for creating new Temporal SDKs without completely starting from
-//! scratch
+//! scratch.
+//!
+//! ## Optional features
+//! - `antithesis_assertions`: Enables integration with Antithesis' Rust SDK. When active, core
+//!   initialization calls `antithesis_sdk::antithesis_init` and critical invariants emit
+//!   Antithesis assertions to aid fuzzing campaigns.
 
 #[cfg(test)]
 #[macro_use]
@@ -12,6 +17,7 @@ extern crate tracing;
 extern crate core;
 
 mod abstractions;
+mod antithesis;
 #[cfg(feature = "debug-plugin")]
 pub mod debug_client;
 #[cfg(feature = "ephemeral-server")]
@@ -86,8 +92,18 @@ pub fn init_worker<CT>(
 where
     CT: Into<sealed::AnyClient>,
 {
+    crate::antithesis::ensure_init();
     let namespace = worker_config.namespace.clone();
     if namespace.is_empty() {
+        #[cfg(feature = "antithesis_assertions")]
+        crate::antithesis::assert_always_failure(
+            "worker namespace must not be empty",
+            ::serde_json::json!({
+                "namespace": namespace.clone(),
+                "task_queue": worker_config.task_queue.clone(),
+                "has_identity_override": worker_config.client_identity_override.is_some(),
+            }),
+        );
         bail!("Worker namespace cannot be empty");
     }
 
@@ -103,6 +119,14 @@ where
     let sticky_q = sticky_q_name_for_worker(&client_ident, worker_config.max_cached_workflows);
 
     if client_ident.is_empty() {
+        #[cfg(feature = "antithesis_assertions")]
+        crate::antithesis::assert_always_failure(
+            "client identity must not be empty",
+            ::serde_json::json!({
+                "namespace": namespace.clone(),
+                "task_queue": worker_config.task_queue.clone(),
+            }),
+        );
         bail!("Client identity cannot be empty. Either lang or user should be setting this value");
     }
 
@@ -132,6 +156,7 @@ pub fn init_replay_worker<I>(rwi: ReplayWorkerInput<I>) -> Result<Worker, anyhow
 where
     I: Stream<Item = HistoryForReplay> + Send + 'static,
 {
+    crate::antithesis::ensure_init();
     info!(
         task_queue = rwi.config.task_queue.as_str(),
         "Registering replay worker"
@@ -290,6 +315,7 @@ impl CoreRuntime {
     where
         F: Fn() + Send + Sync + 'static,
     {
+        crate::antithesis::ensure_init();
         let telemetry = telemetry_init(runtime_options.telemetry_options)?;
         let subscriber = telemetry.trace_subscriber();
         let runtime = tokio_builder
@@ -317,6 +343,7 @@ impl CoreRuntime {
     /// # Panics
     /// If there is no currently active Tokio runtime
     pub fn new_assume_tokio(runtime_options: RuntimeOptions) -> Result<Self, anyhow::Error> {
+        crate::antithesis::ensure_init();
         let telemetry = telemetry_init(runtime_options.telemetry_options)?;
         Ok(Self::new_assume_tokio_initialized_telem(
             telemetry,
@@ -333,6 +360,7 @@ impl CoreRuntime {
         telemetry: TelemetryInstance,
         heartbeat_interval: Option<Duration>,
     ) -> Self {
+        crate::antithesis::ensure_init();
         let runtime_handle = tokio::runtime::Handle::current();
         if let Some(sub) = telemetry.trace_subscriber() {
             set_trace_subscriber_for_current_thread(sub);

crates/sdk/Cargo.toml

@@ -42,5 +42,9 @@ version = "0.1"
 path = "../client"
 version = "0.1"
 
+[features]
+default = []
+antithesis_assertions = ["temporalio-sdk-core/antithesis_assertions"]
+
 [lints]
 workspace = true