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, 50);
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 still
477        // varchar(50) — re-clamp so the insert can't overflow it.
478        utf_truncate(&mut truncated_category, 50);
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        utf_truncate(&mut filename, 50);
499        filepath += &filename;
500        utf_truncate(&mut filepath, 50);
501        messages.push(NewTaskMessage::new(
502          task_id,
503          "info",
504          "loaded_file".to_string(),
505          filename,
506          filepath,
507        ));
508      } else {
509        // Otherwise line is just noise, continue...
510      }
511    }
512  }
513  messages
514}
515
516/// Decodes raw worker-log bytes into a string, **tolerating non-UTF-8 input** (arXiv data is
517/// hostile and workers are unpredictable — W-2). A single invalid byte used to discard the whole
518/// log and force-mark the task `Fatal`, throwing away every real conversion message + the true
519/// status. Instead we decode lossily (invalid sequences → U+FFFD), preserving the real log, and
520/// append a `Warning` line so the encoding issue is recorded *transparently* rather than silently
521/// swallowed.
522fn decode_worker_log(raw: &[u8]) -> String {
523  match str::from_utf8(raw) {
524    Ok(valid) => valid.to_string(),
525    Err(_) => {
526      let mut lossy = String::from_utf8_lossy(raw).into_owned();
527      lossy.push_str(
528        "\nWarning:cortex:non_utf8_log the worker log was not valid UTF-8; decoded lossily\n",
529      );
530      lossy
531    },
532  }
533}
534
535/// Reads + decodes the `cortex.log` entry out of a result `.zip`. Uses the pure-Rust `zip` crate's
536/// **random-access `by_name`** — it seeks straight to `cortex.log` via the archive's central
537/// directory, never decompressing the (potentially large) converted output (the per-task hot path;
538/// ~1.4× libarchive on this op, see `docs/archive/ARCHIVE_RATIONALIZATION.md`). Returns the decoded
539/// log text, or an `Err` describing why it couldn't (a non-zip / corrupt archive, or a missing
540/// `cortex.log` → the task is left `Fatal`), rather than `.expect()`-panicking the dispatch path as
541/// the old libarchive reader did.
542fn read_cortex_log(result: &Path) -> Result<String, String> {
543  let file = File::open(result).map_err(|e| format!("cannot open result archive: {e}"))?;
544  let mut archive = zip::ZipArchive::new(file).map_err(|e| format!("not a readable zip: {e}"))?;
545  let mut entry = archive
546    .by_name("cortex.log")
547    .map_err(|e| format!("no cortex.log entry: {e}"))?;
548  let mut raw = Vec::new();
549  entry
550    .read_to_end(&mut raw)
551    .map_err(|e| format!("reading cortex.log failed: {e}"))?;
552  Ok(decode_worker_log(&raw))
553}
554
555/// Generates a `TaskReport` from a result archive (`.zip`) of a `CorTeX` processing job, following
556/// the `LaTeXML` messaging conventions in its `cortex.log`. Returns **`None` when the archive is
557/// unreadable or empty** (0-byte, truncated, non-zip, or no `cortex.log` entry): that is an
558/// *infrastructure* failure — the worker returned nothing usable — **not** a conversion verdict, so
559/// we do **not** fabricate a terminal `Fatal`. The caller (sink) skips finalizing it, leaving the
560/// task in-flight for the lease reaper, which **retries** it and only dead-letters it (with a
561/// `cortex/never_completed_with_retries` message) after `MAX_DISPATCH_RETRIES` — never a silent,
562/// unretried Fatal (KNOWN_ISSUES D-18). A *readable* `cortex.log` always yields `Some` — a genuine
563/// verdict, even when it parses to `Fatal`.
564pub fn generate_report(task: Task, result: &Path) -> Option<TaskReport> {
565  let log_string = match read_cortex_log(result) {
566    Ok(log_string) => log_string,
567    Err(reason) => {
568      // Infrastructure failure (D-18): don't invent a verdict — defer to the reaper's retry path.
569      println!("-- generate_report: {reason} (result {result:?}); deferring to the reaper (D-18)");
570      return None;
571    },
572  };
573  let mut messages = Vec::new();
574  let mut status = TaskStatus::Fatal; // Fatal by default — overridden by the conversion status line.
575  // Look for the special status message - Fatal otherwise!
576  for message in parse_log(task.id, &log_string).into_iter() {
577    // Invalids are a bit of a workaround for now, they're fatal messages in latexml, but
578    // we want them separated out in cortex
579    let mut skip_message = false;
580    match message {
581      NewTaskMessage::Invalid(ref _log_invalid) => {
582        status = TaskStatus::Invalid;
583      },
584      NewTaskMessage::Info(ref _log_info) => {
585        let message_what = message.what();
586        if message.category() == "conversion" && !message_what.is_empty() {
587          // Adapt status to the CorTeX scheme: cortex_status = -(latexml_status+1)
588          let latexml_scheme_status = match message_what.parse::<i32>() {
589            Ok(num) => num,
590            Err(e) => {
591              println!(
592                "-- generate_report: failed to parse conversion status {message_what:?}: {e:?}"
593              );
594              3 // latexml raw fatal
595            },
596          };
597          let cortex_scheme_status = -(latexml_scheme_status + 1);
598          if status != TaskStatus::Invalid {
599            // Invalid status is final, and derived, all others are set directly from the log.
600            status = TaskStatus::from_raw(cortex_scheme_status);
601          }
602          skip_message = true; // do not record the status message
603        }
604      },
605      _ => {},
606    };
607    if !skip_message {
608      messages.push(message);
609    }
610  }
611
612  Some(TaskReport {
613    task,
614    status,
615    messages,
616  })
617}
618
619/// Returns an open file handle to the task's entry
620pub fn prepare_input_stream(task: &Task) -> Result<File, io::Error> {
621  let entry_path = Path::new(&task.entry);
622  File::open(entry_path)
623}
624
625/// The single source of truth for where a service's **result archive** lives, given a task's source
626/// `entry`. The sink writes it here and the frontend reads it back from here — previously three
627/// call sites re-derived the same `<entry-dir>/<service>.zip` three different ways (a
628/// `Path::parent` and two subtly-different regexes), so this collapses them into one.
629///
630/// `sandbox_id` carries the F-6 fix: a **sandbox** corpus shares the parent's source `entry` paths
631/// in place (owner decision: no source copy), so keying its outputs on `entry` alone would let a
632/// sandbox rerun overwrite the parent's archives. When `Some(id)` (the sandbox's own corpus id) the
633/// archive is name-scoped — `<entry-dir>/<service>.sandbox-<id>.zip` — so a sandbox's outputs never
634/// collide with the parent's (or another sandbox's). `None` keeps the historical
635/// `<entry-dir>/<service>.zip` for ordinary corpora (backward-compatible with existing archives).
636///
637/// Returns `None` if `entry` has no parent directory (a malformed/relative entry) — the caller then
638/// has no result path to write or serve.
639pub fn result_archive_path(
640  entry: &str,
641  service_name: &str,
642  sandbox_id: Option<i32>,
643) -> Option<PathBuf> {
644  let dir = Path::new(entry.trim_end())
645    .parent()
646    .and_then(Path::to_str)?;
647  let stem = match sandbox_id {
648    Some(id) => format!("{service_name}.sandbox-{id}"),
649    None => service_name.to_string(),
650  };
651  Some(PathBuf::from(format!("{dir}/{stem}.zip")))
652}
653
654/// Utility functions, until they find a better place
655pub fn utf_truncate(input: &mut String, maxsize: usize) {
656  let mut utf_maxsize = input.len();
657  if utf_maxsize >= maxsize {
658    {
659      let mut char_iter = input.char_indices();
660      while utf_maxsize >= maxsize {
661        utf_maxsize = match char_iter.next_back() {
662          Some((index, _)) => index,
663          _ => 0,
664        };
665      }
666    } // Extra {} wrap to limit the immutable borrow of char_indices()
667    input.truncate(utf_maxsize);
668  }
669  // eliminate null characters if any
670  *input = input.replace('\x00', "");
671}
672
673/// Generate a random integer useful for temporary DB marks
674pub fn random_mark() -> i32 {
675  let mut rng = rand::rng();
676  let mark_rng: u16 = rng.random();
677  i32::from(mark_rng)
678}
679
680/// Random temporary task `status` sentinel for the `mark_rerun` two-phase reset.
681///
682/// `mark_rerun` stamps the in-scope tasks with this value, then re-selects them by
683/// `status = <sentinel>` to delete their logs and flip them to `TODO`. The sentinel must be
684/// disjoint from **every** value a task can otherwise legitimately hold, or an unrelated task gets
685/// swept into the rerun: live leases occupy `[1, 65536]` (`fetch_tasks` stamps `1 + u16`), `TODO`
686/// is `0`, and completed/Blocked tasks are `< 0`. So we draw a **positive** value strictly *above*
687/// the maximum lease — in `[65537, 131072]` — which collides with none of them. (The old rerun mark
688/// reused `random_mark`'s raw `u16`, which overlapped the live-lease range and aliased `TODO` at 0;
689/// KNOWN_ISSUES R-13.) All temp marks stay `> 0`.
690pub fn rerun_mark() -> i32 {
691  // One past the largest value `fetch_tasks` can stamp on a lease (`1 + u16::MAX` = 65536).
692  const LEASE_CEILING: i32 = u16::MAX as i32 + 1;
693  LEASE_CEILING + 1 + i32::from(rand_in_range(0, u16::MAX))
694}
695
696/// Helper for generating a random i32 in a range, to avoid loading the rng crate + boilerplate
697pub fn rand_in_range(from: u16, to: u16) -> u16 {
698  let mut rng = rand::rng();
699  let mark_rng: u16 = rng.random_range(from..=to);
700  mark_rng
701}
702
703#[cfg(test)]
704mod log_decode_tests {
705  //! W-2: a non-UTF-8 worker log must degrade gracefully (decode lossily + record a warning), not
706  //! get discarded wholesale with the task force-marked Fatal. DB-free, so no L-1 teardown risk.
707  use super::{decode_worker_log, parse_log};
708  use crate::models::LogRecord;
709
710  #[test]
711  fn valid_utf8_passes_through_unchanged() {
712    let valid = "Warning:math:undefined hello\nStatus:conversion:1\n";
713    assert_eq!(decode_worker_log(valid.as_bytes()), valid);
714    assert!(
715      !decode_worker_log(valid.as_bytes()).contains("non_utf8_log"),
716      "no spurious warning is added to a clean log"
717    );
718  }
719
720  #[test]
721  fn non_utf8_decodes_lossily_not_fatal() {
722    // A stray 0xFF byte in an otherwise-real conversion log.
723    let raw = b"Warning:math:undefined bad \xFF byte\nStatus:conversion:1\n";
724    let decoded = decode_worker_log(raw);
725    // The real conversion status + the real message survive (the W-2 regression: previously the
726    // whole log was thrown away and the task force-marked Fatal over this single byte).
727    assert!(
728      decoded.contains("Status:conversion:1"),
729      "the real conversion status survives lossy decoding"
730    );
731    assert!(decoded.contains("Warning:math:undefined"));
732    assert!(
733      decoded.contains('\u{FFFD}'),
734      "the invalid byte became the Unicode replacement char"
735    );
736    // The encoding issue is recorded transparently rather than silently swallowed.
737    assert!(decoded.contains("Warning:cortex:non_utf8_log"));
738    // And it parses into multiple real messages, not a single fatal.
739    assert!(
740      parse_log(1, &decoded).len() >= 2,
741      "real messages are preserved, not collapsed into one fatal"
742    );
743  }
744
745  #[test]
746  fn warn_abbreviation_is_recognized_as_warning() {
747    // latexml-oxide historically emitted the abbreviated `Warn:` token; cortex must file it as a
748    // Warning, not silently default it to Info (the `_ => Info` arm), which once left every warning
749    // task showing "no_messages" in the report. The canonical Perl token `Warning:` behaves the
750    // same.
751    use super::NewTaskMessage;
752    let abbrev = parse_log(42, "Warn:missing_file:rotfloat.sty stubbed\n");
753    assert_eq!(abbrev.len(), 1);
754    assert!(
755      matches!(abbrev[0], NewTaskMessage::Warning(_)),
756      "abbreviated `Warn:` is filed as a Warning, not defaulted to Info"
757    );
758    let canonical = parse_log(42, "Warning:missing_file:x y\n");
759    assert!(matches!(canonical[0], NewTaskMessage::Warning(_)));
760  }
761
762  #[test]
763  fn runtime_line_parses_into_the_fields_mark_done_keys_on() {
764    use super::NewTaskMessage;
765    // New latexml-oxide format `Info:runtime_ms:<N>`: the value is the report category's drill-down
766    // `what` (mark.rs reads `what` for the denormalized runtime), details empty.
767    let new = parse_log(7, "Info:runtime_ms:12345\n");
768    assert_eq!(new.len(), 1);
769    let m = &new[0];
770    assert!(matches!(m, NewTaskMessage::Info(_)));
771    assert_eq!(m.category(), "runtime_ms");
772    assert_eq!(m.what(), "12345");
773    assert_eq!(m.details(), "");
774    // Legacy format `Info:cortex:runtime_ms <N>` must still parse the way the backward-compat arm
775    // expects: value in `details`, `what` is the literal `runtime_ms` (a mixed/rolling fleet).
776    let old = parse_log(7, "Info:cortex:runtime_ms 12345\n");
777    assert_eq!(old.len(), 1);
778    let m = &old[0];
779    assert_eq!(m.category(), "cortex");
780    assert_eq!(m.what(), "runtime_ms");
781    assert_eq!(m.details(), "12345");
782  }
783
784  #[test]
785  fn fatal_invalid_category_is_separated_into_invalid() {
786    // Perl-LaTeXML emits `Fatal('invalid', …)` for an unprocessable input (no TeX source,
787    // PDF-only, binary). cortex must separate that out as its distinct **Invalid** outcome (own
788    // report row, discounted from the total) — NOT a plain conversion Fatal — even though the
789    // severity token is `Fatal`. The bridge keys on the `invalid` category.
790    use super::NewTaskMessage;
791    let m = parse_log(
792      7,
793      "Fatal:invalid:no_tex_source no .tex files found in archive\n",
794    );
795    assert_eq!(m.len(), 1);
796    assert!(
797      matches!(m[0], NewTaskMessage::Invalid(_)),
798      "a Fatal message in the `invalid` category is filed as Invalid, not Fatal"
799    );
800    // A fatal in any OTHER category is still a plain Fatal.
801    let f = parse_log(7, "Fatal:conversion:caught boom\n");
802    assert!(matches!(f[0], NewTaskMessage::Fatal(_)));
803    // And the literal `Invalid:` severity (e.g. the sink's oversize-reject) still maps to Invalid.
804    let i = parse_log(7, "Invalid:size:too_big result exceeds cap\n");
805    assert!(matches!(i[0], NewTaskMessage::Invalid(_)));
806  }
807
808  #[test]
809  fn read_cortex_log_extracts_from_zip_and_errors_gracefully() {
810    use super::read_cortex_log;
811    use std::fs::File;
812    use std::io::Write;
813    let zip_path = std::env::temp_dir().join("cortex_read_log_unit_test.zip");
814    {
815      let mut zw = zip::ZipWriter::new(File::create(&zip_path).unwrap());
816      let opts: zip::write::FileOptions<()> = zip::write::FileOptions::default();
817      // A large output entry first; cortex.log last — `by_name` must seek straight past it.
818      zw.start_file("html/index.html", opts).unwrap();
819      zw.write_all(&vec![b'x'; 200_000]).unwrap();
820      zw.start_file("cortex.log", opts).unwrap();
821      zw.write_all(b"Info:cortex:hello a worker log line\n")
822        .unwrap();
823      zw.finish().unwrap();
824    }
825    let log = read_cortex_log(&zip_path).expect("reads cortex.log out of the zip");
826    assert!(
827      log.contains("hello a worker log line"),
828      "by_name extracted the cortex.log content, got: {log:?}"
829    );
830    // A missing / non-zip path is a graceful Err (task left Fatal), never a panic.
831    assert!(read_cortex_log(std::path::Path::new("/nonexistent/x.zip")).is_err());
832    std::fs::remove_file(&zip_path).ok();
833  }
834
835  #[test]
836  fn generate_report_defers_unreadable_archives_to_the_reaper() {
837    use super::{Task, TaskStatus, generate_report};
838    use std::fs::File;
839    use std::io::Write;
840    let task = || Task {
841      id: 42,
842      service_id: 3,
843      corpus_id: 2,
844      status: 0,
845      entry: "/d/x.zip".to_string(),
846    };
847    // D-18: a 0-byte / missing / non-zip archive is an *infrastructure* failure → None, so the sink
848    // defers to the reaper (retry → dead-letter with a message) instead of a silent terminal Fatal.
849    let empty = std::env::temp_dir().join("cortex_d18_empty_result.zip");
850    File::create(&empty).unwrap(); // 0 bytes — "not a readable zip"
851    assert!(
852      generate_report(task(), &empty).is_none(),
853      "a 0-byte result archive defers to the reaper (None), never a fabricated Fatal"
854    );
855    assert!(
856      generate_report(task(), std::path::Path::new("/nonexistent/x.zip")).is_none(),
857      "a missing result archive defers to the reaper (None)"
858    );
859    // A *readable* cortex.log is always a genuine verdict — Some, even when it parses a conversion
860    // status (here conversion:0 → NoProblem).
861    let good = std::env::temp_dir().join("cortex_d18_good_result.zip");
862    {
863      let mut zw = zip::ZipWriter::new(File::create(&good).unwrap());
864      let opts: zip::write::FileOptions<()> = zip::write::FileOptions::default();
865      zw.start_file("cortex.log", opts).unwrap();
866      zw.write_all(b"Info:conversion:0 clean run\n").unwrap();
867      zw.finish().unwrap();
868    }
869    let report =
870      generate_report(task(), &good).expect("a readable cortex.log yields a verdict (Some)");
871    assert_eq!(
872      report.status,
873      TaskStatus::NoProblem,
874      "conversion:0 status line -> NoProblem"
875    );
876    std::fs::remove_file(&empty).ok();
877    std::fs::remove_file(&good).ok();
878  }
879
880  #[test]
881  fn tab_indented_details_fold_into_the_preceding_message() {
882    // A message header followed by tab-indented continuation lines: the details fold into the one
883    // preceding message (the path that used to run through the `.expect()`).
884    let log = "Error:math:bad first detail\n\tsecond line\n\tthird line\n";
885    let messages = parse_log(7, log);
886    assert_eq!(messages.len(), 1, "one message, the details folded into it");
887    let details = messages[0].details();
888    assert!(details.contains("first detail"), "got: {details:?}");
889    assert!(details.contains("second line"), "got: {details:?}");
890    assert!(details.contains("third line"), "got: {details:?}");
891  }
892
893  #[test]
894  fn orphan_details_line_does_not_panic() {
895    // A hostile log that opens with a tab-indented "details" line before any message header must be
896    // treated as noise, never panic the dispatch path (DESIGN_PRINCIPLES: no `.expect()` on a
897    // parse).
898    let messages = parse_log(
899      7,
900      "\torphan detail with no header\nInfo:cortex:hi a real line\n",
901    );
902    assert_eq!(
903      messages.len(),
904      1,
905      "the orphan line is ignored; the real message survives"
906    );
907  }
908
909  #[test]
910  fn result_archive_path_scopes_sandbox_outputs() {
911    use super::result_archive_path;
912    let entry = "/data/arxiv/1234/5678/source/5678.zip";
913    // Ordinary corpus: the historical path, unchanged (backward-compatible).
914    assert_eq!(
915      result_archive_path(entry, "tex_to_html", None).unwrap(),
916      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.zip")
917    );
918    // Sandbox (F-6): same source dir, but a corpus-id-scoped archive name, so a sandbox rerun can
919    // never overwrite the parent's `tex_to_html.zip`.
920    assert_eq!(
921      result_archive_path(entry, "tex_to_html", Some(42)).unwrap(),
922      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.sandbox-42.zip")
923    );
924    // A trailing newline (DB `text` columns carry them) is trimmed before taking the parent dir.
925    assert_eq!(
926      result_archive_path(&format!("{entry}\n"), "tex_to_html", None).unwrap(),
927      std::path::PathBuf::from("/data/arxiv/1234/5678/source/tex_to_html.zip")
928    );
929    // No parent directory (a blank/root entry) ⇒ no result path: the caller skips the write/serve
930    // instead of panicking.
931    assert!(result_archive_path("", "tex_to_html", None).is_none());
932    assert!(result_archive_path("/", "tex_to_html", None).is_none());
933  }
934}
935
936#[cfg(test)]
937mod rerun_mark_tests {
938  use super::rerun_mark;
939
940  #[test]
941  fn rerun_mark_stays_above_the_lease_space() {
942    // `fetch_tasks` stamps a lease as `1 + u16` (max 65536) and `TODO` is 0; the rerun sentinel
943    // must never land in `[0, 65536]`, or it could sweep a live in-flight lease (or alias `TODO`)
944    // into the rerun and double-dispatch it. All temp marks are `> 0`. KNOWN_ISSUES R-13.
945    for _ in 0..10_000 {
946      let mark = rerun_mark();
947      assert!(
948        mark > i32::from(u16::MAX) + 1,
949        "rerun mark {mark} overlaps the lease space [1, 65536]"
950      );
951      assert!(mark > 0, "temp marks must be positive");
952    }
953  }
954}
955
956#[cfg(test)]
957mod entry_name_tests {
958  use super::entry_document_name;
959
960  #[test]
961  fn extracts_short_document_name() {
962    // The motivating UX example: a download should be named by the entry, not the task id.
963    assert_eq!(
964      entry_document_name("/data/arxmliv/0811/0811.0417/0811.0417.zip"),
965      "0811.0417"
966    );
967    // Dotted arXiv ids keep their internal dot (the extension is the final segment).
968    assert_eq!(
969      entry_document_name("/data/foo/2105.13573.tar"),
970      "2105.13573"
971    );
972    // A trailing-whitespace entry (padded varchar) is trimmed first.
973    assert_eq!(entry_document_name("/d/bar.tex   "), "bar");
974    // No `/name.ext` shape -> falls back to the trimmed entry rather than erroring.
975    assert_eq!(entry_document_name("noslash"), "noslash");
976  }
977}
cortex/helpers.rs

cortex/
helpers.rs
