Working

Move test utils from the Common crate to sdk-core crate

Author: mjameswh

arch_docs/workflow_task_chunking.md

@@ -4,20 +4,20 @@ One source of complexity in Core is the chunking of history into "logical" Workf
 
 Workflow tasks (WFTs) always take the following form in event history:
 
-* \[Preceding Events\] (optional)
-* WFT Scheduled
-* WFT Started
-* WFT Completed
-* \[Commands\] (optional)
+- \[Preceding Events\] (optional)
+- WFT Scheduled
+- WFT Started
+- WFT Completed
+- \[Commands\] (optional)
 
 In the typical case, the "logical" WFT consists of all the commands from the last workflow task,
 any events generated in the interrim, and the scheduled/started preamble. So:
 
-* WFT Completed
-* \[Commands\] (optional)
-* \[Events\] (optional)
-* WFT Scheduled
-* WFT Started
+- WFT Completed
+- \[Commands\] (optional)
+- \[Events\] (optional)
+- WFT Scheduled
+- WFT Started
 
 Commands and events are both "optional" in the sense that:
 
@@ -32,8 +32,22 @@ There may be no events for more nuanced reasons:
    rather as a "protocol message" attached to the task.
 3. Server can forcibly generate a new WFT with some obscure APIs
 
-Core does not consider such empty WFT sequences as worthy of waking lang (on replay - as a new
-task, they always will), since nothing meaningful has happened. Thus, they are grouped together
+## Handling of empty WFTs (March 2026)
+
+Until now, Core would try to avoid waking lang when replaying empty WFT sequences (i.e. a WFT
+Completed immediately followed by a WFT Scheduled and WFT Started, with no commands and no events
+in-between), since they would presumably be no-ops and therefore waste execution resources. Instead,
+the second WFT (and potentially subsequent WFTs) would be collapsed into the first, resulting in a
+single WFT that has the inbound events and start time of the first WFT, but the resulting commands
+of the last WFT in the collapsed sequence.
+
+It was found that there are some particular edge cases where WFT collapsing might result in
+incorrect replay behavior. One particular example of this is the case where an update request would
+be sent after an empty WFT. This led to the development of some heuristics to avoid incorrect
+chunking in presence of known problematic patterns. Unfortunately, it has been found that these
+heuristics may
+
+Thus, they are grouped together task, they always will), since nothing meaningful has happened. Thus, they are grouped together
 as part of a "logical" WFT with the last WFT that had any real work in it.
 
 ## Possible issues as of this writing (5/25)
@@ -44,8 +58,8 @@ NDE.
 
 ### Possible solutions
 
-* Core can attach a flag on WFT completes in order to be explicit that that WFT may be skipped on
+- Core can attach a flag on WFT completes in order to be explicit that that WFT may be skipped on
   replay. IE: During WFT heartbeating for LAs.
-* We could legislate that server should never send empty WFTs. Seemingly the only case of this
+- We could legislate that server should never send empty WFTs. Seemingly the only case of this
   is
   the [obscure api](https://github.com/temporalio/temporal/blob/d189737aa2ed1b07c221abb9fbdd28ecf68f0492/proto/internal/temporal/server/api/adminservice/v1/service.proto#L151)

crates/common/Cargo.toml

@@ -14,7 +14,6 @@ exclude = ["protos/*/.github/*"]
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-history_builders = ["rand"]
 otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp"]
 prometheus = [
   "dep:prometheus",
@@ -25,7 +24,6 @@ prometheus = [
 ]
 envconfig = ["dep:toml", "dep:dirs"]
 serde_serialize = []
-test-utilities = ["history_builders"]
 core-based-sdk = [
   "prometheus",
   "envconfig",
@@ -41,6 +39,7 @@ bon = { workspace = true }
 crc32fast = "1"
 dirs = { version = "6.0", optional = true }
 derive_more = { workspace = true }
+enum-iterator = "2"
 erased-serde = "0.4"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-channel = { version = "0.3", default-features = false, features = [
@@ -74,7 +73,6 @@ prometheus = { version = "0.14", optional = true, default-features = false }
 prost = { workspace = true }
 prost-wkt = "0.7"
 prost-types = { workspace = true }
-rand = { version = "0.10", optional = true }
 ringbuf = { version = "0.4", optional = true }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { workspace = true }

crates/common/src/protos/mod.rs

@@ -6,26 +6,10 @@ pub mod constants;
 /// Utility functions for working with protobuf types.
 pub mod utilities;
 
-#[cfg(feature = "test-utilities")]
-/// Pre-built test histories for common workflow patterns.
-pub mod canned_histories;
-#[cfg(feature = "history_builders")]
-mod history_builder;
-#[cfg(feature = "history_builders")]
-mod history_info;
 mod task_token;
-#[cfg(feature = "test-utilities")]
-pub mod test_utils;
 
 use std::time::Duration;
 
-#[cfg(feature = "history_builders")]
-pub use history_builder::{
-    DEFAULT_ACTIVITY_TYPE, DEFAULT_WORKFLOW_TYPE, TestHistoryBuilder, default_act_sched,
-    default_wes_attribs,
-};
-#[cfg(feature = "history_builders")]
-pub use history_info::HistoryInfo;
 pub use task_token::TaskToken;
 
 /// Payload metadata key that identifies the encoding format.

crates/sdk-core/Cargo.toml

@@ -110,7 +110,7 @@ zip = { version = "8.4", optional = true, default-features = false, features = [
 [dependencies.temporalio-common]
 path = "../common"
 version = "0.2"
-features = ["core-based-sdk", "history_builders", "test-utilities"]
+features = ["core-based-sdk"]
 
 [dependencies.temporalio-client]
 path = "../client"

crates/sdk-core/benches/workflow_replay_bench.rs

@@ -12,13 +12,13 @@ use std::{
     thread,
     time::Duration,
 };
-use temporalio_common::{
-    protos::{DEFAULT_WORKFLOW_TYPE, canned_histories},
-    telemetry::metrics::{MetricKeyValue, MetricParameters, NewAttributes},
-};
+use temporalio_common::telemetry::metrics::{MetricKeyValue, MetricParameters, NewAttributes};
 use temporalio_macros::{workflow, workflow_methods};
 use temporalio_sdk::{SyncWorkflowContext, WorkflowContext, WorkflowResult};
-use temporalio_sdk_core::{CoreRuntime, replay::HistoryForReplay};
+use temporalio_sdk_core::{
+    CoreRuntime,
+    replay::{DEFAULT_WORKFLOW_TYPE, HistoryForReplay, canned_histories},
+};
 
 pub fn criterion_benchmark(c: &mut Criterion) {
     let tokio_runtime = tokio::runtime::Builder::new_current_thread()

crates/sdk-core/src/core_tests/activity_tasks.rs

@@ -1,10 +1,11 @@
 use crate::{
     ActivityHeartbeat, CompleteActivityError, Worker, advance_fut, job_assert, prost_dur,
+    replay::{TestHistoryBuilder, canned_histories},
     test_help::{
         FakeWfResponses, MockPollCfg, MockWorkerInputs, MocksHolder, QueueResponse, WorkerExt,
         WorkflowCachingPolicy, build_fake_worker, build_mock_pollers, build_multihist_mock_sg,
         fanout_tasks, gen_assert_and_reply, mock_manual_poller, mock_poller, mock_worker,
-        poll_and_reply, single_hist_mock_sg, test_worker_cfg,
+        poll_and_reply, single_hist_mock_sg, start_timer_cmd, test_worker_cfg,
     },
     worker::{
         PollerBehavior,
@@ -26,7 +27,6 @@ use std::{
 };
 use temporalio_common::{
     protos::{
-        TestHistoryBuilder, canned_histories,
         coresdk::{
             ActivityTaskCompletion,
             activity_result::{
@@ -52,7 +52,6 @@ use temporalio_common::{
                 RespondActivityTaskFailedResponse, RespondWorkflowTaskCompletedResponse,
             },
         },
-        test_utils::start_timer_cmd,
     },
     worker::WorkerTaskTypes,
 };

crates/sdk-core/src/core_tests/mod.rs

@@ -8,6 +8,7 @@ mod workflow_tasks;
 
 use crate::{
     PollError, Worker,
+    replay::{TestHistoryBuilder, canned_histories},
     test_help::{
         MockPollCfg, build_mock_pollers, mock_worker, single_hist_mock_sg, test_worker_cfg,
     },
@@ -19,7 +20,6 @@ use crate::{
 use futures_util::FutureExt;
 use std::{sync::LazyLock, time::Duration};
 use temporalio_common::protos::{
-    TestHistoryBuilder, canned_histories,
     coresdk::{
         workflow_activation::{WorkflowActivationJob, workflow_activation_job},
         workflow_completion::WorkflowActivationCompletion,

crates/sdk-core/src/core_tests/queries.rs

@@ -1,7 +1,9 @@
 use crate::{
+    replay::{TestHistoryBuilder, canned_histories},
     test_help::{
         MockPollCfg, MocksHolder, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers,
-        hist_to_poll_resp, mock_worker, single_hist_mock_sg,
+        hist_to_poll_resp, mock_worker, query_ok, schedule_activity_cmd, single_hist_mock_sg,
+        start_timer_cmd,
     },
     worker::{
         LEGACY_QUERY_ID, WorkerVersioningStrategy,
@@ -19,7 +21,6 @@ use std::{
 };
 use temporalio_client::MESSAGE_TOO_LARGE_KEY;
 use temporalio_common::protos::{
-    TestHistoryBuilder, canned_histories,
     coresdk::{
         workflow_activation::{
             WorkflowActivationJob, remove_from_cache::EvictionReason, workflow_activation_job,
@@ -40,7 +41,6 @@ use temporalio_common::protos::{
             GetWorkflowExecutionHistoryResponse, RespondWorkflowTaskCompletedResponse,
         },
     },
-    test_utils::{query_ok, schedule_activity_cmd, start_timer_cmd},
 };
 
 #[rstest::rstest]

crates/sdk-core/src/core_tests/replay_flag.rs

@@ -1,16 +1,18 @@
 use crate::{
-    test_help::{MockPollCfg, ResponseType, build_mock_pollers, hist_to_poll_resp, mock_worker},
+    replay::{TestHistoryBuilder, canned_histories},
+    test_help::{
+        MockPollCfg, ResponseType, build_mock_pollers, hist_to_poll_resp, mock_worker, query_ok,
+        start_timer_cmd,
+    },
     worker::{LEGACY_QUERY_ID, client::mocks::mock_worker_client},
 };
 use std::{collections::VecDeque, time::Duration};
 use temporalio_common::protos::{
-    TestHistoryBuilder, canned_histories,
     coresdk::{
         workflow_activation::{WorkflowActivationJob, workflow_activation_job},
         workflow_completion::WorkflowActivationCompletion,
     },
     temporal::api::{enums::v1::EventType, query::v1::WorkflowQuery},
-    test_utils::{query_ok, start_timer_cmd},
 };
 
 #[tokio::test]

crates/sdk-core/src/core_tests/updates.rs

@@ -1,13 +1,13 @@
 use crate::{
     prost_dur,
+    replay::{DEFAULT_ACTIVITY_TYPE, TestHistoryBuilder},
     test_help::{
         MockPollCfg, PollWFTRespExt, ResponseType, WorkerTestHelpers, build_mock_pollers,
         hist_to_poll_resp, mock_worker,
     },
     worker::client::mocks::mock_worker_client,
 };
 use temporalio_common::protos::{
-    DEFAULT_ACTIVITY_TYPE, TestHistoryBuilder,
     coresdk::{
         workflow_activation::{WorkflowActivationJob, workflow_activation_job},
         workflow_commands::{