Consolidate MuxMetrics mux_dropped instrumentation (DSK-132)#173
Merged
Conversation
Record mux drops to RunStats or the local counter, not both; mux_drop_count reads from the active source. Co-authored-by: Cursor <cursoragent@cursor.com>
📝 WalkthroughWalkthroughRunStats に mux tier のドロップ行数を返す public な mux_dropped_line_count() メソッドが追加されました。MuxMetrics の mux_drop_count() と record_mux_drop() は、stats フィールドの有無に応じて処理を分岐し、stats が存在する場合は RunStats 側のカウンタを利用・更新し、存在しない場合のみ従来の内部 AtomicU64 (mux_dropped) を使用するよう変更されました。 Changes
Sequence Diagram(s)本変更は分岐ロジックの追加であり、シーケンス図よりも上記フローチャートで十分説明可能です（省略）。 Possibly related issues
重大度メモ: 本変更はカウンタ管理のロジック整理のみで、非同期処理・K8s・セキュリティへの影響は見られません。ただし 🚥 Pre-merge checks | ❌ 2❌ Failed checks (2 warnings)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/rustern-core/src/runtime/forward.rs`:
- Around line 201-215: The stats-enabled path in mux_drop_count() is reading
from the same MuxMetrics state that stderr_reporter() clears via
snapshot_and_reset(), so the count can lose accumulated drops across intervals.
Update the forward::ForwardRuntime readback so mux_drop_count() uses an
independent counter when stats is Some(...), and keep record_mux_drop()
incrementing that separate read path instead of relying on the reporter-owned
mux_dropped_lines state. Use the existing mux_drop_count(), record_mux_drop(),
and MuxMetrics symbols to locate and split the reporting counter from the
persistent readback counter.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 62249526-2770-470f-b61b-d370aeb3dd95
📒 Files selected for processing (1)
crates/rustern-core/src/runtime/forward.rs
Comment on lines
201
to
215
| pub fn mux_drop_count(&self) -> u64 { | ||
| self.mux_dropped.load(Ordering::Relaxed) | ||
| if let Some(stats) = &self.stats { | ||
| stats.mux_dropped_line_count() | ||
| } else { | ||
| self.mux_dropped.load(Ordering::Relaxed) | ||
| } | ||
| } | ||
|
|
||
| pub fn record_mux_drop(&self) { | ||
| self.mux_dropped.fetch_add(1, Ordering::Relaxed); | ||
| if let Some(stats) = &self.stats { | ||
| stats.record_mux_dropped_line(); | ||
| } else { | ||
| self.mux_dropped.fetch_add(1, Ordering::Relaxed); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# stderr_reporter がテスト実行中に spawn されるか、interval が短いテストケースが存在するか確認
rg -n 'stderr_reporter|RuntimeStatsConfig' crates/rustern-core --type=rust -A3 -B3Repository: daisuke8000/rustern
Length of output: 5713
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Run a focused read-only inspection of the stats implementation and its call sites.
sed -n '1,260p' crates/rustern-core/src/runtime/forward.rs
printf '\n---- mux_forward_core ----\n'
sed -n '1,180p' crates/rustern-core/src/runtime/mux_forward_core.rs
printf '\n---- load_multistream excerpt ----\n'
sed -n '360,410p' crates/rustern-core/tests/load_multistream.rs
printf '\n---- usages of mux_drop_count / record_mux_drop / snapshot_and_reset ----\n'
rg -n 'mux_drop_count|record_mux_drop|snapshot_and_reset|mux_dropped_line_count|mux_dropped_lines' crates/rustern-core/src crates/rustern-core/tests -A2 -B2Repository: daisuke8000/rustern
Length of output: 21570
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# どのタイミングで mux_drop_count() を読むか、load test の流れを確認する。
sed -n '330,365p' crates/rustern-core/tests/load_multistream.rs
printf '\n----\n'
sed -n '410,490p' crates/rustern-core/tests/load_multistream.rs
printf '\n---- mux_metrics tests/usages ----\n'
rg -n 'mux_drop_count\(\)|record_mux_drop\(' crates/rustern-core -A3 -B3Repository: daisuke8000/rustern
Length of output: 6872
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# load_multistream と bench で stats.reporter が有効なまま mux_drop_count() を読む箇所を確認する。
sed -n '1,120p' crates/rustern-core/tests/load_multistream.rs
printf '\n---- bench excerpt ----\n'
sed -n '430,475p' crates/rustern-core/benches/mux_forward.rs
printf '\n---- stats config defaults / docs ----\n'
sed -n '1,120p' crates/rustern-core/src/runtime/config.rsRepository: daisuke8000/rustern
Length of output: 10401
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Run a read-only probe to see whether the reporter can reset the counter before the final read.
python3 - <<'PY'
from pathlib import Path
import re
text = Path('crates/rustern-core/tests/load_multistream.rs').read_text()
m = re.search(r'duration\s*=\s*.*?LOSSY_OBSERVE|LOSSY_OBSERVE', text)
print("LOSSY_OBSERVE present:", bool(m))
for pat in [r'LOSSY_OBSERVE\s*:\s*Duration::from_secs\((\d+)\)', r'TEST_HARD_LIMIT\s*:\s*Duration::from_secs\((\d+)\)', r'load_scale\(\)']:
mm = re.search(pat, text)
print(pat, "=>", mm.group(1) if mm else None)
PYRepository: daisuke8000/rustern
Length of output: 400
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# stats reporter の有無と、bench で mux_drop_count() を読む前に interval が回る条件があるか確認する。
rg -n 'RuntimeStatsConfig\s*\{|stats:\s*Some\(|stats:\s*None\(|stderr_reporter\(' crates/rustern-core/tests crates/rustern-core/benches crates/rustern-core/src -A4 -B4Repository: daisuke8000/rustern
Length of output: 5744
stats 有効時は mux_drop_count() を独立カウンタで読めるようにするべきです
stderr_reporter() が同じ mux_dropped_lines を snapshot_and_reset() で swap(0) するため、stats: Some(...) の経路では interval を跨いだ後の mux_drop_count() が累積値を欠く可能性があります。MuxMetrics 側の readback は reporter から分離してください。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/rustern-core/src/runtime/forward.rs` around lines 201 - 215, The
stats-enabled path in mux_drop_count() is reading from the same MuxMetrics state
that stderr_reporter() clears via snapshot_and_reset(), so the count can lose
accumulated drops across intervals. Update the forward::ForwardRuntime readback
so mux_drop_count() uses an independent counter when stats is Some(...), and
keep record_mux_drop() incrementing that separate read path instead of relying
on the reporter-owned mux_dropped_lines state. Use the existing
mux_drop_count(), record_mux_drop(), and MuxMetrics symbols to locate and split
the reporting counter from the persistent readback counter.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
record_mux_dropwrites toRunStatswhen present, otherwise the local atomic (bench/test path).mux_drop_countreads from the same single source.Test plan
cargo test -p rustern-core --lib runtime::forwardMade with Cursor
Summary by CodeRabbit