[Jetty] Merged Brook #209, which drops the cache-off eviction reason.

CHANGELOG.md

@@ -38,8 +38,8 @@ relevant information.
 
 ### Added
 * `Connection::raw_unary` makes a unary gRPC call by method path, with the request and the
-  response as encoded protobuf. It keeps the connection's TLS, API key, headers, interceptors,
-  metrics and retry policy, so you can call a method the client does not list.
+  response as encoded protobuf. The call uses the connection's TLS, API key, headers,
+  interceptors, metrics and retry policy. Use it to call a method that the client doesn't list.
 
 ### Breaking Changes
 * Workflow replay now reports nondeterminism when a scheduled Nexus operation's service or operation

crates/client/src/lib.rs

@@ -991,7 +991,8 @@ pub struct TemporalServiceClient {
     cloud_svc_client: Box<dyn CloudService>,
     test_svc_client: Box<dyn TestService>,
     health_svc_client: Box<dyn HealthService>,
-    /// Calls by method path on the same service. A client built from mock services has none.
+    /// Makes calls by method path on the same gRPC service. It's `None` for a client built from
+    /// mock services.
     raw_unary: Option<Box<dyn raw::RawUnaryService>>,
 }
 

crates/client/src/raw.rs

@@ -1,6 +1,5 @@
-//! Unary gRPC calls by method path, with encoded protobuf in and out. A crate that generates its
-//! own messages can call a method this client does not list, and keep the connection's TLS, API
-//! key, headers, interceptors, metrics and retries.
+//! Unary gRPC calls by method path, with encoded protobuf messages. A crate that generates its own
+//! messages can use them to call a method that this client doesn't list.
 
 use crate::{Connection, grpc::RawGrpcCaller};
 use bytes::{Buf, BufMut, Bytes};
@@ -93,11 +92,11 @@ impl Decoder for BytesCodec {
 impl Connection {
     /// Makes a unary gRPC call to the method at `path`, such as
     /// `/temporal.api.workflowservice.v1.WorkflowService/GetSystemInfo`, with `request` as the
-    /// encoded request message. Answers with the encoded response message.
+    /// encoded request message. Returns the encoded response message.
     ///
-    /// The call goes through the same stack as the listed methods, with the connection's TLS,
-    /// API key, headers, interceptors, metrics and retry policy. `path` also names the call for
-    /// the retry policy. Request extensions and metadata work as on a listed method. Set the
+    /// The call uses the same stack as the listed methods, with the connection's TLS, API key,
+    /// headers, interceptors, metrics and retry policy. The retry policy uses `path` as the call
+    /// name. Request extensions and metadata work as they do on a listed method. Set the
     /// `temporal-namespace` metadata to route by namespace, as the listed methods do.
     pub async fn raw_unary(
         &self,
@@ -128,11 +127,11 @@ mod tests {
     use tonic::{Code, Request, Status};
     use url::Url;
 
-    /// What the server saw of each call that the system info read did not make.
+    /// The requests that the server got, other than the `GetSystemInfo` call of the connect.
     type Seen = Arc<Mutex<Vec<GrpcRequest>>>;
 
-    /// A connection whose server answers each call with `answer`, after the connect's own
-    /// system info read.
+    /// A connection to a server that answers each call with `answer`, other than the
+    /// `GetSystemInfo` call of the connect.
     async fn connection(
         answer: impl Fn(&GrpcRequest, usize) -> Result<Vec<u8>, Status> + Send + Sync + 'static,
     ) -> (Connection, Seen) {

crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto

@@ -65,23 +65,23 @@ message Start {
     bool is_local = 17;
     // Run ID of this activity execution. Only set for standalone activities.
     string run_id = 19;
-    // For a local activity, the workflow task in which Core dispatched this execution. Retries
-    // that Core runs without a new task keep it. Unset for other activities.
+    // For a local activity, the workflow task in which Core dispatched this execution. The retries
+    // that Core runs without a new workflow task have the same value. Unset for other activities.
     WorkflowTaskAttempt workflow_task = 20;
 }
 
-// One attempt of a workflow task, with the values from the poll response that started it.
+// One attempt of a workflow task, from the poll response and the history of that attempt.
 message WorkflowTaskAttempt {
-    // The id of the task's WorkflowTaskStarted event.
+    // The event id of the task's WorkflowTaskStarted event.
     int64 started_event_id = 1;
     uint32 attempt = 2;
     // When the server started the task.
     google.protobuf.Timestamp started_time = 3;
     // The started event id of the run's last completed workflow task, from the poll response.
     // 0 when no task of the run has completed.
     int64 previous_started_event_id = 4;
-    // The started event ids of the run's failed and timed out workflow tasks in history, in event
-    // order. It holds the last 100 only, to keep each start small.
+    // The started event ids of the run's failed and timed out workflow tasks, in event order. Core
+    // sets only the last 100, so that each start stays small.
     repeated int64 failed_tasks = 5;
 }
 

crates/sdk-core/CHANGELOG.md

@@ -35,8 +35,10 @@ relevant information.
 
 ### Added
 * A Local Activity's `ActivityTask.Start` has `workflow_task` (field 20), with the started event
-  id, attempt and start time of the Workflow Task in which Core dispatched it. Retries that Core
-  runs on its own keep the value.
+  id, attempt and start time of the Workflow Task in which Core dispatched it. `workflow_task`
+  also has `previous_started_event_id`, the started event id of the run's last completed Workflow
+  Task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out
+  Workflow Tasks. The retries that Core runs without a new Workflow Task have the same value.
 
 ### Security
 * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency
@@ -50,8 +52,6 @@ relevant information.
   nondeterminism while retaining compatibility with histories recorded before these checks.
 
 ### Fixed
-* With the workflow cache off, the eviction after a terminal completion that the server accepted
-  has reason `WORKFLOW_EXECUTION_ENDING`, as with a cache, not `CACHE_FULL`.
 * Dev and test server startup now stops waiting at the configured deadline even if a readiness
   request stalls. Timeout messages include the last connection failure when available.
 * Workers with large workflow caches avoid unnecessary CPU usage when no workflow tasks are waiting

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

@@ -1,11 +1,11 @@
-//! The workflow task that each local activity start names.
-
 use crate::{
-    replay::TestHistoryBuilder,
+    replay::{DEFAULT_ACTIVITY_TYPE, TestHistoryBuilder},
     test_help::{
-        MockPollCfg, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers, mock_worker,
-        schedule_local_activity_cmd, start_timer_cmd,
+        MockPollCfg, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers,
+        hist_to_poll_resp, mock_worker, schedule_local_activity_cmd, single_hist_mock_sg,
+        start_timer_cmd,
     },
+    worker::client::mocks::mock_worker_client,
 };
 use std::time::{Duration, SystemTime};
 use temporalio_common::protos::{
@@ -17,7 +17,12 @@ use temporalio_common::protos::{
         workflow_commands::{ActivityCancellationType, ScheduleLocalActivity, workflow_command},
         workflow_completion::WorkflowActivationCompletion,
     },
-    temporal::api::{common::v1::RetryPolicy, enums::v1::EventType, failure::v1::Failure},
+    temporal::api::{
+        common::v1::RetryPolicy,
+        enums::v1::{EventType, WorkflowTaskFailedCause},
+        failure::v1::Failure,
+        workflowservice::v1::GetWorkflowExecutionHistoryResponse,
+    },
 };
 
 async fn complete(
@@ -34,10 +39,10 @@ async fn complete(
         .unwrap();
 }
 
-fn la_cmd() -> workflow_command::Variant {
+fn la_cmd(seq: u32) -> workflow_command::Variant {
     schedule_local_activity_cmd(
-        1,
-        "1",
+        seq,
+        &seq.to_string(),
         ActivityCancellationType::TryCancel,
         Duration::from_secs(60),
     )
@@ -100,7 +105,7 @@ async fn a_local_activity_names_the_live_task_that_dispatched_it() {
     .await;
     let fired = worker.poll_workflow_activation().await.unwrap();
     assert!(!fired.is_replaying);
-    complete(&worker, &fired, vec![la_cmd()]).await;
+    complete(&worker, &fired, vec![la_cmd(1)]).await;
 
     let la = worker.poll_activity_task().await.unwrap();
     let start = start_of(&la);
@@ -123,7 +128,7 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() {
     let mut t = TestHistoryBuilder::default();
     t.add_wfe_started_with_wft_timeout(Duration::from_secs(60));
     t.add_full_wf_task();
-    // The task after it ran the local activity, then the worker died and the task timed out.
+    // The second task ran the local activity. Then the worker stopped and the task timed out.
     t.add_workflow_task_scheduled_and_started();
     t.add_workflow_task_timed_out();
     t.add_workflow_task_scheduled_and_started();
@@ -132,12 +137,11 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() {
     let worker = mock_worker(mock);
 
     let first = worker.poll_workflow_activation().await.unwrap();
-    complete(&worker, &first, vec![la_cmd()]).await;
+    complete(&worker, &first, vec![la_cmd(1)]).await;
 
-    // History has no marker for it, so it runs again, dispatched by the live task.
+    // History has no marker for the local activity, so it runs again in the current task.
     let la = worker.poll_activity_task().await.unwrap();
     assert_eq!(dispatching(start_of(&la)), (9, 1));
-    // The start also names the last completed task and the task that timed out.
     let task = start_of(&la).workflow_task.as_ref().unwrap();
     assert_eq!(task.previous_started_event_id, 3);
     assert_eq!(task.failed_tasks, [6]);
@@ -193,3 +197,173 @@ async fn a_retry_that_core_runs_keeps_the_dispatching_task() {
     assert_eq!(dispatching(start_of(&second_attempt)), (3, 1));
     finish(worker, second_attempt).await;
 }
+
+#[tokio::test]
+async fn local_activities_of_one_activation_name_the_same_task() {
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_workflow_task_scheduled_and_started();
+    let mut mock = build_mock_pollers(MockPollCfg::from_resps(t, [ResponseType::AllHistory]));
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 1);
+    let worker = mock_worker(mock);
+
+    let first = worker.poll_workflow_activation().await.unwrap();
+    complete(&worker, &first, vec![la_cmd(1), la_cmd(2)]).await;
+
+    let one = worker.poll_activity_task().await.unwrap();
+    let two = worker.poll_activity_task().await.unwrap();
+    assert_eq!(dispatching(start_of(&one)), (3, 1));
+    assert_eq!(start_of(&one).workflow_task, start_of(&two).workflow_task);
+    worker
+        .complete_activity_task(ActivityTaskCompletion {
+            task_token: one.task_token,
+            result: Some(ActivityExecutionResult::ok(vec![1].into())),
+        })
+        .await
+        .unwrap();
+    let resolved = worker.poll_workflow_activation().await.unwrap();
+    complete(&worker, &resolved, vec![]).await;
+    finish(worker, two).await;
+}
+
+#[tokio::test]
+async fn a_reschedule_after_a_backoff_timer_names_the_later_task() {
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_full_wf_task();
+    // The first attempt failed with a backoff past the local retry threshold.
+    t.add_local_activity_marker(
+        1,
+        "1",
+        None,
+        Some(Failure::application_failure("retry me".to_string(), false)),
+        |m| m.backoff = Some(Duration::from_secs(10).try_into().unwrap()),
+    );
+    let timer_started = t.add_by_type(EventType::TimerStarted);
+    t.add_timer_fired(timer_started, "1".to_string());
+    t.add_workflow_task_scheduled_and_started();
+    let mut mock = build_mock_pollers(MockPollCfg::from_resps(t, [ResponseType::AllHistory]));
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 1);
+    let worker = mock_worker(mock);
+
+    let first = worker.poll_workflow_activation().await.unwrap();
+    complete(&worker, &first, vec![la_cmd(1)]).await;
+    let backoff = worker.poll_workflow_activation().await.unwrap();
+    assert!(backoff.is_replaying);
+    complete(
+        &worker,
+        &backoff,
+        vec![start_timer_cmd(1, Duration::from_secs(10))],
+    )
+    .await;
+    let fired = worker.poll_workflow_activation().await.unwrap();
+    assert!(!fired.is_replaying);
+    complete(
+        &worker,
+        &fired,
+        vec![
+            ScheduleLocalActivity {
+                seq: 2,
+                activity_id: "1".to_string(),
+                activity_type: DEFAULT_ACTIVITY_TYPE.to_string(),
+                attempt: 2,
+                original_schedule_time: Some(SystemTime::now().into()),
+                start_to_close_timeout: Some(Duration::from_secs(60).try_into().unwrap()),
+                ..Default::default()
+            }
+            .into(),
+        ],
+    )
+    .await;
+
+    let la = worker.poll_activity_task().await.unwrap();
+    assert_eq!(start_of(&la).attempt, 2);
+    assert_eq!(dispatching(start_of(&la)), (9, 1));
+    assert_eq!(
+        start_of(&la)
+            .workflow_task
+            .as_ref()
+            .unwrap()
+            .previous_started_event_id,
+        3
+    );
+    finish(worker, la).await;
+}
+
+#[tokio::test]
+async fn failed_tasks_from_every_history_page_are_named() {
+    let wf_id = "fake_wf_id";
+    let mut t = TestHistoryBuilder::default();
+    t.add_wfe_started_with_wft_timeout(Duration::from_secs(60));
+    t.add_full_wf_task();
+    let timer_started = t.add_by_type(EventType::TimerStarted);
+    t.add_timer_fired(timer_started, "1".to_string());
+    t.add_workflow_task_scheduled_and_started();
+    t.add_workflow_task_failed_with_failure(
+        WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure,
+        Failure::default(),
+    );
+    t.add_full_wf_task();
+    let timer_started = t.add_by_type(EventType::TimerStarted);
+    let second_page_end = timer_started as usize;
+    t.add_timer_fired(timer_started, "2".to_string());
+    t.add_workflow_task_scheduled_and_started();
+    t.add_workflow_task_timed_out();
+    t.add_workflow_task_scheduled_and_started();
+
+    // The poll response has the first page and each history fetch adds one page. The failed task
+    // is on the second page and the timed out task is on the last.
+    let mut first_page = hist_to_poll_resp(&t, wf_id, ResponseType::AllHistory).resp;
+    first_page.history.as_mut().unwrap().events.truncate(6);
+    first_page.next_page_token = vec![1];
+    let mut second_page: GetWorkflowExecutionHistoryResponse =
+        t.get_full_history_info().unwrap().into();
+    second_page
+        .history
+        .as_mut()
+        .unwrap()
+        .events
+        .truncate(second_page_end);
+    second_page.next_page_token = vec![2];
+    let last_page: GetWorkflowExecutionHistoryResponse = t.get_full_history_info().unwrap().into();
+    let mut client = mock_worker_client();
+    client
+        .expect_get_workflow_execution_history()
+        .returning(move |_, _, _| Ok(second_page.clone()))
+        .times(1);
+    client
+        .expect_get_workflow_execution_history()
+        .returning(move |_, _, _| Ok(last_page.clone()))
+        .times(1);
+    let mut mock = single_hist_mock_sg(wf_id, t, [ResponseType::Raw(first_page)], client, true);
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 1);
+    let worker = mock_worker(mock);
+
+    let first = worker.poll_workflow_activation().await.unwrap();
+    complete(
+        &worker,
+        &first,
+        vec![start_timer_cmd(1, Duration::from_secs(1))],
+    )
+    .await;
+    let fired = worker.poll_workflow_activation().await.unwrap();
+    assert!(fired.is_replaying);
+    complete(
+        &worker,
+        &fired,
+        vec![start_timer_cmd(2, Duration::from_secs(1))],
+    )
+    .await;
+    let fired = worker.poll_workflow_activation().await.unwrap();
+    assert!(!fired.is_replaying);
+    complete(&worker, &fired, vec![la_cmd(1)]).await;
+
+    let la = worker.poll_activity_task().await.unwrap();
+    let task = start_of(&la).workflow_task.as_ref().unwrap();
+    assert_eq!(
+        (task.started_event_id, task.previous_started_event_id),
+        (19, 11)
+    );
+    assert_eq!(task.failed_tasks, [8, 16]);
+    finish(worker, la).await;
+}