feat: per-message ack for log topics

crates/taskito-core/BINDING_CONTRACT.md

@@ -160,7 +160,7 @@ itself must match it.
   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.
+  re-reads those messages. The cursor read has no per-message ack (see below).
 - **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.
@@ -184,6 +184,26 @@ itself must match it.
 - Storage: the Diesel backends use a `topics` table (migration `m0006`); Redis uses
   a `topics` hash keyed by name. A shell surfaces `Topic { name, mode, retention_ms,
   created_at }`.
+**Per-message ack** (`lease_topic_messages`/`ack_message`/`nack_message`):
+- An opt-in **consumption choice** on a `log` subscription (not a registration
+  flag): instead of the cursor read, a consumer *leases* messages and acks/nacks
+  each individually, so a poison message never blocks its siblings.
+- `lease_topic_messages(topic, sub, limit, visibility_ms, now)` returns up to
+  `limit` **available** messages oldest-first — never delivered, or a prior lease
+  that expired (`now`-relative) or was nacked and never acked — and (re)leases
+  each for `visibility_ms`. In-flight (leased, un-expired) messages are skipped.
+- `ack_message` ends a delivery (never redelivered); `nack_message` makes it
+  available immediately (vs waiting out the timeout). Both return `false` when
+  there is no un-acked delivery. An un-acked lease that times out is redelivered.
+- Delivery state lives per `(subscription, message)`: Diesel `topic_deliveries`
+  table (migration `m0007`); Redis a `pmdeliv:<topic>:<sub>` hash mirroring it.
+- **Retention**: on a topic consumed purely per-message (every log sub has
+  acked), a message is compacted once every per-message subscriber has acked it;
+  a topic that mixes in a cursor subscriber falls back to `expires_at`. Its
+  delivery state is dropped with it (Diesel deletes the rows, Redis `HDEL`s the
+  fields). Both backends implement this — Diesel via `topic_deliveries`, Redis by
+  scanning the `pmdeliv:*` hashes during the purge sweep. A shell only marshals
+  the three calls; the core owns the state.
 
 **Test vector** (assert byte-exact in each shell that salts keys itself):
 key `evt-42`, topic `orders`, subscription `email` →
@@ -214,7 +234,8 @@ Grouped by concern (enumerated, not exhaustive — read the trait):
 - **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`,
-  `declare_topic`, `get_topic`, `list_declared_topics` (see
+  `declare_topic`, `get_topic`, `list_declared_topics`, `lease_topic_messages`,
+  `ack_message`, `nack_message` (see
   [Topic pub/sub](#topic-pubsub-cross-sdk)).
 
 ## Lifecycle the shell drives

crates/taskito-core/migrations/m0007_topic_deliveries.rs

@@ -0,0 +1,65 @@
+//! Per-message ack for log topics (`0007_topic_deliveries`).
+//!
+//! S28 log subscriptions consume with a single high-water-mark cursor, so one
+//! poisoned message blocks the whole subscription. This adds an opt-in
+//! per-message mode: instead of the cursor read, a consumer *leases* each
+//! message with a visibility timeout and acks or nacks it individually. It's a
+//! consumption choice (which read methods you call), not a registration flag —
+//! delivery state per `(subscription, message)` lives in `topic_deliveries`, and
+//! an un-acked lease that expires is redelivered. Idempotent:
+//! `.if_not_exists()` on the table and index.
+
+use sea_query::{Alias, ColumnDef, Index, Table};
+
+use crate::storage::migrate::{ddl, Backend, Migration, Stmt};
+
+pub struct M0007TopicDeliveries;
+
+fn col(name: &str) -> ColumnDef {
+    ColumnDef::new(Alias::new(name))
+}
+
+fn t(name: &str) -> Alias {
+    Alias::new(name)
+}
+
+impl Migration for M0007TopicDeliveries {
+    fn version(&self) -> &'static str {
+        "0007_topic_deliveries"
+    }
+
+    fn up(&self, b: Backend) -> Vec<Stmt> {
+        // Per-(subscription, message) delivery state. `lease_expires_at` bounds
+        // an in-flight lease (0 = available for redelivery, e.g. after a nack);
+        // `acked` ends the delivery. `attempts` counts (re)deliveries.
+        let deliveries = Table::create()
+            .table(t("topic_deliveries"))
+            .if_not_exists()
+            .col(col("topic").text().not_null())
+            .col(col("subscription_name").text().not_null())
+            .col(col("message_id").text().not_null())
+            .col(col("acked").boolean().not_null().default(false))
+            .col(col("attempts").integer().not_null().default(0))
+            .col(col("lease_expires_at").big_integer().not_null().default(0))
+            .col(col("delivered_at").big_integer().not_null().default(0))
+            .primary_key(
+                Index::create()
+                    .col(t("topic"))
+                    .col(t("subscription_name"))
+                    .col(t("message_id")),
+            )
+            .to_owned();
+
+        // The lease read's anti-join filters by (topic, subscription, availability).
+        let by_sub_lease = Index::create()
+            .if_not_exists()
+            .name("idx_topic_deliveries_sub_lease")
+            .table(t("topic_deliveries"))
+            .col(t("topic"))
+            .col(t("subscription_name"))
+            .col(t("lease_expires_at"))
+            .to_owned();
+
+        vec![ddl(b, &deliveries), ddl(b, &by_sub_lease)]
+    }
+}

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

@@ -352,7 +352,7 @@ macro_rules! impl_diesel_pubsub_ops {
                     .load(&mut conn)?;
                 ids.extend(expired);
 
-                // Then fully-acked messages per topic, within the remaining budget.
+                // Then cursor-fully-acked messages per topic, within the budget.
                 for (topic, floor) in floors {
                     let Some(floor) = floor else { continue };
                     let remaining = limit - ids.len() as i64;
@@ -368,6 +368,78 @@ macro_rules! impl_diesel_pubsub_ops {
                     ids.extend(acked);
                 }
 
+                // Per-message compaction: on a topic consumed purely per-message
+                // (every log sub has delivery rows — mixing with a cursor sub
+                // falls back to expiry-only cleanup), drop the oldest messages
+                // every per-message subscriber has acked. Inlined so it reuses
+                // `conn` — a second pooled connection would deadlock a size-1 pool.
+                let pm_budget = limit - ids.len() as i64;
+                if pm_budget > 0 {
+                    use diesel::dsl::count_star;
+                    use std::collections::HashSet;
+                    // Per-message subscribers per topic (those with delivery rows).
+                    let pm_pairs: Vec<(String, String)> = topic_deliveries::table
+                        .select((topic_deliveries::topic, topic_deliveries::subscription_name))
+                        .distinct()
+                        .load(&mut conn)?;
+                    let mut pm_subs: HashMap<String, HashSet<String>> = HashMap::new();
+                    for (topic, name) in pm_pairs {
+                        pm_subs.entry(topic).or_default().insert(name);
+                    }
+                    // Log subscribers per topic — skip a topic that mixes in a
+                    // cursor reader (a log sub with no delivery rows).
+                    let mut log_by_topic: HashMap<String, Vec<String>> = HashMap::new();
+                    if !pm_subs.is_empty() {
+                        let log_subs: Vec<(String, String)> = topic_subscriptions::table
+                            .filter(
+                                topic_subscriptions::mode
+                                    .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            )
+                            .select((
+                                topic_subscriptions::topic,
+                                topic_subscriptions::subscription_name,
+                            ))
+                            .load(&mut conn)?;
+                        for (topic, name) in log_subs {
+                            log_by_topic.entry(topic).or_default().push(name);
+                        }
+                    }
+                    for (topic, names) in &pm_subs {
+                        let remaining = limit - ids.len() as i64;
+                        if remaining <= 0 {
+                            break;
+                        }
+                        if let Some(logs) = log_by_topic.get(topic) {
+                            if logs.iter().any(|n| !names.contains(n)) {
+                                continue;
+                            }
+                        }
+                        let sub_count = names.len() as i64;
+                        let candidates: Vec<String> = topic_messages::table
+                            .filter(topic_messages::topic.eq(topic))
+                            .order(topic_messages::id.asc())
+                            .select(topic_messages::id)
+                            .limit(remaining)
+                            .load(&mut conn)?;
+                        if candidates.is_empty() {
+                            continue;
+                        }
+                        let acked_counts: Vec<(String, i64)> = topic_deliveries::table
+                            .filter(topic_deliveries::topic.eq(topic))
+                            .filter(topic_deliveries::message_id.eq_any(&candidates))
+                            .filter(topic_deliveries::acked.eq(true))
+                            .group_by(topic_deliveries::message_id)
+                            .select((topic_deliveries::message_id, count_star()))
+                            .load(&mut conn)?;
+                        let counts: HashMap<String, i64> = acked_counts.into_iter().collect();
+                        for id in candidates {
+                            if counts.get(&id).copied().unwrap_or(0) == sub_count {
+                                ids.push(id);
+                            }
+                        }
+                    }
+                }
+
                 if ids.is_empty() {
                     return Ok(0);
                 }
@@ -376,6 +448,12 @@ macro_rules! impl_diesel_pubsub_ops {
                 let removed =
                     diesel::delete(topic_messages::table.filter(topic_messages::id.eq_any(&ids)))
                         .execute(&mut conn)?;
+                // Drop the delivery rows of the purged messages so per-message
+                // state can't outlive the log it tracks.
+                diesel::delete(
+                    topic_deliveries::table.filter(topic_deliveries::message_id.eq_any(&ids)),
+                )
+                .execute(&mut conn)?;
                 Ok(removed as u64)
             }
 
@@ -398,6 +476,134 @@ macro_rules! impl_diesel_pubsub_ops {
                     .load::<TopicRow>(&mut conn)?;
                 Ok(rows.into_iter().map(Into::into).collect())
             }
+
+            /// Lease up to `limit` available messages to `subscription_name` for
+            /// `visibility_ms` (see the `lease_topic_messages` trait method).
+            /// Requires a registered `log` subscription (like the cursor read).
+            /// One bounded anti-join finds the available messages, then a lease
+            /// row is upserted per message in one transaction. Delivery is
+            /// **at-least-once**: the availability window isn't row-locked, so two
+            /// concurrent leasers can briefly claim the same message — handlers
+            /// must be idempotent, the same guarantee as a lease timeout redeliver.
+            pub fn lease_topic_messages(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                limit: i64,
+                visibility_ms: i64,
+                now: i64,
+            ) -> Result<Vec<$crate::storage::records::TopicMessage>> {
+                if limit <= 0 {
+                    return Ok(Vec::new());
+                }
+                let mut conn = self.conn()?;
+                // Only a registered log subscription may lease — a fan-out or
+                // unknown name must never claim from the log (mirrors read/ack).
+                let is_log_sub = 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),
+                    )
+                    .select(topic_subscriptions::subscription_name)
+                    .first::<String>(&mut conn)
+                    .optional()?;
+                if is_log_sub.is_none() {
+                    return Ok(Vec::new());
+                }
+                conn.transaction(|conn| {
+                    // Available = no delivery row (never delivered), or a prior
+                    // lease that expired and was never acked.
+                    let rows: Vec<TopicMessageRow> = topic_messages::table
+                        .left_join(
+                            topic_deliveries::table.on(topic_deliveries::topic
+                                .eq(topic_messages::topic)
+                                .and(topic_deliveries::subscription_name.eq(subscription_name))
+                                .and(topic_deliveries::message_id.eq(topic_messages::id))),
+                        )
+                        .filter(topic_messages::topic.eq(topic))
+                        .filter(
+                            topic_deliveries::message_id
+                                .is_null()
+                                .or(topic_deliveries::acked
+                                    .eq(false)
+                                    .and(topic_deliveries::lease_expires_at.le(now))),
+                        )
+                        .order(topic_messages::id.asc())
+                        .limit(limit)
+                        .select(TopicMessageRow::as_select())
+                        .load(conn)?;
+
+                    let lease_until = now + visibility_ms;
+                    for row in &rows {
+                        let lease = NewTopicDeliveryRow {
+                            topic,
+                            subscription_name,
+                            message_id: &row.id,
+                            acked: false,
+                            attempts: 1,
+                            lease_expires_at: lease_until,
+                            delivered_at: now,
+                        };
+                        diesel::insert_into(topic_deliveries::table)
+                            .values(&lease)
+                            .on_conflict((
+                                topic_deliveries::topic,
+                                topic_deliveries::subscription_name,
+                                topic_deliveries::message_id,
+                            ))
+                            .do_update()
+                            .set((
+                                topic_deliveries::acked.eq(false),
+                                topic_deliveries::attempts.eq(topic_deliveries::attempts + 1),
+                                topic_deliveries::lease_expires_at.eq(lease_until),
+                                topic_deliveries::delivered_at.eq(now),
+                            ))
+                            .execute(conn)?;
+                    }
+                    Ok(rows.into_iter().map(Into::into).collect())
+                })
+            }
+
+            /// Ack one leased message — the delivery is done. Returns false when
+            /// there was no un-acked delivery row to ack.
+            pub fn ack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> Result<bool> {
+                let mut conn = self.conn()?;
+                let affected = diesel::update(
+                    topic_deliveries::table
+                        .find((topic, subscription_name, message_id))
+                        .filter(topic_deliveries::acked.eq(false)),
+                )
+                .set(topic_deliveries::acked.eq(true))
+                .execute(&mut conn)?;
+                Ok(affected > 0)
+            }
+
+            /// Nack one leased message — make it immediately available for
+            /// redelivery (`lease_expires_at = 0`). Returns false when there was
+            /// no un-acked delivery row to nack.
+            pub fn nack_message(
+                &self,
+                topic: &str,
+                subscription_name: &str,
+                message_id: &str,
+            ) -> Result<bool> {
+                let mut conn = self.conn()?;
+                let affected = diesel::update(
+                    topic_deliveries::table
+                        .find((topic, subscription_name, message_id))
+                        .filter(topic_deliveries::acked.eq(false)),
+                )
+                .set(topic_deliveries::lease_expires_at.eq(0))
+                .execute(&mut conn)?;
+                Ok(affected > 0)
+            }
         }
     };
 }