later 0.0.48

//! # later
//!
//! A distributed background job manager and runner for Rust.
//!
//! Later stores job state and delivers work through SQLite or PostgreSQL. Every server using the same database
//! and namespace joins the same worker pool. Renewable leases coordinate job
//! execution across workers and server processes.
//!
//! ## Documentation index
//!
//! - [Workers](#how-workers-run) - local worker count, shared worker pools,
//!   and delivery.
//! - [Storage and SQL data model](crate::storage) - durable records, tables,
//!   retention, and transaction boundaries.
//! - [Retry behavior](#retry-behavior) - per-message and server retry rules.
//! - [Dashboard](#dashboard) - feature-gated job inspection and retention.
//! - [Prometheus metrics](#prometheus-metrics) - feature-gated monitoring.
//! - [Recurring jobs](#recurring-jobs) - cron schedules, upsert-by-identifier,
//!   and the sequential (non-overlapping) mode.
//! - [Sequential partition mode](#sequential-partition-mode) - topic
//!   ordering, partition keys, ownership, and failure behaviour.
//! - [Retained logs](crate::retained_log) - replayable records and consumer
//!   group offsets using the same topic partitions.
//!
//! ## Retained logs
//!
//! Enable `retained-log` with `sqlite` or `postgres` to append immutable bytes
//! and replay them by partition offset. Consumer groups commit offsets and use
//! 30-second, epoch-fenced leases. This is a database-backed log API, not a
//! Kafka protocol implementation: there is no Kafka wire protocol, broker
//! replication, or cross-region quorum.
//!
//! ## Set up

//! ### 1. Import `later` and required dependencies
//!
//! ```toml
//! later = { version = "0.0.28", features = ["sqlite"] }
//! serde = { version = "1.0", features = ["derive"] }
//! ```
//!
//! ### 2. Define some types to use as a payload to the background jobs
//!
//! ```
//! use serde::{Deserialize, Serialize};
//!
//! #[derive(Serialize, Deserialize)] // <- Required derives
//! pub struct SendEmail {
//!     pub address: String,
//!     pub body: String,
//! }
//!
//! // ... more as required
//! ```
//!
//! ### 3. Generate the stub
//!
//! ```
//! # use serde::{Deserialize, Serialize};
//! #
//! # #[derive(Serialize, Deserialize)] // <- Required derives
//! # pub struct SendEmail {
//! #     pub address: String,
//! #     pub body: String,
//! # }
//! later::background_job! {
//!     struct Jobs {
//!         send_email: SendEmail,
//!     }
//! }
//! ```
//!
//! This generates two types
//! * `JobsBuilder` - used to bootstrap the background job server - which can be used to enqueue jobs,
//! * `JobContext<T>` - used to pass application context (`T`) in the handler as well as enqueue jobs,
//!
//! ### 4. Use the generated code to bootstrap the background job server
//!
//! For `struct Jobs` a type `JobsBuilder` will be generated. Use this to bootstrap the server.
//!
//! ```ignore
//! # use serde::{Deserialize, Serialize};
//! #
//! # #[derive(Serialize, Deserialize)]
//! # pub struct SendEmail { pub address: String, pub body: String }
//! # pub struct MyContext {}
//! # later::background_job! {
//! #     struct Jobs {
//! #         send_email: SendEmail,
//! #     }
//! # }
//! use later::{backend::SqliteBackend, storage::Sqlite, BackgroundJobServer, Config};
//!
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()> {
//! // bootstrap the server
//! let ctx = MyContext{ /*..*/ };                  // Any context to pass onto the handlers
//! let storage = Sqlite::new("sqlite://later.db").await?;
//! let backend = SqliteBackend::new("fnf-example", storage)?;
//! let ctx = JobsBuilder::new(
//!     later::Config::builder()
//!         .context(ctx)                       // Pass the context here
//!         .backend(Box::new(backend))          // Configure state and delivery
//!         // ...
//!         .build()
//!     )
//!     // for each payload defined in the `struct Jobs` above
//!     // the generated fn name uses the pattern "with_[name]_handler"
//!     .with_send_email_handler(handle_send_email)     // Pass the handler function
//!     // ..
//!     .build()
//!     .await?;
//!
//! // use ctx.enqueue(SendEmail{ ... }) to enqueue jobs,
//! // or ctx.enqueue_continue(parent_job_id, SendEmail{ ... }) to chain jobs.
//! // this will only accept types defined inside the macro above
//! # Ok(())
//! # }
//! // define handler
//! async fn handle_send_email(
//!         ctx: JobsContext<MyContext>, // JobContext is generated wrapper
//!         payload: SendEmail,
//!     ) -> anyhow::Result<()> {
//!     // handle `payload`
//!
//!     // ctx.app -> Access the MyContext passed during bootstrapping
//!     // ctx.enqueue(_).await to enqueue more jobs
//!     // ctx.enqueue_continue(_).await to chain jobs
//!
//!     Ok(()) // or Err(_) to retry this message
//! }
//! ```
//!
//! This example uses SQLite for state and delivery.
//!
//! ---
//!
//! ## How workers run
//!
//! A [`BackgroundJobServer`] starts six worker tasks by default. Set
//! [`Config::worker_count`] to change that number. The count applies to each
//! server instance, so three server processes configured with four workers
//! can run up to twelve job handlers at once. Every worker owns one delivery
//! consumer and processes one command at a time. A slow job occupies one worker
//! while the other workers continue receiving commands.
//! [`BackgroundJobServer::add_worker`] starts another local worker at runtime.
//! [`BackgroundJobServer::remove_worker`] waits for one worker's current job
//! before stopping it. At least one worker remains.
//! For process shutdown, call [`BackgroundJobServer::shutdown`] with a
//! deadline. It stops new claims, lets active handlers commit, and then
//! deregisters a sequential-processing owner. If the deadline expires,
//! remaining durable work recovers through its normal lease.
//!
//! Server startup waits until every worker consumer is ready. Workers using
//! the same backend database and namespace compete for work, including workers
//! in other processes or on other hosts. SQLite can coordinate several
//! processes only when they can all open the same database file. PostgreSQL
//! can coordinate processes across hosts.
//!
//! For each job delivery, a worker:
//!
//! 1. claims a renewable execution lease so another worker does not run the
//!    same job at the same time;
//! 2. loads the durable job and changes its stage from enqueued to running;
//! 3. calls the generated handler for the stored payload type;
//! 4. stores success, failure, or the next retry; and
//! 5. acknowledges the delivery. A delivery error is reported to the backend
//!    so it can be made available again.
//!
//! A delivery backend may make a command available more than once. The
//! execution lease prevents concurrent handler calls, and the stored job stage
//! makes workers ignore duplicate commands after a job has moved on from the
//! enqueued stage. A regular (non-partitioned) job whose handler is
//! interrupted after the running stage is stored - its worker crashed or was
//! killed mid-handler - is reclaimed by the periodic `PollStuckJobs`
//! maintenance command once its execution lease has been expired for a
//! while: it's retried (or marked failed, once retries are exhausted) the
//! same way a handler error would be. Partitioned/sequential jobs don't need
//! this - a stuck partition head already recovers through its own
//! partition-lease expiry.
//!
//! The same command also repairs jobs the dashboard index shows as `running`
//! that no lease tracks (left by older releases, or whose job record has
//! since expired): a missing record drops the index row, and an ordinary job
//! still `Running` is retried or failed like an abandoned one.
//!
//! Delayed jobs and retries do not occupy a sleeping worker. They remain in
//! durable storage until a periodic command finds that they are ready, then
//! they rejoin the delivery queue. Polling and queue load can add delay, so a
//! scheduled time or retry delay is a lower bound rather than an exact start
//! time.
//!
//! A separate periodic command, `PollExpiredStorage`, physically deletes
//! already-expired rows in small batches. Expiry (`Storage::expire`, and the
//! TTLs `later` sets internally on terminal jobs and dashboard bookkeeping)
//! only ever makes a row invisible to reads - something still has to delete
//! it, or storage only ever grows. This runs automatically; there is nothing
//! to configure.
//!
//! Keep the [`BackgroundJobServer`] handle alive for as long as this process
//! should accept work. Dropping it aborts its local worker and maintenance
//! tasks, including the two periodic commands described above.
//!
//! With the `dashboard` feature, each stage change updates job state and its
//! dashboard projection in the same atomic storage operation. Dashboard work
//! therefore cannot form a second delivery backlog during producer load.
//!
//! ## Fire and forget jobs
//!
//! Fire and forget jobs are made available to a worker immediately. A failed
//! handler is retried according to its resolved retry policy.
//!
//! ```no_run
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct SendEmail { pub address: String, pub body: String }
//! # later::background_job! {
//! #     struct Jobs {
//! #         send_email: SendEmail,
//! #     }
//! # }
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! ctx.enqueue(SendEmail{
//!     address: "hello@rust-lang.org".to_string(),
//!     body: "You rock!".to_string()
//! }).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Continuations
//!
//! One or many jobs can be chained into a workflow. A child becomes available
//! only after its parent succeeds.
//!
//! ```no_run
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct SendEmail { pub address: String, pub body: String }
//! # later::background_job! {
//! #     struct Jobs {
//! #         send_email: SendEmail,
//! #         create_account: CreateAccount,
//! #     }
//! # }
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct CreateAccount { id: String }
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! let email_welcome = ctx.enqueue(SendEmail{
//!     address: "customer@example.com".to_string(),
//!     body: "Creating your account!".to_string()
//! }).await?;
//!
//! let create_account = ctx.enqueue_continue(email_welcome, CreateAccount { id: "accout-1".to_string() }).await?;
//!
//! let email_confirmation = ctx.enqueue_continue(create_account, SendEmail{
//!     address: "customer@example.com".to_string(),
//!     body: "Your account has been created!".to_string()
//! }).await;
//! # Ok(())
//! # }
//! ```
//!
//! ## Delayed jobs
//!
//! A delayed job becomes available at or after a chosen interval or timestamp.
//!
//! ```no_run
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct SendEmail { pub address: String, pub body: String }
//! # later::background_job! {
//! #     struct Jobs {
//! #         send_email: SendEmail,
//! #     }
//! # }
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! // delay
//! ctx.enqueue_delayed(SendEmail{
//!     address: "hello@rust-lang.org".to_string(),
//!     body: "You rock!".to_string()
//! }, std::time::Duration::from_secs(60)).await?;
//!
//! // specific time
//! let run_job_at : chrono::DateTime<chrono::Utc> = todo!();
//! ctx.enqueue_delayed_at(SendEmail{
//!     address: "hello@rust-lang.org".to_string(),
//!     body: "You rock!".to_string()
//! }, run_job_at).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Recurring jobs
//!
//! Run a recurring job on a cron schedule. [`BackgroundJobServerPublisher::enqueue_recurring`]
//! registers a payload and cron expression under an `identifier`; calling it
//! again with the same `identifier` and cron updates the existing
//! definition in place (upsert) rather than creating a second schedule, and
//! returns `None` rather than enqueuing another occurrence - safe to call
//! on every process startup to recreate the schedule, which is the
//! intended pattern; calling it unconditionally on every startup was, once,
//! not safe, and would enqueue one more permanently-queued occurrence per
//! restart. Registering with a *different* cron is treated as a real
//! change and does enqueue a fresh occurrence, returning `Some`. A
//! background poller proactively enqueues each subsequent due occurrence,
//! independent of whether a previous occurrence's handler is still running
//! or ever ran — a crashed process cannot silently stop the schedule.
//!
//! ```no_run
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct SendNewsletter { pub address: String }
//! # later::background_job! {
//! #     struct Jobs {
//! #         send_newsletter: SendNewsletter,
//! #     }
//! # }
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! ctx.enqueue_recurring("send-newsletter-1".to_string(),
//!     SendNewsletter{
//!         address: "hello@rust-lang.org".to_string(),
//!     },
//!     "0 6 1 * * *".to_string() // 6am, 1st day of every month
//! ).await?;
//! # Ok(())
//! # }
//! ```
//!
//! By default occurrences run on the plain unordered path and may overlap,
//! the same as any other job. Use
//! [`BackgroundJobServerPublisher::enqueue_recurring_sequential`] instead
//! when at most one instance of a recurring job — including any jobs
//! chained off an occurrence with
//! [`BackgroundJobServerPublisher::enqueue_recurring_continue`] — may ever
//! be running at once. It reuses the sequential partition mode described
//! below: every occurrence, and every job chained off one, is enqueued into
//! the same topic partition (keyed by `identifier`), so they run in strict
//! order and never overlap. Requires
//! [`Config::recurring_sequential_partitions`] to be set.
//!
//! ```no_run
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct RunReport;
//! # later::background_job! {
//! #     struct Jobs {
//! #         run_report: RunReport,
//! #     }
//! # }
//! # async fn handle_run_report(ctx: JobsContext<()>, _payload: RunReport) -> anyhow::Result<()> {
//! // Chain a follow-up step off the currently-running occurrence itself,
//! // using its own job id. Because this must happen before the handler
//! // returns, the continuation is always ordered ahead of the next
//! // scheduled occurrence.
//! ctx.enqueue_recurring_continue(ctx.job_id().clone(), RunReport).await?;
//! # Ok(())
//! # }
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! ctx.enqueue_recurring_sequential("nightly-report".to_string(),
//!     RunReport,
//!     "0 0 3 * * *".to_string() // 3am daily
//! ).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Sequential partition mode
//!
//! Named topics with a fixed number of Kafka-like partitions give strict,
//! in-order execution within one `(topic, partition)`, while different
//! partitions and topics still run concurrently. Jobs enqueued without a
//! topic keep the default unordered, competing-worker behaviour described
//! above; nothing changes for them.
//!
//! A message type opts in by implementing [`topic::JobPartition`] and being
//! declared with `#[topic("name")]` inside [`background_job!`]. The macro
//! then requires that type to implement [`topic::JobPartition`] and
//! generates the [`topic::JobTopic`] marker the topic-aware enqueue methods
//! need. [`topic::JobPartition::partition_key`] returns an arbitrary
//! [`topic::PartitionKey`] — a customer ID, an order ID, anything that
//! identifies "things that must stay in order relative to each other" —
//! and Later hashes it to a partition with [`topic::partition_for_key`], the
//! way a Kafka producer key is hashed by the partitioner. The caller never
//! names a partition number directly. Register every topic's fixed
//! partition count on [`Config::topics`] before starting the server; an
//! unregistered topic is rejected before anything is written.
//!
//! ```
//! use later::topic::{JobPartition, PartitionKey};
//!
//! #[derive(serde::Serialize, serde::Deserialize)]
//! pub struct ProcessOrder {
//!     pub customer_id: u32,
//!     pub description: String,
//! }
//!
//! // Every order from the same customer must apply in order; different
//! // customers can usually be processed at the same time.
//! impl JobPartition for ProcessOrder {
//!     fn partition_key(&self) -> PartitionKey {
//!         PartitionKey::from(self.customer_id)
//!     }
//! }
//!
//! later::background_job! {
//!     struct Jobs {
//!         #[topic("orders")]
//!         process_order: ProcessOrder,
//!     }
//! }
//! ```
//!
//! Register the topic when building the server, and use the partition-aware
//! enqueue methods for that message type:
//!
//! ```no_run
//! # use later::topic::{JobPartition, PartitionKey};
//! # #[derive(serde::Serialize, serde::Deserialize)]
//! # pub struct ProcessOrder { pub customer_id: u32, pub description: String }
//! # impl JobPartition for ProcessOrder {
//! #     fn partition_key(&self) -> PartitionKey { PartitionKey::from(self.customer_id) }
//! # }
//! # later::background_job! {
//! #     struct Jobs {
//! #         #[topic("orders")]
//! #         process_order: ProcessOrder,
//! #     }
//! # }
//! use later::topic::TopicConfig;
//!
//! # #[tokio::main]
//! # async fn main() -> anyhow::Result<()>{
//! # let context = ();
//! # let backend: Box<dyn later::backend::Backend> = todo!();
//! let config = later::Config::builder()
//!     .context(context)
//!     .backend(backend)
//!     .topics(vec![TopicConfig::new("orders", 8)?])
//!     .build();
//! # let _ = config;
//! # let ctx : later::BackgroundJobServerPublisher = todo!();
//! // Only a type declared with a topic can use the partition-aware methods.
//! ctx.enqueue_to_partition(ProcessOrder {
//!     customer_id: 42,
//!     description: "add item".to_string(),
//! }).await?;
//! # Ok(())
//! # }
//! ```
//!
//! [`BackgroundJobServerPublisher::enqueue_to_partition`],
//! [`BackgroundJobServerPublisher::enqueue_to_partition_delayed`], and
//! [`BackgroundJobServerPublisher::enqueue_continue_to_partition`] mirror
//! [`BackgroundJobServerPublisher::enqueue`],
//! [`BackgroundJobServerPublisher::enqueue_delayed`], and
//! [`BackgroundJobServerPublisher::enqueue_continue`], but assign the job a
//! monotonic sequence
//! within its `(topic, partition)` as part of the same atomic write. The
//! SQLite and PostgreSQL backends also expose `enqueue_to_partition_in` for
//! writing application data, the job, and its sequence in one caller-owned
//! transaction, matching `enqueue_in`.
//!
//! Ordering rules:
//!
//! - The oldest unfinished job in a partition blocks every later job in that
//!   partition, including a retry (which keeps its position), a delayed
//!   head (which blocks until its scheduled time), and a waiting
//!   continuation (which blocks until its parent succeeds, even when the
//!   parent is not itself partitioned or lives in a different partition).
//! - A terminal success or failure releases the next sequence.
//! - Different partitions, and different topics, never block each other,
//!   even when they share the same partition ID.
//!
//! Every server with configured topics registers a heartbeat and runs a
//! bounded number of partition poller tasks (one per worker) that lease and
//! run the oldest ready, unowned head. Which server owns a given
//! `(topic, partition)` is decided by rendezvous hashing over every
//! currently live server (heartbeats expire after 15 seconds), so a small
//! membership change only moves the partitions whose winner changes rather
//! than reshuffling every assignment. A lease keeps one worker's claim
//! exclusive across every process sharing the same database and namespace;
//! losing a race for a lease is normal and just means another worker already
//! owns that partition's head. The lease renews itself while a handler
//! keeps running, so a slow handler does not let another worker reclaim and
//! re-run the same head concurrently. As with the default delivery path, a
//! worker crash can still run the current head job again after its lease
//! expires: order is preserved, but execution is at-least-once rather than
//! exactly-once.
//!
//! Each partition claim also has a fencing epoch. If renewal shows that a
//! lease was lost, Later cancels the handler future and rejects that claim's
//! retry or completion write. Only the owner holding the current epoch can
//! advance the stored partition head. This protects Later's own state during
//! a process pause or a reclaimed lease. It cannot undo an external side
//! effect that began before cancellation, so handlers that call external
//! systems must still be idempotent, normally using the job ID as their key.
//!
//! Adding a worker lets it start claiming newly idle or unassigned
//! partitions within a couple of membership refresh cycles, without waiting
//! for any lease to expire. A job already running when its partition's
//! assignment changes finishes under its current owner; the new owner only
//! takes over once that worker stops claiming the partition's next head.
//! Dropping a [`BackgroundJobServer`] deregisters its worker immediately
//! (best-effort) so other workers stop considering it right away; an
//! ungraceful removal (a crash) is only detected once its heartbeat
//! expires.
//!
//! SQL is always the source of order and ownership. A poller also reacts
//! immediately (instead of waiting out its idle backoff) to a local enqueue
//! or continuation. This wake-up is a pure latency optimization; nothing
//! depends on it arriving, since every poller still scans SQL on its own.
//!
//! If a head job would otherwise block its partition indefinitely (a
//! "poison" job whose retries never succeed, or one you simply need
//! unblocked sooner),
//! [`BackgroundJobServerPublisher::force_fail_partition_head`] moves it
//! straight to a terminal failed state and frees the partition, without
//! waiting out its retry policy and without needing to hold its lease
//! first.
//!
//! See
//! `<https://github.com/mustakimali/later/blob/main/plans/SEQUENTIAL_PARTITION_MODE.md>`
//! for what is planned next.
//!
//! ## Backends
//!
//! | State | Default delivery | Scale | Features |
//! | --- | --- | --- | --- |
//! | SQLite | SQLite | Several processes sharing one local file | `sqlite` |
//! | PostgreSQL | PostgreSQL | Processes on many hosts | `postgres` |
//! Give every backend a namespace. Servers that use the same database and
//! namespace share work; different namespaces stay isolated. The SQL backends
//! also accept an application-owned SQLx pool, which allows application data
//! and a job to be written in the same transaction with `enqueue_in`.
//!
//! ## Retry behavior
//!
//! The default server policy retries six times with exponential backoff and
//! bounded jitter. Override it for the whole server through [`Config`]:
//!
//! ```no_run
//! # let context = ();
//! # let backend: Box<dyn later::backend::Backend> = todo!();
//! use later::{retry::RetryPolicy, Config};
//! use std::time::Duration;
//!
//! let config = Config::builder()
//!     .context(context)
//!     .backend(backend)
//!     .default_retry_policy(RetryPolicy::fixed(3, Duration::from_secs(5)))
//!     .build();
//! ```
//!
//! A message type can take priority over that setting:
//!
//! ```
//! use later::retry::{JobRetryPolicy, RetryPolicy};
//!
//! struct AppContext;
//! struct SendEmail;
//!
//! #[later::async_trait::async_trait]
//! impl JobRetryPolicy for SendEmail {
//!     type Context = AppContext;
//!
//!     async fn retry_policy(&self, _context: &AppContext) -> anyhow::Result<RetryPolicy> {
//!         Ok(RetryPolicy::no_retries())
//!     }
//! }
//! ```
//!
//! ## Dashboard
//! Enable feature `dashboard` to enable the experimental dashboard. The host
//! application is responsible for protecting its route. Dashboard metadata is
//! retained for one day and does not include job payload bytes. Mount the
//! response returned by `BackgroundJobServerPublisher::get_dashboard` in the
//! HTTP framework used by the application.
//!
//! The bundled page also shows stage-transition throughput from the last 60
//! seconds and workers with a heartbeat in the last 15 seconds, plus average
//! and maximum queue-dispatch wait time over that same window - how long a
//! job sat enqueued/delayed/requeued before a worker picked it up, tracked
//! separately for regular jobs and sequential/partitioned ones (a
//! sequential job's wait includes time blocked behind earlier jobs in its
//! partition, so the two are not directly comparable). This is the
//! overhead `later` itself adds before a handler starts, not time spent
//! inside the handler. These stats are stored only when the `dashboard`
//! feature is enabled - no `prometheus` feature needed. Each job response
//! includes total elapsed, handler-processing, and waiting durations computed
//! from its stage history. Job details link to a continuation's parent and to
//! at most 25 jobs that directly continue from it. The jobs table shows only
//! the last 6 characters of each ID (the full ID is still in a hover tooltip
//! and used for the detail link) - later's IDs are lexicographically
//! sortable ULIDs, so the trailing characters are what actually varies
//! between IDs minted close together.
//!
//! Every topic in [`Config::topics`] also gets a badge in the topic menu
//! showing its queue depth (not-yet-completed jobs across every
//! partition), refreshed on the same cycle as everything else on the page,
//! plus a per-partition breakdown in the partition picker once a topic is
//! selected - no separate opt-in needed. Set [`Config::retained_log_lag`]
//! (needs the `retained-log` feature too) to add each configured
//! retained-log topic's consumer lag to the same badges - see
//! [`RetainedLogLag`].
//!
//! The performance panel also shows the entire database's on-disk size (not
//! just Later's own tables) when the storage backend can report one -
//! SQLite and PostgreSQL both can; a backend without a single meaningful
//! database size (Redis, in-memory) shows a dash instead. This is the whole
//! database an application shares with Later, useful for keeping an eye on
//! total footprint even when most of it is application data.
//!
//! ### How the dashboard stays fast at any backlog
//!
//! **Queued jobs are not mirrored.** On SQLite the `enqueued` stage is read
//! from the delivery queue itself - the same rows workers claim from - not
//! from a job-index row per queued job (four b-trees written per enqueue, a
//! million rows for a million-job backlog). Its size is a counter that
//! triggers on the queue table keep exact in the same transaction as each
//! publish, claim, release and delete; the list is a cursor over the queue's
//! integer key, so a page costs the page however deep the queue. A job gets
//! an index row when it starts running, and loses it if it returns to the
//! queue. Jobs on a topic partition are not in the shared queue and stay in
//! the index (their backlog is the topic's own view). PostgreSQL keeps
//! indexing queued jobs: a shared counter row there would serialise
//! concurrent claimers.
//!
//! **The other live stages are counters too** (`delayed`, `waiting`,
//! `running`, `requeued`): triggers on the job index adjust them in the same
//! transaction as each row change, so reading them is a handful of single-row
//! reads. They follow the index's rows rather than each job's previous stage,
//! so a job written twice in the same stage, or a write the revision guard
//! rejects, cannot skew them. A recount at startup seeds both kinds of
//! counter for rows that predate the triggers, in one transaction. Finished
//! stages (`success`, `failed`) are bounded by their expiry and counted from
//! the index.
//!
//! The `count` command computes one snapshot at a time and serves it for a
//! second, so many open dashboards cost one computation. At startup a
//! background task also removes, in small batches, rows earlier dashboard
//! designs left behind (the all-jobs list, per-stage lists of queued and
//! running jobs, the old `stats-*` keys, and index rows for queued jobs).
//!
//! The Monitoring tab's **Queue length** graph is drawn from a per-minute
//! reading of the shared queue's length and the partitions' combined backlog,
//! kept for a week on the server, so it shows a backlog draining over hours
//! and survives a page reload. The plan for the remaining scaling work is in
//! `plans/DASHBOARD_SCALING.md`.
//!
//! A partitioned topic's dashboard history (the list backing its
//! per-partition "partition jobs" view) is the same `later_jobs_index` row
//! every other dashboard view reads, so it shares that row's own terminal-
//! job expiry rather than needing a separate retention setting - or call
//! [`BackgroundJobServerPublisher::truncate_partition_history`] to trim an
//! already-accumulated history down to its most recent N entries in one
//! shot regardless of expiry.
//!
//! ## Prometheus metrics
//!
//! Enable `prometheus` to add process-local worker metrics. Expose the text
//! returned by `BackgroundJobServerPublisher::get_metrics` on an HTTP route
//! that your Prometheus server can scrape. The feature records:
//!
//! - `later_job_transitions_total` for durable stage changes;
//! - `later_job_handler_duration_seconds` for handler latency and outcome per
//!   worker;
//! - `later_job_duration_seconds` and `later_job_wait_duration_seconds` for
//!   completed job workflow latency;
//! - `later_queue_jobs` for durable ready, leased, delayed, retry-waiting, and
//!   dependency-waiting job counts;
//! - `later_worker_commands_total` for successful, failed, and discarded
//!   delivery commands per worker;
//! - `later_workers` for each active and busy worker task; and
//! - for servers with configured topics: `later_partition_claims_total`
//!   (claim attempts, whether or not a head was found),
//!   `later_partition_heads_claimed_total` and `later_partitions_active` by
//!   topic, `later_partition_lease_renewals_total` by topic and outcome,
//!   `later_partition_workers_live` for this process's last-observed live
//!   worker count, `later_partition_backlog` /
//!   `later_partition_oldest_head_age_seconds` by topic for the ready
//!   backlog and how long its oldest head has been waiting, and
//!   `later_partition_queue_depth` by topic and partition for not-yet-completed
//!   jobs currently queued there; and
//! - with `retained-log` also enabled and [`Config::retained_log_lag`] set:
//!   `later_retained_log_consumer_lag` by topic and consumer group, for
//!   records appended but not yet committed by that group.
//!
//! Worker metrics use a numeric `worker_id` that is local to one server.
//! Combine it with Prometheus's scrape `instance` label to distinguish workers
//! running in different processes. Metrics never use a job ID, error message,
//! or partition worker owner ID as a label — only the configured topic name,
//! which stays bounded the way `queue` and `job_type` already are. The one
//! exception is `later_retained_log_consumer_lag`'s `group` label: it is
//! whatever name the application commits offsets under, not a fixed set
//! `later` controls, so keep the number of distinct consumer group names
//! small the same way you would for any other Prometheus label. Values
//! cover only the current process. Prometheus combines several server
//! processes when a query sums their scraped series.
//! Queue depth is refreshed from shared storage every five seconds. Each
//! server reports the same shared value, so use `max` rather than `sum` across
//! Prometheus scrape instances.
#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]
use crate::core::BgJobHandler;

use mq::MqPublisher;
use persist::Persist;
use serde::{Deserialize, Serialize};
use std::{fmt::Display, marker::PhantomData, sync::Arc};
use tokio::task::JoinHandle;
use typed_builder::TypedBuilder;

pub use anyhow;
pub use async_trait;
pub use futures;
pub use later_derive::background_job;
pub use tracing::instrument;

pub mod backend;
mod bg_job_server;
mod bg_job_server_publisher;
mod commands;
/// Traits used by job payloads and generated job handlers.
pub mod core;
/// MessagePack encoding helpers used by Later's wire format.
pub mod encoder;
mod id;
#[cfg(any(feature = "dashboard", feature = "prometheus"))]
mod job_timing;
#[cfg(feature = "prometheus")]
mod metrics;
mod models;
/// Low-level job delivery interfaces and built-in delivery implementations.
pub mod mq;
mod partition;
mod persist;
#[cfg(feature = "retained-log")]
/// Replayable records stored in the existing topic partitions.
pub mod retained_log;
/// Retry limits and delay strategies for failed jobs.
pub mod retry;
#[cfg(feature = "dashboard")]
/// Dashboard response types.
pub mod stats;
/// Low-level state storage interfaces and built-in implementations.
pub mod storage;
/// Named topics and Kafka-like partitions for strictly ordered jobs.
pub mod topic;

/// A UTC timestamp used in persisted job records.
pub type UtcDateTime = chrono::DateTime<chrono::Utc>;

/// The stable identifier assigned to one job execution.
///
/// Display the value to store it in logs or links. Applications receive IDs
/// from enqueue methods and should otherwise treat them as opaque.
#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq)]
pub struct JobId(String);
impl Display for JobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// The caller-selected identifier for a recurring job definition.
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct RecurringJobId(String);
impl Display for RecurringJobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

// ToDo: Remove H - use Box<dyn BgJobHandler<C>>
/// A running job server and its worker tasks.
///
/// The server dereferences to [`BackgroundJobServerPublisher`], so it can also
/// enqueue jobs. [`BackgroundJobServer::add_worker`] and
/// [`BackgroundJobServer::remove_worker`] change this server's concurrency at
/// runtime. Removing a worker lets its current job finish before it exits.
/// Call [`BackgroundJobServer::shutdown`] from a process shutdown handler to
/// drain active work before exit. Dropping the server aborts its local worker
/// and maintenance tasks; durable jobs remain in the configured backend.
pub struct BackgroundJobServer<C, H>
where
    C: Sync + Send + 'static,
    H: BgJobHandler<C> + Sync + Send + 'static,
{
    ctx: PhantomData<C>,
    handler: Arc<H>,
    mq_client: Arc<Box<dyn mq::MqClient>>,
    inproc_cmd_tx: async_std::channel::Sender<models::ChannelCommand>,
    /// `Arc`-wrapped (not just `Mutex`-wrapped) so `supervise_workers` - a
    /// maintenance task spawned during [`BackgroundJobServer::start`],
    /// before this struct exists to hand out `&self` - can hold its own
    /// clone and keep replacing dead or hung workers directly.
    worker_tasks: Arc<std::sync::Mutex<Vec<bg_job_server::WorkerTask>>>,
    worker_scaling: tokio::sync::Mutex<()>,
    /// `Arc`-wrapped for the same reason as `worker_tasks` - shared with
    /// `supervise_workers`, spawned before this struct exists, so a
    /// respawned worker's ID stays unique alongside anything `add_worker`
    /// creates later.
    next_worker_id: Arc<std::sync::atomic::AtomicU32>,
    #[cfg(feature = "dashboard")]
    server_instance_id: String,
    maintenance_tasks: Vec<JoinHandle<anyhow::Result<()>>>,
    /// Set when this server has configured topics; the owner ID registered
    /// in `later_worker`, deregistered best-effort on drop so a graceful
    /// shutdown frees its partitions without waiting for heartbeat expiry.
    partition_owner: Option<String>,
    partition_membership_task: Option<JoinHandle<anyhow::Result<()>>>,
    partition_tasks: std::sync::Mutex<Vec<JoinHandle<anyhow::Result<()>>>>,
    partition_stop: Option<tokio::sync::watch::Sender<bool>>,
}

/// Enqueues jobs and exposes dashboard and metrics data.
///
/// Applications normally get this through the server type generated by
/// [`background_job!`], rather than constructing it directly.
pub struct BackgroundJobServerPublisher {
    publisher: Box<dyn MqPublisher>,
    routing_key: String,
    storage: Arc<Persist>,
    committer: Arc<dyn backend::JobCommitter>,
    default_retry_policy: retry::RetryPolicy,
    topics: Vec<topic::TopicConfig>,
    /// Wakes local partition pollers as soon as work appears, instead of
    /// making them wait out their idle backoff. Notified after this
    /// process's own enqueue/continuation paths commit. Purely a latency optimization: nothing
    /// depends on a notification arriving, since the poller's own periodic
    /// scan already covers a missed or duplicate wake-up.
    partition_wake: Arc<tokio::sync::Notify>,
    #[cfg(feature = "dashboard")]
    stats: stats::Stats,
    retained_log_lag: RetainedLogLag,
    /// Copied from [`Config::vacuum_interval`]; read by the maintenance
    /// backstop loop (`bg_job_server::start_bg_worker_system_ops_ensure_ops_run_on_certain_interval`)
    /// to decide whether/how often to enqueue `ChannelCommand::VacuumDatabase`.
    vacuum_interval: Option<std::time::Duration>,
}

/// Generates a lexicographically sortable ULID string.
pub fn generate_id() -> String {
    rusty_ulid::generate_ulid_string().to_lowercase()
}

#[cfg(test)]
mod generate_id_tests {
    use super::generate_id;

    #[test]
    fn generate_id_is_lowercase() {
        let id = generate_id();

        assert_eq!(id, id.to_lowercase());
        assert!(id.chars().all(|c| !c.is_ascii_uppercase()));
    }

    #[test]
    fn lowercasing_a_ulid_preserves_its_lexicographic_order() {
        // Two real ULIDs differing only in their last character: uniformly
        // lowercasing both must not change which one sorts first.
        let earlier = "01ARZ3NDEKTSV4RRFFQ69G5FAV";
        let later = "01ARZ3NDEKTSV4RRFFQ69G5FAW";

        assert!(earlier < later);
        assert!(earlier.to_lowercase() < later.to_lowercase());
    }
}

/// Bounds for [`Config::worker_autoscale`].
///
/// Every 15 seconds, a maintenance task compares the regular delivery
/// queue's current backlog (ready-to-claim messages, via
/// [`crate::mq::MqClient::queue_depth`]) against the live worker count and
/// adjusts by at most one worker per tick, using
/// [`BackgroundJobServer::add_worker`]/[`BackgroundJobServer::remove_worker`]
/// directly - the same methods available for manual scaling, so autoscaling
/// and a manual call never fight over how many workers should exist, only
/// over when to change it.
///
/// Scales up when the backlog exceeds roughly 20 messages per already-live
/// worker (so a burst is absorbed within a couple of ticks rather than
/// immediately maxing out on a brief spike) and there's room below `max`.
/// Scales down after the backlog has read as empty for three consecutive
/// ticks (about 45 seconds) and there's room above `min` - the delay avoids
/// tearing a worker down moments before the next burst arrives. One worker
/// changed per tick either way, so a large swing in load takes several
/// ticks to fully reflect, deliberately - it trades reaction speed for
/// never overshooting a spike into `max` workers at once.
#[derive(Clone, Copy, Debug)]
pub struct WorkerAutoscale {
    /// Never scales down below this many workers, even at zero backlog.
    pub min: u8,
    /// Never scales up beyond this many workers, regardless of backlog.
    pub max: u8,
}

/// Settings used by a generated job server builder.
///
/// Construct this with [`Config::builder`]. A backend selects both durable
/// state and delivery. The application context is shared immutably with every
/// handler.
#[derive(TypedBuilder)]
pub struct Config<C>
where
    C: Sync + Send + 'static,
{
    /// Application state exposed to each generated job context.
    pub context: C,
    /// The state and delivery backend used by publishers and workers.
    pub backend: Box<dyn backend::Backend>,

    /// Retry count used with the built-in server policy.
    ///
    /// This remains the simple way to change only the retry limit. Setting
    /// `default_retry_policy` replaces both this count and the built-in delay.
    #[builder(default = 6)]
    pub default_retry_count: u8,
    /// Policy used when the message type has no
    /// [`retry::JobRetryPolicy`] implementation.
    ///
    /// When omitted, `default_retry_count` is combined with exponential
    /// backoff and bounded jitter.
    #[builder(default, setter(strip_option))]
    pub default_retry_policy: Option<retry::RetryPolicy>,
    /// Number of worker tasks started by this server. Defaults to six and must
    /// be at least one.
    ///
    /// Each worker processes one delivery at a time. This count applies per
    /// server instance, so the total concurrency grows when more servers using
    /// the same database and namespace are started.
    #[builder(default = 6)]
    pub worker_count: u8,

    /// Automatically adds and removes regular (non-partitioned) workers
    /// between `min` and `max`, based on the regular delivery queue's own
    /// backlog - see [`WorkerAutoscale`]. `None` (the default) keeps the
    /// fixed `worker_count` behavior unchanged.
    ///
    /// Only affects the regular worker pool `worker_count` sizes - topic
    /// partition pollers (one per worker slot, for `Config::topics`) are
    /// unaffected either way, and load-balance across however many workers
    /// exist automatically already (any idle worker can claim any ready
    /// delivery - see [`crate::backend::JobCommitter`]), so there's nothing
    /// separate for autoscaling to route.
    #[builder(default, setter(strip_option))]
    pub worker_autoscale: Option<WorkerAutoscale>,

    /// Named topics available for partitioned enqueue on this server.
    ///
    /// A topic must be registered here before
    /// [`BackgroundJobServerPublisher::enqueue_to_partition`] can use it.
    /// Registering the same topic with a different partition count than a
    /// previous run is rejected; changing a topic's partition count needs an
    /// explicit migration.
    #[builder(default)]
    pub topics: Vec<topic::TopicConfig>,

    /// Enables [`BackgroundJobServerPublisher::enqueue_recurring_sequential`]
    /// and sets the partition count for the reserved topic its occurrences
    /// share.
    ///
    /// A sequential recurring job's identifier is hashed into one of these
    /// partitions, the same way [`Config::topics`] partitions are hashed
    /// from an application-supplied key — two recurring jobs whose
    /// identifiers hash to the same partition serialize against each other
    /// too, so raise this if many distinct sequential recurring jobs are
    /// registered.
    ///
    /// `None` (the default) leaves the reserved topic unregistered, the same
    /// as any other unconfigured [`Config::topics`] entry: partition
    /// worker-membership and polling only start when at least one topic is
    /// registered, so servers that don't use sequential recurring jobs pay
    /// no cost for this feature. Calling
    /// [`BackgroundJobServerPublisher::enqueue_recurring_sequential`]
    /// without setting this returns an error.
    #[builder(default, setter(strip_option))]
    pub recurring_sequential_partitions: Option<u32>,

    /// Reports retained-log consumer lag in the dashboard's topic menu and
    /// the `later_retained_log_consumer_lag` Prometheus gauge. See
    /// [`RetainedLogLag`]. Defaults to disabled - retained-log and the job
    /// queue are independent subsystems (see [`retained_log`]), so this is
    /// the one place they connect, and only when a server actually uses
    /// both.
    #[builder(default)]
    pub retained_log_lag: RetainedLogLag,

    /// How often a background maintenance task reclaims disk space freed by
    /// expired rows (`DELETE` alone only marks pages free for SQLite's own
    /// reuse - it never shrinks the file on disk).
    ///
    /// `None` (the default) never vacuums automatically, matching every
    /// prior release - existing applications see no new behavior on upgrade.
    /// `Some(duration)` runs a full `VACUUM` (SQLite only; other backends
    /// ignore this) at most once every `duration`, tracked the same way the
    /// other maintenance polls track their own last-run time, so it survives
    /// a restart rather than firing again immediately. `VACUUM` rebuilds the
    /// entire database file and needs an exclusive lock on the connection
    /// running it for its duration - pick an interval that fits how large
    /// the database gets between runs, and expect a multi-second pause on a
    /// database that has grown to gigabytes.
    #[builder(default, setter(strip_option))]
    pub vacuum_interval: Option<std::time::Duration>,
}

/// Wires an optional [`retained_log::LogLagSource`] into a [`Config`],
/// gated by the `retained-log` feature so [`Config`] itself needs no
/// feature flag. Build with [`RetainedLogLag::new`], or use
/// [`Default::default`] to leave lag reporting disabled.
#[derive(Clone, Default)]
pub struct RetainedLogLag {
    #[cfg(feature = "retained-log")]
    topics: Vec<String>,
    #[cfg(feature = "retained-log")]
    source: Option<std::sync::Arc<dyn retained_log::LogLagSource>>,
}

#[cfg(feature = "retained-log")]
impl RetainedLogLag {
    /// Reports lag for every partition of each of `topics`, read from
    /// `source` (typically `Arc::new(your_sqlite_or_postgres_log)` - both
    /// [`retained_log::SqliteLog`] and [`retained_log::PostgresLog`]
    /// implement [`retained_log::LogLagSource`]).
    ///
    /// `topics` need not match [`Config::topics`] (job-queue partition
    /// ordering) - a retained-log topic registered with its own
    /// `register_topic` may have no relation to any job-queue topic.
    pub fn new(
        topics: Vec<String>,
        source: std::sync::Arc<dyn retained_log::LogLagSource>,
    ) -> Self {
        Self {
            topics,
            source: Some(source),
        }
    }

    pub(crate) fn source(&self) -> Option<&std::sync::Arc<dyn retained_log::LogLagSource>> {
        self.source.as_ref()
    }

    pub(crate) fn topics(&self) -> &[String] {
        &self.topics
    }
}

/// Runtime settings consumed by [`BackgroundJobServer::start`].
pub struct ServerConfig {
    /// Retry count copied from [`Config`] for compatibility with direct users.
    /// `default_retry_policy` is the policy applied to newly enqueued jobs.
    pub default_retry_count: u8,
    /// Policy stored when jobs are enqueued. A message policy may replace it
    /// after the first handler failure.
    pub default_retry_policy: retry::RetryPolicy,
    /// Number of worker tasks started by this server. Must be at least one.
    pub worker_count: u8,
    /// Copied from [`Config::worker_autoscale`].
    pub worker_autoscale: Option<WorkerAutoscale>,
    /// Named topics available for partitioned enqueue on this server.
    pub topics: Vec<topic::TopicConfig>,
    /// Copied from [`Config::retained_log_lag`].
    pub retained_log_lag: RetainedLogLag,
    /// Copied from [`Config::vacuum_interval`].
    pub vacuum_interval: Option<std::time::Duration>,
}

impl<C: Send + Sync> Config<C> {
    /// Copies worker and retry settings into their runtime representation.
    pub fn to_server_config(&self) -> ServerConfig {
        let default_retry_policy = self.default_retry_policy.clone().unwrap_or_else(|| {
            let default = retry::RetryPolicy::default();
            retry::RetryPolicy::new(self.default_retry_count.into(), default.backoff().clone())
        });
        let mut topics = self.topics.clone();
        if let Some(partitions) = self.recurring_sequential_partitions {
            topics.push(
                topic::TopicConfig::new(topic::RECURRING_SEQUENTIAL_TOPIC, partitions.max(1))
                    .expect(
                        "reserved recurring topic name is always valid, partition count is nonzero",
                    ),
            );
        }
        ServerConfig {
            default_retry_count: self.default_retry_count,
            default_retry_policy,
            worker_count: self.worker_count,
            worker_autoscale: self.worker_autoscale,
            topics,
            retained_log_lag: self.retained_log_lag.clone(),
            vacuum_interval: self.vacuum_interval,
        }
    }

    /// Splits the configuration into the values needed to start a server.
    ///
    /// This validates and consumes the configured [`backend::Backend`], then
    /// registers this server's configured topics with it.
    pub async fn into_runtime(self) -> anyhow::Result<(C, ServerConfig, backend::BackendParts)> {
        let server_config = self.to_server_config();
        let parts = self.backend.into_parts()?;
        parts
            .committer
            .register_topics(&server_config.topics)
            .await?;
        Ok((self.context, server_config, parts))
    }
}

#[cfg(test)]
mod config_tests {
    use super::*;
    use std::time::Duration;

    struct UnusedBackend;

    impl backend::Backend for UnusedBackend {
        fn into_parts(self: Box<Self>) -> anyhow::Result<backend::BackendParts> {
            Err(anyhow::anyhow!("unused test backend"))
        }
    }

    #[test]
    fn server_retry_policy_replaces_or_uses_the_simple_retry_count() {
        let with_count = Config::builder()
            .context(())
            .backend(Box::new(UnusedBackend) as Box<dyn backend::Backend>)
            .default_retry_count(2)
            .build()
            .to_server_config();
        let with_policy = Config::builder()
            .context(())
            .backend(Box::new(UnusedBackend) as Box<dyn backend::Backend>)
            .default_retry_policy(retry::RetryPolicy::fixed(4, Duration::from_secs(3)))
            .build()
            .to_server_config();

        assert_eq!(
            (
                with_count.default_retry_policy.max_retries(),
                with_count.default_retry_policy.backoff(),
                with_policy.default_retry_policy.max_retries(),
                with_policy.default_retry_policy.backoff(),
            ),
            (
                2,
                retry::RetryPolicy::default().backoff(),
                4,
                &retry::RetryBackoff::Fixed {
                    delay: Duration::from_secs(3),
                },
            )
        );
    }
}