@@ -4,6 +4,9 @@
const MAX_BUFFER_LINES: usize = 100;
const MAX_RETRIES: usize = 2;
/// Upper bound on flush passes in `drain()` so a permanently-unreachable
/// server can't hang job completion forever.
const MAX_DRAIN_PASSES: usize = 8;
#[derive(Clone)]
pub struct LogReporter {
@@ -37,7 +40,10 @@
}
}
/// Flush buffered lines to the server. On a failed send the lines are
/// re-queued at the FRONT of the buffer (never dropped), so a later flush —
/// Flush buffered lines to the server.
/// from the periodic timer or `drain()` — retries them. This is what keeps
/// a failing job's error tail from vanishing (anvil-cli#27).
pub async fn flush(&self) {
let lines = {
let mut buf = self.buffer.lock().await;
@@ -47,10 +53,29 @@
std::mem::take(&mut *buf)
};
if !self.send_lines(&lines).await {
// Put the un-sent lines back ahead of anything appended meanwhile,
// preserving overall order.
let mut buf = self.buffer.lock().await;
buf.splice(0..0, lines);
}
}
/// Flush repeatedly until the buffer is empty (or `MAX_DRAIN_PASSES` is
self.send_lines(&lines).await;
/// reached). Call at job completion — after aborting the flush timer — so
/// the job's final output is delivered before the job is marked terminal.
pub async fn drain(&self) {
for _ in 0..MAX_DRAIN_PASSES {
let empty = self.buffer.lock().await.is_empty();
if empty {
return;
}
self.flush().await;
}
}
async fn send_lines(&self, lines: &[String]) {
/// Send one batch, returning whether the server accepted it.
async fn send_lines(&self, lines: &[String]) -> bool {
for attempt in 0..=MAX_RETRIES {
let result = self
.client
@@ -62,14 +87,14 @@
.await;
match result {
Ok(resp) if resp.status().is_success() => return,
Ok(resp) if resp.status().is_success() => return true,
Ok(resp) => {
if attempt < MAX_RETRIES {
eprintln!("log upload failed (status {}), retrying...", resp.status());
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
} else {
eprintln!(
"log upload failed after {} retries, dropping lines",
"log upload failed after {} retries, will re-queue",
MAX_RETRIES
);
}
@@ -79,11 +104,15 @@
eprintln!("log upload error: {e}, retrying...");
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
} else {
eprintln!("log upload error after {} retries: {e}", MAX_RETRIES);
eprintln!(
"log upload error after {} retries, will re-queue: {e}",
MAX_RETRIES
);
}
}
}
}
false
}
/// Start a background flush timer. Cancel by aborting the returned handle.
@@ -96,5 +125,94 @@
reporter.flush().await;
}
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use wiremock::matchers::method;
use wiremock::{Mock, MockServer, ResponseTemplate};
// Collect every log line the mock server actually received, across all
// requests (successful or not).
async fn received_lines(server: &MockServer) -> Vec<String> {
let reqs = server.received_requests().await.unwrap();
let mut out = Vec::new();
for r in reqs {
let v: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
if let Some(arr) = v["lines"].as_array() {
for l in arr {
out.push(l.as_str().unwrap().to_string());
}
}
}
out
}
// The whole point of the CI log pipeline: a transient server failure must
// NOT lose captured log lines — this is the bug that hid job errors
// (anvil-cli#27). The server rejects the first full flush (all internal
// retries), then recovers; every line must still be delivered afterward.
#[tokio::test]
async fn does_not_drop_lines_when_a_flush_fails_then_server_recovers() {
let server = MockServer::start().await;
// Fail every attempt of the first flush (MAX_RETRIES + 1 = 3), then 200.
Mock::given(method("POST"))
.respond_with(ResponseTemplate::new(503))
.up_to_n_times((MAX_RETRIES + 1) as u64)
.with_priority(1)
.mount(&server)
.await;
Mock::given(method("POST"))
.respond_with(ResponseTemplate::new(200))
.with_priority(2)
.mount(&server)
.await;
let reporter = LogReporter::new(&server.uri(), "job1", "tok");
reporter.append("line-A\nline-B").await;
// First flush exhausts its retries against the 503s. A correct reporter
// retains the lines; a broken one drops them here.
reporter.flush().await;
// Deliver whatever remains now that the server is healthy.
reporter.drain().await;
let got = received_lines(&server).await;
assert!(
got.iter().filter(|l| *l == "line-A").count() >= 1,
"line-A was never successfully delivered; got {got:?}"
);
assert!(
got.iter().filter(|l| *l == "line-B").count() >= 1,
"line-B was never successfully delivered; got {got:?}"
);
// A delivery attempt must have happened AFTER the 503 window — proof the
// lines were re-queued rather than dropped.
let reqs = server.received_requests().await.unwrap();
assert!(
reqs.len() > (MAX_RETRIES + 1),
"no send after the failure window (lines were dropped): {} requests",
reqs.len()
);
}
// drain() must deliver the buffered tail even if the periodic timer never
// ran — this is the job's final output (error block + status lines).
#[tokio::test]
async fn drain_delivers_buffered_tail() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.respond_with(ResponseTemplate::new(200))
.mount(&server)
.await;
let reporter = LogReporter::new(&server.uri(), "job2", "tok");
reporter.append("error: could not compile").await;
reporter.drain().await;
let got = received_lines(&server).await;
assert_eq!(got, vec!["error: could not compile".to_string()]);
}
}