Struct RedisBackend 

pub struct RedisBackend { /* private fields */ }

Production-grade Redis implementation of StorageBackend and StreamBackend.

Implementations

impl RedisBackend

pub async fn new(redis_url: impl AsRef<str>) -> Result<Self>

Creates a new RedisBackend from a Redis connection URL (e.g., "redis://127.0.0.1:6379").

pub fn client(&self) -> &Client

Returns reference to the underlying Redis Client.

Trait Implementations

impl Clone for RedisBackend

fn clone(&self) -> RedisBackend

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl StorageBackend for RedisBackend

fn as_stream(&self) -> Option<&dyn StreamBackend>

Returns reference to StreamBackend if supported by this storage implementation.

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

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

Executes backend schema migrations or setup steps.

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

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

Performs a health check to verify backend connectivity and readiness.

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

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

Enqueues a new job into the backend queue.

fn subscribe<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<NotificationStream>> + Send + 'async_trait>>

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

Subscribes to job enqueue notification events for a specific queue.

fn lease_jobs_batch<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queue: &'life1 str, worker_id: &'life2 str, lease_seconds: i64, batch_size: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>

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

Leases up to batch_size runnable jobs for a specified worker ID.

fn lease_jobs_batch_with_ordering<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queue: &'life1 str, worker_id: &'life2 str, lease_seconds: i64, batch_size: i64, ordering: QueueOrdering, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>

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

Leases up to batch_size runnable jobs with specified queue ordering preference (QueueOrdering).

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

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

Reaps expired locks from inactive workers, resetting their status back to queued.

fn start_attempts_batch<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _dataset_ids: &'life1 [String], job_ids: &'life2 [Uuid], _worker_id: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(Uuid, Uuid, i32)>>> + Send + 'async_trait>>

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

Starts job execution attempt records, returning (job_id, attempt_id, attempt_number) tuples.

fn mark_succeeded<'life0, 'life1, 'async_trait>( &'life0 self, job_id: Uuid, _attempt_id: Uuid, worker_id: &'life1 str, _latency_ms: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Marks a single job execution attempt as succeeded.

fn mark_succeeded_batch<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _dataset_id: &'life1 str, updates: &'life2 [(Uuid, Uuid, i32)], worker_id: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Marks a batch of job execution attempts as succeeded.

fn reschedule_for_retry<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, job_id: Uuid, _attempt_id: Uuid, worker_id: &'life1 str, _latency_ms: i32, next_run_at: DateTime<Utc>, _error_code: &'life2 str, _error_message: &'life3 str, _attempt_no: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Records a failed attempt and reschedules the job for a future retry attempt.

fn mark_dlq<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, job_id: Uuid, _attempt_id: Uuid, worker_id: &'life1 str, _latency_ms: i32, reason_code: &'life2 str, _error_code: &'life3 str, _error_message: &'life4 str, _attempt_no: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Records a failed attempt and transitions the job to the Dead-Letter Queue (DLQ).

fn archive_succeeded_older_than<'life0, 'async_trait>( &'life0 self, _cutoff: DateTime<Utc>, _limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

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

Moves succeeded jobs older than cutoff into an archive table or storage location.

fn delete_history_for_succeeded_older_than<'life0, 'async_trait>( &'life0 self, _cutoff: DateTime<Utc>, _limit: i64, ) -> Pin<Box<dyn Future<Output = Result<(u64, u64)>> + Send + 'async_trait>>

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

Prunes attempt audit logs and decision records for succeeded jobs older than cutoff.

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

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

Performs database maintenance (e.g., VACUUM ANALYZE in Postgres, PRAGMA incremental_vacuum in SQLite).

fn extend_lease<'life0, 'life1, 'async_trait>( &'life0 self, job_id: Uuid, worker_id: &'life1 str, lease_seconds: i64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

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

Extends the lease lock expiration for an in-flight running job. Returns true if the lease was extended, false if the job lock was lost or reaped.

fn get_job<'life0, 'async_trait>( &'life0 self, job_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>

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

Fetches a single job record by ID.

fn list_jobs<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queue: Option<&'life1 str>, status: Option<&'life2 str>, limit: i64, _cursor_created_at: Option<DateTime<Utc>>, _cursor_id: Option<Uuid>, ) -> Pin<Box<dyn Future<Output = Result<Vec<JobListItem>>> + Send + 'async_trait>>

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

Fetches a list of jobs matching filters with cursor pagination.

fn replay_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: Uuid, override_queue: Option<&'life1 str>, override_run_at: Option<DateTime<Utc>>, ) -> Pin<Box<dyn Future<Output = Result<Uuid>> + Send + 'async_trait>>

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

Atomically replays a job by ID into the queue.

fn dequeue_and_lease<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queue: &'life1 str, worker_id: &'life2 str, lease_seconds: i64, batch_size: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>, Error>> + Send + 'async_trait>>

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

Dequeues and leases up to batch_size runnable jobs (alias for lease_jobs_batch).

fn complete_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: Uuid, attempt_id: Uuid, worker_id: &'life1 str, latency_ms: i32, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

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

Marks a job attempt as completed (alias for mark_succeeded).

fn retry_job<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, job_id: Uuid, attempt_id: Uuid, worker_id: &'life1 str, latency_ms: i32, next_run_at: DateTime<Utc>, error_code: &'life2 str, error_message: &'life3 str, attempt_no: i32, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

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

Reschedules a job for retry (alias for reschedule_for_retry).

fn fail_job<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, job_id: Uuid, attempt_id: Uuid, worker_id: &'life1 str, latency_ms: i32, reason_code: &'life2 str, error_code: &'life3 str, error_message: &'life4 str, attempt_no: i32, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

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

Moves a job to DLQ on failure (alias for mark_dlq).

impl StreamBackend for RedisBackend

fn publish<'life0, 'life1, 'async_trait>( &'life0 self, stream: &'life1 str, event: NewEvent, ) -> Pin<Box<dyn Future<Output = Result<i64>> + Send + 'async_trait>>

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

Appends a new event to the specified stream log, returning its assigned sequence number.

fn subscribe_stream<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, stream: &'life1 str, _consumer_group: &'life2 str, _last_seq: Option<i64>, ) -> Pin<Box<dyn Future<Output = Result<NotificationStream>> + Send + 'async_trait>>

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

Subscribes to notification events when new entries are appended to a stream.

fn ack<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, stream: &'life1 str, consumer_group: &'life2 str, seq: i64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Acknowledges event processing up to seq for a consumer group on a stream log.

fn read_events<'life0, 'life1, 'async_trait>( &'life0 self, stream: &'life1 str, after_seq: i64, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Event>>> + Send + 'async_trait>>

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

Reads events from a stream with sequence numbers strictly greater than after_seq.

fn consumer_group_info<'life0, 'life1, 'async_trait>( &'life0 self, stream: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<ConsumerGroupStatus>>> + Send + 'async_trait>>

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

Fetches consumer group offset status for a stream log.