From 6ffac5ee71d2a7b6de1431c1db15135cfe0b38b7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 10:38:40 +0530 Subject: [PATCH 1/9] feat(core): add log+cursor pub/sub storage layer S28: a log topic stores one topic_messages row per publish (vs one job per subscriber) and each log subscription tracks a cursor. Adds the topic_messages table (m0005), mode/cursor columns, four Storage methods (publish_message/read_topic_messages/ack_topic_cursor/topic_log_stats) plus purge_topic_messages across SQLite, Postgres, and Redis Streams. --- Cargo.toml | 2 +- .../migrations/m0005_topic_messages.rs | 71 +++++ crates/taskito-core/src/pubsub.rs | 1 + .../src/storage/diesel_common/pubsub.rs | 212 +++++++++++++++ crates/taskito-core/src/storage/mod.rs | 69 +++++ crates/taskito-core/src/storage/models.rs | 54 +++- .../src/storage/postgres/pubsub.rs | 6 +- crates/taskito-core/src/storage/records.rs | 51 ++++ .../src/storage/redis_backend/pubsub.rs | 246 +++++++++++++++++- crates/taskito-core/src/storage/schema.rs | 14 + .../taskito-core/src/storage/sqlite/pubsub.rs | 6 +- .../taskito-core/src/storage/sqlite/tests.rs | 1 + crates/taskito-core/src/storage/traits.rs | 37 ++- .../taskito-core/tests/rust/storage_tests.rs | 97 +++++++ crates/taskito-java/src/queue/pubsub.rs | 2 + crates/taskito-java/src/worker.rs | 2 + crates/taskito-node/src/queue/pubsub.rs | 2 + crates/taskito-python/src/py_queue/pubsub.rs | 2 + 18 files changed, 868 insertions(+), 7 deletions(-) create mode 100644 crates/taskito-core/migrations/m0005_topic_messages.rs diff --git a/Cargo.toml b/Cargo.toml index 5cfd9dde..f093a93d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ chrono = "0.4" cron = "0.16" chrono-tz = "0.10" log = "0.4" -redis = { version = "1.2", features = ["script"] } +redis = { version = "1.2", features = ["script", "streams"] } openssl-sys = { version = "0.9", features = ["vendored"] } pyo3 = { version = "0.29", features = ["multiple-pymethods"] } async-trait = "0.1" diff --git a/crates/taskito-core/migrations/m0005_topic_messages.rs b/crates/taskito-core/migrations/m0005_topic_messages.rs new file mode 100644 index 00000000..2eebb2f9 --- /dev/null +++ b/crates/taskito-core/migrations/m0005_topic_messages.rs @@ -0,0 +1,71 @@ +//! Log-topic message store (`0005_topic_messages`). +//! +//! S28 adds an opt-in log+cursor pub/sub mode. A publish to a log topic writes +//! exactly one `topic_messages` row (vs one `jobs` row per subscriber in +//! fan-out mode), and each log subscription tracks a `cursor` over the log. This +//! migration creates the message store and adds `mode`/`cursor` to the existing +//! `topic_subscriptions` registry. Idempotent: `.if_not_exists()` on the table +//! and index, `add_column` (swallowed duplicate) for the two columns. + +use sea_query::{Alias, ColumnDef, Index, Table}; + +use crate::storage::migrate::{add_column, ddl, Backend, Migration, Stmt}; + +pub struct M0005TopicMessages; + +fn col(name: &str) -> ColumnDef { + ColumnDef::new(Alias::new(name)) +} + +fn t(name: &str) -> Alias { + Alias::new(name) +} + +impl Migration for M0005TopicMessages { + fn version(&self) -> &'static str { + "0005_topic_messages" + } + + fn up(&self, b: Backend) -> Vec { + let messages = Table::create() + .table(t("topic_messages")) + .if_not_exists() + .col(col("id").text().not_null().primary_key()) + .col(col("topic").text().not_null()) + .col(col("payload").blob().not_null()) + .col(col("metadata").text()) + .col(col("notes").text()) + .col(col("created_at").big_integer().not_null()) + .col(col("expires_at").big_integer()) + .to_owned(); + + // Keyset read per topic: `WHERE topic = ? AND id > ? ORDER BY id`. + let by_topic_id = Index::create() + .if_not_exists() + .name("idx_topic_messages_topic_id") + .table(t("topic_messages")) + .col(t("topic")) + .col(t("id")) + .to_owned(); + + // Bounded retention sweep by expiry. + let by_expiry = Index::create() + .if_not_exists() + .name("idx_topic_messages_expires_at") + .table(t("topic_messages")) + .col(t("expires_at")) + .to_owned(); + + vec![ + ddl(b, &messages), + ddl(b, &by_topic_id), + ddl(b, &by_expiry), + add_column( + b, + "topic_subscriptions", + col("mode").text().not_null().default("fanout"), + ), + add_column(b, "topic_subscriptions", col("cursor").text()), + ] + } +} diff --git a/crates/taskito-core/src/pubsub.rs b/crates/taskito-core/src/pubsub.rs index b58e7417..aa1851ff 100644 --- a/crates/taskito-core/src/pubsub.rs +++ b/crates/taskito-core/src/pubsub.rs @@ -218,6 +218,7 @@ mod tests { priority, max_retries, timeout_ms, + mode: crate::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), }) .unwrap(); } diff --git a/crates/taskito-core/src/storage/diesel_common/pubsub.rs b/crates/taskito-core/src/storage/diesel_common/pubsub.rs index d853f552..973dbd67 100644 --- a/crates/taskito-core/src/storage/diesel_common/pubsub.rs +++ b/crates/taskito-core/src/storage/diesel_common/pubsub.rs @@ -153,6 +153,218 @@ macro_rules! impl_diesel_pubsub_ops { subs, counts, oldest, dead, )) } + + /// Append one message to a log topic (id = UUIDv7, so the id is the + /// time-ordered read cursor). O(1), independent of subscriber count. + pub fn publish_message( + &self, + topic: &str, + payload: &[u8], + metadata: Option<&str>, + notes: Option<&str>, + expires_at: Option, + ) -> Result<$crate::storage::records::TopicMessage> { + let mut conn = self.conn()?; + let id = uuid::Uuid::now_v7().to_string(); + let created_at = $crate::job::now_millis(); + let row = NewTopicMessageRow { + id: &id, + topic, + payload, + metadata, + notes, + created_at, + expires_at, + }; + diesel::insert_into(topic_messages::table) + .values(&row) + .execute(&mut conn)?; + Ok($crate::storage::records::TopicMessage { + id, + topic: topic.to_string(), + payload: payload.to_vec(), + metadata: metadata.map(str::to_string), + notes: notes.map(str::to_string), + created_at, + expires_at, + }) + } + + /// Messages after a log subscription's cursor, oldest first, `<= limit`. + /// Cursor resolved server-side; read is exclusive. Unknown subscription + /// or non-positive limit → empty. + pub fn read_topic_messages( + &self, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> Result> { + if limit <= 0 { + return Ok(Vec::new()); + } + let mut conn = self.conn()?; + let cursor: Option> = topic_subscriptions::table + .find((topic, subscription_name)) + .select(topic_subscriptions::cursor) + .first(&mut conn) + .optional()?; + let Some(cursor) = cursor else { + return Ok(Vec::new()); + }; + + let mut query = topic_messages::table + .filter(topic_messages::topic.eq(topic)) + .into_boxed(); + if let Some(after) = cursor { + query = query.filter(topic_messages::id.gt(after)); + } + let rows = query + .order(topic_messages::id.asc()) + .limit(limit) + .select(TopicMessageRow::as_select()) + .load::(&mut conn)?; + Ok(rows.into_iter().map(Into::into).collect()) + } + + /// Advance a log subscription's cursor. Monotonic (never rewinds) and + /// idempotent. Returns false when nothing advanced. + pub fn ack_topic_cursor( + &self, + topic: &str, + subscription_name: &str, + cursor: &str, + ) -> Result { + let mut conn = self.conn()?; + let affected = diesel::update( + topic_subscriptions::table + .filter(topic_subscriptions::topic.eq(topic)) + .filter(topic_subscriptions::subscription_name.eq(subscription_name)) + .filter( + topic_subscriptions::cursor + .is_null() + .or(topic_subscriptions::cursor.lt(cursor)), + ), + ) + .set(topic_subscriptions::cursor.eq(cursor)) + .execute(&mut conn)?; + Ok(affected > 0) + } + + /// Per-log-subscription lag (messages after the cursor) + oldest + /// un-acked age. One aggregate per log subscription; fan-out excluded. + pub fn topic_log_stats(&self) -> Result> { + let mut conn = self.conn()?; + let now = $crate::job::now_millis(); + let subs: Vec<(String, String, Option)> = topic_subscriptions::table + .filter( + topic_subscriptions::mode + .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG), + ) + .select(( + topic_subscriptions::topic, + topic_subscriptions::subscription_name, + topic_subscriptions::cursor, + )) + .load(&mut conn)?; + + let mut out = Vec::with_capacity(subs.len()); + for (topic, subscription_name, cursor) in subs { + let mut query = topic_messages::table + .filter(topic_messages::topic.eq(&topic)) + .into_boxed(); + if let Some(ref after) = cursor { + query = query.filter(topic_messages::id.gt(after.clone())); + } + let (lag, oldest): (i64, Option) = query + .select(( + diesel::dsl::count_star(), + diesel::dsl::min(topic_messages::created_at), + )) + .first(&mut conn)?; + out.push($crate::storage::records::TopicLogStats { + topic, + subscription_name, + cursor, + lag, + oldest_unacked_age_ms: oldest.map(|c| (now - c).max(0)), + }); + } + Ok(out) + } + + /// Drop log messages every subscriber has acked past (id `<=` the min + /// cursor across a topic's log subs) plus any past `expires_at`. + /// Bounded by `limit`. A topic with an un-consumed (NULL-cursor) sub + /// keeps all its messages except expired ones. + pub fn purge_topic_messages(&self, now: i64, limit: i64) -> Result { + use std::collections::HashMap; + if limit <= 0 { + return Ok(0); + } + let mut conn = self.conn()?; + + // Per-topic delete floor: the min cursor across its log subs. + // A topic maps to `None` once any of its subs has read nothing + // (NULL cursor), which excludes it from cursor-based purging. + let subs: Vec<(String, Option)> = topic_subscriptions::table + .filter( + topic_subscriptions::mode + .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG), + ) + .select((topic_subscriptions::topic, topic_subscriptions::cursor)) + .load(&mut conn)?; + let mut floors: HashMap> = HashMap::new(); + for (topic, cursor) in subs { + match floors.entry(topic) { + std::collections::hash_map::Entry::Vacant(slot) => { + slot.insert(cursor); + } + std::collections::hash_map::Entry::Occupied(mut slot) => { + let merged = match (slot.get().clone(), cursor) { + (Some(current), Some(c)) => Some(current.min(c)), + _ => None, + }; + slot.insert(merged); + } + } + } + + let mut ids: Vec = Vec::new(); + // Expired messages first (TTL safety net, ignores cursors). + let expired: Vec = topic_messages::table + .filter(topic_messages::expires_at.is_not_null()) + .filter(topic_messages::expires_at.le(now)) + .select(topic_messages::id) + .limit(limit) + .load(&mut conn)?; + ids.extend(expired); + + // Then fully-acked messages per topic, within the remaining budget. + for (topic, floor) in floors { + let Some(floor) = floor else { continue }; + let remaining = limit - ids.len() as i64; + if remaining <= 0 { + break; + } + let acked: Vec = topic_messages::table + .filter(topic_messages::topic.eq(&topic)) + .filter(topic_messages::id.le(&floor)) + .select(topic_messages::id) + .limit(remaining) + .load(&mut conn)?; + ids.extend(acked); + } + + if ids.is_empty() { + return Ok(0); + } + ids.sort_unstable(); + ids.dedup(); + let removed = + diesel::delete(topic_messages::table.filter(topic_messages::id.eq_any(&ids))) + .execute(&mut conn)?; + Ok(removed as u64) + } } }; } diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index a055297a..025b13a7 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/crates/taskito-core/src/storage/mod.rs @@ -694,6 +694,40 @@ macro_rules! impl_storage { ) -> $crate::error::Result> { self.topic_backlog_stats() } + fn publish_message( + &self, + topic: &str, + payload: &[u8], + metadata: Option<&str>, + notes: Option<&str>, + expires_at: Option, + ) -> $crate::error::Result<$crate::storage::records::TopicMessage> { + self.publish_message(topic, payload, metadata, notes, expires_at) + } + fn read_topic_messages( + &self, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> $crate::error::Result> { + self.read_topic_messages(topic, subscription_name, limit) + } + fn ack_topic_cursor( + &self, + topic: &str, + subscription_name: &str, + cursor: &str, + ) -> $crate::error::Result { + self.ack_topic_cursor(topic, subscription_name, cursor) + } + fn topic_log_stats( + &self, + ) -> $crate::error::Result> { + self.topic_log_stats() + } + fn purge_topic_messages(&self, now: i64, limit: i64) -> $crate::error::Result { + self.purge_topic_messages(now, limit) + } fn record_metric( &self, task_name: &str, @@ -1362,6 +1396,41 @@ impl Storage for StorageBackend { fn topic_backlog_stats(&self) -> Result> { delegate!(self, topic_backlog_stats) } + fn publish_message( + &self, + topic: &str, + payload: &[u8], + metadata: Option<&str>, + notes: Option<&str>, + expires_at: Option, + ) -> Result { + delegate!( + self, + publish_message, + topic, + payload, + metadata, + notes, + expires_at + ) + } + fn read_topic_messages( + &self, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> Result> { + delegate!(self, read_topic_messages, topic, subscription_name, limit) + } + fn ack_topic_cursor(&self, topic: &str, subscription_name: &str, cursor: &str) -> Result { + delegate!(self, ack_topic_cursor, topic, subscription_name, cursor) + } + fn topic_log_stats(&self) -> Result> { + delegate!(self, topic_log_stats) + } + fn purge_topic_messages(&self, now: i64, limit: i64) -> Result { + delegate!(self, purge_topic_messages, now, limit) + } fn record_metric( &self, task_name: &str, diff --git a/crates/taskito-core/src/storage/models.rs b/crates/taskito-core/src/storage/models.rs index 96ddc3a6..a8e890b3 100644 --- a/crates/taskito-core/src/storage/models.rs +++ b/crates/taskito-core/src/storage/models.rs @@ -3,12 +3,12 @@ use serde::{Deserialize, Serialize}; use super::records::{ CircuitBreakerState, JobError, LockInfo, PeriodicTask, RateLimitState, ReplayEntry, - Subscription, TaskLogEntry, TaskMetric, WorkerInfo, + Subscription, TaskLogEntry, TaskMetric, TopicMessage, WorkerInfo, }; use super::schema::{ archived_jobs, circuit_breakers, dashboard_settings, dead_letter, distributed_locks, execution_claims, job_dependencies, job_errors, jobs, periodic_tasks, queue_state, rate_limits, - replay_history, task_logs, task_metrics, topic_subscriptions, workers, + replay_history, task_logs, task_metrics, topic_messages, topic_subscriptions, workers, }; /// A row in the `jobs` table (for SELECT queries). @@ -451,6 +451,10 @@ pub struct SubscriptionRow { pub priority: Option, pub max_retries: Option, pub timeout_ms: Option, + /// Delivery mode: `"fanout"` (default) or `"log"`. + pub mode: String, + /// Log-mode read cursor (last-acked message id); `None` = unread. + pub cursor: Option, } /// Insertable/updatable struct for subscription registrations. @@ -468,6 +472,36 @@ pub struct NewSubscriptionRow<'a> { pub priority: Option, pub max_retries: Option, pub timeout_ms: Option, + pub mode: &'a str, +} + +// ── Topic Messages (log topics) ───────────────────────────────── + +/// A row in the append-only `topic_messages` log. One per publish to a log +/// topic (vs one `jobs` row per subscriber in fan-out mode). +#[derive(Queryable, Selectable, Debug, Clone)] +#[diesel(table_name = topic_messages)] +pub struct TopicMessageRow { + pub id: String, + pub topic: String, + pub payload: Vec, + pub metadata: Option, + pub notes: Option, + pub created_at: i64, + pub expires_at: Option, +} + +/// Insertable struct for a published log message. +#[derive(Insertable, Debug)] +#[diesel(table_name = topic_messages)] +pub struct NewTopicMessageRow<'a> { + pub id: &'a str, + pub topic: &'a str, + pub payload: &'a [u8], + pub metadata: Option<&'a str>, + pub notes: Option<&'a str>, + pub created_at: i64, + pub expires_at: Option, } // ── Distributed Locks ─────────────────────────────────────────── @@ -666,6 +700,22 @@ impl From for Subscription { priority: r.priority, max_retries: r.max_retries, timeout_ms: r.timeout_ms, + mode: r.mode, + cursor: r.cursor, + } + } +} + +impl From for TopicMessage { + fn from(r: TopicMessageRow) -> Self { + TopicMessage { + id: r.id, + topic: r.topic, + payload: r.payload, + metadata: r.metadata, + notes: r.notes, + created_at: r.created_at, + expires_at: r.expires_at, } } } diff --git a/crates/taskito-core/src/storage/postgres/pubsub.rs b/crates/taskito-core/src/storage/postgres/pubsub.rs index 6f192736..b8cb4924 100644 --- a/crates/taskito-core/src/storage/postgres/pubsub.rs +++ b/crates/taskito-core/src/storage/postgres/pubsub.rs @@ -2,7 +2,7 @@ use diesel::prelude::*; use super::super::models::*; use super::super::records::NewSubscription; -use super::super::schema::topic_subscriptions; +use super::super::schema::{topic_messages, topic_subscriptions}; use super::PostgresStorage; use crate::error::Result; @@ -30,8 +30,11 @@ impl PostgresStorage { priority: sub.priority, max_retries: sub.max_retries, timeout_ms: sub.timeout_ms, + mode: &sub.mode, }; + // `cursor` is omitted so a re-registration preserves a log consumer's + // read position. diesel::insert_into(topic_subscriptions::table) .values(&row) .on_conflict(( @@ -47,6 +50,7 @@ impl PostgresStorage { topic_subscriptions::priority.eq(row.priority), topic_subscriptions::max_retries.eq(row.max_retries), topic_subscriptions::timeout_ms.eq(row.timeout_ms), + topic_subscriptions::mode.eq(row.mode), )) .execute(&mut conn)?; diff --git a/crates/taskito-core/src/storage/records.rs b/crates/taskito-core/src/storage/records.rs index 5c82d6cc..70a06e95 100644 --- a/crates/taskito-core/src/storage/records.rs +++ b/crates/taskito-core/src/storage/records.rs @@ -117,6 +117,12 @@ pub struct Subscription { pub max_retries: Option, /// Per-delivery timeout in milliseconds. `None` = queue default. pub timeout_ms: Option, + /// Delivery mode: `"fanout"` (one job per publish, the default) or `"log"` + /// (append one `topic_messages` row per publish; consumer pulls via cursor). + pub mode: String, + /// Log-mode read cursor: the last-acked message id. `None` = unread (start + /// from the beginning). Ignored for fan-out subscriptions. + pub cursor: Option, } /// Registration payload for a topic subscription. @@ -144,6 +150,51 @@ pub struct NewSubscription { pub max_retries: Option, /// Per-delivery timeout in milliseconds. `None` = queue default. pub timeout_ms: Option, + /// Delivery mode: `"fanout"` (default) or `"log"`. See [`Subscription::mode`]. + pub mode: String, +} + +/// Delivery mode marker for the `mode` column. `"log"` opts a subscription into +/// the append-once + cursor model; anything else is treated as fan-out. +pub const SUBSCRIPTION_MODE_LOG: &str = "log"; +/// Default fan-out delivery mode (one job per publish). +pub const SUBSCRIPTION_MODE_FANOUT: &str = "fanout"; + +/// One durable message in a log topic. Unlike fan-out delivery (one `jobs` row +/// per subscriber), a log publish writes exactly one of these and each log +/// subscription advances its own cursor over them. +#[derive(Debug, Clone)] +pub struct TopicMessage { + /// Message id — a time-ordered token that doubles as the read cursor. + /// Opaque to callers (UUIDv7 on Diesel backends, a stream id on Redis). + pub id: String, + /// Topic the message was published to. + pub topic: String, + /// Opaque payload bytes (same codec as fan-out `publish`). + pub payload: Vec, + /// Optional caller metadata (JSON). + pub metadata: Option, + /// Optional structured notes (JSON). + pub notes: Option, + /// Unix-millisecond publish time. + pub created_at: i64, + /// Optional expiry (Unix ms) — a TTL safety net for the retention sweep. + pub expires_at: Option, +} + +/// Backlog snapshot for one log subscription: how far its cursor lags the log. +#[derive(Debug, Clone)] +pub struct TopicLogStats { + /// Topic the subscription reads. + pub topic: String, + /// Subscription name. + pub subscription_name: String, + /// Current read cursor (last-acked id); `None` = nothing acked yet. + pub cursor: Option, + /// Number of messages after the cursor still to be consumed. + pub lag: i64, + /// Age (ms) of the oldest un-acked message; `None` when fully caught up. + pub oldest_unacked_age_ms: Option, } /// One execution measurement for a task. diff --git a/crates/taskito-core/src/storage/redis_backend/pubsub.rs b/crates/taskito-core/src/storage/redis_backend/pubsub.rs index 7d4e86f8..7efec925 100644 --- a/crates/taskito-core/src/storage/redis_backend/pubsub.rs +++ b/crates/taskito-core/src/storage/redis_backend/pubsub.rs @@ -1,12 +1,15 @@ use std::collections::HashSet; +use redis::streams::{StreamId, StreamRangeReply}; use redis::Commands; use serde::{Deserialize, Serialize}; use super::{map_err, RedisStorage}; use crate::error::{QueueError, Result}; use crate::job::{Job, JobStatus}; -use crate::storage::records::{NewSubscription, Subscription}; +use crate::storage::records::{ + NewSubscription, Subscription, TopicLogStats, TopicMessage, SUBSCRIPTION_MODE_LOG, +}; use crate::storage::SubscriptionBacklogStats; /// JSON blob stored at `sub::`. Mirrors @@ -30,6 +33,15 @@ struct SubEntry { max_retries: Option, #[serde(default)] timeout_ms: Option, + // S28 log topics. Older blobs default to fan-out with no cursor. + #[serde(default = "default_mode")] + mode: String, + #[serde(default)] + cursor: Option, +} + +fn default_mode() -> String { + crate::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string() } impl From for Subscription { @@ -46,6 +58,8 @@ impl From for Subscription { priority: e.priority, max_retries: e.max_retries, timeout_ms: e.timeout_ms, + mode: e.mode, + cursor: e.cursor, } } } @@ -123,6 +137,8 @@ impl RedisStorage { priority: sub.priority, max_retries: sub.max_retries, timeout_ms: sub.timeout_ms, + mode: sub.mode.clone(), + cursor: None, }; let blob_key = self.key(&["sub", &sub.topic, &sub.subscription_name]); let by_topic = self.key(&["subs", "by_topic", &sub.topic]); @@ -139,6 +155,7 @@ impl RedisStorage { if let Some(prior) = &prior { entry.active = prior.active; entry.created_at = prior.created_at; + entry.cursor = prior.cursor.clone(); } let json = serde_json::to_string(&entry)?; @@ -478,4 +495,231 @@ impl RedisStorage { .collect(); Ok(out) } + + // ── Log topics (Redis Streams) ────────────────────────────────── + // + // A log topic is a stream at `topic:`; a publish is one `XADD` and + // the returned stream id is the message id / cursor token. Per-subscription + // cursors live in the `SubEntry` blob. This is the Redis fork of the Diesel + // `topic_messages` table (see `diesel_common/pubsub.rs`). + + /// Stream key for a topic's log. + fn topic_stream_key(&self, topic: &str) -> String { + self.key(&["topic", topic]) + } + + /// Parse a `-` stream id into comparable parts (invalid → `(0, 0)`). + fn stream_id_parts(id: &str) -> (u64, u64) { + let mut parts = id.splitn(2, '-'); + let ms = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0); + let seq = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0); + (ms, seq) + } + + /// The next stream id after `id`, used as an exclusive `XTRIM MINID` floor. + fn next_stream_id(id: &str) -> String { + let (ms, seq) = Self::stream_id_parts(id); + match seq.checked_add(1) { + Some(next) => format!("{ms}-{next}"), + None => format!("{}-0", ms + 1), + } + } + + /// Build a [`TopicMessage`] from a stream entry. + fn stream_id_to_message(topic: &str, entry: StreamId) -> TopicMessage { + TopicMessage { + id: entry.id.clone(), + topic: topic.to_string(), + payload: entry.get::>("payload").unwrap_or_default(), + metadata: entry.get::("metadata"), + notes: entry.get::("notes"), + created_at: entry.get::("created_at").unwrap_or_default(), + expires_at: entry.get::("expires_at"), + } + } + + /// Append one message to a log topic (`XADD`); the stream id is the message + /// id. O(1), independent of subscriber count. + pub fn publish_message( + &self, + topic: &str, + payload: &[u8], + metadata: Option<&str>, + notes: Option<&str>, + expires_at: Option, + ) -> Result { + let mut conn = self.conn()?; + let created_at = crate::job::now_millis(); + let stream_key = self.topic_stream_key(topic); + + let mut cmd = redis::cmd("XADD"); + cmd.arg(&stream_key).arg("*"); + cmd.arg("payload").arg(payload); + cmd.arg("created_at").arg(created_at); + if let Some(m) = metadata { + cmd.arg("metadata").arg(m); + } + if let Some(n) = notes { + cmd.arg("notes").arg(n); + } + if let Some(e) = expires_at { + cmd.arg("expires_at").arg(e); + } + let id: String = cmd.query(&mut conn).map_err(map_err)?; + + Ok(TopicMessage { + id, + topic: topic.to_string(), + payload: payload.to_vec(), + metadata: metadata.map(str::to_string), + notes: notes.map(str::to_string), + created_at, + expires_at, + }) + } + + /// Messages after a log subscription's cursor, oldest first, `<= limit`. + /// Cursor resolved from the `SubEntry`; `XRANGE (cursor +` is exclusive. + pub fn read_topic_messages( + &self, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> Result> { + if limit <= 0 { + return Ok(Vec::new()); + } + let mut conn = self.conn()?; + let blob_key = self.key(&["sub", topic, subscription_name]); + let Some(data): Option = conn.get(&blob_key).map_err(map_err)? else { + return Ok(Vec::new()); + }; + let entry: SubEntry = serde_json::from_str(&data)?; + + let start = match &entry.cursor { + Some(cursor) => format!("({cursor}"), + None => "-".to_string(), + }; + let reply: StreamRangeReply = redis::cmd("XRANGE") + .arg(self.topic_stream_key(topic)) + .arg(&start) + .arg("+") + .arg("COUNT") + .arg(limit) + .query(&mut conn) + .map_err(map_err)?; + + Ok(reply + .ids + .into_iter() + .map(|entry| Self::stream_id_to_message(topic, entry)) + .collect()) + } + + /// Advance a log subscription's cursor. Monotonic (never rewinds by stream-id + /// order) and idempotent. Read-modify-write on the `SubEntry` blob — a single + /// puller per subscription is assumed, matching the pull-consumer model. + pub fn ack_topic_cursor( + &self, + topic: &str, + subscription_name: &str, + cursor: &str, + ) -> Result { + let mut conn = self.conn()?; + let blob_key = self.key(&["sub", topic, subscription_name]); + let Some(data): Option = conn.get(&blob_key).map_err(map_err)? else { + return Ok(false); + }; + let mut entry: SubEntry = serde_json::from_str(&data)?; + + let advance = match &entry.cursor { + None => true, + Some(current) => Self::stream_id_parts(cursor) > Self::stream_id_parts(current), + }; + if !advance { + return Ok(false); + } + entry.cursor = Some(cursor.to_string()); + let json = serde_json::to_string(&entry)?; + conn.set::<_, _, ()>(&blob_key, json).map_err(map_err)?; + Ok(true) + } + + /// Per-log-subscription lag + oldest un-acked age. One `XRANGE (cursor +` per + /// log subscription (O(unacked)); fan-out subscriptions are excluded. + pub fn topic_log_stats(&self) -> Result> { + let subs = self.list_subscriptions()?; + let mut conn = self.conn()?; + let now = crate::job::now_millis(); + + let mut out = Vec::new(); + for sub in subs.into_iter().filter(|s| s.mode == SUBSCRIPTION_MODE_LOG) { + let start = match &sub.cursor { + Some(cursor) => format!("({cursor}"), + None => "-".to_string(), + }; + let reply: StreamRangeReply = redis::cmd("XRANGE") + .arg(self.topic_stream_key(&sub.topic)) + .arg(&start) + .arg("+") + .query(&mut conn) + .map_err(map_err)?; + let oldest = reply + .ids + .first() + .and_then(|entry| entry.get::("created_at")) + .map(|created| (now - created).max(0)); + out.push(TopicLogStats { + topic: sub.topic, + subscription_name: sub.subscription_name, + cursor: sub.cursor, + lag: reply.ids.len() as i64, + oldest_unacked_age_ms: oldest, + }); + } + Ok(out) + } + + /// Compact each log topic to the min cursor across its subscriptions via + /// `XTRIM MINID`. Unlike the Diesel backend this ignores `now`/`limit`: XTRIM + /// is one server-side op with no per-message TTL (min-cursor compaction only). + pub fn purge_topic_messages(&self, _now: i64, _limit: i64) -> Result { + use std::collections::HashMap; + let subs = self.list_subscriptions()?; + let mut floors: HashMap> = HashMap::new(); + for sub in subs.into_iter().filter(|s| s.mode == SUBSCRIPTION_MODE_LOG) { + match floors.entry(sub.topic) { + std::collections::hash_map::Entry::Vacant(slot) => { + slot.insert(sub.cursor); + } + std::collections::hash_map::Entry::Occupied(mut slot) => { + let merged = match (slot.get().clone(), sub.cursor) { + (Some(current), Some(cursor)) => Some( + if Self::stream_id_parts(&cursor) < Self::stream_id_parts(¤t) { + cursor + } else { + current + }, + ), + _ => None, + }; + slot.insert(merged); + } + } + } + + let mut conn = self.conn()?; + let mut removed = 0u64; + for (topic, floor) in floors { + let Some(floor) = floor else { continue }; + let n: u64 = redis::cmd("XTRIM") + .arg(self.topic_stream_key(&topic)) + .arg("MINID") + .arg(Self::next_stream_id(&floor)) + .query(&mut conn) + .map_err(map_err)?; + removed += n; + } + Ok(removed) + } } diff --git a/crates/taskito-core/src/storage/schema.rs b/crates/taskito-core/src/storage/schema.rs index 3f573b13..76afd1c0 100644 --- a/crates/taskito-core/src/storage/schema.rs +++ b/crates/taskito-core/src/storage/schema.rs @@ -242,6 +242,20 @@ diesel::table! { priority -> Nullable, max_retries -> Nullable, timeout_ms -> Nullable, + mode -> Text, + cursor -> Nullable, + } +} + +diesel::table! { + topic_messages (id) { + id -> Text, + topic -> Text, + payload -> Binary, + metadata -> Nullable, + notes -> Nullable, + created_at -> BigInt, + expires_at -> Nullable, } } diff --git a/crates/taskito-core/src/storage/sqlite/pubsub.rs b/crates/taskito-core/src/storage/sqlite/pubsub.rs index 0ceb3a3f..347dce33 100644 --- a/crates/taskito-core/src/storage/sqlite/pubsub.rs +++ b/crates/taskito-core/src/storage/sqlite/pubsub.rs @@ -2,7 +2,7 @@ use diesel::prelude::*; use super::super::models::*; use super::super::records::NewSubscription; -use super::super::schema::topic_subscriptions; +use super::super::schema::{topic_messages, topic_subscriptions}; use super::SqliteStorage; use crate::error::Result; @@ -29,8 +29,11 @@ impl SqliteStorage { priority: sub.priority, max_retries: sub.max_retries, timeout_ms: sub.timeout_ms, + mode: &sub.mode, }; + // `cursor` is omitted so a re-registration preserves a log consumer's + // read position. diesel::insert_into(topic_subscriptions::table) .values(&row) .on_conflict(( @@ -46,6 +49,7 @@ impl SqliteStorage { topic_subscriptions::priority.eq(row.priority), topic_subscriptions::max_retries.eq(row.max_retries), topic_subscriptions::timeout_ms.eq(row.timeout_ms), + topic_subscriptions::mode.eq(row.mode), )) .execute(&mut conn)?; diff --git a/crates/taskito-core/src/storage/sqlite/tests.rs b/crates/taskito-core/src/storage/sqlite/tests.rs index 1fe7956c..1582db69 100644 --- a/crates/taskito-core/src/storage/sqlite/tests.rs +++ b/crates/taskito-core/src/storage/sqlite/tests.rs @@ -1721,6 +1721,7 @@ fn make_sub( priority: None, max_retries: None, timeout_ms: None, + mode: "fanout".to_string(), } } diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index a86351b0..9a73e636 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/crates/taskito-core/src/storage/traits.rs @@ -2,7 +2,8 @@ use crate::error::Result; use crate::job::{Job, NewJob}; use crate::storage::records::{ CircuitBreakerState, JobError, LockInfo, NewPeriodicTask, NewSubscription, PeriodicTask, - RateLimitState, ReplayEntry, Subscription, TaskLogEntry, TaskMetric, WorkerInfo, + RateLimitState, ReplayEntry, Subscription, TaskLogEntry, TaskMetric, TopicLogStats, + TopicMessage, WorkerInfo, }; use crate::storage::{DeadJob, DispatchOrder, QueueStats, SubscriptionBacklogStats}; @@ -253,6 +254,40 @@ pub trait Storage: Send + Sync + Clone { /// table scan — safe to poll on a dashboard cadence. fn topic_backlog_stats(&self) -> Result>; + // ── Log topics (append-once + cursor) ─────────────────────────── + + /// Append one message to a log topic and return it (id generated). O(1) — + /// independent of subscriber count, unlike fan-out delivery. + fn publish_message( + &self, + topic: &str, + payload: &[u8], + metadata: Option<&str>, + notes: Option<&str>, + expires_at: Option, + ) -> Result; + /// Messages after a log subscription's cursor, oldest first, up to `limit`. + /// The cursor is resolved server-side from the subscription row; the read is + /// exclusive of the cursor. An empty result means the consumer is caught up. + fn read_topic_messages( + &self, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> Result>; + /// Advance a log subscription's cursor to `cursor` (a message id). Monotonic: + /// never rewinds (a lower/equal cursor is a no-op). Returns false if no + /// subscription matched. + fn ack_topic_cursor(&self, topic: &str, subscription_name: &str, cursor: &str) -> Result; + /// Lag snapshot for every log subscription: messages after the cursor and + /// the oldest un-acked age. Fan-out subscriptions are excluded. + fn topic_log_stats(&self) -> Result>; + /// Purge fully-consumed log messages: for each topic, drop messages whose id + /// is `<=` the minimum cursor across its log subscriptions, plus any past + /// `expires_at`. Bounded by `limit`. Returns the count removed. Caller gates + /// this on the reaper election. + fn purge_topic_messages(&self, now: i64, limit: i64) -> Result; + // ── Metrics operations ────────────────────────────────────────── /// Record one execution measurement for a task. diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 951c2d1e..5b664423 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -1057,6 +1057,7 @@ fn test_topic_subscriptions_crud(s: &impl Storage) { priority: None, max_retries: None, timeout_ms: None, + mode: "fanout".to_string(), }; // Upsert idempotency: re-registering (topic, name) updates in place. @@ -1230,6 +1231,7 @@ fn test_topic_backlog_stats(s: &impl Storage) { priority: None, max_retries: None, timeout_ms: None, + mode: "fanout".to_string(), }; s.register_subscription(&sub("tbs-email", "tbs_send")) .unwrap(); @@ -1289,6 +1291,99 @@ fn test_topic_backlog_stats(s: &impl Storage) { assert!(claimed.task_name == "tbs_send" || claimed.task_name == "tbs_track"); } +/// A log subscription for the given topic/name (mode = "log"). +fn log_sub(topic: &str, name: &str) -> taskito_core::NewSubscription { + taskito_core::NewSubscription { + topic: topic.to_string(), + subscription_name: name.to_string(), + task_name: String::new(), + queue: "default".to_string(), + active: true, + durable: true, + owner_worker_id: None, + created_at: now_millis(), + priority: None, + max_retries: None, + timeout_ms: None, + mode: "log".to_string(), + } +} + +fn test_topic_log_messages(s: &impl Storage) { + let topic = "tlog-msgs"; + s.register_subscription(&log_sub(topic, "reader")).unwrap(); + + // Publish three messages: each is one row, ids are time-ordered. + let m0 = s.publish_message(topic, b"m0", None, None, None).unwrap(); + let m1 = s.publish_message(topic, b"m1", None, None, None).unwrap(); + let m2 = s.publish_message(topic, b"m2", None, None, None).unwrap(); + assert!(m0.id < m1.id && m1.id < m2.id, "ids are monotonic"); + + // Read from the start: all three, oldest first, payloads intact. + let read = s.read_topic_messages(topic, "reader", 10).unwrap(); + assert_eq!( + read.iter().map(|m| m.payload.clone()).collect::>(), + vec![b"m0".to_vec(), b"m1".to_vec(), b"m2".to_vec()] + ); + + // Ack through m1: a re-read returns only what follows (exclusive cursor). + assert!(s.ack_topic_cursor(topic, "reader", &m1.id).unwrap()); + let after = s.read_topic_messages(topic, "reader", 10).unwrap(); + assert_eq!(after.len(), 1); + assert_eq!(after[0].payload, b"m2".to_vec()); + + // Ack is monotonic: acking an older cursor is a no-op. + assert!(!s.ack_topic_cursor(topic, "reader", &m0.id).unwrap()); + assert_eq!(s.read_topic_messages(topic, "reader", 10).unwrap().len(), 1); + + // Lag reflects the one un-acked message; unknown subscription reads empty. + let stats = s.topic_log_stats().unwrap(); + let mine = stats + .iter() + .find(|st| st.topic == topic && st.subscription_name == "reader") + .expect("log subscription appears in stats"); + assert_eq!(mine.lag, 1); + assert!(s + .read_topic_messages(topic, "ghost", 10) + .unwrap() + .is_empty()); + + // Drop the subscription so the global purge/stats in later tests are not + // affected by this topic's leftover cursor. + s.unsubscribe(topic, "reader").unwrap(); +} + +fn test_topic_log_purge(s: &impl Storage) { + let topic = "tlog-purge"; + s.register_subscription(&log_sub(topic, "a")).unwrap(); + s.register_subscription(&log_sub(topic, "b")).unwrap(); + + let m0 = s.publish_message(topic, b"m0", None, None, None).unwrap(); + let _m1 = s.publish_message(topic, b"m1", None, None, None).unwrap(); + let m2 = s.publish_message(topic, b"m2", None, None, None).unwrap(); + + // Only "a" has acked (through m2); "b" has read nothing, so no message is + // safe to drop yet — the floor is the min cursor across all log subs. + assert!(s.ack_topic_cursor(topic, "a", &m2.id).unwrap()); + assert_eq!(s.purge_topic_messages(now_millis(), 100).unwrap(), 0); + + // Once "b" acks through m0, everything at or before m0 is fully consumed. + assert!(s.ack_topic_cursor(topic, "b", &m0.id).unwrap()); + let removed = s.purge_topic_messages(now_millis(), 100).unwrap(); + assert_eq!(removed, 1, "only m0 is at/below the min cursor"); + + // A fresh reader now sees only the surviving messages (m1, m2). + s.register_subscription(&log_sub(topic, "fresh")).unwrap(); + let survivors = s.read_topic_messages(topic, "fresh", 10).unwrap(); + assert_eq!( + survivors + .iter() + .map(|m| m.payload.clone()) + .collect::>(), + vec![b"m1".to_vec(), b"m2".to_vec()] + ); +} + fn test_enqueue_unique_batch(s: &impl Storage) { let q = "q-eub"; let keyed = |uk: &str| { @@ -1377,6 +1472,8 @@ fn run_storage_tests(s: &impl Storage) { test_periodic_crud(s); test_topic_subscriptions_crud(s); test_topic_backlog_stats(s); + test_topic_log_messages(s); + test_topic_log_purge(s); test_circuit_breakers(s); test_execution_claims_purge(s); test_reap_stale_jobs(s); diff --git a/crates/taskito-java/src/queue/pubsub.rs b/crates/taskito-java/src/queue/pubsub.rs index df36e199..7d41ad08 100644 --- a/crates/taskito-java/src/queue/pubsub.rs +++ b/crates/taskito-java/src/queue/pubsub.rs @@ -72,6 +72,8 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu priority: (priority != jint::MIN).then_some(priority), max_retries: (max_retries != jint::MIN).then_some(max_retries), timeout_ms: (timeout_ms != jlong::MIN).then_some(timeout_ms), + // Fan-out by default; the log-mode param is threaded in a later step. + mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), }; queue_handle.storage.register_subscription(&row)?; Ok(()) diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index e0d4f30e..cb8dc765 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -258,6 +258,8 @@ fn register_subscriptions( priority: spec.priority, max_retries: spec.max_retries, timeout_ms: spec.timeout_ms, + // Fan-out by default; the log-mode param is threaded in a later step. + mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), }; storage.register_subscription(&row)?; } diff --git a/crates/taskito-node/src/queue/pubsub.rs b/crates/taskito-node/src/queue/pubsub.rs index 664d67af..e5dd7120 100644 --- a/crates/taskito-node/src/queue/pubsub.rs +++ b/crates/taskito-node/src/queue/pubsub.rs @@ -57,6 +57,8 @@ impl JsQueue { priority, max_retries, timeout_ms, + // Fan-out by default; the log-mode param is threaded in a later step. + mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), }; storage.register_subscription(&row).map_err(to_napi_err) }) diff --git a/crates/taskito-python/src/py_queue/pubsub.rs b/crates/taskito-python/src/py_queue/pubsub.rs index fc789583..9458611b 100644 --- a/crates/taskito-python/src/py_queue/pubsub.rs +++ b/crates/taskito-python/src/py_queue/pubsub.rs @@ -68,6 +68,8 @@ impl PyQueue { priority, max_retries, timeout_ms, + // Fan-out by default; the log-mode param is threaded in a later step. + mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), }; self.storage .register_subscription(&row) From 082c7d015bb68414845e65ba2be2f4d30df5e813 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 10:44:22 +0530 Subject: [PATCH 2/9] feat(core): route log publishes and compact acked messages publish_to_topic writes one topic_messages row when a topic has a log subscriber and fans out jobs only to fan-out subscribers. The retention sweep now compacts log messages every subscriber has acked past. --- crates/taskito-core/src/pubsub.rs | 87 ++++++++++++++++++- .../taskito-core/src/scheduler/maintenance.rs | 13 ++- 2 files changed, 94 insertions(+), 6 deletions(-) diff --git a/crates/taskito-core/src/pubsub.rs b/crates/taskito-core/src/pubsub.rs index aa1851ff..e2f8c644 100644 --- a/crates/taskito-core/src/pubsub.rs +++ b/crates/taskito-core/src/pubsub.rs @@ -1,12 +1,13 @@ -//! Topic pub/sub fan-out: one job per active subscription. +//! Topic pub/sub: fan-out (one job per active subscription) plus opt-in log +//! mode (one durable message per publish, pulled via a per-subscription cursor). //! //! Lives in the core (not the language shells) so every binding shares the -//! same fan-out semantics — in particular the idempotency-key salting, which -//! silently drops deliveries if a shell gets it wrong. +//! same semantics — in particular the idempotency-key salting, which silently +//! drops deliveries if a shell gets it wrong. use crate::error::Result; use crate::job::{Job, NewJob}; -use crate::storage::records::Subscription; +use crate::storage::records::{Subscription, SUBSCRIPTION_MODE_LOG}; use crate::storage::Storage; /// Queue-level fallback delivery settings, used when neither the publish call @@ -70,10 +71,32 @@ pub fn publish_to_topic(storage: &S, request: &PublishRequest) -> Re if subscriptions.is_empty() { return Ok(Vec::new()); } + + // A log topic stores one durable message that consumers pull via cursor; + // fan-out subscribers still get one job each. A topic may mix both modes, + // so write the log row once when any log subscriber exists, then fan out + // only to the fan-out subscribers. + if subscriptions + .iter() + .any(|s| s.mode == SUBSCRIPTION_MODE_LOG) + { + storage.publish_message( + &request.topic, + &request.payload, + request.metadata.as_deref(), + request.notes.as_deref(), + request.expires_at, + )?; + } + let jobs: Vec = subscriptions .iter() + .filter(|sub| sub.mode != SUBSCRIPTION_MODE_LOG) .map(|sub| delivery_job(request, sub)) .collect(); + if jobs.is_empty() { + return Ok(Vec::new()); + } if request.idempotency_key.is_none() { return storage.enqueue_batch(jobs); } @@ -230,6 +253,62 @@ mod tests { assert!(jobs.is_empty()); } + fn subscribe_log(storage: &SqliteStorage, topic: &str, name: &str) { + storage + .register_subscription(&NewSubscription { + topic: topic.to_string(), + subscription_name: name.to_string(), + task_name: String::new(), + queue: "default".to_string(), + active: true, + durable: true, + owner_worker_id: None, + created_at: now_millis(), + priority: None, + max_retries: None, + timeout_ms: None, + mode: SUBSCRIPTION_MODE_LOG.to_string(), + }) + .unwrap(); + } + + #[test] + fn log_topic_writes_one_message_and_no_jobs() { + let storage = SqliteStorage::in_memory().unwrap(); + subscribe_log(&storage, "events", "analytics"); + + // Two publishes → two log rows, zero fan-out jobs, regardless of readers. + assert!(publish_to_topic(&storage, &request("events", None)) + .unwrap() + .is_empty()); + assert!(publish_to_topic(&storage, &request("events", None)) + .unwrap() + .is_empty()); + let msgs = storage + .read_topic_messages("events", "analytics", 10) + .unwrap(); + assert_eq!(msgs.len(), 2); + } + + #[test] + fn mixed_topic_logs_once_and_fans_out_to_fanout_subs() { + let storage = SqliteStorage::in_memory().unwrap(); + subscribe(&storage, "events", "email", "send_email"); + subscribe_log(&storage, "events", "analytics"); + + let jobs = publish_to_topic(&storage, &request("events", None)).unwrap(); + // One fan-out job (email), one log message (analytics). + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].task_name, "send_email"); + assert_eq!( + storage + .read_topic_messages("events", "analytics", 10) + .unwrap() + .len(), + 1 + ); + } + #[test] fn publish_fans_out_one_job_per_subscription() { let storage = SqliteStorage::in_memory().unwrap(); diff --git a/crates/taskito-core/src/scheduler/maintenance.rs b/crates/taskito-core/src/scheduler/maintenance.rs index 76d09b12..08f07147 100644 --- a/crates/taskito-core/src/scheduler/maintenance.rs +++ b/crates/taskito-core/src/scheduler/maintenance.rs @@ -15,6 +15,9 @@ const PERIODIC_DEFAULT_MAX_RETRIES: i32 = 3; /// Default timeout for periodic tasks (ms). const PERIODIC_DEFAULT_TIMEOUT_MS: i64 = 300_000; +/// Max log messages compacted per retention tick, bounding each sweep. +const TOPIC_MESSAGE_PURGE_LIMIT: i64 = 10_000; + /// A retention window in ms as a compact human duration (`7d`, `12h`, `30m`). fn humanize_ms(ms: i64) -> String { const S: i64 = 1000; @@ -234,10 +237,16 @@ impl Scheduler { Some(c) => sweep("purge_task_logs", || self.storage.purge_task_logs(c)), None => 0, }; + // Log topics compact themselves: drop messages every subscriber has + // acked past, plus any expired. Bounded per tick like the other sweeps. + let topic_msgs = sweep("purge_topic_messages", || { + self.storage + .purge_topic_messages(now, TOPIC_MESSAGE_PURGE_LIMIT) + }); - if completed + dead + errors + metrics + logs > 0 { + if completed + dead + errors + metrics + logs + topic_msgs > 0 { info!( - "auto-cleanup: purged {completed} completed, {dead} dead, {errors} errors, {metrics} metrics, {logs} logs" + "auto-cleanup: purged {completed} completed, {dead} dead, {errors} errors, {metrics} metrics, {logs} logs, {topic_msgs} topic messages" ); } From 6d8371f2570caaf64f8f8a9e439ac3266cf69c07 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:19:50 +0530 Subject: [PATCH 3/9] feat(python): add log topic subscribe/read/ack API subscribe_log registers a durable cursor; read_topic pulls decoded messages after it; ack_topic advances it. Adds the TopicMessage dataclass and topic_log_stats. --- crates/taskito-python/src/py_queue/pubsub.rs | 66 ++++++++++- sdks/python/taskito/__init__.py | 2 + sdks/python/taskito/_taskito.pyi | 8 ++ sdks/python/taskito/mixins/pubsub.py | 88 ++++++++++++++- sdks/python/tests/core/test_pubsub_log.py | 110 +++++++++++++++++++ 5 files changed, 270 insertions(+), 4 deletions(-) create mode 100644 sdks/python/tests/core/test_pubsub_log.py diff --git a/crates/taskito-python/src/py_queue/pubsub.rs b/crates/taskito-python/src/py_queue/pubsub.rs index 9458611b..176d5bbc 100644 --- a/crates/taskito-python/src/py_queue/pubsub.rs +++ b/crates/taskito-python/src/py_queue/pubsub.rs @@ -28,6 +28,14 @@ type SubscriptionBacklogTuple = ( Option, ); +/// A log message surfaced to Python: +/// `(id, payload, metadata, notes, created_at)`. +type TopicMessageTuple = (String, Vec, Option, Option, i64); + +/// Per-log-subscription lag surfaced to Python: +/// `(topic, subscription, cursor, lag, oldest_unacked_age_ms)`. +type TopicLogStatsTuple = (String, String, Option, i64, Option); + #[pymethods] impl PyQueue { /// Insert or update a topic subscription (idempotent on topic + name). @@ -35,7 +43,7 @@ impl PyQueue { /// `priority`/`max_retries`/`timeout_ms` (already in milliseconds) persist /// the subscriber task's own delivery settings so a producer-only process /// can apply them without loading the task. - #[pyo3(signature = (topic, subscription_name, task_name, queue="default", durable=true, owner_worker_id=None, priority=None, max_retries=None, timeout_ms=None))] + #[pyo3(signature = (topic, subscription_name, task_name, queue="default", durable=true, owner_worker_id=None, priority=None, max_retries=None, timeout_ms=None, mode="fanout"))] #[allow(clippy::too_many_arguments)] pub fn register_subscription( &self, @@ -48,6 +56,7 @@ impl PyQueue { priority: Option, max_retries: Option, timeout_ms: Option, + mode: &str, ) -> PyResult<()> { // An unowned ephemeral row could never be reaped (cleanup keys off // live worker ids), so it would stay active forever. @@ -68,8 +77,7 @@ impl PyQueue { priority, max_retries, timeout_ms, - // Fan-out by default; the log-mode param is threaded in a later step. - mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), + mode: mode.to_string(), }; self.storage .register_subscription(&row) @@ -144,6 +152,58 @@ impl PyQueue { .collect() } + /// Messages after a log subscription's cursor, oldest first, `<= limit`: + /// `(id, payload, metadata, notes, created_at)`. Cursor resolved server-side. + pub fn read_topic_messages( + &self, + py: Python<'_>, + topic: &str, + subscription_name: &str, + limit: i64, + ) -> PyResult> { + let storage = &self.storage; + Ok(py + .detach(|| storage.read_topic_messages(topic, subscription_name, limit)) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))? + .into_iter() + .map(|m| (m.id, m.payload, m.metadata, m.notes, m.created_at)) + .collect()) + } + + /// Advance a log subscription's cursor to `cursor` (monotonic). Returns + /// false when nothing advanced. + pub fn ack_topic_cursor( + &self, + py: Python<'_>, + topic: &str, + subscription_name: &str, + cursor: &str, + ) -> PyResult { + let storage = &self.storage; + py.detach(|| storage.ack_topic_cursor(topic, subscription_name, cursor)) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + + /// Per-log-subscription lag snapshot: + /// `(topic, subscription, cursor, lag, oldest_unacked_age_ms)`. + pub fn topic_log_stats(&self, py: Python<'_>) -> PyResult> { + let storage = &self.storage; + Ok(py + .detach(|| storage.topic_log_stats()) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))? + .into_iter() + .map(|s| { + ( + s.topic, + s.subscription_name, + s.cursor, + s.lag, + s.oldest_unacked_age_ms, + ) + }) + .collect()) + } + /// Drop ephemeral subscriptions whose owning worker is gone. Runs on the /// heartbeat cadence, after `reap_dead_workers` has pruned the registry. /// diff --git a/sdks/python/taskito/__init__.py b/sdks/python/taskito/__init__.py index d7ff5245..757e0dda 100644 --- a/sdks/python/taskito/__init__.py +++ b/sdks/python/taskito/__init__.py @@ -45,6 +45,7 @@ from taskito.log_config import configure as configure_logging from taskito.mesh import MeshWorker from taskito.middleware import TaskMiddleware +from taskito.mixins.pubsub import TopicMessage from taskito.notes import MAX_NOTE_FIELDS from taskito.proxies.no_proxy import NoProxy from taskito.result import JobResult @@ -119,6 +120,7 @@ "TestMode", "TestResult", "TestResults", + "TopicMessage", "chain", "chord", "chunks", diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index 6ae724a2..1ec26601 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -228,6 +228,7 @@ class PyQueue: priority: int | None = None, max_retries: int | None = None, timeout_ms: int | None = None, + mode: str = "fanout", ) -> None: ... def list_subscriptions( self, topic: str | None = None @@ -240,6 +241,13 @@ class PyQueue: def topic_backlog_stats( self, ) -> list[tuple[str, str, str, str, bool, bool, int, int, int, int | None]]: ... + def read_topic_messages( + self, topic: str, subscription_name: str, limit: int + ) -> list[tuple[str, bytes, str | None, str | None, int]]: ... + def ack_topic_cursor(self, topic: str, subscription_name: str, cursor: str) -> bool: ... + def topic_log_stats( + self, + ) -> list[tuple[str, str, str | None, int, int | None]]: ... def publish( self, topic: str, diff --git a/sdks/python/taskito/mixins/pubsub.py b/sdks/python/taskito/mixins/pubsub.py index b774ad22..7b7dd52e 100644 --- a/sdks/python/taskito/mixins/pubsub.py +++ b/sdks/python/taskito/mixins/pubsub.py @@ -1,9 +1,10 @@ -"""Topic pub/sub: N independent subscribers, each delivered its own job.""" +"""Topic pub/sub: fan-out subscribers (one job each) or log subscribers (pull).""" from __future__ import annotations import json from collections.abc import Callable +from dataclasses import dataclass from typing import TYPE_CHECKING, Any from taskito.notes import validate_and_encode_notes @@ -15,6 +16,22 @@ from taskito.task import TaskWrapper +@dataclass(frozen=True) +class TopicMessage: + """One message pulled from a log topic. + + ``id`` is the cursor token to pass to :meth:`ack_topic`. ``args``/``kwargs`` + are the decoded publish payload; ``metadata``/``notes`` are the caller's. + """ + + id: str + args: tuple[Any, ...] + kwargs: dict[str, Any] + metadata: dict[str, Any] | None + notes: dict[str, Any] | None + created_at: int + + class QueuePubSubMixin: """Publish/subscribe over topics. @@ -89,6 +106,75 @@ def decorator(fn: Callable[..., Any]) -> TaskWrapper: return decorator + def subscribe_log(self, topic: str, name: str) -> None: + """Register a durable **log** subscription (a named cursor over ``topic``). + + Unlike :meth:`subscriber`, a log subscription has no handler: the topic's + publishes are stored once each and this consumer pulls them with + :meth:`read_topic`, advancing its cursor with :meth:`ack_topic`. Writes + immediately to storage, so a producer must have registered it (or called + this) before the publishes it wants to see. + """ + self._inner.register_subscription( + topic=topic, + subscription_name=name, + task_name="", + queue="default", + durable=True, + owner_worker_id=None, + mode="log", + ) + + def read_topic(self, topic: str, name: str, limit: int = 100) -> list[TopicMessage]: + """Pull up to ``limit`` messages after a log subscription's cursor. + + Oldest first, exclusive of the cursor. Decodes each payload with the + queue serializer. Returns an empty list when the consumer is caught up. + Delivery is at-least-once: process, then :meth:`ack_topic` the last id. + """ + messages = [] + for msg_id, payload, metadata, notes, created_at in self._inner.read_topic_messages( + topic, name, limit + ): + args, kwargs = self._serializer.loads(payload) + messages.append( + TopicMessage( + id=msg_id, + args=args, + kwargs=kwargs, + metadata=json.loads(metadata) if metadata is not None else None, + notes=json.loads(notes) if notes is not None else None, + created_at=created_at, + ) + ) + return messages + + def ack_topic(self, topic: str, name: str, cursor: str) -> bool: + """Advance a log subscription's cursor to ``cursor`` (a message id). + + A high-water mark: acking an id acks everything up to and including it. + Monotonic — acking an older id is a no-op. Returns False if nothing moved. + """ + return self._inner.ack_topic_cursor(topic, name, cursor) + + def topic_log_stats(self) -> list[dict[str, Any]]: + """Lag snapshot per log subscription. + + Each entry: ``topic``, ``subscription``, ``cursor`` (``None`` if unread), + ``lag`` (un-acked messages), ``oldest_unacked_age_ms`` (``None`` if + caught up). + """ + return [ + { + "topic": row[0], + "subscription": row[1], + "cursor": row[2], + "lag": row[3], + "oldest_unacked_age_ms": row[4], + } + for row in self._inner.topic_log_stats() + ] + def declare_subscriptions(self) -> None: """Write pending durable subscriptions to storage. diff --git a/sdks/python/tests/core/test_pubsub_log.py b/sdks/python/tests/core/test_pubsub_log.py new file mode 100644 index 00000000..f082ee2a --- /dev/null +++ b/sdks/python/tests/core/test_pubsub_log.py @@ -0,0 +1,110 @@ +"""Log topics (S28): one stored message per publish, pulled via a cursor.""" + +import threading +from typing import Any + +from taskito import Queue, TopicMessage + +PollUntil = Any # the conftest fixture's runtime type + + +class TestLogPublish: + def test_publish_stores_one_message_per_call(self, queue: Queue) -> None: + queue.subscribe_log("events", "analytics") + + # No fan-out jobs; one stored message per publish regardless of readers. + assert queue.publish("events", 1, kind="a") == [] + assert queue.publish("events", 2, kind="b") == [] + + msgs = queue.read_topic("events", "analytics") + assert [(m.args, m.kwargs) for m in msgs] == [((1,), {"kind": "a"}), ((2,), {"kind": "b"})] + assert all(isinstance(m, TopicMessage) for m in msgs) + + def test_read_is_empty_before_any_publish(self, queue: Queue) -> None: + queue.subscribe_log("events", "analytics") + assert queue.read_topic("events", "analytics") == [] + + def test_late_subscriber_misses_earlier_publishes(self, queue: Queue) -> None: + # A log subscription only sees messages published after it registered. + queue.publish("events", "early") # no subscriber yet → nothing stored + queue.subscribe_log("events", "late") + queue.publish("events", "seen") + msgs = queue.read_topic("events", "late") + assert [m.args for m in msgs] == [("seen",)] + + +class TestCursor: + def test_ack_advances_and_is_monotonic(self, queue: Queue) -> None: + queue.subscribe_log("events", "c") + for i in range(3): + queue.publish("events", i) + + msgs = queue.read_topic("events", "c") + assert [m.args[0] for m in msgs] == [0, 1, 2] + + # Ack through the middle message: the next read starts after it. + assert queue.ack_topic("events", "c", msgs[1].id) is True + remaining = queue.read_topic("events", "c") + assert [m.args[0] for m in remaining] == [2] + + # Acking an older cursor never rewinds. + assert queue.ack_topic("events", "c", msgs[0].id) is False + assert [m.args[0] for m in queue.read_topic("events", "c")] == [2] + + def test_unacked_read_is_at_least_once(self, queue: Queue) -> None: + queue.subscribe_log("events", "c") + queue.publish("events", "x") + # Reading without acking (e.g. a crash mid-process) re-delivers. + assert [m.args for m in queue.read_topic("events", "c")] == [("x",)] + assert [m.args for m in queue.read_topic("events", "c")] == [("x",)] + + def test_limit_bounds_the_page(self, queue: Queue) -> None: + queue.subscribe_log("events", "c") + for i in range(5): + queue.publish("events", i) + first = queue.read_topic("events", "c", limit=2) + assert [m.args[0] for m in first] == [0, 1] + queue.ack_topic("events", "c", first[-1].id) + assert [m.args[0] for m in queue.read_topic("events", "c", limit=2)] == [2, 3] + + +class TestMixedTopic: + def test_log_and_fanout_subscribers_coexist( + self, queue: Queue, run_worker: threading.Thread, poll_until: PollUntil + ) -> None: + seen: list[int] = [] + lock = threading.Lock() + + @queue.subscriber("events", name="worker") + def handle(n: int) -> None: + with lock: + seen.append(n) + + queue.declare_subscriptions() + queue.subscribe_log("events", "log") + + # One publish: the fan-out subscriber runs its job... + queue.publish("events", 7) + poll_until(lambda: seen == [7], message="fan-out subscriber should run") + + # ...and the same publish stored one log message for the log subscriber. + assert [m.args for m in queue.read_topic("events", "log")] == [(7,)] + + +class TestLogStats: + def test_lag_reflects_unacked(self, queue: Queue) -> None: + queue.subscribe_log("events", "c") + for i in range(3): + queue.publish("events", i) + + (stat,) = queue.topic_log_stats() + assert stat["topic"] == "events" + assert stat["subscription"] == "c" + assert stat["cursor"] is None + assert stat["lag"] == 3 + + msgs = queue.read_topic("events", "c") + queue.ack_topic("events", "c", msgs[-1].id) + (stat,) = queue.topic_log_stats() + assert stat["lag"] == 0 + assert stat["oldest_unacked_age_ms"] is None From beb994b11a5787b5fe77fa63103d75099c25f6c1 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:29:42 +0530 Subject: [PATCH 4/9] feat(node): add log topic subscribeLog/readTopic/ackTopic subscribeLog registers a durable cursor; readTopic pulls decoded messages after it; ackTopic advances it. Adds TopicMessage/TopicLogStat types and topicLogStats. --- crates/taskito-node/src/convert/mod.rs | 5 +- crates/taskito-node/src/convert/pubsub.rs | 48 ++++++++- crates/taskito-node/src/queue/pubsub.rs | 61 +++++++++++- sdks/node/src/native.ts | 2 + sdks/node/src/queue.ts | 54 ++++++++++ sdks/node/src/types.ts | 15 +++ sdks/node/test/core/pubsub-log.test.ts | 114 ++++++++++++++++++++++ 7 files changed, 293 insertions(+), 6 deletions(-) create mode 100644 sdks/node/test/core/pubsub-log.test.ts diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index f2fed243..e5faee71 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -22,7 +22,10 @@ pub use ops::{ }; pub use outcome::{outcome_to_js, JsOutcome}; pub use periodic::{periodic_to_js, JsPeriodicTask}; -pub use pubsub::{subscription_to_js, JsSubscription}; +pub use pubsub::{ + subscription_to_js, topic_log_stat_to_js, topic_message_to_js, JsSubscription, JsTopicLogStat, + JsTopicMessage, +}; pub use stats::{ dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js, JsDeadJob, JsDeadJobPage, JsJobError, JsMetric, JsStats, JsWorkerRow, diff --git a/crates/taskito-node/src/convert/pubsub.rs b/crates/taskito-node/src/convert/pubsub.rs index 884584c1..96fddaeb 100644 --- a/crates/taskito-node/src/convert/pubsub.rs +++ b/crates/taskito-node/src/convert/pubsub.rs @@ -1,7 +1,8 @@ -//! JS-facing shape of a topic subscription. +//! JS-facing shapes for topic subscriptions and log messages. +use napi::bindgen_prelude::Buffer; use napi_derive::napi; -use taskito_core::storage::records::Subscription; +use taskito_core::storage::records::{Subscription, TopicLogStats, TopicMessage}; /// A topic subscription: routes messages published to `topic` to `taskName` /// jobs on `queue`, one delivery per active subscription. @@ -26,3 +27,46 @@ pub fn subscription_to_js(row: Subscription) -> JsSubscription { durable: row.durable, } } + +/// A message pulled from a log topic. `id` is the cursor token to pass to +/// `ackTopic`; `payload` is the opaque published bytes. +#[napi(object)] +pub struct JsTopicMessage { + pub id: String, + pub payload: Buffer, + pub metadata: Option, + pub notes: Option, + pub created_at: i64, +} + +/// Convert a core [`TopicMessage`] into its JS-facing shape. +pub fn topic_message_to_js(msg: TopicMessage) -> JsTopicMessage { + JsTopicMessage { + id: msg.id, + payload: msg.payload.into(), + metadata: msg.metadata, + notes: msg.notes, + created_at: msg.created_at, + } +} + +/// Lag snapshot for one log subscription. +#[napi(object)] +pub struct JsTopicLogStat { + pub topic: String, + pub subscription: String, + pub cursor: Option, + pub lag: i64, + pub oldest_unacked_age_ms: Option, +} + +/// Convert a core [`TopicLogStats`] into its JS-facing shape. +pub fn topic_log_stat_to_js(stat: TopicLogStats) -> JsTopicLogStat { + JsTopicLogStat { + topic: stat.topic, + subscription: stat.subscription_name, + cursor: stat.cursor, + lag: stat.lag, + oldest_unacked_age_ms: stat.oldest_unacked_age_ms, + } +} diff --git a/crates/taskito-node/src/queue/pubsub.rs b/crates/taskito-node/src/queue/pubsub.rs index e5dd7120..8ae88c6a 100644 --- a/crates/taskito-node/src/queue/pubsub.rs +++ b/crates/taskito-node/src/queue/pubsub.rs @@ -12,7 +12,8 @@ use taskito_core::Storage; use super::JsQueue; use crate::config::PublishOptions; use crate::convert::{ - job_to_js, subscription_to_js, JsJob, JsSubscription, DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, + job_to_js, subscription_to_js, topic_log_stat_to_js, topic_message_to_js, JsJob, + JsSubscription, JsTopicLogStat, JsTopicMessage, DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS, }; use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err}; @@ -35,6 +36,7 @@ impl JsQueue { priority: Option, max_retries: Option, timeout_ms: Option, + mode: Option, ) -> Result<()> { // An owner-less ephemeral row could never be reaped — reject it before // it reaches storage. @@ -43,6 +45,9 @@ impl JsQueue { "an ephemeral subscription (durable=false) requires ownerWorkerId", )); } + let mode = mode.unwrap_or_else(|| { + taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string() + }); let storage = self.storage.clone(); spawn_blocking(move || { let row = NewSubscription { @@ -57,8 +62,7 @@ impl JsQueue { priority, max_retries, timeout_ms, - // Fan-out by default; the log-mode param is threaded in a later step. - mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), + mode, }; storage.register_subscription(&row).map_err(to_napi_err) }) @@ -66,6 +70,57 @@ impl JsQueue { .map_err(join_to_napi_err)? } + /// Messages after a log subscription's cursor, oldest first, `<= limit`. + /// Cursor resolved server-side; the read is exclusive of it. + #[napi] + pub async fn read_topic_messages( + &self, + topic: String, + subscription_name: String, + limit: i64, + ) -> Result> { + let storage = self.storage.clone(); + spawn_blocking(move || { + let messages = storage + .read_topic_messages(&topic, &subscription_name, limit) + .map_err(to_napi_err)?; + Ok(messages.into_iter().map(topic_message_to_js).collect()) + }) + .await + .map_err(join_to_napi_err)? + } + + /// Advance a log subscription's cursor (monotonic). Returns false when + /// nothing advanced. + #[napi] + pub async fn ack_topic_cursor( + &self, + topic: String, + subscription_name: String, + cursor: String, + ) -> Result { + let storage = self.storage.clone(); + spawn_blocking(move || { + storage + .ack_topic_cursor(&topic, &subscription_name, &cursor) + .map_err(to_napi_err) + }) + .await + .map_err(join_to_napi_err)? + } + + /// Lag snapshot per log subscription. + #[napi] + pub async fn topic_log_stats(&self) -> Result> { + let storage = self.storage.clone(); + spawn_blocking(move || { + let stats = storage.topic_log_stats().map_err(to_napi_err)?; + Ok(stats.into_iter().map(topic_log_stat_to_js).collect()) + }) + .await + .map_err(join_to_napi_err)? + } + /// List subscriptions — all of them, or only a topic's active ones. #[napi] pub async fn list_subscriptions(&self, topic: Option) -> Result> { diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index 444c3702..496b99f2 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -34,6 +34,8 @@ export type { JsSubscription, JsTaskInvocation, JsTaskLog, + JsTopicLogStat, + JsTopicMessage, JsWorkerRow, JsWorkflowAdvance, JsWorkflowNode, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index b0165618..3f4a44ab 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -43,6 +43,7 @@ import { } from "./resources"; import { CodecSerializer, + deserializeCall, JsonSerializer, type PayloadCodec, type Serializer, @@ -76,6 +77,8 @@ import type { TaskLog, TaskMap, TaskOptions, + TopicLogStat, + TopicMessage, WorkerInfo, WorkerRunOptions, } from "./types"; @@ -573,6 +576,57 @@ export class Queue { return this.native.listSubscriptions(topic); } + /** + * Register a durable **log** subscription: a named cursor over `topic`. Unlike + * `subscriber`, it has no handler — the topic's publishes are stored once each + * and this consumer pulls them with `readTopic`, advancing with `ackTopic`. + * Writes immediately, so register it before the publishes it should see. + */ + subscribeLog(topic: string, name: string): Promise { + return this.native.registerSubscription( + topic, + name, + "", + "default", + true, + undefined, + undefined, + undefined, + undefined, + "log", + ); + } + + /** + * Pull up to `limit` messages after a log subscription's cursor, oldest first + * and exclusive of it. Each `args` is the deserialized publish payload. Empty + * when caught up. At-least-once: process, then `ackTopic` the last `id`. + */ + async readTopic(topic: string, name: string, limit = 100): Promise { + const messages = await this.native.readTopicMessages(topic, name, limit); + return messages.map((msg) => ({ + id: msg.id, + args: deserializeCall(this.serializer, msg.payload), + metadata: msg.metadata === undefined ? undefined : JSON.parse(msg.metadata), + notes: msg.notes === undefined ? undefined : JSON.parse(msg.notes), + createdAt: msg.createdAt, + })); + } + + /** + * Advance a log subscription's cursor to `cursor` (a message id). A high-water + * mark — acking an id acks everything up to it. Monotonic; resolves false when + * nothing moved. + */ + ackTopic(topic: string, name: string, cursor: string): Promise { + return this.native.ackTopicCursor(topic, name, cursor); + } + + /** Lag snapshot per log subscription. */ + topicLogStats(): Promise { + return this.native.topicLogStats(); + } + /** * Drop ephemeral subscriptions whose owning worker is gone. Workers run * this on their heartbeat cadence; exposed for operational tooling. diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index c78ba97f..523c0889 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -22,10 +22,25 @@ export type { JsStats as Stats, JsSubscription as Subscription, JsTaskLog as TaskLog, + JsTopicLogStat as TopicLogStat, JsWorkerRow as WorkerInfo, MeshWorkerConfig, } from "./native"; +/** One message pulled from a log topic; `id` is the cursor token for `ackTopic`. */ +export interface TopicMessage { + /** Message id — pass to `ackTopic` to advance the cursor past it. */ + id: string; + /** Deserialized positional args from the `publish` call. */ + args: unknown[]; + /** Caller metadata, if any. */ + metadata?: Record; + /** Structured notes, if any. */ + notes?: Record; + /** Unix-millisecond publish time. */ + createdAt: number; +} + /** * Per-job enqueue options. Mirrors the native options, but `notes` is a * structured object here (validated and JSON-encoded before it reaches the diff --git a/sdks/node/test/core/pubsub-log.test.ts b/sdks/node/test/core/pubsub-log.test.ts new file mode 100644 index 00000000..f60c1cf2 --- /dev/null +++ b/sdks/node/test/core/pubsub-log.test.ts @@ -0,0 +1,114 @@ +// Log topics (S28): one stored message per publish, pulled via a cursor. + +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../../src/index"; + +let worker: Worker | undefined; +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function newQueue(): Queue { + return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-pubsub-log-")), "q.db") }); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 4000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return true; + await new Promise((r) => setTimeout(r, 20)); + } + return false; +} + +/** Narrow an array element to non-undefined (strict indexed access). */ +function must(value: T | undefined): T { + if (value === undefined) throw new Error("expected a value"); + return value; +} + +it("stores one message per publish, regardless of readers", async () => { + const queue = newQueue(); + await queue.subscribeLog("events", "analytics"); + + // No fan-out jobs; one stored message per publish. + expect(await queue.publish("events", [1])).toEqual([]); + expect(await queue.publish("events", [2])).toEqual([]); + + const msgs = await queue.readTopic("events", "analytics"); + expect(msgs.map((m) => m.args)).toEqual([[1], [2]]); +}); + +it("advances the cursor on ack and is monotonic", async () => { + const queue = newQueue(); + await queue.subscribeLog("events", "c"); + for (let i = 0; i < 3; i++) await queue.publish("events", [i]); + + const msgs = await queue.readTopic("events", "c"); + expect(msgs.map((m) => m.args[0])).toEqual([0, 1, 2]); + + // Ack through the middle: the next read starts after it. + expect(await queue.ackTopic("events", "c", must(msgs[1]).id)).toBe(true); + expect((await queue.readTopic("events", "c")).map((m) => m.args[0])).toEqual([2]); + + // Acking an older id never rewinds. + expect(await queue.ackTopic("events", "c", must(msgs[0]).id)).toBe(false); + expect((await queue.readTopic("events", "c")).map((m) => m.args[0])).toEqual([2]); +}); + +it("re-delivers un-acked messages (at-least-once)", async () => { + const queue = newQueue(); + await queue.subscribeLog("events", "c"); + await queue.publish("events", ["x"]); + expect((await queue.readTopic("events", "c")).map((m) => m.args)).toEqual([["x"]]); + expect((await queue.readTopic("events", "c")).map((m) => m.args)).toEqual([["x"]]); +}); + +it("bounds a read by limit", async () => { + const queue = newQueue(); + await queue.subscribeLog("events", "c"); + for (let i = 0; i < 5; i++) await queue.publish("events", [i]); + + const first = await queue.readTopic("events", "c", 2); + expect(first.map((m) => m.args[0])).toEqual([0, 1]); + await queue.ackTopic("events", "c", must(first[first.length - 1]).id); + expect((await queue.readTopic("events", "c", 2)).map((m) => m.args[0])).toEqual([2, 3]); +}); + +it("reports lag per log subscription", async () => { + const queue = newQueue(); + await queue.subscribeLog("events", "c"); + for (let i = 0; i < 3; i++) await queue.publish("events", [i]); + + let stat = must((await queue.topicLogStats())[0]); + expect(stat.subscription).toBe("c"); + expect(stat.cursor).toBeUndefined(); + expect(stat.lag).toBe(3); + + const msgs = await queue.readTopic("events", "c"); + await queue.ackTopic("events", "c", must(msgs[msgs.length - 1]).id); + stat = must((await queue.topicLogStats())[0]); + expect(stat.lag).toBe(0); +}); + +it("lets log and fan-out subscribers coexist on one topic", async () => { + const queue = newQueue(); + const seen: number[] = []; + queue.subscriber("events", "worker", (n: number) => { + seen.push(n); + }); + await queue.declareSubscriptions(); + await queue.subscribeLog("events", "log"); + + worker = queue.runWorker({ concurrency: 1 }); + await queue.publish("events", [7]); + + expect(await waitFor(() => seen.length === 1)).toBe(true); + expect(seen).toEqual([7]); + // The same publish stored one log message. + expect((await queue.readTopic("events", "log")).map((m) => m.args)).toEqual([[7]]); +}); From 34cd788e0ca00aec83b6d91f3d81b0fb4ebf549e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:48:16 +0530 Subject: [PATCH 5/9] feat(java): add log topic subscribeLog/readTopic/ackTopic subscribeLog registers a durable cursor; readTopic pulls messages after it (payload as bytes); ackTopic advances it. Adds TopicMessage/TopicLogStat models and topicLogStats. --- Cargo.lock | 1 + crates/taskito-java/Cargo.toml | 1 + crates/taskito-java/src/convert.rs | 54 +++++- crates/taskito-java/src/queue/pubsub.rs | 80 +++++++- .../org/byteveda/taskito/DefaultTaskito.java | 53 ++++++ .../java/org/byteveda/taskito/Taskito.java | 33 ++++ .../taskito/internal/JniQueueBackend.java | 44 ++++- .../taskito/internal/NativeQueue.java | 15 +- .../byteveda/taskito/model/TopicLogStat.java | 35 ++++ .../byteveda/taskito/model/TopicMessage.java | 34 ++++ .../byteveda/taskito/spi/QueueBackend.java | 51 +++++ .../byteveda/taskito/core/PubSubLogTest.java | 180 ++++++++++++++++++ 12 files changed, 572 insertions(+), 9 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/TopicLogStat.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/TopicMessage.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/core/PubSubLogTest.java diff --git a/Cargo.lock b/Cargo.lock index f01ff156..cc2a3d64 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1855,6 +1855,7 @@ name = "taskito-java" version = "0.20.0" dependencies = [ "async-trait", + "base64", "crossbeam-channel", "gethostname", "jni", diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml index a59c3ce4..cf3643b4 100644 --- a/crates/taskito-java/Cargo.toml +++ b/crates/taskito-java/Cargo.toml @@ -22,6 +22,7 @@ mesh = ["dep:taskito-mesh"] taskito-core = { path = "../taskito-core" } taskito-workflows = { path = "../taskito-workflows", optional = true } taskito-mesh = { path = "../taskito-mesh", optional = true } +base64 = "0.22" jni = "0.21" serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index 451eab90..8c34ad4d 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -3,13 +3,14 @@ //! Option and filter structs cross as JSON strings (decoded here); opaque job //! payloads cross as raw `byte[]` and are never interpreted by the core. +use base64::Engine; use serde::{Deserialize, Serialize}; use taskito_core::job::{now_millis, Job, NewJob}; use taskito_core::pubsub::{DeliveryDefaults, PublishRequest}; use taskito_core::resilience::circuit_breaker::CircuitState; use taskito_core::storage::records::{ CircuitBreakerState, JobError, LockInfo, PeriodicTask, ReplayEntry, Subscription, TaskLogEntry, - TaskMetric, WorkerInfo, + TaskMetric, TopicLogStats, TopicMessage, WorkerInfo, }; use taskito_core::storage::{DeadJob, QueueStats}; @@ -242,6 +243,57 @@ impl<'a> From<&'a Subscription> for SubscriptionView<'a> { } } +/// Java-facing view of one log-topic message. The opaque payload crosses as a +/// base64 string (JSON can't carry raw bytes); `metadata`/`notes` stay JSON +/// strings the SDK parses back into maps. `created_at` is Unix milliseconds. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TopicMessageView { + /// Message id — the cursor token passed to `ackTopicCursor`. + pub id: String, + /// Opaque payload, base64-encoded. + pub payload: String, + pub metadata: Option, + pub notes: Option, + pub created_at: i64, +} + +impl From<&TopicMessage> for TopicMessageView { + fn from(m: &TopicMessage) -> Self { + Self { + id: m.id.clone(), + payload: base64::engine::general_purpose::STANDARD.encode(&m.payload), + metadata: m.metadata.clone(), + notes: m.notes.clone(), + created_at: m.created_at, + } + } +} + +/// Java-facing lag snapshot for one log subscription. `cursor` is null until the +/// first ack; `oldest_unacked_age_ms` is null once the subscription is caught up. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TopicLogStatsView { + pub topic: String, + pub subscription: String, + pub cursor: Option, + pub lag: i64, + pub oldest_unacked_age_ms: Option, +} + +impl From<&TopicLogStats> for TopicLogStatsView { + fn from(s: &TopicLogStats) -> Self { + Self { + topic: s.topic.clone(), + subscription: s.subscription_name.clone(), + cursor: s.cursor.clone(), + lag: s.lag, + oldest_unacked_age_ms: s.oldest_unacked_age_ms, + } + } +} + /// Java-facing view of a task's circuit-breaker state. `state` is the lowercase wire /// string (`closed`/`open`/`half_open`); timestamps are Unix milliseconds. #[derive(Serialize)] diff --git a/crates/taskito-java/src/queue/pubsub.rs b/crates/taskito-java/src/queue/pubsub.rs index 7d41ad08..3bc8f219 100644 --- a/crates/taskito-java/src/queue/pubsub.rs +++ b/crates/taskito-java/src/queue/pubsub.rs @@ -15,6 +15,7 @@ use taskito_core::Storage; use crate::backend; use crate::convert::{ build_publish_request, parse_json, to_json, JobView, PublishOptions, SubscriptionView, + TopicLogStatsView, TopicMessageView, }; use crate::ffi::{guard, new_string, read_bytes, read_optional_string, read_string}; @@ -22,10 +23,11 @@ use super::{borrow_queue, to_jboolean}; /// `void registerSubscription(long handle, String topic, String subscriptionName, /// String taskName, String queue, boolean durable, String ownerWorkerIdOrNull, -/// int priority, int maxRetries, long timeoutMs)` — insert or update a -/// subscription (idempotent on topic + name). The three delivery-setting +/// int priority, int maxRetries, long timeoutMs, String mode)` — insert or update +/// a subscription (idempotent on topic + name). The three delivery-setting /// numerics use `i32::MIN` / `i64::MIN` as the "unset → queue default" sentinel, -/// since JNI carries primitives, not boxed nullables. +/// since JNI carries primitives, not boxed nullables. `mode` is `"fanout"` +/// (one job per publish) or `"log"` (append-once + cursor). #[no_mangle] #[allow(clippy::too_many_arguments)] pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSubscription< @@ -43,6 +45,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu priority: jint, max_retries: jint, timeout_ms: jlong, + mode: JString<'local>, ) { guard(&mut env, (), |env| { let queue_handle = unsafe { borrow_queue(handle) }; @@ -51,6 +54,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu let task_name = read_string(env, &task_name)?; let queue = read_string(env, &queue)?; let owner_worker_id = read_optional_string(env, &owner_worker_id)?; + let mode = read_string(env, &mode)?; // An ownerless ephemeral row would never be reaped (the reaper matches // on owner) yet keeps receiving deliveries — reject it up front. if durable == 0 && owner_worker_id.is_none() { @@ -72,8 +76,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu priority: (priority != jint::MIN).then_some(priority), max_retries: (max_retries != jint::MIN).then_some(max_retries), timeout_ms: (timeout_ms != jlong::MIN).then_some(timeout_ms), - // Fan-out by default; the log-mode param is threaded in a later step. - mode: taskito_core::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(), + mode, }; queue_handle.storage.register_subscription(&row)?; Ok(()) @@ -184,3 +187,70 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_publish<'l new_string(env, to_json(&views)?) }) } + +/// `String readTopicMessages(long handle, String topic, String subscriptionName, +/// long limit)` — pull up to `limit` messages after a log subscription's cursor, +/// oldest first and exclusive of it, as a JSON array of `TopicMessageView`. +/// At-least-once: reading without acking re-delivers. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_readTopicMessages<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + topic: JString<'local>, + subscription_name: JString<'local>, + limit: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let topic = read_string(env, &topic)?; + let subscription_name = read_string(env, &subscription_name)?; + let messages = queue + .storage + .read_topic_messages(&topic, &subscription_name, limit)?; + let views: Vec = messages.iter().map(TopicMessageView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `boolean ackTopicCursor(long handle, String topic, String subscriptionName, +/// String cursor)` — advance the subscription's cursor to `cursor` (a message +/// id). A monotonic high-water mark: acking an id acks everything up to it, and +/// acking an older id is a no-op. Returns false when nothing moved. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_ackTopicCursor<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + topic: JString<'local>, + subscription_name: JString<'local>, + cursor: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let topic = read_string(env, &topic)?; + let subscription_name = read_string(env, &subscription_name)?; + let cursor = read_string(env, &cursor)?; + Ok(to_jboolean(queue.storage.ack_topic_cursor( + &topic, + &subscription_name, + &cursor, + )?)) + }) +} + +/// `String topicLogStats(long handle)` — a JSON array of `TopicLogStatsView`, one +/// lag snapshot per log subscription. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_topicLogStats<'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_log_stats()?; + let views: Vec = stats.iter().map(TopicLogStatsView::from).collect(); + new_string(env, to_json(&views)?) + }) +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 7d9b884f..293299fa 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -1,5 +1,6 @@ package org.byteveda.taskito; +import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; import java.time.Duration; @@ -45,6 +46,8 @@ import org.byteveda.taskito.model.Subscription; import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; +import org.byteveda.taskito.model.TopicLogStat; +import org.byteveda.taskito.model.TopicMessage; import org.byteveda.taskito.model.WorkerInfo; import org.byteveda.taskito.model.WorkflowRunInfo; import org.byteveda.taskito.predicates.EnqueueDecision; @@ -887,6 +890,56 @@ public List listTopics() { return new ArrayList<>(topics); } + @Override + public void subscribeLog(String topic, String name) { + // A log subscription has no handler: durable, no owner, no delivery + // settings — just a named cursor over the topic (mode = "log"). + backend.registerSubscription(topic, name, "", "default", true, null, null, null, null, "log"); + } + + @Override + public List readTopic(String topic, String name) { + return readTopic(topic, name, 100); + } + + @Override + public List readTopic(String topic, String name, int limit) { + List rows = + decodeList(backend.readTopicMessagesJson(topic, name, limit), TopicMessageRow.class); + List messages = new ArrayList<>(rows.size()); + for (TopicMessageRow row : rows) { + messages.add(new TopicMessage( + row.id(), row.payload(), parseJsonMap(row.metadata()), parseJsonMap(row.notes()), row.createdAt())); + } + return messages; + } + + @Override + public boolean ackTopic(String topic, String name, String cursor) { + return backend.ackTopicCursor(topic, name, cursor); + } + + @Override + public List topicLogStats() { + return decodeList(backend.topicLogStatsJson(), TopicLogStat.class); + } + + /** + * Native wire view of a log message: the payload crosses as base64 (Jackson + * decodes it into {@code byte[]}); metadata/notes are JSON strings this SDK + * parses back into maps. + */ + private record TopicMessageRow( + @JsonProperty("id") String id, + @JsonProperty("payload") byte[] payload, + @JsonProperty("metadata") String metadata, + @JsonProperty("notes") String notes, + @JsonProperty("createdAt") long createdAt) {} + + private static Map parseJsonMap(String json) { + return json == null ? null : decodeMap(json, Object.class); + } + // ── Workflows ─────────────────────────────────────────────────── @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index e8cf309a..45b43f00 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -35,6 +35,8 @@ import org.byteveda.taskito.model.Subscription; import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; +import org.byteveda.taskito.model.TopicLogStat; +import org.byteveda.taskito.model.TopicMessage; import org.byteveda.taskito.model.WorkerInfo; import org.byteveda.taskito.model.WorkflowRunInfo; import org.byteveda.taskito.predicates.EnqueueGate; @@ -414,6 +416,37 @@ static Builder builder() { /** Distinct topics that currently have at least one subscription. */ List listTopics(); + /** + * Register a durable log subscription: a named cursor over {@code topic}. + * Unlike {@link #subscribe(String, Task)} it has no handler — the topic's + * publishes are stored once each and this consumer pulls them with + * {@link #readTopic(String, String)}, advancing with {@link #ackTopic}. Writes + * immediately, so register it before the publishes it should see. (On Redis the + * log is backed by a Stream, but that is transparent.) + */ + void subscribeLog(String topic, String name); + + /** As {@link #readTopic(String, String, int)} with a default limit of 100. */ + List readTopic(String topic, String name); + + /** + * Pull up to {@code limit} messages after a log subscription's cursor, oldest + * first and exclusive of it. Empty when caught up. Decode each {@code payload} + * with the queue's serializer. At-least-once: process, then {@link #ackTopic} + * the last {@code id}. + */ + List readTopic(String topic, String name, int limit); + + /** + * Advance a log subscription's cursor to {@code cursor} (a message id). A + * high-water mark: acking an id acks everything up to it. Monotonic — acking an + * older id is a no-op. Returns false when nothing moved. + */ + boolean ackTopic(String topic, String name, String cursor); + + /** Lag snapshot per log subscription (cursor position and un-acked backlog). */ + List topicLogStats(); + // ── Workflows ─────────────────────────────────────────────────── /** Submit a workflow DAG; returns a handle to the run. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 2c0d8acc..16adbb95 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -325,6 +325,32 @@ public void registerSubscription( Integer priority, Integer maxRetries, Long timeoutMs) { + // Default to fan-out; the mode-carrying overload handles log subscriptions. + registerSubscription( + topic, + subscriptionName, + taskName, + queue, + durable, + ownerWorkerIdOrNull, + priority, + maxRetries, + timeoutMs, + "fanout"); + } + + @Override + public void registerSubscription( + String topic, + String subscriptionName, + String taskName, + String queue, + boolean durable, + String ownerWorkerIdOrNull, + Integer priority, + Integer maxRetries, + Long timeoutMs, + String mode) { // JNI carries primitives, not boxed nullables — a null delivery setting // crosses as the MIN sentinel the native side reads as "queue default". int nativePriority = priority == null ? Integer.MIN_VALUE : priority; @@ -341,7 +367,8 @@ public void registerSubscription( ownerWorkerIdOrNull, nativePriority, nativeMaxRetries, - nativeTimeoutMs); + nativeTimeoutMs, + mode); return null; }); } @@ -371,6 +398,21 @@ public String publishJson(String topic, byte[] payload, String optionsJson) { return withOpenHandle(() -> NativeQueue.publish(handle, topic, payload, optionsJson)); } + @Override + public String readTopicMessagesJson(String topic, String subscriptionName, long limit) { + return withOpenHandle(() -> NativeQueue.readTopicMessages(handle, topic, subscriptionName, limit)); + } + + @Override + public boolean ackTopicCursor(String topic, String subscriptionName, String cursor) { + return withOpenHandle(() -> NativeQueue.ackTopicCursor(handle, topic, subscriptionName, cursor)); + } + + @Override + public String topicLogStatsJson() { + return withOpenHandle(() -> NativeQueue.topicLogStats(handle)); + } + @Override public String submitWorkflow( String name, diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java index 0bf93b78..2130106a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -146,7 +146,8 @@ public static native long registerPeriodic( * Insert or update a topic subscription (idempotent on topic + name). The * subscriber task's delivery settings persist on the row; {@code priority}/ * {@code maxRetries} of {@link Integer#MIN_VALUE} and {@code timeoutMs} of - * {@link Long#MIN_VALUE} mean "unset — take the queue default". + * {@link Long#MIN_VALUE} mean "unset — take the queue default". {@code mode} is + * {@code "fanout"} (one job per publish) or {@code "log"} (append-once + cursor). */ public static native void registerSubscription( long handle, @@ -158,7 +159,8 @@ public static native void registerSubscription( String ownerWorkerIdOrNull, int priority, int maxRetries, - long timeoutMs); + long timeoutMs, + String mode); /** A JSON array of subscriptions — all of them, or only a topic's active ones. */ public static native String listSubscriptions(long handle, String topicOrNull); @@ -176,6 +178,15 @@ public static native boolean setSubscriptionActive( /** Publish to a topic; returns the created delivery jobs as a JSON array. */ public static native String publish(long handle, String topic, byte[] payload, String optionsJson); + /** Pull messages after a log subscription's cursor, as a JSON array of message views. */ + public static native String readTopicMessages(long handle, String topic, String subscriptionName, long limit); + + /** Advance a log subscription's cursor (monotonic); false if nothing moved. */ + public static native boolean ackTopicCursor(long handle, String topic, String subscriptionName, String cursor); + + /** A JSON array of per-log-subscription lag snapshots. */ + public static native String topicLogStats(long handle); + // ── Worker ────────────────────────────────────────────────────── /** Start a worker; returns its handle. {@code bridge} is a {@code WorkerBridge}. */ public static native long runWorker(long handle, Object bridge, String optionsJson); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/TopicLogStat.java b/sdks/java/src/main/java/org/byteveda/taskito/model/TopicLogStat.java new file mode 100644 index 00000000..e7c3d3f7 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/TopicLogStat.java @@ -0,0 +1,35 @@ +package org.byteveda.taskito.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Lag snapshot for one log subscription: how far its cursor trails the topic. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class TopicLogStat { + public final String topic; + public final String subscription; + + /** Current read cursor (last-acked id), or {@code null} if nothing acked yet. */ + public final String cursor; + + /** Number of messages after the cursor still to be consumed. */ + public final long lag; + + /** Age (ms) of the oldest un-acked message, or {@code null} when caught up. */ + public final Long oldestUnackedAgeMs; + + @JsonCreator + public TopicLogStat( + @JsonProperty("topic") String topic, + @JsonProperty("subscription") String subscription, + @JsonProperty("cursor") String cursor, + @JsonProperty("lag") long lag, + @JsonProperty("oldestUnackedAgeMs") Long oldestUnackedAgeMs) { + this.topic = topic; + this.subscription = subscription; + this.cursor = cursor; + this.lag = lag; + this.oldestUnackedAgeMs = oldestUnackedAgeMs; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/TopicMessage.java b/sdks/java/src/main/java/org/byteveda/taskito/model/TopicMessage.java new file mode 100644 index 00000000..5ca76f8a --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/TopicMessage.java @@ -0,0 +1,34 @@ +package org.byteveda.taskito.model; + +import java.util.Map; + +/** + * One message pulled from a log topic. {@link #id} is the cursor token to pass to + * {@code ackTopic}. {@link #payload} is the opaque publish payload — decode it + * with the queue's serializer. {@code createdAt} is Unix milliseconds. + */ +public final class TopicMessage { + /** Message id; doubles as the cursor token for {@code ackTopic}. */ + public final String id; + + /** Opaque publish payload — decode with the queue's serializer. */ + public final byte[] payload; + + /** Caller metadata, or {@code null} when none was set. */ + public final Map metadata; + + /** Structured notes, or {@code null} when none were set. */ + public final Map notes; + + /** Unix-millisecond publish time. */ + public final long createdAt; + + public TopicMessage( + String id, byte[] payload, Map metadata, Map notes, long createdAt) { + this.id = id; + this.payload = payload; + this.metadata = metadata; + this.notes = notes; + this.createdAt = createdAt; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index 47dd280e..24e7c2e3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -235,6 +235,38 @@ default void registerSubscription( registerSubscription(topic, subscriptionName, taskName, queue, durable, ownerWorkerIdOrNull); } + /** + * Insert or update a subscription with an explicit delivery {@code mode}: + * {@code "fanout"} (one job per publish) or {@code "log"} (append-once + a + * per-subscription cursor pulled via {@link #readTopicMessagesJson}). + * + *

The default drops {@code mode} and delegates to the fan-out form, so a + * backend that predates log topics still registers a fan-out subscription + * instead of throwing. + */ + default void registerSubscription( + String topic, + String subscriptionName, + String taskName, + String queue, + boolean durable, + String ownerWorkerIdOrNull, + Integer priority, + Integer maxRetries, + Long timeoutMs, + String mode) { + registerSubscription( + topic, + subscriptionName, + taskName, + queue, + durable, + ownerWorkerIdOrNull, + priority, + maxRetries, + timeoutMs); + } + /** A JSON array of subscriptions — all of them, or only a topic's active ones. */ default String listSubscriptionsJson(String topicOrNull) { throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); @@ -263,6 +295,25 @@ default String publishJson(String topic, byte[] payload, String optionsJson) { throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); } + /** + * Pull up to {@code limit} messages after a log subscription's cursor (oldest + * first, exclusive), as a JSON array of message views. At-least-once: reading + * without acking re-delivers. + */ + default String readTopicMessagesJson(String topic, String subscriptionName, long limit) { + throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); + } + + /** Advance a log subscription's cursor to {@code cursor} (monotonic); false if nothing moved. */ + default boolean ackTopicCursor(String topic, String subscriptionName, String cursor) { + throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); + } + + /** A JSON array of per-log-subscription lag snapshots. */ + default String topicLogStatsJson() { + throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); + } + // ── Workflows ─────────────────────────────────────────────────── // Optional capability: default to throwing so existing custom backends keep // compiling and fail explicitly only when workflows are actually used. diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubLogTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubLogTest.java new file mode 100644 index 00000000..d0e3202a --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubLogTest.java @@ -0,0 +1,180 @@ +package org.byteveda.taskito.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobStatus; +import org.byteveda.taskito.model.TopicLogStat; +import org.byteveda.taskito.model.TopicMessage; +import org.byteveda.taskito.pubsub.PublishOptions; +import org.byteveda.taskito.serialization.JsonSerializer; +import org.byteveda.taskito.serialization.Serializer; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * S28 — log topics: one stored message per publish, pulled via a per-subscription + * cursor. The core cursor algorithm is unit-tested in Rust; this drives the Java + * surface end-to-end (subscribeLog / readTopic / ackTopic / topicLogStats). + */ +class PubSubLogTest { + + // readTopic returns the raw payload bytes (like getResult); decode them with + // the queue's default serializer to recover the published value. + private static final Serializer SERIALIZER = new JsonSerializer(); + + private Taskito open(Path dir) { + return Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open(); + } + + @Test + void publishStoresOneMessagePerCallReadInOrder(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "analytics"); + + // No fan-out jobs: one stored message per publish, regardless of readers. + assertTrue(queue.publish( + "events", + "a", + PublishOptions.builder() + .notes(Map.of("tenant", "acme")) + .build()) + .isEmpty()); + assertTrue(queue.publish("events", "b").isEmpty()); + assertEquals(0, queue.stats().pending); + + List msgs = queue.readTopic("events", "analytics"); + assertEquals(List.of("a", "b"), decodeAll(msgs)); + // The caller's notes ride along on the stored message. + assertEquals("acme", msgs.get(0).notes.get("tenant")); + } + } + + @Test + void readIsEmptyBeforeAnyPublish(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "analytics"); + assertTrue(queue.readTopic("events", "analytics").isEmpty()); + } + } + + @Test + void ackAdvancesCursorAndIsMonotonic(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "c"); + for (int i = 0; i < 3; i++) { + queue.publish("events", String.valueOf(i)); + } + List msgs = queue.readTopic("events", "c"); + assertEquals(List.of("0", "1", "2"), decodeAll(msgs)); + + // Ack through the middle message: the next read starts after it. + assertTrue(queue.ackTopic("events", "c", msgs.get(1).id)); + assertEquals(List.of("2"), decodeAll(queue.readTopic("events", "c"))); + + // Acking an older cursor never rewinds. + assertFalse(queue.ackTopic("events", "c", msgs.get(0).id)); + assertEquals(List.of("2"), decodeAll(queue.readTopic("events", "c"))); + } + } + + @Test + void unackedReadIsAtLeastOnce(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "c"); + queue.publish("events", "x"); + // Reading without acking (e.g. a crash mid-process) re-delivers. + assertEquals(List.of("x"), decodeAll(queue.readTopic("events", "c"))); + assertEquals(List.of("x"), decodeAll(queue.readTopic("events", "c"))); + } + } + + @Test + void limitBoundsThePage(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "c"); + for (int i = 0; i < 5; i++) { + queue.publish("events", String.valueOf(i)); + } + List first = queue.readTopic("events", "c", 2); + assertEquals(List.of("0", "1"), decodeAll(first)); + queue.ackTopic("events", "c", first.get(first.size() - 1).id); + assertEquals(List.of("2", "3"), decodeAll(queue.readTopic("events", "c", 2))); + } + } + + @Test + void lagReflectsUnacked(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + queue.subscribeLog("events", "c"); + for (int i = 0; i < 3; i++) { + queue.publish("events", String.valueOf(i)); + } + + TopicLogStat stat = single(queue.topicLogStats()); + assertEquals("events", stat.topic); + assertEquals("c", stat.subscription); + assertNull(stat.cursor); + assertEquals(3, stat.lag); + + List msgs = queue.readTopic("events", "c"); + queue.ackTopic("events", "c", msgs.get(msgs.size() - 1).id); + TopicLogStat caughtUp = single(queue.topicLogStats()); + assertEquals(0, caughtUp.lag); + assertNull(caughtUp.oldestUnackedAgeMs); + } + } + + @Test + @Timeout(30) + void logAndFanoutSubscribersCoexist(@TempDir Path dir) { + Task onEvent = Task.of("pubsublog.on_event", String.class); + try (Taskito queue = open(dir)) { + queue.subscribe("events", onEvent); // fan-out: one job per publish + queue.subscribeLog("events", "log"); // log: one stored message per publish + + try (Worker worker = + queue.worker().handle(onEvent, payload -> payload).start()) { + List deliveries = queue.publish("events", "seen"); + + // The fan-out subscriber ran its job to completion... + assertEquals(1, deliveries.size()); + Job done = queue.awaitJob(deliveries.get(0).id, Duration.ofSeconds(10)) + .orElseThrow(); + assertEquals(JobStatus.COMPLETE, done.status); + assertEquals("seen", queue.getResult(done.id, String.class).orElseThrow()); + + // ...and the same publish stored one message for the log subscriber. + assertEquals(List.of("seen"), decodeAll(queue.readTopic("events", "log"))); + } + } + } + + private static List decodeAll(List msgs) { + List out = new ArrayList<>(msgs.size()); + for (TopicMessage msg : msgs) { + out.add(SERIALIZER.deserialize(msg.payload, String.class)); + } + return out; + } + + private static T single(List list) { + assertEquals(1, list.size()); + return list.get(0); + } +} From 80e7d96dae1a9587abc6d62b799b4494d1974ea5 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:58:09 +0530 Subject: [PATCH 6/9] docs(pubsub): document log topics + cross-SDK contract Add the Topic pub/sub section to BINDING_CONTRACT (fanout vs log mode, cursor/ack/at-least-once semantics, salting test vector) and log-topic docs across the shared guide and Python/Node/Java API references. --- crates/taskito-core/BINDING_CONTRACT.md | 42 ++++++ .../docs/java/api-reference/queue/pubsub.mdx | 19 +++ .../docs/node/api-reference/queue/pubsub.mdx | 19 +++ .../python/api-reference/queue/pubsub.mdx | 67 +++++++++ .../docs/shared/guides/core/pubsub.mdx | 131 +++++++++++++++++- 5 files changed, 275 insertions(+), 3 deletions(-) diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 87c8794e..7c9eff35 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/crates/taskito-core/BINDING_CONTRACT.md @@ -126,6 +126,45 @@ input errtype `BoomError`, message `it broke`, traceback `["frame1", "frame2"]` `{"errtype":"BoomError","message":"it broke","traceback":["frame1","frame2"]}` (JSON with those three keys in that order, no extra whitespace). +## Topic pub/sub (cross-SDK) +A subscription (`topic_subscriptions`) routes a topic's publishes to a subscriber. +Its `mode` column selects the delivery model; the core (`pubsub::publish_to_topic`) +owns both, so a shell only marshals arguments. + +| `mode` | On publish | Consumption | +|--------|-----------|-------------| +| `fanout` (default) | one ordinary `jobs` row per active subscriber | the normal dequeue/dispatch path — each delivery is a job | +| `log` | one `topic_messages` row for the whole publish (O(1)) | the shell **pulls**: `read_topic_messages` → process → `ack_topic_cursor` | + +A topic may mix both: a publish writes the log row once **and** fans out to any +fan-out subscribers. + +**Fan-out `unique_key` salting** — the `jobs` unique index is global, so a shell +that keys a publish MUST salt the key per subscriber or all but one delivery is +deduped away. Salt = `::::` (length +prefixes make it injective). Done in the core; a shell that builds `NewJob` rows +itself must match it. + +**Log cursor rules** (`read_topic_messages`/`ack_topic_cursor`): +- The **cursor is an opaque, monotonic token** — a shell stores and passes back + the message `id`, never parses it. Its format differs by backend (UUIDv7 on the + Diesel backends, a `-` stream id on Redis), like the `get_task_logs_after` + cursor note. +- **Reads are exclusive** of the cursor and ordered oldest-first; the cursor is + resolved server-side from the subscription row. +- **Ack is a high-water mark**: acking id `X` acks every message `≤ X`. Monotonic — + acking an older/equal id is a no-op (returns `false`). +- Delivery is **at-least-once**: a consumer that reads but dies before acking + re-reads those messages. There is no per-message ack. +- **Retention** is min-cursor compaction: a message is dropped once every log + subscriber on its topic has acked past it (Diesel deletes `id <= min(cursor)`; + Redis `XTRIM MINID`). A topic with an unread subscriber keeps its backlog. + Diesel also honors an optional per-message `expires_at`; Redis does not. + +**Test vector** (assert byte-exact in each shell that salts keys itself): +key `evt-42`, topic `orders`, subscription `email` → +`evt-42::6:5:ordersemail`. + ## Types the shell produces / consumes - **`Job`** — `job.rs`. Fields incl. `id`, `queue`, `task_name`, `payload: Vec` (opaque), `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`, @@ -149,6 +188,9 @@ Grouped by concern (enumerated, not exhaustive — read the trait): - **Cancellation**: `request_cancel`, `is_cancel_requested`, `mark_cancelled`. - **Resilience**: `try_acquire_token` (rate limit), `count_running_by_task`, `stats_by_queue`. - **Dead-letter**: `move_to_dlq`, `list_dead`, `retry_dead`. +- **Topic pub/sub**: `register_subscription`, `list_subscriptions_for_topic`, `unsubscribe`, + `publish_message`, `read_topic_messages`, `ack_topic_cursor`, `topic_log_stats` (see + [Topic pub/sub](#topic-pubsub-cross-sdk)). ## Lifecycle the shell drives - **Startup**: `register_worker(worker_id, queues, …)`. Worker ID is generated by the shell. diff --git a/docs/content/docs/java/api-reference/queue/pubsub.mdx b/docs/content/docs/java/api-reference/queue/pubsub.mdx index 3e1a542c..a6d385f9 100644 --- a/docs/content/docs/java/api-reference/queue/pubsub.mdx +++ b/docs/content/docs/java/api-reference/queue/pubsub.mdx @@ -14,3 +14,22 @@ 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. | +| `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. | +| `readTopic(topic, name)` / `readTopic(topic, name, limit)` → `List` | Pull up to `limit` (default 100) messages after `name`'s cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then `ackTopic`. | +| `ackTopic(topic, name, cursor)` | Advance a log subscription's cursor to `cursor` (a message `id`) — a monotonic high-water mark. `false` if nothing moved. | +| `topicLogStats()` → `List` | Lag snapshot for every log subscription: `topic`, `subscription`, `cursor`, `lag`, `oldestUnackedAgeMs`. | + +See [Log topics](/java/guides/core/pubsub#log-topics) for the cursor, +at-least-once, and retention semantics. + +## `TopicMessage` + +One message pulled from a log topic, returned by `readTopic`. + +| Field | Type | Description | +|---|---|---| +| `id` | `String` | Cursor token — pass to `ackTopic`. | +| `payload` | `byte[]` | Opaque publish payload. Unlike Python/Node, this is **not** decoded for you — Java is statically typed and there's no target type to infer it into, so decode it yourself with the same serializer the publisher used. | +| `metadata` | `Map` | Caller metadata, decoded, or `null` if none was set. | +| `notes` | `Map` | Structured notes, decoded, or `null` if none were set. | +| `createdAt` | `long` | Unix-millisecond publish time. | diff --git a/docs/content/docs/node/api-reference/queue/pubsub.mdx b/docs/content/docs/node/api-reference/queue/pubsub.mdx index 4d15ab66..3b253a01 100644 --- a/docs/content/docs/node/api-reference/queue/pubsub.mdx +++ b/docs/content/docs/node/api-reference/queue/pubsub.mdx @@ -16,3 +16,22 @@ semantics, lifecycle, cross-SDK topics). | `listSubscriptions(topic?)` → `Promise` | All subscriptions, or one topic's active ones. | | `listTopics()` → `Promise` | Distinct topics with at least one subscription. | | `reapEphemeralSubscriptions()` → `Promise` | Drop ephemeral subscriptions whose owning worker is gone. Runs automatically on the worker heartbeat cadence; exposed for operational tooling. | +| `subscribeLog(topic, name)` → `Promise` | Register a durable log subscription — a named cursor with no handler. Writes immediately, so register it before the publishes it should see. | +| `readTopic(topic, name, limit?)` → `Promise` | Pull up to `limit` (default 100) messages after `name`'s cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then `ackTopic()`. | +| `ackTopic(topic, name, cursor)` → `Promise` | Advance a log subscription's cursor to `cursor` (a message `id`) — a monotonic high-water mark. `false` if nothing moved. | +| `topicLogStats()` → `Promise` | Lag snapshot for every log subscription: `topic`, `subscription`, `cursor`, `lag`, `oldestUnackedAgeMs`. | + +See [Log topics](/node/guides/core/pubsub#log-topics) for the cursor, +at-least-once, and retention semantics. + +## `TopicMessage` + +One message pulled from a log topic, returned by `readTopic()`. + +| Field | Type | Description | +|---|---|---| +| `id` | `string` | Cursor token — pass to `ackTopic()`. | +| `args` | `unknown[]` | Deserialized positional args from the `publish()` call. | +| `metadata` | `Record` | Caller metadata, if any. | +| `notes` | `Record` | Structured notes, if any. | +| `createdAt` | `number` | Unix-millisecond publish time. | diff --git a/docs/content/docs/python/api-reference/queue/pubsub.mdx b/docs/content/docs/python/api-reference/queue/pubsub.mdx index ac17a301..d39d3508 100644 --- a/docs/content/docs/python/api-reference/queue/pubsub.mdx +++ b/docs/content/docs/python/api-reference/queue/pubsub.mdx @@ -82,6 +82,73 @@ no-op. Unrecognized keyword arguments are forwarded as part of the task payload's `kwargs`, same as `enqueue()`. +## Log subscriptions + +See [Log topics](/python/guides/core/pubsub#log-topics) for the cursor, +at-least-once, and retention semantics. + +### `queue.subscribe_log()` + +```python +queue.subscribe_log(topic: str, name: str) -> None +``` + +Register a durable log subscription — a named cursor with no handler. +Writes to storage immediately, so register it before the publishes it +should see. Unlike `@queue.subscriber()`, there's no separate +`declare_subscriptions()` step. + +### `queue.read_topic()` + +```python +queue.read_topic(topic: str, name: str, limit: int = 100) -> list[TopicMessage] +``` + +Pull up to `limit` messages after `name`'s cursor, oldest first and +exclusive of it. Each message's payload is decoded with the queue +serializer. Returns an empty list once the subscription is caught up. +At-least-once — process the batch, then advance the cursor with +`ack_topic()`. + +### `queue.ack_topic()` + +```python +queue.ack_topic(topic: str, name: str, cursor: str) -> bool +``` + +Advance a log subscription's cursor to `cursor` (a message `id`). A +high-water mark — acking an id acks everything up to and including it. +Monotonic: acking an id older than what's already acked is a no-op and +returns `False`. + +### `queue.topic_log_stats()` + +```python +queue.topic_log_stats() -> list[dict[str, Any]] +``` + +Lag snapshot for every log subscription. Each dict contains `topic`, +`subscription`, `cursor` (`None` if nothing has been acked yet), `lag` +(un-acked message count), and `oldest_unacked_age_ms` (`None` when caught +up). + +### `TopicMessage` + +```python +from taskito import TopicMessage +``` + +One message pulled from a log topic, returned by `read_topic()`. + +| Field | Type | Description | +|---|---|---| +| `id` | `str` | Cursor token — pass to `ack_topic()`. | +| `args` | `tuple[Any, ...]` | Decoded positional args from the `publish()` call. | +| `kwargs` | `dict[str, Any]` | Decoded keyword args from the `publish()` call. | +| `metadata` | `dict[str, Any] \| None` | Caller metadata, if any. | +| `notes` | `dict[str, Any] \| None` | Structured notes, if any. | +| `created_at` | `int` | Unix-millisecond publish time. | + ## Managing subscriptions ### `queue.unsubscribe()` diff --git a/docs/content/docs/shared/guides/core/pubsub.mdx b/docs/content/docs/shared/guides/core/pubsub.mdx index e0dbf3d4..5ba22dda 100644 --- a/docs/content/docs/shared/guides/core/pubsub.mdx +++ b/docs/content/docs/shared/guides/core/pubsub.mdx @@ -390,15 +390,140 @@ deliveries. signature. +## Log topics + +Every subscription above is **fan-out**: one job per subscriber, each with +its own retries and DLQ. A **log** subscription is the other shape — a +named, durable cursor with no handler, reading a single stored copy of every +published message at its own pace. Publishing to a topic with log +subscribers writes **one** message no matter how many are reading it (O(1) +publish, versus fan-out's one job write per subscriber), and a topic can mix +both kinds: the same publish stores the log message once *and* fans out jobs +to any fan-out subscribers on it. + +Register a log subscription with queue.subscribe_log(), pull with queue.read_topic(), and advance its cursor with queue.ack_topic().} node={<>Register a log subscription with queue.subscribeLog(), pull with queue.readTopic(), and advance its cursor with queue.ackTopic().} java={<>Register a log subscription with taskito.subscribeLog(), pull with taskito.readTopic(), and advance its cursor with taskito.ackTopic().} /> + + T{{"Topic: orders"}} + T --> L[["One stored message"]] + L -. pull .-> C1["Cursor: audit-log"] + L -. pull .-> C2["Cursor: warehouse-sync"]`} +/> + +Reach for fan-out when each subscriber has its own independent side effect +that needs its own retries and DLQ. Reach for a log subscription for a +replayable stream one consumer works through at its own pace, or to publish +once to many — or unknown — consumers without multiplying writes by +subscriber count. + + + + +```python +queue.subscribe_log("orders", "audit-log") + +queue.publish("orders", 42) # -> [] no fan-out jobs; one message stored + +for msg in queue.read_topic("orders", "audit-log"): + audit_sink.write(msg.args, msg.kwargs) + queue.ack_topic("orders", "audit-log", msg.id) +``` + + + + +```ts +await queue.subscribeLog("orders", "audit-log"); + +await queue.publish("orders", [42]); // -> [] no fan-out jobs; one message stored + +for (const msg of await queue.readTopic("orders", "audit-log")) { + auditSink.write(msg.args); + await queue.ackTopic("orders", "audit-log", msg.id); +} +``` + + + + +```java +// Keep the Serializer you configure on the builder — readTopic() returns a +// raw byte[] payload, and Taskito has no getter to recover it later. +Serializer serializer = new JsonSerializer(); +try (Taskito taskito = Taskito.builder().sqlite("orders.db").serializer(serializer).open()) { + taskito.subscribeLog("orders", "audit-log"); + + taskito.publish("orders", 42); // -> [] no fan-out jobs; one message stored + + for (TopicMessage msg : taskito.readTopic("orders", "audit-log")) { + int orderId = serializer.deserialize(msg.payload, Integer.class); + auditSink.write(orderId); + taskito.ackTopic("orders", "audit-log", msg.id); + } +} +``` + + + + + + +`TopicMessage.payload` is raw `byte[]`, not a decoded value — Java is +statically typed, so there's no target type to infer it into the way +`readTopic()` infers one for Python/Node. Decode it yourself with the same +serializer the publisher used. `metadata` and `notes` are still decoded to +`Map`, same as the fan-out `Job` type. + + + +### The cursor is a high-water mark + +`read_topic()` / `readTopic()` returns messages oldest-first, **exclusive** +of the subscription's cursor — each message's `id` doubles as the cursor +token you pass back. `ack_topic()` / `ackTopic()` advances the cursor to a +given id: acking id X acks everything up to and including it, and it's +monotonic, so acking an id older than what's already acked is a no-op that +returns `false`. + +### At-least-once, same contract as fan-out + +A log subscription gets the same +at-least-once delivery +guarantee as everything else: read, process, then ack. A consumer that +crashes (or otherwise fails) between reading and acking re-reads the same +messages the next time it calls `read_topic()` / `readTopic()` — the cursor +only moves forward on an explicit ack. A log subscription also shares the +same late-join boundary as fan-out subscriptions above: +`subscribe_log()` / `subscribeLog()` only sees messages published after it +registered. + +### Retention compacts, it doesn't expire + +A log message is deleted once every log subscriber on its topic has acked +past it — the retention sweep compacts up to the *minimum* cursor across all +of a topic's log subscribers, on the same background cadence as job +retention. A topic with an unread (or never-yet-read) log subscriber keeps +its entire backlog; nothing is dropped just because it's old. + + + On the Redis backend, log topics are backed by a Redis Stream — an + implementation detail, not a different API. `read_topic()` / `readTopic()` + and `ack_topic()` / `ackTopic()` behave identically across SQLite, + Postgres, and Redis. + + ## Operational notes - **A down durable subscriber doesn't lose messages.** Its deliveries just accumulate as ordinary pending jobs in its queue until a worker resumes consuming it — the same backlog you'd see from any other stalled queue, and visible the same way (`stats()`, the dashboard). -- **Retention is normal job retention.** There's no separate pub/sub log or - replay buffer — a delivery is purged, replayed, or expired exactly like any - other job, via `result_ttl`, `purge_completed()`, and the dead-letter queue. +- **Fan-out retention is normal job retention.** A fan-out delivery is + purged, replayed, or expired exactly like any other job, via `result_ttl`, + `purge_completed()`, and the dead-letter queue. Log topics are the + exception — see "Retention compacts, it doesn't expire" under Log topics + above. - **Publish multiplies writes by subscriber count.** A topic with 20 active subscribers turns every `publish()` call into 20 job rows. For a high-fanout topic, set an aggressive `result_ttl` on the publish (or on the subscriber From 201f24b35b3f1dcf00b5a04617043e25b1502aa3 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 13:17:24 +0530 Subject: [PATCH 7/9] fix(core): restrict log read/ack to log subs, expire on Redis read_topic_messages/ack_topic_cursor now filter mode='log', so a fan-out subscription can't read a mixed topic's log or advance a cursor. Redis purge_topic_messages gains an expires_at XDEL pass, matching the Diesel TTL safety net so a stalled cursor can't block reclamation. --- crates/taskito-core/BINDING_CONTRACT.md | 11 ++- .../src/storage/diesel_common/pubsub.rs | 15 +++- .../src/storage/redis_backend/pubsub.rs | 70 +++++++++++++++---- .../taskito-core/tests/rust/storage_tests.rs | 10 +++ 4 files changed, 91 insertions(+), 15 deletions(-) diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 7c9eff35..f29368b3 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/crates/taskito-core/BINDING_CONTRACT.md @@ -146,6 +146,10 @@ prefixes make it injective). Done in the core; a shell that builds `NewJob` rows itself must match it. **Log cursor rules** (`read_topic_messages`/`ack_topic_cursor`): +- Both are **log-subscription only** — a `fanout` subscription (even on a mixed + topic) reads nothing and acks nothing (`false`). Enforced by a `mode = "log"` + filter on both backends, so a shell can't accidentally leak the log to a + fan-out subscriber. - The **cursor is an opaque, monotonic token** — a shell stores and passes back the message `id`, never parses it. Its format differs by backend (UUIDv7 on the Diesel backends, a `-` stream id on Redis), like the `get_task_logs_after` @@ -153,13 +157,16 @@ itself must match it. - **Reads are exclusive** of the cursor and ordered oldest-first; the cursor is resolved server-side from the subscription row. - **Ack is a high-water mark**: acking id `X` acks every message `≤ X`. Monotonic — - acking an older/equal id is a no-op (returns `false`). + acking an older/equal id is a no-op (returns `false`). Like a Kafka offset commit, + the caller is trusted to pass back an id it actually read. - Delivery is **at-least-once**: a consumer that reads but dies before acking re-reads those messages. There is no per-message ack. - **Retention** is min-cursor compaction: a message is dropped once every log subscriber on its topic has acked past it (Diesel deletes `id <= min(cursor)`; Redis `XTRIM MINID`). A topic with an unread subscriber keeps its backlog. - Diesel also honors an optional per-message `expires_at`; Redis does not. + Both backends also honor an optional per-message `expires_at` as a TTL safety + net (Diesel deletes expired rows; Redis `XDEL`s expired stream entries), so a + stalled or unread cursor can't block reclamation forever. **Test vector** (assert byte-exact in each shell that salts keys itself): key `evt-42`, topic `orders`, subscription `email` → diff --git a/crates/taskito-core/src/storage/diesel_common/pubsub.rs b/crates/taskito-core/src/storage/diesel_common/pubsub.rs index 973dbd67..34e9e504 100644 --- a/crates/taskito-core/src/storage/diesel_common/pubsub.rs +++ b/crates/taskito-core/src/storage/diesel_common/pubsub.rs @@ -203,8 +203,15 @@ macro_rules! impl_diesel_pubsub_ops { return Ok(Vec::new()); } let mut conn = self.conn()?; + // Require a log subscription: a fan-out sub must never read the + // log of a mixed topic. let cursor: Option> = topic_subscriptions::table - .find((topic, subscription_name)) + .filter(topic_subscriptions::topic.eq(topic)) + .filter(topic_subscriptions::subscription_name.eq(subscription_name)) + .filter( + topic_subscriptions::mode + .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG), + ) .select(topic_subscriptions::cursor) .first(&mut conn) .optional()?; @@ -235,10 +242,16 @@ macro_rules! impl_diesel_pubsub_ops { cursor: &str, ) -> Result { let mut conn = self.conn()?; + // Only a log subscription has a cursor; the mode filter also stops + // an ack on a fan-out subscription from writing one. let affected = diesel::update( topic_subscriptions::table .filter(topic_subscriptions::topic.eq(topic)) .filter(topic_subscriptions::subscription_name.eq(subscription_name)) + .filter( + topic_subscriptions::mode + .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG), + ) .filter( topic_subscriptions::cursor .is_null() diff --git a/crates/taskito-core/src/storage/redis_backend/pubsub.rs b/crates/taskito-core/src/storage/redis_backend/pubsub.rs index 7efec925..c107bffc 100644 --- a/crates/taskito-core/src/storage/redis_backend/pubsub.rs +++ b/crates/taskito-core/src/storage/redis_backend/pubsub.rs @@ -595,6 +595,10 @@ impl RedisStorage { return Ok(Vec::new()); }; let entry: SubEntry = serde_json::from_str(&data)?; + // A fan-out sub must never read a mixed topic's log. + if entry.mode != SUBSCRIPTION_MODE_LOG { + return Ok(Vec::new()); + } let start = match &entry.cursor { Some(cursor) => format!("({cursor}"), @@ -631,6 +635,10 @@ impl RedisStorage { return Ok(false); }; let mut entry: SubEntry = serde_json::from_str(&data)?; + // Only a log subscription has a cursor to advance. + if entry.mode != SUBSCRIPTION_MODE_LOG { + return Ok(false); + } let advance = match &entry.cursor { None => true, @@ -680,10 +688,11 @@ impl RedisStorage { Ok(out) } - /// Compact each log topic to the min cursor across its subscriptions via - /// `XTRIM MINID`. Unlike the Diesel backend this ignores `now`/`limit`: XTRIM - /// is one server-side op with no per-message TTL (min-cursor compaction only). - pub fn purge_topic_messages(&self, _now: i64, _limit: i64) -> Result { + /// Compact each log topic: `XTRIM MINID` to the min cursor across its subs, + /// then `XDEL` any entry past `expires_at` (the TTL safety net, so a NULL or + /// stalled cursor can't block reclamation) — matching the Diesel backend. + /// The expiry scan is bounded by `limit` per topic. + pub fn purge_topic_messages(&self, now: i64, limit: i64) -> Result { use std::collections::HashMap; let subs = self.list_subscriptions()?; let mut floors: HashMap> = HashMap::new(); @@ -711,15 +720,52 @@ impl RedisStorage { let mut conn = self.conn()?; let mut removed = 0u64; for (topic, floor) in floors { - let Some(floor) = floor else { continue }; - let n: u64 = redis::cmd("XTRIM") - .arg(self.topic_stream_key(&topic)) - .arg("MINID") - .arg(Self::next_stream_id(&floor)) - .query(&mut conn) - .map_err(map_err)?; - removed += n; + let stream_key = self.topic_stream_key(&topic); + if let Some(floor) = floor { + let trimmed: u64 = redis::cmd("XTRIM") + .arg(&stream_key) + .arg("MINID") + .arg(Self::next_stream_id(&floor)) + .query(&mut conn) + .map_err(map_err)?; + removed += trimmed; + } + removed += self.xdel_expired(&mut conn, &stream_key, now, limit)?; } Ok(removed) } + + /// `XDEL` up to `limit` entries in `stream_key` whose `expires_at` field is + /// at or before `now`, regardless of any subscription cursor. + fn xdel_expired( + &self, + conn: &mut redis::Connection, + stream_key: &str, + now: i64, + limit: i64, + ) -> Result { + let reply: StreamRangeReply = redis::cmd("XRANGE") + .arg(stream_key) + .arg("-") + .arg("+") + .arg("COUNT") + .arg(limit) + .query(conn) + .map_err(map_err)?; + let expired: Vec = reply + .ids + .into_iter() + .filter(|entry| entry.get::("expires_at").is_some_and(|at| at <= now)) + .map(|entry| entry.id) + .collect(); + if expired.is_empty() { + return Ok(0); + } + let deleted: u64 = redis::cmd("XDEL") + .arg(stream_key) + .arg(&expired) + .query(conn) + .map_err(map_err)?; + Ok(deleted) + } } diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 5b664423..fbafac6d 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -1348,6 +1348,16 @@ fn test_topic_log_messages(s: &impl Storage) { .unwrap() .is_empty()); + // A fan-out subscription on the same topic can neither read the log nor + // advance a cursor — the log is for log subscriptions only. + let mut fan = log_sub(topic, "fan"); + fan.mode = "fanout".to_string(); + fan.task_name = "deliver".to_string(); + s.register_subscription(&fan).unwrap(); + assert!(s.read_topic_messages(topic, "fan", 10).unwrap().is_empty()); + assert!(!s.ack_topic_cursor(topic, "fan", &m2.id).unwrap()); + s.unsubscribe(topic, "fan").unwrap(); + // Drop the subscription so the global purge/stats in later tests are not // affected by this topic's leftover cursor. s.unsubscribe(topic, "reader").unwrap(); From 6038710a5415254889407fab43932e0e0680d08f Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 13:17:49 +0530 Subject: [PATCH 8/9] fix(java): reject log mode on backends that lack it The mode-aware registerSubscription default silently downgraded 'log' to fan-out; now it throws UnsupportedOperationException for any non-fanout mode, keeping the compatibility fallback for fan-out only. --- .../main/java/org/byteveda/taskito/spi/QueueBackend.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index 24e7c2e3..659eb713 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -240,9 +240,9 @@ default void registerSubscription( * {@code "fanout"} (one job per publish) or {@code "log"} (append-once + a * per-subscription cursor pulled via {@link #readTopicMessagesJson}). * - *

The default drops {@code mode} and delegates to the fan-out form, so a - * backend that predates log topics still registers a fan-out subscription - * instead of throwing. + *

The default preserves fan-out compatibility for backends that predate + * log topics — {@code "fanout"} delegates to the mode-less form — but rejects + * {@code "log"} rather than silently downgrading it to fan-out. */ default void registerSubscription( String topic, @@ -255,6 +255,9 @@ default void registerSubscription( Integer maxRetries, Long timeoutMs, String mode) { + if (!"fanout".equals(mode)) { + throw new UnsupportedOperationException("log topics not supported by this backend"); + } registerSubscription( topic, subscriptionName, From 6cd76fd34429d172d388ec4639e077824057bcec Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 19 Jul 2026 13:17:49 +0530 Subject: [PATCH 9/9] docs(pubsub): scope fan-out operational notes Backlog accumulation and per-subscriber write amplification are fan-out properties; a log topic writes one row per publish regardless of subscriber count. --- docs/content/docs/shared/guides/core/pubsub.mdx | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/docs/content/docs/shared/guides/core/pubsub.mdx b/docs/content/docs/shared/guides/core/pubsub.mdx index 5ba22dda..3751f596 100644 --- a/docs/content/docs/shared/guides/core/pubsub.mdx +++ b/docs/content/docs/shared/guides/core/pubsub.mdx @@ -515,20 +515,22 @@ its entire backlog; nothing is dropped just because it's old. ## Operational notes -- **A down durable subscriber doesn't lose messages.** Its deliveries just +- **A down fan-out subscriber doesn't lose messages.** Its deliveries just accumulate as ordinary pending jobs in its queue until a worker resumes consuming it — the same backlog you'd see from any other stalled queue, and - visible the same way (`stats()`, the dashboard). + visible the same way (`stats()`, the dashboard). (A down log subscriber is + the same story from the other side: its cursor just stops advancing.) - **Fan-out retention is normal job retention.** A fan-out delivery is purged, replayed, or expired exactly like any other job, via `result_ttl`, `purge_completed()`, and the dead-letter queue. Log topics are the exception — see "Retention compacts, it doesn't expire" under Log topics above. -- **Publish multiplies writes by subscriber count.** A topic with 20 active - subscribers turns every `publish()` call into 20 job rows. For a high-fanout - topic, set an aggressive `result_ttl` on the publish (or on the subscriber - tasks themselves) so completed deliveries don't pile up in storage between - cleanup passes. +- **Fan-out publish multiplies writes by subscriber count.** A fan-out topic + with 20 active subscribers turns every `publish()` call into 20 job rows. For + a high-fanout topic, set an aggressive `result_ttl` on the publish (or on the + subscriber tasks themselves) so completed deliveries don't pile up in storage + between cleanup passes. (A log topic writes one row per publish regardless of + subscriber count — this multiplication is a fan-out property only.) ## See also