Chapter 03

实操：从抓取到并发调度

02 章讲清了机制：错误如何被类型追踪、依赖如何编译期注入、fiber 如何调度与中断。这一章把它们变成会跑的代码——三阶递进：完整 worked 示例 → 留决策点的 partial → 端到端的开放练习。每一步都指回前两章的概念。

本章你将建立的 schema

把一个会 reject 的 fetch 改写成类型化错误的 Effect（落地 01.6 + 01.3）
用 Context.Tag + Layer 注入并替换实现（落地 01.7 + 02.3）
写出有界并发 + 单项超时 + 失败隔离的调度（落地 02.4 + 02.5）

3.0环境准备

Effect 要求 TypeScript 5.5+ 且开启 strict——类型推断重度依赖它。下面在 Node 24 上跑通。

mkdir effect-lab && cd effect-lab
npm init -y
npm i effect
npm i -D tsx typescript   # tsx 直接跑 .ts；需要 TS 5.5+

# tsconfig.json 至少要有：
#   "strict": true
#   "target": "ES2022"   （Effect.gen 用到现代生成器）
npx tsx hello.ts

常见环境失败

① strict 没开 → 类型推断退化，R/E 经常被推成 unknown，编译信息一团乱。② TS 低于 5.5 → 直接 yield* effect 报错（旧版需已废弃的 _ 适配器）。两者都先检查再排查代码。

3.1示例 1 · Worked：类型化抓取（完整）

目标：把"会失败的网络抓取"包成 Effect<Weather, HttpError>，失败进入 E 通道而不是 reject。这是后面一切的地基。

图 3.1同一份数据，上排是动作、下排是类型。注意：最后一格把失败也收进了类型（HttpError 进 E 通道）——这正是和 Promise<Weather> 的区别：失败不再隐身。

import { Effect, Data } from "effect"

// ① 类型化错误（落地 01.6）
class HttpError extends Data.TaggedError("HttpError")<{
  readonly status: number
  readonly url: string
}> {}

interface Weather {
  readonly city: string
  readonly tempC: number
}

// 真实场景这里是网络调用；用假数据模拟一个返回 Promise 的抓取
const rawFetch = (city: string): Promise<Weather> =>
  Promise.resolve({ city, tempC: 21 })

// ② tryPromise 把会 reject 的 Promise 包进 Effect，reject 转成 HttpError（落地 01.3）
const fetchWeather = (city: string): Effect.Effect<Weather, HttpError> =>
  Effect.tryPromise({
    try: () => rawFetch(city),
    catch: () => new HttpError({ status: 0, url: `weather/${city}` }),
  })

// ③ pipe + map 组合（落地 01.5）
const program = fetchWeather("Tokyo").pipe(
  Effect.map((w) => `${w.city}: ${w.tempC}°C`),
)

// ④ 在最外层 run（落地 01.4）
Effect.runPromise(program).then(console.log) // "Tokyo: 21°C"

逐行映射回概念

① Data.TaggedError：定义带 _tag 的错误 → 01.6。它让失败有了可被 catchTag 区分的身份。
② Effect.tryPromise：边界闸门 → 01.3。catch 的返回类型 HttpError 决定了 E 通道。
③ pipe(Effect.map(...))：组合 → 01.5。注意此刻什么都没跑，program 仍是值 → 01.1。
④ runPromise：唯一边界 → 01.4。

想一想

fetchWeather("Tokyo") 这一行执行了网络抓取吗？

展开答案（先停 10 秒）

没有。它只是构造了一个 Effect 值（01.1 惰性）。真正的抓取发生在 Effect.runPromise(program) 那一刻。把 fetchWeather("Tokyo") 赋给一个变量、放进数组、传给别的函数，都不会触发抓取。

3.2示例 2 · Partial：注入服务（留决策点）

把"怎么抓取"从业务里抽出来，做成一个 HttpClient 服务。这样测试时给假实现、生产给真实现，业务代码一行不改。下面的模板留了两个关键决策点——不是无关填空，每个都对应一个设计选择。

import { Effect, Context, Layer, Data } from "effect"

class HttpError extends Data.TaggedError("HttpError")<{ status: number }> {}

class HttpClient extends Context.Tag("app/HttpClient")<
  HttpClient,
  // ───── 决策点 ① ─────
  // get 接收 url，应该返回什么类型的 Effect？
  // 提示：它可能以 HttpError 失败，调用方需要在签名里看见这一点。
  { readonly get: (url: string) => ___①___ }
>() {}

const fetchWeather = (city: string) =>
  Effect.gen(function* () {
    const client = yield* HttpClient            // R 出现 HttpClient
    return yield* client.get(`weather/${city}`)
  })

// 测试用实现：不联网，直接给假数据
const HttpClientTest = Layer.succeed(HttpClient, {
  get: (_url) => Effect.succeed({ city: "Test", tempC: 0 }),
})

// ───── 决策点 ② ─────
// 让 fetchWeather 能 run —— R 必须从 HttpClient 变成 never。怎么做？
const runnable = fetchWeather("Tokyo").pipe( ___②___ )

Effect.runPromise(runnable).then(console.log)

先自己填，再看答案

决策点 ① 该填什么类型？决策点 ② 该用哪个算子？

参考答案 + 决策路径

① Effect.Effect<unknown, HttpError>。A 用 unknown 让客户端保持通用（调用方再 decode）；关键是 E 必须写上 HttpError——否则调用方的 fetchWeather 签名里看不到"会失败"，02.2 的类型追踪就断了。

② Effect.provide(HttpClientTest)。fetchWeather("Tokyo") 的类型是 Effect<unknown, HttpError, HttpClient>，R 还挂着 HttpClient，不 provide 就编译不过（02.3：R 不为 never 不能 run）。换成生产实现，只需把这里换成 Effect.provide(HttpClientLive)——业务代码零改动。

洞察 · 这就是可测试性

同一个 fetchWeather，测试里 provide(HttpClientTest)、生产里 provide(HttpClientLive)。依赖在最外层注入，业务逻辑只依赖 Tag 这个接口——这是 02.3 那张备选方案表里"选中"那一行的实际收益。

3.3示例 3 · 开放练习：并发工具调度

这一题贴近真实的 Agent 后端：并发跑一批工具，每个有超时，单个失败不能炸掉整批，超时的工具要被干净地取消。要求用到 02.4（有界并发）+ 02.5（超时/中断）+ 02.2（用 Exit 隔离失败）。

图 3.2三阶的脚手架（深色"已给"）从上到下递减，你写的部分（红色）递增。注意：递减的是"代码给到多少"，递增的是"你做多少判别决策"——到示例 3，框架只剩签名，决策全在你。

import { Effect, Data, Exit } from "effect"

class ToolError extends Data.TaggedError("ToolError")<{
  readonly tool: string
  readonly reason: string
}> {}

interface Tool {
  readonly name: string
  readonly run: Effect.Effect<string, ToolError>
}

// 任务：并发跑所有 tool，满足——
//   · 并发上限 2（别一次打爆下游）
//   · 每个 tool 最多等 2 秒，超时即取消该 tool
//   · 单个 tool 失败/超时，不影响其他 tool 的结果
//   · 返回每个 tool 的 { name, exit }
const runTools = (tools: ReadonlyArray<Tool>) => {
  // 你的实现
}

验收 checklist（可观测的行为）

给 3 个 tool、并发设 2 时，第三个要等前两个之一完成才开始（不是一拥而上）。
一个 tool 故意 Effect.sleep("5 seconds")，它在 2 秒时被取消，其余 tool 正常返回。
一个 tool 故意 Effect.fail，整批不抛、不中止，该 tool 的 exit 是 Failure，其余是 Success。

参考实现（自己写完再展开）

import { Effect, Data, Exit, Console } from "effect"

class ToolError extends Data.TaggedError("ToolError")<{
  readonly tool: string
  readonly reason: string
}> {}

interface Tool {
  readonly name: string
  readonly run: Effect.Effect<string, ToolError>
}

const runTools = (tools: ReadonlyArray<Tool>) =>
  Effect.all(
    tools.map((t) =>
      t.run.pipe(
        Effect.timeout("2 seconds"),   // 02.5：超时即中断该 fiber，清理照跑
        Effect.exit,                   // 02.2：成功/失败/超时都收成 Exit，失败被隔离
        Effect.map((exit) => ({ name: t.name, exit })),
      ),
    ),
    { concurrency: 2 },                 // 02.4：有界并发，最多 2 个在飞
  )

// --- 演示 ---
const tools: Tool[] = [
  { name: "fast", run: Effect.succeed("ok").pipe(Effect.delay("100 millis")) },
  { name: "slow", run: Effect.succeed("late").pipe(Effect.delay("5 seconds")) },
  { name: "boom", run: Effect.fail(new ToolError({ tool: "boom", reason: "x" })) },
]

const main = runTools(tools).pipe(
  Effect.tap((rows) =>
    Effect.forEach(rows, (r) =>
      Console.log(`${r.name}: ${Exit.isSuccess(r.exit) ? "OK " + r.exit.value : "FAILED"}`),
    ),
  ),
)

Effect.runPromise(main)
// fast: OK ok
// boom: FAILED               （fail 被隔离在自己的 Exit 里）
// slow: FAILED               （2 秒超时被中断，没拖到 5 秒）

关键决策说明：用 Effect.exit 而不是直接 Effect.all 收集——因为 Effect.all 默认一个失败就整批短路并中断其余（fail-fast）。把每个 tool 先 exit 成值，失败就被"降级"成数据，整批永远成功，逐项检查 Exit。这正是 02.2"错误是值"的实战用法。

§本章 self-check

先合上教程作答，再展开对照。

示例 1 里，把 Effect.tryPromise 换成 Effect.promise 会丢掉什么？
示例 2 决策点 ② 不写 provide 会怎样——运行时报错还是编译报错？
示例 3 里为什么必须用 Effect.exit，直接 Effect.all 收集结果会怎样？

答案（先做完再展开）

丢掉错误类型。Effect.promise 假定 Promise 永不 reject，E 通道是 never；真 reject 时会变成缺陷（02.2），绕过 catchTag。会失败的抓取必须用 tryPromise。
编译报错。runnable 的 R 还是 HttpClient，runPromise 要求 R=never，类型不匹配——编译期就被挡下（02.3：这是特性）。
Effect.all 默认 fail-fast：第一个失败的 tool 会让整批短路、中断其余、整体以该错误失败。用 Effect.exit 先把每项失败降级成 Exit 值，整批才不会被单个失败炸掉。

进阶挑战 · 刚好够不着

给整批加一个总预算

在示例 3 基础上：除了每个 tool 各自 2 秒超时，再给整批一个 5 秒总预算——5 秒到了，还没跑完的 tool 全部取消，已完成的结果照常返回。signature 不变。

提示（卡住再展开）

给 runTools(tools) 整体再套一层。但直接 Effect.timeout("5 seconds") 超时会让整个 Effect 失败、丢掉已完成结果。换个思路：用 Effect.timeoutTo（超时返回一个降级值而非失败），或先 Effect.exit 整批再 race 一个 5 秒 sleep。核心是区分"超时=失败"和"超时=拿已有的部分结果"。