从零实现持久化执行：先把工作流变成可验证的纯状态机

为什么不是再写一个任务队列普通任务队列擅长“把函数放到另一台机器执行”却很难回答更麻烦的问题进程在扣款之后、写订单之前崩溃重启后究竟应该从哪里继续如果把整个函数重新跑一遍扣款可能发生两次如果直接标记失败已经发生的副作用又无人处理。持久化执行的核心不是更快地调度函数而是让一次跨越数分钟、数天甚至数月的业务过程在任意中断后仍能从确定位置恢复。本系列贯穿项目叫miniflow。我们用一个“旅行预订”流程持续演进锁定预算、预订航班、预订酒店、最终确认。后续会故意加入崩溃、重复投递、并发抢占、长定时器、补偿以及代码升级。第一篇暂时不碰数据库和网络先解决最容易被低估的基础把“下一步做什么”从一串隐含在调用栈里的控制流变成显式、可测试的数据变换。许多工作流系统的第一个坑是一上来就写执行器。执行器里同时查状态、调接口、捕获异常、推进游标最终得到一个无法重放的巨大循环。更稳妥的边界是决策函数只读取历史状态和一个事件返回新状态以及待执行命令真正的网络调用由外层执行。这样决策层没有时间、随机数、网络和全局变量相同输入永远得到相同输出。这个性质叫确定性也是以后恢复的地基。建模事件、命令与不可变状态事件描述“已经发生什么”例如预算已锁定命令描述“希望外部做什么”例如请求预订航班。两者不能混用。如果决策函数自己调用航司 API重放历史时就会再次订票如果它只产生book_flight命令重放时便能识别这条命令过去已经完成。状态使用冻结的数据类迫使每次转移构造新值避免测试之间共享可变对象。把下面内容保存为miniflow.py。decide是本系列最重要的函数后九篇都会复用或扩展它。注意未知事件不是默默忽略而是立即失败。悄悄吞掉事件会让旧版本程序看似运行实际却在错误状态上继续推进这是事件驱动系统里很危险的“兼容性幻觉”。fromdataclassesimportdataclass,replacefromtypingimportNamedTupledataclass(frozenTrue)classTripState:trip_id:strphase:strnewbudget_token:str|NoneNoneflight_ref:str|NoneNonehotel_ref:str|NoneNoneclassEvent(NamedTuple):kind:strdata:dictclassCommand(NamedTuple):kind:strdata:dictdefdecide(state:TripState,event:Event)-tuple[TripState,list[Command]]:ifstate.phasenewandevent.kindtrip_requested:newreplace(state,phaselocking_budget)returnnew,[Command(lock_budget,{trip_id:state.trip_id})]ifstate.phaselocking_budgetandevent.kindbudget_locked:newreplace(state,phasebooking_flight,budget_tokenevent.data[token])returnnew,[Command(book_flight,{trip_id:state.trip_id})]ifstate.phasebooking_flightandevent.kindflight_booked:newreplace(state,phasebooking_hotel,flight_refevent.data[ref])returnnew,[Command(book_hotel,{trip_id:state.trip_id})]ifstate.phasebooking_hotelandevent.kindhotel_booked:newreplace(state,phaseconfirmed,hotel_refevent.data[ref])returnnew,[]raiseValueError(finvalid transition:{state.phase}{event.kind})运行输出模块定义成功无标准输出用一条历史证明状态转移下面的脚本不是“演示式伪代码”而是可以直接运行的最小测试驱动器。保存为demo_101.py与miniflow.py放在同一目录。它明确列出历史事件每次把上一步状态作为下一步输入并打印新阶段和命令。命令只被打印没有真正执行这恰好体现决策与副作用的分界。fromminiflowimportTripState,Event,decide history[Event(trip_requested,{city:成都}),Event(budget_locked,{token:B-7}),Event(flight_booked,{ref:F-8}),Event(hotel_booked,{ref:H-9}),]stateTripState(trip-001)emitted[]forposition,eventinenumerate(history,start1):state,commandsdecide(state,event)emitted.extend(command.kindforcommandincommands)print(position,event.kind,,state.phase,[command.kindforcommandincommands])assertstate.phaseconfirmedassertstate.flight_refF-8assertstate.hotel_refH-9assertemitted[lock_budget,book_flight,book_hotel]print(final:,state)运行输出1 trip_requested locking_budget [lock_budget] 2 budget_locked booking_flight [book_flight] 3 flight_booked booking_hotel [book_hotel] 4 hotel_booked confirmed [] final: TripState(trip_idtrip-001, phaseconfirmed, budget_tokenB-7, flight_refF-8, hotel_refH-9)为什么纯函数边界值得坚持第一它让失败语义清楚。事件进入之前世界没有变化事件成功应用之后新状态和命令一起产生。虽然 Python 返回两个值不是数据库事务但这个结构为下一篇把两者原子写入 SQLite 留出了位置。第二它让测试组合简单。无需启动 worker、伪造 HTTP 或等待定时器只要枚举事件序列即可覆盖业务分支。第三它让历史具备解释力。看到flight_booked就知道航班副作用曾成功而不是从某个布尔字段猜测。一个非平凡踩坑是“把当前阶段当成唯一事实”。阶段只是历史的折叠结果不能替代历史。假设状态显示booking_hotel我们仍不知道航班命令是否发送过两次、哪次成功、响应是什么。当前篇用阶段保障合法转移下一篇则保存每个事实。另一个坑是事件负载随意使用字典。字典方便起步却会把拼写错误推迟到运行期工程中应给事件做版本化校验。我们会在第八篇专门处理模式升级现在先保持依赖为零让决定性边界一眼可见。还要记住确定性不等于“永远按同一业务结果运行”。航司价格当然会变化但变化应作为一次外部活动的结果写入事件然后成为确定历史的一部分。决策函数不能在重放时重新查询价格、读取当前日期或生成随机 ID。若确实需要这些值就发出命令让执行器获取后以事件返回。这条规则看似拘谨实际是持久化执行能够跨崩溃延续的根本。本篇产出与下一步现在我们有了miniflow.py中的TripState、Event、Command和decide以及一条可重复验证的旅行历史。它仍然只活在内存里进程退出就消失。下一篇会原样复用这些类型和函数把输入事件与产生的命令在一个 SQLite 事务里追加到日志得到第一个真正可恢复的持久化核心。 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。