Worker · mcpd

Che cos’è un worker

Un worker tiene i corpi dei passi che si possono prendere in carico e chiede lavoro al daemon. Gira dove gira il tuo codice, con i tuoi import, il tuo pool di connessioni e le tue credenziali, e il daemon non ne vede mai nessuno. Il daemon gli consegna un lease con l’input reso dal template e i passi precedenti; il worker esegue il corpo e riferisce com’è andata. Il rapporto è tutto qui.

Ogni SDK ne porta uno. Registra i workflow che tiene, resta in long poll per i lease, manda heartbeat mentre un corpo gira, e risponde a ogni rifiuto in base al suo motivo. L’esempio qui sotto è lo stesso programma della pagina Workflow; questa è la parte che lo serve.

servire i corpi

// 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)

Il protocollo

Cinque chiamate, e un worker che le sa parlare si può scrivere usando direttamente l’API HTTP, senza un SDK. Il worker di ogni SDK è un client di queste cinque e di nient’altro.

One lease

The daemon learns that a body is running from started, that the worker is alive from the heartbeats, and how the step ended from complete or fail. Miss the heartbeats and the lease is reclaimed; a later call about it is refused with a reason, never lost.

Chiamata	Che cosa fa
`POST /v1/api/workers/register`	Annuncia il worker e archivia le definizioni che tiene. La risposta porta `lease_ttl_ms` e `heartbeat_ms`, i due numeri su cui gira tutto il resto del protocollo.
`POST /v1/api/steps/claim`	Offre fino a `max` passi eseguibili, tenendo aperta la richiesta fino a `wait_ms` quando non ce n’è nessuno. Il daemon tiene aperta una presa in carico al massimo per quattro quinti del suo timeout di richiesta, 24 secondi con il valore predefinito, e risponde subito quando smette, quindi una presa in carico può tornare vuota prima di `wait_ms`: fanne un’altra. Una presa in carico può essere ristretta a una sola esecuzione, ed è così che uno script porta fino in fondo la propria.
`POST /v1/api/steps/{lease}/started`	Dichiara che il corpo è partito. Prima del corpo, sempre: è quello che dice al daemon la differenza fra offerto e in esecuzione.
`POST /v1/api/steps/{lease}/heartbeat`	Allunga il lease. È l’unica prova di vita che il daemon ha di un worker.
`POST /v1/api/steps/{lease}/complete`, `/fail`	Chiude il passo: l’output, una modifica allo stato dell’esecuzione e il consumo che il daemon valorizza; oppure un errore con la classe che decide se ci sarà un nuovo tentativo.

I lease

Un lease è un tentativo, tenuto da un worker, per un tempo limitato. Porta l’input reso dal template, i passi precedenti indicizzati per id, lo stato e la correlazione dell’esecuzione, una chiave che resta stabile fra i nuovi tentativi di questo passo, e gli id di traccia e di span dello span del passo che il daemon ha aperto. Un corpo che registra span propri attraverso il contesto del passo li mette sotto quello span, così la chiamata al modello dentro un passo atterra nella traccia dell’esecuzione alla profondità giusta.

Un lease per cui il worker smette di mandare heartbeat scade. Il daemon segna il tentativo come orfano e, dopo una finestra di recupero, lo rimette in coda sotto at_least_once oppure lo fa fallire sotto at_most_once. Un worker che torna da una pausa e scrive su un lease già ripreso non viene ignorato: gli viene detto.

I quattro rifiuti

Una chiamata di un worker che il daemon non può accettare è rifiutata con un 409 che porta uno di quattro motivi. Sono un insieme chiuso, e il worker di ogni SDK gestisce ciascuno per nome, perché un worker che li riduce a un solo ramo abbandona passi che ha già finito o riferisce un fallimento al posto di un successo.

Motivo	Significato	Che cosa fa il worker
`already_applied`	Un complete o un fail precedente di questo stesso lease è arrivato e la sua risposta è andata persa.	Lo tratta come un successo e non scrive altro.
`fenced`	Esiste un tentativo più recente, oppure un altro lease ha chiuso questo.	Abbandona il lavoro e non scrive più niente per questo lease. Un heartbeat respinto dal fencing annulla anche, attraverso il contesto, il corpo in esecuzione.
`run_terminal`	L’esecuzione è finita, perché è arrivata in fondo, è fallita, è scaduta o è stata annullata.	Si ferma, e smette di prendere in carico lavoro per quell’esecuzione.
`not_started`	Un complete o un fail è arrivato per un tentativo `at_most_once` il cui `started` non è mai arrivato.	Lo segnala in modo evidente come un difetto del worker. I worker degli SDK non lo sanno produrre; uno scritto a mano sì.

Il corpo

Un corpo riceve un contesto, il suo input reso dal template e i passi precedenti. Il contesto è il posto dove riferisce quello che il daemon deve tenere: una modifica allo stato dell’esecuzione, fusa al momento del complete con le chiavi di primo livello che sostituiscono; i token che ha speso, che il daemon valorizza con lo stesso listino con cui valorizza gli span; e un segnale di annullamento che scatta quando il lease è perso o l’esecuzione finisce, così un corpo che lo passa alle proprie operazioni di I/O si ferma quando deve. I token si sommano da soli: uno span che il corpo registra attraverso il contesto, e ogni span sottostante, aggiunge il proprio consumo a quello del passo quando finisce, così un corpo che registra le sue chiamate al modello come span non somma mai a mano, e set_usage è l’override esplicito per un corpo che non ne registra nessuna.

Un corpo finisce restituendo il suo output oppure sollevando un errore. Solleva l’errore ritentabile dell’SDK per avere un nuovo tentativo dentro il budget del passo; solleva il fallimento tipizzato del passo per nominare qualunque altra classe; qualsiasi altra cosa è fatale. Il messaggio va al daemon e compare nel diario. Lo stack trace no.

Un lease per una definizione, una versione o un passo che il worker non tiene fallisce con unservable e il corpo non gira mai. Il tentativo torna in coda per un worker che invece lo tiene, senza consumare il budget dei nuovi tentativi, ed è questo che lascia girare fianco a fianco due versioni di un worker durante un deploy.

Scalare in orizzontale

Fai girare quanti worker vuoi, tutti collegati a un solo daemon. Ognuno prende in carico fino al limite della propria concorrenza, il daemon consegna ogni tentativo a uno solo di loro, e il fencing fa sì che un lease che in qualche modo due worker tengono insieme si risolva in un solo vincitore. L’id di un worker è registrato su ogni tentativo che ha preso, quindi il diario dice quale processo ha eseguito un passo. Dai un nome a ogni processo, oppure lascia che l’SDK lo ricavi dall’host.

Una presa in carico è un long poll, quindi una flotta ferma costa al daemon una richiesta aperta per worker e nessun traffico. Mentre aspetta, una presa in carico cerca lavoro quando parte un’esecuzione, finisce un passo, scatta una scadenza o un evento riprende un’attesa, e altrimenti al massimo due volte al secondo. Una flotta al lavoro è limitata dalla concorrenza dei worker stessi, e un corpo che dura un’ora costa un heartbeat ogni dieci secondi e nient’altro.

Le chiavi

Le chiamate del worker vogliono l’ambito worker; avviare, segnalare e annullare esecuzioni vogliono execute. Entrambi ricadono sulle chiavi di scrittura quando nessuno li imposta, e una chiave admin li tiene tutti e due. Un worker che serve soltanto passi non ha bisogno di una chiave capace di avviare un’esecuzione, e un webhook che pubblica soltanto eventi non ha bisogno di una chiave capace di prenderne in carico una. Per le opzioni, vedi Configurazione.

Le chiamate del worker non vengono registrate nel registro di audit, anche se un lease porta l’input del passo e gli output dei passi precedenti. Sono un ciclo di una macchina che prende in carico molte volte al minuto, e una riga per ogni presa in carico seppellirebbe le letture fatte da una persona. Rileggere un’esecuzione attraverso gli endpoint delle esecuzioni viene registrato; vedi Sicurezza.

Poi

Avviare e seguire un’esecuzione, e rispondere a un’attesa: Esecuzioni.
La definizione che il worker serve: Workflow.
Ogni endpoint del protocollo dei worker, generato dalla specifica: API HTTP, Workers.