Skip to content

Consolidate MuxMetrics mux_dropped instrumentation (DSK-132) #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
daisuke8000 merged 1 commit into from
Jul 5, 2026
Merged

Consolidate MuxMetrics mux_dropped instrumentation (DSK-132) #173

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions crates/rustern-core/src/runtime/forward.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,10 @@ impl RunStats {
self.mux_dropped_lines.fetch_add(1);
}

pub fn mux_dropped_line_count(&self) -> u64 {
self.mux_dropped_lines.load()
}

pub fn snapshot_and_reset(&self) -> RunStatsSnapshot {
RunStatsSnapshot {
active_streams: self.active_streams.load(),
Expand Down Expand Up @@ -195,13 +199,18 @@ impl MuxMetrics {
}

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);
}
}
Comment on lines 201 to 215

@coderabbitai coderabbitai Bot Jul 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 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 -B3

Repository: 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 -B2

Repository: 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 -B3

Repository: 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.rs

Repository: 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)
PY

Repository: 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 -B4

Repository: 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.

}
Expand Down