test largescale mq#34
Conversation
| ) | ||
| if not result_cached and (not result_path.exists() or result_path.read_text(encoding="utf-8") != raw): | ||
| result_path.write_text(raw, encoding="utf-8") | ||
| result_cached = True |
There was a problem hiding this comment.
[P1] 不要在校验前冻结第一次快照。这里在 json.loads 和 completion 校验之前就把 result_cached 设为 true；而 coordinator 使用非原子的 Path.write_text 写结果，所以第一次读取可能只是半截 JSON。我用两次读取（先 {"runs":，再完整 SUCCESS JSON）复现到：本函数返回 SUCCESS，但本地 benchmark_result.json 仍永久保留半截内容，后续 collect/resume 会读到损坏 artifact。请在每次 raw 变化时同步，或只在完整校验通过后标记缓存完成。
| watch_thread.join(timeout=2) | ||
| if watch_thread.is_alive(): | ||
| logging.warning(f"MPMC channel {self.mpmc_id} watch thread did not stop before timeout") | ||
| watch_thread.join() |
There was a problem hiding this comment.
[P1] 这里会让公开 close() 永久卡住。上面的 cancel() 明确是 best-effort，异常还会被捕获；如果 gRPC watch 没有因此退出，随后无超时的 join() 就永远不返回（watch thread 本身是 daemon）。请恢复有界 join 并记录仍存活的线程，或先提供能够保证解除 iterator 阻塞的取消契约。
|
|
||
|
|
||
| class FluxonBlockingStore: | ||
| class FluxonBlockingStore(KvLeaseApi): |
There was a problem hiding this comment.
[P1] 这里把 profiler wrapper 伪装成了 channel 所需的 client：它只继承 KvLeaseApi，却依赖私有 _client 并新增大量纯透传方法，随后被传入签名声明为 KvClient 的公开 MQ 路径。这既没有满足 KvClient 的强类型契约，也正是仓库规则禁止的 forwarding/duck-typing compatibility layer。请让 MQ 直接使用唯一的底层 self._store，把这个 wrapper 仅留在需要采集 blocking KV 指标的内部 benchmark 路径。
| from types import SimpleNamespace | ||
| from unittest import mock | ||
|
|
||
| from fluxon_test_stack.mpmc_readiness import evaluate_mpmc_topology_ready |
There was a problem hiding this comment.
[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。
| reason: str, | ||
| ) -> None: | ||
| try: | ||
| close_result = channel._discard() |
There was a problem hiding this comment.
[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，或把创建与发布放进一个可回滚的所有权契约。
| final_error: Optional[ApiError] = None | ||
| max_attempts = 3 | ||
| for attempt_idx in range(max_attempts): | ||
| fast_bind_res = _try_bind_existing_channel_without_lock() |
There was a problem hiding this comment.
[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。
|
|
||
| round_state = PreparedMPMCRound() | ||
| self._prepared_mpmc_round = round_state | ||
| prepare_retry_deadline_ts = time.monotonic() + cluster_ready_timeout_s |
There was a problem hiding this comment.
[P1] 这个 retry deadline 的起点早于 producer 的实际启动。下面 producer 线程会被刻意 defer 到 coordinator START 之后才 thread.start()；初始 READY 屏障如果消耗了大部分 cluster_ready_timeout_seconds，线程启动时这个 deadline 已经过期，第一次可重试的 KV/MQ 初始化错误就会立即变成 deadline exceeded。请在线程真正启动时创建 deadline，或在 START 后重置它。
| tokio::pin!(bind_future); | ||
| tokio::select! { | ||
| biased; | ||
| _ = shutdown.wait_closed() => { |
There was a problem hiding this comment.
[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 点取消。
| "instance_key": instance_key, | ||
| "kind": kind, | ||
| "name": name, | ||
| "direction": "Backward", |
There was a problem hiding this comment.
[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。
| reason=f"unsupported MPMC role: {role!r}", | ||
| ) | ||
|
|
||
| return MPMCTopologyReadiness(ready=True, reason="ready") |
There was a problem hiding this comment.
[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 并保留原因。
| try: | ||
| if self.mpsc_consumer is not None: | ||
| self.mpsc_consumer.request_shutdown() | ||
| self.mpsc_consumer.close().unwrap() |
There was a problem hiding this comment.
[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
temporarily deployed
to
OPENAI_API_KEY
GitHub Actions
Inactive
1 participant
No description provided.