cortex/
helpers.rs

1// Copyright 2015-2025 Deyan Ginev. See the LICENSE
2// file at the top-level directory of this distribution.
3//
4// Licensed under the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>.
5// This file may not be copied, modified, or distributed
6// except according to those terms.
7
8//! Helper structures and methods for Task
9use rand::RngExt;
10use regex::Regex;
11use std::fs::File;
12use std::io;
13use std::io::Read;
14use std::path::{Path, PathBuf};
15use std::str;
16use std::sync::LazyLock;
17
18use diesel::pg::PgConnection;
19use diesel::result::Error;
20
21use crate::concerns::CortexInsertable;
22use crate::models::{
23  LogError, LogFatal, LogInfo, LogInvalid, LogRecord, LogWarning, NewLogError, NewLogFatal,
24  NewLogInfo, NewLogInvalid, NewLogWarning, Task,
25};
26
27static MESSAGE_LINE_REGEX: LazyLock<Regex> =
28  LazyLock::new(|| Regex::new(r"^([^ :]+):([^ :]+):([^ ]+)(\s(.*))?$").unwrap());
29/// "(Loading... file" message regex
30pub static LOADING_LINE_REGEX: LazyLock<Regex> = LazyLock::new(|| {
31  Regex::new(r"^\((?:Loading|Processing definitions)\s(.+/)?([^/]+[^.])\.\.\.(\s|$)").unwrap()
32});
33/// The short document name within an entry path: everything between the last `/` and the last `.`.
34static ENTRY_DOCUMENT_NAME_REGEX: LazyLock<Regex> =
35  LazyLock::new(|| Regex::new(r"^.+/(.+)\..+$").unwrap());
36
37/// The short document name shown in reports and used for download filenames — an entry path's
38/// basename without its directory or extension (e.g. `/data/…/0811.0417/0811.0417.zip` →
39/// `0811.0417`). Falls back to the trimmed entry when the path doesn't match the `…/name.ext`
40/// shape.
41pub fn entry_document_name(entry: &str) -> String {
42  let trimmed = entry.trim_end();
43  ENTRY_DOCUMENT_NAME_REGEX.replace(trimmed, "$1").to_string()
44}
45
46#[derive(Clone, Copy, PartialEq, Eq, Debug)]
47/// An enumeration of the expected task statuses. A small value type (every variant is a unit or a
48/// single `i32`) that round-trips through `i32` via [`TaskStatus::raw`]/[`TaskStatus::from_raw`],
49/// so it is `Copy`.
50pub enum TaskStatus {
51  /// currently queued for processing
52  TODO,
53  /// everything went smoothly
54  NoProblem,
55  /// minor issues
56  Warning,
57  /// major issues
58  Error,
59  /// critical/panic issues
60  Fatal,
61  /// invalid task, fatal + discard from statistics
62  Invalid,
63  /// currently blocked by dependencies
64  Blocked(i32),
65  /// currently being processed (marker identifies batch)
66  Queued(i32),
67}
68
69#[derive(Clone, Debug)]
70/// In-progress task, with dispatch metadata
71pub struct TaskProgress {
72  /// the `Task` struct being tracked
73  pub task: Task,
74  /// time of entering the job queue / first dispatch
75  pub created_at: i64,
76  /// number of dispatch retries
77  pub retries: i64,
78  /// the effective lease / visibility timeout (seconds) for THIS task, captured at dispatch time
79  /// from the owning service's `lease_timeout_seconds` override or the global dispatcher default
80  /// (D-17). Stored per-task so a config change never re-times an already-leased task, and so one
81  /// dispatcher can serve fast (latexml-oxide) and slow (Perl) services with different leases.
82  pub lease_timeout_seconds: i64,
83}
84impl TaskProgress {
85  /// What is the latest admissible time for this task to be completed? The base deadline is this
86  /// task's captured lease / visibility timeout (`lease_timeout_seconds`); each retry extends it by
87  /// another full timeout (`(retries + 1) × timeout`), so a task that keeps timing out backs off
88  /// rather than re-leasing ever-faster.
89  pub fn expected_at(&self) -> i64 {
90    self.created_at + ((self.retries + 1) * self.lease_timeout_seconds)
91  }
92}
93
94#[derive(Clone, Debug)]
95/// Completed task, with its processing status and report messages
96pub struct TaskReport {
97  /// the `Task` we are reporting on
98  pub task: Task,
99  /// the reported processing status
100  pub status: TaskStatus,
101  /// a vector of `TaskMessage` log entries
102  pub messages: Vec<NewTaskMessage>,
103}
104
105#[derive(Clone, Debug)]
106/// Enum for all types of reported messages for a given Task, as per the `LaTeXML` convention
107/// One of "invalid", "fatal", "error", "warning" or "info"
108pub enum TaskMessage {
109  /// Debug/low-priroity messages
110  Info(LogInfo),
111  /// Soft/resumable problem messages
112  Warning(LogWarning),
113  /// Hard/recoverable problem messages
114  Error(LogError),
115  /// Critical/unrecoverable problem messages
116  Fatal(LogFatal),
117  /// Invalid tasks, work can not begin
118  Invalid(LogInvalid),
119}
120impl LogRecord for TaskMessage {
121  fn task_id(&self) -> i64 {
122    use crate::helpers::TaskMessage::*;
123    match *self {
124      Info(ref record) => record.task_id(),
125      Warning(ref record) => record.task_id(),
126      Error(ref record) => record.task_id(),
127      Fatal(ref record) => record.task_id(),
128      Invalid(ref record) => record.task_id(),
129    }
130  }
131  fn category(&self) -> &str {
132    use crate::helpers::TaskMessage::*;
133    match *self {
134      Info(ref record) => record.category(),
135      Warning(ref record) => record.category(),
136      Error(ref record) => record.category(),
137      Fatal(ref record) => record.category(),
138      Invalid(ref record) => record.category(),
139    }
140  }
141  fn what(&self) -> &str {
142    use crate::helpers::TaskMessage::*;
143    match *self {
144      Info(ref record) => record.what(),
145      Warning(ref record) => record.what(),
146      Error(ref record) => record.what(),
147      Fatal(ref record) => record.what(),
148      Invalid(ref record) => record.what(),
149    }
150  }
151  fn details(&self) -> &str {
152    use crate::helpers::TaskMessage::*;
153    match *self {
154      Info(ref record) => record.details(),
155      Warning(ref record) => record.details(),
156      Error(ref record) => record.details(),
157      Fatal(ref record) => record.details(),
158      Invalid(ref record) => record.details(),
159    }
160  }
161  fn set_details(&mut self, new_details: String) {
162    use crate::helpers::TaskMessage::*;
163    match *self {
164      Info(ref mut record) => record.set_details(new_details),
165      Warning(ref mut record) => record.set_details(new_details),
166      Error(ref mut record) => record.set_details(new_details),
167      Fatal(ref mut record) => record.set_details(new_details),
168      Invalid(ref mut record) => record.set_details(new_details),
169    }
170  }
171  fn severity(&self) -> &str {
172    use crate::helpers::TaskMessage::*;
173    match *self {
174      Info(ref record) => record.severity(),
175      Warning(ref record) => record.severity(),
176      Error(ref record) => record.severity(),
177      Fatal(ref record) => record.severity(),
178      Invalid(ref record) => record.severity(),
179    }
180  }
181}
182
183impl TaskStatus {
184  /// Maps the enumeration into the raw ints for the Task store
185  pub fn raw(&self) -> i32 {
186    match *self {
187      TaskStatus::TODO => 0,
188      TaskStatus::NoProblem => -1,
189      TaskStatus::Warning => -2,
190      TaskStatus::Error => -3,
191      TaskStatus::Fatal => -4,
192      TaskStatus::Invalid => -5,
193      TaskStatus::Blocked(x) | TaskStatus::Queued(x) => x,
194    }
195  }
196  /// Maps the enumeration into the raw severity string for the Task store logs / frontend reports
197  pub fn to_key(&self) -> String {
198    match *self {
199      TaskStatus::NoProblem => "no_problem",
200      TaskStatus::Warning => "warning",
201      TaskStatus::Error => "error",
202      TaskStatus::Fatal => "fatal",
203      TaskStatus::TODO => "todo",
204      TaskStatus::Invalid => "invalid",
205      TaskStatus::Blocked(_) => "blocked",
206      TaskStatus::Queued(_) => "queued",
207    }
208    .to_string()
209  }
210  /// Maps the enumeration into the Postgresql table name expected to hold messages for this
211  /// status
212  pub fn to_table(&self) -> String {
213    match *self {
214      TaskStatus::Warning => "log_warnings",
215      TaskStatus::Error => "log_errors",
216      TaskStatus::Fatal => "log_fatals",
217      TaskStatus::Invalid => "log_invalids",
218      _ => "log_infos",
219    }
220    .to_string()
221  }
222  /// Maps from the raw Task store value into the enumeration
223  pub fn from_raw(num: i32) -> Self {
224    match num {
225      0 => TaskStatus::TODO,
226      -1 => TaskStatus::NoProblem,
227      -2 => TaskStatus::Warning,
228      -3 => TaskStatus::Error,
229      -4 => TaskStatus::Fatal,
230      -5 => TaskStatus::Invalid,
231      num if num < -5 => TaskStatus::Blocked(num),
232      _ => TaskStatus::Queued(num),
233    }
234  }
235  /// Maps from the raw severity log values into the enumeration
236  pub fn from_key(key: &str) -> Option<Self> {
237    match key.to_lowercase().as_str() {
238      "no_problem" => Some(TaskStatus::NoProblem),
239      "warning" => Some(TaskStatus::Warning),
240      "error" => Some(TaskStatus::Error),
241      "todo" => Some(TaskStatus::TODO),
242      "in_progress" => Some(TaskStatus::TODO),
243      "invalid" => Some(TaskStatus::Invalid),
244      "blocked" => Some(TaskStatus::Blocked(-6)),
245      "queued" => Some(TaskStatus::Queued(1)),
246      "fatal" => Some(TaskStatus::Fatal),
247      _ => None,
248    }
249  }
250  /// Returns all raw severity strings as a vector
251  pub fn keys() -> Vec<String> {
252    [
253      "no_problem",
254      "warning",
255      "error",
256      "fatal",
257      "invalid",
258      "todo",
259      "blocked",
260      "queued",
261    ]
262    .iter()
263    .map(|&x| x.to_string())
264    .collect::<Vec<_>>()
265  }
266}
267
268#[derive(Clone, Debug)]
269/// Enum for all types of reported messages for a given Task, as per the `LaTeXML` convention
270/// One of "invalid", "fatal", "error", "warning" or "info"
271pub enum NewTaskMessage {
272  /// Debug/low-priroity messages
273  Info(NewLogInfo),
274  /// Soft/resumable problem messages
275  Warning(NewLogWarning),
276  /// Hard/recoverable problem messages
277  Error(NewLogError),
278  /// Critical/unrecoverable problem messages
279  Fatal(NewLogFatal),
280  /// Invalid tasks, work can not begin
281  Invalid(NewLogInvalid),
282}
283impl LogRecord for NewTaskMessage {
284  fn task_id(&self) -> i64 {
285    use crate::helpers::NewTaskMessage::*;
286    match *self {
287      Info(ref record) => record.task_id(),
288      Warning(ref record) => record.task_id(),
289      Error(ref record) => record.task_id(),
290      Fatal(ref record) => record.task_id(),
291      Invalid(ref record) => record.task_id(),
292    }
293  }
294  fn category(&self) -> &str {
295    use crate::helpers::NewTaskMessage::*;
296    match *self {
297      Info(ref record) => record.category(),
298      Warning(ref record) => record.category(),
299      Error(ref record) => record.category(),
300      Fatal(ref record) => record.category(),
301      Invalid(ref record) => record.category(),
302    }
303  }
304  fn what(&self) -> &str {
305    use crate::helpers::NewTaskMessage::*;
306    match *self {
307      Info(ref record) => record.what(),
308      Warning(ref record) => record.what(),
309      Error(ref record) => record.what(),
310      Fatal(ref record) => record.what(),
311      Invalid(ref record) => record.what(),
312    }
313  }
314  fn details(&self) -> &str {
315    use crate::helpers::NewTaskMessage::*;
316    match *self {
317      Info(ref record) => record.details(),
318      Warning(ref record) => record.details(),
319      Error(ref record) => record.details(),
320      Fatal(ref record) => record.details(),
321      Invalid(ref record) => record.details(),
322    }
323  }
324  fn set_details(&mut self, new_details: String) {
325    use crate::helpers::NewTaskMessage::*;
326    match *self {
327      Info(ref mut record) => record.set_details(new_details),
328      Warning(ref mut record) => record.set_details(new_details),
329      Error(ref mut record) => record.set_details(new_details),
330      Fatal(ref mut record) => record.set_details(new_details),
331      Invalid(ref mut record) => record.set_details(new_details),
332    }
333  }
334
335  fn severity(&self) -> &str {
336    use crate::helpers::NewTaskMessage::*;
337    match *self {
338      Info(ref record) => record.severity(),
339      Warning(ref record) => record.severity(),
340      Error(ref record) => record.severity(),
341      Fatal(ref record) => record.severity(),
342      Invalid(ref record) => record.severity(),
343    }
344  }
345}
346impl CortexInsertable for NewTaskMessage {
347  fn create(&self, connection: &mut PgConnection) -> Result<usize, Error> {
348    use crate::helpers::NewTaskMessage::*;
349    match *self {
350      Info(ref record) => record.create(connection),
351      Warning(ref record) => record.create(connection),
352      Error(ref record) => record.create(connection),
353      Fatal(ref record) => record.create(connection),
354      Invalid(ref record) => record.create(connection),
355    }
356  }
357}
358
359impl NewTaskMessage {
360  /// Instantiates an appropriate insertable LogRecord object based on the raw message components
361  pub fn new(
362    task_id: i64,
363    severity: &str,
364    category: String,
365    what: String,
366    details: String,
367  ) -> NewTaskMessage {
368    match severity.to_lowercase().as_str() {
369      // Canonical Perl-LaTeXML token is "Warning"; tolerate the abbreviated "Warn" too so a
370      // producer that emits `Warn:` doesn't silently land in the `_ => Info` default (which once
371      // misfiled every latexml-oxide warning as Info).
372      "warning" | "warn" => NewTaskMessage::Warning(NewLogWarning {
373        task_id,
374        category,
375        what,
376        details,
377      }),
378      "error" => NewTaskMessage::Error(NewLogError {
379        task_id,
380        category,
381        what,
382        details,
383      }),
384      // Perl-LaTeXML emits `Fatal('invalid', …)` for an *unprocessable input* (no TeX source,
385      // PDF-only, binary, …): a fatal whose CATEGORY is `invalid`. CorTeX treats that as its
386      // distinct **Invalid** outcome — its own report row, discounted from the total — not a plain
387      // conversion Fatal. This is the bridge that realizes the long-noted intent in
388      // `generate_report` ("they're fatal messages in latexml, but we want them separated out in
389      // cortex"); the literal `invalid:` severity below resolves to the same record.
390      "fatal" if category == "invalid" => NewTaskMessage::Invalid(NewLogInvalid {
391        task_id,
392        category,
393        what,
394        details,
395      }),
396      "fatal" => NewTaskMessage::Fatal(NewLogFatal {
397        category,
398        task_id,
399        what,
400        details,
401      }),
402      "invalid" => NewTaskMessage::Invalid(NewLogInvalid {
403        task_id,
404        category,
405        what,
406        details,
407      }),
408      _ => NewTaskMessage::Info(NewLogInfo {
409        task_id,
410        category,
411        what,
412        details,
413      }), // unknown severity will be treated as info
414    }
415  }
416}
417
418/// Parses a log string which follows the `LaTeXML` convention
419/// (described at [the Manual](http://dlmf.nist.gov/LaTeXML/manual/errorcodes/index.html))
420pub fn parse_log(task_id: i64, log: &str) -> Vec<NewTaskMessage> {
421  let mut messages: Vec<NewTaskMessage> = Vec::new();
422  let mut in_details_mode = false;
423
424  for line in log.lines() {
425    // Skip empty lines
426    if line.is_empty() {
427      continue;
428    }
429    // If we have found a message header and we're collecting details:
430    if in_details_mode {
431      // If the line starts with tab, we are indeed reading in details
432      if line.starts_with('\t') {
433        // Append the details line to the last message. `in_details_mode` is only ever set true
434        // right after a message is pushed, so `messages` is non-empty here by invariant — but we
435        // never `.expect()`-panic the dispatch path on a hostile log (DESIGN_PRINCIPLES): if that
436        // invariant is ever broken, treat the orphan details line as noise and carry on rather than
437        // aborting the whole task's log parse.
438        if let Some(mut last_message) = messages.pop() {
439          let mut truncated_details = last_message.details().to_string() + "\n" + line;
440          utf_truncate(&mut truncated_details, 2000);
441          last_message.set_details(truncated_details);
442          messages.push(last_message);
443        }
444        continue; // This line has been consumed, next
445      } else {
446        // Otherwise, no tab at the line beginning means last message has ended
447        in_details_mode = false;
448        if in_details_mode {} // hacky? disable "unused" warning
449      }
450    }
451    // Since this isn't a details line, check if it's a message line:
452    if let Some(cap) = MESSAGE_LINE_REGEX.captures(line) {
453      // Indeed a message, so record it
454      // We'll need to do some manual truncations, since the POSTGRESQL wrapper prefers
455      //   panicking to auto-truncating (would not have been the Perl way, but Rust is Rust)
456      let mut truncated_severity = cap
457        .get(1)
458        .map_or("", |m| m.as_str())
459        .to_string()
460        .to_lowercase();
461      utf_truncate(&mut truncated_severity, 50);
462      let mut truncated_category = cap.get(2).map_or("", |m| m.as_str()).to_string();
463      utf_truncate(&mut truncated_category, 100);
464      let mut truncated_what = cap.get(3).map_or("", |m| m.as_str()).to_string();
465      // `what` is varchar(200) (widened from 50 so math-parser footprints — `ambiguous_math` /
466      // `unparsed_math` token-type signatures — fit as the groupable key; the full stream stays in
467      // `details`). `category` stays varchar(50).
468      utf_truncate(&mut truncated_what, 200);
469      let mut truncated_details = cap.get(5).map_or("", |m| m.as_str()).to_string();
470      utf_truncate(&mut truncated_details, 2000);
471
472      if truncated_severity == "fatal" && truncated_category == "invalid" {
473        truncated_severity = "invalid".to_string();
474        truncated_category = truncated_what;
475        truncated_what = "all".to_string();
476        // The swap just moved a (now up-to-200-char) `what` into `category`, which is
477        // varchar(100) — re-clamp so the insert can't overflow it.
478        utf_truncate(&mut truncated_category, 100);
479      }
480
481      let message = NewTaskMessage::new(
482        task_id,
483        &truncated_severity,
484        truncated_category,
485        truncated_what,
486        truncated_details,
487      );
488      // Prepare to record follow-up lines with the message details:
489      in_details_mode = true;
490      // Add to the array of parsed messages
491      messages.push(message);
492    } else {
493      in_details_mode = false; // not a details line.
494      if let Some(cap) = LOADING_LINE_REGEX.captures(line) {
495        // Special case is a "Loading..." info messages
496        let mut filepath = cap.get(1).map_or("", |m| m.as_str()).to_string();
497        let mut filename = cap.get(2).map_or("", |m| m.as_str()).to_string();
498        // `what` (the filename) is a groupable key, capped at 100 — its column is
499        // varchar(200). The full path goes to `details` (varchar(2000)); the old 50
500        // cap cut real texmf paths mid-directory — for tcolorbox.sty the directory
501        // alone is exactly 50 chars, so the filename was dropped outright, leaving
502        // an unusable source path.
503        utf_truncate(&mut filename, 100);
504        filepath += &filename;
505        utf_truncate(&mut filepath, 2000);
506        messages.push(NewTaskMessage::new(
507          task_id,
508          "info",
509          "loaded_file".to_string(),
510          filename,
511          filepath,
512        ));
513      } else {
514        // Otherwise line is just noise, continue...
515      }
516    }
517  }
518  messages
519}
520
521/// Decodes raw worker-log bytes into a string, **tolerating non-UTF-8 input** (arXiv data is
522/// hostile and workers are unpredictable — W-2). A single invalid byte used to discard the whole
523/// log and force-mark the task `Fatal`, throwing away every real conversion message + the true
524/// status. Instead we decode lossily (invalid sequences → U+FFFD), preserving the real log, and
525/// append a `Warning` line so the encoding issue is recorded *transparently* rather than silently
526/// swallowed.
527fn decode_worker_log(raw: &[u8]) -> String {
528  match str::from_utf8(raw) {
529    Ok(valid) => valid.to_string(),
530    Err(_) => {
531      let mut lossy = String::from_utf8_lossy(raw).into_owned();
532      lossy.push_str(
533        "\nWarning:cortex:non_utf8_log the worker log was not valid UTF-8; decoded lossily\n",
534      );
535      lossy
536    },
537  }
538}
539
540/// Reads + decodes the `cortex.log` entry out of a result `.zip`. Uses the pure-Rust `zip` crate's
541/// **random-access `by_name`** — it seeks straight to `cortex.log` via the archive's central
542/// directory, never decompressing the (potentially large) converted output (the per-task hot path;
543/// ~1.4× libarchive on this op, see `docs/archive/ARCHIVE_RATIONALIZATION.md`). Returns the decoded
544/// log text, or an `Err` describing why it couldn't (a non-zip / corrupt archive, or a missing
545/// `cortex.log` → the task is left `Fatal`), rather than `.expect()`-panicking the dispatch path as
546/// the old libarchive reader did.
547fn read_cortex_log(result: &Path) -> Result<String, String> {
548  let file = File::open(result).map_err(|e| format!("cannot open result archive: {e}"))?;
549  let mut archive = zip::ZipArchive::new(file).map_err(|e| format!("not a readable zip: {e}"))?;
550  let mut entry = archive
551    .by_name("cortex.log")
552    .map_err(|e| format!("no cortex.log entry: {e}"))?;
553  let mut raw = Vec::new();
554  entry
555    .read_to_end(&mut raw)
556    .map_err(|e| format!("reading cortex.log failed: {e}"))?;
557  Ok(decode_worker_log(&raw))
558}
559
560/// Generates a `TaskReport` from a result archive (`.zip`) of a `CorTeX` processing job, following
561/// the `LaTeXML` messaging conventions in its `cortex.log`. Returns **`None` when the archive is
562/// unreadable or empty** (0-byte, truncated, non-zip, or no `cortex.log` entry): that is an
563/// *infrastructure* failure — the worker returned nothing usable — **not** a conversion verdict, so
564/// we do **not** fabricate a terminal `Fatal`. The caller (sink) skips finalizing it, leaving the
565/// task in-flight for the lease reaper, which **retries** it and only dead-letters it (with a
566/// `cortex/never_completed_with_retries` message) after `MAX_DISPATCH_RETRIES` — never a silent,
567/// unretried Fatal (KNOWN_ISSUES D-18). A *readable* `cortex.log` always yields `Some` — a genuine
568/// verdict, even when it parses to `Fatal`.
569pub fn generate_report(task: Task, result: &Path) -> Option<TaskReport> {
570  let log_string = match read_cortex_log(result) {
571    Ok(log_string) => log_string,
572    Err(reason) => {
573      // Infrastructure failure (D-18): don't invent a verdict — defer to the reaper's retry path.
574      println!("-- generate_report: {reason} (result {result:?}); deferring to the reaper (D-18)");
575      return None;
576    },
577  };
578  let mut messages = Vec::new();
579  let mut status = TaskStatus::Fatal; // Fatal by default — overridden by the conversion status line.
580  // Look for the special status message - Fatal otherwise!
581  for message in parse_log(task.id, &log_string).into_iter() {
582    // Invalids are a bit of a workaround for now, they're fatal messages in latexml, but
583    // we want them separated out in cortex
584    let mut skip_message = false;
585    match message {
586      NewTaskMessage::Invalid(ref _log_invalid) => {
587        status = TaskStatus::Invalid;
588      },
589      NewTaskMessage::Info(ref _log_info) => {
590        let message_what = message.what();
591        if message.category() == "conversion" && !message_what.is_empty() {
592          // Adapt status to the CorTeX scheme: cortex_status = -(latexml_status+1)
593          let latexml_scheme_status = match message_what.parse::<i32>() {
594            Ok(num) => num,
595            Err(e) => {
596              println!(
597                "-- generate_report: failed to parse conversion status {message_what:?}: {e:?}"
598              );
599              3 // latexml raw fatal
600            },
601          };
602          let cortex_scheme_status = -(latexml_scheme_status + 1);
603          if status != TaskStatus::Invalid {
604            // Invalid status is final, and derived, all others are set directly from the log.
605            status = TaskStatus::from_raw(cortex_scheme_status);
606          }
607          skip_message = true; // do not record the status message
608        }
609      },
610      _ => {},
611    };
612    if !skip_message {
613      messages.push(message);
614    }
615  }
616
617  Some(TaskReport {
618    task,
619    status,
620    messages,
621  })
622}
623
624/// Returns an open file handle to the task's entry
625pub fn prepare_input_stream(task: &Task) -> Result<File, io::Error> {
626  let entry_path = Path::new(&task.entry);
627  File::open(entry_path)
628}
629
630/// The single source of truth for where a service's **result archive** lives, given a task's source
631/// `entry`. The sink writes it here and the frontend reads it back from here — previously three
632/// call sites re-derived the same `<entry-dir>/<service>.zip` three different ways (a
633/// `Path::parent` and two subtly-different regexes), so this collapses them into one.
634///
635/// `sandbox_id` carries the F-6 fix: a **sandbox** corpus shares the parent's source `entry` paths
636/// in place (owner decision: no source copy), so keying its outputs on `entry` alone would let a
637/// sandbox rerun overwrite the parent's archives. When `Some(id)` (the sandbox's own corpus id) the
638/// archive is name-scoped — `<entry-dir>/<service>.sandbox-<id>.zip` — so a sandbox's outputs never
639/// collide with the parent's (or another sandbox's). `None` keeps the historical
640/// `<entry-dir>/<service>.zip` for ordinary corpora (backward-compatible with existing archives).
641///
642/// Returns `None` if `entry` has no parent directory (a malformed/relative entry) — the caller then
643/// has no result path to write or serve.
644pub fn result_archive_path(
645  entry: &str,
646  service_name: &str,
647  sandbox_id: Option<i32>,
648) -> Option<PathBuf> {
649  let dir = Path::new(entry.trim_end())
650    .parent()
651    .and_then(Path::to_str)?;
652  let stem = match sandbox_id {
653    Some(id) => format!("{service_name}.sandbox-{id}"),
654    None => service_name.to_string(),
655  };
656  Some(PathBuf::from(format!("{dir}/{stem}.zip")))
657}
658
659/// Utility functions, until they find a better place
660pub fn utf_truncate(input: &mut String, maxsize: usize) {
661  let mut utf_maxsize = input.len();
662  if utf_maxsize >= maxsize {
663    {
664      let mut char_iter = input.char_indices();
665      while utf_maxsize >= maxsize {
666        utf_maxsize = match char_iter.next_back() {
667          Some((index, _)) => index,
668          _ => 0,
669        };
670      }
671    } // Extra {} wrap to limit the immutable borrow of char_indices()
672    input.truncate(utf_maxsize);
673  }
674  // eliminate null characters if any
675  *input = input.replace('\x00', "");
676}
677
678/// Generate a random integer useful for temporary DB marks
679pub fn random_mark() -> i32 {
680  let mut rng = rand::rng();
681  let mark_rng: u16 = rng.random();
682  i32::from(mark_rng)
683}
684
685/// Random temporary task `status` sentinel for the `mark_rerun` two-phase reset.
686///
687/// `mark_rerun` stamps the in-scope tasks with this value, then re-selects them by
688/// `status = <sentinel>` to delete their logs and flip them to `TODO`. The sentinel must be
689/// disjoint from **every** value a task can otherwise legitimately hold, or an unrelated task gets
690/// swept into the rerun: live leases occupy `[1, 65536]` (`fetch_tasks` stamps `1 + u16`), `TODO`
691/// is `0`, and completed/Blocked tasks are `< 0`. So we draw a **positive** value strictly *above*
692/// the maximum lease — in `[65537, 131072]` — which collides with none of them. (The old rerun mark
693/// reused `random_mark`'s raw `u16`, which overlapped the live-lease range and aliased `TODO` at 0;
694/// KNOWN_ISSUES R-13.) All temp marks stay `> 0`.
695pub fn rerun_mark() -> i32 {
696  // One past the largest value `fetch_tasks` can stamp on a lease (`1 + u16::MAX` = 65536).
697  const LEASE_CEILING: i32 = u16::MAX as i32 + 1;
698  LEASE_CEILING + 1 + i32::from(rand_in_range(0, u16::MAX))
699}
700
701/// Helper for generating a random i32 in a range, to avoid loading the rng crate + boilerplate
702pub fn rand_in_range(from: u16, to: u16) -> u16 {
703  let mut rng = rand::rng();
704  let mark_rng: u16 = rng.random_range(from..=to);
705  mark_rng
706}
707
708#[cfg(test)]
709mod log_decode_tests {
710  //! W-2: a non-UTF-8 worker log must degrade gracefully (decode lossily + record a warning), not
711  //! get discarded wholesale with the task force-marked Fatal. DB-free, so no L-1 teardown risk.
712  use super::{decode_worker_log, parse_log};
713  use crate::models::LogRecord;
714
715  #[test]
716  fn valid_utf8_passes_through_unchanged() {
717    let valid = "Warning:math:undefined hello\nStatus:conversion:1\n";
718    assert_eq!(decode_worker_log(valid.as_bytes()), valid);
719    assert!(
720      !decode_worker_log(valid.as_bytes()).contains("non_utf8_log"),
721      "no spurious warning is added to a clean log"
722    );
723  }
724
725  #[test]
726  fn non_utf8_decodes_lossily_not_fatal() {
727    // A stray 0xFF byte in an otherwise-real conversion log.
728    let raw = b"Warning:math:undefined bad \xFF byte\nStatus:conversion:1\n";
729    let decoded = decode_worker_log(raw);
730    // The real conversion status + the real message survive (the W-2 regression: previously the
731    // whole log was thrown away and the task force-marked Fatal over this single byte).
732    assert!(
733      decoded.contains("Status:conversion:1"),
734      "the real conversion status survives lossy decoding"
735    );
736    assert!(decoded.contains("Warning:math:undefined"));
737    assert!(
738      decoded.contains('\u{FFFD}'),
739      "the invalid byte became the Unicode replacement char"
740    );
741    // The encoding issue is recorded transparently rather than silently swallowed.
742    assert!(decoded.contains("Warning:cortex:non_utf8_log"));
743    // And it parses into multiple real messages, not a single fatal.
744    assert!(
745      parse_log(1, &decoded).len() >= 2,
746      "real messages are preserved, not collapsed into one fatal"
747    );
748  }
749
750  #[test]
751  fn warn_abbreviation_is_recognized_as_warning() {
752    // latexml-oxide historically emitted the abbreviated `Warn:` token; cortex must file it as a
753    // Warning, not silently default it to Info (the `_ => Info` arm), which once left every warning
754    // task showing "no_messages" in the report. The canonical Perl token `Warning:` behaves the
755    // same.
756    use super::NewTaskMessage;
757    let abbrev = parse_log(42, "Warn:missing_file:rotfloat.sty stubbed\n");
758    assert_eq!(abbrev.len(), 1);
759    assert!(
760      matches!(abbrev[0], NewTaskMessage::Warning(_)),
761      "abbreviated `Warn:` is filed as a Warning, not defaulted to Info"
762    );
763    let canonical = parse_log(42, "Warning:missing_file:x y\n");
764    assert!(matches!(canonical[0], NewTaskMessage::Warning(_)));
765  }
766
767  #[test]
768  fn runtime_line_parses_into_the_fields_mark_done_keys_on() {
769    use super::NewTaskMessage;
770    // New latexml-oxide format `Info:runtime_ms:<N>`: the value is the report category's drill-down
771    // `what` (mark.rs reads `what` for the denormalized runtime), details empty.
772    let new = parse_log(7, "Info:runtime_ms:12345\n");
773    assert_eq!(new.len(), 1);
774    let m = &new[0];
775    assert!(matches!(m, NewTaskMessage::Info(_)));
776    assert_eq!(m.category(), "runtime_ms");
777    assert_eq!(m.what(), "12345");
778    assert_eq!(m.details(), "");
779    // Legacy format `Info:cortex:runtime_ms <N>` must still parse the way the backward-compat arm
780    // expects: value in `details`, `what` is the literal `runtime_ms` (a mixed/rolling fleet).
781    let old = parse_log(7, "Info:cortex:runtime_ms 12345\n");
782    assert_eq!(old.len(), 1);
783    let m = &old[0];
784    assert_eq!(m.category(), "cortex");
785    assert_eq!(m.what(), "runtime_ms");
786    assert_eq!(m.details(), "12345");
787  }
788
789  #[test]
790  fn fatal_invalid_category_is_separated_into_invalid() {
791    // Perl-LaTeXML emits `Fatal('invalid', …)` for an unprocessable input (no TeX source,
792    // PDF-only, binary). cortex must separate that out as its distinct **Invalid** outcome (own
793    // report row, discounted from the total) — NOT a plain conversion Fatal — even though the
794    // severity token is `Fatal`. The bridge keys on the `invalid` category.
795    use super::NewTaskMessage;
796    let m = parse_log(
797      7,
798      "Fatal:invalid:no_tex_source no .tex files found in archive\n",
799    );
800    assert_eq!(m.len(), 1);
801    assert!(
802      matches!(m[0], NewTaskMessage::Invalid(_)),
803      "a Fatal message in the `invalid` category is filed as Invalid, not Fatal"
804    );
805    // A fatal in any OTHER category is still a plain Fatal.
806    let f = parse_log(7, "Fatal:conversion:caught boom\n");
807    assert!(matches!(f[0], NewTaskMessage::Fatal(_)));
808    // And the literal `Invalid:` severity (e.g. the sink's oversize-reject) still maps to Invalid.
809    let i = parse_log(7, "Invalid:size:too_big result exceeds cap\n");
810    assert!(matches!(i[0], NewTaskMessage::Invalid(_)));
811  }
812
813  #[test]
814  fn read_cortex_log_extracts_from_zip_and_errors_gracefully() {
815    use super::read_cortex_log;
816    use std::fs::File;
817    use std::io::Write;
818    let zip_path = std::env::temp_dir().join("cortex_read_log_unit_test.zip");
819    {
820      let mut zw = zip::ZipWriter::new(File::create(&zip_path).unwrap());
821      let opts: zip::write::FileOptions<()> = zip::write::FileOptions::default();
822      // A large output entry first; cortex.log last — `by_name` must seek straight past it.
823      zw.start_file("html/index.html", opts).unwrap();
824      zw.write_all(&vec![b'x'; 200_000]).unwrap();
825      zw.start_file("cortex.log", opts).unwrap();
826      zw.write_all(b"Info:cortex:hello a worker log line\n")
827        .unwrap();
828      zw.finish().unwrap();
829    }
830    let log = read_cortex_log(&zip_path).expect("reads cortex.log out of the zip");
831    assert!(
832      log.contains("hello a worker log line"),
833      "by_name extracted the cortex.log content, got: {log:?}"
834    );
835    // A missing / non-zip path is a graceful Err (task left Fatal), never a panic.
836    assert!(read_cortex_log(std::path::Path::new("/nonexistent/x.zip")).is_err());
837    std::fs::remove_file(&zip_path).ok();
838  }
839
840  #[test]
841  fn generate_report_defers_unreadable_archives_to_the_reaper() {
842    use super::{Task, TaskStatus, generate_report};
843    use std::fs::File;
844    use std::io::Write;
845    let task = || Task {
846      id: 42,
847      service_id: 3,
848      corpus_id: 2,
849      status: 0,
850      entry: "/d/x.zip".to_string(),
851    };
852    // D-18: a 0-byte / missing / non-zip archive is an *infrastructure* failure → None, so the sink
853    // defers to the reaper (retry → dead-letter with a message) instead of a silent terminal Fatal.
854    let empty = std::env::temp_dir().join("cortex_d18_empty_result.zip");
855    File::create(&empty).unwrap(); // 0 bytes — "not a readable zip"
856    assert!(
857      generate_report(task(), &empty).is_none(),
858      "a 0-byte result archive defers to the reaper (None), never a fabricated Fatal"
859    );
860    assert!(
861      generate_report(task(), std::path::Path::new("/nonexistent/x.zip")).is_none(),
862      "a missing result archive defers to the reaper (None)"
863    );
864    // A *readable* cortex.log is always a genuine verdict — Some, even when it parses a conversion
865    // status (here conversion:0 → NoProblem).
866    let good = std::env::temp_dir().join("cortex_d18_good_result.zip");
867    {
868      let mut zw = zip::ZipWriter::new(File::create(&good).unwrap());
869      let opts: zip::write::FileOptions<()> = zip::write::FileOptions::default();
870      zw.start_file("cortex.log", opts).unwrap();
871      zw.write_all(b"Info:conversion:0 clean run\n").unwrap();
872      zw.finish().unwrap();
873    }
874    let report =
875      generate_report(task(), &good).expect("a readable cortex.log yields a verdict (Some)");
876    assert_eq!(
877      report.status,
878      TaskStatus::NoProblem,
879      "conversion:0 status line -> NoProblem"
880    );
881    std::fs::remove_file(&empty).ok();
882    std::fs::remove_file(&good).ok();
883  }
884
885  #[test]
886  fn tab_indented_details_fold_into_the_preceding_message() {
887    // A message header followed by tab-indented continuation lines: the details fold into the one
888    // preceding message (the path that used to run through the `.expect()`).
889    let log = "Error:math:bad first detail\n\tsecond line\n\tthird line\n";
890    let messages = parse_log(7, log);
891    assert_eq!(messages.len(), 1, "one message, the details folded into it");
892    let details = messages[0].details();
893    assert!(details.contains("first detail"), "got: {details:?}");
894    assert!(details.contains("second line"), "got: {details:?}");
895    assert!(details.contains("third line"), "got: {details:?}");
896  }
897
898  #[test]
899  fn orphan_details_line_does_not_panic() {
900    // A hostile log that opens with a tab-indented "details" line before any message header must be
901    // treated as noise, never panic the dispatch path (DESIGN_PRINCIPLES: no `.expect()` on a
902    // parse).
903    let messages = parse_log(
904      7,
905      "\torphan detail with no header\nInfo:cortex:hi a real line\n",
906    );
907    assert_eq!(
908      messages.len(),
909      1,
910      "the orphan line is ignored; the real message survives"
911    );
912  }
913
914  #[test]
915  fn loaded_file_details_keep_the_full_source_path() {
916    use super::NewTaskMessage;
917    // The `(Loading ...` special case: `what` is the groupable filename, `details` the full path.
918    // tcolorbox's texmf directory is exactly 50 chars, so the old 50-char cap on the joined path
919    // dropped the filename outright and recorded a bare, unusable directory.
920    let messages = parse_log(
921      7,
922      "(Loading /usr/share/texlive/texmf-dist/tex/latex/tcolorbox/tcolorbox.sty...\n",
923    );
924    assert_eq!(messages.len(), 1, "the Loading line parses to one message");
925    let m = &messages[0];
926    assert!(matches!(m, NewTaskMessage::Info(_)));
927    assert_eq!(m.category(), "loaded_file");
928    assert_eq!(m.what(), "tcolorbox.sty", "the filename stays groupable");
929    assert_eq!(
930      m.details(),
931      "/usr/share/texlive/texmf-dist/tex/latex/tcolorbox/tcolorbox.sty",
932      "details keeps the whole path, filename included"
933    );
934  }
935
936  #[test]
937  fn result_archive_path_scopes_sandbox_outputs() {
938    use super::result_archive_path;
939    let entry = "/data/arxiv/1234/5678/source/5678.zip";
940    // Ordinary corpus: the historical path, unchanged (backward-compatible).
941    assert_eq!(
942      result_archive_path(entry, "tex_to_html", None).unwrap(),
943      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.zip")
944    );
945    // Sandbox (F-6): same source dir, but a corpus-id-scoped archive name, so a sandbox rerun can
946    // never overwrite the parent's `tex_to_html.zip`.
947    assert_eq!(
948      result_archive_path(entry, "tex_to_html", Some(42)).unwrap(),
949      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.sandbox-42.zip")
950    );
951    // A trailing newline (DB `text` columns carry them) is trimmed before taking the parent dir.
952    assert_eq!(
953      result_archive_path(&format!("{entry}\n"), "tex_to_html", None).unwrap(),
954      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.zip")
955    );
956    // No parent directory (a blank/root entry) ⇒ no result path: the caller skips the write/serve
957    // instead of panicking.
958    assert!(result_archive_path("", "tex_to_html", None).is_none());
959    assert!(result_archive_path("/", "tex_to_html", None).is_none());
960  }
961}
962
963#[cfg(test)]
964mod rerun_mark_tests {
965  use super::rerun_mark;
966
967  #[test]
968  fn rerun_mark_stays_above_the_lease_space() {
969    // `fetch_tasks` stamps a lease as `1 + u16` (max 65536) and `TODO` is 0; the rerun sentinel
970    // must never land in `[0, 65536]`, or it could sweep a live in-flight lease (or alias `TODO`)
971    // into the rerun and double-dispatch it. All temp marks are `> 0`. KNOWN_ISSUES R-13.
972    for _ in 0..10_000 {
973      let mark = rerun_mark();
974      assert!(
975        mark > i32::from(u16::MAX) + 1,
976        "rerun mark {mark} overlaps the lease space [1, 65536]"
977      );
978      assert!(mark > 0, "temp marks must be positive");
979    }
980  }
981}
982
983#[cfg(test)]
984mod entry_name_tests {
985  use super::entry_document_name;
986
987  #[test]
988  fn extracts_short_document_name() {
989    // The motivating UX example: a download should be named by the entry, not the task id.
990    assert_eq!(
991      entry_document_name("/data/arxmliv/0811/0811.0417/0811.0417.zip"),
992      "0811.0417"
993    );
994    // Dotted arXiv ids keep their internal dot (the extension is the final segment).
995    assert_eq!(
996      entry_document_name("/data/foo/2105.13573.tar"),
997      "2105.13573"
998    );
999    // A trailing-whitespace entry (padded varchar) is trimmed first.
1000    assert_eq!(entry_document_name("/d/bar.tex   "), "bar");
1001    // No `/name.ext` shape -> falls back to the trimmed entry rather than erroring.
1002    assert_eq!(entry_document_name("noslash"), "noslash");
1003  }
1004}
cortex/helpers.rs

cortex/
helpers.rs
