diff --git "a/fluxon_doc_cn/blog/blog_2_Fluxon KV \346\212\200\346\234\257\346\267\261\345\272\246\350\247\243\346\236\220\357\274\232\350\247\222\350\211\262\345\210\206\345\261\202\343\200\201\350\257\273\345\206\231\346\227\266\345\272\217\344\270\216\345\271\266\345\217\221\346\216\247\345\210\266.md" "b/fluxon_doc_cn/blog/blog_2_Fluxon KV \346\212\200\346\234\257\346\267\261\345\272\246\350\247\243\346\236\220\357\274\232\350\247\222\350\211\262\345\210\206\345\261\202\343\200\201\350\257\273\345\206\231\346\227\266\345\272\217\344\270\216\345\271\266\345\217\221\346\216\247\345\210\266.md" new file mode 100644 index 0000000..56a790f --- /dev/null +++ "b/fluxon_doc_cn/blog/blog_2_Fluxon KV \346\212\200\346\234\257\346\267\261\345\272\246\350\247\243\346\236\220\357\274\232\350\247\222\350\211\262\345\210\206\345\261\202\343\200\201\350\257\273\345\206\231\346\227\266\345\272\217\344\270\216\345\271\266\345\217\221\346\216\247\345\210\266.md" @@ -0,0 +1,559 @@ +# Fluxon KV 深度解析:面向 AI 数据面的分层架构、通信优化与可扩展设计 + +## 一、背景:AI 数据面下的 KV 挑战 + +讨论分布式 KV 系统在 AI 数据面里的演进,先要拆清两个不同的 `KV`:传统系统里的 Key-Value,和 Transformer Attention 里的 Key/Value Cache。这里的 `KV Cache` 指模型推理时由 QKV Attention 产生并在后续 token 中复用的 Key/Value 状态,它和通用 Key-Value 存储不是同一个概念。 + +Mooncake / `MooncakeStore` 是近几年最有代表性的 `KV Cache` 场景分布式 KV 系统;它以 Kimi 线上服务为背景,证明了长上下文推理里的瓶颈已经不只是 GPU 算子本身,`KV Cache` 池化、放置和传输也会直接决定服务吞吐和 SLO。其论文 [Mooncake: Trading More Storage for Less Computation — A KVCache-centric Architecture for Serving LLM Chatbot](https://www.usenix.org/conference/fast25/presentation/qin) 获得了 `FAST '25` Best Paper。 + +Fluxon KV 沿着这个方向继续往前推进。Mooncake 从 `KV Cache` 池化、全局调度和 Transfer Engine 视角打开了长上下文推理的数据面问题;Fluxon KV 在类似的大对象复用背景下,把问题继续推到 Python 多进程 AI 服务的工程底座里。 + +实际尝试使用一些开源方案时,我们依旧遇到很多问题:Python worker 崩溃时,共享内存和本机数据面容易被一起拖下水,甚至出现 core dump、状态机无法复位等正确性和稳定性问题;底层 native 组件或通信层异常时,业务进程、缓存上下文和跨节点连接会相互干扰;每个业务进程都参与全量 P2P 时,节点数和 worker 数一上来,连接拓扑很快膨胀;通信层如果不能同时支持 TCP 和 RDMA,并在 RDMA 不稳定时稳定兜底,长稳运行就很难;RDMA 如果只耦合在数据面传输里,控制面 RPC 仍然吃不到同一套低延迟通信能力;系统重启后,成员发现、路由、缓存失效和本机资源也需要能自恢复收敛。 + +### 1. 现有 KV / KV Cache 路径在 AI 场景的缺口 + +在传统业务系统里,Redis 这类远端 KV 已经把业务进程和存储服务拆开了,但它的主路径基本是 TCP 网络通信:写入一个 `key`,读取一段 value,再配上过期时间和容量上限。这个模型适合高并发小对象访问,但没有为同机多进程共享 MB 级 tensor 对象提供共享内存快路径。 + +`MooncakeStore` 这类面向 `KV Cache` 的系统进一步证明了大对象缓存池化的价值,重点展示了多机缓存池化和跨机复用带来的收益。把视角放到 Python 多进程 AI 服务和更通用的数据面时,工程架构上仍有更细粒度的优化空间: + +- **同机快路径不足**:Redis 这类远端 KV 天然走网络路径,`MooncakeStore` 这类外部缓存路径也更偏跨机复用;当同一台机器上的多个 Python worker 读取同一个 MB 级对象时,如果缺少共享内存和 mmap 级别的本机交接,对象仍可能在多个进程地址空间里反复 materialize。 +- **业务进程和数据面生命周期耦合**:如果共享内存管理、传输状态或缓存副本直接绑在 Python worker 生命周期上,worker 异常退出、滚动升级或弹性扩缩容都会扰动底层数据面。例如一次 segment 生命周期错误、非法内存访问或 native 组件崩溃,可能同时影响业务请求、缓存上下文和本机数据面资源。 +- **连接扩散**:每个业务进程如果都直接参与跨节点互联,进程数上升会直接放大连接拓扑。例如同一台机器上启动 8 个推理 worker,如果每个 worker 都和远端节点建连接,跨机连接数会按 worker 数膨胀;扩缩容、滚动升级或异常重启时,控制面也要处理更多成员变更和连接抖动。 +- **资源治理割裂**:框架级缓存和外部缓存经常同处单机 CPU 内存,但它们通常是两个分离实现的系统,各自为了隔离复杂度维护独立的索引、放置和驱逐逻辑。例如 `SGLang` 的 `L2` 保留框架内索引以提供最低延迟访问,作为 `L3` 的 `MooncakeStore` 负责跨机复用;两者没有天然互通的资源视图,`L2` 内存也难以进入统一索引、放置和驱逐治理,进而增加缓存穿越、对象交接开销和冗余驻留带来的内存浪费。 +- **缓存重复驻留**:每个业务进程如果都维护自己的本地副本,热点对象会在同机内反复驻留。例如 `MooncakeStore` 这类外部缓存路径缺少本机共享内存快路径时,同一台机器上的多个 worker 多次 `get` 同一个对象,数据可能被 materialize 到多个进程各自的地址空间里。这个复制和重复驻留开销来自本机进程间共享能力不足,业务语义本身并不要求付出这部分成本。 + +### 2. AI workload 的数据面特征 + +这些问题在 AI workload 里集中爆发。对象形态和运行方式已经变了,传统小对象 KV 的直觉很容易失灵。 + +- **对象大**:常见形态包括 `KV Cache`、latent cache、中间态 tensor、消息 payload 和文件切片。控制面不该直接搬这些 payload,也不该让它们在多个进程或多套缓存层级里反复复制。 +- **对象热**:同一个 `key` 可能被大量并发请求读取。cache miss 需要合并,副本扩散需要受控,否则热点读取很容易打成远程回源风暴。 +- **单机业务进程数受 GPU 卡数约束**:AI 服务里的 worker 经常独占单卡或多卡,单机业务进程数通常不会无限扩张。但每个进程手里的对象都很重,同机 worker 之间又有复用需求。优化重点会转向让少量高价值进程共享同一套本机数据面对象。 +- **跨边界流动**:数据会跨请求、跨进程、跨节点、跨子集群流动。单进程内存对象的模型已经不够用,系统必须显式处理放置、传输和路由。 +- **生命周期长短不一**:有些对象只是一次读取的临时结果,有些对象需要被 lease 保活,有些 holder 仍被用户代码持有。系统必须说清楚:谁持有、何时可见、何时释放。 +- **Python 多进程是常态**:很多语言栈会优先把并发压在一个进程内,用线程或协程减少切换和内存复制;Python 受 `GIL` 和 native 扩展边界影响,AI 服务为了并行度、隔离性和 GPU worker 编排,经常回到多进程模型。Python worker、Producer / Consumer、推理服务实例都可能频繁启停,数据面容量和跨节点连接不能跟着业务进程数量线性膨胀。 + +想象一下,如果把这些脏活累活全塞进 Python 业务进程会怎样?每个 worker 都要自己维护路由、建连接、管副本、做回收,还要兜住共享内存和 native 组件的异常。进程数一多,连接拓扑直接膨胀;缓存各管各的,同机内存被重复驻留塞满;缓存层级再一多,CPU 内存又被切成几套难以统一治理的资源池。 + +这条路走不通。Fluxon KV 的破局思路是把职责切开:业务进程只负责接入,本机常驻 `owner` 负责共享内存、allocation 和跨机传输,`master` 负责全局权威状态。让业务的归业务,数据面的归数据面。 + +### 3. 核心洞察 + +> 核心判断很直接:AI 数据面里的 KV 系统,不能只当远端字典看。它必须同时治理大对象、热点回源、跨边界路由、生命周期、同机复用和最终一致性边界。 + +- **数据特征是不可变、MB 级大对象**:`KV Cache`、latent cache 和中间态 tensor 提交后通常不再原地修改,所以系统可以把控制面元数据和真实 payload 拆开,重复写入也可以在进入大对象传输前尽早收敛。 +- **热点读取要先折叠再回源**:同一个 `key` 可能被大量并发请求读取,所以并发读请求需要按 key 合并,已存在的对象需要直接复用,避免热点 miss 打成远程回源风暴。 +- **单机 worker 规模有上限,中心化 `master` 可接受**:控制面很重要,要做 allocation 决策和 KV 路由;但它只处理元数据,不搬 payload。AI 数据面里的对象通常是 MB 级,单机 worker 数又受 GPU 卡数约束,一次中心化决策的成本相对 payload 传输可控,也为后续按负载、热度、GPU 拓扑和子集群做放置优化留下入口。 +- **角色分层解决生命周期耦合**:`external` 可以随业务进程弹性启停,`owner` 保持共享内存、allocation 和跨机连接常驻,`master` 保留全局版本和生命周期权威。 +- **同机复用优先压低重复驻留**:多个 `external` 共享本机 `owner` 的 mmap、allocation 和本地副本缓存,热点对象不需要在每个业务进程里各自 materialize 一份。 +- **最终一致性模型适合 KV Cache**:上下文通常是反复读取、持续增长、提交后不可变的大对象,读路径对性能敏感;异步失效、副本提升和批量异步延迟删除,比强同步清理更契合这个访问模型。 + +后文沿这组判断展开:第二章先说明为什么要拆成 `master / owner / external`,为什么 allocation 决策和 KV 路由可以收敛到中心化 `master`,以及为什么同机共享内存应该收敛到常驻 `owner`;第三章说明不可变大对象如何对应到分段读写、请求合并和本地复用;第四章先把 `Allocation / Segment / Holder` 生命周期链路讲清楚;第五章说明最终一致性、异步失效和延迟删除如何进入特化策略接口;第六章给出项目入口和近期推理路径演进。 + +--- + +## 二、架构破局:Master-Owner-External 三层分离 + +生命周期解耦、中心化控制面和同机复用这几条判断,都指向同一个架构动作:把 `master / owner / external` 拆开。`external` 只做业务接入,`owner` 常驻承载本机共享内存和跨机连接,`master` 保留全局版本、allocation 决策和生命周期权威。 + +可以把 Fluxon KV 类比成一个现代物流系统: + +- `master` 是全局调度中心,只负责记账、发号、路线规划和生命周期登记,不亲自搬运 payload。 +- `owner` 是本地常驻仓库和物流站,持有共享内存、allocation、本地副本缓存和跨节点传输能力,负责对象的实际存取与搬运。 +- `external` 是业务方或取件人,可以随请求、worker 或服务实例动态出现,只向本地 `owner` 下单,不自己维护仓库和车队。 + +### 2.1 架构全景图 + +先看全局形态。这张图要抓住 payload 和控制面的边界:`master` 只做调度和生命周期登记,真实对象停在 `owner` 数据面。 + +```mermaid +flowchart TB + subgraph EXTERNALS[external clients] + direction LR + E1[external\n业务进程 A] + E2[external\n业务进程 B] + E3[external\n业务进程 C] + end + + subgraph OWNERS[owner 间互联] + direction LR + O1[owner A\n共享内存与 allocation] + NET[TCP / RDMA\n协议动态适应] + O2[owner B\n共享内存与 allocation] + end + + M[master\n版本路由与生命周期] + ETCD[(etcd)] + + E1 --> O1 + E2 --> O1 + E3 --> O2 + + O1 <--> NET + NET <--> O2 + + O1 --> M + O2 --> M + + M -.成员发现与 readiness.-> ETCD +``` + +### 2.2 分层架构:角色边界与职责收口 + +这里的 `allocation` 指 `owner` 管理的内存分配对象,`holder` 指 `get` 返回后约束对象生命周期的用户可见引用。 + +如果继续沿用前面的物流类比,可以这样理解: + +| 概念 | 物流类比 | 工程含义 | +| --- | --- | --- | +| `master` | 调度中心 | 不碰货,只发“入库单 / 提货单”,记录哪个 `owner` 有哪个 key-version | +| `owner` | 本地仓储和物流站 | 真正管理货架、mmap 内存、本地副本和跨城物流 | +| `external` | 取件人 / 业务方 | Python 业务进程,只拿取件码和本机入口,不掌握仓库钥匙 | +| `allocation` | 货架预留 | 控制面拍板“这块 slot 归这次 put/get 用” | +| `segment lease` | 运输保价单 | 跨库 transport 期间托住本地 segment,避免货还在路上货架先被拆 | +| `holder` | 取件码 | 业务代码持有的借用凭证,释放前对象不能被系统回收 | + +| 角色 | 持有什么 | 不承担什么 | 主要收益 | +| --- | --- | --- | --- | +| **`master`** | `kv_routes`、`inflight_puts`、`inflight_gets`、allocation 预留与回收权威、lease 绑定、holder 生命周期、delete 广播入口 | 业务 payload bytes | 放置、allocation、版本、lease 和生命周期决策有单一权威入口 | +| **`owner`** | 共享内存 segment、mmap、allocation、本地副本缓存、owner 侧 holder | 业务进程私有状态 | 本机共享内存和副本缓存常驻,业务进程重启不会直接扰动本机数据面容量 | +| **`external`** | 业务入口、本地弱缓存、pending put 上下文、external holder | 集群容量贡献、owner-owner 互联 | Python 业务进程可以轻量接入和弹性扩缩容 | + +这套角色分层先解决三个归属问题:allocation 决策归 `master`,本机共享内存和副本缓存归 `owner`,业务入口和用户可见 holder 归 `external`。落到实现上,有几个直接的技术点: + +- **allocation 决策和 KV 路由中心化,内存承载在 `owner`**:`master` 在 `PutStart` / `GetStart` 里一次控制面 `RTT` 同时完成目标 `owner` 选择、allocation 预留和当前 KV 路由更新,避免请求方先用一致性哈希定位节点、再远端试分配、失败后重试的多轮交互。AI 数据面里的对象通常是 MB 级大对象,一次中心化调度的控制面成本相对 payload 搬运成本更可控;保留中心化调度入口,也为后续按负载、热度、拓扑、GPU 位置或子集群做更复杂的放置优化留下空间。`owner` 承载真实的共享内存 segment、mmap 和本地副本缓存,同机多个 `external` 通过同一套 owner 本机对象访问热点数据。 +- **payload 不经过 `master`**:`master` 只做控制面决策,传输发生在 `owner` 和 transfer engine 路径上。 +- **共享内存由 `owner` 承载**:同机 `external` 复用 `owner` 的 mmap,不需要每个业务进程各自维护一份数据面对象,从而提升同机热点对象的缓存命中率,并让命中路径接近内存级访问性能。 + +### 2.3 通信平面 + +通信优化和前面的三层切分是相辅相成的。`master / owner / external` 先把跨机互联和数据面常驻资源收敛到 `owner`,减少业务进程直接参与的通信面;通信平面再负责让这些角色完成成员发现、协议选择和连接维护。Fluxon 用 `etcd` 存储成员元数据,让拓扑发现、协议选择和连接状态从业务进程里拆出来。`master`、`owner`、`external` 都通过 `ClusterManager` 注册成员信息,元数据里包含角色、`local_ipc_root`、`shared_storage_node_id`、`rdma_control` 等字段。 + +这部分有几个值得展开的技术点: + +- **用成员元数据驱动连接规划**:每个 member 都能从 `etcd` 观察当前成员集合和关键元数据,形成一份可用于连接规划的拓扑快照。成员表里记录角色、本机 IPC 根目录、共享存储绑定关系和 RDMA 控制面配置,让通信层可以基于同一份元数据决定本机 IPC、跨机 TCP / RDMA,或 relay / forwarding。 +- **把跨机连接收敛到 `owner`**:`external` 只附着到本机 `owner`,不参与 owner-owner 网状互联。跨节点连接主要发生在 `owner` 之间,业务进程从 1 个扩到 8 个时,不会把跨机连接和远端路由状态同步放大 8 倍。 +- **同机入口走共享内存和 `iceoryx2`**:`external` 启动时 attach 到 `owner` 发布的共享内存 bundle,value payload 本体主要通过 mmap 暴露;控制消息和本机入口走 `iceoryx2`。`iceoryx2` 是面向同机进程间通信的 Rust IPC / shared-memory 通信库,适合把本机控制消息和对象交接留在共享内存路径内,避免同机对象交接绕行跨机传输协议栈。 +- **跨机路径复用 `P2pModule + transfer_engine`**:`owner <-> master` 和 `owner <-> owner` 统一走 `fluxon_commu` 的 `P2pModule + transfer_engine`。底层按部署选择 TCP 或 RDMA,并可按成员元数据和直连条件动态调整;RDMA 不只服务 payload 传输,也可以承载控制面 RPC 和数据面传输两类通信路径。能走 busy polling 的路径优先 busy poll,业务进程不需要各自维护一套传输轮询和连接状态。 +- **协议切换和中继收敛在通信层**:业务进程只接入本机 `owner`;跨节点直连、协议切换、中继转发都由 `owner` 和 `fluxon_commu` 处理,避免把网络拓扑判断和失败重试逻辑扩散到 Python 业务进程里。 + +### 2.4 强大的 Rust 生态与并发底座 + +角色分层和通信平面能保持相对克制的实现复杂度,也依赖 Rust 生态里成熟的高并发基础库。Fluxon 不需要为每个热路径状态手写一套容器和队列,可以把不同类型的并发状态放到合适的现成结构上: + +- **`DashMap` / `DashSet`**:用于高并发访问的 peer 表、传输状态、缓存索引和路由类状态,减少全局锁争用。 +- **`crossbeam`**:用于线程间 channel、队列和 cache padding 等基础并发结构,支撑 relay worker、传输完成队列和后台任务协作。 +- **`moka`**:用于高并发缓存和容量控制,例如 `inflight_puts` / `inflight_gets` 这类在途状态表,以及节点侧副本缓存控制器。它让短生命周期状态和缓存淘汰逻辑不用重新实现一套。 +- **`sharded-slab`**:用于 RPC pending call 表。P2P RPC 的 wire `task_id` 直接来自 slab key,包括 generation bits,让 pending call 的分配、查找和释放落在一个并发友好的权威表里。 + +下面这组结构节选能看到 Fluxon KV 的核心状态形状:慢路径在途状态单独放,稳定路由保存当前 key-version 的副本集合和 lease 绑定,派生索引异步维护。 + +```rust +#[derive(Clone)] +pub struct InflightPutInfo { + pub node_id: NodeID, + pub key: String, + pub req_node_id: NodeID, + pub len: u64, + pub src_target_allocation: Arc>>, +} + +#[derive(Clone)] +pub struct InflightGetInfo { + pub put_id: PutIDForAKey, + pub src_node_id: NodeID, + pub key: String, + pub req_node_id: NodeID, + pub len: u64, + pub allocation: Arc, + pub route: Arc, + pub allocation_mode: GetAllocationMode, +} + +pub struct OneKvNodesRoutes { + pub put_id: PutIDForAKey, + pub lease_id: Option, + pub nodes_replicas: RwLock>, + pub get_durable_slots_used: AtomicU32, +} + +pub struct MasterKvRouterInner { + pub inflight_puts: moka::future::Cache<(String, u64, u32), InflightPutInfo>, + pub inflight_put_key_counts: Arc>, + pub inflight_gets: moka::future::Cache, + pub kv_routes: DashMap>, + pub prefix_index: ARwLock, +} +``` + +这段代码对应前面的设计判断:`PutStart / GetStart` 把慢传输挂到 `inflight_*`,`PutDone / GetDone` 再回到 `kv_routes` 提交稳定状态;同 `key` 写入准入用 `inflight_put_key_counts` 收敛,副本集合和 lease 绑定则跟着 key-version 一起存在。 + +--- + +## 三、核心链路:读写时序与状态流转 + +不可变 MB 级大对象和热点读取这两条判断落到读写链路上,就是把 payload 和控制面元数据分开处理。`put` 先登记版本和 allocation,再写入或传输 payload,最后提交稳定路由;`get` 先复用本地命中,真正 miss 才回源并合并并发请求;`delete` 先改变权威可见性,再让缓存失效和回收异步推进。 + +从内存视角看,一个 MB 级对象的主路径如下。看图时抓住一件事:`master` 管单据,`owner` 管货,transfer engine 管跨节点搬运。 + +```mermaid +flowchart LR + U[external user space\n业务对象 / tensor view] + E[external client\n轻量入口] + S[(owner shared memory\nmmap + allocation)] + O[local owner\n本机数据面] + T[transfer engine\nTCP / RDMA] + RO[(remote owner shared memory\nreplica allocation)] + M[master\nroute / lease / holder meta] + + U --> E + E -->|本机写入 / mmap 暴露| S + S --> O + O -->|跨节点 payload 传输| T + T --> RO + + E -. PutStart / GetStart .-> M + O -. PutDone / GetDone / Delete .-> M + M -. allocation / route / holder .-> O +``` + +这张图的重点是区分权威元数据和 payload 所在路径:`master` 决定路由、allocation、lease 和 holder,真实大对象留在 `owner` 共享内存与传输引擎路径上。它只描述具体路径,不把局部 mmap 或传输快路径概括成全链路零拷贝。 + +### 3.1 Put:传输前登记,传输后提交 + +`put` 的主链路采用分段状态流转: + +```text +PutStart -> 数据写入 / 传输 -> PutDone +``` + +对于 `external` 调用方,真实入口还会多一层本机 `owner` 代理。下图的关键是:大对象传输期间,`master` 持有的是在途状态,不搬 payload,也不长期占住稳定路由。 + +```mermaid +sequenceDiagram + participant E as external + participant O as owner + participant M as master + participant TO as target owner + + E->>O: ExternalPutStartReq + O->>M: PutStartReq + Note right of M: 选择目标 owner 与 allocation\n生成 put_id\n记录 inflight_puts + M-->>O: PutStartResp + O-->>E: ExternalPutStartResp + + Note over E,O: external 写入 owner 共享内存 + E->>O: ExternalPutTransferEndReq + O->>TO: transfer_data_no_copy + O->>M: PutDoneReq + Note right of M: 校验在途状态\n绑定 lease(可选)\n替换 kv_routes + M-->>O: PutDoneResp + O-->>E: ExternalPutTransferEndResp +``` + +这条链路确立了三个稳定边界: + +1. **`PutStart` 只登记在途状态**:`master` 选择放置目标、分配 `put_id` 并放入 `inflight_puts`。此时稳定路由 `kv_routes` 未更新,避免大对象传输期间长期占住主路由状态。 +2. **payload 由 `owner` 数据面搬运**:`external` 先写入本机 `owner` 共享内存,后续由 `owner` 负责本地或跨节点传输。`master` 不搬运业务 bytes。 +3. **`PutDone` 才提交稳定版本**:传输完成并回到 `master` 提交后,`kv_routes[key]` 才会替换。如果这是覆盖写,旧版本路由会进入异步 delete 广播管线,后续清理旧副本和客户端缓存。 + +关于 `put_id` 的设计: + +- `put_id` 形状是 `(put_time_ms, put_version)`。 +- `put_time_ms` 来自 `master` 处理 `PutStart` 时的毫秒时间。 +- `put_version` 是 `master` 侧按 `key` 维护的递增计数。 +- 不同 `key` 之间不承诺全局唯一,在在途表里会和 `key` 一起组成 `(key, put_time_ms, put_version)`。 + +这个设计让并发写入有明确身份。后续 `PutDoneReq` 或 `PutRevokeReq` 必须带着同一个 id 回来,`master` 才能找到对应的在途写入并决定提交或回收。 + +### 3.2 Get:本地命中优先与远程 miss 合并 + +`get` 的主链路为: + +```text +GetStart -> 数据复用 / 传输 -> GetDone +``` + +真实读取会优先尝试本地弱缓存或 `owner` 本地副本。图里的 `external weak cache hit` 是本地快路径,只在 `external` 进程内完成;只有 miss 后才进入 `owner` 和 `master`。看图时注意两把 per-key miss lock:它们的目标都是把并发回源压成少量真实请求。 + +```mermaid +sequenceDiagram + participant E as external + participant O as owner + participant M as master + participant SO as source owner + + alt external weak cache hit + Note over E: 本地弱缓存命中\n不进入 owner / master + E-->>E: 返回 ExternalMemHolder + else external weak cache miss + Note over E: 获取 per-key miss lock\n二次检查缓存 + E->>O: ExternalGetReq + + alt owner local cache hit + O-->>E: ExternalGetResp(offset, len, holder_id) + else owner local cache miss + Note over O: 获取 owner 侧 per-key miss lock + O->>M: GetStartReq + Note right of M: 读取 kv_routes\n短读锁复制副本快照\n选择源 replica 与目标 allocation + M-->>O: GetStartResp + + O->>SO: transfer_data_no_copy + O->>M: GetDoneReq + Note right of M: 校验 put_id\n创建 holder_id\n必要时提升副本 + M-->>O: GetDoneResp + O-->>E: ExternalGetResp(offset, len, holder_id) + end + end +``` + +核心结论是:热路径优先本地命中,真正 miss 后才进入 `master`。`external` 和 `owner` 都有 per-key 的 `AMapLock` 进行 miss 折叠。`master` 在 `GetStart` 中短暂读取 `nodes_replicas` 并复制成局部快照,避免长时间持有副本表读锁。 + +当前 `get` 有三种分配模式: + +| 模式 | 触发条件 | 完成后的状态 | +| --- | --- | --- | +| **`ReuseReplica`** | 请求节点已经有该 `key` 副本 | 直接复用本地 allocation,不触发真实传输 | +| **`DurableReplica`** | 请求节点需要新建可复用副本 | `GetDone` 后提升为稳定副本,受并发槽位限制 | +| **`Temporary`** | 只需要服务本次读取 | 返回 holder,但不进入稳定副本集合 | + +`DurableReplica` 有额外并发控制:同一 `key` 最多同时保留 2 个 durable get 槽位。这样热点对象可以扩散副本,但不会因为一轮并发 miss 失控地扩张副本数。 + +### 3.3 Delete:权威路由先删,缓存异步失效 + +`delete` 的权威动作发生在 `master`,采用“先删路由,后清缓存”的策略: + +下图的重点是:`DeleteResp` 不等待所有本地缓存同步清空。权威可见性先变,物理清理走后台聚合。 + +```mermaid +sequenceDiagram + participant E as external + participant O as owner + participant M as master + participant OO as other owner + + E->>O: ExternalDeleteReq + O->>M: DeleteReq + Note right of M: 删除 kv_routes\n删除 prefix_index + M-->>O: DeleteResp + O-->>E: ExternalDeleteResp + + M-->>OO: BatchDeleteClientKvMetaCacheReq + Note over OO: 批量清理 owner 本地缓存 +``` + +`delete` 先删除 `kv_routes`,让新的权威读取看不到这个 `key`;客户端弱缓存失效、`owner` 本地缓存清理进入后台管线,后续通过异步批量聚合 RPC 推进,尽可能减少删除对读写热路径的性能扰动。代价是失效传播是异步的:`DeleteResp` 返回时,所有缓存未必都已经同步消失。 + +如果 `key` 不存在,当前实现返回 `KeyNotFound`,不会把不存在的删除静默当成成功。 + +--- + +## 四、Allocation / Segment / Holder 生命周期 + +拿到一个 `MemHolder`,是不是就等于对象生命周期安全了?还不够。真正要回答的是几个正确性问题: + +- `holder` 还在用户代码手里时,底层 slot 会不会被提前回收? +- transport 还在异步读写本地地址时,segment 会不会被 `unregister / close` 掉? +- `ExternalMemHolder` drop 后,release ack 会不会漏掉,导致 master 永远托住 allocation? +- owner / external 退出或重启后,旧代际的 holder ack 会不会误删新代际状态? + +Fluxon KV 用三层 authority 来回答这些问题:`master allocation -> owner segment -> holder`。类比到前面的物流系统,`allocation` 是货架预留,`segment lease` 是运输保价单,`holder` 是业务侧取件码。三层分别管不同的不变量,不能互相替代。 + +下图先给静态关系。看图时抓住三条不变量:`master` 托住 slot,`owner` 托住本地 segment,`holder` 托住用户借用。 + +```mermaid +flowchart TD + A["master allocation\nInflightPutAllocation / InflightGetInfo.allocation"] --> B["owner segment\nmmap / registered memory"] + B --> C["segment lease\n跨库 transport 保活"] + C --> D["holder\nUserMemHolder / ExternalMemHolder / holder_id"] + + A1["slot 是否仍被系统持有"]:::note + B1["本地段地址是否仍然有效"]:::note + C1["传输期间 segment 不被提前关闭"]:::note + D1["上层调用方是否仍在合法借用"]:::note + + A --- A1 + B --- B1 + C --- C1 + D --- D1 + + classDef note fill:#f6f1e8,stroke:#8b6f47,color:#1e1c19; +``` + +`Allocation` 决定哪块 slot 还活着、何时可回收;`owner segment` 和 segment lease 保证 mmap / registered memory 在本机访问和跨库传输期间不会提前失效;`holder` 约束业务代码是否还在合法借用读取结果。正确性边界可以压成一句话:只有 holder 借用、segment transport 和 master allocation 都结束后,这块内存才允许真正回到 allocator。 + +静态关系之外,更关键的是时序:先占货架,再运输,最后发取件码。`allocation` 先于真实 payload 搬运被控制面托住,`segment lease` 只覆盖传输期间的本地段保活,`holder` 则在 `get_done` 之后才暴露给上层业务。 + +```mermaid +sequenceDiagram + participant E as external + participant O as owner + participant TE as transfer engine + participant M as master + + E->>O: ExternalGetReq(key) + O->>M: GetStart(key) + Note right of M: 选择源副本\n分配 target allocation\n写入 inflight_gets + M-->>O: GetStartResp(get_id, source, target) + + alt 本地已有副本 + O->>O: 复用本机 allocation + else 需要跨 owner 搬运 + O->>TE: transfer_data_no_copy(source -> target) + TE->>O: 申请 segment lease + Note over O,TE: 传输期间 segment 不能 unregister / close + TE-->>O: transfer done + end + + O->>M: GetDone(get_id) + Note right of M: allocation 从 inflight_gets\n转入 get_holding\n生成 holder_id + M-->>O: GetDoneResp(holder_id) + O-->>E: ExternalGetResp(offset, len, holder_id) + E->>E: 构造 ExternalMemHolder +``` + +这条链路里,`holder_id` 不是凭空指向一段地址。它背后先有 `master` 侧托住的 `Allocation`,再有 `owner` 侧真实可访问的 mmap / registered memory,最后才是业务代码拿到的 `ExternalMemHolder` 或 `UserMemHolder`。如果发生跨库传输,`segment lease` 只负责保证传输期间本地 segment 不会被提前卸载;它不替代 `Allocation`,也不把业务值变成可修改对象。 + +释放链路同样不是一跳完成。正确路径是分层释放:external 先释放对 owner holder 的借用,owner 再在本地 holder 引用归零后向 master ack,master 最后删除 `get_holding` 并释放 allocation 强引用。也就是说,external 不直接改 master 状态;它只能把“我不再借用”告诉绑定 owner。 + +这里最容易担心的是 ack 会不会漏。当前设计把 release ack 放进 `delete_ack_batch` 这类后台管线,但批量化只改变发送方式,不改变语义:ack item 先入队,队列按 holder / target 做短窗口聚合和去重,随后通过 `BatchDeleteAckReq` 发给 master;master 收到后按 `(client_id, holder_id)` 删除 `get_holding`。重复 ack 可以被视为幂等删除,漏发或目标暂时不可达则留在后台管线重试和存活检查里,不会因为“为了聚合”就直接丢掉 holder release 语义。 + +节点退出和重启也要单独看。正常退出时,owner 本地 `UserMemHolder` 通过引用计数阻止 `close()` 过早收尾;只要还有用户层 holder 活着,owner 不能把底层生命周期当作已经结束。external 侧的 `ExternalMemHolder` 则带着 owner 的 `node_start_time` 代际。owner 收到 `ExternalDeleteAckReq` 时会校验这个代际,旧 owner 崩溃重启后,即使 `node_id` 复用,旧 holder ack 也不能误删新 owner 代际的状态。对于失效广播,目标也会做存活和代际检查;目标成员已经退出或代际变化时,旧广播不会继续打到错误对象上。硬崩溃这类场景下,已经消失的进程当然不能再发送 drop ack;这里的正确性底线是“不误释放、不误删新代际状态”,后续回收再交给成员失效、lease / delete 和后台清理路径推进。 + +下图把 release ack 和节点代际放在一起看。重点是:批量聚合负责降 RPC 数量,代际校验负责挡住重启后的误释放,二者一起保证生命周期收尾不会越权。 + +```mermaid +sequenceDiagram + participant E as external + participant O as owner + participant OB as owner ack batch + participant M as master + participant MB as master invalidation batch + participant OO as other owner + + E->>O: ExternalDeleteAckReq(holder_id, owner_start_time) + Note right of O: 校验 owner 代际\n删除 external_get_holding + O->>OB: enqueue OwnerDeleteAckItem + Note over OB: 短窗口聚合\n按 holder 去重与合并\n失败重试 / 存活检查 + OB->>M: BatchDeleteAckReq(holder_ids...) + Note right of M: 删除 get_holding\nallocation 强引用归零后回收 slot + + M->>MB: enqueue DeleteKeyInfo + Note over MB: delete / 覆盖写后的缓存失效\n按目标 owner 聚合 + MB->>OO: BatchDeleteClientKvMetaCacheReq(keys..., node_start_time) + Note right of OO: 异步清理本地副本缓存 +``` + +这里的聚合请求覆盖两类路径:一类是 `owner -> master` 的 holder release ack,另一类是 `master -> owner` 或 `owner -> external` 的缓存失效广播。它们的业务语义不同,但调度骨架相似:先进入队列,再按目标短窗口聚合、去重、检查目标代际是否仍然存活,最后批量发送。这样 `master` 保留 holder 与 allocation 的权威状态,`owner` 承接本机 external 借用关系,释放和失效又不会在热点 key 或大量 holder drop 时放大成细粒度 RPC 风暴。 + +因此这一节真正想表达的不是“多了几个对象名”,而是几条正确性边界: + +| 场景 | 正确性保证 | +| --- | --- | +| 用户还持有 holder | `master.get_holding` 托住 holder -> allocation 映射,owner 本地引用计数阻止过早关闭 | +| transport 还在访问本地地址 | `segment lease` / segment read guard 阻止 segment 提前 `unregister / close` | +| holder release ack 批量发送 | ack 先入队,再聚合、去重、重试和代际存活检查;批量化不改变 release 语义 | +| owner / external 正常退出 | holder 引用归零后才进入 release ack;owner close 不会越过仍存活的用户 holder | +| owner / external 重启或目标不可达 | 请求、ack 和广播目标携带 `node_start_time` 并做存活检查;旧代际 holder 不能误删新代际状态 | +| 成员硬崩溃 | 不依赖已经消失的进程继续发 ack;系统先保证不误释放,后续通过成员失效、lease / delete 和后台清理路径收敛 | + +更完整的生命周期拆解见 [KV 设计 4 - Allocation / Segment / Holder 生命周期](https://tele-ai.github.io/Fluxon/cn/design/kv_4_allocation_segment_holder%E7%94%9F%E5%91%BD%E5%91%A8%E6%9C%9F/)。 + +--- + +## 五、特化策略接口:并发控制与生命周期治理 + +最终一致性这条判断落到并发和生命周期上,就是一个很现实的问题:哪些地方必须同步,哪些地方可以晚一点? + +Fluxon KV 的回答是:权威可见性要短路径收敛,物理清理和派生索引可以最终一致。也就是接受主路由、派生索引和缓存失效之间的短暂不同步,但把不同步约束在明确边界内。并发设计的核心,是别让慢操作占住共享状态:大对象传输不拖住主状态机,热点读取按 key 合并,删除和回收走后台批处理。 + +整体并发控制矩阵如下。扫这张表时可以只看一条主线:热路径短,慢路径折叠,清理路径异步化。 + +| 压力点 | 当前机制 | 结果 | +| --- | --- | --- | +| 大对象传输慢 | `PutStart / PutDone`、`GetStart / GetDone` 分离 | `master` 不在传输期间持有主路由写锁 | +| 同 `key` 并发 miss | `external` 和 `owner` 都有 per-key miss lock | 同 `key` miss 折叠成少量回源请求 | +| 读取副本表被长流程拖住 | `nodes_replicas` 短读锁复制快照 | 源副本选择和分配准备不长期占住共享读锁 | +| 读取时版本被覆盖 | `get_done` 校验在途 `put_id` 与当前 `kv_routes` | 旧快照无法误提升为新版本副本 | +| delete 和覆盖写需要清理旧缓存 | `delete_broadcast` 后台管线 | 提交路径保持短,缓存失效异步推进 | +| lease key 生命周期不同 | `lease_id` 固化到 `OneKvNodesRoutes` | 热路径不需要额外探测 lease manager | + +### 5.1 同 key 并发写:轻量级准入控制 + +`reject_if_inflight_same_key` 解决的是一个具体问题:调用方希望同一 `key` 不要同时出现多个在途写入。这也是 `MooncakeStore` 这类 `KV Cache` 外部缓存路径的默认取舍:同一个 `key` 已经有在途写入时,新的 put 直接 fail-fast,避免同一上下文的大对象被重复传输和重复提交。 + +开启后,`master` 在 `put_start` 阶段检查 `inflight_put_key_counts`。如果同一 `key` 已经有在途写入,直接返回 `KeyBeingWritten`。未开启时,当前实现允许同 `key` 并发 put,最终以后提交成功的版本替换前一个稳定版本。 + +这里用轻量计数索引解决准入问题,不给 `key` 加一把覆盖传输全过程的大锁。完整在途上下文仍放在 `inflight_puts`,计数索引只做按 `key` 聚合的准入判断,不阻塞其他 `key`。 + +### 5.2 Lease 机制:将保活语义绑定到 key-version + +`lease_id` 主要是为 Fluxon MQ 的大 payload 保活语义抽出来的能力。Fluxon 最早的工程动机之一,是 `VAE` 解耦异构训练里的跨资源池数据交接:`Producer` 和 `Consumer` 分布在不同资源池中独立扩缩容,通过中间态 `Payload` 完成异步交接。 + +这个场景对 KV 底座提出了一个明确要求:消息还没有被消费前,payload 必须保留;消费完成或 lease 到期后,系统又要能回收这部分大对象。传统固定成员集合通信更适合训练进程组内部同步通信,难以直接覆盖动态成员、异步交接、背压、消息保活和跨资源池弹性调度。Fluxon MQ 选择让控制面只保留消息壳、成员拓扑与 offset,大 payload 直接复用 KV 数据面搬运;因此 KV 需要提供一条稳定的 lease 绑定路径。 + +KV 的 `PutOptionalArgs` 当前对 Python 稳定公开的字段主要是: + +| 参数 | 稳定语义 | +| --- | --- | +| `lease_id` | 在 `put_done` 时把当前 `key` 版本绑定到指定 lease | +| `reject_if_inflight_same_key` | 在 `put_start` 阶段拒绝同 `key` 已有在途写入的请求 | + +`lease_id` 的关键设计是:lease 绑定这次提交出来的 key-version,语义不落到某个模糊的 `key` 全局状态上。这让 MQ 可以把“消息未消费前 payload 必须保留”的语义收敛到 KV 版本路由,而不需要另建一套大对象保活和回收系统。 + +这个设计落到实现上有几个约束: + +- `OneKvNodesRoutes.lease_id` 成为稳定路由对象的一部分。 +- 只有显式传入 `lease_id`,该次 put 才是 lease put。 +- `lease_id=None` 表示纯非 lease put,不会回退到最近一次 lease。 +- lease key 不进入普通 moka 副本缓存,避免被普通缓存淘汰语义间接管理。 +- `get` 热路径只读取 `route.lease_id`,不需要再向 lease manager 做额外探测。 + +Rust 内部还支持 `preferred_sub_cluster`,用于影响 `put_start` 的目标放置。它当前是内部已有能力,主要被 Fluxon MQ 通过 Rust 接口使用,还没有完整暴露成 Python 稳定公开契约。因此本文只把它作为内部实现能力说明,不把它写成普通 Python 用户可依赖的稳定参数。 + +### 5.3 派生索引与最终一致性 + +`prefix_index` 是从 `kv_routes` 派生出的前缀索引,当前主要用于 MQ 的容量背压限制。`put_done` 提交后,前缀索引更新可以在后台推进,所以它不承诺 put 时立即可见的强一致性。 + +这个边界把主路由和派生索引区分开,也避免把 MQ 背压辅助路径误解成 KV 主读取路径。 + +对 `KV Cache` 这类提交后不可变、读取频繁、随上下文增长而累积的对象,最终一致性的关键价值在于把“可见性决策”和“物理清理”拆开。新版本提交或删除请求先更新权威路由,让后续读请求看到稳定结果;旧副本、客户端弱缓存和 holder release ack 可以进入后台队列,按节点、key 或 holder 聚合后批量异步延迟删除。这样读写提交路径不需要等待所有副本同步清空,也不会因为单个热点上下文的清理放大成大量同步 RPC。 + +--- + +## 六、加入 Tele-AI Fluxon 项目 + +Fluxon 当前开源的是一套面向 AI 数据面的统一底座,包含 `KV/RPC`、`MQ` 和 `FS` 三类接口。`KV/RPC` 侧已经可以通过 Quick Start 体验 `master / owner / external` 的最小链路。 + +需要说明的是,本文描述的是当前开源仓库里 Fluxon KV 的初版通用架构。最近的工作里,我们在推理路径上继续把最终一致性的边界推到更极致:对本地命中的 `KV Cache`,让内存操作尽可能停留在本机直接路径;针对推理里的特化前缀 `get`,补上批量化和分组对齐接口,让一次请求可以更贴近模型侧的 prefix 访问形态;在 `SGLang` 集成里,进一步把内存操作下沉到 CUDA kernel 消费路径,减少 Python / C++ 中间层对象组织和拷贝开销。在对应的推理集成与测试场景下,这条专用路径已经表现出超过 `HiCache + Mooncake` 组合的端到端推理吞吐和延迟结果。 + +这套系统还在快速演进。我们开源 Fluxon,是希望邀请更多基础设施开发者一起找问题、提需求、补场景。无论你关心 Rust 高并发、LLM 推理缓存、跨节点大对象传输,还是正在被 Python 多进程和内存治理折磨,都欢迎来仓库里交流。 + +5 分钟快速体验可以直接跑 `KV Quick Start`: + +```bash +docker run --rm -it --network host \ + hanbaoaaa/fluxon_quick_start:0.2.1 \ + --mode kv \ + --etcd-client-port 12379 \ + --master-p2p-port 31000 \ + --panel-port 18080 \ + --greptime-http-port 14000 \ + --kv-http-port 8083 +``` + +进入容器后执行: + +```text +put demo:hello world +get demo:hello +del demo:hello +``` + +更多入口: + +- GitHub 仓库:[Tele-AI/Fluxon](https://github.com/Tele-AI/Fluxon),欢迎 Star、Issue 和 PR +- 中文文档:[Fluxon 中文文档](https://tele-ai.github.io/Fluxon/cn/) +- 用户文档:[Fluxon 用户文档](https://tele-ai.github.io/Fluxon/cn/user_doc/) +- KV 和 RPC 接口:[用户 - 3 - KV/RPC 接口](https://tele-ai.github.io/Fluxon/cn/user_doc/%E7%94%A8%E6%88%B7---3---KV-RPC%E6%8E%A5%E5%8F%A3/) +- 开发者文档:[Fluxon 开发者文档](https://tele-ai.github.io/Fluxon/cn/dev_doc/) diff --git "a/fluxon_doc_cn/dev_doc/\345\274\200\345\217\221\350\200\205 - 3 - \346\226\207\346\241\243\345\206\231\344\275\234\350\247\204\347\272\246.md" "b/fluxon_doc_cn/dev_doc/\345\274\200\345\217\221\350\200\205 - 3 - \346\226\207\346\241\243\345\206\231\344\275\234\350\247\204\347\272\246.md" index 75ed5b7..d2e2ed4 100644 --- "a/fluxon_doc_cn/dev_doc/\345\274\200\345\217\221\350\200\205 - 3 - \346\226\207\346\241\243\345\206\231\344\275\234\350\247\204\347\272\246.md" +++ "b/fluxon_doc_cn/dev_doc/\345\274\200\345\217\221\350\200\205 - 3 - \346\226\207\346\241\243\345\206\231\344\275\234\350\247\204\347\272\246.md" @@ -5,6 +5,14 @@ - 让读者能在最短时间内抓住稳定结论。 - 让文档和代码、公开接口、实际行为保持一致。 +在进入细则前,先记住这几个高频判断: + +- 结论优先,展开随后;读者应该先看到稳定说法,再决定是否继续下钻。 +- 同一概念只保留一个名字;术语漂移会直接削弱可读性和可信度。 +- 开头按什么轴提出问题,后文就按什么轴回答;不要半路更换分类方法。 +- 英文文档优先写行业里自然成立的表达,不做逐词直译。 +- 能用列表、表格、图说清的内容,不要硬压成长段线性 prose。 + ## 1. 通用规则 - 先写结论,再写展开。读者应在开头 30 秒内知道“这篇文档在回答什么问题”。 diff --git a/fluxon_doc_en/dev_doc/Developer - 3 - Documentation Writing Rules.md b/fluxon_doc_en/dev_doc/Developer - 3 - Documentation Writing Rules.md index 4577714..dee7076 100644 --- a/fluxon_doc_en/dev_doc/Developer - 3 - Documentation Writing Rules.md +++ b/fluxon_doc_en/dev_doc/Developer - 3 - Documentation Writing Rules.md @@ -5,6 +5,14 @@ This page defines how Fluxon user docs, developer docs, and design docs should b - Let readers reach the stable conclusion as quickly as possible. - Keep docs aligned with code, public contracts, and actual behavior. +Before the detailed rules, keep these high-frequency judgments in mind: + +- Put the conclusion first and the expansion after it. Readers should see the stable statement before deciding whether to drill deeper. +- Keep one name for one concept. Terminology drift directly weakens readability and trust. +- Answer later sections on the same axis used by the opening question list. Do not switch classification schemes halfway through the page. +- In English docs, prefer expressions that are natural in the domain instead of translating Chinese source text word by word. +- If a list, table, or diagram can explain the point cleanly, do not force it into a long block of linear prose. + ## 1. General Rules - Lead with the conclusion, then expand. A reader should know what the page answers within the first 30 seconds. diff --git a/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_base.py b/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_base.py index f992c2d..ad6f11e 100644 --- a/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_base.py +++ b/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_base.py @@ -44,9 +44,7 @@ def _find_project_root(start: Path) -> Path: if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -from typing import Dict, List, Optional, Tuple - -import etcd3 +from typing import Any, Dict, List, Optional, Tuple from fluxon_py.api_ext_chan import ( # noqa: E402 MPMCChanConsumer, @@ -82,12 +80,12 @@ def _find_project_root(start: Path) -> Path: KV_SVC_TYPE, CHAN_CONFIG_TEST, TEST_TIMEOUT_SECONDS, - ETCD_HOST, - ETCD_PORT, setup_test_environment, new_test_consumer, new_test_producer, run_with_argmatrix, + is_transient_etcd_control_failure as _is_transient_etcd_failure, + etcd_control_call_with_retry as _etcd_call_with_retry, ) from fluxon_py.tests.test_lib import pre_kill_existing_test_processes_by_script_name # noqa: E402 @@ -118,6 +116,35 @@ def _atomic_stdout_write_line(line: str) -> None: os.write(sys.stdout.fileno(), payload) +def _read_log_tail(path: str, *, max_lines: int = 120) -> str: + try: + with open(path, "r", encoding="utf-8", errors="replace") as handle: + lines = handle.readlines() + except FileNotFoundError: + return f"" + except Exception as exc: # noqa: BLE001 + return f"" + return "".join(lines[-max_lines:]).rstrip() + + +def _format_subprocess_failure( + process_type: str, + identifier: str, + rc: int, + log_file: str, + *, + early: bool, +) -> str: + phase = "exited early" if early else "failed" + return ( + f"{process_type} {identifier} {phase} with return code {rc}. " + f"Log file: {log_file}\n" + "--- child log tail ---\n" + f"{_read_log_tail(log_file)}\n" + "--- end child log tail ---" + ) + + INITIAL_PRODUCERS_COUNT = 3 INITIAL_CONSUMERS_COUNT = 2 NEW_PRODUCERS_MIN_COUNT = 1 @@ -264,7 +291,6 @@ def run_producer(env: "ChannelState", args: argparse.Namespace) -> None: ) assert isinstance(producer, MPMCChanProducer) print(f"[Producer-{args.producer_id}] Started") - etcd_client = producer.etcd_client try: for index in range(args.message_count): unique_id = str(uuid.uuid4()) @@ -292,11 +318,22 @@ def run_producer(env: "ChannelState", args: argparse.Namespace) -> None: except Exception as exc: # noqa: BLE001 print(f"[Producer-{args.producer_id}] Error: {exc}") _atomic_stdout_write_line(f"{PRODUCER_CRASH_MARKER} {args.producer_id}") + try: + producer.close().unwrap() + except Exception as close_exc: # noqa: BLE001 + print( + f"[Producer-{args.producer_id}] close after error failed: " + f"{close_exc}" + ) raise - finally: - print(f"[Producer-{args.producer_id}] Finished") - _atomic_stdout_write_line(f"{PRODUCER_NORMAL_EXIT_MARKER} {args.producer_id}") + try: producer.close().unwrap() + except Exception as exc: # noqa: BLE001 + print(f"[Producer-{args.producer_id}] close failed: {exc}") + _atomic_stdout_write_line(f"{PRODUCER_CRASH_MARKER} {args.producer_id}") + raise + print(f"[Producer-{args.producer_id}] Finished") + _atomic_stdout_write_line(f"{PRODUCER_NORMAL_EXIT_MARKER} {args.producer_id}") finally: configure_backend(env, backend_type=prev_type, backend_ip=prev_ip) release(env, store_key) @@ -324,11 +361,14 @@ def run_consumer(env: "ChannelState", args: argparse.Namespace) -> None: f"{consumer.mpmc_channel.mpmc_member_id}", flush=True, ) - etcd_client = consumer.etcd_client - etcd_client.put( - f"/test_mpmc_consumer/{args.consumer_id}", - b"dummy_value", - consumer.mpmc_channel.mpmc_global_lease, + _etcd_call_with_retry( + f"publish consumer presence for consumer_id={args.consumer_id}", + lambda client: client.put( + f"/test_mpmc_consumer/{args.consumer_id}", + b"dummy_value", + lease_id=int(consumer.mpmc_channel.mpmc_global_lease.id), + ), + max_attempts=3, ) consumed_count = 0 @@ -344,13 +384,19 @@ def run_consumer(env: "ChannelState", args: argparse.Namespace) -> None: # Periodically check whether all producers are done. now = time.time() if now - last_producer_check >= producer_done_check_interval: - stop_flag, _ = etcd_client.get(f"/test_mpmc_stop_producer") + stop_flag = _etcd_call_with_retry( + "read producer stop flag", + lambda client: client.get("/test_mpmc_stop_producer"), + max_attempts=3, + ) all_producers_done = stop_flag is not None last_producer_check = now # External stop signal from harness. - stop_flag, _ = etcd_client.get( - f"/test_mpmc_stop_consumer/{args.consumer_id}" + stop_flag = _etcd_call_with_retry( + f"read consumer stop flag for consumer_id={args.consumer_id}", + lambda client: client.get(f"/test_mpmc_stop_consumer/{args.consumer_id}"), + max_attempts=3, ) if stop_flag: logging.info( @@ -417,31 +463,47 @@ def run_consumer(env: "ChannelState", args: argparse.Namespace) -> None: except Exception as exc: # noqa: BLE001 print(f"[Consumer-{args.consumer_id}] Error: {exc}") _atomic_stdout_write_line(f"{CONSUMER_CRASH_MARKER} {args.consumer_id}") + try: + consumer.close().unwrap() + except Exception as close_exc: # noqa: BLE001 + print( + f"[Consumer-{args.consumer_id}] close after error failed: " + f"{close_exc}" + ) raise - finally: - print( - f"[Consumer-{args.consumer_id}] Finished, consumed" - f" {consumed_count} messages" + try: + _etcd_call_with_retry( + f"delete consumer presence for consumer_id={args.consumer_id}", + lambda client: client.delete(f"/test_mpmc_consumer/{args.consumer_id}"), + max_attempts=3, ) - _atomic_stdout_write_line(f"{CONSUMER_NORMAL_EXIT_MARKER} {args.consumer_id}") - etcd_client.delete(f"/test_mpmc_consumer/{args.consumer_id}") consumer.close().unwrap() + except Exception as exc: # noqa: BLE001 + print(f"[Consumer-{args.consumer_id}] cleanup failed: {exc}") + _atomic_stdout_write_line(f"{CONSUMER_CRASH_MARKER} {args.consumer_id}") + raise + print( + f"[Consumer-{args.consumer_id}] Finished, consumed" + f" {consumed_count} messages" + ) + _atomic_stdout_write_line(f"{CONSUMER_NORMAL_EXIT_MARKER} {args.consumer_id}") finally: configure_backend(env, backend_type=prev_type, backend_ip=prev_ip) release(env, store_key) def clean_etcd() -> None: - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: + def _clean(etcd_client: Any) -> None: etcd_client.delete_prefix("/mpmc_channels") etcd_client.delete_prefix("/channels") etcd_client.delete_prefix("/test_mpmc_stop_consumer") etcd_client.delete_prefix("/test_mpmc_consumer") etcd_client.delete_prefix("/test_mpmc_stop_producer") + _etcd_call_with_retry("clean MPMC test etcd prefixes", _clean) + def _wait_until_lease_revoked( - etcd_client: etcd3.Etcd3Client, lease_id: int, *, timeout_s: float = 10.0, @@ -449,7 +511,11 @@ def _wait_until_lease_revoked( deadline = time.time() + timeout_s while True: try: - info = etcd_client.get_lease_info(int(lease_id)) + ttl_val = _etcd_call_with_retry( + f"read lease ttl for lease_id={lease_id}", + lambda client: client.lease_ttl(int(lease_id)), + max_attempts=3, + ) except Exception as exc: # noqa: BLE001 msg = str(exc).lower() if "not found" in msg or "requested lease not found" in msg: @@ -459,7 +525,6 @@ def _wait_until_lease_revoked( f"lease revoke verification failed for lease_id={lease_id}: {exc}" ) from exc else: - ttl_val = getattr(info, "TTL", None) if not isinstance(ttl_val, int): raise RuntimeError( f"invalid TTL returned for lease_id={lease_id}: {ttl_val!r}" @@ -493,14 +558,19 @@ def test_mpmc_member_lease_expiry_closes_owner() -> None: chan_id = producer.get_chan_id() lease_id = int(producer.mpmc_channel.mpmc_member_lease.id) - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - mpsc_meta_before = list(etcd_client.get_prefix("/channels/meta/")) - assert len(mpsc_meta_before) == 0, ( - "fresh MPMC producer should not create any sub-MPSC metadata before the first put, " - f"found {len(mpsc_meta_before)} keys" - ) - etcd_client.revoke_lease(lease_id) - _wait_until_lease_revoked(etcd_client, lease_id) + mpsc_meta_before = _etcd_call_with_retry( + "list MPSC metadata before member lease revoke", + lambda client: list(client.get_prefix("/channels/meta/")), + ) + assert len(mpsc_meta_before) == 0, ( + "fresh MPMC producer should not create any sub-MPSC metadata before the first put, " + f"found {len(mpsc_meta_before)} keys" + ) + _etcd_call_with_retry( + f"revoke member lease lease_id={lease_id}", + lambda client: client.revoke_lease(lease_id), + ) + _wait_until_lease_revoked(lease_id) first_put = producer.put_data( { @@ -516,11 +586,13 @@ def test_mpmc_member_lease_expiry_closes_owner() -> None: assert first_err.channel_id == chan_id assert producer.shutdown_ctl.closed, "producer must mark itself closed after member lease loss" - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - mpsc_meta_after = list(etcd_client.get_prefix("/channels/meta/")) - assert len(mpsc_meta_after) == 0, ( - "dead member lease must stop sub-MPSC creation before any new channel meta is published" - ) + mpsc_meta_after = _etcd_call_with_retry( + "list MPSC metadata after member lease revoke", + lambda client: list(client.get_prefix("/channels/meta/")), + ) + assert len(mpsc_meta_after) == 0, ( + "dead member lease must stop sub-MPSC creation before any new channel meta is published" + ) second_put = producer.put_data( { @@ -643,11 +715,12 @@ def scenario_dynamic_producer_consumer( # Map all process handles by identifier (producer_id/consumer_id) process_handles_by_id: Dict[str, Tuple[str, subprocess.Popen, str]] = {} joined_ids: set[str] = set() - etcd_client = etcd3.client(ETCD_HOST, ETCD_PORT) initial_consumers_id: List[str] = [] dyn_consumers: List[str] = [] recovered_consumers: List[str] = [] test_mpmc_id: Optional[str] = None + scan_stop_event = threading.Event() + scan_thread: Optional[threading.Thread] = None def fail_fast_on_subprocess_error(*, process_type_filter: Optional[str] = None) -> None: for identifier, (process_type, proc, log_file) in process_handles_by_id.items(): @@ -658,8 +731,13 @@ def fail_fast_on_subprocess_error(*, process_type_filter: Optional[str] = None) continue if rc != 0: raise RuntimeError( - f"{process_type} {identifier} exited early with return code {rc}. " - f"Check log file for details: {log_file}" + _format_subprocess_failure( + process_type, + identifier, + rc, + log_file, + early=True, + ) ) def wait_all_of_type(process_type: str, *, timeout_s: int) -> None: @@ -681,8 +759,13 @@ def wait_all_of_type(process_type: str, *, timeout_s: int) -> None: print(f"Log file: {log_file}") continue raise RuntimeError( - f"{ptype} {identifier} failed with return code {proc.returncode}." - f" Check log file for details: {log_file}" + _format_subprocess_failure( + ptype, + identifier, + proc.returncode, + log_file, + early=False, + ) ) if not running: @@ -716,9 +799,13 @@ def _extract_mpmc_member_id_from_consumer_log(log_file: str) -> Optional[int]: def _count_ready_keys_for_member_id(member_id: int) -> int: assert test_mpmc_id is not None, "test_mpmc_id must be initialized before counting ready keys" - ready_chans_kvs = list(etcd_client.get_prefix(_new_mpmc_ready_channels_prefix(test_mpmc_id))) + ready_prefix = _new_mpmc_ready_channels_prefix(test_mpmc_id) + ready_chans_kvs = _etcd_call_with_retry( + f"count ready keys for member_id={member_id}", + lambda client: list(client.get_prefix(ready_prefix)), + ) count = 0 - for value, _meta in ready_chans_kvs: + for _key, value in ready_chans_kvs: if value is None: continue if value.decode() == str(member_id): @@ -728,17 +815,21 @@ def _count_ready_keys_for_member_id(member_id: int) -> int: def _list_ready_keys_for_member_id(member_id: int) -> List[str]: assert test_mpmc_id is not None, "test_mpmc_id must be initialized before listing ready keys" member_id_str = str(member_id) - ready_chans_kvs = list(etcd_client.get_prefix(_new_mpmc_ready_channels_prefix(test_mpmc_id))) + ready_prefix = _new_mpmc_ready_channels_prefix(test_mpmc_id) + ready_chans_kvs = _etcd_call_with_retry( + f"list ready keys for member_id={member_id}", + lambda client: list(client.get_prefix(ready_prefix)), + ) keys: List[str] = [] - for value, meta in ready_chans_kvs: + for key, value in ready_chans_kvs: if value is None: continue if value.decode() != member_id_str: continue try: - keys.append(meta.key.decode()) + keys.append(key.decode()) except Exception: - keys.append(repr(meta.key)) + keys.append(repr(key)) return keys def _wait_ready_keys_cleared_for_member_id(member_id: int, timeout_s: float) -> List[str]: @@ -755,12 +846,19 @@ def _wait_ready_keys_cleared_for_member_id(member_id: int, timeout_s: float) -> def _wait_for_test_mpmc_id() -> str: deadline = time.time() + float(TEST_TIMEOUT_SECONDS) while True: - value, _meta = etcd_client.get(NEW_OR_BIND_MAPPING_KEY) + value = _etcd_call_with_retry( + f"read MPMC mapping key {NEW_OR_BIND_MAPPING_KEY}", + lambda client: client.get(NEW_OR_BIND_MAPPING_KEY), + ) if value is not None: raw = value.decode().strip() if raw.isdigit(): candidate = raw - meta_val, _meta = etcd_client.get(_new_mpmc_meta_key(candidate)) + meta_key = _new_mpmc_meta_key(candidate) + meta_val = _etcd_call_with_retry( + f"read MPMC meta key {meta_key}", + lambda client: client.get(meta_key), + ) if meta_val is not None: return candidate logging.warning( @@ -782,7 +880,10 @@ def _wait_for_test_mpmc_id() -> str: def _assert_test_mpmc_id_stable() -> None: assert test_mpmc_id is not None - value, _meta = etcd_client.get(NEW_OR_BIND_MAPPING_KEY) + value = _etcd_call_with_retry( + f"assert stable MPMC mapping key {NEW_OR_BIND_MAPPING_KEY}", + lambda client: client.get(NEW_OR_BIND_MAPPING_KEY), + ) if value is None: raise RuntimeError( f"etcd key {NEW_OR_BIND_MAPPING_KEY!r} disappeared during test; expected mpmc_id={test_mpmc_id}" @@ -872,22 +973,25 @@ def consumer_process_cmd(consumer_id: str) -> List[str]: str(prefetch), ] - def scan_producer_offset() -> None: + def scan_producer_offset(stop_event: threading.Event) -> None: # Use dedicated scan logger (single file mpmc_scan_offset.log) logger = _scan_logger mpsc_producer_offset_pair: Dict[int, Dict[int, List[int]]] = {} - scan_client = etcd3.client(ETCD_HOST, ETCD_PORT) mpsc_chans: List[int] = [] def get_chans() -> List[int]: prefix = mpsc._new_etcd_meta_key_prefix() + meta_kvs = _etcd_call_with_retry( + "scan MPSC channel metadata keys", + lambda client: list(client.get_prefix(prefix)), + ) return [ - int(meta.key.decode().split("/")[-1]) - for _, meta in scan_client.get_prefix(prefix) + int(key.decode().split("/")[-1]) + for key, _value in meta_kvs ] try: - while True: + while not stop_event.is_set(): if mpsc_producer_offset_pair: sum_unconsumed_count = 0 sum_consumed_count = 0 @@ -925,11 +1029,14 @@ def get_chans() -> List[int]: ) logger.info(">>> sum_unconsumed_count: %s", sum_unconsumed_count) logger.info(">>> sum_consumed_count: %s", sum_consumed_count) - ready_chans_kvs = list(scan_client.get_prefix(_READY_CHANNELS_BASE_PREFIX)) + ready_chans_kvs = _etcd_call_with_retry( + "scan all MPMC ready channel keys", + lambda client: list(client.get_prefix(_READY_CHANNELS_BASE_PREFIX)), + ) mpmc_ids: set[str] = set() ready_mpscs: List[str] = [] - for value, meta in ready_chans_kvs: - key = meta.key.decode() + for key_bytes, value in ready_chans_kvs: + key = key_bytes.decode() mpmc_id = key.split("/")[-2] mpsc_id = key.split("/")[-1] mpmc_ids.add(mpmc_id) @@ -950,10 +1057,10 @@ def get_chans() -> List[int]: logger.info("all_mpscs: %s", mpsc_chans) for mpsc_id in mpsc_chans: mpsc_producer_offset_pair.setdefault(mpsc_id, {}) - producer_offset_kvs = list( - scan_client.get_prefix( - mpsc._new_produce_offset_of_all_producer_key(mpsc_id) - ) + producer_offset_prefix = mpsc._new_produce_offset_of_all_producer_key(mpsc_id) + producer_offset_kvs = _etcd_call_with_retry( + f"scan producer offsets for mpsc_id={mpsc_id}", + lambda client: list(client.get_prefix(producer_offset_prefix)), ) logger.info( "mpsc %s producer_offset_kvs: %s", @@ -961,8 +1068,8 @@ def get_chans() -> List[int]: producer_offset_kvs, ) mpsc_producer_offsets = { - int(meta.key.decode().split("/")[-1]): int(value.decode()) - for value, meta in producer_offset_kvs + int(key.decode().split("/")[-1]): int(value.decode()) + for key, value in producer_offset_kvs } logger.info( "mpsc %s producer_offsets dict: %s", @@ -981,7 +1088,10 @@ def get_chans() -> List[int]: consume_offset_key = mpsc._new_consume_offset_of_one_producer_key( mpsc_id, str(mpsc_producer_key) ) - consume_value, _ = scan_client.get(consume_offset_key) + consume_value = _etcd_call_with_retry( + f"read consume offset {consume_offset_key}", + lambda client: client.get(consume_offset_key), + ) logger.info( "mpsc_id: %s mpsc_producer_key: %s consume_offset_key: %s " "mpsc_consume_offset: %s", @@ -994,9 +1104,12 @@ def get_chans() -> List[int]: mpsc_producer_offset_pair[mpsc_id][mpsc_producer_key][1] = int( consume_value.decode() ) - time.sleep(5) - finally: - scan_client.close() + stop_event.wait(5) + except Exception as exc: # noqa: BLE001 + if stop_event.is_set() and _is_transient_etcd_failure(exc): + logger.info("scan_producer_offset stopped during transient etcd failure: %s", exc) + return + logger.exception("scan_producer_offset stopped unexpectedly") try: # NOTE: `new_or_bind` has a race window during first-time channel creation when multiple @@ -1022,7 +1135,11 @@ def get_chans() -> List[int]: initial_consumers_id.append(consumer_id) start_processes() - scan_thread = threading.Thread(target=scan_producer_offset, daemon=True) + scan_thread = threading.Thread( + target=scan_producer_offset, + args=(scan_stop_event,), + daemon=True, + ) scan_thread.start() print("=== Starting dynamic management phase ===") @@ -1058,6 +1175,7 @@ def get_chans() -> List[int]: start_processes() time.sleep(5) + fail_fast_on_subprocess_error(process_type_filter="producer") _assert_test_mpmc_id_stable() for phase in range(CONSUMER_CRASH_RECOVER_PHASES): @@ -1071,8 +1189,11 @@ def get_chans() -> List[int]: consumes_to_stop.append(consumer_id) for consumer_id in consumes_to_stop: - etcd_client.put( - f"/test_mpmc_stop_consumer/{consumer_id}", b"dummy_value" + _etcd_call_with_retry( + f"publish stop flag for consumer_id={consumer_id}", + lambda client, consumer_id=consumer_id: client.put( + f"/test_mpmc_stop_consumer/{consumer_id}", b"dummy_value" + ), ) for consumer_id in consumes_to_stop: @@ -1102,7 +1223,10 @@ def get_chans() -> List[int]: member_id = _extract_mpmc_member_id_from_consumer_log(log_file) # Verify that ready keys exist (global view) before join - ready_keys_before = list(etcd_client.get_prefix(_READY_CHANNELS_BASE_PREFIX)) + ready_keys_before = _etcd_call_with_retry( + f"list ready keys before joining consumer_id={consumer_id}", + lambda client: list(client.get_prefix(_READY_CHANNELS_BASE_PREFIX)), + ) logging.info( "Before join: total ready keys=%d (all mpmc) for consumer %s", len(ready_keys_before), @@ -1110,8 +1234,10 @@ def get_chans() -> List[int]: ) if member_id is not None: assert test_mpmc_id is not None - ready_keys_under_test = list( - etcd_client.get_prefix(_new_mpmc_ready_channels_prefix(test_mpmc_id)) + test_ready_prefix = _new_mpmc_ready_channels_prefix(test_mpmc_id) + ready_keys_under_test = _etcd_call_with_retry( + f"list test ready keys before joining consumer_id={consumer_id}", + lambda client: list(client.get_prefix(test_ready_prefix)), ) logging.info( "Before join: test_mpmc_id=%s ready_keys=%d, consumer %s mpmc_member_id=%s ready_keys_by_member=%d", @@ -1151,7 +1277,10 @@ def get_chans() -> List[int]: leftover_keys = [] # Verify that ready keys have been deleted for this consumer after sleep - ready_keys_after = list(etcd_client.get_prefix(_READY_CHANNELS_BASE_PREFIX)) + ready_keys_after = _etcd_call_with_retry( + f"list ready keys after joining consumer_id={consumer_id}", + lambda client: list(client.get_prefix(_READY_CHANNELS_BASE_PREFIX)), + ) logging.info( "After join and sleep: total ready keys=%d (all mpmc) for consumer %s", len(ready_keys_after), @@ -1159,8 +1288,10 @@ def get_chans() -> List[int]: ) if member_id is not None: assert test_mpmc_id is not None - ready_keys_under_test_after = list( - etcd_client.get_prefix(_new_mpmc_ready_channels_prefix(test_mpmc_id)) + test_ready_prefix_after = _new_mpmc_ready_channels_prefix(test_mpmc_id) + ready_keys_under_test_after = _etcd_call_with_retry( + f"list test ready keys after joining consumer_id={consumer_id}", + lambda client: list(client.get_prefix(test_ready_prefix_after)), ) left = _count_ready_keys_for_member_id(member_id) logging.info( @@ -1205,9 +1336,12 @@ def debug_all_ready_channels() -> None: "debug_all_ready_channels after close consumers %s", consumes_to_stop, ) - ready_chans_kvs = list(etcd_client.get_prefix(_READY_CHANNELS_BASE_PREFIX)) - for value, meta in ready_chans_kvs: - key = meta.key.decode() + ready_chans_kvs = _etcd_call_with_retry( + "debug all ready channels after closing consumers", + lambda client: list(client.get_prefix(_READY_CHANNELS_BASE_PREFIX)), + ) + for key_bytes, value in ready_chans_kvs: + key = key_bytes.decode() mpmc_id = key.split("/")[-2] mpsc_id = key.split("/")[-1] logging.info( @@ -1233,6 +1367,7 @@ def debug_all_ready_channels() -> None: recovered_consumers.append(consumer_id) start_processes() time.sleep(5) + fail_fast_on_subprocess_error(process_type_filter="producer") _assert_test_mpmc_id_stable() join_timeout_s = int(TEST_TIMEOUT_SECONDS) @@ -1241,7 +1376,10 @@ def debug_all_ready_channels() -> None: wait_all_of_type("producer", timeout_s=join_timeout_s) # 2) Notify consumers that no more producers will publish; they will exit after idle timeout. - etcd_client.put("/test_mpmc_stop_producer", b"dummy_value") + _etcd_call_with_retry( + "publish producer stop flag", + lambda client: client.put("/test_mpmc_stop_producer", b"dummy_value"), + ) # 3) Wait remaining consumers to drain and exit (or fail fast). wait_all_of_type("consumer", timeout_s=join_timeout_s) @@ -1251,7 +1389,11 @@ def debug_all_ready_channels() -> None: verify_exit_status(subprocesses) print("=== MPMC Dynamic Test PASSED ===") finally: - etcd_client.close() + scan_stop_event.set() + if scan_thread is not None: + scan_thread.join(timeout=10) + if scan_thread.is_alive(): + logging.warning("scan_producer_offset did not stop before timeout") def verify_production_consumption_counts( diff --git a/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_quick_and_weighted_consume.py b/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_quick_and_weighted_consume.py index e234cce..693abf4 100644 --- a/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_quick_and_weighted_consume.py +++ b/fluxon_py/tests/test_api_chan_mpmc/test_api_chan_mpmc_quick_and_weighted_consume.py @@ -17,8 +17,6 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Tuple -import etcd3 - # Ensure absolute imports work when running this file directly import os as _os import sys as _sys @@ -37,14 +35,13 @@ ) from fluxon_py.tests.test_lib import ( # noqa: E402 CHAN_CONFIG_TEST, - ETCD_HOST, - ETCD_PORT, TEST_TIMEOUT_SECONDS, setup_test_environment, new_test_consumer, new_test_producer, load_test_fluxon_cluster_name, run_with_argmatrix, + etcd_control_call_with_retry as _etcd_call_with_retry, ) from fluxon_py.api_ext_chan import ( # noqa: E402 _new_unique_lock_key, @@ -267,7 +264,10 @@ def wait_for_processes(processes: List[Tuple[str, subprocess.Popen, str]]) -> No if proc.returncode != 0: raise RuntimeError( f"Process {process_type} failed (log: {log_file})," - f" return code: {proc.returncode}" + f" return code: {proc.returncode}\n" + "--- child log tail ---\n" + f"{_read_log_tail(log_file)}\n" + "--- end child log tail ---" ) @@ -448,27 +448,29 @@ def consumer_fairness_tolerance(total_consumed: int) -> int: def clean_namespace() -> None: if LOG_DIR.exists(): shutil.rmtree(LOG_DIR) - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: + + def _clean(etcd_client: Any) -> None: # Delete unique mapping and its lock key to keep this scenario deterministic across reruns. etcd_client.delete(_new_unique_mapping_key(CHANNEL_KEY)) etcd_client.delete(_new_unique_lock_key(CHANNEL_KEY)) etcd_client.delete_prefix("/mpmc_channels") etcd_client.delete_prefix("/channels") - # Delete all producer done keys with correct format for p_idx in range(PRODUCER_COUNT): producer_id = f"P{p_idx}" etcd_client.delete(f"{PRODUCER_DONE_KEY}_{producer_id}") + _etcd_call_with_retry("clean quick fair consume namespace", _clean) + def reset_producer_done_flag() -> None: - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - # Delete legacy unsuffixed key (if any) + def _reset(etcd_client: Any) -> None: etcd_client.delete(PRODUCER_DONE_KEY) - # Delete all per-producer done keys for p_idx in range(PRODUCER_COUNT): producer_id = f"P{p_idx}" etcd_client.delete(f"{PRODUCER_DONE_KEY}_{producer_id}") + _etcd_call_with_retry("reset producer done flags", _reset) + def _read_log_tail(path: str, *, max_lines: int = 80) -> str: try: @@ -505,31 +507,34 @@ def _wait_unique_key_mapping( bootstrap_log: str, ) -> str: deadline = time.time() + float(timeout_seconds) - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - while time.time() < deadline: - bootstrap_state = _bootstrap_process_state( - bootstrap_proc=bootstrap_proc, - bootstrap_log=bootstrap_log, + mapping_key = _new_unique_mapping_key(CHANNEL_KEY) + while time.time() < deadline: + bootstrap_state = _bootstrap_process_state( + bootstrap_proc=bootstrap_proc, + bootstrap_log=bootstrap_log, + ) + if bootstrap_state is not None: + raise RuntimeError( + "Bootstrap producer exited before publishing channel mapping: " + f"unique_key={CHANNEL_KEY!r} {bootstrap_state}" ) - if bootstrap_state is not None: - raise RuntimeError( - "Bootstrap producer exited before publishing channel mapping: " - f"unique_key={CHANNEL_KEY!r} {bootstrap_state}" - ) - value, _ = etcd_client.get(_new_unique_mapping_key(CHANNEL_KEY)) - if value is not None: - try: - chan_id = value.decode("utf-8") - except Exception as err: # noqa: BLE001 - raise RuntimeError( - f"Invalid channel mapping value for unique_key={CHANNEL_KEY!r}: {value!r}, err={err}" - ) from None - if chan_id.isdigit(): - return chan_id + value = _etcd_call_with_retry( + f"wait channel mapping unique_key={CHANNEL_KEY!r}", + lambda etcd_client: etcd_client.get(mapping_key), + ) + if value is not None: + try: + chan_id = value.decode("utf-8") + except Exception as err: # noqa: BLE001 raise RuntimeError( - f"Invalid channel mapping for unique_key={CHANNEL_KEY!r}: {chan_id!r} (expected digit-only chan_id)" - ) - time.sleep(0.2) + f"Invalid channel mapping value for unique_key={CHANNEL_KEY!r}: {value!r}, err={err}" + ) from None + if chan_id.isdigit(): + return chan_id + raise RuntimeError( + f"Invalid channel mapping for unique_key={CHANNEL_KEY!r}: {chan_id!r} (expected digit-only chan_id)" + ) + time.sleep(0.2) bootstrap_state = _bootstrap_process_state( bootstrap_proc=bootstrap_proc, bootstrap_log=bootstrap_log, @@ -610,8 +615,11 @@ def run_producer(env, args: Dict[str, Any]) -> None: # Close returns Result[OkNone, ApiError]; consume explicitly producer.close().unwrap() finally: - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - etcd_client.put(f"{PRODUCER_DONE_KEY}_{producer_id}", str(produced)) + done_key = f"{PRODUCER_DONE_KEY}_{producer_id}" + _etcd_call_with_retry( + f"mark producer {producer_id} done", + lambda etcd_client: etcd_client.put(done_key, str(produced).encode()), + ) finally: configure_backend(env, backend_type=prev_type, backend_ip=prev_ip) @@ -647,74 +655,76 @@ def run_consumer(env, args: Dict[str, Any]) -> None: start_time = time.monotonic() max_deadline = start_time + MAX_CONSUMER_RUNTIME try: - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - all_producers_done = False - last_producer_check = time.monotonic() - producer_check_interval = 0.5 + all_producers_done = False + last_producer_check = time.monotonic() + producer_check_interval = 0.5 + + while True: + now = time.monotonic() + + if now - last_producer_check >= producer_check_interval: + if not all_producers_done: + done_count = 0 + for p_idx in range(PRODUCER_COUNT): + producer_id = f"P{p_idx}" + done_key = f"{PRODUCER_DONE_KEY}_{producer_id}" + value = _etcd_call_with_retry( + f"read producer {producer_id} done flag", + lambda etcd_client, key=done_key: etcd_client.get(key), + ) + if value is not None: + done_count += 1 + all_producers_done = done_count == PRODUCER_COUNT + if all_producers_done: + msg = f"🎉 Consumer {consumer_id}: All {PRODUCER_COUNT} producers done! consumed={consumed}" + print(msg, file=sys.stdout, flush=True) + last_producer_check = now - while True: - now = time.monotonic() + res = consumer.get_data(batch_size=1, try_time=1) - if now - last_producer_check >= producer_check_interval: - if not all_producers_done: - done_count = 0 - for p_idx in range(PRODUCER_COUNT): - producer_id = f"P{p_idx}" - value, _metadata = etcd_client.get(f"{PRODUCER_DONE_KEY}_{producer_id}") - if value is not None: - done_count += 1 - all_producers_done = done_count == PRODUCER_COUNT - if all_producers_done: - import sys - msg = f"🎉 Consumer {consumer_id}: All {PRODUCER_COUNT} producers done! consumed={consumed}" - print(msg, file=sys.stdout, flush=True) - last_producer_check = now - - res = consumer.get_data(batch_size=1, try_time=1) - - if res is None: - now = time.monotonic() + if res is None: + now = time.monotonic() + if now >= max_deadline: + raise RuntimeError( + f"Consumer {consumer_id} get_data returned None unexpectedly" + ) + elif res.is_ok(): + success = res.unwrap() + now = time.monotonic() + if isinstance(success, list) and success: + consumed += 1 + last_activity = now + if isinstance(success[0], dict): + msg_key = str(success[0]["unique_id"]) + if msg_key.startswith("quick-msg-"): + parts = msg_key.split("-") + if len(parts) >= 3: + producer_id_str = parts[2] + producer_consumed_counts[producer_id_str] = producer_consumed_counts.get(producer_id_str, 0) + 1 + else: + err = res.unwrap_error() + now = time.monotonic() + if isinstance(err, MessageConsumptionNoNewMessageError): if now >= max_deadline: raise RuntimeError( - f"Consumer {consumer_id} get_data returned None unexpectedly" + f"Consumer {consumer_id} exceeded max runtime with no new message" ) - elif res.is_ok(): - success = res.unwrap() - now = time.monotonic() - if isinstance(success, list) and success: - consumed += 1 - last_activity = now - if isinstance(success[0], dict): - msg_key = str(success[0]["unique_id"]) - if msg_key.startswith("quick-msg-"): - parts = msg_key.split("-") - if len(parts) >= 3: - producer_id_str = parts[2] - producer_consumed_counts[producer_id_str] = producer_consumed_counts.get(producer_id_str, 0) + 1 else: - err = res.unwrap_error() - now = time.monotonic() - if isinstance(err, MessageConsumptionNoNewMessageError): - if now >= max_deadline: - raise RuntimeError( - f"Consumer {consumer_id} exceeded max runtime with no new message" - ) - else: - raise RuntimeError( - f"Consumer {consumer_id} get_data failed: {err}" - ) - - idle_time = now - last_activity - if all_producers_done and idle_time >= idle_timeout: - print( - f"✅ Consumer {consumer_id} exiting: all producers done and idle for {idle_time:.1f}s (consumed {consumed} messages)" - ) - break - if now >= max_deadline: raise RuntimeError( - f"Consumer {consumer_id} exceeded max runtime" + f"Consumer {consumer_id} get_data failed: {err}" ) - time.sleep(PRODUCER_DONE_POLL_INTERVAL) + + idle_time = now - last_activity + if all_producers_done and idle_time >= idle_timeout: + print( + f"✅ Consumer {consumer_id} exiting: all producers done and idle for {idle_time:.1f}s (consumed {consumed} messages)" + ) + break + if now >= max_deadline: + raise RuntimeError( + f"Consumer {consumer_id} exceeded max runtime" + ) + time.sleep(PRODUCER_DONE_POLL_INTERVAL) if consumed == 0: raise AssertionError( f"Consumer {consumer_id} did not receive any message" diff --git a/fluxon_py/tests/test_api_chan_mpmc/test_rebind_client.py b/fluxon_py/tests/test_api_chan_mpmc/test_rebind_client.py index 5f42fab..92a967b 100644 --- a/fluxon_py/tests/test_api_chan_mpmc/test_rebind_client.py +++ b/fluxon_py/tests/test_api_chan_mpmc/test_rebind_client.py @@ -12,9 +12,7 @@ import sys import time from pathlib import Path -from typing import List, Tuple - -import etcd3 +from typing import Any, List, Optional, Tuple # Bootstrap import path to project root so absolute imports always work CURRENT_DIR = Path(__file__).resolve().parent @@ -33,8 +31,6 @@ def _find_project_root(start: Path) -> Path: from fluxon_py.api_error import MessageConsumptionNoNewMessageError # noqa: E402 from fluxon_py.logging import init_logger # noqa: E402 from fluxon_py.tests.test_lib import ( # noqa: E402 - ETCD_HOST, - ETCD_PORT, KV_SVC_TYPE, KV_SVC_IP, setup_test_environment, @@ -44,6 +40,10 @@ def _find_project_root(start: Path) -> Path: new_shared_stores, load_test_fluxon_cluster_name, run_with_argmatrix, + etcd_control_call_with_retry as _etcd_call_with_retry, + etcd_control_delete_prefix as _etcd_delete_prefix, + etcd_control_get as _etcd_get, + etcd_control_put as _etcd_put, ) from fluxon_py.kvclient import KvClientType, new_store # noqa: E402 from fluxon_py.kvclient.kvclient_interface import KvClient # noqa: E402 @@ -116,22 +116,20 @@ def _wait_fluxon_member_absent(instance_key: str, *, timeout_s: int = 45) -> Non cluster = load_test_fluxon_cluster_name() key = f"/fluxon_kv_member_base/{cluster}/members/{instance_key}" deadline = time.time() + float(timeout_s) - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - while True: - val = etcd_client.get(key)[0] - if val is None: - return - if time.time() >= deadline: - raise RuntimeError( - f"member key still exists after wait: {key}. Previous lease not expired" - ) - # Progress logging is handled by the caller; keep quiet here. - time.sleep(1.0) + while True: + val = _etcd_get(key) + if val is None: + return + if time.time() >= deadline: + raise RuntimeError( + f"member key still exists after wait: {key}. Previous lease not expired" + ) + # Progress logging is handled by the caller; keep quiet here. + time.sleep(1.0) # ------------------- Local CLI for subprocess workers ------------------- import argparse -from typing import Optional def _build_parser() -> argparse.ArgumentParser: @@ -277,14 +275,25 @@ def _create_store(env: ChannelState, instance_key: str) -> KvClient: # ------------------- Local verification and cleanup ------------------- def clean_etcd() -> None: - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - etcd_client.delete_prefix("/mpmc_channels") - etcd_client.delete_prefix("/channels") - etcd_client.delete_prefix("/test_mpmc_stop_consumer") - etcd_client.delete_prefix("/test_mpmc_consumer") - etcd_client.delete_prefix("/test_mpmc_stop_producer") - etcd_client.delete_prefix(PRODUCER_PAUSE_KEY) - etcd_client.delete_prefix("/test_mpmc_rebind") + _etcd_delete_prefix("/mpmc_channels") + _etcd_delete_prefix("/channels") + _etcd_delete_prefix("/test_mpmc_stop_consumer") + _etcd_delete_prefix("/test_mpmc_consumer") + _etcd_delete_prefix("/test_mpmc_stop_producer") + _etcd_delete_prefix(PRODUCER_PAUSE_KEY) + _etcd_delete_prefix("/test_mpmc_rebind") + + +def _read_log_tail(path: str, *, max_lines: int = 80) -> str: + try: + lines = Path(path).read_text(encoding="utf-8", errors="replace").splitlines() + except FileNotFoundError: + return f"" + except OSError as exc: + return f"" + if not lines: + return "" + return "\n".join(lines[-max_lines:]) def verify_production_consumption_counts( @@ -407,12 +416,11 @@ def run_producer(env, args: argparse.Namespace) -> None: print(f"[Producer-{args.producer_id}] Started", flush=True) try: import uuid, random - etcd_client = producer.etcd_client index = 0 while True: # Check stop first - stop_flag, _ = etcd_client.get("/test_mpmc_stop_producer") + stop_flag = _etcd_get("/test_mpmc_stop_producer") if stop_flag: logging.info( f"[RBD-STOP] Producer-{args.producer_id} stop flag detected" @@ -422,7 +430,7 @@ def run_producer(env, args: argparse.Namespace) -> None: i=0 while True: i+=1 - pause_flag, _ = etcd_client.get(PRODUCER_PAUSE_KEY) + pause_flag = _etcd_get(PRODUCER_PAUSE_KEY) if not pause_flag: logging.info( f"[RBD-RESUME] Producer-{args.producer_id} resumed" @@ -430,7 +438,7 @@ def run_producer(env, args: argparse.Namespace) -> None: break logging.info(f"[RBD-PAUSE] Producer-{args.producer_id} paused, loop i {i}") # allow quick reaction to stop while paused - stop_flag, _ = etcd_client.get("/test_mpmc_stop_producer") + stop_flag = _etcd_get("/test_mpmc_stop_producer") if stop_flag: logging.info( f"[RBD-STOP] Producer-{args.producer_id} stop while paused" @@ -441,7 +449,7 @@ def run_producer(env, args: argparse.Namespace) -> None: break # Read current loop index to embed into message key for verification per loop try: - loop_val, _ = etcd_client.get(REBIND_LOOP_KEY) + loop_val = _etcd_get(REBIND_LOOP_KEY) loop_idx = int(loop_val.decode()) if loop_val else -1 except Exception: loop_idx = -1 @@ -466,7 +474,7 @@ def run_producer(env, args: argparse.Namespace) -> None: ) print(f"PRODUCE_MARKER: {args.producer_id}:{msg_id}") # Track production per loop in etcd for gating - etcd_client.put( + _etcd_put( f"/test_mpmc_rebind/produced/{loop_idx}/{args.producer_id}/{unique_id}", b"", ) @@ -528,11 +536,10 @@ def run_consumer(env, args: argparse.Namespace) -> None: f"[Consumer-{args.consumer_id}] Started with mpmc consumer {consumer.mpmc_channel.mpmc_member_id}", flush=True, ) - etcd_client = consumer.etcd_client - etcd_client.put( + _etcd_put( f"/test_mpmc_consumer/{args.consumer_id}", b"dummy_value", - consumer.mpmc_channel.mpmc_global_lease, + lease_id=int(consumer.mpmc_channel.mpmc_global_lease.id), ) logging.info( f"[RBD-REGISTER] Consumer-{args.consumer_id} registered in etcd" @@ -581,7 +588,7 @@ def run_consumer(env, args: argparse.Namespace) -> None: raise ValueError(f"Invalid loop index in message id: {unique_id_str}") li = int(li_str) if li >= 0: - etcd_client.put( + _etcd_put( f"/test_mpmc_rebind/consumed/{li}/{args.consumer_id}/{unique_id_str}", b"", ) @@ -621,9 +628,7 @@ def run_consumer(env, args: argparse.Namespace) -> None: ) break time.sleep(0.5) - stop_flag, _ = etcd_client.get( - f"/test_mpmc_stop_consumer/{args.consumer_id}" - ) + stop_flag = _etcd_get(f"/test_mpmc_stop_consumer/{args.consumer_id}") if stop_flag: # enter draining mode: keep getting until one timeout/no-data if not draining: @@ -680,11 +685,10 @@ def _once(prefetch: int) -> None: ) shutil.rmtree("logs", ignore_errors=True) clean_etcd() - with etcd3.client(ETCD_HOST, ETCD_PORT) as etcd_client: - if etcd_client.get("/test_mpmc_stop_producer")[0] is not None: - raise RuntimeError( - "precondition failed: /test_mpmc_stop_producer exists before test start" - ) + if _etcd_get("/test_mpmc_stop_producer") is not None: + raise RuntimeError( + "precondition failed: /test_mpmc_stop_producer exists before test start" + ) logging.info("[RBD-CTL-ETCD-CLEAN] cleared test prefixes") os.makedirs("logs", exist_ok=True) @@ -730,91 +734,90 @@ def spawn(process_type: str, cmd: List[str], identifier: str) -> None: ) # Repeatedly stop and restart a single consumer while producers keep producing - etcd_client = etcd3.client(ETCD_HOST, ETCD_PORT) # initialize loop index for producers to tag messages - etcd_client.put(REBIND_LOOP_KEY, b"0") + _etcd_put(REBIND_LOOP_KEY, b"0") logging.info("[RBD-CTL-LOOPKEY] set loop_idx=0") - try: - for i in range(LOOPS - 1): - logging.info(f"[RBD-CTL-LOOP] round={i} active_window={ACTIVE_WINDOW_SEC}s") - # Soft window to allow production - time.sleep(ACTIVE_WINDOW_SEC) - - # Pause producers and stop current consumer to drain until last get_data times out - etcd_client.put(PRODUCER_PAUSE_KEY, b"1") - logging.info("[RBD-CTL-PAUSE] producers paused") - etcd_client.put( - f"/test_mpmc_stop_consumer/{current_consumer}", b"dummy_value" - ) - logging.info( - f"[RBD-CTL-STOP-CONS] request stop consumer={current_consumer}" - ) - while True: - status, _ = etcd_client.get( - f"/test_mpmc_consumer/{current_consumer}" - ) - if not status: - break - time.sleep(0.5) - logging.info( - f"[RBD-CTL-WAIT-CONS] consumer exited id={current_consumer}" - ) - - # Switch to next loop index now that previous consumer fully drained and exited - etcd_client.put(REBIND_LOOP_KEY, str(i + 1).encode()) - logging.info(f"[RBD-CTL-LOOPKEY] set loop_idx={i+1}") + for i in range(LOOPS - 1): + logging.info(f"[RBD-CTL-LOOP] round={i} active_window={ACTIVE_WINDOW_SEC}s") + # Soft window to allow production + time.sleep(ACTIVE_WINDOW_SEC) - # Short gap, then start next consumer for next loop and resume producers - time.sleep(INACTIVE_GAP_SEC) - next_consumer = f"C{i+1}" - print( - f"[rebind_client] starting next consumer {next_consumer}", - flush=True, - ) - spawn( - "consumer", - _consumer_cmd( - env.backend_type, env.backend_ip, next_consumer, prefetch - ), - next_consumer, - ) - current_consumer = next_consumer - # Resume producers for next round - etcd_client.delete(PRODUCER_PAUSE_KEY) - logging.info("[RBD-CTL-RESUME] producers resumed") - - # After last loop index set, stop producers, then stop last consumer (which drains before exit) - etcd_client.put(PRODUCER_PAUSE_KEY, b"1") - logging.info("[RBD-CTL-FINAL-PAUSE] producers paused before shutdown") - etcd_client.put("/test_mpmc_stop_producer", b"dummy_value") - logging.info("[RBD-CTL-STOP-PROD] stop producers signaled") - for process_type, proc, log_file in subprocesses: - if process_type != "producer": - continue - logging.info(f"[RBD-CTL-WAIT-PROD] waiting producer log={log_file}") - proc.wait() - if proc.returncode != 0: - raise RuntimeError( - f"producer failed with return code {proc.returncode}. Check log: {log_file}" - ) - logging.info("[RBD-CTL-PROD-DONE] producers exited") - # Stop the last consumer and wait for consumers to exit (drains until last get timeout) - etcd_client.put( - f"/test_mpmc_stop_consumer/{current_consumer}", b"dummy_value" + # Pause producers and stop current consumer to drain until last get_data times out + _etcd_put(PRODUCER_PAUSE_KEY, b"1") + logging.info("[RBD-CTL-PAUSE] producers paused") + _etcd_put(f"/test_mpmc_stop_consumer/{current_consumer}", b"dummy_value") + logging.info( + f"[RBD-CTL-STOP-CONS] request stop consumer={current_consumer}" ) + while True: + status = _etcd_get(f"/test_mpmc_consumer/{current_consumer}") + if not status: + break + time.sleep(0.5) logging.info( - f"[RBD-CTL-STOP-LAST-CONS] request stop consumer={current_consumer}" + f"[RBD-CTL-WAIT-CONS] consumer exited id={current_consumer}" + ) + + # Switch to next loop index now that previous consumer fully drained and exited + _etcd_put(REBIND_LOOP_KEY, str(i + 1).encode()) + logging.info(f"[RBD-CTL-LOOPKEY] set loop_idx={i+1}") + + # Short gap, then start next consumer for next loop and resume producers + time.sleep(INACTIVE_GAP_SEC) + next_consumer = f"C{i+1}" + print( + f"[rebind_client] starting next consumer {next_consumer}", + flush=True, + ) + spawn( + "consumer", + _consumer_cmd( + env.backend_type, env.backend_ip, next_consumer, prefetch + ), + next_consumer, + ) + current_consumer = next_consumer + # Resume producers for next round + _etcd_call_with_retry( + f"delete producer pause key {PRODUCER_PAUSE_KEY}", + lambda client: client.delete(PRODUCER_PAUSE_KEY), ) - finally: - etcd_client.close() + logging.info("[RBD-CTL-RESUME] producers resumed") + + # After last loop index set, stop producers, then stop last consumer (which drains before exit) + _etcd_put(PRODUCER_PAUSE_KEY, b"1") + logging.info("[RBD-CTL-FINAL-PAUSE] producers paused before shutdown") + _etcd_put("/test_mpmc_stop_producer", b"dummy_value") + logging.info("[RBD-CTL-STOP-PROD] stop producers signaled") + for process_type, proc, log_file in subprocesses: + if process_type != "producer": + continue + logging.info(f"[RBD-CTL-WAIT-PROD] waiting producer log={log_file}") + proc.wait() + if proc.returncode != 0: + raise RuntimeError( + f"producer failed with return code {proc.returncode}. Check log: {log_file}\n" + "--- child log tail ---\n" + f"{_read_log_tail(log_file)}\n" + "--- end child log tail ---" + ) + logging.info("[RBD-CTL-PROD-DONE] producers exited") + # Stop the last consumer and wait for consumers to exit (drains until last get timeout) + _etcd_put(f"/test_mpmc_stop_consumer/{current_consumer}", b"dummy_value") + logging.info( + f"[RBD-CTL-STOP-LAST-CONS] request stop consumer={current_consumer}" + ) for process_type, proc, log_file in subprocesses: logging.info(f"[RBD-CTL-WAIT] waiting {process_type} log={log_file}") proc.wait() if proc.returncode != 0: raise RuntimeError( - f"{process_type} failed with return code {proc.returncode}. Check log: {log_file}" + f"{process_type} failed with return code {proc.returncode}. Check log: {log_file}\n" + "--- child log tail ---\n" + f"{_read_log_tail(log_file)}\n" + "--- end child log tail ---" ) logging.info("[RBD-CTL-ALL-DONE] all subprocesses exited") diff --git a/fluxon_py/tests/test_lib.py b/fluxon_py/tests/test_lib.py index 9be7003..9700f69 100644 --- a/fluxon_py/tests/test_lib.py +++ b/fluxon_py/tests/test_lib.py @@ -8,6 +8,7 @@ from logging import Logger from fluxon_py.logging import init_logger, update_log_level import multiprocessing +import threading from typing import List from fluxon_py.kvclient.kvclient_interface import KvClient from fluxon_py import FluxonKvClientConfig @@ -15,7 +16,7 @@ from fluxon_py import ChanType, ChanRole, chan_new, chan_bind, MPSCChanConsumer, MPMCChanConsumer, MPSCChanProducer, MPMCChanProducer from typing import Optional, Dict, Union from fluxon_py import api_ext_chan -from typing import Any, Callable, Iterable +from typing import Any, Callable, Iterable, TypeVar import signal from typing import Tuple import subprocess @@ -123,6 +124,91 @@ def load_test_chan_config(*, config_path: Optional[Path] = None) -> Dict[str, in TEST_ARGMATRIX: Dict[str, Iterable[Any]] = { "prefetch": (0,), } +T = TypeVar("T") +_TEST_ETCD_CLIENT_LOCK = threading.Lock() +_TEST_ETCD_CLIENT: Optional[Any] = None + + +def etcd_control_client() -> Any: + """Return the cached PyO3 etcd client for process-level test control-plane ops.""" + global _TEST_ETCD_CLIENT + with _TEST_ETCD_CLIENT_LOCK: + if _TEST_ETCD_CLIENT is None: + from fluxon_py.tool import import_fluxon_pyo3_local + + fluxon_pyo3 = import_fluxon_pyo3_local() + _TEST_ETCD_CLIENT = fluxon_pyo3.EtcdKvClient([f"{ETCD_HOST}:{ETCD_PORT}"]) + return _TEST_ETCD_CLIENT + + +def is_transient_etcd_control_failure(exc: BaseException) -> bool: + rendered = f"{type(exc).__name__}: {exc}" + lowered = rendered.lower() + return ( + "unavailable" in lowered + or "etcdserver: request timed out" in rendered + or "timed out" in lowered + or "timeout" in lowered + or "connection" in lowered + or "transport" in lowered + or "broken pipe" in lowered + or "closed" in lowered + ) + + +def etcd_control_call_with_retry( + context: str, + operation: Callable[[Any], T], + *, + max_attempts: int = 5, +) -> T: + if max_attempts <= 0: + raise ValueError(f"max_attempts must be positive, got {max_attempts!r}") + + last_exc: Optional[BaseException] = None + for attempt in range(1, max_attempts + 1): + try: + client = etcd_control_client() + return operation(client) + except Exception as exc: # noqa: BLE001 + last_exc = exc + if (not is_transient_etcd_control_failure(exc)) or attempt == max_attempts: + raise + sleep_s = min(0.2 * attempt, 2.0) + logging.warning( + "transient etcd failure during %s; retrying in %.1fs " + "(attempt %d/%d): %s", + context, + sleep_s, + attempt, + max_attempts, + exc, + ) + time.sleep(sleep_s) + + assert last_exc is not None + raise RuntimeError(f"etcd operation {context!r} failed") from last_exc + + +def etcd_control_get(key: str) -> Optional[bytes]: + return etcd_control_call_with_retry(f"get {key}", lambda client: client.get(key)) + + +def etcd_control_put(key: str, value: bytes, lease_id: Optional[int] = None) -> None: + if lease_id is None: + etcd_control_call_with_retry(f"put {key}", lambda client: client.put(key, value)) + return + etcd_control_call_with_retry( + f"put {key} with lease", + lambda client: client.put(key, value, lease_id=lease_id), + ) + + +def etcd_control_delete_prefix(prefix: str) -> int: + return etcd_control_call_with_retry( + f"delete_prefix {prefix}", + lambda client: client.delete_prefix(prefix), + ) def run_with_argmatrix( diff --git a/fluxon_py/tests/test_pyo3_etcd.py b/fluxon_py/tests/test_pyo3_etcd.py new file mode 100644 index 0000000..4c79451 --- /dev/null +++ b/fluxon_py/tests/test_pyo3_etcd.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import os +import sys +import time +import unittest +import uuid +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from setup_and_pack.utils.repo_config_utils import ( # noqa: E402 + _verify_host_port, + load_test_etcd_address_from_test_config, +) +from fluxon_py.tool import import_fluxon_pyo3_local # noqa: E402 + + +_ETCD_ADDRESS = load_test_etcd_address_from_test_config() +ETCD_HOST, _ETCD_PORT = _verify_host_port(_ETCD_ADDRESS, field="test_config.yaml.etcd_address") +ETCD_PORT = int(_ETCD_PORT) + + +class TestPyO3EtcdKvClient(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.endpoint = f"{ETCD_HOST}:{ETCD_PORT}" + cls.fluxon_pyo3 = import_fluxon_pyo3_local() + + def setUp(self) -> None: + self.prefix = f"/fluxon_py_tests/pyo3_etcd/{os.getpid()}/{uuid.uuid4().hex}/" + self.client = self.fluxon_pyo3.EtcdKvClient([self.endpoint]) + self.addCleanup(self._cleanup_prefix) + + def _cleanup_prefix(self) -> None: + self.client.delete_prefix(self.prefix) + + def _wait_key_deleted(self, key: str, *, timeout_s: float = 10.0) -> None: + deadline = time.time() + timeout_s + while time.time() < deadline: + if self.client.get(key) is None: + return + time.sleep(0.2) + self.fail(f"timed out waiting for etcd key to be deleted: {key}") + + def _assert_lease_not_live(self, lease_id: int) -> None: + try: + ttl = self.client.lease_ttl(lease_id) + except RuntimeError as exc: + if "not found" in str(exc).lower(): + return + raise + self.assertLessEqual(ttl, 0) + + def test_kv_prefix_and_delete_roundtrip(self) -> None: + key_a = self.prefix + "a" + key_b = self.prefix + "nested/b" + + self.assertIsNone(self.client.get(key_a)) + self.client.put(key_a, b"value-a") + self.client.put(key_b, b"value-b") + + self.assertEqual(self.client.get(key_a), b"value-a") + other_client = self.fluxon_pyo3.EtcdKvClient([self.endpoint]) + self.assertEqual(other_client.get(key_b), b"value-b") + + rows = sorted( + (key.decode("utf-8"), value.decode("utf-8")) + for key, value in self.client.get_prefix(self.prefix) + ) + self.assertEqual(rows, [(key_a, "value-a"), (key_b, "value-b")]) + + self.assertTrue(self.client.delete(key_a)) + self.assertFalse(self.client.delete(key_a)) + self.assertIsNone(self.client.get(key_a)) + + self.assertEqual(self.client.delete_prefix(self.prefix), 1) + self.assertEqual(self.client.delete_prefix(self.prefix), 0) + self.assertIsNone(self.client.get(key_b)) + + def test_put_with_lease_and_revoke_deletes_key(self) -> None: + lease_mgr = self.fluxon_pyo3.LeaseManagerHandle() + lease: Any = lease_mgr.allocate_etcd_lease([self.endpoint], 30, False) + lease_id = int(lease.id) + key = self.prefix + "leased" + + self.client.put(key, b"leased-value", lease_id=lease_id) + self.assertEqual(self.client.get(key), b"leased-value") + self.assertGreater(self.client.lease_ttl(lease_id), 0) + + self.client.revoke_lease(lease_id) + self._wait_key_deleted(key) + self._assert_lease_not_live(lease_id) + + def test_lock_exclusive_and_context_manager_release(self) -> None: + lock_name = self.prefix + "lock" + lock_a = self.fluxon_pyo3.EtcdLock([self.endpoint], lock_name, 10, 1.0) + lock_b = self.fluxon_pyo3.EtcdLock([self.endpoint], lock_name, 10, 0.5) + + self.assertFalse(lock_a.held) + self.assertIsNone(lock_a.lease_id) + + self.assertTrue(lock_a.acquire()) + self.assertTrue(lock_a.held) + self.assertIsInstance(lock_a.lease_id, int) + + self.assertFalse(lock_b.acquire()) + self.assertFalse(lock_b.held) + self.assertIsNone(lock_b.lease_id) + + first_lease_id = int(lock_a.lease_id) + self.assertTrue(lock_a.release()) + self.assertFalse(lock_a.held) + self.assertIsNone(lock_a.lease_id) + self.assertFalse(lock_a.release()) + self._assert_lease_not_live(first_lease_id) + + with self.fluxon_pyo3.EtcdLock([self.endpoint], lock_name, 10, 1.0) as held_lock: + self.assertTrue(held_lock.held) + context_lease_id = int(held_lock.lease_id) + self._assert_lease_not_live(context_lease_id) + + +if __name__ == "__main__": + raise SystemExit(unittest.main()) diff --git a/fluxon_release/test_rsc/source/prepare.yaml b/fluxon_release/test_rsc/source/prepare.yaml index fb0c5ac..d9c842f 100644 --- a/fluxon_release/test_rsc/source/prepare.yaml +++ b/fluxon_release/test_rsc/source/prepare.yaml @@ -21,6 +21,10 @@ python_runtime: source: wheel - pinned: pytest==8.3.5 source: wheel + - pinned: tomli==2.2.1 + source: wheel + - pinned: exceptiongroup==1.3.0 + source: wheel zerorpc: requirements: - pinned: zerorpc==0.6.3 diff --git a/fluxon_rs/fluxon_pyo3/build.rs b/fluxon_rs/fluxon_pyo3/build.rs index 3ff8c5b..90c419b 100644 --- a/fluxon_rs/fluxon_pyo3/build.rs +++ b/fluxon_rs/fluxon_pyo3/build.rs @@ -1,6 +1,7 @@ use std::{ - env, + env, fs, path::{Path, PathBuf}, + process::Command, }; const DEFAULT_RUNTIME_SEARCH_SUBDIRS: &[&str] = &[ @@ -13,7 +14,53 @@ const DEFAULT_RUNTIME_SEARCH_SUBDIRS: &[&str] = &[ ]; const CLOSED_SDK_RUNTIME_ROOT_DIR_NAMES: &[&str] = &["native_runtime", "vendor_runtime"]; +const PYTHON_TEST_EMBED_LINK_ARGS_SCRIPT: &str = r#" +import sysconfig + +args = [] +seen = set() + +def add(arg): + arg = str(arg or "").strip() + if arg and arg not in seen: + seen.add(arg) + args.append(arg) + +for key in ("LIBPL", "LIBDIR"): + path = sysconfig.get_config_var(key) + if path: + add("-L" + path) + +libname = "" +ldlibrary = sysconfig.get_config_var("LDLIBRARY") or sysconfig.get_config_var("LIBRARY") or "" +if ldlibrary.startswith("libpython"): + stem = ldlibrary[3:] + for suffix in (".so", ".a", ".dylib"): + pos = stem.find(suffix) + if pos >= 0: + stem = stem[:pos] + break + libname = stem + +if not libname: + version = sysconfig.get_config_var("VERSION") or "" + abiflags = sysconfig.get_config_var("ABIFLAGS") or "" + if version: + libname = "python" + version + abiflags + +if libname: + add("-l" + libname) + +for key in ("LIBS", "SYSLIBS"): + for arg in (sysconfig.get_config_var(key) or "").split(): + add(arg) + +print("\n".join(args)) +"#; + fn main() { + emit_python_test_embed_link_args(); + let target_dir = get_target_dir(); let runtime_search_subdirs = load_runtime_search_subdirs(); let runtime_root_dir_names = runtime_root_dir_names(); @@ -59,6 +106,105 @@ fn main() { println!("cargo:rerun-if-changed=../target/release/"); } +fn emit_python_test_embed_link_args() { + println!("cargo:rerun-if-env-changed=PYTHON"); + + let python = env::var("PYTHON") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| "python3".to_string()); + + let output = match Command::new(&python) + .arg("-c") + .arg(PYTHON_TEST_EMBED_LINK_ARGS_SCRIPT) + .output() + { + Ok(output) => output, + Err(err) => { + write_python_test_link_source( + None, + &format!( + "failed to query Python embed link args with {}: {}", + python, err + ), + ); + println!( + "cargo:warning=failed to query Python embed link args with {}: {}", + python, err + ); + return; + } + }; + + if !output.status.success() { + write_python_test_link_source( + None, + &format!( + "failed to query Python embed link args with {}: status={} stderr={}", + python, + output.status, + String::from_utf8_lossy(&output.stderr).trim() + ), + ); + println!( + "cargo:warning=failed to query Python embed link args with {}: status={} stderr={}", + python, + output.status, + String::from_utf8_lossy(&output.stderr).trim() + ); + return; + } + + let mut python_link_lib = None; + for arg in String::from_utf8_lossy(&output.stdout) + .lines() + .map(str::trim) + .filter(|arg| !arg.is_empty()) + { + if let Some(path) = arg.strip_prefix("-L") { + if !path.is_empty() { + println!("cargo:rustc-link-search=native={path}"); + } + continue; + } + + if let Some(lib) = arg.strip_prefix("-l") { + if lib.starts_with("python") { + python_link_lib = Some(lib.to_string()); + } + } + } + + if let Some(lib) = python_link_lib { + write_python_test_link_source(Some(&lib), ""); + } else { + let message = format!( + "Python embed link args from {} did not include a libpython entry", + python + ); + write_python_test_link_source(None, &message); + println!( + "cargo:warning=Python embed link args from {} did not include a libpython entry", + python + ); + } +} + +fn write_python_test_link_source(python_link_lib: Option<&str>, message: &str) { + let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap()); + let path = out_dir.join("python_test_link.rs"); + let source = match python_link_lib { + Some(lib) => format!("#[link(name = {lib:?})]\nunsafe extern \"C\" {{}}\n"), + None => format!("compile_error!({:?});\n", message), + }; + fs::write(&path, source).expect("write generated Python test link source"); + println!( + "cargo:rustc-env=FLUXON_PYO3_TEST_PYTHON_LINK_RS={}", + path.display() + ); +} + fn get_target_dir() -> PathBuf { if let Ok(target_dir) = env::var("CARGO_TARGET_DIR") { let path = PathBuf::from(target_dir); diff --git a/fluxon_rs/fluxon_pyo3/src/etcd.rs b/fluxon_rs/fluxon_pyo3/src/etcd.rs index 2ceace8..f3c27d7 100644 --- a/fluxon_rs/fluxon_pyo3/src/etcd.rs +++ b/fluxon_rs/fluxon_pyo3/src/etcd.rs @@ -1,12 +1,417 @@ use etcd_client as etcd; +use fluxon_util::auto_clean_map::{AutoCleanMap, AutoCleanMapEntry}; use fluxon_util::run_async_from_sync::SyncAsyncBridge; use pyo3::prelude::*; +use pyo3::pybacked::PyBackedBytes; +use pyo3::types::{PyBytes, PyList, PyTuple}; use pyo3::{PyErr, PyObject}; -use std::sync::Arc; +use std::future::Future; +use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; use tokio::runtime::Runtime; use tracing::debug; +fn normalize_raw_endpoint(endpoint: &str) -> PyResult { + let endpoint = endpoint.trim(); + if endpoint.is_empty() { + return Err(PyErr::new::( + "etcd endpoint must be non-empty raw host:port", + )); + } + if endpoint.contains("://") { + return Err(PyErr::new::(format!( + "etcd endpoint must be raw host:port without scheme, got: {}", + endpoint + ))); + } + Ok(format!("http://{}", endpoint)) +} + +fn normalize_raw_endpoints(endpoints: Vec, component: &str) -> PyResult> { + if endpoints.is_empty() { + return Err(PyErr::new::(format!( + "{} requires at least one endpoint", + component + ))); + } + let mut normalized = Vec::with_capacity(endpoints.len()); + for endpoint in endpoints { + normalized.push(normalize_raw_endpoint(&endpoint)?); + } + Ok(normalized) +} + +struct EtcdKvBackend { + endpoints: Vec, + client: tokio::sync::RwLock>, +} + +impl EtcdKvBackend { + fn new(endpoints: Vec) -> Self { + Self { + endpoints, + client: tokio::sync::RwLock::new(None), + } + } + + async fn client(&self) -> anyhow::Result { + { + let guard = self.client.read().await; + if let Some(client) = guard.as_ref() { + return Ok(client.clone()); + } + } + + let mut guard = self.client.write().await; + if let Some(client) = guard.as_ref() { + return Ok(client.clone()); + } + + let client = etcd::Client::connect(self.endpoints.clone(), None) + .await + .map_err(|e| { + anyhow::anyhow!( + "failed to connect etcd endpoints={:?}: {:?}", + self.endpoints, + e + ) + })?; + *guard = Some(client.clone()); + Ok(client) + } + + async fn clear_client(&self) { + let mut guard = self.client.write().await; + *guard = None; + } +} + +fn etcd_kv_backend_map() -> &'static AutoCleanMap, EtcdKvBackend> { + static MAP: OnceLock, EtcdKvBackend>> = OnceLock::new(); + MAP.get_or_init(|| AutoCleanMap::new()) +} + +fn is_reconnectable_etcd_error(err: &etcd::Error) -> bool { + is_reconnectable_etcd_error_text(&format!("{:?}", err)) +} + +fn is_reconnectable_etcd_error_text(msg: &str) -> bool { + let msg = msg.to_ascii_lowercase(); + msg.contains("unavailable") + || msg.contains("connection") + || msg.contains("transport") + || msg.contains("timed out") + || msg.contains("timeout") + || msg.contains("broken pipe") + || msg.contains("closed") +} + +async fn run_etcd_op( + backend: AutoCleanMapEntry, EtcdKvBackend>, + context: String, + mut op: F, +) -> anyhow::Result +where + F: FnMut(etcd::Client) -> Fut, + Fut: Future>, +{ + let mut last_err = None; + for attempt in 1..=2 { + let client = backend.client().await?; + match op(client).await { + Ok(value) => return Ok(value), + Err(err) => { + let should_retry = attempt == 1 && is_reconnectable_etcd_error(&err); + last_err = Some(err); + if should_retry { + backend.clear_client().await; + continue; + } + let err = last_err.take().expect("etcd error must be recorded"); + return Err(anyhow::anyhow!("{}: {:?}", context, err)); + } + } + } + + let err = last_err.expect("etcd retry loop must record the last error"); + Err(anyhow::anyhow!("{}: {:?}", context, err)) +} + +#[pyclass(name = "EtcdKvClient")] +pub struct PyEtcdKvClient { + rt: Arc, + endpoints: Vec, + backend: AutoCleanMapEntry, EtcdKvBackend>, +} + +#[pymethods] +impl PyEtcdKvClient { + #[new] + fn new(endpoints: Vec) -> PyResult { + let endpoints = normalize_raw_endpoints(endpoints, "EtcdKvClient")?; + let backend = etcd_kv_backend_map() + .get_or_init(endpoints.clone(), || EtcdKvBackend::new(endpoints.clone())); + Ok(Self { + rt: crate::mpsc::get_global_runtime(), + endpoints, + backend, + }) + } + + fn get(&self, py: Python<'_>, key: String) -> PyResult>> { + if key.is_empty() { + return Err(PyErr::new::( + "etcd get key must not be empty", + )); + } + + let backend = self.backend.clone(); + let key_for_op = key.clone(); + let value = py + .allow_threads(|| { + self.rt.run_async_from_sync(async move { + let resp = run_etcd_op( + backend, + format!("etcd get failed for key={}", key), + move |mut client| { + let key = key_for_op.clone(); + async move { client.get(key, None).await } + }, + ) + .await?; + Ok::>, anyhow::Error>( + resp.kvs().first().map(|kv| kv.value().to_vec()), + ) + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.get: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(value.map(|raw| PyBytes::new_bound(py, &raw).into())) + } + + fn get_prefix(&self, py: Python<'_>, prefix: String) -> PyResult { + if prefix.is_empty() { + return Err(PyErr::new::( + "etcd get_prefix prefix must not be empty", + )); + } + + let backend = self.backend.clone(); + let prefix_for_op = prefix.clone(); + let rows = py + .allow_threads(|| { + self.rt.run_async_from_sync(async move { + let resp = run_etcd_op( + backend, + format!("etcd get_prefix failed for prefix={}", prefix), + move |mut client| { + let prefix = prefix_for_op.clone(); + async move { + client + .get(prefix, Some(etcd::GetOptions::new().with_prefix())) + .await + } + }, + ) + .await?; + Ok::, Vec)>, anyhow::Error>( + resp.kvs() + .iter() + .map(|kv| (kv.key().to_vec(), kv.value().to_vec())) + .collect(), + ) + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.get_prefix: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string()))?; + + let out = PyList::empty_bound(py); + for (key, value) in rows { + let item = PyTuple::new_bound( + py, + [ + PyBytes::new_bound(py, &key).into_py(py), + PyBytes::new_bound(py, &value).into_py(py), + ], + ); + out.append(item)?; + } + Ok(out.into_any().into_py(py)) + } + + #[pyo3(signature = (key, value, lease_id=None))] + fn put( + &self, + py: Python<'_>, + key: String, + value: PyBackedBytes, + lease_id: Option, + ) -> PyResult<()> { + if key.is_empty() { + return Err(PyErr::new::( + "etcd put key must not be empty", + )); + } + if let Some(lease_id) = lease_id { + if lease_id <= 0 { + return Err(PyErr::new::(format!( + "lease_id must be positive, got {}", + lease_id + ))); + } + } + + let backend = self.backend.clone(); + let key_for_op = key.clone(); + let value = value.as_ref().to_vec(); + py.allow_threads(|| { + self.rt.run_async_from_sync(async move { + run_etcd_op( + backend, + format!("etcd put failed for key={}", key), + move |mut client| { + let key = key_for_op.clone(); + let value = value.clone(); + async move { + let opts = lease_id.map(|id| etcd::PutOptions::new().with_lease(id)); + client.put(key, value, opts).await.map(|_| ()) + } + }, + ) + .await?; + Ok::<(), anyhow::Error>(()) + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.put: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string())) + } + + fn delete(&self, py: Python<'_>, key: String) -> PyResult { + if key.is_empty() { + return Err(PyErr::new::( + "etcd delete key must not be empty", + )); + } + + let backend = self.backend.clone(); + let key_for_op = key.clone(); + py.allow_threads(|| { + self.rt.run_async_from_sync(async move { + run_etcd_op( + backend, + format!("etcd delete failed for key={}", key), + move |mut client| { + let key = key_for_op.clone(); + async move { + client + .delete(key, None) + .await + .map(|resp| resp.deleted() > 0) + } + }, + ) + .await + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.delete: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string())) + } + + fn delete_prefix(&self, py: Python<'_>, prefix: String) -> PyResult { + if prefix.is_empty() { + return Err(PyErr::new::( + "etcd delete_prefix prefix must not be empty", + )); + } + + let backend = self.backend.clone(); + let prefix_for_op = prefix.clone(); + py.allow_threads(|| { + self.rt.run_async_from_sync(async move { + run_etcd_op( + backend, + format!("etcd delete_prefix failed for prefix={}", prefix), + move |mut client| { + let prefix = prefix_for_op.clone(); + async move { + client + .delete(prefix, Some(etcd::DeleteOptions::new().with_prefix())) + .await + .map(|resp| resp.deleted()) + } + }, + ) + .await + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.delete_prefix: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string())) + } + + fn lease_ttl(&self, py: Python<'_>, lease_id: i64) -> PyResult { + if lease_id <= 0 { + return Err(PyErr::new::(format!( + "lease_id must be positive, got {}", + lease_id + ))); + } + + let backend = self.backend.clone(); + py.allow_threads(|| { + self.rt.run_async_from_sync(async move { + run_etcd_op( + backend, + format!("etcd lease_ttl failed for lease_id={}", lease_id), + move |mut client| async move { + client + .lease_time_to_live(lease_id, None) + .await + .map(|resp| resp.ttl()) + }, + ) + .await + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.lease_ttl: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string())) + } + + fn revoke_lease(&self, py: Python<'_>, lease_id: i64) -> PyResult<()> { + if lease_id <= 0 { + return Err(PyErr::new::(format!( + "lease_id must be positive, got {}", + lease_id + ))); + } + + let backend = self.backend.clone(); + py.allow_threads(|| { + self.rt.run_async_from_sync(async move { + run_etcd_op( + backend, + format!("etcd revoke_lease failed for lease_id={}", lease_id), + move |mut client| async move { client.lease_revoke(lease_id).await.map(|_| ()) }, + ) + .await + }) + }) + .map_err(|e| anyhow::anyhow!("runtime bridge failed in EtcdKvClient.revoke_lease: {}", e)) + .map_err(|e| PyErr::new::(e.to_string()))? + .map_err(|e| PyErr::new::(e.to_string())) + } + + fn __repr__(&self) -> String { + format!("", self.endpoints) + } +} + #[pyclass(name = "EtcdLock")] pub struct PyEtcdLock { rt: Arc, @@ -28,11 +433,7 @@ impl PyEtcdLock { ttl_seconds: i64, timeout_seconds: Option, ) -> PyResult { - if endpoints.is_empty() { - return Err(PyErr::new::( - "EtcdLock requires at least one endpoint", - )); - } + let endpoints = normalize_raw_endpoints(endpoints, "EtcdLock")?; if ttl_seconds <= 0 { return Err(PyErr::new::(format!( "EtcdLock ttl_seconds must be > 0, got {}", @@ -266,3 +667,119 @@ impl PyEtcdLock { ) } } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + static TEST_KEY_SEQ: AtomicUsize = AtomicUsize::new(1); + + fn unique_test_endpoints() -> Vec { + let seq = TEST_KEY_SEQ.fetch_add(1, Ordering::Relaxed); + vec![format!("http://unit-test-etcd-backend-{}", seq)] + } + + #[test] + fn normalize_raw_endpoint_accepts_raw_host_port() { + assert_eq!( + normalize_raw_endpoint(" 127.0.0.1:2379 ").unwrap(), + "http://127.0.0.1:2379" + ); + } + + #[test] + fn normalize_raw_endpoint_rejects_empty_or_schemed_endpoint() { + assert!(normalize_raw_endpoint("").is_err()); + assert!(normalize_raw_endpoint(" ").is_err()); + assert!(normalize_raw_endpoint("http://127.0.0.1:2379").is_err()); + assert!(normalize_raw_endpoint("https://127.0.0.1:2379").is_err()); + } + + #[test] + fn normalize_raw_endpoints_requires_at_least_one_endpoint() { + assert!(normalize_raw_endpoints(Vec::new(), "EtcdKvClient").is_err()); + assert_eq!( + normalize_raw_endpoints( + vec!["127.0.0.1:2379".to_string(), "localhost:2380".to_string()], + "EtcdKvClient", + ) + .unwrap(), + vec![ + "http://127.0.0.1:2379".to_string(), + "http://localhost:2380".to_string() + ] + ); + } + + #[test] + fn etcd_kv_client_constructor_normalizes_raw_endpoints() { + let client = PyEtcdKvClient::new(vec!["127.0.0.1:2379".to_string()]).unwrap(); + assert_eq!(client.endpoints, vec!["http://127.0.0.1:2379"]); + } + + #[test] + fn etcd_lock_constructor_normalizes_raw_endpoints() { + let lock = PyEtcdLock::new( + vec!["127.0.0.1:2379".to_string()], + "/unit-test/lock".to_string(), + 10, + Some(1.0), + ) + .unwrap(); + assert_eq!(lock.endpoints, vec!["http://127.0.0.1:2379"]); + } + + #[test] + fn etcd_lock_constructor_rejects_schemed_endpoints() { + assert!( + PyEtcdLock::new( + vec!["http://127.0.0.1:2379".to_string()], + "/unit-test/lock".to_string(), + 10, + Some(1.0), + ) + .is_err() + ); + } + + #[test] + fn etcd_kv_backend_map_reuses_and_auto_cleans_live_entries() { + let endpoints = unique_test_endpoints(); + let map = etcd_kv_backend_map(); + assert!(map.with_existing(&endpoints, |_| ()).is_none()); + + { + let entry_a = + map.get_or_init(endpoints.clone(), || EtcdKvBackend::new(endpoints.clone())); + assert!(map.with_existing(&endpoints, |_| ()).is_some()); + + { + let entry_b = map.get_or_init(endpoints.clone(), || { + panic!("live backend entry should be reused") + }); + assert!(std::ptr::eq(&*entry_a, &*entry_b)); + } + + assert!(map.with_existing(&endpoints, |_| ()).is_some()); + } + + assert!(map.with_existing(&endpoints, |_| ()).is_none()); + } + + #[test] + fn reconnectable_error_text_matches_transient_transport_failures() { + assert!(is_reconnectable_etcd_error_text("StatusCode::UNAVAILABLE")); + assert!(is_reconnectable_etcd_error_text( + "etcdserver: request timed out" + )); + assert!(is_reconnectable_etcd_error_text("transport error")); + assert!(is_reconnectable_etcd_error_text("connection closed")); + assert!(is_reconnectable_etcd_error_text("broken pipe")); + + assert!(!is_reconnectable_etcd_error_text( + "requested lease not found" + )); + assert!(!is_reconnectable_etcd_error_text("permission denied")); + } +} diff --git a/fluxon_rs/fluxon_pyo3/src/lib.rs b/fluxon_rs/fluxon_pyo3/src/lib.rs index a73591f..631f1ea 100644 --- a/fluxon_rs/fluxon_pyo3/src/lib.rs +++ b/fluxon_rs/fluxon_pyo3/src/lib.rs @@ -50,6 +50,11 @@ use std::os::fd::IntoRawFd; use std::time::Duration; use tokio::runtime::Runtime; +// Unit tests build a native test binary, so PyO3's extension-module mode must +// link libpython there. The cdylib/wheel build keeps Python symbols unresolved. +#[cfg(test)] +include!(env!("FLUXON_PYO3_TEST_PYTHON_LINK_RS")); + mod memholder; pub use memholder::{ExternalMemHolder, MemHolder}; mod flatdict_zerocopy; @@ -58,7 +63,7 @@ pub use kvfuture::KvFuture; mod error; mod etcd; mod mpsc; // Python ApiError constructors and MPSC error mapping -pub use etcd::PyEtcdLock; +pub use etcd::{PyEtcdKvClient, PyEtcdLock}; pub use mpsc::{MpscConsumerHandle, MpscContext, MpscProducerHandle}; mod lease_manager; pub use lease_manager::{LeaseManagerHandle, PyGeneralLease, PyLeaseBackendUid}; @@ -4154,6 +4159,7 @@ fn fluxon_pyo3(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; // English note: keep the `from fluxon_pyo3 import LeaseManagerHandle` import path stable for Python users. m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/fluxon_test_stack/tests/test_test_rsc_prepare_yaml.py b/fluxon_test_stack/tests/test_test_rsc_prepare_yaml.py index f723fb9..d81b59e 100644 --- a/fluxon_test_stack/tests/test_test_rsc_prepare_yaml.py +++ b/fluxon_test_stack/tests/test_test_rsc_prepare_yaml.py @@ -26,6 +26,8 @@ def test_base_dependency_set_includes_pytest(self) -> None: self.assertIsInstance(requirements, list) pinned = {item.get("pinned") for item in requirements if isinstance(item, dict)} self.assertIn("pytest==8.3.5", pinned) + self.assertIn("tomli==2.2.1", pinned) + self.assertIn("exceptiongroup==1.3.0", pinned) if __name__ == "__main__": diff --git a/fluxon_test_stack/tests/test_top_attention_index_helper.py b/fluxon_test_stack/tests/test_top_attention_index_helper.py index dcec087..7857366 100644 --- a/fluxon_test_stack/tests/test_top_attention_index_helper.py +++ b/fluxon_test_stack/tests/test_top_attention_index_helper.py @@ -62,6 +62,12 @@ def test_top_attention_scene_id_uses_stable_prefix(self) -> None: "ci_top_attention_bin_kvtest", ) + def test_kv_py_core_keeps_pyo3_etcd_integration_test(self) -> None: + entry_text = ( + REPO_ROOT / "fluxon_test_stack" / "top_attention_test_index" / "_kv_py_core.py" + ).read_text(encoding="utf-8") + self.assertIn("fluxon_py/tests/test_pyo3_etcd.py", entry_text) + if __name__ == "__main__": unittest.main() diff --git a/fluxon_test_stack/top_attention_test_index/_kv_py_core.py b/fluxon_test_stack/top_attention_test_index/_kv_py_core.py index 924ca7a..6459407 100755 --- a/fluxon_test_stack/top_attention_test_index/_kv_py_core.py +++ b/fluxon_test_stack/top_attention_test_index/_kv_py_core.py @@ -11,6 +11,7 @@ def main() -> int: return run_pytest( "Flat index entry for Python KV backend core smoke tests.", [ + "fluxon_py/tests/test_pyo3_etcd.py", "fluxon_py/tests/test_backend.py", "fluxon_py/tests/test_backend_fallback_close.py", ], diff --git a/scripts/preview_doc_html.py b/scripts/preview_doc_html.py new file mode 100755 index 0000000..18f5093 --- /dev/null +++ b/scripts/preview_doc_html.py @@ -0,0 +1,547 @@ +#!/usr/bin/env python3 +"""Generate a standalone HTML preview for a Markdown document.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_OUTPUT_ROOT = REPO_ROOT / ".dever/html_preview" + + +HTML_TEMPLATE = """ + + + + + {title} + + + + + + +
+
{title}
+
+ Waiting + + +
+
+ +
+
+
+ + + + + + + + + + + +""" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Generate a standalone HTML preview for a Markdown document with Mermaid fullscreen zoom/pan support." + ) + parser.add_argument( + "markdown_path", + type=Path, + help="Markdown input path.", + ) + parser.add_argument( + "--output", + type=Path, + default=None, + help=f"HTML output path. Default: {DEFAULT_OUTPUT_ROOT.relative_to(REPO_ROOT)}/.html", + ) + parser.add_argument( + "--open", + action="store_true", + help="Open the generated HTML file with the system browser.", + ) + return parser.parse_args() + + +def default_output_path(input_path: Path) -> Path: + try: + relative_input = input_path.relative_to(REPO_ROOT) + except ValueError: + digest = hashlib.sha256(str(input_path).encode("utf-8")).hexdigest()[:12] + relative_input = Path("external") / f"{input_path.stem}-{digest}{input_path.suffix}" + return (DEFAULT_OUTPUT_ROOT / relative_input).with_suffix(".html") + + +def infer_html_lang(input_path: Path, markdown: str) -> str: + parts = set(input_path.parts) + if "fluxon_doc_cn" in parts or input_path.name == "README_CN.md": + return "zh-CN" + if "fluxon_doc_en" in parts or input_path.name == "README.md": + return "en" + return "zh-CN" if re.search(r"[\u4e00-\u9fff]", markdown) else "en" + + +def first_markdown_heading(markdown: str, fallback: str) -> str: + for line in markdown.splitlines(): + stripped = line.strip() + if stripped.startswith("# "): + return stripped[2:].strip() or fallback + return fallback + + +def open_in_browser(path: Path) -> None: + if sys.platform == "darwin": + subprocess.run(["open", str(path)], check=False) + elif os.name == "nt": + os.startfile(str(path)) # type: ignore[attr-defined] + else: + subprocess.run(["xdg-open", str(path)], check=False) + + +def main() -> int: + args = parse_args() + input_path = args.markdown_path.resolve() + + if not input_path.exists(): + print(f"Input markdown does not exist: {input_path}", file=sys.stderr) + return 1 + if not input_path.is_file(): + print(f"Input path is not a file: {input_path}", file=sys.stderr) + return 1 + + output_path = args.output.resolve() if args.output else default_output_path(input_path) + + markdown = input_path.read_text(encoding="utf-8") + title = first_markdown_heading(markdown, input_path.stem) + html = HTML_TEMPLATE.format( + title=title, + html_lang=infer_html_lang(input_path, markdown), + markdown_json=json.dumps(markdown, ensure_ascii=False), + ) + + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(html, encoding="utf-8") + + print(f"Generated HTML preview: {output_path}") + print("Open it in a browser. Mermaid fullscreen zoom/pan uses CDN assets, so internet access is required.") + + if args.open: + open_in_browser(output_path) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())