Cancellation fundamentals

.gitignore

@@ -10,3 +10,4 @@ Cargo.lock
 # Ignore generated protobuf files
 src/protos/*.rs
 !src/protos/mod.rs
+/tarpaulin-report.html

Cargo.toml

@@ -18,14 +18,16 @@ displaydoc = "0.1"
 env_logger = "0.8"
 futures = "0.3"
 log = "0.4"
-opentelemetry-jaeger = "0.10"
-opentelemetry = "0.11.2"
+once_cell = "1.5"
+opentelemetry-jaeger = "0.11"
+opentelemetry = "0.12"
 prost = "0.7"
 prost-types = "0.7"
+slotmap = "1.0"
 thiserror = "1.0"
 tokio = { version = "1.1", features = ["rt", "rt-multi-thread"] }
 tracing = { version = "0.1", features = ["log"] }
-tracing-opentelemetry = "0.10"
+tracing-opentelemetry = "0.11"
 tracing-subscriber = "0.2"
 url = "2.2"
 rand = "0.8.3"

fsm/Cargo.toml

@@ -6,7 +6,5 @@ edition = "2018"
 license-file = "LICENSE.txt"
 
 [dependencies]
-thiserror = "1.0"
-derive_more = "0.99"
 state_machine_procmacro = { path = "state_machine_procmacro" }
 state_machine_trait = { path = "state_machine_trait" }

fsm/state_machine_procmacro/src/lib.rs

@@ -329,7 +329,9 @@ impl StateMachineDefinition {
             .flat_map(|t| vec![t.from.clone(), t.to.clone()])
             .collect();
         let state_variants = states.iter().map(|s| {
+            let statestr = s.to_string();
             quote! {
+                #[display(fmt=#statestr)]
                 #s(#s)
             }
         });
@@ -349,7 +351,7 @@ impl StateMachineDefinition {
             }
         };
         let states_enum = quote! {
-            #[derive(::derive_more::From, Clone)]
+            #[derive(::derive_more::From, Clone, ::derive_more::Display)]
             #visibility enum #state_enum_name {
                 #(#state_variants),*
             }

protos/local/core_interface.proto

@@ -63,11 +63,13 @@ message WFActivationJob {
         StartWorkflowTaskAttributes start_workflow = 1;
         // A timer has fired, allowing whatever was waiting on it (if anything) to proceed
         TimerFiredTaskAttributes timer_fired = 2;
-        // Workflow was reset. The randomness seed has to be updated.
-        RandomSeedUpdatedAttributes random_seed_updated = 3;
+        // A timer was canceled
+        TimerCanceledTaskAttributes timer_canceled = 3;
+        // Workflow was reset. The randomness seed must be updated.
+        RandomSeedUpdatedAttributes random_seed_updated = 4;
 
-        QueryWorkflowJob query_workflow = 4;
-        CancelWorkflowTaskAttributes cancel_workflow = 5;
+        QueryWorkflowJob query_workflow = 5;
+        CancelWorkflowTaskAttributes cancel_workflow = 6;
     }
 }
 
@@ -94,6 +96,10 @@ message TimerFiredTaskAttributes {
     string timer_id = 1;
 }
 
+message TimerCanceledTaskAttributes {
+    string timer_id = 1;
+}
+
 message RandomSeedUpdatedAttributes {
     uint64 randomness_seed = 1;
 }

src/core_tracing.rs

@@ -0,0 +1,24 @@
+use once_cell::sync::OnceCell;
+use opentelemetry_jaeger::Uninstall;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
+
+static TRACING_INIT: OnceCell<Uninstall> = OnceCell::new();
+const TRACING_ENABLE_ENV_VAR: &str = "TEMPORAL_CORE_TRACING";
+
+pub(crate) fn tracing_init() {
+    let _ = env_logger::try_init();
+    if std::env::var(TRACING_ENABLE_ENV_VAR).is_ok() {
+        TRACING_INIT.get_or_init(|| {
+            let (tracer, uninstall) = opentelemetry_jaeger::new_pipeline()
+                .with_service_name("coresdk")
+                .install()
+                .unwrap();
+            let opentelemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+            tracing_subscriber::registry()
+                .with(opentelemetry)
+                .try_init()
+                .unwrap();
+            uninstall
+        });
+    }
+}