|
| 1 | +use std::time::{Duration, Instant}; |
| 2 | +use tailflow_core::{ |
| 3 | + config::RestartPolicy, |
| 4 | + ingestion::{process::ProcessSource, Source}, |
| 5 | + new_bus, |
| 6 | +}; |
| 7 | + |
| 8 | +/// Collect up to `limit` records from `rx` within `timeout`. |
| 9 | +async fn collect( |
| 10 | + mut rx: tailflow_core::LogReceiver, |
| 11 | + limit: usize, |
| 12 | + timeout: Duration, |
| 13 | +) -> Vec<String> { |
| 14 | + let deadline = Instant::now() + timeout; |
| 15 | + let mut payloads = Vec::new(); |
| 16 | + while payloads.len() < limit && Instant::now() < deadline { |
| 17 | + let remaining = deadline.saturating_duration_since(Instant::now()); |
| 18 | + match tokio::time::timeout(remaining, rx.recv()).await { |
| 19 | + Ok(Ok(r)) => payloads.push(r.payload), |
| 20 | + _ => break, |
| 21 | + } |
| 22 | + } |
| 23 | + payloads |
| 24 | +} |
| 25 | + |
| 26 | +// ── RestartPolicy::Never ────────────────────────────────────────────────────── |
| 27 | + |
| 28 | +#[tokio::test] |
| 29 | +async fn never_policy_does_not_restart_on_failure() { |
| 30 | + let (tx, rx) = new_bus(); |
| 31 | + let src = |
| 32 | + ProcessSource::new("test", "echo hello && exit 1").with_restart(RestartPolicy::Never, 50); |
| 33 | + |
| 34 | + tokio::spawn(async move { Box::new(src).run(tx).await }); |
| 35 | + |
| 36 | + let payloads = collect(rx, 10, Duration::from_secs(3)).await; |
| 37 | + // Should get exactly one "hello", no restart synthetic record. |
| 38 | + assert_eq!(payloads.iter().filter(|p| p.as_str() == "hello").count(), 1); |
| 39 | + assert!( |
| 40 | + !payloads.iter().any(|p| p.contains("[tailflow]")), |
| 41 | + "unexpected restart record with Never policy" |
| 42 | + ); |
| 43 | +} |
| 44 | + |
| 45 | +// ── RestartPolicy::OnFailure ────────────────────────────────────────────────── |
| 46 | + |
| 47 | +#[tokio::test] |
| 48 | +async fn on_failure_restarts_after_non_zero_exit() { |
| 49 | + let (tx, rx) = new_bus(); |
| 50 | + // Exits non-zero — should trigger a restart synthetic record. |
| 51 | + let src = ProcessSource::new("test", "exit 1").with_restart(RestartPolicy::OnFailure, 50); |
| 52 | + |
| 53 | + tokio::spawn(async move { Box::new(src).run(tx).await }); |
| 54 | + |
| 55 | + let payloads = collect(rx, 5, Duration::from_secs(3)).await; |
| 56 | + assert!( |
| 57 | + payloads |
| 58 | + .iter() |
| 59 | + .any(|p| p.contains("[tailflow]") && p.contains("restarting")), |
| 60 | + "expected restart record after non-zero exit, got: {payloads:?}" |
| 61 | + ); |
| 62 | +} |
| 63 | + |
| 64 | +#[tokio::test] |
| 65 | +async fn on_failure_does_not_restart_after_clean_exit() { |
| 66 | + let (tx, rx) = new_bus(); |
| 67 | + // Exits zero — should NOT restart. |
| 68 | + let src = ProcessSource::new("test", "echo done").with_restart(RestartPolicy::OnFailure, 50); |
| 69 | + |
| 70 | + tokio::spawn(async move { Box::new(src).run(tx).await }); |
| 71 | + |
| 72 | + let payloads = collect(rx, 10, Duration::from_secs(2)).await; |
| 73 | + assert!( |
| 74 | + !payloads.iter().any(|p| p.contains("[tailflow]")), |
| 75 | + "unexpected restart record after clean exit" |
| 76 | + ); |
| 77 | +} |
| 78 | + |
| 79 | +// ── RestartPolicy::Always ───────────────────────────────────────────────────── |
| 80 | + |
| 81 | +#[tokio::test] |
| 82 | +async fn always_policy_restarts_after_clean_exit() { |
| 83 | + let (tx, rx) = new_bus(); |
| 84 | + // Exits zero — Always policy should still restart. |
| 85 | + let src = ProcessSource::new("test", "echo hi").with_restart(RestartPolicy::Always, 50); |
| 86 | + |
| 87 | + tokio::spawn(async move { Box::new(src).run(tx).await }); |
| 88 | + |
| 89 | + let payloads = collect(rx, 10, Duration::from_secs(3)).await; |
| 90 | + assert!( |
| 91 | + payloads |
| 92 | + .iter() |
| 93 | + .any(|p| p.contains("[tailflow]") && p.contains("restarting")), |
| 94 | + "expected restart record with Always policy, got: {payloads:?}" |
| 95 | + ); |
| 96 | +} |
| 97 | + |
| 98 | +// ── Backoff ─────────────────────────────────────────────────────────────────── |
| 99 | + |
| 100 | +#[tokio::test] |
| 101 | +async fn restart_delay_is_respected() { |
| 102 | + let (tx, rx) = new_bus(); |
| 103 | + // 200 ms initial delay so we can measure it. |
| 104 | + let src = ProcessSource::new("test", "exit 1").with_restart(RestartPolicy::OnFailure, 200); |
| 105 | + |
| 106 | + tokio::spawn(async move { Box::new(src).run(tx).await }); |
| 107 | + |
| 108 | + let start = Instant::now(); |
| 109 | + // Wait for the first restart synthetic record. |
| 110 | + let payloads = collect(rx, 2, Duration::from_secs(3)).await; |
| 111 | + let elapsed = start.elapsed(); |
| 112 | + |
| 113 | + assert!( |
| 114 | + payloads.iter().any(|p| p.contains("[tailflow]")), |
| 115 | + "expected restart record" |
| 116 | + ); |
| 117 | + // At least the configured delay elapsed before the record appeared. |
| 118 | + assert!( |
| 119 | + elapsed >= Duration::from_millis(150), |
| 120 | + "restart fired too fast ({elapsed:?})" |
| 121 | + ); |
| 122 | +} |
0 commit comments