Merge branch 'master' into antithesis-instrumentation

Author: THardy98

crates/client/src/raw.rs

@@ -1575,6 +1575,8 @@ proxier! {
     (get_connectivity_rule, cloudreq::GetConnectivityRuleRequest, cloudreq::GetConnectivityRuleResponse);
     (get_connectivity_rules, cloudreq::GetConnectivityRulesRequest, cloudreq::GetConnectivityRulesResponse);
     (delete_connectivity_rule, cloudreq::DeleteConnectivityRuleRequest, cloudreq::DeleteConnectivityRuleResponse);
+    (set_service_account_namespace_access, cloudreq::SetServiceAccountNamespaceAccessRequest, cloudreq::SetServiceAccountNamespaceAccessResponse);
+    (validate_account_audit_log_sink, cloudreq::ValidateAccountAuditLogSinkRequest, cloudreq::ValidateAccountAuditLogSinkResponse);
 }
 
 proxier! {

crates/common/protos/api_cloud_upstream/VERSION

@@ -1 +1 @@
-v0.7.1
+v0.8.0

crates/common/protos/api_cloud_upstream/temporal/api/cloud/account/v1/message.proto

@@ -10,6 +10,7 @@ option ruby_package = "Temporalio::Api::Cloud::Account::V1";
 option csharp_namespace = "Temporalio.Api.Cloud.Account.V1";
 
 import "temporal/api/cloud/resource/v1/message.proto";
+import "temporal/api/cloud/sink/v1/message.proto";
 
 message MetricsSpec {
     // The ca cert(s) in PEM format that clients connecting to the metrics endpoint can use for authentication.
@@ -44,3 +45,20 @@ message Account {
     // Information related to metrics.
     Metrics metrics = 6;
 }
+
+// AuditLogSinkSpec is only used by Audit Log
+message AuditLogSinkSpec {
+  // Name of the sink e.g. "audit_log_01"
+  string name = 1;
+
+  oneof sink_type {
+    // The KinesisSpec when destination_type is Kinesis
+    api.cloud.sink.v1.KinesisSpec kinesis_sink = 2;
+
+    // The PubSubSpec when destination_type is PubSub
+    api.cloud.sink.v1.PubSubSpec pub_sub_sink = 3;
+  }
+
+  // Enabled indicates whether the sink is enabled or not.
+  bool enabled = 4;
+}

crates/common/protos/api_cloud_upstream/temporal/api/cloud/cloudservice/v1/request_response.proto

@@ -22,7 +22,7 @@ import "temporal/api/cloud/connectivityrule/v1/message.proto";
 
 message GetUsersRequest {
     // The requested size of the page to retrieve - optional.
-    // Cannot exceed 1000. Defaults to 100. 
+    // Cannot exceed 1000. Defaults to 100.
     int32 page_size = 1;
     // The page token if this is continuing from another response - optional.
     string page_token = 2;
@@ -144,8 +144,8 @@ message CreateNamespaceResponse {
 
 message GetNamespacesRequest {
     // The requested size of the page to retrieve.
-    // Cannot exceed 1000. 
-    // Optional, defaults to 100. 
+    // Cannot exceed 1000.
+    // Optional, defaults to 100.
     int32 page_size = 1;
     // The page token if this is continuing from another response.
     // Optional, defaults to empty.
@@ -387,7 +387,7 @@ message DeleteApiKeyResponse {
 
 message GetNexusEndpointsRequest {
     // The requested size of the page to retrieve - optional.
-    // Cannot exceed 1000. Defaults to 100. 
+    // Cannot exceed 1000. Defaults to 100.
     int32 page_size = 1;
 
     // The page token if this is continuing from another response - optional.
@@ -694,6 +694,25 @@ message UpdateServiceAccountResponse {
     temporal.api.cloud.operation.v1.AsyncOperation async_operation = 1;
 }
 
+message SetServiceAccountNamespaceAccessRequest {
+    // The ID of the service account to update.
+    string service_account_id = 1;
+    // The namespace to set permissions for.
+    string namespace = 2;
+    // The namespace access to assign the service account.
+    temporal.api.cloud.identity.v1.NamespaceAccess access = 3;
+    // The version of the service account for which this update is intended for.
+    // The latest version can be found in the GetServiceAccount response.
+    string resource_version = 4;
+    // The ID to use for this async operation - optional.
+    string async_operation_id = 5;
+}
+
+message SetServiceAccountNamespaceAccessResponse {
+    // The async operation.
+    temporal.api.cloud.operation.v1.AsyncOperation async_operation = 1;
+}
+
 message DeleteServiceAccountRequest {
     // The ID of the service account to delete;
     string service_account_id = 1;
@@ -715,18 +734,18 @@ message GetUsageRequest {
     // Must be: within the last 90 days from the current date.
     // Must be: midnight UTC time.
     google.protobuf.Timestamp start_time_inclusive = 1;
-    
+
     // Filter for UTC time < - optional.
     // Defaults to: start of the next UTC day.
     // Must be: within the last 90 days from the current date.
     // Must be: midnight UTC time.
     google.protobuf.Timestamp end_time_exclusive = 2;
-    
+
     // The requested size of the page to retrieve - optional.
     // Each count corresponds to a single object - per day per namespace
     // Cannot exceed 1000. Defaults to 100.
     int32 page_size = 3;
-    
+
     // The page token if this is continuing from another response - optional.
     string page_token = 4;
 }
@@ -853,11 +872,11 @@ message ValidateNamespaceExportSinkResponse {
 message UpdateNamespaceTagsRequest {
     // The namespace to set tags for.
     string namespace = 1;
-    // A list of tags to add or update. 
-    // If a key of an existing tag is added, the tag's value is updated. 
+    // A list of tags to add or update.
+    // If a key of an existing tag is added, the tag's value is updated.
     // At least one of tags_to_upsert or tags_to_remove must be specified.
     map<string, string> tags_to_upsert = 2;
-    // A list of tag keys to remove. 
+    // A list of tag keys to remove.
     // If a tag key doesn't exist, it is silently ignored.
     // At least one of tags_to_upsert or tags_to_remove must be specified.
     repeated string tags_to_remove = 3;
@@ -868,7 +887,7 @@ message UpdateNamespaceTagsRequest {
 message UpdateNamespaceTagsResponse {
     // The async operation.
     temporal.api.cloud.operation.v1.AsyncOperation async_operation = 1;
-} 
+}
 
 message CreateConnectivityRuleRequest {
     // The connectivity rule specification.
@@ -928,3 +947,11 @@ message DeleteConnectivityRuleResponse {
     // The async operation
     temporal.api.cloud.operation.v1.AsyncOperation async_operation = 1;
 }
+
+message ValidateAccountAuditLogSinkRequest {
+  // The audit log sink spec that will be validated
+  temporal.api.cloud.account.v1.AuditLogSinkSpec spec = 1;
+}
+
+message ValidateAccountAuditLogSinkResponse {
+}

crates/common/protos/api_cloud_upstream/temporal/api/cloud/cloudservice/v1/service.proto

@@ -21,7 +21,7 @@ service CloudService {
             get: "/cloud/users",
         };
     }
-    
+
     // Get a user
     rpc GetUser(GetUserRequest) returns (GetUserResponse) {
         option (google.api.http) = {
@@ -192,7 +192,7 @@ service CloudService {
             get: "/cloud/nexus/endpoints",
         };
     }
-    
+
     // Get a nexus endpoint
     rpc GetNexusEndpoint(GetNexusEndpointRequest) returns (GetNexusEndpointResponse) {
         option (google.api.http) = {
@@ -320,6 +320,14 @@ service CloudService {
         };
     }
 
+    // Set a service account's access to a namespace.
+    rpc SetServiceAccountNamespaceAccess(SetServiceAccountNamespaceAccessRequest) returns (SetServiceAccountNamespaceAccessResponse) {
+        option (google.api.http) = {
+            post: "/cloud/namespaces/{namespace}/service-accounts/{service_account_id}/access",
+            body: "*"
+        };
+    }
+
     // Delete a service account.
     rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {
         option (google.api.http) = {
@@ -358,7 +366,7 @@ service CloudService {
         };
     }
 
-    // Get an export sink 
+    // Get an export sink
     rpc GetNamespaceExportSink(GetNamespaceExportSinkRequest) returns (GetNamespaceExportSinkResponse) {
         option (google.api.http) = {
             get: "/cloud/namespaces/{namespace}/export-sinks/{name}"
@@ -432,4 +440,13 @@ service CloudService {
             delete: "/cloud/connectivity-rules/{connectivity_rule_id}"
         };
     }
-}
+
+    // Validate customer audit log sink is accessible from Temporal's workflow by delivering an empty file to the specified sink.
+    // The operation verifies that the sink is correctly configured, accessible and ready to receive audit logs.
+    rpc ValidateAccountAuditLogSink(ValidateAccountAuditLogSinkRequest) returns (ValidateAccountAuditLogSinkResponse) {
+        option (google.api.http) = {
+            post: "/cloud/account/audit-logs/sink/validate",
+            body: "*"
+        };
+    }
+}

crates/common/protos/api_cloud_upstream/temporal/api/cloud/operation/v1/message.proto

@@ -41,12 +41,12 @@ message AsyncOperation {
     google.protobuf.Timestamp finished_time = 8;
 
     enum State {
-	STATE_UNSPECIFIED = 0;
-	STATE_PENDING = 1;     // The operation is pending.
-	STATE_IN_PROGRESS = 2; // The operation is in progress.
-	STATE_FAILED = 3;      // The operation failed, check failure_reason for more details.
-	STATE_CANCELLED = 4;   // The operation was cancelled.
-	STATE_FULFILLED = 5;   // The operation was fulfilled.
+        STATE_UNSPECIFIED = 0;
+        STATE_PENDING = 1;     // The operation is pending.
+        STATE_IN_PROGRESS = 2; // The operation is in progress.
+        STATE_FAILED = 3;      // The operation failed, check failure_reason for more details.
+        STATE_CANCELLED = 4;   // The operation was cancelled.
+        STATE_FULFILLED = 5;   // The operation was fulfilled.
         STATE_REJECTED = 6;    // The operation was rejected.
     }
 }

crates/common/protos/api_cloud_upstream/temporal/api/cloud/sink/v1/message.proto

@@ -38,4 +38,26 @@ message GCSSpec {
 
     // The region of the gcs bucket
     string region = 4;
+}
+
+message KinesisSpec {
+    // The role Temporal Cloud assumes when writing records to Kinesis
+    string role_name = 1;
+
+    // Destination Kinesis endpoint arn for temporal to send data to.
+    string destination_uri = 2;
+
+    // The sink's region.
+    string region = 3;
+}
+
+message PubSubSpec{
+    // The customer service account id that Temporal Cloud impersonates for writing records to customer's pubsub topic
+    string service_account_id = 1;
+
+    // Destination pubsub topic name for us
+    string topic_name = 2;
+
+    // The gcp project id of pubsub topic and service account
+    string gcp_project_id = 3;
 }

crates/sdk-core-c-bridge/src/client.rs

@@ -1058,6 +1058,17 @@ async fn call_cloud_service(client: &CoreClient, call: &RpcCallOptions) -> anyho
         "DeleteConnectivityRule" => {
             rpc_call_on_trait!(client, call, CloudService, delete_connectivity_rule)
         }
+        "SetServiceAccountNamespaceAccess" => {
+            rpc_call_on_trait!(
+                client,
+                call,
+                CloudService,
+                set_service_account_namespace_access
+            )
+        }
+        "ValidateAccountAuditLogSink" => {
+            rpc_call_on_trait!(client, call, CloudService, validate_account_audit_log_sink)
+        }
         rpc => Err(anyhow::anyhow!("Unknown RPC call {rpc}")),
     }
 }

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

@@ -27,6 +27,7 @@ use std::{
     },
     time::Duration,
 };
+use temporalio_client::MESSAGE_TOO_LARGE_KEY;
 use temporalio_common::{
     Worker as WorkerTrait,
     errors::PollError,
@@ -3070,3 +3071,54 @@ async fn both_normal_and_sticky_pollers_poll_concurrently() {
         "At peak, there should be exactly 2 pollers active at the same time"
     );
 }
+
+#[tokio::test]
+async fn grpc_message_too_large_doesnt_spam_task_fails() {
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_workflow_task_scheduled_and_started();
+
+    let mut mh = MockPollCfg::from_resp_batches(
+        "fake_wf_id",
+        t,
+        [
+            ResponseType::AllHistory,
+            ResponseType::AllHistory,
+            ResponseType::AllHistory,
+            ResponseType::AllHistory,
+            ResponseType::AllHistory,
+            ResponseType::AllHistory,
+        ],
+        mock_worker_client(),
+    );
+    mh.num_expected_fails = 1;
+    let mut times = 1;
+    mh.completion_mock_fn = Some(Box::new(move |_| {
+        if times <= 5 {
+            let mut err = tonic::Status::new(tonic::Code::ResourceExhausted, "message too large");
+            // This key is what we look for
+            err.metadata_mut().insert(MESSAGE_TOO_LARGE_KEY, 1.into());
+            times += 1;
+            Err(err)
+        } else {
+            Ok(Default::default())
+        }
+    }));
+
+    let mut mock = build_mock_pollers(mh);
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 1);
+    let core = mock_worker(mock);
+
+    // Since the mock makes us fail 5 times, we should succeed on the sixth
+    for _ in 1..=5 {
+        let act = core.poll_workflow_activation().await.unwrap();
+        core.complete_workflow_activation(WorkflowActivationCompletion::empty(&act.run_id))
+            .await
+            .unwrap();
+        core.handle_eviction().await;
+    }
+    let act = core.poll_workflow_activation().await.unwrap();
+    core.complete_execution(&act.run_id).await;
+    core.drain_pollers_and_shutdown().await;
+    // Mock only expects 1 task failure, and would fail here if we spammed
+}