Fix new clippy lints & integ test recompiles (#923)

Author: Sushisource

.cargo/config.toml

@@ -4,7 +4,7 @@
 CLI_VERSION_OVERRIDE = "v1.3.1-priority.0"
 
 [alias]
-integ-test = ["run", "--package", "temporal-sdk-core", "--example", "integ_runner", "--"]
+integ-test = ["test", "--features", "temporal-sdk-core-protos/serde_serialize", "--package", "temporal-sdk-core", "--test", "integ_runner", "--"]
 lint = ["clippy", "--workspace", "--examples", "--all-features",
     "--test", "integ_tests", "--test", "heavy_tests", "--test", "manual_tests",
     "--", "--D", "warnings"]

.github/workflows/per-pr.yml

@@ -21,7 +21,7 @@ jobs:
           submodules: recursive
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.85.0
+          toolchain: 1.87.0
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -56,7 +56,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.85.0
+          toolchain: 1.87.0
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -95,7 +95,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.85.0
+          toolchain: 1.87.0
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -120,7 +120,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.85.0
+          toolchain: 1.87.0
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -145,7 +145,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.85.0
+          toolchain: 1.87.0
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:

AGENTS.md

@@ -1,6 +1,7 @@
 # Contributor Guidance for `sdk-core`
 
-This repository provides a Rust workspace for the Temporal Core SDK and related crates. Use this document as your quick reference when submitting pull requests.
+This repository provides a Rust workspace for the Temporal Core SDK and related crates. Use this
+document as your quick reference when submitting pull requests.
 
 ## Where Things Are
 
@@ -18,12 +19,13 @@ This repository provides a Rust workspace for the Temporal Core SDK and related
 ## Repo Specific Utilities
 
 - `.cargo/config.toml` defines useful cargo aliases:
-  - `cargo lint` – run clippy on workspace crates
-  - `cargo test-lint` – run clippy on tests
-  - `cargo integ-test` – run the integration test runner
+    - `cargo lint` – run clippy on workspace crates
+    - `cargo test-lint` – run clippy on tests
+    - `cargo integ-test` – run the integration test runner
 - `cargo-tokio-console.sh` – run any cargo command with the `tokio-console` feature
 - `integ-with-otel.sh` – run integration tests with OpenTelemetry enabled
-- `.cargo/multi-worker-manual-test` – helper script for spawning multiple workers during manual testing
+- `.cargo/multi-worker-manual-test` – helper script for spawning multiple workers during manual
+  testing
 
 ## Building and Testing
 
@@ -33,7 +35,7 @@ The following commands are enforced for each pull request (see `README.md`):
 cargo build            # build all crates
 cargo test             # run unit tests
 cargo integ-test       # integration tests (starts ephemeral server by default)
-cargo test --test heavy_tests  # load tests
+cargo test --test heavy_tests  # load tests -- agents do not need to run this and should not
 ```
 
 Additional checks:
@@ -48,7 +50,8 @@ Documentation can be generated with `cargo doc`.
 ## Expectations for Pull Requests
 
 - Format and lint your code before submitting.
-- Ensure all tests pass locally. Integration tests may require a running Temporal server or the ephemeral server started by `cargo integ-test`.
+- Ensure all tests pass locally. Integration tests may require a running Temporal server or the
+  ephemeral server started by `cargo integ-test`.
 - Keep commit messages short and in the imperative mood.
 - Provide a clear PR description outlining what changed and why.
 - Reviewers expect new features or fixes to include corresponding tests when applicable.
@@ -64,5 +67,7 @@ Reviewers will look for:
 
 ## Notes
 
-- Fetch workflow histories with `cargo run --bin histfetch <workflow_id> [run_id]` (binary lives in `test-utils`).
-- Protobuf files under `sdk-core-protos/protos/api_upstream` are a git subtree; see `README.md` for update instructions.
+- Fetch workflow histories with `cargo run --bin histfetch <workflow_id> [run_id]` (binary lives in
+  `test-utils`).
+- Protobuf files under `sdk-core-protos/protos/api_upstream` are a git subtree; see `README.md` for
+  update instructions.

core/Cargo.toml

@@ -129,11 +129,16 @@ test = false
 name = "workflow_replay"
 harness = false
 
-# This is maybe a bit hacky, but we call the runner an "example" because that gets it compiling with
-# the dev-dependencies, which we want.
-[[example]]
+# The integration test runner should compile with the same configuration as the
+# rest of the integration tests so that artifacts are shared and no additional
+# compilation is required when switching between using the runner and running
+# tests directly.
+[[test]]
 name = "integ_runner"
 path = "../tests/runner.rs"
+harness = false
+test = false
+required-features = ["temporal-sdk-core-protos/serde_serialize"]
 
 [lints]
 workspace = true

core/src/ephemeral_server/mod.rs

@@ -546,7 +546,7 @@ async fn download_and_extract(
     // We have to map the error type to an io error
     let stream = resp
         .bytes_stream()
-        .map(|item| item.map_err(|err| io::Error::new(io::ErrorKind::Other, err)));
+        .map(|item| item.map_err(io::Error::other));
 
     // Since our tar/zip impls use sync IO, we have to create a bridge and run
     // in a blocking closure.

core/src/internal_flags.rs

@@ -40,6 +40,7 @@ pub(crate) enum CoreInternalFlags {
     TooHigh = u32::MAX,
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum InternalFlags {
     Enabled {

core/src/protosext/mod.rs

@@ -260,6 +260,7 @@ pub(crate) struct CompleteLocalActivityData {
     pub(crate) result: Result<Payload, Failure>,
 }
 
+#[allow(clippy::result_large_err)]
 pub(crate) fn validate_activity_completion(
     status: &activity_execution_result::Status,
 ) -> Result<(), CompleteActivityError> {

core/src/worker/activities.rs

@@ -175,7 +175,7 @@ pub(crate) struct WorkerActivityTasks {
 #[derive(derive_more::From)]
 enum ActivityTaskSource {
     PendingCancel(PendingActivityCancel),
-    PendingStart(Result<(PermittedTqResp<PollActivityTaskQueueResponse>, bool), PollError>),
+    PendingStart(Box<Result<(PermittedTqResp<PollActivityTaskQueueResponse>, bool), PollError>>),
 }
 
 impl WorkerActivityTasks {
@@ -209,7 +209,7 @@ impl WorkerActivityTasks {
         let complete_notify = Arc::new(Notify::new());
         let source_stream = stream::select_with_strategy(
             UnboundedReceiverStream::new(cancels_rx).map(ActivityTaskSource::from),
-            starts_stream.map(ActivityTaskSource::from),
+            starts_stream.map(|a| ActivityTaskSource::from(Box::new(a))),
             |_: &mut ()| PollNext::Left,
         );
 

core/src/worker/activities/local_activities.rs

@@ -893,6 +893,7 @@ impl TimeoutBag {
     /// as request to schedule it arrives.
     ///
     /// Returns error in the event the activity is *already* timed out
+    #[allow(clippy::result_large_err)]
     fn new(
         new_la: &NewLocalAct,
         cancel_chan: UnboundedSender<CancelOrTimeout>,

core/src/worker/nexus.rs

@@ -71,7 +71,7 @@ impl NexusManager {
         let task_stream_input = stream::select_with_strategy(
             UnboundedReceiverStream::new(cancels_rx).map(TaskStreamInput::from),
             source_stream
-                .map(TaskStreamInput::from)
+                .map(|p| TaskStreamInput::from(Box::new(p)))
                 .chain(stream::once(async move { TaskStreamInput::SourceComplete })),
             |_: &mut ()| PollNext::Left,
         );
@@ -244,90 +244,92 @@ where
         self.source_stream
             .filter_map(move |t| {
                 let res = match t {
-                    TaskStreamInput::Poll(Ok(t)) => {
-                        if let Some(dur) = t.resp.sched_to_start() {
-                            self.metrics.nexus_task_sched_to_start_latency(dur);
-                        };
+                    TaskStreamInput::Poll(t) => match *t {
+                        Ok(t) => {
+                            if let Some(dur) = t.resp.sched_to_start() {
+                                self.metrics.nexus_task_sched_to_start_latency(dur);
+                            };
 
-                        let tt = TaskToken(t.resp.task_token.clone());
-                        let mut timeout_task = None;
-                        if let Some(timeout_str) = t
-                            .resp
-                            .request
-                            .as_ref()
-                            .and_then(|r| r.header.get(REQUEST_TIMEOUT_HEADER))
-                        {
-                            if let Ok(timeout_dur) = parse_request_timeout(timeout_str) {
-                                let tt_clone = tt.clone();
-                                let cancels_tx = self.cancels_tx.clone();
-                                timeout_task = Some(tokio::task::spawn(async move {
-                                    tokio::time::sleep(timeout_dur).await;
-                                    debug!(
+                            let tt = TaskToken(t.resp.task_token.clone());
+                            let mut timeout_task = None;
+                            if let Some(timeout_str) = t
+                                .resp
+                                .request
+                                .as_ref()
+                                .and_then(|r| r.header.get(REQUEST_TIMEOUT_HEADER))
+                            {
+                                if let Ok(timeout_dur) = parse_request_timeout(timeout_str) {
+                                    let tt_clone = tt.clone();
+                                    let cancels_tx = self.cancels_tx.clone();
+                                    timeout_task = Some(tokio::task::spawn(async move {
+                                        tokio::time::sleep(timeout_dur).await;
+                                        debug!(
                                         task_token=%tt_clone,
                                         "Timing out nexus task due to elapsed local timeout timer"
                                     );
-                                    let _ = cancels_tx.send(CancelNexusTask {
-                                        task_token: tt_clone.0,
-                                        reason: NexusTaskCancelReason::TimedOut.into(),
-                                    });
-                                }));
-                            } else {
-                                // This could auto-respond and fail the nexus task, but given that
-                                // the server is going to try to parse this as well, and all we're
-                                // doing with this parsing is notifying the handler of a local
-                                // timeout, it seems reasonable to rely on server to handle this.
-                                warn!(
+                                        let _ = cancels_tx.send(CancelNexusTask {
+                                            task_token: tt_clone.0,
+                                            reason: NexusTaskCancelReason::TimedOut.into(),
+                                        });
+                                    }));
+                                } else {
+                                    // This could auto-respond and fail the nexus task, but given that
+                                    // the server is going to try to parse this as well, and all we're
+                                    // doing with this parsing is notifying the handler of a local
+                                    // timeout, it seems reasonable to rely on server to handle this.
+                                    warn!(
                                     "Failed to parse nexus timeout header value '{}'",
                                     timeout_str
                                 );
+                                }
                             }
-                        }
 
-                        let (service, operation, request_kind) = t
-                            .resp
-                            .request
-                            .as_ref()
-                            .and_then(|r| r.variant.as_ref())
-                            .map(|v| match v {
-                                Variant::StartOperation(s) => (
-                                    s.service.to_owned(),
-                                    s.operation.to_owned(),
-                                    RequestKind::Start,
-                                ),
-                                Variant::CancelOperation(c) => (
-                                    c.service.to_owned(),
-                                    c.operation.to_owned(),
-                                    RequestKind::Cancel,
-                                ),
-                            })
-                            .unwrap_or_default();
-                        self.outstanding_task_map.lock().insert(
-                            tt,
-                            NexusInFlightTask {
-                                request_kind,
-                                timeout_task,
-                                scheduled_time: t
-                                    .resp
-                                    .request
-                                    .as_ref()
-                                    .and_then(|r| r.scheduled_time)
-                                    .and_then(|t| t.try_into().ok()),
-                                start_time: Instant::now(),
-                                _permit: t.permit.into_used(NexusSlotInfo { service, operation }),
-                            },
-                        );
-                        Some(Ok(NexusTask {
-                            variant: Some(nexus_task::Variant::Task(t.resp)),
-                        }))
-                    }
+                            let (service, operation, request_kind) = t
+                                .resp
+                                .request
+                                .as_ref()
+                                .and_then(|r| r.variant.as_ref())
+                                .map(|v| match v {
+                                    Variant::StartOperation(s) => (
+                                        s.service.to_owned(),
+                                        s.operation.to_owned(),
+                                        RequestKind::Start,
+                                    ),
+                                    Variant::CancelOperation(c) => (
+                                        c.service.to_owned(),
+                                        c.operation.to_owned(),
+                                        RequestKind::Cancel,
+                                    ),
+                                })
+                                .unwrap_or_default();
+                            self.outstanding_task_map.lock().insert(
+                                tt,
+                                NexusInFlightTask {
+                                    request_kind,
+                                    timeout_task,
+                                    scheduled_time: t
+                                        .resp
+                                        .request
+                                        .as_ref()
+                                        .and_then(|r| r.scheduled_time)
+                                        .and_then(|t| t.try_into().ok()),
+                                    start_time: Instant::now(),
+                                    _permit: t.permit.into_used(NexusSlotInfo { service, operation }),
+                                },
+                            );
+                            Some(Ok(NexusTask {
+                                variant: Some(nexus_task::Variant::Task(t.resp)),
+                            }))
+                        },
+                        Err(e) => Some(Err(PollError::TonicError(e)))
+                    },
                     TaskStreamInput::Cancel(c) => Some(Ok(NexusTask {
                         variant: Some(nexus_task::Variant::CancelTask(c)),
                     })),
                     TaskStreamInput::SourceComplete => {
                         source_done.cancel();
                         None
                     }
-                    TaskStreamInput::Poll(Err(e)) => Some(Err(PollError::TonicError(e))),
                 };
                 async move { res }
             })
@@ -379,7 +381,7 @@ enum RequestKind {
 
 #[derive(derive_more::From)]
 enum TaskStreamInput {
-    Poll(NexusPollItem),
+    Poll(Box<NexusPollItem>),
     Cancel(CancelNexusTask),
     SourceComplete,
 }