feat: per-subscription backlog and lag metrics

crates/taskito-core/src/pubsub.rs

@@ -116,6 +116,18 @@ fn delivery_job(request: &PublishRequest, sub: &SubscriptionRow) -> NewJob {
     }
 }
 
+/// Inverse of [`delivery_notes`]: pull `topic`/`subscription` back out of a
+/// job's notes JSON when both are present. Feeds the indexed
+/// `jobs.topic`/`jobs.subscription_name` columns (and their `dead_letter`
+/// mirrors) at insert time, so backlog/lag aggregation runs off an index
+/// instead of a JSON scan — without adding pub/sub-only fields to `NewJob`.
+pub fn extract_topic_subscription(notes: Option<&str>) -> Option<(String, String)> {
+    let obj: serde_json::Map<String, serde_json::Value> = serde_json::from_str(notes?).ok()?;
+    let topic = obj.get("topic")?.as_str()?.to_string();
+    let subscription = obj.get("subscription")?.as_str()?.to_string();
+    Some((topic, subscription))
+}
+
 /// Stamp `topic` and `subscription` into the caller's notes object so every
 /// delivery is filterable per subscriber without a schema change.
 fn delivery_notes(request: &PublishRequest, sub: &SubscriptionRow) -> String {

crates/taskito-core/src/storage/diesel_common/dead_letter.rs

@@ -25,6 +25,11 @@ macro_rules! impl_diesel_dead_letter_ops {
                     .and_then(|m| serde_json::from_str::<serde_json::Value>(m).ok())
                     .and_then(|v| v.get("__dlq_retry_count")?.as_i64())
                     .unwrap_or(0) as i32;
+                // Carry the delivery's subscription attribution into the DLQ so
+                // per-subscription dead-letter depth is countable.
+                let (topic, subscription_name) =
+                    $crate::pubsub::extract_topic_subscription(job.notes.as_deref())
+                        .map_or((None, None), |(t, s)| (Some(t), Some(s)));
 
                 self.write_transaction(|conn| {
                     let dlq_row = NewDeadLetterRow {
@@ -46,6 +51,8 @@ macro_rules! impl_diesel_dead_letter_ops {
                         result_ttl_ms: job.result_ttl_ms,
                         namespace: job.namespace.as_deref(),
                         dlq_retry_count,
+                        topic: topic.as_deref(),
+                        subscription_name: subscription_name.as_deref(),
                     };
 
                     diesel::insert_into(dead_letter::table)
@@ -190,6 +197,11 @@ macro_rules! impl_diesel_dead_letter_ops {
                 };
 
                 let job = new_job.into_job();
+                // Re-attribute the resurrected delivery so it counts against the
+                // subscription's backlog again, not just its DLQ depth.
+                let (topic, subscription_name) =
+                    $crate::pubsub::extract_topic_subscription(job.notes.as_deref())
+                        .map_or((None, None), |(t, s)| (Some(t), Some(s)));
 
                 let result = conn.transaction(|conn| {
                     let row = super::super::models::NewJobRow {
@@ -212,6 +224,8 @@ macro_rules! impl_diesel_dead_letter_ops {
                         result_ttl_ms: job.result_ttl_ms,
                         namespace: job.namespace.as_deref(),
                         has_deps: job.has_deps,
+                        topic: topic.as_deref(),
+                        subscription_name: subscription_name.as_deref(),
                     };
 
                     diesel::insert_into(jobs::table)

crates/taskito-core/src/storage/diesel_common/jobs.rs

@@ -162,6 +162,11 @@ macro_rules! impl_diesel_job_ops {
                         Self::validate_dependency(conn, dep_id)?;
                     }
 
+                    // Attribute pub/sub deliveries to their subscription so
+                    // backlog stats index by it; `None` for ordinary jobs.
+                    let (topic, subscription_name) =
+                        $crate::pubsub::extract_topic_subscription(job.notes.as_deref())
+                            .map_or((None, None), |(t, s)| (Some(t), Some(s)));
                     let row = NewJobRow {
                         id: &job.id,
                         queue: &job.queue,
@@ -182,6 +187,8 @@ macro_rules! impl_diesel_job_ops {
                         result_ttl_ms: job.result_ttl_ms,
                         namespace: job.namespace.as_deref(),
                         has_deps: job.has_deps,
+                        topic: topic.as_deref(),
+                        subscription_name: subscription_name.as_deref(),
                     };
 
                     diesel::insert_into(jobs::table)
@@ -224,10 +231,21 @@ macro_rules! impl_diesel_job_ops {
 
                 let jobs: Vec<Job> = new_jobs.into_iter().map(|nj| nj.into_job()).collect();
 
+                // Pre-compute subscription attribution so the owned strings
+                // outlive the borrowing rows built below.
+                let attribution: Vec<(Option<String>, Option<String>)> = jobs
+                    .iter()
+                    .map(|job| {
+                        $crate::pubsub::extract_topic_subscription(job.notes.as_deref())
+                            .map_or((None, None), |(t, s)| (Some(t), Some(s)))
+                    })
+                    .collect();
+
                 self.write_transaction(|conn| {
                     let rows: Vec<NewJobRow> = jobs
                         .iter()
-                        .map(|job| NewJobRow {
+                        .zip(&attribution)
+                        .map(|(job, (topic, subscription_name))| NewJobRow {
                             id: &job.id,
                             queue: &job.queue,
                             task_name: &job.task_name,
@@ -247,6 +265,8 @@ macro_rules! impl_diesel_job_ops {
                             result_ttl_ms: job.result_ttl_ms,
                             namespace: job.namespace.as_deref(),
                             has_deps: job.has_deps,
+                            topic: topic.as_deref(),
+                            subscription_name: subscription_name.as_deref(),
                         })
                         .collect();
 
@@ -299,6 +319,9 @@ macro_rules! impl_diesel_job_ops {
                             Self::validate_dependency(conn, dep_id)?;
                         }
 
+                        let (topic, subscription_name) =
+                            $crate::pubsub::extract_topic_subscription(job.notes.as_deref())
+                                .map_or((None, None), |(t, s)| (Some(t), Some(s)));
                         let row = NewJobRow {
                             id: &job.id,
                             queue: &job.queue,
@@ -319,6 +342,8 @@ macro_rules! impl_diesel_job_ops {
                             result_ttl_ms: job.result_ttl_ms,
                             namespace: job.namespace.as_deref(),
                             has_deps: job.has_deps,
+                            topic: topic.as_deref(),
+                            subscription_name: subscription_name.as_deref(),
                         };
 
                         diesel::insert_into(jobs::table)

crates/taskito-core/src/storage/diesel_common/migrations.rs

@@ -325,6 +325,13 @@ pub fn create_indexes() -> &'static [&'static str] {
         // the partial index keeps durable rows (owner NULL) out of the reaper's way.
         "CREATE INDEX IF NOT EXISTS idx_topic_subs_owner
                 ON topic_subscriptions(owner_worker_id) WHERE owner_worker_id IS NOT NULL",
+        // Backlog/lag aggregation groups pub/sub deliveries by (topic, subscription,
+        // status). Partial so only the small set of pub/sub jobs is indexed — the
+        // dashboard query never touches the mass of ordinary jobs.
+        "CREATE INDEX IF NOT EXISTS idx_jobs_subscription_status
+                ON jobs(topic, subscription_name, status) WHERE subscription_name IS NOT NULL",
+        "CREATE INDEX IF NOT EXISTS idx_dead_letter_subscription
+                ON dead_letter(topic, subscription_name) WHERE subscription_name IS NOT NULL",
     ]
 }
 
@@ -396,6 +403,14 @@ pub fn alter_statements(d: &Dialect) -> Vec<String> {
         format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}priority INTEGER"),
         format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}max_retries INTEGER"),
         format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}timeout_ms {bi}"),
+        // Pub/sub delivery attribution: publish stamps the topic + subscription
+        // onto each delivery job (and its dead_letter row) so backlog/lag stats
+        // aggregate off an index instead of scanning the notes JSON. NULL for
+        // every non-pub/sub job, so the partial indexes stay small.
+        format!("ALTER TABLE jobs ADD COLUMN {ife}topic TEXT"),
+        format!("ALTER TABLE jobs ADD COLUMN {ife}subscription_name TEXT"),
+        format!("ALTER TABLE dead_letter ADD COLUMN {ife}topic TEXT"),
+        format!("ALTER TABLE dead_letter ADD COLUMN {ife}subscription_name TEXT"),
     ]
 }
 

crates/taskito-core/src/storage/diesel_common/pubsub.rs

@@ -92,6 +92,64 @@ macro_rules! impl_diesel_pubsub_ops {
 
                 Ok(affected as u64)
             }
+
+            /// Backlog/lag snapshot per registered subscription. Four bounded,
+            /// index-backed queries (subscription skeleton + pending/running
+            /// counts + oldest-pending age + dead counts) merged in Rust — the
+            /// same live+aggregate shape as `stats_all_queues`, never a full
+            /// table scan. Counts are always a direct read of real state, so
+            /// they cannot drift the way a maintained counter would.
+            pub fn topic_backlog_stats(
+                &self,
+            ) -> Result<Vec<$crate::storage::SubscriptionBacklogStats>> {
+                use $crate::job::JobStatus;
+                use $crate::storage::schema::{dead_letter, jobs};
+                let mut conn = self.conn()?;
+
+                let subs = topic_subscriptions::table
+                    .select($crate::storage::models::SubscriptionRow::as_select())
+                    .load::<$crate::storage::models::SubscriptionRow>(&mut conn)?;
+
+                // Pending + running counts, scoped to pub/sub-tagged rows only.
+                let counts: Vec<(String, String, i32, i64)> = jobs::table
+                    .filter(jobs::subscription_name.is_not_null())
+                    .filter(
+                        jobs::status.eq_any([JobStatus::Pending as i32, JobStatus::Running as i32]),
+                    )
+                    .group_by((jobs::topic, jobs::subscription_name, jobs::status))
+                    .select((
+                        jobs::topic.assume_not_null(),
+                        jobs::subscription_name.assume_not_null(),
+                        jobs::status,
+                        diesel::dsl::count(jobs::id),
+                    ))
+                    .load(&mut conn)?;
+
+                let oldest: Vec<(String, String, Option<i64>)> = jobs::table
+                    .filter(jobs::subscription_name.is_not_null())
+                    .filter(jobs::status.eq(JobStatus::Pending as i32))
+                    .group_by((jobs::topic, jobs::subscription_name))
+                    .select((
+                        jobs::topic.assume_not_null(),
+                        jobs::subscription_name.assume_not_null(),
+                        diesel::dsl::min(jobs::created_at),
+                    ))
+                    .load(&mut conn)?;
+
+                let dead: Vec<(String, String, i64)> = dead_letter::table
+                    .filter(dead_letter::subscription_name.is_not_null())
+                    .group_by((dead_letter::topic, dead_letter::subscription_name))
+                    .select((
+                        dead_letter::topic.assume_not_null(),
+                        dead_letter::subscription_name.assume_not_null(),
+                        diesel::dsl::count(dead_letter::id),
+                    ))
+                    .load(&mut conn)?;
+
+                Ok($crate::storage::merge_backlog_stats(
+                    subs, counts, oldest, dead,
+                ))
+            }
         }
     };
 }

crates/taskito-core/src/storage/mod.rs

@@ -42,6 +42,88 @@ pub struct QueueStats {
     pub cancelled: i64,
 }
 
+/// Per-subscription backlog/lag snapshot for the pub/sub dashboard. One entry
+/// per *registered* subscription — durable or ephemeral, active or paused —
+/// even at zero backlog, so the dashboard renders the full subscriber list
+/// from a single call. Counts are computed live off the delivery-attribution
+/// indexes; they can never drift the way a maintained counter would.
+#[derive(Debug, Clone)]
+pub struct SubscriptionBacklogStats {
+    pub topic: String,
+    pub subscription_name: String,
+    pub task_name: String,
+    pub queue: String,
+    pub active: bool,
+    pub durable: bool,
+    pub pending: i64,
+    pub running: i64,
+    pub dead: i64,
+    /// Milliseconds since the oldest still-pending delivery was created.
+    /// `None` when the subscription currently has no pending backlog.
+    pub oldest_pending_age_ms: Option<i64>,
+}
+
+/// Merge the four `topic_backlog_stats` aggregate queries into one row per
+/// registered subscription. Seeds a zeroed entry per subscription so idle
+/// subscriptions still appear, then folds in the pending/running counts,
+/// oldest-pending age (converted to a millisecond age), and dead counts.
+/// Shared by the Diesel and Redis backends.
+pub(crate) fn merge_backlog_stats(
+    subs: Vec<models::SubscriptionRow>,
+    counts: Vec<(String, String, i32, i64)>,
+    oldest: Vec<(String, String, Option<i64>)>,
+    dead: Vec<(String, String, i64)>,
+) -> Vec<SubscriptionBacklogStats> {
+    use crate::job::{now_millis, JobStatus};
+    use std::collections::HashMap;
+
+    let mut by_key: HashMap<(String, String), SubscriptionBacklogStats> = subs
+        .into_iter()
+        .map(|s| {
+            (
+                (s.topic.clone(), s.subscription_name.clone()),
+                SubscriptionBacklogStats {
+                    topic: s.topic,
+                    subscription_name: s.subscription_name,
+                    task_name: s.task_name,
+                    queue: s.queue,
+                    active: s.active,
+                    durable: s.durable,
+                    pending: 0,
+                    running: 0,
+                    dead: 0,
+                    oldest_pending_age_ms: None,
+                },
+            )
+        })
+        .collect();
+
+    for (topic, name, status, count) in counts {
+        if let Some(stats) = by_key.get_mut(&(topic, name)) {
+            if status == JobStatus::Pending as i32 {
+                stats.pending = count;
+            } else if status == JobStatus::Running as i32 {
+                stats.running = count;
+            }
+        }
+    }
+
+    let now = now_millis();
+    for (topic, name, oldest_created) in oldest {
+        if let (Some(stats), Some(created)) = (by_key.get_mut(&(topic, name)), oldest_created) {
+            stats.oldest_pending_age_ms = Some((now - created).max(0));
+        }
+    }
+
+    for (topic, name, count) in dead {
+        if let Some(stats) = by_key.get_mut(&(topic, name)) {
+            stats.dead = count;
+        }
+    }
+
+    by_key.into_values().collect()
+}
+
 #[derive(Debug, Clone)]
 pub struct DeadJob {
     pub id: String,
@@ -387,6 +469,11 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<u64> {
                 self.reap_ephemeral_subscriptions(live_worker_ids)
             }
+            fn topic_backlog_stats(
+                &self,
+            ) -> $crate::error::Result<Vec<$crate::storage::SubscriptionBacklogStats>> {
+                self.topic_backlog_stats()
+            }
             fn record_metric(
                 &self,
                 task_name: &str,
@@ -950,6 +1037,9 @@ impl Storage for StorageBackend {
     fn reap_ephemeral_subscriptions(&self, live_worker_ids: &[String]) -> Result<u64> {
         delegate!(self, reap_ephemeral_subscriptions, live_worker_ids)
     }
+    fn topic_backlog_stats(&self) -> Result<Vec<SubscriptionBacklogStats>> {
+        delegate!(self, topic_backlog_stats)
+    }
     fn record_metric(
         &self,
         task_name: &str,

crates/taskito-core/src/storage/models.rs

@@ -93,6 +93,10 @@ pub struct NewJobRow<'a> {
     pub result_ttl_ms: Option<i64>,
     pub namespace: Option<&'a str>,
     pub has_deps: bool,
+    /// Set on pub/sub deliveries (derived from `notes`) so backlog/lag stats
+    /// index by subscription; `None` for ordinary jobs.
+    pub topic: Option<&'a str>,
+    pub subscription_name: Option<&'a str>,
 }
 
 /// A row in the `dead_letter` table.
@@ -137,6 +141,8 @@ pub struct NewDeadLetterRow<'a> {
     pub result_ttl_ms: Option<i64>,
     pub namespace: Option<&'a str>,
     pub dlq_retry_count: i32,
+    pub topic: Option<&'a str>,
+    pub subscription_name: Option<&'a str>,
 }
 
 /// A row in the `rate_limits` table.