reusex::pipeline::JobRunner Class Reference

FIFO, single-worker, in-process pipeline job runner. More...

#include <JobRunner.hpp>

Public Member Functions
	JobRunner (std::filesystem::path project, StageExecutor executor=default_stage_executor())
	JobRunner (std::filesystem::path project, StageExecutor executor, JobRunnerOptions options)
	As above, with an explicit job-store policy.
	~JobRunner ()
	Requests cancellation of anything in flight and joins the worker.
	JobRunner (const JobRunner &)=delete
JobRunner &	operator= (const JobRunner &)=delete
const std::filesystem::path &	project () const noexcept
	The project every job of this runner targets.
std::string	submit (JobStage stage, std::string parameters={})
	Enqueue a stage run.
bool	cancel (std::string_view id)
	Request cancellation.
std::optional< JobRecord >	job (std::string_view id) const
	Snapshot of one job, or nullopt if the id is unknown or evicted.
std::vector< JobRecord >	jobs () const
	Snapshot of every retained job, most recently submitted first.
size_t	queued_count () const
	Number of jobs still waiting to start.
bool	is_busy () const
	True while a job is executing.
void	wait_idle ()
	Block until the queue is empty and no job is running.
WriterLease	try_acquire_writer (std::chrono::milliseconds timeout)
	Try to take the project's exclusive writer lock, giving up after timeout.
size_t	add_listener (JobListener listener)
	Register a listener. Returns a token for remove_listener().
void	remove_listener (size_t token)
	Unregister a listener. Safe to call with an unknown token.

Detailed Description

FIFO, single-worker, in-process pipeline job runner.

JOB LIFETIME: the store is in-memory and bounded (#286). Terminal jobs past JobRunnerOptions::max_terminal_jobs are evicted oldest-first by submission order, so job() returning nullopt means "unknown **or evicted**". Nothing survives a restart.

WHY NOT PERSIST: rux gui is an in-process server for a single project, and a job record is a view of work the process itself is doing — a queue position, a cancel token, a live progress counter. None of that is meaningful once the process is gone, and persisting it would make the runner the second writer of a history the pipeline already writes. The durable record is pipeline_log, which every stage writes and which carries the driving job id in pipeline_log.parameters.job_id; that join is the documented recovery path for a client holding an id the runner no longer knows. Durable job identity — jobs that outlive the worker executing them — belongs to Phase 6, where ruxd keeps them in PostgreSQL and one job can migrate between workers; adding a second, weaker persistence layer here first would only have to be undone.

Definition at line 173 of file JobRunner.hpp.

Constructor & Destructor Documentation

JobRunner() [1/3]

reusex::pipeline::JobRunner::JobRunner ( std::filesystem::path project, StageExecutor executor = default_stage_executor() )

explicit

Parameters

project	The .rux project every job of this runner operates on.
executor	Stage execution strategy; defaults to the real pipeline. Injecting a fake makes the state machine testable.

References project().

Referenced by JobRunner(), and operator=().

JobRunner() [2/3]

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

As above, with an explicit job-store policy.

References project().

~JobRunner()

reusex::pipeline::JobRunner::~JobRunner

(

)

Requests cancellation of anything in flight and joins the worker.

JobRunner() [3/3]

reusex::pipeline::JobRunner::JobRunner ( const JobRunner & )

delete

References JobRunner().

Member Function Documentation

add_listener()

size_t reusex::pipeline::JobRunner::add_listener

(

JobListener

listener

)

Register a listener. Returns a token for remove_listener().

cancel()

bool reusex::pipeline::JobRunner::cancel

(

std::string_view

id

)

Request cancellation.

• queued job -> immediately cancelled, never executed
• running job -> cancel token set; the stage stops at its next check
• terminal -> no-op

  Returns

  false if no job with that id exists.

is_busy()

bool reusex::pipeline::JobRunner::is_busy

(

)

const

True while a job is executing.

job()

std::optional< JobRecord > reusex::pipeline::JobRunner::job

(

std::string_view

id

)

const

Snapshot of one job, or nullopt if the id is unknown or evicted.

The two are indistinguishable here by design: the runner does not keep a tombstone for a job it has dropped. A caller that needs to tell them apart looks in pipeline_log, which is the durable record — every stage run is a row there, joined back to its job via pipeline_log.parameters.job_id.

jobs()

std::vector< JobRecord > reusex::pipeline::JobRunner::jobs

(

)

const

Snapshot of every retained job, most recently submitted first.

Terminal jobs beyond JobRunnerOptions::max_terminal_jobs have been evicted and do not appear; queued and running jobs always do.

operator=()

JobRunner & reusex::pipeline::JobRunner::operator= ( const JobRunner & )

delete

References JobRunner().

project()

const std::filesystem::path & reusex::pipeline::JobRunner::project ( ) const

noexcept

The project every job of this runner targets.

Referenced by JobRunner(), and JobRunner().

queued_count()

size_t reusex::pipeline::JobRunner::queued_count

(

)

const

Number of jobs still waiting to start.

remove_listener()

void reusex::pipeline::JobRunner::remove_listener

(

size_t

token

)

Unregister a listener. Safe to call with an unknown token.

submit()

std::string reusex::pipeline::JobRunner::submit	(	JobStage	stage,
		std::string	parameters = {} )

Enqueue a stage run.

Returns the new job id.

Parameters

parameters

JSON object string; "" means "stage defaults".

Exceptions

std::runtime_error

if parameters is not a JSON object.

References submit().

Referenced by submit().

try_acquire_writer()

WriterLease reusex::pipeline::JobRunner::try_acquire_writer ( std::chrono::milliseconds timeout )

nodiscard

Try to take the project's exclusive writer lock, giving up after timeout.

The worker holds this lock for the entire execution of a stage, so a caller that obtains it knows no stage is midway through writing. This is the mechanism that lets a second writer (the GUI's editor endpoints) exist at all without racing the pipeline: sqlite would serialize the two connections anyway, but only at statement granularity and only by returning SQLITE_BUSY — which cannot protect a read-modify-write such as renaming one entry of a whole-cloud label map.

Parameters

timeout

How long to wait. Keep it short on a request path: a running stage holds the lock for minutes, and answering "busy" promptly is far better than stalling the caller until it does.

Returns

A held lease, or an unheld one (!owns_lock()) on timeout.

Do not call this from a JobListener: the worker publishes events while holding the lock, so waiting on it from a listener would deadlock.

wait_idle()

void reusex::pipeline::JobRunner::wait_idle

(

)

Block until the queue is empty and no job is running.

Test and shutdown helper; do not call from a listener.

---

The documentation for this class was generated from the following file:

• libs/reusex/include/pipeline/JobRunner.hpp