save

Author: YunchuWang

durabletask-azuremanaged/CHANGELOG.md

@@ -13,6 +13,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   worker profiles, and running sandbox activity workers.
 - Added optional activity versions to sandbox worker profile declarations and
   sandbox worker registration.
+- Required sandbox workers to fail fast when `DTS_SANDBOX_ID` is missing,
+  treated `DTS_SANDBOX_PROVIDER` as optional metadata, and added context
+  manager support to `SandboxActivitiesClient`.
 
 ## v1.5.0
 

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py

@@ -41,6 +41,12 @@ def __init__(
     def close(self) -> None:
         self._transport.close()
 
+    def __enter__(self) -> "SandboxActivitiesClient":
+        return self
+
+    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
+        self.close()
+
     def enable_sandbox_activities(self) -> None:
         """Declare all configured sandbox worker profiles with Durable Task Scheduler."""
         worker_profiles = build_sandbox_worker_profiles()

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker.py

@@ -7,6 +7,7 @@
 
 from typing import Iterator, Optional
 
+import grpc
 from azure.core.credentials import TokenCredential
 from azure.identity import ManagedIdentityCredential
 
@@ -45,6 +46,7 @@ def __init__(self) -> None:
         resolved_secure_channel = _resolve_secure_channel(resolved_host_address)
         resolved_token_credential = _resolve_token_credential()
         resolved_max_concurrent_activities = _resolve_max_concurrent_activities()
+        resolved_dts_sandbox_identifier = _resolve_dts_sandbox_identifier()
         resolved_sandbox_provider = _resolve_sandbox_provider()
         concurrency_options = ConcurrencyOptions(
             maximum_concurrent_activity_work_items=resolved_max_concurrent_activities)
@@ -66,7 +68,7 @@ def __init__(self) -> None:
         self._sandbox_activities: list[SandboxActivity] = []
         self._sandbox_max_activities = resolved_max_concurrent_activities
         self._sandbox_provider = resolved_sandbox_provider
-        self._sandbox_dts_sandbox_identifier = os.getenv("DTS_SANDBOX_ID")
+        self._sandbox_dts_sandbox_identifier = resolved_dts_sandbox_identifier
         self._sandbox_heartbeat_interval_seconds = 2.0
         self._sandbox_registration_stop = threading.Event()
         self._sandbox_registration_thread: Optional[threading.Thread] = None
@@ -143,6 +145,11 @@ def _run_sandbox_registration_loop(self) -> None:
             except Exception as ex:
                 if self._sandbox_registration_stop.is_set():
                     break
+                if not _is_retriable_registration_failure(ex):
+                    self._sandbox_logger.error(
+                        "Sandbox activity worker registration failed permanently: %s", ex)
+                    self._sandbox_registration_stop.set()
+                    break
                 self._sandbox_logger.warning("Sandbox activity worker registration failed: %s", ex)
                 delay = random.uniform(0, retry_delay)
                 self._sandbox_registration_stop.wait(delay)
@@ -216,19 +223,59 @@ def _resolve_token_credential() -> TokenCredential | None:
     return ManagedIdentityCredential(client_id=client_id.strip())
 
 
-def _resolve_sandbox_provider() -> str:
+def _resolve_sandbox_provider() -> Optional[str]:
     sandbox_provider = os.getenv("DTS_SANDBOX_PROVIDER")
-    if not sandbox_provider:
+    if not sandbox_provider or not sandbox_provider.strip():
+        return None
+    return sandbox_provider.strip()
+
+
+def _resolve_dts_sandbox_identifier() -> str:
+    dts_sandbox_identifier = os.getenv("DTS_SANDBOX_ID")
+    if not dts_sandbox_identifier or not dts_sandbox_identifier.strip():
         raise ValueError(
-            "Sandbox worker requires DTS_SANDBOX_PROVIDER to be injected in the "
+            "Sandbox worker requires DTS_SANDBOX_ID to be injected in the "
             "sandbox environment.")
 
-    normalized = sandbox_provider.strip()
-    if normalized.lower() not in ("sandbox", "acasessionpool"):
-        raise ValueError(
-            "Sandbox worker requires DTS_SANDBOX_PROVIDER to be Sandbox or AcaSessionPool.")
+    return dts_sandbox_identifier.strip()
+
+
+_RETRIABLE_REGISTRATION_STATUS_CODES = {
+    grpc.StatusCode.CANCELLED,
+    grpc.StatusCode.DEADLINE_EXCEEDED,
+    grpc.StatusCode.INTERNAL,
+    grpc.StatusCode.RESOURCE_EXHAUSTED,
+    grpc.StatusCode.UNAVAILABLE,
+    grpc.StatusCode.UNKNOWN,
+}
+
+_PERMANENT_FAILED_PRECONDITION_DETAILS = (
+    "worker profile",
+    "registered activit",
+    "max activit",
+    "sandbox identifier",
+    "sandbox id",
+)
+
+
+def _is_retriable_registration_failure(ex: Exception) -> bool:
+    if isinstance(ex, grpc.RpcError):
+        status_code = ex.code()
+        if status_code in _RETRIABLE_REGISTRATION_STATUS_CODES:
+            return True
+        if status_code == grpc.StatusCode.FAILED_PRECONDITION:
+            details = _rpc_error_details(ex).casefold()
+            return not any(
+                detail in details
+                for detail in _PERMANENT_FAILED_PRECONDITION_DETAILS)
+        return False
+
+    return isinstance(ex, OSError)
+
 
-    return normalized
+def _rpc_error_details(ex: grpc.RpcError) -> str:
+    details = ex.details()
+    return details if isinstance(details, str) else ""
 
 
 def _resolve_max_concurrent_activities() -> int:

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker_messages.py

@@ -5,6 +5,7 @@
 
 from durabletask.azuremanaged.internal import sandbox_service_pb2 as pb
 from durabletask.azuremanaged.preview.sandboxes.helpers import SandboxActivity
+from durabletask.azuremanaged.preview.sandboxes.helpers import normalize_required
 from durabletask.azuremanaged.preview.sandboxes.helpers import resolve_activities
 
 
@@ -29,13 +30,17 @@ def build_sandbox_worker_start(
     if not resolved_activities:
         raise ValueError("Sandbox activity worker registration requires at least one registered activity.")
 
+    resolved_dts_sandbox_identifier = normalize_required(
+        dts_sandbox_identifier,
+        "Sandbox activity worker registration requires a DTS sandbox ID.")
+
     message = pb.SandboxActivityWorkerMessage(
         start=pb.SandboxActivityWorkerStart(
             task_hub=taskhub.strip(),
             worker_profile_id=worker_profile_id.strip(),
             max_activities_count=max_activities_count,
             sandbox_provider=_parse_sandbox_provider(sandbox_provider),
-            dts_sandbox_identifier=(dts_sandbox_identifier or "").strip()))
+            dts_sandbox_identifier=resolved_dts_sandbox_identifier))
     message.start.activities.extend([
         pb.SandboxActivity(name=activity.name, version=activity.version or "")
         for activity in resolved_activities
@@ -53,10 +58,11 @@ def build_sandbox_worker_heartbeat(active_activities_count: int) -> pb.SandboxAc
 
 
 def _parse_sandbox_provider(sandbox_provider: Optional[str]) -> "pb.SandboxProviderKind":
-    if not sandbox_provider:
+    if not sandbox_provider or not sandbox_provider.strip():
         return pb.SANDBOX_PROVIDER_KIND_UNSPECIFIED
-    if sandbox_provider.lower() == "sandbox":
+    normalized = sandbox_provider.strip().lower()
+    if normalized == "sandbox":
         return pb.SANDBOX_PROVIDER_KIND_SANDBOX
-    if sandbox_provider.lower() == "acasessionpool":
+    if normalized == "acasessionpool":
         return pb.SANDBOX_PROVIDER_KIND_ACA_SESSION_POOL
     return pb.SANDBOX_PROVIDER_KIND_UNSPECIFIED

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker_profiles.py

@@ -34,13 +34,6 @@ class SandboxWorkerProfileImageOptions:
 class SandboxWorkerProfileOptions:
     """Options for a decorated sandbox worker profile."""
 
-    @dataclass(frozen=True)
-    class Activity:
-        """Activity name and optional version for a sandbox worker profile."""
-
-        name: str
-        version: Optional[str]
-
     worker_profile_id: str
     image: SandboxWorkerProfileImageOptions = field(default_factory=SandboxWorkerProfileImageOptions)
     scheduler_managed_identity_client_id: str = ""
@@ -60,7 +53,7 @@ def add_activity(
             name=normalize_required(activity_name, "Sandbox activity name is required."),
             version=(version.strip() if version and version.strip() else None)))
 
-    def add_activities(self, activities: Iterable[Activity]) -> None:
+    def add_activities(self, activities: Iterable[SandboxActivity]) -> None:
         """Add activity names and versions to the sandbox worker profile."""
         for activity in activities:
             self.add_activity(activity.name, activity.version)

examples/sandboxes/README.md

@@ -51,14 +51,15 @@ metadata API can carry an optional activity version for future activity-version
 support, but this sample intentionally omits it. The declarer and remote worker
 both use `activities.py` so they stay in sync.
 
-The remote worker code cannot pass Durable Task Scheduler runtime settings to the SDK. In a
-sandbox, `SandboxWorker()` reads `DTS_ENDPOINT`,
-`DTS_TASK_HUB`, `DTS_WORKER_PROFILE_ID`, `DTS_SANDBOX_MAX_ACTIVITIES`,
-`DTS_SANDBOX_PROVIDER`, `DTS_AUTHENTICATION`, `DTS_UMI_CLIENT_ID`, and
-`DTS_SANDBOX_ID` from environment variables injected by Durable Task Scheduler.
-The worker requires `DTS_AUTHENTICATION=ManagedIdentity` and reports its
-registered activity identities when it connects. Durable Task Scheduler validates
-they match the worker_profile before advertising worker capacity.
+The remote worker code cannot pass Durable Task Scheduler runtime settings to
+the SDK. In a sandbox, `SandboxWorker()` reads `DTS_ENDPOINT`, `DTS_TASK_HUB`,
+`DTS_WORKER_PROFILE_ID`, `DTS_SANDBOX_MAX_ACTIVITIES`, `DTS_AUTHENTICATION`,
+`DTS_UMI_CLIENT_ID`, and `DTS_SANDBOX_ID` from environment variables injected by
+Durable Task Scheduler. It also reads optional `DTS_SANDBOX_PROVIDER` metadata
+when present. The worker requires `DTS_AUTHENTICATION=ManagedIdentity` and
+reports its sandbox ID plus registered activity identities when it connects.
+Durable Task Scheduler validates they match the worker_profile before
+advertising worker capacity.
 
 ## Build the remote worker image
 

examples/sandboxes/main_app.py

@@ -93,12 +93,12 @@ def configure(self, options) -> None:
 print(f"Using endpoint: {endpoint}")
 print(f"Declaring sandbox activity image: {container_image}")
 
-sandboxes_client = SandboxActivitiesClient(
-    host_address=endpoint,
-    secure_channel=secure_channel,
-    taskhub=taskhub_name,
-    token_credential=credential)
-sandboxes_client.enable_sandbox_activities()
+with SandboxActivitiesClient(
+        host_address=endpoint,
+        secure_channel=secure_channel,
+        taskhub=taskhub_name,
+        token_credential=credential) as sandboxes_client:
+    sandboxes_client.enable_sandbox_activities()
 
 with DurableTaskSchedulerWorker(
         host_address=endpoint,