await本身不控制流向，真正决定流走向的是await后的条件判断和逻辑分支；应将清洗规则封装为返回语义化状态的promise函数，并用await+if/else串联调用，支持并行、短路、上下文透传与错误降级。
直接用 await 实现“动态分流”是个常见误解——await 本身不控制流向，它只暂停当前 async 函数执行，等待 Promise 解析。真正决定流走向的是你写在 await 后的条件判断和逻辑分支。关键在于把清洗决策建模为可 await 的状态计算，再用自然的 if/else 或 switch 组织后续节点调用。
把清洗规则变成可 await 的状态函数
不要让清洗逻辑散落在回调或 then 链里。每个清洗节点应封装为返回 Promise 的函数，其 resolve 值是明确的状态标识（如 "valid"、"missing_phone"、"suspicious_amount"），而不是直接操作数据或跳转。
- 例如：
async function validateContact(record) { ... return record.phone ? "valid" : "missing_phone"; } - 再如：
async function checkAmount(record) { ... return Math.abs(record.amount) > 1e6 ? "suspicious_amount" : "normal"; } - 这些函数内部可含 I/O（查码表、调风控 API）、计算或缓存读取，但对外只输出语义化状态
用 await + 条件语句串联清洗路径
主清洗流程用 async 函数编写，每步 await 一个状态函数，然后根据返回值决定下一步调哪个清洗节点——这比硬编码 pipe 链更灵活，也比事件总线更轻量。
const contactStatus = await validateContact(record);if (contactStatus === "missing_phone") { await enrichPhone(record); }else if (contactStatus === "valid") { await normalizeAddress(record); }const amountStatus = await checkAmount(record);if (amountStatus === "suspicious_amount") { await triggerReview(record); }
支持多路并行与短路退出
某些判断可并行执行（如同时校验手机号和邮箱格式），用 Promise.allSettled 汇总结果；而高危状态（如 "fraud_flag"）应立即终止后续清洗，避免冗余处理。
const [phoneRes, emailRes] = await Promise.allSettled([validatePhone(r), validateEmail(r)]);if (phoneRes.status === "rejected" || emailRes.status === "rejected") throw new ValidationError("contact_invalid");- 遇到
throw或显式return，后续 await 不再执行，天然实现短路
保持上下文透传与错误隔离
每个清洗节点接收统一结构的上下文对象（如 { record, metadata, logger }），而非原始 record。这样状态函数可往 context 中追加字段（如 context.flags = ["enriched_phone"]），供下游节点感知，避免重复判断。
- 错误不抛到顶层，而是捕获后转为状态：
try { await callRiskApi(r); } catch { return "risk_api_unavailable"; } - 这样即使某个节点失败，管道仍可按降级策略继续（如 API 不可用时走规则兜底）
- 所有节点共用同一 context 引用，状态累积可见，无需全局变量或 event bus
