Workers

Ce qu’est un worker

Un worker détient les corps des étapes réclamables et demande du travail au daemon. Il tourne là où tourne votre code, avec vos imports, votre pool de connexions et vos secrets, et le daemon n’en voit jamais rien. Le daemon lui remet un bail avec l’entrée évaluée et les étapes précédentes ; le worker exécute le corps et rend compte du déroulement. Toute la relation tient là.

Chaque SDK en fournit un. Il enregistre les workflows qu’il détient, réclame des baux en long polling, envoie des heartbeats pendant qu’un corps tourne, et répond à chaque refus selon son motif. L’exemple ci-dessous est le même programme que celui de la page Workflows ; en voici la partie qui le sert.

// Serve step bodies for as long as this process lives. The worker registers
// the definition, long-polls for leases, heartbeats while a body runs, and
// answers every 409 by its reason.
const worker = new Worker(mcpd, { concurrency: 8 }).register(support)
void worker.serve()

# Serve step bodies for as long as this process lives. serve() blocks, so it
# runs on a thread of its own when the process does other work too.
worker = Worker(mcpd, concurrency=8).register(support)
threading.Thread(target=worker.serve, daemon=True).start()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
let worker = Worker::builder(&mcpd).concurrency(8).register(support).build();
let serving = {
    let worker = worker.clone();
    tokio::spawn(async move { worker.serve().await })
};

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
worker := mcpd.NewWorker(m, mcpd.WorkerConcurrency(8)).Register(support)
go func() { _ = worker.Serve(ctx) }()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
val worker = Worker(mcpd, concurrency = 8).register(support)
launch { worker.serve() }

# Serve step bodies for as long as this node lives: register, long-poll for
# leases, heartbeat while a body runs, answer every 409 by its reason.
{:ok, worker} = StepWorker.start_link(mcpd, workflows: [support], concurrency: 8)
:ok = StepWorker.serve(worker)

Le protocole

Cinq appels, et un worker qui les parle peut s’écrire directement sur l’API HTTP, sans SDK. Le worker de chaque SDK est un client de ces cinq appels, et rien de plus.

Les baux

Un bail est une tentative, tenue par un worker, pour une durée bornée. Il porte l’entrée évaluée, les étapes précédentes indexées par identifiant, l’état et la corrélation de l’exécution, une clé stable d’une tentative à l’autre de cette étape, et les identifiants de trace et de span du span d’étape que le daemon a ouvert. Un corps qui enregistre ses propres spans par le contexte d’étape les place sous ce span, si bien que l’appel de modèle à l’intérieur d’une étape atterrit à la bonne profondeur dans la trace de l’exécution.

Un bail dont le worker cesse d’envoyer les heartbeats expire. Le daemon marque la tentative comme orpheline puis, passée une fenêtre de reprise, la remet en file sous at_least_once ou la fait échouer sous at_most_once. Un worker qui revient d’une pause et poste sur un bail repris n’est pas ignoré : on le lui dit.

Les quatre refus

Un appel de worker que le daemon ne peut pas accepter est refusé avec un 409 portant l’un des quatre motifs. La liste est fermée, et le worker de chaque SDK traite chaque motif par son nom, car un worker qui les réunit dans une seule branche abandonne des étapes qu’il a déjà terminées ou signale un échec pour une réussite.

Le corps

Un corps reçoit un contexte, son entrée évaluée et les étapes précédentes. Le contexte est l’endroit où il déclare ce que le daemon doit garder : un patch de l’état de l’exécution, fusionné au complete avec remplacement des clés de premier niveau ; les tokens qu’il a dépensés, que le daemon tarife avec la même grille que les spans ; et un signal d’annulation qui part quand le bail est perdu ou quand l’exécution se termine, pour qu’un corps qui le transmet à ses propres entrées-sorties s’arrête au bon moment. Les tokens s’additionnent d’eux-mêmes : un span que le corps enregistre par le contexte, et tout span sous lui, ajoute sa consommation à celle de l’étape en se terminant, si bien qu’un corps qui enregistre ses appels de modèle comme des spans n’a jamais de somme à faire à la main, et set_usage est la surcharge explicite pour celui qui n’en enregistre aucun.

Un corps se termine en rendant sa sortie ou en levant une erreur. Levez l’erreur réessayable du SDK pour être réessayé dans le budget de l’étape ; levez un échec d’étape typé pour nommer toute autre classe ; le reste est fatal. Le message est envoyé au daemon et affiché dans le journal. La pile d’appels, non.

Un bail portant sur une définition, une version ou une étape que le worker ne détient pas échoue en unservable et le corps ne tourne jamais. La tentative retourne dans la file pour un worker qui, lui, la détient, sans dépenser le budget de nouvelles tentatives, et c’est ce qui permet à deux versions d’un worker de tourner côte à côte pendant un déploiement.

Monter en charge

Faites tourner autant de workers que vous voulez sur un même daemon. Chacun réclame jusqu’à sa propre limite de concurrence, le daemon remet chaque tentative à exactement l’un d’eux, et le fencing fait qu’un bail que deux workers détiendraient malgré tout ne laisse qu’un gagnant. L’identifiant d’un worker est consigné sur chaque tentative qu’il a prise, donc le journal dit quel processus a exécuté une étape. Donnez un nom à chaque processus, ou laissez le SDK en dériver un depuis l’hôte.

Une réclamation est un long poll : une flotte au repos coûte au daemon une requête ouverte par worker et aucun trafic. Pendant qu’elle attend, une réclamation cherche du travail quand une exécution démarre, qu’une étape se termine, qu’une échéance se déclenche ou qu’un événement reprend une attente, et au plus deux fois par seconde le reste du temps. Une flotte chargée est bornée par la concurrence des workers eux-mêmes, et un corps qui prend une heure coûte un heartbeat toutes les dix secondes, et rien d’autre.

Les clés

Les appels de worker demandent la portée worker ; démarrer, signaler et annuler des exécutions demandent execute. Les deux retombent sur les clés d’écriture quand elles ne sont pas définies, et une clé admin porte les deux. Un worker qui ne fait que servir des étapes n’a besoin d’aucune clé capable de démarrer une exécution, et un webhook qui ne fait que publier des événements n’a besoin d’aucune clé capable d’en réclamer une. Voir Configuration pour les indicateurs.

Les appels de worker ne sont pas enregistrés dans le journal d’audit, bien qu’un bail porte l’entrée de l’étape et les sorties des étapes précédentes. Ils forment une boucle de machine qui réclame de nombreuses fois par minute, et une ligne par réclamation noierait les lectures faites par une personne. Relire une exécution par les endpoints d’exécution est enregistré ; voir Sécurité.

Ensuite

• Démarrer et suivre une exécution, et répondre à une attente : Exécutions.
• La définition que le worker sert : Workflows.
• Tous les endpoints du protocole des workers, générés depuis la spécification : API HTTP, Workers.