diff --git a/docs/features/worker-sessions.md b/docs/features/worker-sessions.md index c30857f66bb..d1579350725 100644 --- a/docs/features/worker-sessions.md +++ b/docs/features/worker-sessions.md @@ -81,6 +81,10 @@ on a `schedule_activity` workflow-task command under `worker_session`. } ``` +## Python service workers {#python-service-workers} + +The [Python worker-session example](../polyglot/python.md#worker-sessions) creates `WorkerSessionOptions`, schedules two sequential remote activities, declares the holder capability and handles an exhausted activity failure. It also shows rebuilding disposable worker-local state after holder loss and explains why explicit close or TTL expiry needs a fresh session id. The recipe is checked with published Python SDK 2.9.0 and Server 2.11.0, including the Server's scheduler/maintenance role. + ## Lifecycle Session creation is lazy. The matching layer creates or reacquires a session diff --git a/docs/polyglot/python.md b/docs/polyglot/python.md index a9c9b61ac22..25f5b087a2c 100644 --- a/docs/polyglot/python.md +++ b/docs/polyglot/python.md @@ -1279,6 +1279,84 @@ Use `PassthroughWorkerInterceptor` as a base class when you only need one hook. Implement `WorkerInterceptor` directly when you want type checkers to force all hooks to be present. +### Worker sessions {#worker-sessions} + +Use `WorkerSessionOptions` when sequential remote activities need one worker's local resource. This example is checked with Python SDK **2.9.0** and Server **2.11.0**; see the [portable affinity version requirements](./portable-worker-affinity.md). Use a unique `job_id` for each workflow run so a new run does not reuse a terminal session id. + +The cache below is disposable. Each activity can rebuild it from the durable business input, so a replacement holder does not depend on a previous process's memory: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Register both activities and declare the required capability on the worker: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Start `inspect_document_workflow` on the `documents` task queue with `[job_id]` as input. The first activity lazily acquires the session; the second uses the same holder while its lease remains valid. `max_concurrent_activities=1` bounds simultaneous activities in that session. Queue routing and capability checks still apply. These short leases/timeouts demonstrate bounded behavior; size them for your real work and heartbeat well before expiry. + +After holder loss or lease expiry, `allow_reacquire_after_failure=True` permits another eligible worker to reacquire the session. Its local resource must be rebuilt. Already committed activity results replay, while uncommitted effects can repeat: use durable resource identifiers and idempotent external operations. A session does not migrate memory, files or open handles. + +Explicit close and absolute TTL expiry are terminal for that session id. A later activity cannot reopen it and can remain unadmitted until its configured timeout is enforced. Run the Server's scheduler/maintenance role from the [published deployment recipe](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); an API-only process is insufficient for timeout enforcement. The workflow handles exhausted activity failure with `ActivityFailed` and returns `resource_unavailable`; the application can then choose a fresh session and rebuild from the business input. This example does not promise an immediate session-closed exception or automatically reopen a terminal session. + +For explicit holder lifecycle operations, use `worker.worker_session(options)` and its `create()`, `renew()` and `close(reason)` methods. Graceful worker shutdown closes its held sessions. See the [shared lifecycle contract](../features/worker-sessions.md#lifecycle). + ## Logging The SDK uses Python's standard `logging` module with structured logger names: diff --git a/i18n/de/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/de/docusaurus-plugin-content-docs/current/features/worker-sessions.md index 5ebe42cd945..2e029c47094 100644 --- a/i18n/de/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/de/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -79,6 +79,10 @@ JSON-Form unter `worker_session` eines Workflow-Aufgabenbefehls } ``` +## Python-Service-Worker {#python-service-workers} + +Das [Python-Beispiel für Worker-Sessions](../polyglot/python.md#worker-sessions) erstellt `WorkerSessionOptions`, plant zwei aufeinanderfolgende Remote-Activities, deklariert die Capability des Inhabers und behandelt einen endgültigen Activity-Fehler. Es zeigt außerdem den Neuaufbau verwerfbaren lokalen Zustands nach dem Verlust des Inhabers und erklärt, warum explizites Schließen oder TTL-Ablauf eine neue Session-ID erfordert. Das Rezept wurde mit den veröffentlichten Versionen Python SDK 2.9.0 und Server 2.11.0 einschließlich der Scheduler-/Wartungsrolle des Servers geprüft. + ## Lebenszyklus {#lifecycle} Sitzungen werden bei Bedarf erstellt. Die Zuordnungsschicht erstellt oder diff --git a/i18n/de/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/de/docusaurus-plugin-content-docs/current/polyglot/python.md index 874300689f6..d6dd6bf11cd 100644 --- a/i18n/de/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/de/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1257,6 +1257,84 @@ Verwenden Sie `PassthroughWorkerInterceptor` als Basisklasse für nur einen Hook. Implementieren Sie `WorkerInterceptor` direkt, wenn Typprüfer das Vorhandensein aller Hooks erzwingen sollen. +### Worker-Sessions {#worker-sessions} + +Verwenden Sie `WorkerSessionOptions`, wenn aufeinanderfolgende Remote-Activities eine lokale Ressource desselben Workers brauchen. Dieses Beispiel wurde mit Python SDK **2.9.0** und Server **2.11.0** geprüft; siehe die [Versionsanforderungen für portable Affinität](./portable-worker-affinity.md). Verwenden Sie pro Workflow-Ausführung eine eindeutige `job_id`, damit ein neuer Lauf keine Session-ID im Endzustand wiederverwendet. + +Der folgende Cache ist verwerfbar. Jede Activity kann ihn aus dauerhaft gespeicherten Geschäftseingaben neu aufbauen. Ein neuer Inhaber hängt daher nicht vom Speicher des vorherigen Prozesses ab: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Registrieren Sie beide Activities und deklarieren Sie die erforderliche Capability am Worker: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Starten Sie `inspect_document_workflow` in der Task-Queue `documents` mit `[job_id]` als Eingabe. Die erste Activity erwirbt die Session bei Bedarf; die zweite nutzt denselben Inhaber, solange die Lease gültig ist. `max_concurrent_activities=1` begrenzt gleichzeitig laufende Activities in dieser Session. Queue-Routing und Capability-Prüfungen gelten weiterhin. Diese kurzen Leases und Timeouts zeigen begrenztes Verhalten; passen Sie sie an die tatsächliche Arbeit an und senden Sie Heartbeats deutlich vor Ablauf. + +Nach dem Verlust des Inhabers oder Ablauf der Lease erlaubt `allow_reacquire_after_failure=True` einem anderen geeigneten Worker, die Session erneut zu erwerben. Er muss die lokale Ressource neu aufbauen. Bereits bestätigte Activity-Ergebnisse werden wiedergegeben, während unbestätigte Effekte sich wiederholen können: Verwenden Sie dauerhafte Ressourcenkennungen und idempotente externe Operationen. Eine Session migriert weder Speicher noch Dateien oder offene Handles. + +Explizites Schließen und der Ablauf der absoluten TTL sind für diese Session-ID endgültig. Eine spätere Activity kann sie nicht erneut öffnen und bleibt möglicherweise bis zur Durchsetzung ihres konfigurierten Timeouts unzugelassen. Führen Sie die Scheduler-/Wartungsrolle des Servers aus dem [veröffentlichten Deployment-Rezept](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml) aus; ein reiner API-Prozess reicht zur Timeout-Durchsetzung nicht aus. Der Workflow behandelt den endgültigen Activity-Fehler mit `ActivityFailed` und gibt `resource_unavailable` zurück. Die Anwendung kann dann eine neue Session wählen und die Ressource aus den Geschäftseingaben neu aufbauen. Das Beispiel verspricht weder eine sofortige Ausnahme wegen einer geschlossenen Session noch ein automatisches Wiederöffnen einer endgültig beendeten Session. + +Für explizite Operationen am Lebenszyklus des Inhabers verwenden Sie `worker.worker_session(options)` und dessen Methoden `create()`, `renew()` und `close(reason)`. Beim geordneten Herunterfahren schließt der Worker seine gehaltenen Sessions. Siehe den [gemeinsamen Lebenszyklusvertrag](../features/worker-sessions.md#lifecycle). + ## Logging {#logging} Das SDK verwendet Pythons Standardmodul `logging` mit strukturierten Logger-Namen: diff --git a/i18n/de/source-hashes.json b/i18n/de/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/de/source-hashes.json +++ b/i18n/de/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/es/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/es/docusaurus-plugin-content-docs/current/features/worker-sessions.md index 764448740a6..c2a711fb272 100644 --- a/i18n/es/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/es/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -71,6 +71,10 @@ La instantánea de opciones guarda el mismo contrato en `activity_options.worker } ``` +## Workers de servicio en Python {#python-service-workers} + +El [ejemplo de sesiones de worker en Python](../polyglot/python.md#worker-sessions) crea `WorkerSessionOptions`, programa dos actividades remotas secuenciales, declara la capacidad del titular y maneja el fallo definitivo de actividad. También muestra cómo reconstruir el estado local desechable tras perder al titular y explica por qué el cierre explícito o la expiración del TTL requieren un identificador de sesión nuevo. La receta se ha comprobado con Python SDK 2.9.0 y Server 2.11.0 publicados, incluido el rol de planificador/mantenimiento del Server. + ## Ciclo de vida {#lifecycle} La creación es diferida. La capa de asignación crea o readquiere la sesión cuando la primera tarea de actividad de la sesión se admite en un worker capaz. Un worker también puede crearla explícitamente antes de consultar tareas mediante `POST /api/worker/sessions`. diff --git a/i18n/es/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/es/docusaurus-plugin-content-docs/current/polyglot/python.md index a24979805a2..0d3a350960b 100644 --- a/i18n/es/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/es/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1226,6 +1226,84 @@ worker = Worker( Usa `PassthroughWorkerInterceptor` como base si solo necesitas un hook. Implementa `WorkerInterceptor` directamente para que la comprobación de tipos exija todos. +### Sesiones de worker {#worker-sessions} + +Use `WorkerSessionOptions` cuando varias actividades remotas secuenciales necesiten un recurso local del mismo worker. Este ejemplo se ha comprobado con Python SDK **2.9.0** y Server **2.11.0**; consulte los [requisitos de versión de afinidad portable](./portable-worker-affinity.md). Use un `job_id` único por ejecución del workflow para que una ejecución nueva no reutilice el identificador de una sesión terminal. + +La caché siguiente es desechable. Cada actividad puede reconstruirla a partir de la entrada de negocio duradera, de modo que el nuevo titular no dependa de la memoria del proceso anterior: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Registre ambas actividades y declare la capacidad requerida en el worker: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Inicie `inspect_document_workflow` en la cola de tareas `documents` con `[job_id]` como entrada. La primera actividad adquiere la sesión de forma diferida; la segunda usa el mismo titular mientras su concesión siga vigente. `max_concurrent_activities=1` limita las actividades simultáneas de esa sesión. El enrutamiento de cola y las comprobaciones de capacidad siguen aplicándose. Estas concesiones y timeouts breves demuestran un comportamiento acotado; ajústelos al trabajo real y envíe heartbeats con suficiente antelación a la expiración. + +Tras perder al titular o expirar la concesión, `allow_reacquire_after_failure=True` permite que otro worker apto readquiera la sesión. Debe reconstruir el recurso local. Los resultados de actividad ya confirmados se reproducen, pero los efectos no confirmados pueden repetirse: use identificadores duraderos e interacciones externas idempotentes. Una sesión no migra memoria, archivos ni descriptores abiertos. + +El cierre explícito y la expiración del TTL absoluto son terminales para ese identificador de sesión. Una actividad posterior no puede reabrirlo y puede permanecer sin admisión hasta que se aplique el timeout configurado. Ejecute el rol de planificador/mantenimiento del Server de la [receta de despliegue publicada](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); un proceso solo de API no basta para aplicar los timeouts. El workflow maneja el fallo definitivo de actividad con `ActivityFailed` y devuelve `resource_unavailable`; la aplicación puede elegir una sesión nueva y reconstruir a partir de la entrada de negocio. Este ejemplo no promete una excepción inmediata de sesión cerrada ni reabre automáticamente una sesión terminal. + +Para gestionar explícitamente el ciclo de vida del titular, use `worker.worker_session(options)` y sus métodos `create()`, `renew()` y `close(reason)`. El apagado ordenado del worker cierra sus sesiones. Consulte el [contrato de ciclo de vida compartido](../features/worker-sessions.md#lifecycle). + ## Registros {#logging} El SDK usa `logging` estándar con nombres estructurados: diff --git a/i18n/es/source-hashes.json b/i18n/es/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/es/source-hashes.json +++ b/i18n/es/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/fr/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/fr/docusaurus-plugin-content-docs/current/features/worker-sessions.md index e9762f0fabc..cbc81c4d755 100644 --- a/i18n/fr/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/fr/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -84,6 +84,10 @@ sous `worker_session`. } ``` +## Workers de service Python {#python-service-workers} + +L’[exemple de session de worker Python](../polyglot/python.md#worker-sessions) crée `WorkerSessionOptions`, planifie deux activités distantes séquentielles, déclare la capacité du détenteur et traite un échec définitif d’activité. Il montre aussi comment reconstruire l’état local jetable après la perte du détenteur et explique pourquoi une fermeture explicite ou l’expiration du TTL impose un nouvel identifiant de session. La recette est vérifiée avec les versions publiées du SDK Python 2.9.0 et de Server 2.11.0, y compris le rôle d’ordonnancement/maintenance du Server. + ## Cycle de vie {#lifecycle} La création de session est différée. La couche d’appariement crée ou réacquiert diff --git a/i18n/fr/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/fr/docusaurus-plugin-content-docs/current/polyglot/python.md index 17e03bb44e5..c80bcbe39a9 100644 --- a/i18n/fr/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/fr/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1254,6 +1254,84 @@ Utilisez `PassthroughWorkerInterceptor` comme base pour un seul hook. Implément directement `WorkerInterceptor` si les vérificateurs de types doivent imposer la présence de tous les hooks. +### Sessions de worker {#worker-sessions} + +Utilisez `WorkerSessionOptions` lorsque des activités distantes séquentielles doivent utiliser une ressource locale du même worker. Cet exemple est vérifié avec le SDK Python **2.9.0** et Server **2.11.0** ; consultez les [versions requises pour l’affinité portable](./portable-worker-affinity.md). Utilisez un `job_id` unique par exécution de workflow afin qu’une nouvelle exécution ne réutilise pas l’identifiant d’une session terminale. + +Le cache ci-dessous est jetable. Chaque activité peut le reconstruire à partir de l’entrée métier durable ; le nouveau détenteur ne dépend donc pas de la mémoire du processus précédent : + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Enregistrez les deux activités et déclarez la capacité requise sur le worker : + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Démarrez `inspect_document_workflow` dans la file de tâches `documents` avec `[job_id]` en entrée. La première activité acquiert la session à la demande ; la seconde utilise le même détenteur tant que son bail reste valide. `max_concurrent_activities=1` limite les activités simultanées dans cette session. Le routage de file et les contrôles de capacité s’appliquent toujours. Ces baux et timeouts courts illustrent un comportement borné ; dimensionnez-les pour le travail réel et envoyez les heartbeats bien avant l’expiration. + +Après la perte du détenteur ou l’expiration du bail, `allow_reacquire_after_failure=True` permet à un autre worker admissible de réacquérir la session. Il doit reconstruire la ressource locale. Les résultats d’activité déjà validés sont relus, mais les effets non validés peuvent se répéter : utilisez des identifiants durables et des opérations externes idempotentes. Une session ne migre ni mémoire, ni fichiers, ni descripteurs ouverts. + +La fermeture explicite et l’expiration du TTL absolu sont terminales pour cet identifiant de session. Une activité ultérieure ne peut pas le rouvrir et peut rester sans admission jusqu’à l’application du timeout configuré. Exécutez le rôle d’ordonnancement/maintenance du Server de la [recette de déploiement publiée](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml) ; un processus API seul ne suffit pas à appliquer les timeouts. Le workflow traite l’échec définitif d’activité avec `ActivityFailed` et renvoie `resource_unavailable` ; l’application peut ensuite choisir une nouvelle session et reconstruire depuis l’entrée métier. Cet exemple ne promet ni exception immédiate de session fermée, ni réouverture automatique d’une session terminale. + +Pour gérer explicitement le cycle de vie du détenteur, utilisez `worker.worker_session(options)` et ses méthodes `create()`, `renew()` et `close(reason)`. L’arrêt normal du worker ferme les sessions qu’il détient. Consultez le [contrat de cycle de vie commun](../features/worker-sessions.md#lifecycle). + ## Journalisation {#logging} Le SDK utilise le module standard `logging` avec des noms de logger structurés : diff --git a/i18n/fr/source-hashes.json b/i18n/fr/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/fr/source-hashes.json +++ b/i18n/fr/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/ja/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/ja/docusaurus-plugin-content-docs/current/features/worker-sessions.md index 92958ce8425..c1218fcd32e 100644 --- a/i18n/ja/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/ja/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -71,6 +71,10 @@ $manifest = $session->activity(AssembleManifestActivity::class, $frames['path']) } ``` +## Python のサービス Worker {#python-service-workers} + +[Python の Worker セッション例](../polyglot/python.md#worker-sessions)では、`WorkerSessionOptions` を作成し、2 つのリモートアクティビティを順番にスケジュールし、保持者の capability を宣言して、試行を使い切ったアクティビティの失敗を処理します。保持者の喪失後に破棄可能な Worker ローカル状態を再構築する方法と、明示的なクローズや TTL 満了後に新しいセッション ID が必要な理由も説明します。この手順は公開済みの Python SDK 2.9.0 と Server 2.11.0 で検証済みで、Server のスケジューラー/メンテナンスロールも含みます。 + ## ライフサイクル {#lifecycle} セッションは遅延作成されます。最初のセッション内タスクが要件を満たす Worker に受け入れられると、マッチング層が作成または再取得します。Worker はポーリング前に `POST /api/worker/sessions` で明示的に作成することもできます。 diff --git a/i18n/ja/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/ja/docusaurus-plugin-content-docs/current/polyglot/python.md index 3ee4320f025..2e6c25763e7 100644 --- a/i18n/ja/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/ja/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1041,6 +1041,84 @@ worker = Worker( 1フックだけが必要な場合は `PassthroughWorkerInterceptor` を基底クラスにします。型チェッカーですべてのフックの存在を強制したい場合は `WorkerInterceptor` を直接実装します。 +### Worker セッション {#worker-sessions} + +複数のリモートアクティビティが順番に同じ Worker のローカルリソースを使う場合は、`WorkerSessionOptions` を使用します。この例は Python SDK **2.9.0** と Server **2.11.0** で検証済みです。[ポータブルなアフィニティのバージョン要件](./portable-worker-affinity.md)も参照してください。ワークフローの各実行で一意の `job_id` を使い、新しい実行で終端状態のセッション ID を再利用しないようにします。 + +以下のキャッシュは破棄可能です。各アクティビティは永続化された業務入力から再構築できるため、交代した保持者が前のプロセスのメモリに依存することはありません。 + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +両方のアクティビティを登録し、Worker で必要な capability を宣言します。 + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +`documents` タスクキューで、入力を `[job_id]` として `inspect_document_workflow` を開始します。最初のアクティビティが必要に応じてセッションを取得し、リースが有効な間は次のアクティビティも同じ保持者を使います。`max_concurrent_activities=1` は、そのセッション内で同時実行するアクティビティを制限します。キューのルーティングと capability の検査も適用されます。この短いリースとタイムアウトは有限の動作を示すためのものです。実際の作業に合わせて設定し、期限より十分前にハートビートを送信してください。 + +保持者の喪失やリース切れの後は、`allow_reacquire_after_failure=True` により、条件を満たす別の Worker がセッションを再取得できます。その Worker はローカルリソースを再構築する必要があります。確定済みのアクティビティ結果はリプレイされますが、未確定の副作用は繰り返される可能性があります。永続的なリソース識別子と冪等な外部操作を使用してください。セッションはメモリ、ファイル、開いたハンドルを移行しません。 + +明示的なクローズと絶対 TTL の満了は、そのセッション ID に対する終端状態です。後続のアクティビティは再度開けず、設定したタイムアウトが適用されるまで実行が許可されない場合があります。[公開済みのデプロイ手順](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml)にある Server のスケジューラー/メンテナンスロールを実行してください。API プロセスだけではタイムアウトは適用されません。ワークフローは試行を使い切ったアクティビティの失敗を `ActivityFailed` で処理し、`resource_unavailable` を返します。その後、アプリケーションは新しいセッションを選び、業務入力からリソースを再構築できます。この例は、セッション終了例外の即時送出や終端セッションの自動再開を保証しません。 + +保持者のライフサイクルを明示的に操作するには、`worker.worker_session(options)` とその `create()`、`renew()`、`close(reason)` を使用します。Worker の正常なシャットダウンでは、保持中のセッションを閉じます。[共通のライフサイクル契約](../features/worker-sessions.md#lifecycle)を参照してください。 + ## ログ {#logging} SDK は構造化したロガー名で Python 標準の `logging` モジュールを使います。 diff --git a/i18n/ja/source-hashes.json b/i18n/ja/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/ja/source-hashes.json +++ b/i18n/ja/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/pt-BR/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/pt-BR/docusaurus-plugin-content-docs/current/features/worker-sessions.md index 5bc01c2d177..bb8126d3042 100644 --- a/i18n/pt-BR/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/pt-BR/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -82,6 +82,10 @@ JSON em `worker_session` de um comando de tarefa de workflow `schedule_activity` } ``` +## Workers de serviço em Python {#python-service-workers} + +O [exemplo de sessão de worker em Python](../polyglot/python.md#worker-sessions) cria `WorkerSessionOptions`, agenda duas atividades remotas sequenciais, declara a capacidade do titular e trata uma falha definitiva de atividade. Ele também mostra como reconstruir estado local descartável após a perda do titular e explica por que o fechamento explícito ou a expiração do TTL exigem um identificador de sessão novo. A receita foi verificada com Python SDK 2.9.0 e Server 2.11.0 publicados, incluindo a função de agendamento/manutenção do Server. + ## Ciclo de vida {#lifecycle} A criação da sessão é adiada. A camada de correspondência cria ou readquire diff --git a/i18n/pt-BR/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/pt-BR/docusaurus-plugin-content-docs/current/polyglot/python.md index eaef30e8aef..a8a82d8f13d 100644 --- a/i18n/pt-BR/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/pt-BR/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1254,6 +1254,84 @@ Use `PassthroughWorkerInterceptor` como classe base se precisar de um único hook. Implemente `WorkerInterceptor` diretamente quando quiser que os verificadores de tipo exijam todos os hooks. +### Sessões de worker {#worker-sessions} + +Use `WorkerSessionOptions` quando atividades remotas sequenciais precisarem de um recurso local do mesmo worker. Este exemplo foi verificado com Python SDK **2.9.0** e Server **2.11.0**; consulte os [requisitos de versão da afinidade portátil](./portable-worker-affinity.md). Use um `job_id` único por execução do workflow para que uma nova execução não reutilize o identificador de uma sessão terminal. + +O cache abaixo é descartável. Cada atividade pode reconstruí-lo a partir da entrada de negócio durável, de modo que o novo titular não dependa da memória do processo anterior: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Registre as duas atividades e declare a capacidade exigida no worker: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Inicie `inspect_document_workflow` na fila de tarefas `documents` com `[job_id]` como entrada. A primeira atividade adquire a sessão de forma tardia; a segunda usa o mesmo titular enquanto o lease estiver válido. `max_concurrent_activities=1` limita as atividades simultâneas nessa sessão. O roteamento de fila e as verificações de capacidade continuam valendo. Estes leases/timeouts curtos demonstram comportamento limitado; dimensione-os para o trabalho real e envie heartbeats bem antes da expiração. + +Após a perda do titular ou a expiração do lease, `allow_reacquire_after_failure=True` permite que outro worker apto readquira a sessão. Ele precisa reconstruir o recurso local. Resultados de atividade já confirmados são reproduzidos, mas efeitos não confirmados podem se repetir: use identificadores duráveis e operações externas idempotentes. Uma sessão não migra memória, arquivos ou handles abertos. + +O fechamento explícito e a expiração do TTL absoluto são terminais para esse identificador de sessão. Uma atividade posterior não pode reabri-lo e pode continuar sem admissão até que o timeout configurado seja aplicado. Execute a função de agendamento/manutenção do Server da [receita de implantação publicada](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); um processo somente de API não basta para aplicar timeouts. O workflow trata a falha definitiva de atividade com `ActivityFailed` e retorna `resource_unavailable`; a aplicação pode então escolher uma sessão nova e reconstruir a partir da entrada de negócio. O exemplo não promete uma exceção imediata de sessão fechada nem reabre automaticamente uma sessão terminal. + +Para operações explícitas do ciclo de vida do titular, use `worker.worker_session(options)` e os métodos `create()`, `renew()` e `close(reason)`. O encerramento normal do worker fecha as sessões que ele mantém. Consulte o [contrato de ciclo de vida compartilhado](../features/worker-sessions.md#lifecycle). + ## Logging {#logging} O SDK usa o módulo padrão `logging` do Python com nomes estruturados de logger: diff --git a/i18n/pt-BR/source-hashes.json b/i18n/pt-BR/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/pt-BR/source-hashes.json +++ b/i18n/pt-BR/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/uk/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/uk/docusaurus-plugin-content-docs/current/features/worker-sessions.md index f4c4022a330..786262694aa 100644 --- a/i18n/uk/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/uk/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -80,6 +80,10 @@ $manifest = $session->activity(AssembleManifestActivity::class, $frames['path']) } ``` +## Сервісні worker на Python {#python-service-workers} + +[Приклад worker session на Python](../polyglot/python.md#worker-sessions) створює `WorkerSessionOptions`, планує дві послідовні віддалені activity, оголошує capability власника й обробляє остаточну помилку activity. Він також показує відновлення тимчасового локального стану після втрати власника та пояснює, чому явне закриття чи завершення TTL потребує нового ідентифікатора сесії. Рецепт перевірено з опублікованими Python SDK 2.9.0 і Server 2.11.0, включно з роллю scheduler/maintenance Server. + ## Життєвий цикл {#lifecycle} Сесія створюється ліниво. Шар підбору завдань створює або повторно отримує diff --git a/i18n/uk/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/uk/docusaurus-plugin-content-docs/current/polyglot/python.md index 5eaa468ffc6..7cd24faa8ee 100644 --- a/i18n/uk/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/uk/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1237,6 +1237,84 @@ worker = Worker( Реалізуйте `WorkerInterceptor` напряму, якщо перевірка типів має вимагати наявності всіх hook. +### Worker sessions {#worker-sessions} + +Використовуйте `WorkerSessionOptions`, коли послідовним віддаленим activity потрібен локальний ресурс одного worker. Приклад перевірено з Python SDK **2.9.0** і Server **2.11.0**; див. [вимоги версій для переносимої прив’язки](./portable-worker-affinity.md). Задавайте унікальний `job_id` для кожного запуску workflow, щоб новий запуск не використовував завершений ідентифікатор сесії. + +Кеш нижче є тимчасовим. Кожна activity може відновити його з надійно збережених бізнес-вхідних даних, тому новий власник не залежить від пам’яті попереднього процесу: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +Зареєструйте обидві activity й оголосіть потрібну capability на worker: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +Запустіть `inspect_document_workflow` у task queue `documents`, передавши `[job_id]`. Перша activity ліниво отримує сесію; друга використовує того самого власника, доки lease дійсний. `max_concurrent_activities=1` обмежує одночасні activity в сесії. Маршрутизація черги й перевірки capability залишаються чинними. Короткі lease і таймаути демонструють обмежену поведінку; підбирайте їх для реальної роботи й надсилайте heartbeat задовго до завершення строку. + +Після втрати власника або завершення lease параметр `allow_reacquire_after_failure=True` дозволяє іншому придатному worker отримати сесію повторно. Він має відновити локальний ресурс. Уже зафіксовані результати activity відтворюються, але незафіксовані ефекти можуть повторитися: використовуйте надійні ідентифікатори ресурсів та ідемпотентні зовнішні операції. Сесія не переносить пам’ять, файли чи відкриті дескриптори. + +Явне закриття та завершення абсолютного TTL є остаточними для цього ідентифікатора сесії. Наступна activity не відкриє його повторно й може залишатися недопущеною, доки runtime не застосує налаштований таймаут. Запускайте роль scheduler/maintenance Server з [опублікованого рецепта розгортання](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); лише API-процесу недостатньо для застосування таймаутів. Workflow обробляє остаточну помилку activity через `ActivityFailed` і повертає `resource_unavailable`; тоді застосунок може вибрати нову сесію й відновити ресурс із бізнес-вхідних даних. Приклад не обіцяє негайного винятку закритої сесії чи автоматичного відкриття завершеної сесії. + +Для явного життєвого циклу власника використовуйте `worker.worker_session(options)` та методи `create()`, `renew()` і `close(reason)`. Коректне завершення worker закриває утримувані ним сесії. Див. [спільний контракт життєвого циклу](../features/worker-sessions.md#lifecycle). + ## Журналювання {#logging} SDK використовує стандартний модуль `logging` Python зі структурованими назвами: diff --git a/i18n/uk/source-hashes.json b/i18n/uk/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/uk/source-hashes.json +++ b/i18n/uk/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83", diff --git a/i18n/zh-Hans/docusaurus-plugin-content-docs/current/features/worker-sessions.md b/i18n/zh-Hans/docusaurus-plugin-content-docs/current/features/worker-sessions.md index 952e330b6e8..5d65e0ab093 100644 --- a/i18n/zh-Hans/docusaurus-plugin-content-docs/current/features/worker-sessions.md +++ b/i18n/zh-Hans/docusaurus-plugin-content-docs/current/features/worker-sessions.md @@ -71,6 +71,10 @@ $manifest = $session->activity(AssembleManifestActivity::class, $frames['path']) } ``` +## Python 服务 Worker {#python-service-workers} + +[Python Worker 会话示例](../polyglot/python.md#worker-sessions)创建 `WorkerSessionOptions`,安排两个顺序执行的远程活动,声明持有者能力,并处理重试耗尽后的活动失败。它还展示了持有者丢失后如何重建可丢弃的 Worker 本地状态,并解释为何显式关闭或 TTL 到期后需要新会话 ID。该方案已使用发布的 Python SDK 2.9.0 和 Server 2.11.0 验证,包括 Server 调度器/维护角色。 + ## 生命周期 {#lifecycle} 会话按需创建。首个会话内活动任务被具备所需能力的 Worker 接纳时,匹配层创建或重新获取会话。Worker 也可以在轮询前调用 `POST /api/worker/sessions` 显式创建会话。 diff --git a/i18n/zh-Hans/docusaurus-plugin-content-docs/current/polyglot/python.md b/i18n/zh-Hans/docusaurus-plugin-content-docs/current/polyglot/python.md index 16a8ccf6cf1..f954b1e893e 100644 --- a/i18n/zh-Hans/docusaurus-plugin-content-docs/current/polyglot/python.md +++ b/i18n/zh-Hans/docusaurus-plugin-content-docs/current/polyglot/python.md @@ -1022,6 +1022,84 @@ worker = Worker( 仅需一个钩子时,将 `PassthroughWorkerInterceptor` 作为基类。希望类型检查器强制提供全部钩子时,直接实现 `WorkerInterceptor`。 +### Worker 会话 {#worker-sessions} + +当多个顺序执行的远程活动需要同一个 Worker 的本地资源时,使用 `WorkerSessionOptions`。本示例已使用 Python SDK **2.9.0** 和 Server **2.11.0** 验证;参见[可移植亲和性的版本要求](./portable-worker-affinity.md)。每次工作流运行应使用唯一的 `job_id`,避免新运行重复使用已终止的会话 ID。 + +下方缓存是可丢弃的。每个活动都能根据持久保存的业务输入重建缓存,因此替代持有者不依赖上一个进程的内存: + +```python +from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow + +# Disposable process-local state. Rebuild it from durable business input. +documents = {} + +def ensure_document(job_id): + rebuilt = job_id not in documents + if rebuilt: + documents[job_id] = {"text": "document:" + job_id} + return documents[job_id], rebuilt + +@activity.defn(name="prepare_document") +async def prepare_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt} + +@activity.defn(name="inspect_document") +async def inspect_document(job_id): + document, rebuilt = ensure_document(job_id) + return {"holder": activity.context().info.worker_id, + "rebuilt": rebuilt, "text": document["text"]} + +@workflow.defn(name="inspect_document_workflow") +class InspectDocumentWorkflow: + def run(self, ctx, job_id): + session = WorkerSessionOptions( + session_id="documents:" + job_id, + requirements=("fs:documents",), + lease_seconds=5, + ttl_seconds=60, + max_concurrent_activities=1, + allow_reacquire_after_failure=True, + ) + options = dict(worker_session=session, + start_to_close_timeout=5, + schedule_to_start_timeout=12, + schedule_to_close_timeout=20, + retry_policy={"max_attempts": 1}) + try: + prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options) + inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options) + except ActivityFailed as error: + if error.activity_type not in {"prepare_document", "inspect_document"}: + raise + return {"status": "resource_unavailable", "job_id": job_id} + return {"status": "inspected", "prepared": prepared, "inspected": inspected} +``` + +注册两个活动,并在 Worker 上声明所需能力: + +```python +from durable_workflow import Worker + +worker = Worker( + client, + task_queue="documents", + workflows=[InspectDocumentWorkflow], + activities=[prepare_document, inspect_document], + capabilities=["fs:documents"], +) +await worker.run() +``` + +在 `documents` 任务队列启动 `inspect_document_workflow`,输入为 `[job_id]`。第一个活动按需获取会话;只要租约有效,第二个活动就使用同一持有者。`max_concurrent_activities=1` 限制该会话内同时执行的活动数。队列路由和能力检查仍然生效。这些较短的租约与超时用于演示有界行为;请按实际工作调整,并在到期之前留出充足时间发送心跳。 + +持有者丢失或租约到期后,`allow_reacquire_after_failure=True` 允许另一个符合条件的 Worker 重新获取会话。它必须重建本地资源。已提交的活动结果会被重放,而未提交的副作用可能重复:请使用持久资源标识和幂等的外部操作。会话不会迁移内存、文件或已打开的句柄。 + +显式关闭和绝对 TTL 到期对该会话 ID 都是终态。后续活动无法重新打开它,可能一直无法获准执行,直到运行时处理已配置的超时。请运行[已发布部署方案](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml)中的 Server 调度器/维护角色;仅有 API 进程不足以执行超时处理。工作流通过 `ActivityFailed` 处理重试耗尽后的活动失败,并返回 `resource_unavailable`;应用随后可以选择新会话,根据业务输入重建资源。本示例不承诺立即抛出会话已关闭异常,也不会自动重新打开终态会话。 + +需要显式管理持有者生命周期时,使用 `worker.worker_session(options)` 及其 `create()`、`renew()`、`close(reason)` 方法。Worker 正常关闭时会关闭其持有的会话。参见[共享生命周期约定](../features/worker-sessions.md#lifecycle)。 + ## 日志 {#logging} SDK 使用 Python 标准 `logging` 模块及结构化日志器名称: diff --git a/i18n/zh-Hans/source-hashes.json b/i18n/zh-Hans/source-hashes.json index d08bd4bdab9..ad4fc84a1e5 100644 --- a/i18n/zh-Hans/source-hashes.json +++ b/i18n/zh-Hans/source-hashes.json @@ -55,7 +55,7 @@ "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06", "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3", "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8", - "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d", + "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c", "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb", "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967", "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3", @@ -82,7 +82,7 @@ "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989", "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959", "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c", - "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648", + "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028", "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3", "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58", "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83",