Publish Budget#

PublishBudget 把 workspace publication 的运维时间边界写进 task metadata。它只适用于 workspace task，用来让 Conductor 的 responseTimeoutSeconds 覆盖 LakeFS merge、Conductor completion update、worker shutdown grace 和 heartbeat slack。

这个页面面向运行时维护者和需要给 workspace task 配置发布预算的任务作者。TaskDef 字段映射见 ../task-authoring/controls-and-taskdef.md；LakeFS 发布顺序见 workspace-publication.md。

何时需要配置#

默认 TaskControls(timeout=TimeoutPolicy(response_seconds=600)) 只表达 Conductor response timeout。对短任务或没有真实 LakeFS publication 压力的任务，这通常足够。

当 workspace task 会修改大量 object、LakeFS merge latency 已经有观测值，或 worker shutdown 与 Conductor result update 需要明确留量时，配置 publish_budget：

from pathlib import Path

from pydantic import BaseModel, Field

from perago import PublishBudget, TaskControls, WorkspaceSpec, task


class BuildFeaturesParams(BaseModel):
    feature_set: str


class BuildFeaturesOutput(BaseModel):
    row_count: int = Field(ge=0)


@task(
    name="features.build",
    owner_email="data@example.com",
    workspace=WorkspaceSpec(prefix="/audio/render"),
    controls=TaskControls(
        publish_budget=PublishBudget(
            observed_merge_p99_seconds=20,
            safety_margin_seconds=10,
            lakefs_merge_timeout_seconds=45,
            conductor_completion_timeout_seconds=15,
            worker_shutdown_grace_seconds=30,
            heartbeat_interval_seconds=10,
        ),
    ),
)
def build_features(workspace: Path, params: BuildFeaturesParams) -> BuildFeaturesOutput:
    ...

Required/optional/generated 字段边界：

  • required: PublishBudget 的 6 个字段都必须显式提供。

  • optional: TaskControls.publish_budget 可以省略；省略时使用 TimeoutPolicy.response_seconds。

  • conditional: publish_budget 只允许配置在带 WorkspaceSpec(...) 的 workspace task 上。

  • generated: responseTimeoutSeconds 由 PublishBudget.response_timeout_seconds 派生并写入 TaskDef。

  • forbidden: 不能把 publish_budget 配在 workspace-free task 上；不能在 PublishBudget 里声明未知字段或 exactly-once 语义。

字段语义#

字段

约束

说明

observed_merge_p99_seconds

>= 0

目标 workload 下已观测到的 LakeFS merge 高分位延迟。

safety_margin_seconds

>= 0

覆盖观测抖动、网络抖动和小幅数据量增长的安全余量。

lakefs_merge_timeout_seconds

>= 1

传给 LakeFS merge SDK request 的 timeout。必须覆盖观测 p99 加安全余量。

conductor_completion_timeout_seconds

>= 1

worker 向 Conductor 回写最终 task result 的 request timeout。

worker_shutdown_grace_seconds

>= 1

publication 后预留给 worker 停止、清理和进程退出的时间。

heartbeat_interval_seconds

>= 1

预留给 Conductor response timeout/heartbeat 机制的 slack。

PublishBudget 是 frozen Pydantic model，并拒绝额外字段。配置错误会在模块导入、perago check 或 perago extract 阶段暴露为 validation error。

Response Timeout 计算#

PublishBudget.response_timeout_seconds 的公式是：

lakefs_merge_timeout_seconds
+ conductor_completion_timeout_seconds
+ worker_shutdown_grace_seconds
+ heartbeat_interval_seconds

上面的示例会生成：

45 + 15 + 30 + 10 = 100

因此 perago extract 写出的 TaskDef 会包含：

{
  "responseTimeoutSeconds": 100
}

如果同时配置了 TimeoutPolicy(response_seconds=999) 和 publish_budget，publish_budget 优先。PublishBudget 本身不会写入 TaskDef JSON，也不会出现在 Conductor input/output 中。

Runtime 使用位置#

PublishBudget 同时影响 TaskDef 和 worker runtime：

位置

使用字段

行为

TaskDef generation

response_timeout_seconds

生成 Conductor responseTimeoutSeconds。

LakeFS publish

lakefs_merge_timeout_seconds

作为 merge request timeout 传给 LakeFS SDK。

Conductor result update

conductor_completion_timeout_seconds

作为 task completion update request timeout 传给 Conductor client。

observed_merge_p99_seconds 和 safety_margin_seconds 不直接传给外部系统。它们只用于校验 lakefs_merge_timeout_seconds 是否覆盖 observed_merge_p99_seconds + safety_margin_seconds。

配置流程#

  1. 在接近真实数据量的 workspace task 上观测 LakeFS merge latency。

  2. 选取稳定窗口内的 p99 或更保守分位，填入 observed_merge_p99_seconds。

  3. 根据网络、object 数量增长和 LakeFS 负载变化选择 safety_margin_seconds。

  4. 设置 lakefs_merge_timeout_seconds >= observed_merge_p99_seconds + safety_margin_seconds。

  5. 为 Conductor completion update 和 worker shutdown 分别设置明确预算。

  6. 运行 perago check 验证 task definition，再运行 perago extract 检查生成的 responseTimeoutSeconds。

不要把 lakefs_merge_timeout_seconds 设成远小于观测值的探测性 timeout。merge timeout 或连接错误后，runtime 不能假设 publish 一定没有发生；需要先用 commit metadata 判断是否已经存在匹配的 publication。

常见拒绝形状#

merge timeout 小于观测值加安全余量会失败：

PublishBudget(
    observed_merge_p99_seconds=20,
    safety_margin_seconds=10,
    lakefs_merge_timeout_seconds=29,
    conductor_completion_timeout_seconds=15,
    worker_shutdown_grace_seconds=30,
    heartbeat_interval_seconds=10,
)

workspace-free task 配置 publish_budget 会失败：

@task(
    name="metadata.validate",
    owner_email="data@example.com",
    controls=TaskControls(publish_budget=budget),
)
def validate_metadata(params: ValidateMetadataParams) -> ValidateMetadataOutput:
    ...

额外字段会失败：

PublishBudget(
    observed_merge_p99_seconds=20,
    safety_margin_seconds=10,
    lakefs_merge_timeout_seconds=45,
    conductor_completion_timeout_seconds=15,
    worker_shutdown_grace_seconds=30,
    heartbeat_interval_seconds=10,
    exact_once=True,
)

PublishBudget 是运维时间预算，不是 exactly-once publication 证明。Perago MVP 的恢复边界仍然是 attempt fence、publish fence、metadata classification 和 fail closed。