Struct JobRepository 

pub struct JobRepository { /* private fields */ }

Implementations

impl JobRepository

pub fn new(pool: PgPool) -> Self

pub fn pool(&self) -> &PgPool

pub async fn enqueue( &self, input: NewJob, ) -> Result<EnqueueOutcome, DomainError>

Insert a job. When an idempotency_key is set and a non-terminal (or succeeded) row already exists for (kind, key), returns the existing row rather than inserting a duplicate.

pub async fn append_log( &self, job_id: JobId, attempt_no: Option<i32>, level: &str, message: &str, fields: Value, ) -> Result<JobLogRecord, DomainError>

pub async fn list_logs( &self, id: JobId, after_id: Option<i64>, limit: u32, ) -> Result<Vec<JobLogRecord>, DomainError>

pub async fn dequeue( &self, queues: &[String], worker_id: &str, lease: Duration, ) -> Result<Option<SbolJob>, DomainError>

Dequeue one job from the given queues, taking a lease for the worker. Uses FOR UPDATE SKIP LOCKED so concurrent workers across the cluster never contend on the same row.

pub async fn renew_lease( &self, job_id: JobId, worker_id: &str, lease: Duration, ) -> Result<bool, DomainError>

Extend the lease on a running job. Returns false when the lease has already been stolen (different leased_by) or the job has transitioned out of running.

pub async fn mark_succeeded( &self, job_id: JobId, worker_id: &str, result: Option<Value>, ) -> Result<(), DomainError>

pub async fn mark_failed( &self, job_id: JobId, worker_id: &str, error: &str, ) -> Result<JobStatus, DomainError>

Mark a failed attempt. If the job still has retries left, transition it back to queued with exponential backoff (60s * 2^attempts, capped at 1 hour). Otherwise it lands in dead.

pub async fn reap_expired_leases(&self) -> Result<u64, DomainError>

Re-queue jobs whose lease has expired. Workers run this on a slow timer; one losing the race is harmless — UPDATE ... WHERE status = 'running' is idempotent under concurrent reapers.

pub async fn get(&self, id: JobId) -> Result<Option<SbolJob>, DomainError>

pub async fn list_attempts( &self, id: JobId, ) -> Result<Vec<JobAttempt>, DomainError>

Per-attempt audit log for a job, newest first. Returns an empty vector when the job exists but has not been dequeued yet, and also when the job id does not exist — callers that care about existence should pair this with Self::get.

pub async fn list( &self, filter: &ListJobsFilter, ) -> Result<Vec<SbolJob>, DomainError>

pub async fn cancel(&self, id: JobId) -> Result<bool, DomainError>

Cancel a job. Only queued jobs cancel cleanly; running jobs are marked cancelled but the worker still finishes its current step (cooperative cancellation is a v2 concern).

pub async fn current_status( &self, id: JobId, ) -> Result<Option<JobStatus>, DomainError>

Used by handler dispatch and the worker’s heartbeat: peek the status without taking a lease.

pub async fn queue_depth_snapshot( &self, ) -> Result<Vec<QueueDepthRow>, DomainError>

Aggregate row count grouped by (status, queue) for the non-terminal lifecycle states. Drives the /metrics sbol_db_jobs_queue_depth gauge. Cheap given the partial dequeue index covers status = 'queued'.

pub async fn oldest_queued_age( &self, ) -> Result<Vec<OldestQueuedAge>, DomainError>

Oldest available_at per queue among queued rows whose available_at <= now(). Lets ops alert on stalled queues without chasing individual job ids.

pub async fn recent_failure_count( &self, within_hours: i64, ) -> Result<i64, DomainError>

Trait Implementations

impl Clone for JobRepository

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl JobQueue for JobRepository

fn enqueue<'life0, 'async_trait>( &'life0 self, input: NewJob, ) -> Pin<Box<dyn Future<Output = Result<EnqueueOutcome, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn append_log<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: JobId, attempt_no: Option<i32>, level: &'life1 str, message: &'life2 str, fields: Value, ) -> Pin<Box<dyn Future<Output = Result<JobLogRecord, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn list_logs<'life0, 'async_trait>( &'life0 self, id: JobId, after_id: Option<i64>, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<JobLogRecord>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn dequeue<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queues: &'life1 [String], worker_id: &'life2 str, lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<Option<SbolJob>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn renew_lease<'life0, 'life1, 'async_trait>( &'life0 self, job_id: JobId, worker_id: &'life1 str, lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<bool, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn mark_succeeded<'life0, 'life1, 'async_trait>( &'life0 self, job_id: JobId, worker_id: &'life1 str, result: Option<Value>, ) -> Pin<Box<dyn Future<Output = Result<(), DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn mark_failed<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: JobId, worker_id: &'life1 str, error: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<JobStatus, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn reap_expired_leases<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<u64, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get<'life0, 'async_trait>( &'life0 self, id: JobId, ) -> Pin<Box<dyn Future<Output = Result<Option<SbolJob>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn list_attempts<'life0, 'async_trait>( &'life0 self, id: JobId, ) -> Pin<Box<dyn Future<Output = Result<Vec<JobAttempt>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn list<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ListJobsFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<SbolJob>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn cancel<'life0, 'async_trait>( &'life0 self, id: JobId, ) -> Pin<Box<dyn Future<Output = Result<bool, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn current_status<'life0, 'async_trait>( &'life0 self, id: JobId, ) -> Pin<Box<dyn Future<Output = Result<Option<JobStatus>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn queue_depth_snapshot<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<QueueDepthRow>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn oldest_queued_age<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<OldestQueuedAge>, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn recent_failure_count<'life0, 'async_trait>( &'life0 self, within_hours: i64, ) -> Pin<Box<dyn Future<Output = Result<i64, DomainError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Count jobs that reached a terminal failure (failed or dead) in the last within_hours. Backed by Self::list so every backend reports the same number without backend-specific SQL; the dashboard’s failures tile reads it.