Search attributes for workflow indexing
The starter code demonstrates setting SearchAttributes when starting a Workflow: SearchAttributes.from_pairs([('DocumentId', document.document_id), ('SubmitterEmail', document.submitter_email), ('ApprovalStatus', 'SUBMITTED')]). After the Workflow completes, the status can be updated: client.update_workflow_search_attributes(workflow_id, SearchAttributes.from_pairs([('ApprovalStatus', result['status'].upper())])). This enables filtering and searching for Workflows by document ID, submitter, and approval status in the Temporal UI and via list queries.
Workflow.wait_condition for signal-based control flow
workflow.wait_condition(lambda_predicate, timeout=timedelta(...)) blocks the Workflow until the condition becomes True or the timeout expires. In the approval pattern: wait_condition(lambda: self._pending_decision is not None or self._state.status == DocumentStatus.WITHDRAWN) waits for either a decision Signal or a withdrawal Signal. When either Signal handler sets the corresponding state, the condition becomes True and the wait unblocks. This allows the Workflow to react to external events without consuming compute.
Workflow allHandlersFinished for clean completion
Before returning from a Workflow, it is good practice to call await workflow.wait_condition(workflow.all_handlers_finished) to ensure all in-flight Signal and Update handlers complete before the Workflow result is returned. This prevents potential race conditions where a Signal arrives just as the Workflow is completing. The approval pattern calls this before the final return statement and before calling continue_as_new().
ApprovalDecision timestamp set by Workflow
When an ApprovalDecision arrives via Signal, the decided_at field is initially empty. The Workflow stamps the decision with the processing time using the deterministic Workflow clock: decision.decided_at = workflow.now().isoformat(). This ensures all timestamps in the approval process use the Server's time (via workflow.now()), not the Signal sender's local clock, maintaining determinism and audit trail accuracy.
Workflow resource-aware routing example
Example workflow routing Activities to appropriate Task Queues based on resource requirements:
```python
from datetime import timedelta
from dataclasses import dataclass
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from task_queues import STANDARD_CPU_QUEUE, GPU_ML_QUEUE, HIGH_MEMORY_QUEUE
@dataclass
class MLPipelineRequest:
pipeline_id: str
dataset_url: str
model_type: str
customer_id: str
@workflow.defn
class MLPipelineWorkflow:
"""ML pipeline workflow with resource-aware routing."""
@workflow.run
async def run(self, request: MLPipelineRequest) -> dict:
workflow.logger.info(f"Starting ML pipeline {request.pipeline_id}")
# Validate data (CPU queue)
validation_result = await workflow.execute_activity(
"validate_data",
{"dataset_url": request.dataset_url},
task_queue=STANDARD_CPU_QUEUE,
start_to_close_timeout=timedelta(minutes=2),
)
if not validation_result["valid"]:
return {"status": "validation_failed", "error": validation_result["error"]}
# Process large dataset (High-memory queue)
processed_data = await workflow.execute_activity(
"process_large_dataset",
{"dataset_url": request.dataset_url, "pipeline_id": request.pipeline_id},
task_queue=HIGH_MEMORY_QUEUE,
start_to_close_timeout=timedelta(hours=1),
)
# Train model (GPU queue)
model_result = await workflow.execute_activity(
"train_model",
{
"model_type": request.model_type,
"data_path": processed_data["output_path"],
"pipeline_id": request.pipeline_id,
},
task_queue=GPU_ML_QUEUE,
start_to_close_timeout=timedelta(hours=4),
)
# Generate embeddings (GPU queue)
embeddings = await workflow.execute_activity(
"generate_embeddings",
{
"model_path": model_result["model_path"],
"customer_id": request.customer_id,
},
task_queue=GPU_ML_QUEUE,
start_to_close_timeout=timedelta(minutes=30),
)
# Store results (CPU queue)
await workflow.execute_activity(
"store_results",
{
"pipeline_id": request.pipeline_id,
"embeddings": embeddings,
"model_metrics": model_result["metrics"],
},
task_queue=STANDARD_CPU_QUEUE,
start_to_close_timeout=timedelta(minutes=5),
)
return {
"status": "completed",
"pipeline_id": request.pipeline_id,
"model_path": model_result["model_path"],
"embeddings_count": len(embeddings["vectors"]),
}
```
Each workflow.execute_activity() call includes a task_queue parameter specifying which resource-specific queue to use based on Activity requirements.
ParentClosePolicyAbandon allows child workflows to continue after parent completion
The ParentClosePolicyAbandon setting instructs the Temporal Service to allow a Child Workflow to continue running even after the parent Workflow completes. This decouples the client's synchronous request from the long-running Time-To-Live (TTL) wait state. When the Child Workflow enters a sleep state, it consumes no Worker memory, allowing the pattern to scale across thousands of concurrent requests efficiently.
Durable timers guarantee resource cleanup
Temporal persists the state of Workflows, including durably storing and starting Timers. By using durable Timers in a Child Workflow, capacity increases are guaranteed to revert reliably after the specified duration, preventing runaway costs if operators would otherwise forget to manually reduce limits.
ProvisionTRUWorkflow execution flow in Go SDK
The ProvisionTRUWorkflow first executes AddTRUs Activity with a 2-minute StartToCloseTimeout and retry policy, waiting for it to complete synchronously to ensure capacity is available before returning. It then configures ChildWorkflowOptions with WorkflowID derived from namespace and ParentClosePolicy set to enums.PARENT_CLOSE_POLICY_ABANDON. It executes DeprovisionTRUWorkflow as a Child Workflow and calls GetChildWorkflowExecution().Get() to block execution only until the Temporal Service confirms the Child Workflow has started. Once started, the parent completes, leaving the child to run independently.
DeprovisionTRUWorkflow implementation with sleep timer
The DeprovisionTRUWorkflow accepts a DeprovisionTRUInput containing the namespace and MinutesToProvision duration. It calls workflow.Sleep(ctx, time.Duration(input.MinutesToProvision)*time.Minute) to wait for the TTL duration without consuming Worker memory. After the sleep completes, it executes the RemoveTRUs Activity with a 2-minute StartToCloseTimeout and the same retry policy as the provisioning workflow. The workflow returns the error from the Activity execution.
Service handler for capacity provisioning requests
The HandleProvisionRequest function checks if a provisioning workflow is already running using c.GetWorkflow() to retrieve the workflow run ID. If one exists, it returns an error. It also checks for a pre-existing deprovisioning workflow and cancels it if found. It creates client.StartWorkflowOptions with a generated ID (format: provision-{namespace}) and TaskQueue (capacity-management), then calls c.ExecuteWorkflow without waiting for completion, returning immediately after the Temporal Service accepts the request.