1use std::collections::HashMap;
2use std::future::Future;
3use std::sync::atomic::{
4 AtomicBool,
5 Ordering,
6};
7use std::sync::Arc;
8use std::time::Duration;
9
10use bytes::Bytes;
11use chrono::{
12 DateTime,
13 Utc,
14};
15use reqwest::header::{
16 HeaderMap,
17 CONTENT_RANGE,
18};
19use serde::de::DeserializeOwned;
20use serde::{
21 Deserialize,
22 Serialize,
23};
24use tokio::sync::mpsc;
25
26use crate::error::{
27 PluginError,
28 PluginResult,
29};
30use crate::http::{
31 check_response,
32 map_send_error,
33};
34use crate::types::PipelineStatus;
35
36pub const LOG_CHUNK_STALL_TIMEOUT: Duration = Duration::from_secs(60);
37
38pub async fn next_log_chunk(response: &mut reqwest::Response) -> PluginResult<Option<Bytes>> {
39 match tokio::time::timeout(LOG_CHUNK_STALL_TIMEOUT, response.chunk()).await {
40 Err(_) => Err(PluginError::NetworkError(
41 "Log download stalled for 60 seconds".into(),
42 )),
43 Ok(chunk) => chunk.map_err(|e| map_send_error(e, response.url())),
44 }
45}
46
47pub const CONTENT_HASH_SEED: u64 = 0xcbf2_9ce4_8422_2325;
48
49pub fn content_hash(bytes: &[u8]) -> u64 {
50 extend_content_hash(CONTENT_HASH_SEED, bytes)
51}
52
53pub fn extend_content_hash(hash: u64, bytes: &[u8]) -> u64 {
54 bytes.iter().fold(hash, |hash, &byte| {
55 (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3)
56 })
57}
58
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
60#[serde(rename_all = "lowercase")]
61pub enum LogNodeKind {
62 Stage,
63 Job,
64 Step,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
68#[serde(rename_all = "lowercase")]
69pub enum LogAvailability {
70 Ready,
71 Partial,
72 Pending,
73}
74
75#[derive(Debug, Clone, Serialize, Deserialize)]
76pub struct LogNode {
77 pub id: String,
78 pub parent_id: Option<String>,
79 pub kind: LogNodeKind,
80 pub name: String,
81 pub status: PipelineStatus,
82 pub started_at: Option<DateTime<Utc>>,
83 pub finished_at: Option<DateTime<Utc>>,
84 pub log: Option<LogAvailability>,
85 pub web_url: Option<String>,
86}
87
88pub struct RunRef<'a> {
89 pub pipeline_id: &'a str,
90 pub run_number: i64,
91 pub run_id: &'a str,
92 pub metadata: &'a HashMap<String, serde_json::Value>,
93}
94
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96#[serde(transparent)]
97pub struct LogCursor(serde_json::Value);
98
99impl LogCursor {
100 pub fn new<T: Serialize>(position: &T) -> PluginResult<Self> {
101 serde_json::to_value(position)
102 .map(Self)
103 .map_err(|error| PluginError::SerializationError(error.to_string()))
104 }
105
106 pub fn parse<T: DeserializeOwned>(&self) -> Option<T> {
107 serde_json::from_value(self.0.clone()).ok()
108 }
109}
110
111#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
112#[serde(rename_all = "lowercase")]
113pub enum LogReadMode {
114 Append,
115 Replace,
116}
117
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct LogRead {
120 pub mode: LogReadMode,
121 pub cursor: Option<LogCursor>,
122}
123
124impl LogRead {
125 pub fn replace(cursor: Option<LogCursor>) -> Self {
126 Self {
127 mode: LogReadMode::Replace,
128 cursor,
129 }
130 }
131
132 pub fn append(cursor: LogCursor) -> Self {
133 Self {
134 mode: LogReadMode::Append,
135 cursor: Some(cursor),
136 }
137 }
138}
139
140#[derive(Debug, Clone, Copy, PartialEq, Eq)]
141pub enum LineMarker {
142 Plain,
143 GroupStart,
144 GroupEnd,
145 SectionStart { open: bool },
146 OpenSection,
147 Error,
148 Warning,
149 Notice,
150 Debug,
151 Command,
152}
153
154#[derive(Debug, Clone, Copy, PartialEq, Eq)]
155pub struct ParsedLine<'a> {
156 pub timestamp: Option<DateTime<Utc>>,
157 pub marker: LineMarker,
158 pub body: &'a [u8],
159}
160
161pub trait LogDialect: Send + Sync {
162 fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a>;
163
164 fn is_step_start(&self, _line: &ParsedLine<'_>) -> bool {
165 false
166 }
167
168 fn infers_errors(&self) -> bool {
169 true
170 }
171}
172
173pub struct PlainDialect;
174
175impl LogDialect for PlainDialect {
176 fn parse<'a>(&self, line: &'a [u8]) -> ParsedLine<'a> {
177 ParsedLine {
178 timestamp: None,
179 marker: LineMarker::Plain,
180 body: line,
181 }
182 }
183}
184
185#[derive(Debug, Clone, Copy, PartialEq, Eq)]
186pub struct ContentRange {
187 pub span: Option<(u64, u64)>,
188 pub total: Option<u64>,
189}
190
191impl ContentRange {
192 pub fn parse(headers: &HeaderMap) -> Option<Self> {
193 let value = headers
194 .get(CONTENT_RANGE)?
195 .to_str()
196 .ok()?
197 .strip_prefix("bytes ")?;
198 let (span, total) = value.split_once('/')?;
199 let span = match span.trim() {
200 "*" => None,
201 span => {
202 let (start, end) = span.split_once('-')?;
203 Some((start.parse().ok()?, end.parse().ok()?))
204 }
205 };
206 let total = match total.trim() {
207 "*" => None,
208 total => Some(total.parse().ok()?),
209 };
210 Some(Self { span, total })
211 }
212}
213
214#[derive(Debug, Clone, Copy, PartialEq, Eq)]
215pub enum RangedRead {
216 Appended(u64),
217 Replaced(u64),
218 Refetch,
219}
220
221pub struct LogSink {
222 tx: mpsc::Sender<Bytes>,
223 bytes: u64,
224 restarted: Arc<AtomicBool>,
225}
226
227impl LogSink {
228 pub fn new(tx: mpsc::Sender<Bytes>) -> Self {
229 Self {
230 tx,
231 bytes: 0,
232 restarted: Arc::new(AtomicBool::new(false)),
233 }
234 }
235
236 pub fn restart_signal(&self) -> Arc<AtomicBool> {
237 Arc::clone(&self.restarted)
238 }
239
240 pub fn restart(&mut self) -> PluginResult<()> {
241 if self.bytes > 0 {
242 return Err(PluginError::Internal(
243 "A log read can only restart before it writes".into(),
244 ));
245 }
246 self.restarted.store(true, Ordering::SeqCst);
247 Ok(())
248 }
249
250 pub fn restarted(&self) -> bool {
251 self.restarted.load(Ordering::SeqCst)
252 }
253
254 pub fn bytes_written(&self) -> u64 {
255 self.bytes
256 }
257
258 pub async fn write(&mut self, chunk: Bytes) -> PluginResult<()> {
259 if chunk.is_empty() {
260 return Ok(());
261 }
262 self.bytes += chunk.len() as u64;
263 self.tx
264 .send(chunk)
265 .await
266 .map_err(|_| PluginError::Internal("Log reader stopped".into()))
267 }
268
269 pub async fn pipe(&mut self, response: reqwest::Response) -> PluginResult<()> {
270 self.pipe_after(response, 0).await
271 }
272
273 pub async fn pipe_after(
274 &mut self, mut response: reqwest::Response, mut skip: u64,
275 ) -> PluginResult<()> {
276 while let Some(chunk) = next_log_chunk(&mut response).await? {
277 let cut = usize::try_from(skip).map_or(chunk.len(), |s| s.min(chunk.len()));
278 skip -= cut as u64;
279 self.write(chunk.slice(cut..)).await?;
280 }
281 Ok(())
282 }
283
284 pub async fn pipe_range(
285 &mut self, response: reqwest::Response, offset: u64,
286 ) -> PluginResult<RangedRead> {
287 let range = ContentRange::parse(response.headers());
288 match response.status().as_u16() {
289 206 if range
290 .and_then(|r| r.span)
291 .is_some_and(|(start, _)| start == offset) =>
292 {
293 self.pipe(response).await?;
294 Ok(RangedRead::Appended(offset + self.bytes))
295 }
296 416 if range.is_some_and(|r| r.span.is_none() && r.total == Some(offset)) => {
297 Ok(RangedRead::Appended(offset))
298 }
299 200 => {
300 self.restart()?;
301 self.pipe(response).await?;
302 Ok(RangedRead::Replaced(self.bytes))
303 }
304 206 | 416 => Ok(RangedRead::Refetch),
305 _ => {
306 check_response(response).await?;
307 Ok(RangedRead::Refetch)
308 }
309 }
310 }
311
312 pub async fn pipe_resumed<F, Fut>(
313 &mut self, offset: Option<u64>, mut fetch: F,
314 ) -> PluginResult<(LogReadMode, u64)>
315 where
316 F: FnMut(Option<u64>) -> Fut,
317 Fut: Future<Output = PluginResult<reqwest::Response>>,
318 {
319 if let Some(offset) = offset {
320 match self.pipe_range(fetch(Some(offset)).await?, offset).await? {
321 RangedRead::Appended(next) => return Ok((LogReadMode::Append, next)),
322 RangedRead::Replaced(next) => return Ok((LogReadMode::Replace, next)),
323 RangedRead::Refetch => self.restart()?,
324 }
325 }
326 self.pipe(check_response(fetch(None).await?).await?).await?;
327 Ok((LogReadMode::Replace, self.bytes))
328 }
329}
330
331#[cfg(test)]
332mod tests {
333 use std::io::{
334 Read,
335 Write,
336 };
337 use std::net::TcpListener;
338
339 use super::*;
340
341 async fn serve(raw: &'static str) -> reqwest::Response {
342 let _ = rustls::crypto::ring::default_provider().install_default();
343 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
344 let addr = listener.local_addr().unwrap();
345 std::thread::spawn(move || {
346 if let Ok((mut stream, _)) = listener.accept() {
347 let mut buf = [0u8; 4096];
348 let _ = stream.read(&mut buf);
349 let _ = stream.write_all(raw.as_bytes());
350 }
351 });
352 reqwest::Client::new()
353 .get(format!("http://{addr}/log"))
354 .send()
355 .await
356 .unwrap()
357 }
358
359 async fn drain(mut rx: mpsc::Receiver<Bytes>) -> Vec<u8> {
360 let mut got = Vec::new();
361 while let Some(chunk) = rx.recv().await {
362 got.extend_from_slice(&chunk);
363 }
364 got
365 }
366
367 async fn ranged(
368 raw: &'static str, offset: u64,
369 ) -> (PluginResult<RangedRead>, Vec<u8>, bool, u64) {
370 let response = serve(raw).await;
371 let (tx, rx) = mpsc::channel(16);
372 let mut sink = LogSink::new(tx);
373 let outcome = sink.pipe_range(response, offset).await;
374 let restarted = sink.restarted();
375 let written = sink.bytes_written();
376 drop(sink);
377 (outcome, drain(rx).await, restarted, written)
378 }
379
380 #[tokio::test]
381 async fn pipe_delivers_chunked_bytes_exactly() {
382 let response = serve(
383 "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
384 5\r\nhello\r\n7\r\n world\n\r\n3\r\nend\r\n0\r\n\r\n",
385 )
386 .await;
387 let (tx, rx) = mpsc::channel(16);
388 let mut sink = LogSink::new(tx);
389 sink.pipe(response).await.unwrap();
390 assert_eq!(sink.bytes_written(), 15);
391 drop(sink);
392 assert_eq!(drain(rx).await, b"hello world\nend");
393 }
394
395 #[tokio::test]
396 async fn pipe_after_drops_the_overlap_across_chunks() {
397 let response = serve(
398 "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
399 2\r\nab\r\n3\r\ncde\r\n1\r\nf\r\n0\r\n\r\n",
400 )
401 .await;
402 let (tx, rx) = mpsc::channel(16);
403 let mut sink = LogSink::new(tx);
404 sink.pipe_after(response, 4).await.unwrap();
405 assert_eq!(sink.bytes_written(), 2);
406 drop(sink);
407 assert_eq!(drain(rx).await, b"ef");
408 }
409
410 #[test]
411 fn content_hash_is_fnv1a_64_and_continues_across_chunks() {
412 assert_eq!(content_hash(b""), 0xcbf2_9ce4_8422_2325);
413 assert_eq!(content_hash(b"a"), 0xaf63_dc4c_8601_ec8c);
414 assert_eq!(content_hash(b"foobar"), 0x8594_4171_f739_67e8);
415 assert_eq!(
416 extend_content_hash(content_hash(b"foo"), b"bar"),
417 content_hash(b"foobar")
418 );
419 }
420
421 #[test]
422 fn content_range_reads_spans_totals_and_unknowns() {
423 let parse = |value: &str| {
424 let mut headers = HeaderMap::new();
425 headers.insert(CONTENT_RANGE, value.parse().unwrap());
426 ContentRange::parse(&headers)
427 };
428 assert_eq!(
429 parse("bytes 5-9/10"),
430 Some(ContentRange {
431 span: Some((5, 9)),
432 total: Some(10)
433 })
434 );
435 assert_eq!(
436 parse("bytes 5-9/*"),
437 Some(ContentRange {
438 span: Some((5, 9)),
439 total: None
440 })
441 );
442 assert_eq!(
443 parse("bytes */10"),
444 Some(ContentRange {
445 span: None,
446 total: Some(10)
447 })
448 );
449 for malformed in ["items 5-9/10", "bytes 5-9", "bytes a-9/10", "bytes 5-/10"] {
450 assert_eq!(parse(malformed), None, "{malformed}");
451 }
452 assert_eq!(ContentRange::parse(&HeaderMap::new()), None);
453 }
454
455 #[tokio::test]
456 async fn range_reply_starting_at_the_offset_appends() {
457 let (outcome, got, restarted, _) = ranged(
458 "HTTP/1.1 206 Partial Content\r\nContent-Range: bytes 5-9/10\r\nContent-Length: 5\r\n\r\nworld",
459 5,
460 )
461 .await;
462 assert_eq!(outcome.unwrap(), RangedRead::Appended(10));
463 assert_eq!(got, b"world");
464 assert!(!restarted);
465 }
466
467 #[tokio::test]
468 async fn range_reply_starting_elsewhere_or_unmarked_is_refetched_unwritten() {
469 for raw in [
470 "HTTP/1.1 206 Partial Content\r\nContent-Range: bytes 6-9/10\r\nContent-Length: 4\r\n\r\norld",
471 "HTTP/1.1 206 Partial Content\r\nContent-Length: 5\r\n\r\nworld",
472 "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */3\r\nContent-Length: 0\r\n\r\n",
473 "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Length: 0\r\n\r\n",
474 "HTTP/1.1 204 No Content\r\n\r\n",
475 ] {
476 let (outcome, got, restarted, written) = ranged(raw, 5).await;
477 assert_eq!(outcome.unwrap(), RangedRead::Refetch, "{raw}");
478 assert!(got.is_empty() && !restarted && written == 0, "{raw}");
479 }
480 }
481
482 #[tokio::test]
483 async fn range_not_satisfiable_at_the_exact_total_appends_nothing() {
484 let (outcome, got, restarted, _) = ranged(
485 "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */5\r\nContent-Length: 0\r\n\r\n",
486 5,
487 )
488 .await;
489 assert_eq!(outcome.unwrap(), RangedRead::Appended(5));
490 assert!(got.is_empty() && !restarted);
491 }
492
493 #[tokio::test]
494 async fn full_body_after_a_range_restarts_before_writing() {
495 let (outcome, got, restarted, _) =
496 ranged("HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nhelloworld", 5).await;
497 assert_eq!(outcome.unwrap(), RangedRead::Replaced(10));
498 assert_eq!(got, b"helloworld");
499 assert!(restarted);
500 }
501
502 #[tokio::test]
503 async fn range_reply_with_a_server_error_fails() {
504 let (outcome, got, restarted, _) = ranged(
505 "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
506 5,
507 )
508 .await;
509 assert!(matches!(
510 outcome,
511 Err(PluginError::ServiceUnavailable { .. })
512 ));
513 assert!(got.is_empty() && !restarted);
514 }
515
516 #[tokio::test]
517 async fn write_after_reader_dropped_is_internal() {
518 let (tx, rx) = mpsc::channel(1);
519 drop(rx);
520 let mut sink = LogSink::new(tx);
521 let err = sink.write(Bytes::from_static(b"x")).await.unwrap_err();
522 assert!(matches!(err, PluginError::Internal(_)));
523 }
524}