Workers

O que é um worker

Um worker guarda os corpos das etapas reivindicáveis e pede trabalho ao daemon. Ele roda onde o seu código roda, com os seus imports, o seu pool de conexões e as suas credenciais, e o daemon nunca vê nada disso. O daemon entrega a ele um lease com a entrada já renderizada e as etapas anteriores; o worker roda o corpo e relata como foi. É essa a relação inteira.

Todo SDK traz um. Ele registra os workflows que guarda, faz long polling por leases, manda heartbeat enquanto um corpo roda, e responde a cada recusa pelo motivo dela. O exemplo abaixo é o mesmo programa da página Workflows; esta é a parte que o serve.

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

O protocolo

Cinco chamadas, e um worker que as fala pode ser escrito contra a API HTTP sem SDK nenhum. O worker de cada SDK é um cliente delas e de mais nada.

Leases

Um lease é uma tentativa, segurada por um worker, por um tempo limitado. Ele carrega a entrada renderizada, as etapas anteriores indexadas por id, o estado e a correlação da execução, uma chave que é estável entre as repetições desta etapa, e os ids de trace e de span do span de etapa que o daemon abriu. Um corpo que registra os próprios spans pelo contexto da etapa os coloca sob aquele span, então a chamada de modelo dentro de uma etapa cai no trace da execução na profundidade certa.

Um lease para o qual o worker deixa de mandar heartbeat vence. O daemon marca a tentativa como órfã e, passada uma janela de recuperação, a devolve à fila sob at_least_once ou a faz falhar sob at_most_once. Um worker que volta de uma pausa e manda algo contra um lease já recuperado não é ignorado: ele é avisado.

As quatro recusas

Uma chamada de worker que o daemon não pode aceitar é recusada com um 409 que carrega um de quatro motivos. Eles são fechados, e o worker de cada SDK trata cada um pelo nome, porque um worker que junta todos em um ramo só abandona etapas que já terminou ou relata falha onde houve sucesso.

O corpo

Um corpo recebe um contexto, a entrada já renderizada e as etapas anteriores. O contexto é onde ele relata o que o daemon deve guardar: um patch no estado da execução, mesclado no complete com as chaves de primeiro nível substituindo as antigas; os tokens que ele gastou, que o daemon precifica com a mesma tabela com que precifica spans; e um sinal de cancelamento que dispara quando o lease se perde ou a execução acaba, para que um corpo que passe esse sinal à própria E/S pare na hora certa. Os tokens se somam sozinhos: um span que o corpo registra pelo contexto, e todo span abaixo dele, soma o uso ao da etapa quando termina, então um corpo que registra as chamadas de modelo como spans nunca soma à mão, e set_usage é a substituição explícita para um corpo que não registra nenhuma.

Um corpo termina devolvendo a saída dele ou levantando um erro. Levante o erro repetível do SDK para ser repetido dentro do orçamento da etapa; levante uma falha de etapa tipada para nomear qualquer outra classe; qualquer outra coisa é fatal. A mensagem vai para o daemon e aparece no diário. Um stack trace não vai.

Um lease de uma definição, de uma versão ou de uma etapa que o worker não guarda falha como unservable e o corpo nunca roda. A tentativa volta para a fila, para um worker que a guarde, sem gastar o orçamento de repetições, que é o que deixa duas versões de um worker rodarem lado a lado durante uma implantação.

Escalar horizontalmente

Rode quantos workers quiser contra um daemon. Cada um reivindica até a própria concorrência, o daemon entrega cada tentativa a exatamente um deles, e o fence faz com que um lease que dois workers de algum modo seguram termine com um vencedor só. O id de um worker fica gravado em toda tentativa que ele pegou, então o diário diz qual processo rodou uma etapa. Dê um nome a cada processo, ou deixe o SDK derivar um a partir do host.

Uma reivindicação é um long poll, então uma frota ociosa custa ao daemon uma requisição aberta por worker e nenhum tráfego. Enquanto espera, uma reivindicação procura trabalho quando uma execução começa, uma etapa termina, um prazo dispara ou um evento retoma uma espera, e no máximo duas vezes por segundo fora isso. Uma frota ocupada é limitada pela concorrência dos próprios workers, e um corpo que leva uma hora custa um heartbeat a cada dez segundos e mais nada.

Chaves

As chamadas de worker pedem o escopo worker; iniciar, sinalizar e cancelar execuções pedem execute. Os dois recaem sobre as chaves de escrita quando não estão definidos, e uma chave de administração cobre ambos. Um worker que só serve etapas não precisa de uma chave capaz de iniciar uma execução, e um webhook que só publica eventos não precisa de uma chave capaz de reivindicar uma etapa. Veja Configuração para as opções.

As chamadas de worker não são registradas no log de auditoria, embora um lease carregue a entrada da etapa e as saídas das etapas anteriores. Elas são um laço de máquina que reivindica muitas vezes por minuto, e uma linha por reivindicação enterraria as leituras que uma pessoa fez. Ler uma execução de volta pelos endpoints de execução é registrado; veja Segurança.

A seguir

• Iniciar e acompanhar uma execução, e responder a uma espera: Execuções.
• A definição que o worker serve: Workflows.
• Todos os endpoints do protocolo de worker, gerados a partir da especificação: API HTTP, workers.