Pular para o conteúdo principal

Ação assíncrona

Quando usar: a operação demora mais do que uma requisição HTTP deve durar (varredura, geração de relatório, chamada a terceiro), ou precisa sobreviver a um restart do pod. A regra é simples: requisição HTTP não fica aberta esperando trabalho.

O padrão da plataforma: a rota responde 202 com o corpo {"action": {…}} e um Location da rota de status; quem executa é um Runner dentro do serviço, com lease, retry e desfecho registrado.

O descritor​

nnp/api.yaml — a rota async e a rota de status (obrigatória quando existe rota async):

actions: {status_route: actions.get}
routes:
  - {id: items.scan, method: POST, path: "/items/{item_id}/scan", permission: sample.item.scan, idempotency: required, mode: async, max_body_bytes: 4096}
  - {id: actions.get, method: GET, path: "/actions/{action_id}", permission: sample.item.read}

O evento terminal da operação é uma trilha de audit declarada — ele sai no audit.yaml, não no api.yaml:

trails:
  operations:
    audience: tenant
    events:
      item.scan: {schema_version: 1, description: Varredura concluída ou falha, payload_schema: audit/operations/item.scan.v1.schema.json}

Aceitar: barato e transacional​

java/src/main/java/.../app/ScansController.java:

@NnsRoute("items.scan")
ResponseEntity<byte[]> scan(
    Ctx ctx, ServiceContext sc, ServiceChain.Dispatch d, Match m, NnsBody body) {
  String id = m.params().get("item_id");
  return service.mutation(
      ctx,
      sc,
      d,
      body,
      (tx, b) -> {
        if (!isBlank(b) && decodeEmpty(b) == null) {
          throw Wire.invalidBody();
        }
        String tenant = sc.requireTenantId();
        Item it = ItemStore.get(ctx, tx, tenant, id, true);
        if (service.scanBusy(ctx, it)) {
          throw SampleService.scanInProgress(it);
        }
        Action act =
            Actions.newAction(
                    ctx, service.actionConfig, SampleService.SCAN_TYPE, Map.of("item_id", it.id()))
                .withResource("item", it.id());
        service.actions.create(ctx, tx, act);
        ItemStore.markScanPending(ctx, tx, tenant, it.id(), act.id(), act.createdAt());
        return Actions.acceptedResponse(act);
      });
}

O aceite grava três coisas na mesma transação: a reserva da Idempotency-Key, a Action (nn_actions) e o vínculo no item. Repetir a chamada com a mesma chave devolve a mesma resposta — a Action original — em vez de criar uma segunda varredura.

O estado da operação é uma leitura, com a vista montada a partir da Action:

@NnsRoute("actions.get")
ResponseEntity<byte[]> action(Ctx ctx, ServiceContext sc, Match m) {
  Action a = service.actions.get(ctx, sc.requireTenantId(), m.params().get("action_id"));
  return Wire.json(200, Map.of("action", actionView(a)));
}

Executar: o handler​

O tipo da Action é a permissão qualificada da rota que a aceita (sample.item.scan), e é por ele que o Runner encontra o código:

public static final String SCAN_TYPE = SampleModule.ID + ".item.scan";

java/src/main/java/.../app/SampleService.java:

/** O handler da Action. Idempotente: repetido depois de crash, regrava o mesmo resultado. */
public Map<String, Object> runScan(Ctx ctx, Action act) {
  String itemId = itemIdOf(act.input());
  if (itemId.isEmpty()) {
    throw ApiError.of(Code.VALIDATION_FAILED, "scan_input_invalid", "Entrada da varredura sem item_id.");
  }
  boolean mine;
  try {
    mine = ItemStore.markScanRunning(ctx, db, act.tenantId(), itemId, act.id(), now());
  } catch (RuntimeException e) {
    throw dbUnavailable();   // 503 dependency_unavailable/database_unavailable, retryable
  }
  if (!mine) {
    // Item apagado, ou a varredura dele é outra: falha DEFINITIVA, sem retry.
    throw ItemStore.notFound();
  }
  progress(ctx, "varrendo o item");
  sleep(ctx, scanDelay);     // NnsBackground.sleep: acorda no cancelamento

  Item it = ItemStore.get(ctx, db, act.tenantId(), itemId, false);
  List<Finding> findings = ScanEngine.checks(it);
  int score = ScanEngine.score(findings);
  boolean recorded =
      ItemStore.recordScan(
          ctx, db, act.tenantId(), itemId, act.id(), ScanEngine.ENGINE, config.releaseId(),
          findings, score, now());
  if (!recorded) {
    throw ItemStore.notFound();
  }
  Map<String, Object> out = new LinkedHashMap<>();
  out.put("item_id", itemId);
  out.put("engine", ScanEngine.ENGINE);
  out.put("checks", findings.size());
  out.put("warnings", ScanEngine.warnings(findings));
  out.put("score", score);
  out.put("marker", SampleModule.MARKER);
  out.put("completed_by", config.releaseId());
  return out;
}

Quatro regras que fazem a diferença:

  • A posse é um predicado do banco (scan_action_id = <a Action>): repetir a execução depois de um crash regrava o mesmo resultado em vez de abrir uma segunda varredura.
  • Erro classificado: o que pode melhorar (banco fora) é dependency_unavailable com retryable: true; o que não pode (item apagado) é 404 — o Runner não insiste.
  • A espera é do SDK (NnsBackground.sleep(ctx, d)): ela acorda quando o contexto é cancelado, e o Runner devolve a Action em vez de segurá-la até o fim do lease.
  • O que o handler devolve é o corpo do evento terminal — é ele que fica no result da Action e no payload do evento operations/item.scan.

Registrando o Runner​

java/src/main/java/.../app/SampleConfiguration.java:

@Bean
ActionRunner sampleActionRunner(
    ActionStore store, NnsConfig config, NnsModule module, SampleService service,
    TxAuditEmitter emitter, Logger logger) {
  ActionRunner.Builder b =
      ActionRunner.builder(store, config.releaseId())
          .handler(SampleService.SCAN_TYPE, service::runScan)
          .logger(logger.with("worker", "action-runner"));
  if (emitter.transactional() && module.audit().isPresent()) {
    b.emitter(emitter).audit(module.audit().get());
  }
  return b.build();
}

/** O Runner roda em TODA réplica de toda release, como laço longo do ciclo de vida. */
@Bean
SmartInitializingSingleton sampleActionRunnerLoop(
    ObjectProvider<ActionRunner> runner, ObjectProvider<NnsBackground> background) {
  return () -> {
    ActionRunner r = runner.getIfAvailable();
    NnsBackground bg = background.getIfAvailable();
    if (r != null && bg != null) {
      bg.loop("action-runner", r::run);
    }
  };
}

Mais o Reaper (fecha as Actions de uma release que saiu do ar) e o contador do /ready:

@Bean
ActionReaper sampleActionReaper(ActionStore store, NnsConfig config, RoleWatcher role, Logger logger) {
  return ActionReaper.of(store, config.releaseId(), () -> role.refresh().live())
      .withLogger(logger.with("worker", "action-reaper"))
      .withAdoptable(List.of(SampleService.SCAN_TYPE));
}

@Bean
NnsPendingActions samplePendingActions(ActionStore store, NnsConfig config) {
  return ctx -> store.pending(ctx, config.releaseId());
}

O cliente vê assim​

POST /v1/sample/items/0be1d3c6-6f3e-4a6b-a4f5-2a2f2a9f0d10/scan
Idempotency-Key: 8e14f0c6-...

202 Accepted
Location: /v1/sample/actions/1c9b0b7e-...
{"action":{"id":"1c9b0b7e-...","type":"sample.item.scan","status":"pending", ...}}
GET /v1/sample/actions/1c9b0b7e-...

200 OK
{"action":{"id":"1c9b0b7e-...","status":"succeeded","output":{"item_id":"0be1d3c6-...","score":100, ...}}}

Como testar​

Em MySQL real, com Runner de verdade (java/src/test/java/.../app/ScanActionMysqlTest.java):

@Test
void runnerClosesTheActionWithTheResultAndTheItemKeepsIt() {
  Item it = newItem("Cartão");
  Action act = acceptScan(it);
  assertThat(act.status()).isEqualTo(Status.PENDING);

  assertThat(runner.processOnce(CTX)).isEqualTo(1);

  Action done = actions.get(CTX, TENANT, act.id());
  assertThat(done.status()).isEqualTo(Status.SUCCEEDED);
  assertThat(done.result().get())
      .containsEntry("item_id", it.id())
      .containsEntry("engine", ScanEngine.ENGINE)
      .containsEntry("score", 100.0)
      .containsEntry("completed_by", RELEASE);
  assertThat(ItemStore.get(CTX, db, TENANT, it.id(), false).scanStatus()).isEqualTo(Scan.SUCCEEDED);
}

@Test
void aRetryAfterACrashCompletesTheScanInsteadOfDuplicatingIt() {
  Item it = newItem("Duplicado");
  Action act = acceptScan(it);
  // A tentativa anterior marcou o item como `running` e morreu antes de gravar o resultado.
  assertThat(ItemStore.markScanRunning(CTX, db, TENANT, it.id(), act.id(), act.createdAt())).isTrue();
  Map<String, Object> result = service.runScan(CTX, act);
  assertThat(result).containsEntry("score", 100).containsEntry("item_id", it.id());
}

Erros comuns​

  • Fazer o trabalho dentro da requisição e responder 200 no fim: a requisição fica presa, o gateway corta no prazo e o cliente não sabe se a operação aconteceu.
  • Esquecer actions.status_route — a partida recusa: uma rota async sem rota de status não teria como ser acompanhada.
  • Responder 202 sem Location — use Actions.acceptedResponse(act), que monta os dois.
  • Não classificar o erro — tudo virando 500 faz o Runner desistir depois das tentativas de algo que era só indisponibilidade momentânea; e um retryable: true num erro definitivo faz o contrário.
  • Criar a Action fora da transação do vínculo — sobra Action sem dono (ou item marcado sem Action) quando algo falha no meio.
  • Fazer Thread.sleep — o desligamento não espera por você e o lease vence. Use NnsBackground.sleep(ctx, …), que devolve o controle no cancelamento.
  • Registrar o mesmo tipo em dois Runners da mesma release — o tipo passa a ter dois donos.
  • Rodar o Runner só na release ativa — o Runner roda em toda réplica de toda release: a Action pertence a uma release, e é ela que a executa.
  • Devolver dado sensível no resultado — o resultado vai para o cliente e para a trilha de audit.