Skip to content

test(session-net): isolate connection metrics assertions #692

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

Draft
VerifiedOrganic wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

test(session-net): isolate connection metrics assertions #692

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9c41765
test(session-net): reproduce global metrics race
VerifiedOrganic Aug 15, 2026
537f4c5
test(session-net): isolate connection metrics accounting
VerifiedOrganic Aug 15, 2026
7a62722
chore: merge current main into issue 576 lane
VerifiedOrganic Aug 16, 2026
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
112 changes: 62 additions & 50 deletions crates/opc-session-net/src/consensus.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,13 @@ fn map_protocol_error(error: &ProtocolError) -> SessionConsensusPeerError {
}

fn record_consensus_server_connection_failure(error: &ProtocolError) {
#[cfg(test)]
if matches!(
error,
ProtocolError::Io(error) if error.kind() == io::ErrorKind::TimedOut
) {
crate::test_support::record_connection_timeout_failure();
}
match error {
ProtocolError::Io(error) if error.kind() == io::ErrorKind::TimedOut => {
&METRICS.session_net_connection_failure_timeout
Expand Down Expand Up @@ -274,6 +281,8 @@ fn record_consensus_server_connection_outcome(result: &Result<(), ProtocolError>
METRICS
.session_net_connection_successes
.fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
crate::test_support::record_connection_success();
}
Err(error) => record_consensus_server_connection_failure(error),
}
Expand Down Expand Up @@ -3407,9 +3416,6 @@ mod tests {

#[tokio::test(start_paused = true)]
async fn claimant_records_explicit_retirement_before_dropping_stale_ready_connection() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;
let (_server_binding, client_binding) = bindings();
let control = SessionReauthenticationControl::new();
let resolver: RemoteAddrResolver =
Expand Down Expand Up @@ -3470,9 +3476,6 @@ mod tests {

#[tokio::test(start_paused = true)]
async fn claimant_records_material_retirement_before_dropping_stale_ready_connection() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;
let (_server_binding, client_binding) = bindings();
let material = crate::test_support::RotatableClientMaterial::new(
"spiffe://test-domain/tenant/test/ns/default/sa/session/nf/smf/instance/1",
Expand Down Expand Up @@ -3750,9 +3753,6 @@ mod tests {

#[tokio::test(start_paused = true)]
async fn consensus_soft_timeout_classifies_pending_connect_without_abandoning() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;
let (_server_binding, client_binding) = bindings();
let resolver: RemoteAddrResolver =
Arc::new(|| Box::pin(std::future::pending::<io::Result<SocketAddr>>()));
Expand Down Expand Up @@ -3798,9 +3798,6 @@ mod tests {

#[tokio::test(start_paused = true)]
async fn configured_ceiling_also_reserves_post_connect_rpc_time() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;
let (_server_binding, client_binding) = bindings();
let resolver: RemoteAddrResolver =
Arc::new(|| Box::pin(std::future::pending::<io::Result<SocketAddr>>()));
Expand Down Expand Up @@ -3950,42 +3947,50 @@ mod tests {
bytes
}

#[derive(Clone, Copy)]
struct ConnectionOutcomeMetricSnapshot {
idle_retirements: u64,
timeout_failures: u64,
successes: u64,
drain_started: u64,
drain_completed: u64,
}

fn connection_outcome_metrics() -> ConnectionOutcomeMetricSnapshot {
ConnectionOutcomeMetricSnapshot {
idle_retirements: METRICS
.session_net_lifecycle_retirement_idle_timeout
.load(Ordering::Relaxed),
timeout_failures: METRICS
.session_net_connection_failure_timeout
.load(Ordering::Relaxed),
successes: METRICS
.session_net_connection_successes
.load(Ordering::Relaxed),
drain_started: METRICS
.session_net_lifecycle_drain_started
.load(Ordering::Relaxed),
drain_completed: METRICS
.session_net_lifecycle_drain_completed
.load(Ordering::Relaxed),
}
fn connection_outcome_metrics() -> crate::test_support::ConnectionOutcomeMetricSnapshot {
crate::test_support::CONNECTION_OUTCOME_TEST_ACCOUNTING
.try_with(|accounting| accounting.snapshot())
.expect("connection outcome accounting scope")
}

fn record_test_idle_retirement() {
let lifecycle = ConnectionLifecycle::new(
test_consensus_lifecycle_policy(),
tokio::time::Instant::now(),
None,
None,
0,
None,
)
.expect("test connection lifecycle");
lifecycle.record_forced_retirement(RetirementReason::IdleTimeout);
}

#[tokio::test]
async fn connection_outcome_delta_isolated_from_writer_outside_test_scope() {
let accounting = Arc::new(crate::test_support::ConnectionOutcomeTestAccounting::default());
crate::test_support::CONNECTION_OUTCOME_TEST_ACCOUNTING
.scope(accounting, async {
let before = connection_outcome_metrics();

tokio::spawn(async { record_test_idle_retirement() })
.await
.expect("outside metric writer");
record_test_idle_retirement();

let after = connection_outcome_metrics();
assert_eq!(after.idle_retirements, before.idle_retirements + 1);
assert_eq!(after.timeout_failures, before.timeout_failures);
assert_eq!(after.successes, before.successes);
assert_eq!(after.drain_started, before.drain_started + 1);
assert_eq!(after.drain_completed, before.drain_completed + 1);
})
.await;
}

async fn wait_for_drain_completion(minimum: u64) {
tokio::time::timeout(Duration::from_secs(1), async {
while METRICS
.session_net_lifecycle_drain_completed
.load(Ordering::Relaxed)
< minimum
{
while connection_outcome_metrics().drain_completed < minimum {
tokio::task::yield_now().await;
}
})
Expand Down Expand Up @@ -4029,12 +4034,7 @@ mod tests {
(result, writer)
}

#[tokio::test]
async fn consensus_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;

async fn assert_consensus_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let before_idle = connection_outcome_metrics();
let (idle_result, acknowledgement) = dispatch_after_authentication(&[]).await;
record_consensus_server_connection_outcome(&idle_result);
Expand Down Expand Up @@ -4112,6 +4112,18 @@ mod tests {
assert!(after_handshake.timeout_failures > before_handshake.timeout_failures);
}

#[tokio::test]
async fn consensus_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let accounting = Arc::new(crate::test_support::ConnectionOutcomeTestAccounting::default());
crate::test_support::CONNECTION_OUTCOME_TEST_ACCOUNTING
.scope(
accounting,
assert_consensus_server_distinguishes_authenticated_idle_from_active_frame_timeout(
),
)
.await;
}

#[tokio::test]
async fn consensus_pre_hello_generation_retirement_emits_one_no_dispatch_control() {
let (server_binding, _client_binding) = bindings();
Expand Down
17 changes: 17 additions & 0 deletions crates/opc-session-net/src/lifecycle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -659,6 +659,9 @@ const UNRECORDED_RETIREMENT_REASON: u8 = u8::MAX;
struct LifecycleConnectionMetrics {
state: AtomicU8,
hard_overrun_recorded: AtomicBool,
// Retain test attribution when lifecycle ownership moves between tasks.
#[cfg(test)]
test_accounting: Option<Arc<crate::test_support::ConnectionOutcomeTestAccounting>>,
}

impl LifecycleConnectionMetrics {
Expand All @@ -669,6 +672,8 @@ impl LifecycleConnectionMetrics {
Self {
state: AtomicU8::new(LIFECYCLE_METRIC_ACTIVE),
hard_overrun_recorded: AtomicBool::new(false),
#[cfg(test)]
test_accounting: crate::test_support::current_connection_outcome_test_accounting(),
}
}

Expand All @@ -692,6 +697,10 @@ impl LifecycleConnectionMetrics {
METRICS
.session_net_lifecycle_drain_started
.fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
if let Some(accounting) = &self.test_accounting {
accounting.record_drain_started();
}
}

fn record_hard_overrun(&self) {
Expand All @@ -712,6 +721,10 @@ impl Drop for LifecycleConnectionMetrics {
METRICS
.session_net_lifecycle_drain_completed
.fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
if let Some(accounting) = &self.test_accounting {
accounting.record_drain_completed();
}
}
_ => decrement_gauge(&METRICS.session_net_lifecycle_active_connections),
}
Expand Down Expand Up @@ -1017,6 +1030,10 @@ impl ConnectionLifecycle {
.store(reason as u8, Ordering::Release);
self.metrics.begin_draining();
reason.retirement_counter().fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
if let Some(accounting) = &self.metrics.test_accounting {
accounting.record_retirement(reason);
}
tracing::debug!(reason = reason.as_str(), "session connection retired");
}

Expand Down
64 changes: 26 additions & 38 deletions crates/opc-session-net/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,13 @@ fn connection_failure_reason(error: &ProtocolError) -> &'static str {
}

fn record_server_connection_failure(error: &ProtocolError) {
#[cfg(test)]
if matches!(
error,
ProtocolError::Io(error) if error.kind() == std::io::ErrorKind::TimedOut
) {
crate::test_support::record_connection_timeout_failure();
}
match error {
ProtocolError::Io(error) if error.kind() == std::io::ErrorKind::TimedOut => {
&METRICS.session_net_connection_failure_timeout
Expand All @@ -263,6 +270,8 @@ fn record_server_connection_outcome(result: &Result<(), ProtocolError>) {
METRICS
.session_net_connection_successes
.fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
crate::test_support::record_connection_success();
}
Err(error) => record_server_connection_failure(error),
}
Expand Down Expand Up @@ -3927,42 +3936,15 @@ mod tests {
bytes
}

#[derive(Clone, Copy)]
struct ConnectionOutcomeMetricSnapshot {
idle_retirements: u64,
timeout_failures: u64,
successes: u64,
drain_started: u64,
drain_completed: u64,
}

fn connection_outcome_metrics() -> ConnectionOutcomeMetricSnapshot {
ConnectionOutcomeMetricSnapshot {
idle_retirements: METRICS
.session_net_lifecycle_retirement_idle_timeout
.load(Ordering::Relaxed),
timeout_failures: METRICS
.session_net_connection_failure_timeout
.load(Ordering::Relaxed),
successes: METRICS
.session_net_connection_successes
.load(Ordering::Relaxed),
drain_started: METRICS
.session_net_lifecycle_drain_started
.load(Ordering::Relaxed),
drain_completed: METRICS
.session_net_lifecycle_drain_completed
.load(Ordering::Relaxed),
}
fn connection_outcome_metrics() -> crate::test_support::ConnectionOutcomeMetricSnapshot {
crate::test_support::CONNECTION_OUTCOME_TEST_ACCOUNTING
.try_with(|accounting| accounting.snapshot())
.expect("connection outcome accounting scope")
}

async fn wait_for_drain_completion(minimum: u64) {
tokio::time::timeout(Duration::from_secs(1), async {
while METRICS
.session_net_lifecycle_drain_completed
.load(Ordering::Relaxed)
< minimum
{
while connection_outcome_metrics().drain_completed < minimum {
tokio::task::yield_now().await;
}
})
Expand Down Expand Up @@ -3998,12 +3980,7 @@ mod tests {
(result, writer)
}

#[tokio::test]
async fn generic_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let _guard = crate::test_support::SESSION_CONNECTION_METRICS_TEST_LOCK
.lock()
.await;

async fn assert_generic_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let before_idle = connection_outcome_metrics();
let (idle_result, acknowledgement) = dispatch_after_authentication(&[]).await;
record_server_connection_outcome(&idle_result);
Expand Down Expand Up @@ -4072,6 +4049,17 @@ mod tests {
assert!(after_handshake.timeout_failures > before_handshake.timeout_failures);
}

#[tokio::test]
async fn generic_server_distinguishes_authenticated_idle_from_active_frame_timeout() {
let accounting = Arc::new(crate::test_support::ConnectionOutcomeTestAccounting::default());
crate::test_support::CONNECTION_OUTCOME_TEST_ACCOUNTING
.scope(
accounting,
assert_generic_server_distinguishes_authenticated_idle_from_active_frame_timeout(),
)
.await;
}

#[tokio::test]
async fn pre_hello_generation_retirement_emits_one_explicit_no_dispatch_control() {
let reauthentication = SessionReauthenticationControl::new();
Expand Down
Loading