test largescale mq

[ActivePeter]

PR #34: Harden MPMC lifecycle, P2P recovery, and large-scale MQ validation

Suggested PR title:

fix(mq): harden MPMC lifecycle, P2P recovery, and large-scale validation

English

Summary

This PR adds a dedicated large-scale MPMC validation path and fixes the MQ, lease, shutdown, and P2P issues exposed by high-concurrency runs.

The resulting contract separates three concerns clearly:

• MQ endpoint lifecycle owns local runtime teardown and role-specific cleanup.
• Lease handles contribute keepalive; cleanup ownership is defined by the MQ object that owns each distributed key.
• Benchmark completion proves both distributed lifecycle convergence and zero failed operations.

The PR also updates the test stack, same-host process orchestration, closed SDK artifacts, user documentation, and regression coverage required to exercise these contracts continuously.

Problems exposed at scale

Large MPMC runs exposed several independent failure modes:

• Existing producers serialized on a distributed unique-channel lock even after the channel mapping had already been published.
• Producers eagerly bound every ready sub-MPSC, creating a large startup burst of handles, etcd operations, and lease registrations.
• The MPMC create lock could expire while a sub-MPSC was still being constructed. A stale writer could then overwrite a newer channel-list snapshot or publish channel and ready state inconsistently.
• Parent MPMC and child MPSC lease ownership was ambiguous. Some child handles only carried lease IDs instead of real keepalive contributors, while close paths could revoke or release state owned by other members.
• Concurrent close, blocked put/get, lazy bind, commit sequencing, and KV teardown could wait on one another or miss a shutdown notification.
• Closed-SDK P2P failures lost their transport category, so transient route failures could bypass recovery or end as a generic timeout with the useful root cause discarded.
• Benchmark nodes could enter the timed window before every MPMC runtime endpoint was ready.
• The large-scale validator checked lifecycle completion and positive successful-operation counts, but did not reject total_failed_ops > 0.

Public MQ lifecycle contract

The user-facing shutdown order is now explicit:

producer.close().unwrap("close MQ producer failed")
# or consumer.close().unwrap("close MQ consumer failed")

store.close().unwrap("close KV store failed")

If one KvClient owns multiple MQ endpoints, every producer and consumer must be closed and its Result consumed before closing the store.

Endpoint close() now has one canonical responsibility boundary:

• publish shutdown to active put/get/bind operations;
• close every endpoint-owned subchannel;
• request and await Rust MQ framework shutdown;
• stop watchers, actors, keepalive guards, and local handles;
• make one best-effort delete pass for endpoint-owned role, ready, membership, and weight keys;
• return an ApiError when required local teardown fails, leaving the endpoint retryable.

Distributed leased-key deletion remains best effort. A delete timeout or backend error is logged as a warning, local teardown continues, and the corresponding backend TTL performs final reclamation. Create, bind, publish, ready-claim, message acknowledgement, and offset-update operations retain their strong error contract.

KvClient now owns a weak-reference child-close registry. Direct MPSC endpoints and outer MPMC endpoints register their public close() method before entering blocking construction. KvClient.close() stops new registrations, waits for in-progress child construction, closes every registered endpoint, consumes every Result, and only then tears down the native KV backend. A child close failure keeps the backend open and remains retryable.

MPMC topology and binding fixes

Existing-channel fast bind

new_or_bind_with_unique_key() now reads an already-published unique mapping and binds it without taking the distributed create lock. The read has bounded transient retries. The lock remains authoritative only for channel creation and stale-mapping cleanup.

The bind path returns each newly constructed endpoint directly instead of publishing it through a process-global get/delete handoff, so concurrent callers retain their own handle.

Lazy producer subchannel binding

An MPMC producer now tracks ready subchannel IDs in its local scheduling queue without eagerly constructing a producer for every sub-MPSC. It binds the selected route on first use and reuses that local producer afterward.

The scheduler keeps ready channels rotating locally, while shutdown is rechecked before and after a potentially blocking lazy bind. If close wins the race, the new sub-producer is closed immediately; a cleanup failure preserves it for a later close retry.

Atomic subchannel publication and claiming

The MPMC create lock is now treated as an optimization rather than the final consistency authority. Publication compares the exact channel-list value read before sub-MPSC construction:

• Producer publication uses a snapshot CAS and rolls back an unpublished subchannel if another creator wins.
• Consumer publication atomically appends the channel ID and creates its ready key in one etcd transaction.
• A stale consumer writer only retries when the channel list advanced monotonically.
• If another creator published an unclaimed subchannel, the loser rolls back its temporary channel and tries to claim the published one.
• Unexpected duplicate IDs, non-monotonic snapshots, and rollback failures are surfaced as errors.

Existing unready subchannels are claimed with a ready-key CAS before consumer binding. This prevents two consumers from binding the same sub-MPSC from stale snapshots.

Producer membership publication

Rust MPSC producer binding now publishes membership and producer weight in one transaction. The path has:

• a five-second per-RPC timeout;
• three bounded attempts;
• idempotent reconciliation when a timed-out request may already have committed;
• conflict detection when existing membership does not match the intended lease and value;
• cleanup of both membership and weight after terminal failure or shutdown.

Shutdown is checked before metadata load, before publication, during retry, and after publication so a close race cannot leave a newly published producer behind unnoticed.

Lease ownership and channel construction

The lease model now separates keepalive contribution from cleanup ownership:

• GeneralLease represents a real local keepalive contribution.
• Dropping the final local guard unregisters future keepalive work; it does not revoke the backend lease or delete shared keys.
• GeneralLease::Borrowed and the revoke_on_drop branch are removed.
• Normal close explicitly deletes only keys owned by that endpoint; shared metadata, payload state, and allocator state remain until their final contributor stops and TTL expires.

Fluxon KV payload lease allocation and keepalive now stay in native Rust. LeaseBackendUid::KvClient carries async Rust operations backed by the actual Framework, and its identity includes both cluster and client instance. This avoids Python callbacks, blocking-pool round trips, and accidental backend reuse between different clients in the same cluster.

MPSC creation prepares every fallible lease handle before publishing channel metadata. Once /channels/meta/{chan_id} is visible, construction can return a complete ChanManager without a later remote registration leaving a half-created channel.

ChanManager::new_mpmc_subchannel_with_chan_id() is the canonical existing-subchannel path for both producers and consumers. It:

• reuses the parent's registered etcd backend instead of opening one connection per subchannel;
• loads metadata with a bounded timeout;
• validates parent global, member, and payload lease overrides;
• restores the subchannel global-long lease;
• registers real member, global, global-long, and payload keepalive handles.

New MPMC subchannels reuse the parent metadata/global lease, parent member lease, and shared payload lease. The subchannel still owns a separate long lease for its local producer/consumer ID allocator.

The creator and existing-attach branches remain explicit. A creator keeps the MPMC metadata, payload, allocator-cluster, and member leases. A top-level existing attach initially keeps only its member lease; bound sub-MPSC handles add metadata/global and payload contributions. The allocator-cluster lease remains creator-owned in the current implementation.

Shutdown and notification fixes

Rust MQ shutdown now uses one fixed shutdown worker. Concurrent close requests are linearized into a single Framework::shutdown() call, and a separate completion channel publishes its final result. The old per-context task that waited for KV shutdown before closing MQ is removed.

Python endpoint close state is split deliberately:

• the shutdown flag stops new operations;
• local detachment records that teardown has started;
• only _close_done means the complete public close contract has finished.

This keeps close idempotent and retryable without treating an early stop signal as successful teardown.

Tokio state waits now use persistent state as authority and Notify only as a wake-up hint. The shared notify_state helper removes hand-written single-future polling patterns. MQ commit sequencing registers the waiter before rechecking state, responds promptly to an already-closed consumer, and gives shutdown priority when sequence advancement races close.

The Python signal adapter now executes potentially blocking Ctrl-C cleanup on a dedicated listener thread. This avoids running endpoint close from a Python signal frame while the interrupted thread may still own an MQ operation lock.

Consumer handoff and cleanup ordering

MPMC consumer shutdown now closes the child MPSC consumer and confirms its membership cleanup before deleting the ready key.

• If membership deletion succeeds, the ready key is removed so another consumer can claim the subchannel.
• If membership deletion cannot be confirmed, the ready key is retained and both records expire under the same parent member lease.

This prevents a new consumer from observing an available ready route while the previous consumer membership is still visible.

Unpublished subchannels use a stricter internal rollback path. It first runs the canonical endpoint close(), then deletes and verifies all channel-scoped state with bounded retries. This rollback is internal transaction cleanup and does not create a second public shutdown API.

P2P readiness and recovery

Closed-SDK P2P runtime errors now retain their detail and map to the transport-level SendFailed category. RPC retry returns the last concrete transport error instead of replacing it with a generic timeout after all attempts.

External KV initialization now waits for a stable owner route before reporting success. The readiness check:

• treats known connection/readiness failures as transient;
• requires consecutive successful observations;
• has a fixed deadline and polling interval;
• stops promptly during framework shutdown.

External put recovery distinguishes three owner states:

• same generation: refresh the local mapping and wait for the existing route;
• changed generation: wait for owner remap, update the base address, and wait for the new route;
• temporarily absent: wait for membership to reappear instead of treating absence as an unchanged owner.

P2P recovery is bounded to three attempts and preserves the original operation error when route recovery itself times out. Both regular puts and flat-dict pointer puts follow the same contract.

The closed communication SDK binaries, manifest, consumer bindings, and contract surface are updated together so the open Rust facade and packaged runtime agree on the new error behavior.

Large-scale MQ validation

A dedicated bare-local entrypoint now runs large-scale MPMC without testbed or testrunner indirection. It materializes explicit configs and starts:

• etcd;
• GreptimeDB;
• one Fluxon master;
• the requested owner processes;
• one benchmark coordinator;
• one process for every producer and consumer node.

The CI job consumes the packaged Fluxon release and currently runs:

• 2 owners with 1 GiB each;
• 80 producer processes;
• 8 consumer processes;
• 1 worker thread per process;
• a 90-second timed run with a 60-second metric warmup;
• 256-byte values.

The runner owns explicit port allocation, nofile validation, process-group shutdown, resource snapshots, status files, log tails, and large runtime-data cleanup. Text diagnostics are retained after cleanup.

The distributed MPMC benchmark now has explicit lifecycle gates:

1. every node registers and receives its test configuration;
2. consumers establish the required subchannel topology;
3. one coordinator-selected producer prefeeds every ready subchannel;
4. consumers drain and validate prefeed messages;
5. every node reports runtime_ready after its runtime endpoint is prepared;
6. the coordinator releases runtime_start only after all expected nodes are ready;
7. the timed measurement window begins after that release;
8. every node reports a terminal result and waits for the round gate.

Completion metadata now records registered, ready, runtime-ready, reported, and pending node IDs and counts. Missing-node outcomes remain explicit and cannot silently reopen a failed round.

Final large-scale success requires every run to satisfy all lifecycle gates, contain positive total and successful operation counts, and provide an integer total_failed_ops equal to 0. Missing, boolean, malformed, or non-zero failed-operation counts are rejected. A failed validation writes failure.json, prints process log tails, cleans up all managed processes, and exits non-zero without writing a successful summary.

Supporting CI and process-lifecycle changes

The GitHub Actions DAG now packages the Fluxon release once and reuses that artifact in the regular test job and the dedicated large-scale MQ job.

Same-host test infrastructure now supports multiple logical nodes explicitly:

• generated daemonsets accept and validate a preset NODE_ID;
• coordinator and worker ports are assigned per logical node;
• high-count local worker startup is throttled;
• the runtime raises the file-descriptor limit before starting large topologies.

Generated bare-process supervision now allows delayed child creation, observes one stable child without rejecting additional managed children, uses a longer startup deadline, and stops every supervisor identity that may own a service. Successful test runs remove bulky runtime data while preserving the diagnostics needed for review.

Legacy forwarding wrappers and parallel internal surfaces touched by this work are removed in favor of the canonical MQ, lease, P2P, and process-lifecycle entrypoints.

Compatibility and current boundaries

• Valid MPSC and MPMC results with total_failed_ops = 0 remain accepted.
• try_get_data is removed; use get_data(..., try_time=0) for non-blocking consumer behavior.
• User code must consume close() results and must not call private lease-manager, shutdown-controller, or MQ-framework objects.
• Payload lease loss is surfaced as an error and closes the affected producer; the implementation does not silently rebuild a channel or allocate a replacement lease.
• Best-effort distributed-key cleanup may leave keys visible until TTL after a close warning.
• Top-level existing MPMC attach does not by itself keep every shared lease alive; actual contribution begins at the paths documented above.

Validation

Regression coverage includes:

• concurrent MPMC channel publication and stale-writer CAS conflicts;
• lazy producer binding, fair route rotation, and close-during-bind;
• blocked put/get shutdown and retryable local cleanup failures;
• membership-before-ready consumer handoff;
• KV parent/child close races and construction linearization;
• native Fluxon KV lease backend identity and keepalive ownership;
• Tokio notification and commit-sequencer shutdown races;
• P2P error classification, retry, route readiness, owner replacement, and shutdown;
• runtime-ready, prefeed, round-completion, and missing-node benchmark gates;
• strict rejection of non-zero total_failed_ops;
• bare-local port planning, runtime materialization, cleanup, and result validation.

Current checks:

• package-wheel: passed;
• ci-2-virt-node: passed;
• ci-large-scale-mq: passed;
• build-image: passed;
• focused large-scale MQ contract suite: 9 tests passed;
• focused CI contract suite: 31 tests passed;
• Python compilation checks for the modified validation paths passed.

---

中文

建议 PR 标题：

fix(mq): 修复 MPMC 生命周期、P2P 恢复与大规模 MQ 校验

摘要

本 PR 新增独立的大规模 MPMC 验证路径，并修复高并发压测暴露出的 MQ、lease、关闭流程和 P2P 问题。

修改后的契约明确区分三类责任：

• MQ endpoint 生命周期负责本地 runtime 回收和角色专属状态清理。
• Lease handle 只负责贡献 keepalive；分布式 key 的清理权由实际拥有该 key 的 MQ 对象决定。
• Benchmark 成功必须同时证明分布式生命周期已经收敛，并且不存在失败操作。

本 PR 同步更新了持续验证这些契约所需的 test stack、同机多进程编排、closed SDK 产物、用户文档和回归测试。

大规模场景暴露的问题

大规模 MPMC 运行暴露了多类相互独立的问题：

• Channel mapping 已经发布后，existing producer 仍会串行争抢分布式 unique-channel lock。
• Producer 启动时会立即绑定所有 ready 子 MPSC，集中产生大量 handle、etcd 操作和 lease 注册。
• 创建子 MPSC 的耗时可能超过 MPMC create lock 的 TTL。持有旧快照的 writer 随后可能覆盖更新后的 channel list，或者让 channel 与 ready 状态发布不一致。
• Parent MPMC 与 child MPSC 的 lease ownership 不清晰。部分 child 只保存 lease id，没有形成真实 keepalive contribution；部分关闭路径又可能影响其他 member 仍在使用的状态。
• 并发 close、阻塞中的 put/get、lazy bind、commit 排序和 KV teardown 之间可能互相等待，或者错过关闭通知。
• Closed SDK 返回的 P2P 错误会丢失 transport 分类，导致瞬时路由错误无法进入恢复逻辑，或在最终被替换为缺少根因的通用 timeout。
• 所有 MPMC runtime endpoint 尚未就绪时，benchmark 就可能进入计时窗口。
• Large-scale validator 只检查生命周期完成和成功操作数大于零，没有拒绝 total_failed_ops > 0。

MQ 公共生命周期契约

面向用户的关闭顺序现在明确为：

producer.close().unwrap("close MQ producer failed")
# consumer 进程则调用：
consumer.close().unwrap("close MQ consumer failed")

store.close().unwrap("close KV store failed")

同一个 KvClient 上存在多个 MQ endpoint 时，必须先关闭全部 producer / consumer 并消费每个 Result，再关闭 store。

Endpoint close() 现在只有一条规范实现路径，负责：

• 向正在执行的 put/get/bind 发布关闭信号；
• 关闭该 endpoint 拥有的所有子通道；
• 请求并等待 Rust MQ framework 完成 shutdown；
• 停止 watcher、actor、keepalive guard 和本地 handle；
• 对 endpoint 自己拥有的 role、ready、membership、weight key 做一次尽力删除；
• 必需的本地 teardown 失败时返回 ApiError，并保留再次调用 close() 重试的能力。

绑定 lease 的分布式 key 仍采用尽力删除。Delete 超时或 backend 错误只记录 WARN，本地 teardown 继续执行，最终由相应 backend lease TTL 回收。Create、bind、publish、ready claim、消息确认和 offset 更新等运行中操作继续保持强错误契约。

KvClient 新增基于弱引用的 child-close registry。直接 MPSC endpoint 和外层 MPMC endpoint 在进入可能阻塞的构造前注册自己的公共 close()。KvClient.close() 会停止接收新注册，等待构造中的 child 完成，逐个调用并消费全部 endpoint 的关闭结果，最后才关闭 native KV backend。Child 关闭失败时 backend 保持打开，后续仍可重试。

MPMC 拓扑与绑定修复

Existing-channel 快速绑定

new_or_bind_with_unique_key() 在 unique mapping 已经发布时直接读取并绑定，不再获取分布式 create lock。读取失败有明确上限的重试；只有创建 channel 和清理 stale mapping 时才进入 lock 路径。

绑定路径直接返回本次新构造的 endpoint，不再通过进程级全局表执行 get/delete 交接，保证并发调用者各自持有自己的 handle。

Producer 子通道懒绑定

MPMC producer 现在只把 ready 子通道 ID 加入本地调度队列，不在初始化阶段为每个子 MPSC 创建 producer。选中某条路由后才首次绑定，并在后续复用对应的本地 producer。

调度器在本地轮转 ready channel。可能阻塞的 lazy bind 前后都会重新检查 shutdown。如果 close 赢得竞争，新建的子 producer 会立即关闭；清理失败时保留该对象，交给后续 close() 重试。

子通道原子发布与 claim

MPMC create lock 现在只是优化手段，最终一致性由 etcd compare-and-swap 保证。发布时比较构造子 MPSC 前读取到的完整 channel-list 原值：

• Producer 使用 snapshot CAS 发布 channel list；竞争失败时回滚未发布的子通道。
• Consumer 在同一个 etcd transaction 中追加 channel ID 并创建 ready key。
• Stale consumer writer 只在 channel list 单调前进时重试。
• 如果其他 creator 已发布一个尚未 claim 的子通道，竞争失败者会回滚自己的临时 channel，再尝试 claim 已发布的 channel。
• Duplicate ID、非单调快照以及回滚失败都会显式返回错误。

绑定 existing unready 子通道前，consumer 先通过 ready-key CAS 完成 claim，避免两个 consumer 基于旧快照绑定同一个子 MPSC。

Producer membership 发布

Rust MPSC producer 现在用同一个 transaction 发布 membership 和 producer weight。该路径包含：

• 单次 RPC 五秒超时；
• 最多三次尝试；
• 对“请求超时但可能已经提交”的情况做幂等状态核对；
• 发现已有 membership 与预期 lease/value 不一致时返回冲突；
• 最终失败或 shutdown 时同时清理 membership 和 weight。

Metadata 加载前、发布前、重试中以及发布后都会检查 shutdown，避免关闭竞争留下无人管理的新 producer 状态。

Lease ownership 与 channel 构造

Lease 模型现在明确拆分 keepalive contribution 和 cleanup ownership：

• GeneralLease 代表一次真实的本地 keepalive contribution。
• 最后一个本地 guard 释放后，只注销后续 keepalive，不 revoke backend lease，也不删除 shared key。
• 删除 GeneralLease::Borrowed 和 revoke_on_drop 分支。
• 正常关闭只显式删除当前 endpoint 拥有的 key；shared metadata、payload 和 allocator state 等待最后一个 contributor 停止后由 TTL 回收。

Fluxon KV payload lease 的分配和 keepalive 现在全部留在原生 Rust。LeaseBackendUid::KvClient 持有基于真实 Framework 的 async Rust operation，backend identity 同时包含 cluster 和 client instance。这样可以避免 Python callback、blocking pool 往返，以及同一 cluster 下不同 client 误复用已经关闭的 backend。

MPSC 创建会在发布 channel metadata 前完成所有可能失败的 lease handle 准备。/channels/meta/{chan_id} 一旦可见，构造流程就可以返回完整的 ChanManager，不会在后续远程注册失败时遗留半创建 channel。

Producer 和 consumer 的 existing MPMC 子通道统一通过 ChanManager::new_mpmc_subchannel_with_chan_id()：

• 复用 parent 已注册的 etcd backend，避免每个子通道各建一条连接；
• 使用有界超时加载 metadata；
• 校验 parent global、member 和 payload lease override；
• 恢复子通道的 global-long lease；
• 注册真实的 member、global、global-long 和 payload keepalive handle。

新建 MPMC 子通道复用 parent metadata/global lease、parent member lease 和 shared payload lease；子通道仍为自己的 producer/consumer ID allocator 持有独立 long lease。

Creator 与 existing attach 的分支保持显式区分。Creator 持有 MPMC metadata、payload、allocator-cluster 和 member lease；顶层 existing attach 初始只持有自己的 member lease，后续绑定的子 MPSC 再贡献 metadata/global 与 payload keepalive。当前 allocator-cluster lease 仍由 creator 持有。

Shutdown 与通知修复

Rust MQ 现在使用一个固定 shutdown worker。并发 close request 被线性化成一次 Framework::shutdown()，最终结果通过独立 completion channel 发布。旧的“每个 context 等待 KV shutdown 后再关闭 MQ”的 bridge task 已删除。

Python endpoint 的关闭状态被明确拆分：

• shutdown flag 只负责阻止新操作；
• local detach 状态表示 teardown 已开始；
• 只有 _close_done 才表示公共 close() 的全部阶段都已完成。

这样可以在保持 close 幂等的同时，避免把提前发布的停止信号误判为完整关闭，并允许本地清理失败后重试。

Tokio 等待现在以持久状态为权威，Notify 只作为唤醒提示。公共 notify_state helper 替换了手写的单 future poll。MQ commit sequencer 会先注册 waiter，再重新检查状态；consumer 已关闭时立即返回；sequence advance 与 close 竞争时优先处理 shutdown。

Python 信号适配器将可能阻塞的 Ctrl-C 清理放到专用 listener thread 中执行，避免主线程在持有 MQ operation lock 的 PyO3 调用中被 signal frame 打断后直接执行 close。

Consumer handoff 与清理顺序

MPMC consumer 关闭时，先关闭 child MPSC consumer 并确认 membership 清理，再决定是否删除 ready key：

• Membership 删除成功后，删除 ready key，让其他 consumer 可以 claim 该子通道。
• Membership 删除无法确认时，保留 ready key，让两类记录随同一条 parent member lease 一起过期。

这可以避免旧 consumer membership 仍可见时，新 consumer 已经把该 route 当作可用通道接管。

未发布子通道使用更严格的内部 rollback。它先复用规范的 endpoint close()，再通过有界重试删除并验证全部 channel-scoped state。该路径只用于创建事务回滚，不形成第二套公共关闭接口。

P2P 就绪与恢复

Closed SDK 的 P2P runtime error 现在保留原始 detail，并映射为 transport 层 SendFailed。RPC 重试全部失败时返回最后一个真实 transport error，不再统一替换为通用 timeout。

External KV 初始化在返回成功前会等待稳定的 owner route。Readiness 检查会：

• 将已知 connection/readiness 错误视为瞬时错误；
• 要求连续成功观测；
• 使用固定 deadline 和 poll interval；
• framework shutdown 时立即停止。

External put 恢复会区分三种 owner 状态：

• generation 未变化：刷新本地映射并等待原 route；
• generation 已变化：等待 owner remap，更新 base address，再等待新 route；
• owner 暂时不在 membership 中：等待 membership 恢复，不再把 absence 当作原 owner 未变化。

P2P 恢复最多尝试三次；route recovery 自身超时时保留原始 operation error。普通 put 与 flat-dict pointer put 使用相同契约。

Closed communication SDK 二进制、manifest、consumer binding 和 contract surface 同步更新，确保 open Rust facade 与发布包中的 runtime 对新错误契约理解一致。

大规模 MQ 验证

新增独立的 bare-local large-scale MPMC 入口，不经过 testbed 或 testrunner。它显式生成配置并启动：

• etcd；
• GreptimeDB；
• 一个 Fluxon master；
• 指定数量的 owner 进程；
• 一个 benchmark coordinator；
• 每个 producer / consumer node 对应的独立进程。

CI 使用打包后的 Fluxon release，当前规模为：

• 2 个 owner，每个 1 GiB；
• 80 个 producer 进程；
• 8 个 consumer 进程；
• 每进程 1 个 worker thread；
• 90 秒计时运行，其中 60 秒为 metric warmup；
• 256-byte value。

Runner 负责显式端口分配、nofile 校验、进程组关闭、资源快照、状态文件、日志尾部和大体积运行数据清理，并在清理后保留文本诊断。

分布式 MPMC benchmark 现在使用明确的阶段门禁：

1. 所有 node 注册并取得测试配置；
2. Consumer 建立所需的子通道拓扑；
3. Coordinator 指定一个 producer leader，向所有 ready 子通道发送 prefeed；
4. Consumer drain 并验证 prefeed message；
5. 每个 node 在 runtime endpoint 准备完成后上报 runtime_ready；
6. Coordinator 只在全部预期 node 就绪后释放 runtime_start；
7. 收到 release 后才开始计时窗口；
8. 所有 node 上报 terminal result 并等待 round gate。

Completion metadata 现在记录 registered、ready、runtime-ready、reported 和 pending node 的 ID 与数量。缺失 node 会形成明确的失败结果，已经失败的 round 不会被迟到结果重新打开。

Large-scale 最终成功要求每个 run 满足全部生命周期门禁，总操作数和成功操作数均大于零，并且 total_failed_ops 必须是整数 0。缺失、布尔值、类型错误或任何非零失败操作数都会被拒绝。校验失败时写入 failure.json、打印相关进程日志尾部、清理全部受管进程并返回非零退出码，不会生成成功 summary。

CI 与进程生命周期配套修改

GitHub Actions DAG 现在只打包一次 Fluxon release，并由普通测试 job 和独立 large-scale MQ job 复用该 artifact。

同机测试基础设施显式支持多个逻辑节点：

• 生成的 daemonset 接受并校验预设 NODE_ID；
• 为不同逻辑节点分配独立 coordinator / worker 端口；
• 高节点数的本地启动会节流；
• 启动大规模拓扑前提高文件描述符上限。

Bare-process supervisor 现在允许冷启动时延迟创建 child，观察一个稳定 child 时允许存在其他受管 child，扩大 startup deadline，并在 stop 时覆盖所有可能拥有该 service 的 supervisor identity。成功测试会删除大体积 runtime 数据，同时保留 review 所需的诊断文件。

本次修改触及的旧 forwarding wrapper 和平行内部入口已经删除，统一调用规范的 MQ、lease、P2P 和 process-lifecycle 实现。

兼容性与当前边界

• total_failed_ops = 0 的合法 MPSC / MPMC 结果仍会正常通过。
• 删除 try_get_data；非阻塞 consumer 调用统一使用 get_data(..., try_time=0)。
• 用户代码必须消费 close() 的结果，不能操作私有 lease manager、shutdown controller 或 MQ framework。
• Payload lease 丢失会作为错误暴露并关闭对应 producer；实现不会静默重建 channel 或分配替代 lease。
• 分布式 key 尽力删除失败后，相关 key 可能继续可见直到 TTL 到期。
• 顶层 existing MPMC attach 本身不会保持所有 shared lease 存活，实际 contribution 范围以上述注册路径为准。

验证

回归覆盖包括：

• 并发 MPMC channel 发布和 stale-writer CAS 冲突；
• Producer lazy bind、公平 route 轮转和 bind 期间 close；
• 阻塞 put/get 的关闭，以及可重试的本地清理失败；
• Consumer membership-before-ready handoff；
• KV parent/child close 竞争和构造线性化；
• 原生 Fluxon KV lease backend identity 与 keepalive ownership；
• Tokio notification 与 commit sequencer shutdown 竞争；
• P2P 错误分类、重试、route readiness、owner replacement 和 shutdown；
• Runtime-ready、prefeed、round completion 和 missing-node benchmark gate；
• 严格拒绝非零 total_failed_ops；
• Bare-local 端口规划、runtime 生成、清理和结果校验。

当前检查结果：

• package-wheel：通过；
• ci-2-virt-node：通过；
• ci-large-scale-mq：通过；
• build-image：通过；
• Large-scale MQ 定向契约测试：9 项通过；
• CI 定向契约测试：31 项通过；
• 修改涉及的 Python 校验路径通过编译检查。

[ActivePeter]

[P1] 不要在校验前冻结第一次快照。这里在 json.loads 和 completion 校验之前就把 result_cached 设为 true；而 coordinator 使用非原子的 Path.write_text 写结果，所以第一次读取可能只是半截 JSON。我用两次读取（先 {"runs":，再完整 SUCCESS JSON）复现到：本函数返回 SUCCESS，但本地 benchmark_result.json 仍永久保留半截内容，后续 collect/resume 会读到损坏 artifact。请在每次 raw 变化时同步，或只在完整校验通过后标记缓存完成。

[ActivePeter]

[P1] 这里会让公开 close() 永久卡住。上面的 cancel() 明确是 best-effort，异常还会被捕获；如果 gRPC watch 没有因此退出，随后无超时的 join() 就永远不返回（watch thread 本身是 daemon）。请恢复有界 join 并记录仍存活的线程，或先提供能够保证解除 iterator 阻塞的取消契约。

[ActivePeter]

[P1] 这里把 profiler wrapper 伪装成了 channel 所需的 client：它只继承 KvLeaseApi，却依赖私有 _client 并新增大量纯透传方法，随后被传入签名声明为 KvClient 的公开 MQ 路径。这既没有满足 KvClient 的强类型契约，也正是仓库规则禁止的 forwarding/duck-typing compatibility layer。请让 MQ 直接使用唯一的底层 self._store，把这个 wrapper 仅留在需要采集 blocking KV 指标的内部 benchmark 路径。

[ActivePeter]

[P1] 这个新增测试不能按仓库规定的 standalone process 入口执行：从仓库根目录运行 python3 fluxon_test_stack/tests/test_mpmc_readiness_contract.py 会在此处直接报 ModuleNotFoundError: No module named 'fluxon_test_stack'，后面的可选 ImportError/skip 逻辑也到不了。请在第一次包导入前把 repo root 加入 sys.path（与其他直接脚本测试一致），不要依赖外部 PYTHONPATH。

[ActivePeter]

[P2] _discard() 并没有回滚已经创建的子 channel。对带 parent_mpmc_id 的 handle，本 PR 的 _close_local 只删除 membership/weight；同时 lease Drop 已改为不 revoke，因此 CAS loser 已写入的 /channels/meta/<mpsc_id> 和 allocator 状态不会被删除。该 meta 还绑定 shared MPMC global lease，会被其他 member 持续续租，形成不在 mpsc_channels 列表中的长期孤儿。请为未发布 channel 增加完整 rollback，或把创建与发布放进一个可回滚的所有权契约。

[ActivePeter]

[P1] 无锁 fast-bind 仍然经过进程全局的 CHANID_2_NODES 临时槽：chan_bind() 按 (chan_type, chan_id) 写入，_bind_existing_channel() 随后再 get/delete。两个线程在同一进程并发绑定同一个 unique key 时会互相覆盖；其中一个可能取走另一个对象，另一个看到 missing/触发 delete KeyError，并泄漏自己的 handle。原来的分布式锁同时也序列化了这个本地槽。请给该段加进程内同步，或让 chan_bind 直接返回构造出的对象而不经全局 registry。

[ActivePeter]

[P1] 这个 retry deadline 的起点早于 producer 的实际启动。下面 producer 线程会被刻意 defer 到 coordinator START 之后才 thread.start()；初始 READY 屏障如果消耗了大部分 cluster_ready_timeout_seconds，线程启动时这个 deadline 已经过期，第一次可重试的 KV/MQ 初始化错误就会立即变成 deadline exceeded。请在线程真正启动时创建 deadline，或在 START 后重置它。

[ActivePeter]

[P2] 直接在这里赢得 select 会 drop bind_future，但被包装的创建/绑定流程不是 cancellation-safe：create_mpsc_channel 会先 grant/register lease、写 channel meta，再继续多个 await；bind 也会写 membership。shutdown 在这些副作用之后到达时，future 被丢弃且没有 rollback，尤其 override global lease 下的 meta 可能由 MPMC peers 长期续租。请增加 drop/rollback guard，或让构造完成后再显式关闭结果，而不是在任意 await 点取消。

[ActivePeter]

[P1] 这个请求与当前 Ops log contract 不匹配：agent 的 LogReadDirection::Backward 分支要求 cursor，缺失时直接返回 cursor is required for Backward reads；这里没有发送 cursor，因此新增的 collect 路径只会得到 400/error JSON，写不出 workload_log_tail.txt。无 cursor 的 Forward 分支已经实现了“返回最后 max_bytes”语义，请改用它，或先取得文件末尾 cursor 再做 Backward read。

[ActivePeter]

[P1] 缺失的 authority observation 不能落到 ready。get_cluster_info_snapshot() 会捕获 etcd/channel 查询异常并把 total_mpsc_channels、ready_channels、active_consumers 留为 None；当前所有检查都带 is not None，所以三项全缺失时 producer 和 consumer 都会从这里返回 ready=True，在控制面读取失败时反而放开 READY 屏障。请先要求该角色所需字段全部可用，再评估数量；读取失败应保持 waiting 并保留原因。

[ActivePeter]

[P1] 这里的 subchannel close() 会以 delete_membership=False 完成并把 handle 标记为已关闭；随后本函数立即删除 ready key，但 /channels/<id>/consumer/consumer_<n> 仍保留到 member lease TTL。另一个 MPMC consumer 因 ready key 已空会马上 claim/bind，producer 的 consumer watch 此时看到两个 membership，会进入代码中明确的 Invalid consumer binding state，导致该 channel 在 TTL 窗口内无法 put。此前的主动 lease 回收正是为避免 rebind 重叠；现在应在交出 ready key 前显式删除这个 consumer 自己的 membership（并与唤醒阻塞 get 的 shutdown 动作分开）。

[ActivePeter]

[P1] 请在写入 artifact 前过滤凭据。这里把命中的 YAML、日志和 deploy_result 原样复制；本 PR 提交的 codex-failure-context/repository/runner-temp/ci_test_list.ci.yaml 已包含 secret_key 字段。随后 workflow 会上传该目录，Codex prompt 还要求读取其中所有 YAML，因此最终 report 阶段的 redaction 已经太晚，原值早已进入 artifact 和模型输入。请在这个复制边界做结构化脱敏或收紧 allowlist，并覆盖显式加入的 ci_test_list.ci.yaml。

[ActivePeter]

[P1] 这里会把含失败占位的 run 判成成功。coordinator 的 wait_for_completion() 在 consumer 未上报时会插入 failed_operations=1 / forced_missing_consumer_result_timeout 占位，返回 True，随后写出 completion.status=SUCCESS；而当前谓词完全不检查 total_failed_ops 或 aggregated_error_details。我用 total_ops=1008, total_successful_ops=1000, total_failed_ops=8 调用本函数，仍被接受。这样未上报的 consumer 只要之后以 0 退出，large-scale job 就会打印 SUCCESS。请至少拒绝 forced placeholder，通常还应要求 total_failed_ops == 0。

[ActivePeter]

[P2] 请把 MPSC endpoint 的 close() 线性化；consumer 的同名实现也有相同竞态。_closed_local 只保护上面的 detach，但每个调用都会继续执行这里；线程 A 在 native MpscContext.close(&mut self) 里释放 GIL 等待 shutdown 时，线程 B 会看到 _closed_local=True，却仍对同一个 self._ctx 再次调用 close()，从 PyO3 得到 already-mutably-borrowed/cleanup error。仓库已经给 KvClient 和 MPMC close 加了 _close_lock，MPSC 也需要同一契约，并且只让一个调用执行 context 和 membership cleanup。

[ActivePeter]

补充：如果 close() 的契约本身要求支持并发调用，仅在 Python 加锁只能修复外层入口。这里更合适的做法是消除 MpscContext.close() 中的 mq_framework.take()：context 创建时启动一个固定的 Rust shutdown task，由该 task 持有 framework；Python 线性化后的唯一 close 调用只通过 watch/control channel 把状态从 open 切到 close-requested。shutdown task 收到信号后执行 Framework::shutdown()，再通过独立的 completion channel 发布最终成功/失败结果，close() 等到该结果后返回。这样不存在“没有 take 到就误判为已关闭”的分支。控制 channel 不能只有单向 bool 后立即返回，因为公开 close 契约要求 native teardown 完成并返回其 Result，之后调用方才能关闭 KvClient。Python 锁继续覆盖 context close 与 membership cleanup，并仅在全部完成后设置 _close_done。

[ActivePeter]

复核当前 head 的增量；以下是新增且未与已有线程重复的问题。

[ActivePeter]

[P1] 保留 ready key 到 MPMC member lease TTL，仍不能保证 existing sub-consumer 的旧 membership 先消失。该 bind 走 ChanManager::new_with_chan_id，membership 使用另一个本地 MPSC member lease；两个 keepalive 的最后续租时刻独立，因此 ready lease 可能先过期，channel 再次可 claim 时旧 /channels/.../consumer/... 仍可见，重新出现双 consumer 窗口。删除失败路径需要让 ready 与 membership 绑定同一 lease，或持有 ready ownership 直到确认那条 MPSC lease/membership 已失效；等待另一个 lease 的 TTL 不能提供顺序保证。

[ActivePeter]

[P1] 这里在释放 self.lock 后才标记 terminal/event，与 handle_report_results() 的 _upsert_test_result() 存在竞态，而且 missing_consumers 是更早取得的快照。迟到 consumer 若在 placeholder 插入前完成 upsert，当前路径仍会 append 同一 node_id 的第二条占位结果，聚合时会重复计数；若在解锁后到达，又可能在 waiter 生成 summary 时替换占位，结果取决于调度。请在同一临界区内重新计算仍缺失节点、按 node_id upsert 占位并冻结 terminal snapshot，同时明确拒绝或忽略 terminal 后的迟到上报。

[ActivePeter]

[P2] 这个 bridge 只等待 KV shutdown；显式 MpscContext.close() 已完成时，它仍持有 MqFrameworkShutdown 并挂在 KV task registry 里。每个 MPSC endpoint 都会创建一个 context 并注册一条 bridge，因此同一 KvClient 上反复创建/关闭 endpoint 会线性累积休眠 task 和 watch 状态，直到最终 store.close()。请让 bridge 同时观察 MQ completion 后退出（或由一个 worker 统一等待 control/KV 两种关闭信号），使显式 close 后该 registry task 能被 reaper 回收。