Remove serializable WF inputs feature (#701)

Author: Sushisource

.cargo/config.toml

@@ -1,8 +1,6 @@
 [alias]
 integ-test = ["run", "--package", "temporal-sdk-core", "--example", "integ_runner", "--"]
-wf-input-replay = ["run", "--package", "temporal-sdk-core", "--features", "save_wf_inputs",
-                   "--example", "wf_input_replay", "--"]
 lint = ["clippy", "--workspace", "--examples", "--all-features",
-        "--test", "integ_tests", "--test", "heavy_tests", "--", "--D", "warnings"]
+    "--test", "integ_tests", "--test", "heavy_tests", "--", "--D", "warnings"]
 test-lint = ["clippy", "--all", "--all-features", "--examples", "--workspace",
-             "--tests", "--", "--D", "warnings"]
+    "--tests", "--", "--D", "warnings"]

.github/workflows/heavy.yml

@@ -30,4 +30,4 @@ jobs:
       - uses: actions-rs/cargo@v1
         with:
           command: integ-test
-          args: -c "--release" -c "--features=save_wf_inputs" -t heavy_tests -- --test-threads 1
+          args: -c "--release" -t heavy_tests -- --test-threads 1

README.md

@@ -34,7 +34,7 @@ This repo is composed of multiple crates:
 - temporal-sdk-core `./core` - The Core implementation
 - temporal-sdk `./sdk` - A (currently prototype) Rust SDK built on top of Core. Used for testing.
 - rustfsm `./fsm` - Implements a procedural macro used by core for defining state machines
-    (contains subcrates). It is temporal agnostic.
+  (contains subcrates). It is temporal agnostic.
 
 Visualized (dev dependencies are in blue):
 
@@ -51,7 +51,7 @@ You can build and test the project using cargo:
 Run integ tests with `cargo integ-test`. By default it will start an ephemeral server. You can also
 use an already-running server by passing `-s external`.
 
-Run load tests with `cargo test --features=save_wf_inputs --test heavy_tests`.
+Run load tests with `cargo test --test heavy_tests`.
 
 ## Formatting
 
@@ -86,8 +86,8 @@ equivalent.
 
 ## Proto files
 
-This repo uses a subtree for upstream protobuf files. The path `sdk-core-protos/protos/api_upstream` is a
-subtree. To update it, use:
+This repo uses a subtree for upstream protobuf files. The path `sdk-core-protos/protos/api_upstream`
+is a subtree. To update it, use:
 
 `git pull --squash --rebase=false -s subtree ssh://git@github.com/temporalio/api.git master --allow-unrelated-histories`
 

core-api/Cargo.toml

@@ -21,10 +21,8 @@ derive_builder = { workspace = true }
 derive_more = { workspace = true }
 opentelemetry = { workspace = true, optional = true }
 prost-types = { workspace = true }
-serde = { version = "1.0", default_features = false, features = ["derive"] }
 serde_json = "1.0"
 thiserror = "1.0"
-tokio = "1.24"
 tonic = { workspace = true }
 tracing-core = "0.1"
 url = "2.3"

core-api/src/worker.rs

@@ -1,11 +1,10 @@
 use std::time::Duration;
-use tokio::sync::mpsc::UnboundedSender;
 
 const MAX_OUTSTANDING_WFT_DEFAULT: usize = 100;
 const MAX_CONCURRENT_WFT_POLLS_DEFAULT: usize = 5;
 
 /// Defines per-worker configuration options
-#[derive(Debug, Clone, derive_builder::Builder, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, derive_builder::Builder)]
 #[builder(setter(into), build_fn(validate = "Self::validate"))]
 #[non_exhaustive]
 pub struct WorkerConfig {
@@ -115,14 +114,6 @@ pub struct WorkerConfig {
     #[builder(default = "5")]
     pub fetching_concurrency: usize,
 
-    // TODO: Move this out - dependency on tokio should not exist just for this
-    /// If set, and the `save_wf_inputs` feature is enabled in core, will be sent a serialized
-    /// instance of every input to workflow state in order. This is for testing purposes, SDK
-    /// implementations never need to care about it.
-    #[builder(default)]
-    #[serde(skip)]
-    pub wf_state_inputs: Option<UnboundedSender<Vec<u8>>>,
-
     /// If set, core will issue cancels for all outstanding activities after shutdown has been
     /// initiated and this amount of time has elapsed.
     #[builder(default)]

core/Cargo.toml

@@ -14,10 +14,6 @@ categories = ["development-tools"]
 
 [features]
 default = ["otel"]
-# Do not enable this feature when building production SDKs. If we ever want a user in the field to
-# record WF input data, we can build them a custom SDK or they can build - it adds significant extra
-# code size in the form of [de]serializers.
-save_wf_inputs = ["rmp-serde", "temporal-sdk-core-protos/serde_serialize"]
 otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp",
     "dep:opentelemetry-prometheus", "dep:hyper", "dep:hyper-util", "dep:http-body-util"]
 tokio-console = ["console-subscriber"]
@@ -59,7 +55,6 @@ prost-types = { version = "0.5", package = "prost-wkt-types" }
 rand = "0.8.3"
 reqwest = { version = "0.11", features = ["json", "stream", "rustls-tls", "tokio-rustls"], default-features = false, optional = true }
 ringbuf = "0.3"
-rmp-serde = { version = "1.1", optional = true }
 serde = "1.0"
 serde_json = "1.0"
 siphasher = "1.0"
@@ -115,7 +110,6 @@ test = false
 name = "heavy_tests"
 path = "../tests/heavy_tests.rs"
 test = false
-required-features = ["save_wf_inputs"]
 
 [[bench]]
 name = "workflow_replay"
@@ -126,8 +120,3 @@ harness = false
 [[example]]
 name = "integ_runner"
 path = "../tests/runner.rs"
-
-[[example]]
-name = "wf_input_replay"
-path = "../tests/wf_input_replay.rs"
-required-features = ["save_wf_inputs"]

core/src/abstractions.rs

@@ -222,18 +222,6 @@ impl OwnedMeteredSemPermit {
 
 #[derive(Debug)]
 pub(crate) struct UsedMeteredSemPermit(OwnedMeteredSemPermit);
-impl UsedMeteredSemPermit {
-    #[cfg(feature = "save_wf_inputs")]
-    pub(crate) fn fake_deserialized() -> Self {
-        let sem = Arc::new(Semaphore::new(1));
-        let inner = sem.try_acquire_owned().unwrap();
-        Self(OwnedMeteredSemPermit {
-            inner,
-            unused_claimants: None,
-            record_fn: Box::new(|_| {}),
-        })
-    }
-}
 
 macro_rules! dbg_panic {
   ($($arg:tt)*) => {

core/src/lib.rs

@@ -38,8 +38,6 @@ pub use temporal_sdk_core_api as api;
 pub use temporal_sdk_core_protos as protos;
 pub use temporal_sdk_core_protos::TaskToken;
 pub use url::Url;
-#[cfg(feature = "save_wf_inputs")]
-pub use worker::replay_wf_state_inputs;
 pub use worker::{Worker, WorkerConfig, WorkerConfigBuilder};
 
 use crate::{

core/src/protosext/protocol_messages.rs

@@ -8,10 +8,6 @@ use temporal_sdk_core_protos::temporal::api::{
 
 /// A decoded & verified of a [Message] that came with a WFT.
 #[derive(Debug, Clone, PartialEq)]
-#[cfg_attr(
-    feature = "save_wf_inputs",
-    derive(serde::Serialize, serde::Deserialize)
-)]
 pub struct IncomingProtocolMessage {
     pub id: String,
     pub protocol_instance_id: String,
@@ -44,10 +40,6 @@ impl TryFrom<Message> for IncomingProtocolMessage {
 /// All the protocol [Message] bodies Core understands that might come to us when receiving a new
 /// WFT.
 #[derive(Debug, Clone, PartialEq)]
-#[cfg_attr(
-    feature = "save_wf_inputs",
-    derive(serde::Serialize, serde::Deserialize)
-)]
 pub enum IncomingProtocolMessageBody {
     UpdateRequest(UpdateRequest),
 }
@@ -71,10 +63,6 @@ impl TryFrom<Option<prost_types::Any>> for IncomingProtocolMessageBody {
 }
 
 #[derive(Debug, Clone, PartialEq)]
-#[cfg_attr(
-    feature = "save_wf_inputs",
-    derive(serde::Serialize, serde::Deserialize)
-)]
 pub struct UpdateRequest {
     pub name: String,
     pub headers: HashMap<String, Payload>,

core/src/worker/activities/local_activities.rs

@@ -55,10 +55,6 @@ pub(crate) struct LocalInFlightActInfo {
 }
 
 #[derive(Debug, Clone)]
-#[cfg_attr(
-    feature = "save_wf_inputs",
-    derive(serde::Serialize, serde::Deserialize)
-)]
 pub(crate) enum LocalActivityExecutionResult {
     Completed(Success),
     Failed(ActFail),
@@ -104,10 +100,6 @@ impl LocalActivityExecutionResult {
 }
 
 #[derive(Debug, Clone)]
-#[cfg_attr(
-    feature = "save_wf_inputs",
-    derive(serde::Serialize, serde::Deserialize)
-)]
 pub(crate) struct LocalActivityResolution {
     pub seq: u32,
     pub result: LocalActivityExecutionResult,