feat: expose topic_stats in the Node and Java shells

CHANGELOG.md

@@ -27,6 +27,10 @@ underlying Rust crates are released together, in lock-step.
 
 - **Node `request` resource scope.** A fresh instance on every `useResource()` call, each
   disposed when the task ends — matching the Java scope of the same name.
+- **Pub/sub backlog stats in the Node and Java SDKs.** `queue.topicStats(topic?)` (Node) and
+  `topicStats()` / `topicStats(topic)` (Java) return the per-subscription snapshot Python has
+  as `topic_stats()`: `pending`, `running`, `dead`, and `oldestPendingAgeMs`, with a row for
+  every registered subscription even at zero backlog.
 - Job outcome events report how long the task ran: `durationMs()` on Java's `OutcomeEvent`,
   `durationMs` on Node's, and `duration_ms` on Python's job event payloads. Java also gains
   `NodeSnapshot.durationMs()` / `compensationDurationMs()` and `TaskContext.elapsedMs()`.

crates/taskito-java/src/convert.rs

@@ -12,7 +12,7 @@ use taskito_core::storage::records::{
     CircuitBreakerState, JobError, LockInfo, PeriodicTask, ReplayEntry, Subscription, TaskLogEntry,
     TaskMetric, Topic, TopicLogStats, TopicMessage, WorkerInfo,
 };
-use taskito_core::storage::{DeadJob, QueueStats};
+use taskito_core::storage::{DeadJob, QueueStats, SubscriptionBacklogStats};
 
 use crate::error::BindingError;
 
@@ -294,6 +294,41 @@ impl From<&TopicLogStats> for TopicLogStatsView {
     }
 }
 
+/// Java-facing backlog snapshot for one topic subscription. One entry per
+/// *registered* subscription — paused and ephemeral ones included — even at zero
+/// backlog. `oldest_pending_age_ms` is null when nothing is pending.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TopicStatsView<'a> {
+    pub topic: &'a str,
+    pub subscription: &'a str,
+    pub task_name: &'a str,
+    pub queue: &'a str,
+    pub active: bool,
+    pub durable: bool,
+    pub pending: i64,
+    pub running: i64,
+    pub dead: i64,
+    pub oldest_pending_age_ms: Option<i64>,
+}
+
+impl<'a> From<&'a SubscriptionBacklogStats> for TopicStatsView<'a> {
+    fn from(s: &'a SubscriptionBacklogStats) -> Self {
+        Self {
+            topic: &s.topic,
+            subscription: &s.subscription_name,
+            task_name: &s.task_name,
+            queue: &s.queue,
+            active: s.active,
+            durable: s.durable,
+            pending: s.pending,
+            running: s.running,
+            dead: s.dead,
+            oldest_pending_age_ms: s.oldest_pending_age_ms,
+        }
+    }
+}
+
 /// Java-facing view of a declared topic. `retention_ms` is null when the backlog
 /// is kept until consumed; `created_at` is Unix milliseconds.
 #[derive(Serialize)]

crates/taskito-java/src/queue/pubsub.rs

@@ -15,7 +15,7 @@ use taskito_core::Storage;
 use crate::backend;
 use crate::convert::{
     build_publish_request, parse_json, to_json, JobView, PublishOptions, SubscriptionView,
-    TopicLogStatsView, TopicMessageView, TopicView,
+    TopicLogStatsView, TopicMessageView, TopicStatsView, TopicView,
 };
 use crate::ffi::{guard, new_string, read_bytes, read_optional_string, read_string};
 
@@ -154,6 +154,24 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_setSubscri
     })
 }
 
+/// `String topicBacklogStats(long handle)` — a JSON array of `TopicStatsView`,
+/// one backlog snapshot per registered subscription (even at zero backlog).
+/// Counts are aggregated live off the delivery-attribution indexes, so they can
+/// never drift the way a maintained counter would.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_topicBacklogStats<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let stats = queue.storage.topic_backlog_stats()?;
+        let views: Vec<TopicStatsView> = stats.iter().map(TopicStatsView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
 /// `long reapEphemeralSubscriptions(long handle)` — drop ephemeral subscriptions
 /// whose owning worker is gone; returns the count removed.
 #[no_mangle]

crates/taskito-node/src/convert/mod.rs

@@ -23,8 +23,8 @@ pub use ops::{
 pub use outcome::{outcome_to_js, JsOutcome};
 pub use periodic::{periodic_to_js, JsPeriodicTask};
 pub use pubsub::{
-    subscription_to_js, topic_log_stat_to_js, topic_message_to_js, topic_to_js, JsSubscription,
-    JsTopic, JsTopicLogStat, JsTopicMessage,
+    subscription_to_js, topic_log_stat_to_js, topic_message_to_js, topic_stat_to_js, topic_to_js,
+    JsSubscription, JsTopic, JsTopicLogStat, JsTopicMessage, JsTopicStat,
 };
 pub use stats::{
     dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js,

crates/taskito-node/src/convert/pubsub.rs

@@ -3,6 +3,7 @@
 use napi::bindgen_prelude::Buffer;
 use napi_derive::napi;
 use taskito_core::storage::records::{Subscription, Topic, TopicLogStats, TopicMessage};
+use taskito_core::storage::SubscriptionBacklogStats;
 
 /// A topic subscription: routes messages published to `topic` to `taskName`
 /// jobs on `queue`, one delivery per active subscription.
@@ -28,6 +29,43 @@ pub fn subscription_to_js(row: Subscription) -> JsSubscription {
     }
 }
 
+/// Backlog snapshot for one topic subscription. One entry per *registered*
+/// subscription — durable or ephemeral, active or paused — even at zero
+/// backlog, so a caller renders the full subscriber list from a single call.
+#[napi(object)]
+pub struct JsTopicStat {
+    pub topic: String,
+    pub subscription: String,
+    pub task_name: String,
+    pub queue: String,
+    pub active: bool,
+    pub durable: bool,
+    /// Deliveries waiting to run.
+    pub pending: i64,
+    /// Deliveries currently executing.
+    pub running: i64,
+    /// Deliveries in the dead-letter queue.
+    pub dead: i64,
+    /// Age (ms) of the oldest still-pending delivery, absent at zero backlog.
+    pub oldest_pending_age_ms: Option<i64>,
+}
+
+/// Convert a core [`SubscriptionBacklogStats`] into its JS-facing shape.
+pub fn topic_stat_to_js(stat: SubscriptionBacklogStats) -> JsTopicStat {
+    JsTopicStat {
+        topic: stat.topic,
+        subscription: stat.subscription_name,
+        task_name: stat.task_name,
+        queue: stat.queue,
+        active: stat.active,
+        durable: stat.durable,
+        pending: stat.pending,
+        running: stat.running,
+        dead: stat.dead,
+        oldest_pending_age_ms: stat.oldest_pending_age_ms,
+    }
+}
+
 /// A message pulled from a log topic. `id` is the cursor token to pass to
 /// `ackTopic`; `payload` is the opaque published bytes.
 #[napi(object)]

crates/taskito-node/src/queue/pubsub.rs

@@ -12,9 +12,9 @@ use taskito_core::Storage;
 use super::JsQueue;
 use crate::config::PublishOptions;
 use crate::convert::{
-    job_to_js, subscription_to_js, topic_log_stat_to_js, topic_message_to_js, topic_to_js, JsJob,
-    JsSubscription, JsTopic, JsTopicLogStat, JsTopicMessage, DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY,
-    DEFAULT_TIMEOUT_MS,
+    job_to_js, subscription_to_js, topic_log_stat_to_js, topic_message_to_js, topic_stat_to_js,
+    topic_to_js, JsJob, JsSubscription, JsTopic, JsTopicLogStat, JsTopicMessage, JsTopicStat,
+    DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS,
 };
 use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err};
 
@@ -274,6 +274,20 @@ impl JsQueue {
         .map_err(join_to_napi_err)?
     }
 
+    /// Backlog snapshot per registered subscription — one entry each, even at
+    /// zero backlog. Counts are aggregated live off the delivery-attribution
+    /// indexes, so they can never drift the way a maintained counter would.
+    #[napi]
+    pub async fn topic_backlog_stats(&self) -> Result<Vec<JsTopicStat>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let stats = storage.topic_backlog_stats().map_err(to_napi_err)?;
+            Ok(stats.into_iter().map(topic_stat_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
+
     /// Drop ephemeral subscriptions whose owning worker is gone. Runs on the
     /// heartbeat cadence. Returns the number of subscriptions removed.
     ///

docs/content/docs/java/api-reference/queue/pubsub.mdx

@@ -14,6 +14,7 @@ semantics, lifecycle, cross-SDK topics).
 | `pauseSubscription(topic, name)` / `resumeSubscription(topic, name)` | Stop/resume deliveries without unregistering; `false` if none matched. |
 | `listSubscriptions()` / `listSubscriptions(topic)` | Every subscription, or one topic's active ones. |
 | `listTopics()` | Distinct topics with at least one subscription. |
+| `topicStats()` / `topicStats(topic)` → `List<TopicStat>` | Backlog snapshot per subscription, across all topics or filtered to one: `topic`, `subscription`, `taskName`, `queue`, `active`, `durable`, `pending`, `running`, `dead`, `oldestPendingAgeMs`. Every registered subscription appears — paused and ephemeral ones included — even at zero backlog. Computed live off indexed columns, so it is safe to poll. |
 | `subscribeLog(topic, name)` | Register a durable log subscription — a named cursor with no handler. Writes immediately, so register it before the publishes it should see. |
 | `logConsumer(topic, name, payloadType, handler)` / `logConsumer(topic, name, payloadType, handler, options)` | Register a **managed consumer**: the durable log subscription plus, once a worker runs, a daemon thread that pulls messages, decodes each into `payloadType`, invokes `handler`, and advances the cursor. `LogConsumerOptions` sets `pollIntervalMs` (default 1000), `batchSize` (default 100), and `onError` (`"retry"` leaves a failed message un-acked to re-read, default; `"skip"` acks past it). |
 | `declareTopic(name)` / `declareTopic(name, retention)` | Declare a log topic so its publishes are retained even with no subscriber (removing the late-join boundary). `retention` (a `Duration`) bounds a sub-less backlog. Idempotent. |

docs/content/docs/node/api-reference/queue/pubsub.mdx

@@ -15,6 +15,7 @@ semantics, lifecycle, cross-SDK topics).
 | `pauseSubscription(topic, name)` / `resumeSubscription(topic, name)` → `Promise<boolean>` | Stop/resume deliveries without unregistering. |
 | `listSubscriptions(topic?)` → `Promise<Subscription[]>` | All subscriptions, or one topic's active ones. |
 | `listTopics()` → `Promise<string[]>` | Distinct topics with at least one subscription. |
+| `topicStats(topic?)` → `Promise<TopicStat[]>` | Backlog snapshot per subscription, optionally filtered to one topic: `topic`, `subscription`, `taskName`, `queue`, `active`, `durable`, `pending`, `running`, `dead`, `oldestPendingAgeMs`. Every registered subscription appears — paused and ephemeral ones included — even at zero backlog. Computed live off indexed columns, so it is safe to poll. |
 | `reapEphemeralSubscriptions()` → `Promise<number>` | Drop ephemeral subscriptions whose owning worker is gone. Runs automatically on the worker heartbeat cadence; exposed for operational tooling. |
 | `subscribeLog(topic, name)` → `Promise<void>` | Register a durable log subscription — a named cursor with no handler. Writes immediately, so register it before the publishes it should see. |
 | `logConsumer(topic, name, handler, opts?)` → `this` | Register a **managed consumer**: the durable log subscription plus, once a worker runs, a poll loop that pulls messages, calls `handler(...args)` per message (the handler may return a `Promise` — it is awaited), and advances the cursor. `opts`: `pollIntervalMs` (default 1000), `batchSize` (default 100), `onError` (`"retry"` leaves a failed message un-acked to re-read, default; `"skip"` acks past it). |

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -52,6 +52,7 @@
 import org.byteveda.taskito.model.Topic;
 import org.byteveda.taskito.model.TopicLogStat;
 import org.byteveda.taskito.model.TopicMessage;
+import org.byteveda.taskito.model.TopicStat;
 import org.byteveda.taskito.model.WorkerInfo;
 import org.byteveda.taskito.model.WorkflowRunInfo;
 import org.byteveda.taskito.predicates.EnqueueDecision;
@@ -924,6 +925,26 @@ public List<Subscription> listSubscriptions(String topic) {
         return decodeList(backend.listSubscriptionsJson(topic), Subscription.class);
     }
 
+    @Override
+    public List<TopicStat> topicStats() {
+        return decodeList(backend.topicBacklogStatsJson(), TopicStat.class);
+    }
+
+    @Override
+    public List<TopicStat> topicStats(String topic) {
+        // A null topic is "no filter", matching listSubscriptions(null).
+        if (topic == null) {
+            return topicStats();
+        }
+        List<TopicStat> filtered = new ArrayList<>();
+        for (TopicStat stat : topicStats()) {
+            if (stat.topic.equals(topic)) {
+                filtered.add(stat);
+            }
+        }
+        return filtered;
+    }
+
     @Override
     public List<String> listTopics() {
         Set<String> topics = new LinkedHashSet<>();

sdks/java/src/main/java/org/byteveda/taskito/Taskito.java

@@ -41,6 +41,7 @@
 import org.byteveda.taskito.model.Topic;
 import org.byteveda.taskito.model.TopicLogStat;
 import org.byteveda.taskito.model.TopicMessage;
+import org.byteveda.taskito.model.TopicStat;
 import org.byteveda.taskito.model.WorkerInfo;
 import org.byteveda.taskito.model.WorkflowRunInfo;
 import org.byteveda.taskito.predicates.EnqueueGate;
@@ -459,6 +460,17 @@ static Builder builder() {
     /** One topic's active subscriptions. */
     List<Subscription> listSubscriptions(String topic);
 
+    /**
+     * Backlog snapshot per subscription, across all topics. Every registered
+     * subscription appears — paused and ephemeral ones included — even with nothing
+     * queued, so the full subscriber list comes from one call. Counts are computed
+     * live off indexed columns, so this is safe to poll.
+     */
+    List<TopicStat> topicStats();
+
+    /** As {@link #topicStats()}, filtered to one topic; a {@code null} topic means no filter. */
+    List<TopicStat> topicStats(String topic);
+
     /** Distinct topics that currently have at least one subscription. */
     List<String> listTopics();
 

sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java

@@ -393,6 +393,11 @@ public boolean setSubscriptionActive(String topic, String subscriptionName, bool
         return withOpenHandle(() -> NativeQueue.setSubscriptionActive(handle, topic, subscriptionName, active));
     }
 
+    @Override
+    public String topicBacklogStatsJson() {
+        return withOpenHandle(() -> NativeQueue.topicBacklogStats(handle));
+    }
+
     @Override
     public long reapEphemeralSubscriptions() {
         return withOpenHandle(() -> NativeQueue.reapEphemeralSubscriptions(handle));

sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java

@@ -178,6 +178,9 @@ public static native void registerSubscription(
     public static native boolean setSubscriptionActive(
             long handle, String topic, String subscriptionName, boolean active);
 
+    /** A JSON array of per-subscription backlog snapshots, one per registered subscription. */
+    public static native String topicBacklogStats(long handle);
+
     /** Drop ephemeral subscriptions whose owning worker is gone; returns the count removed. */
     public static native long reapEphemeralSubscriptions(long handle);
 

sdks/java/src/main/java/org/byteveda/taskito/model/TopicStat.java

@@ -0,0 +1,62 @@
+package org.byteveda.taskito.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Backlog snapshot for one topic subscription: how much of its fan-out is outstanding. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public final class TopicStat {
+    public final String topic;
+
+    /** Stable subscription identity; unique per topic. */
+    public final String subscription;
+
+    /** Task enqueued for each published message. */
+    public final String taskName;
+
+    /** Queue deliveries are enqueued into. */
+    public final String queue;
+
+    /** Whether the subscription currently receives deliveries (false = paused). */
+    public final boolean active;
+
+    /** Whether the registration persists across restarts (false = ephemeral). */
+    public final boolean durable;
+
+    /** Deliveries waiting to run. */
+    public final long pending;
+
+    /** Deliveries currently executing. */
+    public final long running;
+
+    /** Deliveries in the dead-letter queue. */
+    public final long dead;
+
+    /** Age (ms) of the oldest still-pending delivery, or {@code null} at zero backlog. */
+    public final Long oldestPendingAgeMs;
+
+    @JsonCreator
+    public TopicStat(
+            @JsonProperty("topic") String topic,
+            @JsonProperty("subscription") String subscription,
+            @JsonProperty("taskName") String taskName,
+            @JsonProperty("queue") String queue,
+            @JsonProperty("active") boolean active,
+            @JsonProperty("durable") boolean durable,
+            @JsonProperty("pending") long pending,
+            @JsonProperty("running") long running,
+            @JsonProperty("dead") long dead,
+            @JsonProperty("oldestPendingAgeMs") Long oldestPendingAgeMs) {
+        this.topic = topic;
+        this.subscription = subscription;
+        this.taskName = taskName;
+        this.queue = queue;
+        this.active = active;
+        this.durable = durable;
+        this.pending = pending;
+        this.running = running;
+        this.dead = dead;
+        this.oldestPendingAgeMs = oldestPendingAgeMs;
+    }
+}

sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java

@@ -294,6 +294,11 @@ default boolean setSubscriptionActive(String topic, String subscriptionName, boo
         throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
     }
 
+    /** A JSON array of per-subscription backlog snapshots, one per registered subscription. */
+    default String topicBacklogStatsJson() {
+        throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);
+    }
+
     /** Drop ephemeral subscriptions whose owning worker is gone; returns the count removed. */
     default long reapEphemeralSubscriptions() {
         throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED);