使用关联 ID 避免重复执行-未完纪

可以把 system.correlationId 作为幂等键，用它把相关执行归为一组、追踪跨子流程的执行链，并跳过重复处理。

本指南介绍两种模式：对于 API 触发的流程，在创建执行时设置关联 ID；对于 Webhook 触发、幂等键在执行开始后才可用的流程，则使用自定义标签。

`system.correlationId` 如何工作

它是每次执行都有的内置系统标签。
默认值是执行自身的 ID，并且会自动传播到子流程：同一条执行链中的执行共享根执行的 system.correlationId。
创建执行时，可以将它设为任意稳定的业务标识，例如支付意图、消息 UUID 或事件键。
执行创建后，该值不可变，不能在流程执行过程中修改。

何时使用

如果同一个业务事件可能到达多次，而你希望只处理一次，可以使用 system.correlationId。典型情形包括：

通过 API 触发的支付处理。
消费消息队列的事件驱动流水线。
由调用方控制触发、并在创建执行时已持有稳定业务键的流程。

对于 Webhook 触发的流程，如果幂等键在执行创建之后才从请求头中取得，应使用自定义标签。后文“Webhook 触发的流程”将介绍这一模式。

在创建执行时设置关联 ID

创建执行时，通过 labels 查询参数传入 system.correlationId。标签采用 key:value 格式。

把 {your-tenant} 替换为你的租户 ID，可在 Administration → Tenants（管理 → 租户）中查看。

curl -X POST "http://localhost:8080/api/v1/{your-tenant}/executions/company.team/payments?labels=system.correlationId:payment-ORD-123" \
  -H "Authorization: Bearer {your-api-token}"

调用方已经持有稳定业务键时，适合采用这种方式。示例中的租户和 API 令牌均为占位值。

尽早跳过重复执行

在创建时设置关联 ID 后，检查是否已经存在相同键的成功执行。存在时，就跳过当前运行。

同时把幂等键作为 labels 查询参数和流程输入传入，让流程能在重复检查中引用它：

curl -X POST "http://localhost:8080/api/v1/{your-tenant}/executions/company.team/payment_flow_guarded?labels=system.correlationId:payment-ORD-123" \
  -H "Authorization: Bearer {your-api-token}" \
  -F "idempotencyKey=payment-ORD-123"

id: payment_flow_guarded
namespace: company.team

inputs:
  - id: idempotencyKey
    type: STRING

tasks:
  - id: check_existing
    type: io.kestra.plugin.core.http.Request
    uri: "http://localhost:8080/api/v1/{{ kv('KESTRA_TENANT') }}/executions/search?filters[labels][EQUALS][system.correlationId]={{ inputs.idempotencyKey }}&filters[namespace][EQUALS]=company.team&filters[flowId][EQUALS]=payment_flow_guarded&filters[state][IN]=SUCCESS&size=1"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('KESTRA_API_TOKEN') }}"

  - id: maybe_skip
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not (outputs.check_existing.body contains '\"total\":0') }}"
    then:
      - id: skip
        type: io.kestra.plugin.core.log.Log
        message: "Duplicate {{ inputs.idempotencyKey }} skipped; already succeeded"
    else:
      - id: continue
        type: io.kestra.plugin.core.log.Log
        message: "First time for {{ inputs.idempotencyKey }}, proceed"

分别把租户 ID 和 API 令牌保存为 KV 值和 Secret。

Webhook 触发的流程

执行创建时，system.correlationId 会自动分配，之后无法修改。对于 Webhook 触发的流程，提供方的幂等键只有在执行开始后才可用。可以使用 Labels 任务把它保存到自定义标签中，再使用该标签检查重复执行。

id: payment_webhook
namespace: company.team

variables:
  idem_key: "{{ trigger.headers['Idempotency-Key'] | first }}"

tasks:
  - id: set_idempotency_key
    type: io.kestra.plugin.core.execution.Labels
    labels:
      idempotency.key: "{{ vars.idem_key }}"

  - id: check_existing
    type: io.kestra.plugin.core.http.Request
    uri: "http://localhost:8080/api/v1/{{ kv('KESTRA_TENANT') }}/executions/search?filters[labels][EQUALS][idempotency.key]={{ vars.idem_key }}&filters[namespace][EQUALS]=company.team&filters[flowId][EQUALS]=payment_webhook&filters[state][IN]=SUCCESS&size=1"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('KESTRA_API_TOKEN') }}"

  - id: maybe_skip
    type: io.kestra.plugin.core.flow.If
    condition: "{{ not (outputs.check_existing.body contains '\"total\":0') }}"
    then:
      - id: skip
        type: io.kestra.plugin.core.log.Log
        message: "Duplicate {{ vars.idem_key }} skipped"
    else:
      - id: process_payment
        type: io.kestra.plugin.core.log.Log
        message: "Charge payment for {{ vars.idem_key }}"

triggers:
  - id: webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: payment-events

使用关联 ID 进行运维

界面筛选：在 Executions（执行）页面添加标签筛选条件 system.correlationId:your-key，查看整条执行链。
API 搜索：在执行搜索 API 中使用 filters[labels][EQUALS][system.correlationId]={value} 和 filters[state][IN]=SUCCESS 查询参数，以编程方式审计或检测重复执行。
子流程：关联 ID 会自动传播，因此下游执行共享相同的 system.correlationId，无需额外配置。