24944862965

Committed 26 Apr 2026 01:05AM UTC coverage: 85.418% (-0.2%) from 85.647%

Build # 24944862965

Build Type

push

github

Committed by

maxlambrecht

Commit Message

feat(spiffe)!: mark X509SourceError and JwtSourceError as non_exhaustive

Signed-off-by: Max Lambrecht <maxlambrecht@gmail.com>

Coverage Stats

5969 of 6988 relevant lines covered (85.42%)

803.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.53

/spiffe/src/workload_api/supervisor_common.rs

//! Common supervisor utilities shared between X.509 and JWT source supervisors.
//!
//! This module contains reusable components for managing Workload API connections,
//! error tracking, and backoff policies.

use std::time::Duration;
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;

/// Supervisor policy: maximum number of consecutive identical errors before suppressing WARN logs.
///
/// Applies to:
/// - client creation failures
/// - stream connection failures
/// - update validation rejections
pub(crate) const MAX_CONSECUTIVE_SAME_ERROR: u32 = 3;

/// Stream connection phase for diagnostics (distinguishes initial sync from steady-state supervisor).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum StreamPhase {
    /// Initial sync phase during source construction.
    InitialSync,
    /// Steady-state supervisor loop maintaining the stream.
    Supervisor,
}

/// Allocation-free key type for error tracking categories.
///
/// Used by `ErrorTracker` to group errors for log suppression without requiring
/// string literals or heap allocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum ErrorKey {
    /// Client creation failures.
    ClientCreation,
    /// Stream connection failures.
    StreamConnect,
    /// Update rejection failures.
    UpdateRejected,
    /// No identity issued (expected transient state).
    NoIdentityIssued,
}

/// Helper for tracking repeated errors to suppress log noise.
///
/// This tracks consecutive occurrences of the same error kind and suppresses
/// log warnings after the first N consecutive occurrences. For the first N
/// consecutive occurrences of each error kind, logs are emitted at WARN level.
/// After that, logs are downgraded to DEBUG level to reduce noise.
///
/// When a different error kind occurs or errors stop, the counter resets.
pub(crate) struct ErrorTracker {
    last_error_kind: Option<ErrorKey>,
    consecutive_same_error: u32,
    max_consecutive: u32,
}

impl ErrorTracker {
    pub(crate) const fn new(max_consecutive: u32) -> Self {
        Self {
            last_error_kind: None,
            consecutive_same_error: 0,
            max_consecutive,
        }
    }

    pub(crate) fn record_error(&mut self, error_kind: ErrorKey) -> bool {
        let should_warn = self.last_error_kind != Some(error_kind)
            || self.consecutive_same_error < self.max_consecutive;

        if self.last_error_kind == Some(error_kind) {
            self.consecutive_same_error += 1;
        } else {
            self.consecutive_same_error = 1;
            self.last_error_kind = Some(error_kind);
        }

        should_warn
    }

    pub(crate) const fn reset(&mut self) {
        self.consecutive_same_error = 0;
        self.last_error_kind = None;
    }

    pub(crate) const fn consecutive_count(&self) -> u32 {
        self.consecutive_same_error
    }

    pub(crate) const fn last_error_kind(&self) -> Option<ErrorKey> {
        self.last_error_kind
    }
}

pub(crate) async fn sleep_or_cancel(token: &CancellationToken, dur: Duration) -> bool {
    tokio::select! {
        () = token.cancelled() => true,
        () = sleep(dur) => false,
    }
}

/// Exponential backoff with small jitter.
///
/// Computes the next backoff duration by:
/// 1. Doubling the current duration (exponential growth)
/// 2. Clamping to the maximum duration
/// 3. Adding small jitter (0-10% of the base) to prevent synchronized reconnect storms
///
/// The jitter is especially important in container fleets that start simultaneously.
///
/// Note: Jitter is calculated in milliseconds, which may result in sub-millisecond
/// precision loss for very small durations. This is acceptable for backoff purposes.
pub(crate) fn next_backoff(current: Duration, max: Duration) -> Duration {
    let cur = u64::try_from(current.as_millis()).unwrap_or(u64::MAX);
    let max = u64::try_from(max.as_millis()).unwrap_or(u64::MAX);

    let base = (cur.saturating_mul(2)).min(max);
    if base == 0 {
        return Duration::from_millis(0);
    }

    let jitter = base / 10;
    let add = if jitter > 0 {
        fastrand::u64(0..=jitter)
    } else {
        0
    };

    // Subtract jitter range from base before adding random jitter, so the
    // result stays within [base - jitter, base] instead of being clamped to
    // exactly `max` when base == max.
    let jitter_base = base.saturating_sub(jitter);
    Duration::from_millis(jitter_base.saturating_add(add))
}

/// Slower backoff policy for "no identity issued" condition.
///
/// This is an expected transient state (workload may not be registered yet),
/// so we use a gentler backoff: starts at 1s, exponential up to the effective
/// maximum (the lesser of the user-configured `max` and the default 10s cap),
/// with jitter.
pub(crate) fn next_backoff_for_no_identity(current: Duration, max: Duration) -> Duration {
    const MIN_BACKOFF_MS: u64 = 1000; // 1 second
    const DEFAULT_MAX_BACKOFF_MS: u64 = 10000; // 10 seconds

    let max_ms = u64::try_from(max.as_millis()).unwrap_or(u64::MAX);
    let effective_max = max_ms.min(DEFAULT_MAX_BACKOFF_MS);

    let current_with_min = current.max(Duration::from_millis(MIN_BACKOFF_MS));
    next_backoff(current_with_min, Duration::from_millis(effective_max))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn next_backoff_at_max_preserves_jitter() {
        let max = Duration::from_secs(30);
        let lo = max.saturating_sub(max / 10);

        // Run multiple times to exercise the random jitter path.
        for _ in 0..100 {
            let result = next_backoff(max, max);
            // Result should be in [max - 10%, max].
            assert!(
                result >= lo && result <= max,
                "expected backoff in [{lo:?}, {max:?}], got {result:?}"
            );
        }

        // Verify that not all values are identical (jitter is present).
        let mut results = std::collections::HashSet::new();
        for _ in 0..100 {
            results.insert(next_backoff(max, max).as_millis());
        }
        assert!(
            results.len() > 1,
            "expected jitter to produce varying results, got {results:?}"
        );
    }

    #[test]
    fn no_identity_backoff_starts_at_minimum_1s() {
        // Even with a very small current backoff, the minimum is clamped to 1s,
        // then doubled to 2s by next_backoff. With jitter (0-10%), the result
        // lands in [1.8s, 2.0s].
        let result =
            next_backoff_for_no_identity(Duration::from_millis(100), Duration::from_secs(30));
        assert!(
            result >= Duration::from_millis(1800),
            "expected >= 1800ms (2s - 10% jitter), got {}ms",
            result.as_millis()
        );
    }

    #[test]
    fn no_identity_backoff_respects_default_10s_cap() {
        // Even with a high user-configured max, the effective max is 10s.
        let result = next_backoff_for_no_identity(Duration::from_secs(8), Duration::from_secs(60));
        assert!(
            result <= Duration::from_secs(11),
            "expected <= 11s (10s + jitter), got {}ms",
            result.as_millis()
        );
    }

    #[test]
    fn no_identity_backoff_respects_user_max_below_default() {
        // If user-configured max is 3s (below the 10s default), that should be the cap.
        let result = next_backoff_for_no_identity(Duration::from_secs(2), Duration::from_secs(3));
        assert!(
            result <= Duration::from_millis(3300),
            "expected <= 3.3s (3s + jitter), got {}ms",
            result.as_millis()
        );
    }

    #[test]
    fn no_identity_backoff_grows_exponentially() {
        let first = next_backoff_for_no_identity(Duration::from_secs(1), Duration::from_secs(30));
        let second = next_backoff_for_no_identity(first, Duration::from_secs(30));

        // Second backoff should be larger than first (exponential growth)
        assert!(
            second > first,
            "expected growth: first={}ms, second={}ms",
            first.as_millis(),
            second.as_millis()
        );
    }
}

1	//! Common supervisor utilities shared between X.509 and JWT source supervisors.
2	//!
3	//! This module contains reusable components for managing Workload API connections,
4	//! error tracking, and backoff policies.
5
6	use std::time::Duration;
7	use tokio::time::sleep;
8	use tokio_util::sync::CancellationToken;
9
10	/// Supervisor policy: maximum number of consecutive identical errors before suppressing WARN logs.
11	///
12	/// Applies to:
13	/// - client creation failures
14	/// - stream connection failures
15	/// - update validation rejections
16	pub(crate) const MAX_CONSECUTIVE_SAME_ERROR: u32 = 3;
17
18	/// Stream connection phase for diagnostics (distinguishes initial sync from steady-state supervisor).
19	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20	pub(crate) enum StreamPhase {
21	/// Initial sync phase during source construction.
22	InitialSync,
23	/// Steady-state supervisor loop maintaining the stream.
24	Supervisor,
25	}
26
27	/// Allocation-free key type for error tracking categories.
28	///
29	/// Used by `ErrorTracker` to group errors for log suppression without requiring
30	/// string literals or heap allocation.
31	#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
32	pub(crate) enum ErrorKey {
33	/// Client creation failures.
34	ClientCreation,
35	/// Stream connection failures.
36	StreamConnect,
37	/// Update rejection failures.
38	UpdateRejected,
39	/// No identity issued (expected transient state).
40	NoIdentityIssued,
41	}
42
43	/// Helper for tracking repeated errors to suppress log noise.
44	///
45	/// This tracks consecutive occurrences of the same error kind and suppresses
46	/// log warnings after the first N consecutive occurrences. For the first N
47	/// consecutive occurrences of each error kind, logs are emitted at WARN level.
48	/// After that, logs are downgraded to DEBUG level to reduce noise.
49	///
50	/// When a different error kind occurs or errors stop, the counter resets.
51	pub(crate) struct ErrorTracker {
52	last_error_kind: Option<ErrorKey>,
53	consecutive_same_error: u32,
54	max_consecutive: u32,
55	}
56
57	impl ErrorTracker {
58	pub(crate) const fn new(max_consecutive: u32) -> Self {	107✔
59	Self {	107✔
60	last_error_kind: None,	107✔
61	consecutive_same_error: 0,	107✔
62	max_consecutive,	107✔
63	}	107✔
64	}	107✔
65
66	pub(crate) fn record_error(&mut self, error_kind: ErrorKey) -> bool {	×
67	let should_warn = self.last_error_kind != Some(error_kind)	×
68	\|\| self.consecutive_same_error < self.max_consecutive;	×
69
70	if self.last_error_kind == Some(error_kind) {	×
71	self.consecutive_same_error += 1;	×
72	} else {	×
73	self.consecutive_same_error = 1;	×
74	self.last_error_kind = Some(error_kind);	×
75	}	×
76
77	should_warn	×
78	}	×
79
80	pub(crate) const fn reset(&mut self) {	72✔
81	self.consecutive_same_error = 0;	72✔
82	self.last_error_kind = None;	72✔
83	}	72✔
84
85	pub(crate) const fn consecutive_count(&self) -> u32 {	19✔
86	self.consecutive_same_error	19✔
87	}	19✔
88
89	pub(crate) const fn last_error_kind(&self) -> Option<ErrorKey> {	216✔
90	self.last_error_kind	216✔
91	}	216✔
92	}
93
94	pub(crate) async fn sleep_or_cancel(token: &CancellationToken, dur: Duration) -> bool {	×
95	tokio::select! {	×
96	() = token.cancelled() => true,	×
97	() = sleep(dur) => false,	×
98	}
99	}	×
100
101	/// Exponential backoff with small jitter.
102	///
103	/// Computes the next backoff duration by:
104	/// 1. Doubling the current duration (exponential growth)
105	/// 2. Clamping to the maximum duration
106	/// 3. Adding small jitter (0-10% of the base) to prevent synchronized reconnect storms
107	///
108	/// The jitter is especially important in container fleets that start simultaneously.
109	///
110	/// Note: Jitter is calculated in milliseconds, which may result in sub-millisecond
111	/// precision loss for very small durations. This is acceptable for backoff purposes.
112	pub(crate) fn next_backoff(current: Duration, max: Duration) -> Duration {	1,230✔
113	let cur = u64::try_from(current.as_millis()).unwrap_or(u64::MAX);	1,230✔
114	let max = u64::try_from(max.as_millis()).unwrap_or(u64::MAX);	1,230✔
115
116	let base = (cur.saturating_mul(2)).min(max);	1,230✔
117	if base == 0 {	1,230✔
118	return Duration::from_millis(0);	×
119	}	1,230✔
120
121	let jitter = base / 10;	1,230✔
122	let add = if jitter > 0 {	1,230✔
123	fastrand::u64(0..=jitter)	1,230✔
124	} else {
125	0	×
126	};
127
128	// Subtract jitter range from base before adding random jitter, so the
129	// result stays within [base - jitter, base] instead of being clamped to
130	// exactly `max` when base == max.
131	let jitter_base = base.saturating_sub(jitter);	1,230✔
132	Duration::from_millis(jitter_base.saturating_add(add))	1,230✔
133	}	1,230✔
134
135	/// Slower backoff policy for "no identity issued" condition.
136	///
137	/// This is an expected transient state (workload may not be registered yet),
138	/// so we use a gentler backoff: starts at 1s, exponential up to the effective
139	/// maximum (the lesser of the user-configured `max` and the default 10s cap),
140	/// with jitter.
141	pub(crate) fn next_backoff_for_no_identity(current: Duration, max: Duration) -> Duration {	30✔
142	const MIN_BACKOFF_MS: u64 = 1000; // 1 second
143	const DEFAULT_MAX_BACKOFF_MS: u64 = 10000; // 10 seconds
144
145	let max_ms = u64::try_from(max.as_millis()).unwrap_or(u64::MAX);	30✔
146	let effective_max = max_ms.min(DEFAULT_MAX_BACKOFF_MS);	30✔
147
148	let current_with_min = current.max(Duration::from_millis(MIN_BACKOFF_MS));	30✔
149	next_backoff(current_with_min, Duration::from_millis(effective_max))	30✔
150	}	30✔
151
152	#[cfg(test)]
153	mod tests {
154	use super::*;
155
156	#[test]
157	fn next_backoff_at_max_preserves_jitter() {	6✔
158	let max = Duration::from_secs(30);	6✔
159	let lo = max.saturating_sub(max / 10);	6✔
160
161	// Run multiple times to exercise the random jitter path.
162	for _ in 0..100 {	6✔
163	let result = next_backoff(max, max);	600✔
164	// Result should be in [max - 10%, max].
165	assert!(	600✔
166	result >= lo && result <= max,	600✔
167	"expected backoff in [{lo:?}, {max:?}], got {result:?}"
168	);
169	}
170
171	// Verify that not all values are identical (jitter is present).
172	let mut results = std::collections::HashSet::new();	6✔
173	for _ in 0..100 {	600✔
174	results.insert(next_backoff(max, max).as_millis());	600✔
175	}	600✔
176	assert!(	6✔
177	results.len() > 1,	6✔
178	"expected jitter to produce varying results, got {results:?}"
179	);
180	}	6✔
181
182	#[test]
183	fn no_identity_backoff_starts_at_minimum_1s() {	6✔
184	// Even with a very small current backoff, the minimum is clamped to 1s,
185	// then doubled to 2s by next_backoff. With jitter (0-10%), the result
186	// lands in [1.8s, 2.0s].
187	let result =	6✔
188	next_backoff_for_no_identity(Duration::from_millis(100), Duration::from_secs(30));	6✔
189	assert!(	6✔
190	result >= Duration::from_millis(1800),	6✔
191	"expected >= 1800ms (2s - 10% jitter), got {}ms",
192	result.as_millis()	×
193	);
194	}	6✔
195
196	#[test]
197	fn no_identity_backoff_respects_default_10s_cap() {	6✔
198	// Even with a high user-configured max, the effective max is 10s.
199	let result = next_backoff_for_no_identity(Duration::from_secs(8), Duration::from_secs(60));	6✔
200	assert!(	6✔
201	result <= Duration::from_secs(11),	6✔
202	"expected <= 11s (10s + jitter), got {}ms",
203	result.as_millis()	×
204	);
205	}	6✔
206
207	#[test]
208	fn no_identity_backoff_respects_user_max_below_default() {	6✔
209	// If user-configured max is 3s (below the 10s default), that should be the cap.
210	let result = next_backoff_for_no_identity(Duration::from_secs(2), Duration::from_secs(3));	6✔
211	assert!(	6✔
212	result <= Duration::from_millis(3300),	6✔
213	"expected <= 3.3s (3s + jitter), got {}ms",
214	result.as_millis()	×
215	);
216	}	6✔
217
218	#[test]
219	fn no_identity_backoff_grows_exponentially() {	6✔
220	let first = next_backoff_for_no_identity(Duration::from_secs(1), Duration::from_secs(30));	6✔
221	let second = next_backoff_for_no_identity(first, Duration::from_secs(30));	6✔
222
223	// Second backoff should be larger than first (exponential growth)
224	assert!(	6✔
225	second > first,	6✔
226	"expected growth: first={}ms, second={}ms",
227	first.as_millis(),	×
228	second.as_millis()	×
229	);
230	}	6✔
231	}

maxlambrecht / rust-spiffe / 24944862965

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous