Effect 执行计划生命周期事件用onEvent观测 ExecutionPlan 的每次尝试【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect执行计划ExecutionPlan是 Effect 中用于按顺序尝试多个后备执行步骤的机制一个计划包含一个或多个步骤每个步骤提供一套Context或Layer资源并可通过attempts、while、schedule定义重试策略运行时按顺序尝试直到某个步骤成功或计划被耗尽。本文聚焦该机制的一个重要可观测性能力通过Effect.withExecutionPlan与Stream.withExecutionPlan的可选onEvent处理器从效应外部观察每次尝试的生命周期事件用于日志与指标采集。读完本文你将掌握事件模型AttemptStart/AttemptSuccess/AttemptFailure、事件编号语义、异常场景中断、缺陷、观察者自身失败下的行为以及如何在真实代码中落地观测。背景ExecutionPlan 与尝试Attempt语义在进入事件 API 之前先明确尝试的定义。在 ExecutionPlan.ts 的模块文档中ExecutionPlan被描述为描述运行效应或流的有序后备步骤ordered fallback steps每个步骤提供Context.Context或Layer.Layer字段provide可选地配置attempts该步骤最多尝试次数、while基于输入决定是否继续尝试的谓词、schedule重试时间调度运行时按顺序尝试各步骤直到工作流成功或计划耗尽计划支持merge按顺序拼接多个计划的步骤与captureRequirements用当前上下文捕获需求。一次尝试指在某个步骤提供资源后执行一次完整的效应或流。步骤之间可能因为资源构建失败Layer 失败、效应失败、流部分输出后失败等原因而发生失败切换同一步骤内部则可能因attempts/schedule配置而重试多次。而ExecutionPlan.CurrentMetadata一个Context.ReferenceMetadata默认值为{ attempt: 0, stepIndex: 0 }记录了当前运行中尝试的元数据供运行中的代码读取。核心 APIonEvent处理器本变更changeset 条目为Effect.withExecutionPlan与Stream.withExecutionPlan增加了可选的onEvent处理器。其公开签名定义在 Effect.ts 与 Stream.ts// Effect 版本 Effect.withExecutionPlanA, E, R( effect, plan, options?: { readonly onEvent?: ((event: ExecutionPlan.EventE | PlanE) Effectvoid, never, RX) | undefined } ): EffectA, E | PlanE, ExcludeR, Provides | PlanR | RX // Stream 版本 Stream.withExecutionPlan( policy, options?: { readonly preventFallbackOnPartialStream?: boolean | undefined readonly onEvent?: ((event: ExecutionPlan.EventE | PolicyE) Effectvoid, never, RX) | undefined } )几点关键约束来自 Effect.ts 的 API 文档处理器的返回值类型固定为Effectvoid, never, RX即处理器不允许失败错误类型为never这保证观察永远不会改变计划的最终结果处理器在每次尝试之前和之后被内联等待awaited inline因此事件是严格有序发射的但这也意味着应保持处理器廉价避免拖慢被观测的效应处理器的需求requirementsRX会被追加到结果效应上如果处理器依赖某个服务比如指标上报客户端该服务会成为最终效应的需求终端事件AttemptSuccess/AttemptFailure像 finalizer 一样运行即使尝试被中断终端事件也一定会发射详见下文中断与缺陷一节。最简单的用法即 changeset 中的示例import { Effect } from effect Effect.withExecutionPlan(program, plan, { onEvent: (event) Effect.log(execution plan event, event) })事件模型ExecutionPlan.Event标记联合onEvent收到的参数是一个ExecutionPlan.EventE定义为三个变体的标记联合tagged union其类型定义位于 ExecutionPlan.ts事件_tag携带字段语义AttemptStartAttemptStartattempt、stepAttempt、stepIndex一次尝试开始之前发射AttemptSuccessAttemptSuccessattempt、stepAttempt、stepIndex、duration尝试成功成功的尝试会结束整个计划因此它总是最后的事件AttemptFailureEAttemptFailureattempt、stepAttempt、stepIndex、duration、cause尝试失败cause是完整的失败原因关键语义每个AttemptStart之后恰好跟随一个终端事件AttemptSuccess或AttemptFailure两者配成一对不存在有头无尾的情况AttemptFailure.cause携带完整的CauseE因此不仅预期的错误typed error会被上报缺陷defect与中断interruption也会被如实报告durationAttemptSuccess/AttemptFailure是本次尝试的经过时间elapsed time类型为Duration.Duration可直接用于指标统计之后计划是否继续重试或切换步骤由该步骤的attempts、while、schedule决定若紧接着再次发射AttemptStart则说明又进行了一次尝试。编号语义attempt与stepAttempt事件编号与ExecutionPlan.CurrentMetadata完全对应是理解事件序列的关键ExecutionPlan.ts 与 changeset 均明确说明attempt是跨步骤累计的 1-based 尝试号它从计划第一次尝试计为 1每次新尝试无论发生在哪个步骤递增与CurrentMetadata.attempt完全一致stepAttempt是当前步骤内的 1-based 尝试号进入一个新步骤时重置为 1同一步骤内重试则递增stepIndex是 0-based 的步骤索引。在内部实现 executionPlan.ts 中makeEventEmitter用闭包维护lastStepIndex与stepAttempt当meta.stepIndex变化时把stepAttempt重置为 0然后每次尝试stepAttempt从而得到与元数据一致的编号。也就是说事件编号不是独立的另一套计数器而是与运行中代码通过CurrentMetadata读到的编号同一套值。事件序列示例Effect 与 Stream 的实测行为以下事件序列均来自仓库中 ExecutionPlan.test.ts 的describe(onEvent)测试用例simplify仅把cause压平成错误值便于断言。场景一单步骤首次即成功对Effect.succeed(1)套用ExecutionPlan.make({ provide: Context.empty() })AttemptStart { attempt: 1, stepAttempt: 1, stepIndex: 0 } AttemptSuccess { attempt: 1, stepAttempt: 1, stepIndex: 0, duration }场景二单步骤内多次重试任务前两次返回fail-1、fail-2第三次成功计划为{ provide: Context.empty(), attempts: 3 }AttemptStart { attempt: 1, stepAttempt: 1, stepIndex: 0 } AttemptFailure { attempt: 1, stepAttempt: 1, stepIndex: 0, error: fail-1 } AttemptStart { attempt: 2, stepAttempt: 2, stepIndex: 0 } AttemptFailure { attempt: 2, stepAttempt: 2, stepIndex: 0, error: fail-2 } AttemptStart { attempt: 3, stepAttempt: 3, stepIndex: 0 } AttemptSuccess { attempt: 3, stepAttempt: 3, stepIndex: 0 }注意同一步骤内attempt与stepAttempt同步递增。场景三跨步骤失败切换failover计划为两个步骤每步attempts: 2任务同样第三次才成功AttemptStart { attempt: 1, stepAttempt: 1, stepIndex: 0 } AttemptFailure { attempt: 1, stepAttempt: 1, stepIndex: 0, error: fail-1 } AttemptStart { attempt: 2, stepAttempt: 2, stepIndex: 0 } AttemptFailure { attempt: 2, stepAttempt: 2, stepIndex: 0, error: fail-2 } AttemptStart { attempt: 3, stepAttempt: 1, stepIndex: 1 } // 步骤切换stepAttempt 重置 AttemptSuccess { attempt: 3, stepAttempt: 1, stepIndex: 1 }这里attempt继续累计到 3而stepAttempt在新步骤重置为 1——这正是跨步骤累计 / 步骤内 1-based 语义的直观体现。测试还验证了AttemptStart事件中的attempt/stepIndex与任务内部通过ExecutionPlan.CurrentMetadata读到的元数据逐一相等。场景四Stream 的后备切换对Stream.unwrap(Effect.map(Service, (_) _.stream))应用三步计划步骤 A、B 的流分别以错误A、B失败步骤 C 输出[1, 2, 3]AttemptStart { attempt: 1, stepAttempt: 1, stepIndex: 0 } AttemptFailure { attempt: 1, stepAttempt: 1, stepIndex: 0, error: A } AttemptStart { attempt: 2, stepAttempt: 1, stepIndex: 1 } AttemptFailure { attempt: 2, stepAttempt: 1, stepIndex: 1, error: B } AttemptStart { attempt: 3, stepAttempt: 1, stepIndex: 2 } AttemptSuccess { attempt: 3, stepAttempt: 1, stepIndex: 2 }场景五步骤资源Layer构建失败第一步的Layer.effect(Service, Effect.fail(nope))在构建资源时就失败AttemptStart { attempt: 1, stepAttempt: 1, stepIndex: 0 } AttemptFailure { attempt: 1, stepAttempt: 1, stepIndex: 0, error: nope } AttemptStart { attempt: 2, stepAttempt: 1, stepIndex: 1 } AttemptSuccess { attempt: 2, stepAttempt: 1, stepIndex: 1 }这证明 Layer 构建失败同样会触发完整的AttemptStart/AttemptFailure配对观测者无需区分资源失败与业务失败。边界行为中断、缺陷与观察者自身失败事件机制在异常场景下的行为是把它用于生产观测的关键前提。中断Interruption测试用Effect.never挂起任务在AttemptStart之后中断 fiber事件序列仍为恰好两个AttemptStartAttemptFailureAttemptFailure.cause满足Cause.hasInterrupts(failure.cause)——中断以AttemptFailure上报且携带中断原因这是因为终端事件像 finalizer 一样运行内部实现中尝试被包裹在Effect.uninterruptibleMaskEffect.onExit里executionPlan.ts即使尝试被中断onExit也会执行emitter.end发射终端事件。缺陷Defect对Effect.die(boom)应用含attempts: 2的计划结果仍是Exit.die(boom)缺陷不会被重试事件为AttemptStartAttemptFailure且cause中携带该缺陷——缺陷也被如实上报而不会静默消失。观察者自身失败不影响结果一个关键设计onEvent处理器的返回类型是Effectvoid, never, RX不允许失败但若处理器内部主动抛出缺陷测试中用Effect.die(observer-defect)结果如何测试断言源效应/流的结果完全不受影响Exit.succeed事件仍完整配对发射[AttemptStart, AttemptSuccess]。这是因为内部实现用effect.ignoreCause(onEvent(event))包装处理器调用executionPlan.ts观察者的缺陷被忽略观测永远不会改变计划的执行结果。同理在 Stream 版本中测试也验证了观察者缺陷不改变流的结果、也不造成事件不配对。实践基于事件构建日志与指标结合上述语义一个典型的观测用法是把onEvent与 Effect 的日志、指标设施组合。由于处理器是Effectvoid, never, RX它可以使用Effect.log、Effect.metric等工具也可以依赖注入的服务其需求会并入最终效应import { Context, Effect, ExecutionPlan, Layer, Metric } from effect // 假设一个上报客户端作为服务注入 class Metrics extends Context.ServiceMetrics()(Metrics, { make: Effect.succeed({ record: (name: string, value: number) Effect.void }) }) {} const plan ExecutionPlan.make( { provide: Layer.empty, attempts: 2 }, { provide: Layer.empty } ) const program Effect.withExecutionPlan(work, plan, { onEvent: (event) Effect.gen(function*() { const metrics yield* Effect.service(Metrics) yield* metrics.record(plan.${event._tag.toLowerCase()}, 1) // 终端事件携带 duration可直接用于耗时指标 if (event._tag AttemptFailure) { yield* Effect.log(attempt ${event.attempt} failed on step ${event.stepIndex}, event.cause) } }) })需要注意处理器的RX需求会追加到结果效应上因此若处理器依赖Metricsprogram的类型会要求提供该服务与计划步骤自身需求的并集。此外由于处理器被内联等待且严格有序应避免在其中执行慢操作如同步网络请求以免拖慢被观测效应的吞吐。与流式场景的结合部分输出与preventFallbackOnPartialStreamStream.withExecutionPlan的onEvent语义与Effect版本一致Stream.ts 明确说明参见Effect.withExecutionPlan的处理程序语义但流场景有两个特有细节值得注意部分输出后的回退fallback after partial stream默认情况下一个已发出部分元素的失败步骤仍可回退到下一个步骤测试中[1,2,3]输出后失败再回退后总输出[1,2,3,1,2,3]事件仍为完整的AttemptFailure 下一次AttemptStart若设置preventFallbackOnPartialStream: true则部分输出后不再回退直接以该步骤的错误失败事件也仅有一对AttemptStart/AttemptFailure。下游提前停止消费当下游消费者提前停止拉取例如计划外的Stream.take时被截断的尝试会报告为AttemptSuccess——因为停止是消费者决定的并非源失败。底层机制速览事件发射的核心实现在 executionPlan.tsmakeEventEmitterL28-L84维护stepAttempt计数与时钟begin在尝试前发射AttemptStart并记录起始时间end在尝试结算时根据Exit发射AttemptSuccess成功或AttemptFailure携带完整cause并用monotonicTimeNanosUnsafe计算durationwithExecutionPlanL87-L188通过uninterruptibleMaskonExit把尝试与事件发射绑定L150-L155保证中断时终端事件仍发射观察者通过ignoreCauseL34隔离确保其缺陷不改变计划结果事件编号复用CurrentMetadata的attempt与stepIndex通过provideMeta更新L137-L146并在begin中按stepIndex变化重置stepAttempt保证了事件编号与运行中元数据的完全一致。小结onEvent让ExecutionPlan从只关心结果的黑盒变成每个尝试均可观测的透明流程AttemptStart/AttemptSuccess/AttemptFailure三个事件严格配对、编号与CurrentMetadata一致、cause完整覆盖错误/缺陷/中断、观察者本身不影响计划结果。这使其可以直接支撑基于执行计划的重试与故障切换场景下的日志审计、错误分类与耗时指标而无需侵入业务代码。相关类型定义、公开 API 与测试用例可分别在 ExecutionPlan.ts、Effect.ts、Stream.ts 与 ExecutionPlan.test.ts 中继续深挖。【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考