feat: add opt-in log+cursor pub/sub

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"

crates/taskito-core/BINDING_CONTRACT.md

@@ -126,6 +126,52 @@ 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 = `<key>::<topic_len>:<name_len>:<topic><name>` (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`):
+- 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 `<ms>-<seq>` 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`). 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.
+  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` →
+`evt-42::6:5:ordersemail`.
+
 ## Types the shell produces / consumes
 - **`Job`** — `job.rs`. Fields incl. `id`, `queue`, `task_name`, `payload: Vec<u8>` (opaque),
   `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`,
@@ -149,6 +195,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.

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<Stmt> {
+        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()),
+        ]
+    }
+}

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<S: Storage>(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<NewJob> = 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);
     }
@@ -218,6 +241,7 @@ mod tests {
                 priority,
                 max_retries,
                 timeout_ms,
+                mode: crate::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(),
             })
             .unwrap();
     }
@@ -229,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();

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"
             );
         }