From 3969b33b8c9d5558f48efff285f99896a6e9999f Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 1 Jul 2026 12:13:46 -0700 Subject: [PATCH 1/6] Respect SDK flags already written in history regardless of GetSystemInfo --- crates/sdk-core/src/internal_flags.rs | 226 +++++++++++--------------- 1 file changed, 98 insertions(+), 128 deletions(-) diff --git a/crates/sdk-core/src/internal_flags.rs b/crates/sdk-core/src/internal_flags.rs index b03086f68..861ade531 100644 --- a/crates/sdk-core/src/internal_flags.rs +++ b/crates/sdk-core/src/internal_flags.rs @@ -1,11 +1,7 @@ //! Utilities for and tracking of internal versions which alter history in incompatible ways //! so that we can use older code paths for workflows executed on older core versions. -use itertools::Either; -use std::{ - collections::{BTreeSet, HashSet}, - iter, -}; +use std::collections::{BTreeSet, HashSet}; use temporalio_common::protos::temporal::api::{ history::v1::WorkflowTaskCompletedEventAttributes, sdk::v1::WorkflowTaskCompletedMetadata, workflowservice::v1::get_system_info_response, @@ -41,20 +37,17 @@ pub enum CoreInternalFlags { TooHigh = u32::MAX, } -#[allow(clippy::large_enum_variant)] #[derive(Debug, Clone, PartialEq, Eq)] -pub(crate) enum InternalFlags { - Enabled { - core: BTreeSet, - lang: BTreeSet, - core_since_last_complete: HashSet, - lang_since_last_complete: HashSet, - last_sdk_name: String, - last_sdk_version: String, - sdk_name: String, - sdk_version: String, - }, - Disabled, +pub(crate) struct InternalFlags { + can_write_sdk_metadata: bool, + core: BTreeSet, + lang: BTreeSet, + core_since_last_complete: HashSet, + lang_since_last_complete: HashSet, + last_sdk_name: String, + last_sdk_version: String, + sdk_name: String, + sdk_version: String, } impl InternalFlags { @@ -63,89 +56,64 @@ impl InternalFlags { sdk_name: String, sdk_version: String, ) -> Self { - match server_capabilities.sdk_metadata { - true => Self::Enabled { - core: Default::default(), - lang: Default::default(), - core_since_last_complete: Default::default(), - lang_since_last_complete: Default::default(), - last_sdk_name: "".to_string(), - last_sdk_version: "".to_string(), - sdk_name, - sdk_version, - }, - false => Self::Disabled, + Self { + can_write_sdk_metadata: server_capabilities.sdk_metadata, + core: Default::default(), + lang: Default::default(), + core_since_last_complete: Default::default(), + lang_since_last_complete: Default::default(), + last_sdk_name: "".to_string(), + last_sdk_version: "".to_string(), + sdk_name, + sdk_version, } } pub(crate) fn add_from_complete(&mut self, e: &WorkflowTaskCompletedEventAttributes) { - if let Self::Enabled { - core, - lang, - last_sdk_name, - last_sdk_version, - .. - } = self - && let Some(metadata) = e.sdk_metadata.as_ref() - { - core.extend( + if let Some(metadata) = e.sdk_metadata.as_ref() { + self.core.extend( metadata .core_used_flags .iter() .map(|u| CoreInternalFlags::from_u32(*u)), ); - lang.extend(metadata.lang_used_flags.iter()); + self.lang.extend(metadata.lang_used_flags.iter()); if !metadata.sdk_name.is_empty() { - *last_sdk_name = metadata.sdk_name.clone(); + self.last_sdk_name = metadata.sdk_name.clone(); } if !metadata.sdk_version.is_empty() { - *last_sdk_version = metadata.sdk_version.clone(); + self.last_sdk_version = metadata.sdk_version.clone(); } } } pub(crate) fn add_lang_used(&mut self, flags: impl IntoIterator) { - if let Self::Enabled { - lang_since_last_complete, - .. - } = self - { - lang_since_last_complete.extend(flags); + if self.can_write_sdk_metadata { + self.lang_since_last_complete.extend(flags); } } - /// Returns true if this flag may currently be used. If `should_record` is true, always returns - /// true and records the flag as being used, for taking later via - /// [Self::gather_for_wft_complete]. + /// Returns true if this flag may currently be used. If `should_record` is true, returns true + /// and records the flag only when new SDK metadata can be written. pub(crate) fn try_use(&mut self, flag: CoreInternalFlags, should_record: bool) -> bool { - match self { - Self::Enabled { - core, - core_since_last_complete, - .. - } => { - if should_record { - core_since_last_complete.insert(flag); - true - } else { - core.contains(&flag) - } + if should_record { + if self.can_write_sdk_metadata { + self.core_since_last_complete.insert(flag); + true + } else { + false } - // If the server does not support the metadata field, we must assume we can never use - // any internal flags since they can't be recorded for future use - Self::Disabled => false, + } else { + self.core.contains(&flag) } } /// Writes all known core flags to the set which should be recorded in the current WFT if not /// already known. Must only be called if not replaying. pub(crate) fn write_all_known(&mut self) { - if let Self::Enabled { - core_since_last_complete, - .. - } = self - { - core_since_last_complete.extend(CoreInternalFlags::all_except_too_high()); + if self.can_write_sdk_metadata { + self.core_since_last_complete + .extend(CoreInternalFlags::all_except_too_high()); } } @@ -153,63 +121,50 @@ impl InternalFlags { /// the last WFT complete. The returned value can be combined with other data before sending the /// WFT complete. pub(crate) fn gather_for_wft_complete(&mut self) -> WorkflowTaskCompletedMetadata { - match self { - Self::Enabled { - core_since_last_complete, - lang_since_last_complete, - core, - lang, - last_sdk_name, - last_sdk_version, - sdk_name, - sdk_version, - } => { - let core_newly_used: Vec<_> = core_since_last_complete - .iter() - .filter(|f| !core.contains(f)) - .map(|p| *p as u32) - .collect(); - let lang_newly_used: Vec<_> = lang_since_last_complete - .iter() - .filter(|f| !lang.contains(f)) - .copied() - .collect(); - core.extend(core_since_last_complete.iter()); - lang.extend(lang_since_last_complete.iter()); - let sdk_name = if last_sdk_name != sdk_name { - sdk_name.clone() - } else { - "".to_string() - }; - let sdk_version = if last_sdk_version != sdk_version { - sdk_version.clone() - } else { - "".to_string() - }; - WorkflowTaskCompletedMetadata { - core_used_flags: core_newly_used, - lang_used_flags: lang_newly_used, - sdk_name, - sdk_version, - } - } - Self::Disabled => WorkflowTaskCompletedMetadata::default(), + if !self.can_write_sdk_metadata { + return WorkflowTaskCompletedMetadata::default(); + } + let core_newly_used: Vec<_> = self + .core_since_last_complete + .iter() + .filter(|f| !self.core.contains(f)) + .map(|p| *p as u32) + .collect(); + let lang_newly_used: Vec<_> = self + .lang_since_last_complete + .iter() + .filter(|f| !self.lang.contains(f)) + .copied() + .collect(); + self.core.extend(self.core_since_last_complete.iter()); + self.lang.extend(self.lang_since_last_complete.iter()); + let sdk_name = if self.last_sdk_name != self.sdk_name { + self.sdk_name.clone() + } else { + "".to_string() + }; + let sdk_version = if self.last_sdk_version != self.sdk_version { + self.sdk_version.clone() + } else { + "".to_string() + }; + WorkflowTaskCompletedMetadata { + core_used_flags: core_newly_used, + lang_used_flags: lang_newly_used, + sdk_name, + sdk_version, } } pub(crate) fn all_lang(&self) -> impl Iterator + '_ { - match self { - Self::Enabled { lang, .. } => Either::Left(lang.iter().copied()), - Self::Disabled => Either::Right(iter::empty()), - } + self.lang.iter().copied() } pub(crate) fn last_sdk_version(&self) -> Option<&str> { - match self { - InternalFlags::Enabled { - last_sdk_version, .. - } if !last_sdk_version.is_empty() => Some(last_sdk_version), - InternalFlags::Enabled { .. } | InternalFlags::Disabled => None, + if !self.last_sdk_version.is_empty() { + Some(&self.last_sdk_version) + } else { + None } } } @@ -235,30 +190,45 @@ mod tests { use super::*; use temporalio_common::protos::temporal::api::workflowservice::v1::get_system_info_response::Capabilities; - #[allow(clippy::derivable_impls)] // Only want this in test impl Default for InternalFlags { fn default() -> Self { - Self::Disabled + Self::new(&Capabilities::default(), "".to_string(), "".to_string()) } } #[test] - fn disabled_in_capabilities_disables() { + fn metadata_disabled_honors_flags_from_history() { let mut f = InternalFlags::new( &Capabilities::default(), "name".to_string(), "ver".to_string(), ); - f.add_lang_used([1]); f.add_from_complete(&WorkflowTaskCompletedEventAttributes { sdk_metadata: Some(WorkflowTaskCompletedMetadata { core_used_flags: vec![1], - lang_used_flags: vec![], + lang_used_flags: vec![2], sdk_name: "".to_string(), sdk_version: "".to_string(), }), ..Default::default() }); + + assert!(f.try_use(CoreInternalFlags::IdAndTypeDeterminismChecks, false)); + assert!(f.all_lang().any(|flag| flag == 2)); + } + + #[test] + fn metadata_disabled_does_not_record_new_flags() { + let mut f = InternalFlags::new( + &Capabilities::default(), + "name".to_string(), + "ver".to_string(), + ); + f.add_lang_used([1]); + + assert!(!f.try_use(CoreInternalFlags::IdAndTypeDeterminismChecks, true)); + + f.write_all_known(); let gathered = f.gather_for_wft_complete(); assert_matches!(gathered.core_used_flags.as_slice(), &[]); assert_matches!(gathered.lang_used_flags.as_slice(), &[]); From 155d4af94309f2c83b9b77589e277114151e8a3c Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 1 Jul 2026 12:29:43 -0700 Subject: [PATCH 2/6] Narrow GetSystemInfo unimplemented fallback --- CHANGELOG.md | 3 ++ crates/client/src/lib.rs | 83 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 84 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b7e5639bb..70fe5db79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ to docs, or any other relevant information. dependency tree free of `ring`. ### Fixed +* `GetSystemInfo` connection initialization now only falls back to empty server capabilities when + `UNIMPLEMENTED` indicates the RPC method is missing. Other `UNIMPLEMENTED` responses are + reported as connection errors. * Awaiting a Nexus operation's result (`StartedNexusOperation::result()`) no longer trips nondeterminism detection ("a waker was invoked by a non-SDK source", TMPRL1100) on replay. The result future is a `Shared`, whose internal waker machinery must be polled inside an `SdkWakeGuard` diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 967507c14..cab393df8 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -255,7 +255,16 @@ impl Connection { { Ok(sysinfo) => sysinfo.into_inner().capabilities, Err(status) => match status.code() { - Code::Unimplemented => None, + Code::Unimplemented + if { + let msg = status.message().to_lowercase(); + msg.contains("unknown method") + || msg.contains("unknown service") + || msg.contains("method not found") + } => + { + None + } _ => return Err(ClientConnectError::SystemInfoCallError(status)), }, } @@ -1362,9 +1371,20 @@ pub(crate) use dbg_panic; #[cfg(test)] mod tests { use super::*; - use tonic::metadata::Ascii; + use crate::callback_based::CallbackBasedGrpcService; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tonic::{Status, metadata::Ascii}; use url::Url; + fn connection_options_for_system_info_test( + service_override: CallbackBasedGrpcService, + ) -> ConnectionOptions { + ConnectionOptions::new(Url::parse("http://localhost:7233").unwrap()) + .service_override(service_override) + .dns_load_balancing(None) + .build() + } + #[test] fn applies_headers() { // Initial header set @@ -1520,6 +1540,65 @@ mod tests { assert!(opts.keep_alive.is_none()); } + #[tokio::test] + async fn get_system_info_unknown_method_falls_back_to_empty_capabilities() { + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_clone = attempts.clone(); + let service_override = CallbackBasedGrpcService { + callback: Arc::new(move |req| { + let attempts = attempts_clone.clone(); + Box::pin(async move { + assert_eq!(req.rpc, "GetSystemInfo"); + attempts.fetch_add(1, Ordering::SeqCst); + Err(Status::unimplemented( + "unknown method GetSystemInfo for service \ + temporal.api.workflowservice.v1.WorkflowService", + )) + }) + }), + }; + + let connection = + Connection::connect(connection_options_for_system_info_test(service_override)) + .await + .unwrap(); + + assert!(connection.capabilities().is_none()); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn get_system_info_non_missing_unimplemented_fails_connect() { + let attempts = Arc::new(AtomicUsize::new(0)); + let attempts_clone = attempts.clone(); + let service_override = CallbackBasedGrpcService { + callback: Arc::new(move |req| { + let attempts = attempts_clone.clone(); + Box::pin(async move { + assert_eq!(req.rpc, "GetSystemInfo"); + attempts.fetch_add(1, Ordering::SeqCst); + Err(Status::unimplemented("backend temporarily unimplemented")) + }) + }), + }; + + let err = + match Connection::connect(connection_options_for_system_info_test(service_override)) + .await + { + Ok(_) => panic!("connection should fail"), + Err(err) => err, + }; + + assert!(matches!( + err, + ClientConnectError::SystemInfoCallError(status) + if status.code() == Code::Unimplemented + && status.message() == "backend temporarily unimplemented" + )); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + mod tls_custom_verifier_tests { use super::*; use tokio_rustls::rustls::{ From 7030bb804c792d32281ac42dad7c10d2a40027f2 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 1 Jul 2026 13:09:05 -0700 Subject: [PATCH 3/6] Disable compression if GSI call says it won't work --- CHANGELOG.md | 2 + crates/client/src/lib.rs | 64 ++++-- .../tests/integ_tests/client_tests.rs | 182 +++++++++++++++++- 3 files changed, 221 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 70fe5db79..b125b9dd6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ to docs, or any other relevant information. * `GetSystemInfo` connection initialization now only falls back to empty server capabilities when `UNIMPLEMENTED` indicates the RPC method is missing. Other `UNIMPLEMENTED` responses are reported as connection errors. +* Connection initialization now retries once with gRPC compression disabled if the eager + `GetSystemInfo` call fails because the server cannot decompress gzip. * Awaiting a Nexus operation's result (`StartedNexusOperation::result()`) no longer trips nondeterminism detection ("a waker was invoked by a non-SDK source", TMPRL1100) on replay. The result future is a `Shared`, whose internal waker machinery must be polled inside an `SdkWakeGuard` diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index cab393df8..9a57440d8 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -161,16 +161,32 @@ struct ConnectionInner { impl Connection { /// Connect to a Temporal service. - pub async fn connect(options: ConnectionOptions) -> Result { - let dns_lb_opts = dns::validate_and_get_dns_lb(&options)?.cloned(); - // The callback-based override transport cannot decode compressed request bodies, so - // compression is forced off whenever a service override is in use. - let compression = if options.service_override.is_some() { - GrpcCompression::None - } else { - options.grpc_compression - }; - let (service, dns_task) = if let Some(service_override) = options.service_override { + pub async fn connect(mut options: ConnectionOptions) -> Result { + if options.service_override.is_some() { + options.grpc_compression = GrpcCompression::None; + } + + let first_result = Self::connect_once(&options).await; + if options.grpc_compression == GrpcCompression::Gzip + && let Err(ClientConnectError::SystemInfoCallError(status)) = &first_result + && status.code() == Code::Unimplemented + && { + let msg = status.message().to_lowercase(); + msg.contains("gzip") + && (msg.contains("decompress") + || msg.contains("grpc-encoding") + || msg.contains("compressor")) + } + { + options.grpc_compression = GrpcCompression::None; + return Self::connect_once(&options).await; + } + first_result + } + + async fn connect_once(options: &ConnectionOptions) -> Result { + let dns_lb_opts = dns::validate_and_get_dns_lb(options)?.cloned(); + let (service, dns_task) = if let Some(service_override) = options.service_override.clone() { ( GrpcMetricSvc { inner: ChannelOrGrpcOverride::GrpcOverride(service_override), @@ -180,7 +196,7 @@ impl Connection { None, ) } else if let Some(dns_opts) = &dns_lb_opts { - let (channel, sender) = dns::create_balanced_channel(&options).await?; + let (channel, sender) = dns::create_balanced_channel(options).await?; let handle = dns::spawn_dns_reresolution( sender, options.target.clone(), @@ -246,7 +262,7 @@ impl Connection { headers: headers.clone(), }; let svc = InterceptedService::new(service, interceptor); - let mut svc_client = TemporalServiceClient::new(svc, compression); + let mut svc_client = TemporalServiceClient::new(svc, options.grpc_compression); let capabilities = if !options.skip_get_system_info { match svc_client @@ -261,6 +277,8 @@ impl Connection { msg.contains("unknown method") || msg.contains("unknown service") || msg.contains("method not found") + || (msg.contains("getsysteminfo") + && msg.contains("is unimplemented")) } => { None @@ -274,11 +292,11 @@ impl Connection { Ok(Self { inner: Arc::new(ConnectionInner { service: svc_client, - retry_options: options.retry_options, - identity: options.identity, + retry_options: options.retry_options.clone(), + identity: options.identity.clone(), headers, - client_name: options.client_name, - client_version: options.client_version, + client_name: options.client_name.clone(), + client_version: options.client_version.clone(), capabilities, workers: Arc::new(ClientWorkerSet::new()), _dns_task: dns_task, @@ -1540,8 +1558,15 @@ mod tests { assert!(opts.keep_alive.is_none()); } + #[rstest::rstest] + #[case( + "unknown method GetSystemInfo for service temporal.api.workflowservice.v1.WorkflowService" + )] + #[case("Method temporal.api.workflowservice.v1.WorkflowService/GetSystemInfo is unimplemented")] #[tokio::test] - async fn get_system_info_unknown_method_falls_back_to_empty_capabilities() { + async fn get_system_info_missing_method_falls_back_to_empty_capabilities( + #[case] message: &'static str, + ) { let attempts = Arc::new(AtomicUsize::new(0)); let attempts_clone = attempts.clone(); let service_override = CallbackBasedGrpcService { @@ -1550,10 +1575,7 @@ mod tests { Box::pin(async move { assert_eq!(req.rpc, "GetSystemInfo"); attempts.fetch_add(1, Ordering::SeqCst); - Err(Status::unimplemented( - "unknown method GetSystemInfo for service \ - temporal.api.workflowservice.v1.WorkflowService", - )) + Err(Status::unimplemented(message)) }) }), }; diff --git a/crates/sdk-core/tests/integ_tests/client_tests.rs b/crates/sdk-core/tests/integ_tests/client_tests.rs index 28957aea0..35ed911db 100644 --- a/crates/sdk-core/tests/integ_tests/client_tests.rs +++ b/crates/sdk-core/tests/integ_tests/client_tests.rs @@ -1,6 +1,6 @@ use crate::common::{ CoreWfStarter, NAMESPACE, - fake_grpc_server::{GenericService, fake_server}, + fake_grpc_server::{FakeServer, GenericService, fake_server}, get_integ_server_options, http_proxy::HttpProxy, }; @@ -16,20 +16,21 @@ use std::{ collections::HashMap, env, sync::{ - Arc, + Arc, Mutex, atomic::{AtomicUsize, Ordering}, }, time::Duration, }; use temporalio_client::{ - Connection, Namespace, RETRYABLE_ERROR_CODES, RetryOptions, UntypedWorkflow, - grpc::WorkflowService, proxy::HttpConnectProxyOptions, + Connection, GrpcCompression, Namespace, RETRYABLE_ERROR_CODES, RetryOptions, UntypedWorkflow, + errors::ClientConnectError, grpc::WorkflowService, proxy::HttpConnectProxyOptions, }; use temporalio_common::protos::temporal::api::{ cloud::cloudservice::v1::GetNamespaceRequest, workflowservice::v1::{ - DescribeNamespaceRequest, GetWorkflowExecutionHistoryRequest, ListNamespacesRequest, - RespondActivityTaskCanceledResponse, + DescribeNamespaceRequest, GetSystemInfoResponse, GetWorkflowExecutionHistoryRequest, + ListNamespacesRequest, RespondActivityTaskCanceledResponse, SignalWorkflowExecutionRequest, + SignalWorkflowExecutionResponse, get_system_info_response, }, }; #[cfg(unix)] @@ -79,6 +80,175 @@ async fn calls_get_system_info() { assert!(connection.capabilities().is_some()); } +#[derive(Clone)] +enum CompressionTestBehavior { + RejectGzipGetSystemInfo, + GenericGzipUnimplemented, + UnknownMethod, +} + +async fn compression_test_server( + behavior: CompressionTestBehavior, +) -> (FakeServer, Arc>>) { + let records = Arc::new(Mutex::new(Vec::new())); + let records_clone = records.clone(); + let fs = fake_server(move |req| { + let path = req.uri().path().to_string(); + let encoding = req + .headers() + .get("grpc-encoding") + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .to_string(); + records_clone + .lock() + .unwrap() + .push((path.clone(), encoding.clone())); + + let behavior = behavior.clone(); + async move { + match behavior { + CompressionTestBehavior::UnknownMethod => Status::unimplemented( + "unknown method GetSystemInfo for service \ + temporal.api.workflowservice.v1.WorkflowService", + ) + .into_http(), + CompressionTestBehavior::GenericGzipUnimplemented + if path.ends_with("/GetSystemInfo") => + { + Status::unimplemented("gzip is unavailable for this backend").into_http() + } + CompressionTestBehavior::RejectGzipGetSystemInfo + if path.ends_with("/GetSystemInfo") && encoding == "gzip" => + { + Status::unimplemented( + "grpc: Decompressor is not installed for grpc-encoding \"gzip\"", + ) + .into_http() + } + _ if path.ends_with("/GetSystemInfo") => make_ok_response(GetSystemInfoResponse { + capabilities: Some(get_system_info_response::Capabilities { + sdk_metadata: true, + ..Default::default() + }), + ..Default::default() + }), + _ if path.ends_with("/SignalWorkflowExecution") => { + make_ok_response(SignalWorkflowExecutionResponse::default()) + } + _ => Response::new(Body::empty()), + } + } + .boxed() + }) + .await; + (fs, records) +} + +fn compression_test_options( + fs: &FakeServer, + compression: GrpcCompression, +) -> temporalio_client::ConnectionOptions { + let mut opts = get_integ_server_options(); + opts.target = format!("http://localhost:{}", fs.addr.port()) + .parse::() + .unwrap(); + opts.retry_options = RetryOptions::no_retries(); + opts.grpc_compression = compression; + opts +} + +#[tokio::test] +async fn gzip_get_system_info_failure_reconnects_without_compression() { + let (fs, records) = + compression_test_server(CompressionTestBehavior::RejectGzipGetSystemInfo).await; + + let mut connection = Connection::connect(compression_test_options(&fs, GrpcCompression::Gzip)) + .await + .unwrap(); + assert!(connection.capabilities().unwrap().sdk_metadata); + + WorkflowService::signal_workflow_execution( + &mut connection, + SignalWorkflowExecutionRequest::default().into_request(), + ) + .await + .unwrap(); + + let records = records.lock().unwrap().clone(); + assert_eq!(records.len(), 3); + assert!(records[0].0.ends_with("/GetSystemInfo")); + assert_eq!(records[0].1, "gzip"); + assert!(records[1].0.ends_with("/GetSystemInfo")); + assert_eq!(records[1].1, ""); + assert!(records[2].0.ends_with("/SignalWorkflowExecution")); + assert_eq!(records[2].1, ""); + + fs.shutdown().await; +} + +#[tokio::test] +async fn compression_none_does_not_retry_compression_fallback() { + let (fs, records) = + compression_test_server(CompressionTestBehavior::RejectGzipGetSystemInfo).await; + + let connection = Connection::connect(compression_test_options(&fs, GrpcCompression::None)) + .await + .unwrap(); + assert!(connection.capabilities().unwrap().sdk_metadata); + + let records = records.lock().unwrap().clone(); + assert_eq!(records.len(), 1); + assert!(records[0].0.ends_with("/GetSystemInfo")); + assert_eq!(records[0].1, ""); + + fs.shutdown().await; +} + +#[tokio::test] +async fn generic_gzip_unimplemented_does_not_reconnect_without_compression() { + let (fs, records) = + compression_test_server(CompressionTestBehavior::GenericGzipUnimplemented).await; + + let err = match Connection::connect(compression_test_options(&fs, GrpcCompression::Gzip)).await + { + Ok(_) => panic!("connection should fail"), + Err(err) => err, + }; + + assert!(matches!( + err, + ClientConnectError::SystemInfoCallError(status) + if status.code() == Code::Unimplemented + && status.message() == "gzip is unavailable for this backend" + )); + + let records = records.lock().unwrap().clone(); + assert_eq!(records.len(), 1); + assert!(records[0].0.ends_with("/GetSystemInfo")); + assert_eq!(records[0].1, "gzip"); + + fs.shutdown().await; +} + +#[tokio::test] +async fn unknown_method_unimplemented_does_not_trigger_compression_reconnect() { + let (fs, records) = compression_test_server(CompressionTestBehavior::UnknownMethod).await; + + let connection = Connection::connect(compression_test_options(&fs, GrpcCompression::Gzip)) + .await + .unwrap(); + + assert!(connection.capabilities().is_none()); + + let records = records.lock().unwrap().clone(); + assert_eq!(records.len(), 1); + assert!(records[0].0.ends_with("/GetSystemInfo")); + assert_eq!(records[0].1, "gzip"); + + fs.shutdown().await; +} + #[tokio::test] async fn per_call_timeout_respected_whole_client() { let opts = get_integ_server_options(); From b2e54076022a87b67c0b5bad524f3ab194737f8e Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 1 Jul 2026 13:58:36 -0700 Subject: [PATCH 4/6] Don't check for gzip in unimplemented str --- crates/client/src/lib.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 9a57440d8..f029dbd4b 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -172,10 +172,9 @@ impl Connection { && status.code() == Code::Unimplemented && { let msg = status.message().to_lowercase(); - msg.contains("gzip") - && (msg.contains("decompress") - || msg.contains("grpc-encoding") - || msg.contains("compressor")) + msg.contains("decompress") + || msg.contains("grpc-encoding") + || msg.contains("compressor") } { options.grpc_compression = GrpcCompression::None; From 3e457d915bae2068075cd637201f54a96ce708ec Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Wed, 1 Jul 2026 15:39:48 -0700 Subject: [PATCH 5/6] Comment fix --- crates/sdk-core/src/internal_flags.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/sdk-core/src/internal_flags.rs b/crates/sdk-core/src/internal_flags.rs index 861ade531..8155a7f67 100644 --- a/crates/sdk-core/src/internal_flags.rs +++ b/crates/sdk-core/src/internal_flags.rs @@ -93,8 +93,8 @@ impl InternalFlags { } } - /// Returns true if this flag may currently be used. If `should_record` is true, returns true - /// and records the flag only when new SDK metadata can be written. + /// Returns true if this flag may currently be used. If `should_record` is true, and new SDK + /// metadata can be written, returns true and records the flag. pub(crate) fn try_use(&mut self, flag: CoreInternalFlags, should_record: bool) -> bool { if should_record { if self.can_write_sdk_metadata { From c993837299da0e130417724b327b7ad023c7a8ad Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Thu, 2 Jul 2026 09:13:41 -0700 Subject: [PATCH 6/6] Chris' suggestion --- crates/sdk-core/tests/integ_tests/client_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/sdk-core/tests/integ_tests/client_tests.rs b/crates/sdk-core/tests/integ_tests/client_tests.rs index 35ed911db..eaf9773e3 100644 --- a/crates/sdk-core/tests/integ_tests/client_tests.rs +++ b/crates/sdk-core/tests/integ_tests/client_tests.rs @@ -114,7 +114,7 @@ async fn compression_test_server( ) .into_http(), CompressionTestBehavior::GenericGzipUnimplemented - if path.ends_with("/GetSystemInfo") => + if path.ends_with("/GetSystemInfo") && encoding == "gzip" => { Status::unimplemented("gzip is unavailable for this backend").into_http() }