JobRunner.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2026 Povl Filip Sonne-Frederiksen
//
// SPDX-License-Identifier: GPL-3.0-or-later
 
#pragma once
 
// In-process pipeline job runner (#265, Phase 1).
//
// Owns a FIFO queue of stage runs and a single worker thread, bridges the
// global core::IProgressObserver singleton into per-job progress events, and
// exposes a submit / status / cancel surface that both `rux gui` (today) and
// ruxd (Phase 6) can put behind the same HTTP contract.
//
// WHY ONE WORKER: progress reporting goes through the process-global observer
// registered with core::set_progress_observer(). Two stages running
// concurrently would interleave into a single observer with no way to tell
// their events apart, so jobs are serialized. The stages themselves are
// TBB-parallel internally and already saturate the machine.
//
// THREAD SAFETY: every public member is safe to call from any thread. The
// ProjectDB instance used to execute a job is created and destroyed on the
// worker thread and never escapes it — ProjectDB itself is not thread-safe.
//
// WRITER EXCLUSION: the runner also owns the project's *writer lock* (see
// try_acquire_writer). The worker holds it for the whole execution of a stage,
// so anything else that wants to write to the same `.rux` — the GUI's editor
// endpoints, for instance — can take the same lock and be genuinely exclusive
// with the pipeline rather than merely hoping sqlite sorts it out.
 
#include "reusex/core/stages.hpp"
#include "reusex/pipeline/stages.hpp"
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <filesystem>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
 
namespace reusex::pipeline {

enum class JobStatus {
  queued,
  running,
  succeeded,
  failed,
  cancelled,
};

std::string_view to_string(JobStatus status);

std::optional<JobStatus> parse_job_status(std::string_view name);

bool is_terminal(JobStatus status);

std::string iso8601_utc_now();

struct JobRecord {
  std::string id; 
  JobStage stage = JobStage::clouds;
  JobStatus status = JobStatus::queued;
  std::string parameters;   
  std::string error;        
  std::string submitted_at; 
  std::string started_at;   
  std::string finished_at;  
  bool cancel_requested = false;

  core::Stage progress_stage = core::Stage::idle;
  size_t progress_current = 0;
  size_t progress_total = 0; 

  std::string result_summary;
  std::vector<StageArtifact> result_outputs;
};

struct JobEvent {
  uint64_t sequence = 0;
 
  enum class Type {
    submitted, 
    started,   
    progress,  
    finished,  
  };
 
  Type type = Type::submitted;
  std::string timestamp; 
  JobRecord job;
};

std::string_view to_string(JobEvent::Type type);

using JobListener = std::function<void(const JobEvent &)>;

using WriterLease = std::unique_lock<std::timed_mutex>;

struct JobRunnerOptions {
  size_t max_terminal_jobs = 256;
};

class JobRunner {
    public:
  explicit JobRunner(std::filesystem::path project,
                     StageExecutor executor = default_stage_executor());

  JobRunner(std::filesystem::path project, StageExecutor executor,
            JobRunnerOptions options);

  ~JobRunner();
 
  JobRunner(const JobRunner &) = delete;
  JobRunner &operator=(const JobRunner &) = delete;

  const std::filesystem::path &project() const noexcept;

  std::string submit(JobStage stage, std::string parameters = {});

  bool cancel(std::string_view id);

  std::optional<JobRecord> job(std::string_view id) const;

  std::vector<JobRecord> jobs() const;

  size_t queued_count() const;

  bool is_busy() const;

  void wait_idle();

  [[nodiscard]] WriterLease
  try_acquire_writer(std::chrono::milliseconds timeout);

  size_t add_listener(JobListener listener);

  void remove_listener(size_t token);
 
    private:
  class Impl;
  std::unique_ptr<Impl> impl_;
};
 
} // namespace reusex::pipeline