feat: replace stringly-typed closed-set params with enums

CHANGELOG.md

@@ -5,6 +5,32 @@ All notable changes to taskito are documented here. The format is based on
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html). All SDKs (Python, Node, Java) and the
 underlying Rust crates are released together, in lock-step.
 
+## Unreleased
+
+### Changed
+
+- **BREAKING (Python): resource scopes renamed to match the other SDKs.**
+  `ResourceScope.TASK` meant *checked out of a bounded pool* and `REQUEST` meant *fresh per
+  task* — the reverse of what Java and Node call those names, so the same `scope="task"` code
+  behaved differently per SDK. Python now uses `POOLED` (was `TASK`) and `TASK` (was `REQUEST`);
+  `REQUEST` is gone, and `scope="request"` raises. `scope="task"` still resolves but now builds
+  per task instead of pooling, so pooled resources must move to `scope="pooled"` — passing
+  `pool_size`/`pool_min` with any other scope now raises, which catches that case.
+- Closed-set parameters across the core and all three SDKs take enums instead of bare strings:
+  built-in proxy ids, interception mode, predicate `on_false`, log-consumer `on_error`, gate
+  `on_timeout`, workflow diagram format, fan-out strategy, dispatch order, task-log level,
+  workflow run state, dashboard role, subscription mode, and worker status. Wire and stored
+  values are unchanged, and existing string callers keep working (Java keeps its `String`
+  overloads, deprecated).
+
+### Added
+
+- **Node `request` resource scope.** A fresh instance on every `useResource()` call, each
+  disposed when the task ends — matching the Java scope of the same name.
+- Job outcome events report how long the task ran: `durationMs()` on Java's `OutcomeEvent`,
+  `durationMs` on Node's, and `duration_ms` on Python's job event payloads. Java also gains
+  `NodeSnapshot.durationMs()` / `compensationDurationMs()` and `TaskContext.elapsedMs()`.
+
 ## 0.21.0
 
 Overload-controls and retention release. The queue gains admission and load-shedding controls,

crates/taskito-core/src/pubsub.rs

@@ -7,14 +7,17 @@
 
 use crate::error::{QueueError, Result};
 use crate::job::{now_millis, Job, NewJob};
-use crate::storage::records::{Subscription, SUBSCRIPTION_MODE_LOG};
+use crate::storage::records::{Subscription, SubscriptionMode};
 use crate::storage::Storage;
 
 /// Validate a topic declaration before a backend persists it: only `"log"`
 /// topics are declarable, and a retention window must be non-negative (a
 /// negative one would expire messages immediately or overflow `now + retention`).
-pub(crate) fn validate_topic_declaration(mode: &str, retention_ms: Option<i64>) -> Result<()> {
-    if mode != SUBSCRIPTION_MODE_LOG {
+pub(crate) fn validate_topic_declaration(
+    mode: SubscriptionMode,
+    retention_ms: Option<i64>,
+) -> Result<()> {
+    if !mode.is_log() {
         return Err(QueueError::Config(format!(
             "only \"log\" topics are declarable, got {mode:?}"
         )));
@@ -85,9 +88,7 @@ pub struct PublishRequest {
 /// on the unique index; unkeyed publishes use one batch insert.
 pub fn publish_to_topic<S: Storage>(storage: &S, request: &PublishRequest) -> Result<Vec<Job>> {
     let subscriptions = storage.list_subscriptions_for_topic(&request.topic)?;
-    let has_log_sub = subscriptions
-        .iter()
-        .any(|s| s.mode == SUBSCRIPTION_MODE_LOG);
+    let has_log_sub = subscriptions.iter().any(|s| s.mode.is_log());
 
     // 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.
@@ -121,7 +122,7 @@ pub fn publish_to_topic<S: Storage>(storage: &S, request: &PublishRequest) -> Re
 
     let jobs: Vec<NewJob> = subscriptions
         .iter()
-        .filter(|sub| sub.mode != SUBSCRIPTION_MODE_LOG)
+        .filter(|sub| !sub.mode.is_log())
         .map(|sub| delivery_job(request, sub))
         .collect();
     if jobs.is_empty() {
@@ -271,7 +272,7 @@ mod tests {
                 priority,
                 max_retries,
                 timeout_ms,
-                mode: crate::storage::records::SUBSCRIPTION_MODE_FANOUT.to_string(),
+                mode: SubscriptionMode::Fanout,
             })
             .unwrap();
     }
@@ -297,7 +298,7 @@ mod tests {
                 priority: None,
                 max_retries: None,
                 timeout_ms: None,
-                mode: SUBSCRIPTION_MODE_LOG.to_string(),
+                mode: SubscriptionMode::Log,
             })
             .unwrap();
     }
@@ -446,7 +447,7 @@ mod tests {
     fn declared_log_topic_retains_with_zero_subscribers() {
         let storage = SqliteStorage::in_memory().unwrap();
         storage
-            .declare_topic("events", SUBSCRIPTION_MODE_LOG, None)
+            .declare_topic("events", SubscriptionMode::Log, None)
             .unwrap();
 
         // No subscriber at publish time, but the topic is declared → retained.
@@ -478,7 +479,7 @@ mod tests {
     fn declared_topic_retention_sets_message_expiry() {
         let storage = SqliteStorage::in_memory().unwrap();
         storage
-            .declare_topic("events", SUBSCRIPTION_MODE_LOG, Some(60_000))
+            .declare_topic("events", SubscriptionMode::Log, Some(60_000))
             .unwrap();
         publish_to_topic(&storage, &request("events", None)).unwrap();
 
@@ -494,7 +495,7 @@ mod tests {
         assert!(storage.get_topic("events").unwrap().is_none());
 
         storage
-            .declare_topic("events", SUBSCRIPTION_MODE_LOG, Some(1000))
+            .declare_topic("events", SubscriptionMode::Log, Some(1000))
             .unwrap();
         let first = storage.get_topic("events").unwrap().unwrap();
         assert_eq!(first.name, "events");
@@ -506,7 +507,7 @@ mod tests {
         // regression that overwrote created_at could still pass this assertion.
         std::thread::sleep(std::time::Duration::from_millis(2));
         storage
-            .declare_topic("events", SUBSCRIPTION_MODE_LOG, Some(2000))
+            .declare_topic("events", SubscriptionMode::Log, Some(2000))
             .unwrap();
         let second = storage.get_topic("events").unwrap().unwrap();
         assert_eq!(second.retention_ms, Some(2000));
@@ -517,9 +518,11 @@ mod tests {
     #[test]
     fn declare_topic_rejects_bad_mode_and_negative_retention() {
         let storage = SqliteStorage::in_memory().unwrap();
-        assert!(storage.declare_topic("events", "fanout", None).is_err());
         assert!(storage
-            .declare_topic("events", SUBSCRIPTION_MODE_LOG, Some(-1))
+            .declare_topic("events", SubscriptionMode::Fanout, None)
+            .is_err());
+        assert!(storage
+            .declare_topic("events", SubscriptionMode::Log, Some(-1))
             .is_err());
         // Nothing was persisted on rejection.
         assert!(storage.get_topic("events").unwrap().is_none());

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

@@ -210,7 +210,7 @@ macro_rules! impl_diesel_pubsub_ops {
                     .filter(topic_subscriptions::subscription_name.eq(subscription_name))
                     .filter(
                         topic_subscriptions::mode
-                            .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                     )
                     .select(topic_subscriptions::cursor)
                     .first(&mut conn)
@@ -250,7 +250,7 @@ macro_rules! impl_diesel_pubsub_ops {
                         .filter(topic_subscriptions::subscription_name.eq(subscription_name))
                         .filter(
                             topic_subscriptions::mode
-                                .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                                .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                         )
                         .filter(
                             topic_subscriptions::cursor
@@ -271,7 +271,7 @@ macro_rules! impl_diesel_pubsub_ops {
                 let subs: Vec<(String, String, Option<String>)> = topic_subscriptions::table
                     .filter(
                         topic_subscriptions::mode
-                            .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                     )
                     .select((
                         topic_subscriptions::topic,
@@ -322,7 +322,7 @@ macro_rules! impl_diesel_pubsub_ops {
                 let subs: Vec<(String, Option<String>)> = topic_subscriptions::table
                     .filter(
                         topic_subscriptions::mode
-                            .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                     )
                     .select((topic_subscriptions::topic, topic_subscriptions::cursor))
                     .load(&mut conn)?;
@@ -393,7 +393,7 @@ macro_rules! impl_diesel_pubsub_ops {
                         let log_subs: Vec<(String, String)> = topic_subscriptions::table
                             .filter(
                                 topic_subscriptions::mode
-                                    .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                                    .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                             )
                             .select((
                                 topic_subscriptions::topic,
@@ -504,7 +504,7 @@ macro_rules! impl_diesel_pubsub_ops {
                     .filter(topic_subscriptions::subscription_name.eq(subscription_name))
                     .filter(
                         topic_subscriptions::mode
-                            .eq($crate::storage::records::SUBSCRIPTION_MODE_LOG),
+                            .eq($crate::storage::records::SubscriptionMode::Log.as_str()),
                     )
                     .select(topic_subscriptions::subscription_name)
                     .first::<String>(&mut conn)

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

@@ -22,12 +22,16 @@ macro_rules! impl_diesel_worker_ops {
             }
 
             /// Update the status of a worker.
-            pub fn update_worker_status(&self, worker_id: &str, status: &str) -> Result<()> {
+            pub fn update_worker_status(
+                &self,
+                worker_id: &str,
+                status: $crate::storage::records::WorkerStatus,
+            ) -> Result<()> {
                 let mut conn = self.conn()?;
 
                 diesel::update(workers::table)
                     .filter(workers::worker_id.eq(worker_id))
-                    .set(workers::status.eq(status))
+                    .set(workers::status.eq(status.as_str()))
                     .execute(&mut conn)?;
 
                 Ok(())

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

@@ -731,7 +731,7 @@ macro_rules! impl_storage {
             fn declare_topic(
                 &self,
                 name: &str,
-                mode: &str,
+                mode: $crate::storage::records::SubscriptionMode,
                 retention_ms: Option<i64>,
             ) -> $crate::error::Result<()> {
                 self.declare_topic(name, mode, retention_ms)
@@ -893,7 +893,7 @@ macro_rules! impl_storage {
             fn update_worker_status(
                 &self,
                 worker_id: &str,
-                status: &str,
+                status: $crate::storage::records::WorkerStatus,
             ) -> $crate::error::Result<()> {
                 self.update_worker_status(worker_id, status)
             }
@@ -1476,7 +1476,12 @@ impl Storage for StorageBackend {
     fn purge_topic_messages(&self, now: i64, limit: i64) -> Result<u64> {
         delegate!(self, purge_topic_messages, now, limit)
     }
-    fn declare_topic(&self, name: &str, mode: &str, retention_ms: Option<i64>) -> Result<()> {
+    fn declare_topic(
+        &self,
+        name: &str,
+        mode: records::SubscriptionMode,
+        retention_ms: Option<i64>,
+    ) -> Result<()> {
         delegate!(self, declare_topic, name, mode, retention_ms)
     }
     fn get_topic(&self, name: &str) -> Result<Option<records::Topic>> {
@@ -1634,7 +1639,7 @@ impl Storage for StorageBackend {
     fn heartbeat(&self, worker_id: &str, resource_health: Option<&str>) -> Result<()> {
         delegate!(self, heartbeat, worker_id, resource_health)
     }
-    fn update_worker_status(&self, worker_id: &str, status: &str) -> Result<()> {
+    fn update_worker_status(&self, worker_id: &str, status: records::WorkerStatus) -> Result<()> {
         delegate!(self, update_worker_status, worker_id, status)
     }
     fn list_workers(&self) -> Result<Vec<records::WorkerInfo>> {

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

@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
 
 use super::records::{
     CircuitBreakerState, JobError, LockInfo, PeriodicTask, RateLimitState, ReplayEntry,
-    Subscription, TaskLogEntry, TaskMetric, Topic, TopicMessage, WorkerInfo,
+    Subscription, SubscriptionMode, TaskLogEntry, TaskMetric, Topic, TopicMessage, WorkerInfo,
 };
 use super::schema::{
     archived_jobs, circuit_breakers, dashboard_settings, dead_letter, distributed_locks,
@@ -740,7 +740,7 @@ impl From<SubscriptionRow> for Subscription {
             priority: r.priority,
             max_retries: r.max_retries,
             timeout_ms: r.timeout_ms,
-            mode: r.mode,
+            mode: SubscriptionMode::from_wire(&r.mode),
             cursor: r.cursor,
         }
     }
@@ -764,7 +764,7 @@ impl From<TopicRow> for Topic {
     fn from(r: TopicRow) -> Self {
         Topic {
             name: r.name,
-            mode: r.mode,
+            mode: SubscriptionMode::from_wire(&r.mode),
             retention_ms: r.retention_ms,
             created_at: r.created_at,
         }

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

@@ -1,7 +1,7 @@
 use diesel::prelude::*;
 
 use super::super::models::*;
-use super::super::records::NewSubscription;
+use super::super::records::{NewSubscription, SubscriptionMode};
 use super::super::schema::{topic_deliveries, topic_messages, topic_subscriptions, topics};
 use super::PostgresStorage;
 use crate::error::Result;
@@ -11,12 +11,17 @@ crate::storage::diesel_common::impl_diesel_pubsub_ops!(PostgresStorage);
 impl PostgresStorage {
     /// Declare a topic (idempotent upsert on `name`). `created_at` is preserved
     /// on re-declaration; only `mode`/`retention_ms` are updated.
-    pub fn declare_topic(&self, name: &str, mode: &str, retention_ms: Option<i64>) -> Result<()> {
+    pub fn declare_topic(
+        &self,
+        name: &str,
+        mode: SubscriptionMode,
+        retention_ms: Option<i64>,
+    ) -> Result<()> {
         crate::pubsub::validate_topic_declaration(mode, retention_ms)?;
         let mut conn = self.conn()?;
         let row = NewTopicRow {
             name,
-            mode,
+            mode: mode.as_str(),
             retention_ms,
             created_at: crate::job::now_millis(),
         };
@@ -53,7 +58,7 @@ impl PostgresStorage {
             priority: sub.priority,
             max_retries: sub.max_retries,
             timeout_ms: sub.timeout_ms,
-            mode: &sub.mode,
+            mode: sub.mode.as_str(),
         };
 
         // `cursor` is omitted so a re-registration preserves a log consumer's