diff --git a/crates/taskito-core/src/pubsub.rs b/crates/taskito-core/src/pubsub.rs index ba0f46a2..5cf382ef 100644 --- a/crates/taskito-core/src/pubsub.rs +++ b/crates/taskito-core/src/pubsub.rs @@ -4,16 +4,13 @@ //! same fan-out semantics — in particular the idempotency-key salting, which //! silently drops deliveries if a shell gets it wrong. -use std::collections::HashMap; - use crate::error::Result; use crate::job::{Job, NewJob}; use crate::storage::models::SubscriptionRow; use crate::storage::Storage; -/// Per-task delivery settings a subscriber's task declared at registration. -/// Shells pass what their task registry knows; unknown tasks fall back to -/// the queue-level defaults. +/// Queue-level fallback delivery settings, used when neither the publish call +/// nor the subscription row specifies a value. #[derive(Clone, Copy)] pub struct DeliveryDefaults { pub priority: i32, @@ -22,9 +19,9 @@ pub struct DeliveryDefaults { } /// Everything a publish shares across the jobs it fans out to. Per-subscriber -/// routing (task name, queue) comes from the subscription registry; delivery -/// settings resolve per field as explicit publish override, then the -/// subscriber task's own defaults, then the queue defaults. +/// routing (task name, queue) and delivery settings come from the subscription +/// registry; delivery settings resolve per field as explicit publish override, +/// then the subscription row's persisted setting, then the queue default. pub struct PublishRequest { pub topic: String, /// Wire-envelope payload bytes; every subscriber receives the same body. @@ -46,7 +43,6 @@ pub struct PublishRequest { pub result_ttl_ms: Option, pub namespace: Option, pub queue_defaults: DeliveryDefaults, - pub task_defaults: HashMap, } /// Fan a message out to every active subscription of `topic` as ordinary @@ -85,19 +81,28 @@ fn salted_unique_key(key: &str, topic: &str, subscription_name: &str) -> String } fn delivery_job(request: &PublishRequest, sub: &SubscriptionRow) -> NewJob { - let task = request - .task_defaults - .get(&sub.task_name) - .copied() - .unwrap_or(request.queue_defaults); + // Resolve each field independently: explicit publish override, then the + // subscription's persisted setting, then the queue default. Persisting on + // the row is what lets a producer-only process apply a subscriber's own + // retry/timeout/priority without ever loading its task code. + let defaults = request.queue_defaults; NewJob { queue: sub.queue.clone(), task_name: sub.task_name.clone(), payload: request.payload.clone(), - priority: request.priority.unwrap_or(task.priority), + priority: request + .priority + .or(sub.priority) + .unwrap_or(defaults.priority), scheduled_at: request.scheduled_at, - max_retries: request.max_retries.unwrap_or(task.max_retries), - timeout_ms: request.timeout_ms.unwrap_or(task.timeout_ms), + max_retries: request + .max_retries + .or(sub.max_retries) + .unwrap_or(defaults.max_retries), + timeout_ms: request + .timeout_ms + .or(sub.timeout_ms) + .unwrap_or(defaults.timeout_ms), unique_key: request .idempotency_key .as_ref() @@ -158,11 +163,23 @@ mod tests { max_retries: 3, timeout_ms: 300_000, }, - task_defaults: HashMap::new(), } } fn subscribe(storage: &SqliteStorage, topic: &str, name: &str, task: &str) { + subscribe_with(storage, topic, name, task, None, None, None); + } + + #[allow(clippy::too_many_arguments)] + fn subscribe_with( + storage: &SqliteStorage, + topic: &str, + name: &str, + task: &str, + priority: Option, + max_retries: Option, + timeout_ms: Option, + ) { storage .register_subscription(&NewSubscriptionRow { topic, @@ -173,6 +190,9 @@ mod tests { durable: true, owner_worker_id: None, created_at: now_millis(), + priority, + max_retries, + timeout_ms, }) .unwrap(); } @@ -227,28 +247,30 @@ mod tests { } #[test] - fn delivery_settings_resolve_override_then_task_then_queue() { + fn delivery_settings_resolve_override_then_subscription_then_queue() { + use std::collections::HashMap; let storage = SqliteStorage::in_memory().unwrap(); - subscribe(&storage, "orders", "email", "send_email"); + // send_email persisted its own settings on the subscription row; + // audit_log has none → queue defaults. + subscribe_with( + &storage, + "orders", + "email", + "send_email", + Some(5), + Some(0), + Some(60_000), + ); subscribe(&storage, "orders", "audit", "audit_log"); - let mut req = request("orders", None); - req.task_defaults.insert( - "send_email".to_string(), - DeliveryDefaults { - priority: 5, - max_retries: 0, - timeout_ms: 60_000, - }, - ); - let jobs = publish_to_topic(&storage, &req).unwrap(); + let jobs = publish_to_topic(&storage, &request("orders", None)).unwrap(); let by_task: HashMap<_, _> = jobs.iter().map(|j| (j.task_name.as_str(), j)).collect(); - // send_email declared its own settings; audit_log falls back to queue defaults. assert_eq!(by_task["send_email"].max_retries, 0); assert_eq!(by_task["send_email"].priority, 5); + assert_eq!(by_task["send_email"].timeout_ms, 60_000); assert_eq!(by_task["audit_log"].max_retries, 3); - // An explicit publish-level override beats both. + // An explicit publish-level override beats both the row and the default. let mut req = request("orders", None); req.max_retries = Some(9); let jobs = publish_to_topic(&storage, &req).unwrap(); diff --git a/crates/taskito-core/src/storage/diesel_common/migrations.rs b/crates/taskito-core/src/storage/diesel_common/migrations.rs index 23dd94e5..b2f34f6b 100644 --- a/crates/taskito-core/src/storage/diesel_common/migrations.rs +++ b/crates/taskito-core/src/storage/diesel_common/migrations.rs @@ -275,6 +275,9 @@ pub fn create_tables(d: &Dialect) -> Vec { durable {bool_true}, owner_worker_id TEXT, created_at {bi} NOT NULL, + priority INTEGER, + max_retries INTEGER, + timeout_ms {bi}, PRIMARY KEY (topic, subscription_name) )" ), @@ -387,6 +390,12 @@ pub fn alter_statements(d: &Dialect) -> Vec { format!("ALTER TABLE jobs ADD COLUMN {ife}has_deps {bool_false}"), // DLQ auto-retry counter: tracks how many times an entry was auto-retried format!("ALTER TABLE dead_letter ADD COLUMN {ife}dlq_retry_count INTEGER NOT NULL DEFAULT 0"), + // Per-subscription delivery settings: let publish_to_topic apply each + // subscriber's own retry/timeout/priority even from a producer process + // that never loaded the subscriber task. NULL = fall back to queue defaults. + format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}priority INTEGER"), + format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}max_retries INTEGER"), + format!("ALTER TABLE topic_subscriptions ADD COLUMN {ife}timeout_ms {bi}"), ] } diff --git a/crates/taskito-core/src/storage/models.rs b/crates/taskito-core/src/storage/models.rs index c8365b7d..40796536 100644 --- a/crates/taskito-core/src/storage/models.rs +++ b/crates/taskito-core/src/storage/models.rs @@ -421,6 +421,11 @@ pub struct SubscriptionRow { pub durable: bool, pub owner_worker_id: Option, pub created_at: i64, + /// Per-subscription delivery settings persisted at registration so + /// `publish_to_topic` applies them cross-process. `None` = queue default. + pub priority: Option, + pub max_retries: Option, + pub timeout_ms: Option, } /// Insertable/updatable struct for subscription registrations. @@ -435,6 +440,9 @@ pub struct NewSubscriptionRow<'a> { pub durable: bool, pub owner_worker_id: Option<&'a str>, pub created_at: i64, + pub priority: Option, + pub max_retries: Option, + pub timeout_ms: Option, } // ── Distributed Locks ─────────────────────────────────────────── diff --git a/crates/taskito-core/src/storage/postgres/pubsub.rs b/crates/taskito-core/src/storage/postgres/pubsub.rs index ea579ae3..3c4a31b4 100644 --- a/crates/taskito-core/src/storage/postgres/pubsub.rs +++ b/crates/taskito-core/src/storage/postgres/pubsub.rs @@ -30,6 +30,9 @@ impl PostgresStorage { topic_subscriptions::queue.eq(sub.queue), topic_subscriptions::durable.eq(sub.durable), topic_subscriptions::owner_worker_id.eq(sub.owner_worker_id), + topic_subscriptions::priority.eq(sub.priority), + topic_subscriptions::max_retries.eq(sub.max_retries), + topic_subscriptions::timeout_ms.eq(sub.timeout_ms), )) .execute(&mut conn)?; diff --git a/crates/taskito-core/src/storage/redis_backend/pubsub.rs b/crates/taskito-core/src/storage/redis_backend/pubsub.rs index 852dd688..83205b42 100644 --- a/crates/taskito-core/src/storage/redis_backend/pubsub.rs +++ b/crates/taskito-core/src/storage/redis_backend/pubsub.rs @@ -20,6 +20,14 @@ struct SubEntry { durable: bool, owner_worker_id: Option, created_at: i64, + // Older blobs (registered before per-subscription settings) lack these; + // default to None so they resolve to queue defaults, matching prior behavior. + #[serde(default)] + priority: Option, + #[serde(default)] + max_retries: Option, + #[serde(default)] + timeout_ms: Option, } impl From for SubscriptionRow { @@ -33,6 +41,9 @@ impl From for SubscriptionRow { durable: e.durable, owner_worker_id: e.owner_worker_id, created_at: e.created_at, + priority: e.priority, + max_retries: e.max_retries, + timeout_ms: e.timeout_ms, } } } @@ -84,6 +95,9 @@ impl RedisStorage { durable: sub.durable, owner_worker_id: sub.owner_worker_id.map(|s| s.to_string()), created_at: sub.created_at, + priority: sub.priority, + max_retries: sub.max_retries, + timeout_ms: sub.timeout_ms, }; let blob_key = self.key(&["sub", sub.topic, sub.subscription_name]); let by_topic = self.key(&["subs", "by_topic", sub.topic]); diff --git a/crates/taskito-core/src/storage/schema.rs b/crates/taskito-core/src/storage/schema.rs index bc0ea3ae..db7b6197 100644 --- a/crates/taskito-core/src/storage/schema.rs +++ b/crates/taskito-core/src/storage/schema.rs @@ -235,6 +235,9 @@ diesel::table! { durable -> Bool, owner_worker_id -> Nullable, created_at -> BigInt, + priority -> Nullable, + max_retries -> Nullable, + timeout_ms -> Nullable, } } diff --git a/crates/taskito-core/src/storage/sqlite/pubsub.rs b/crates/taskito-core/src/storage/sqlite/pubsub.rs index f932f087..9a470c83 100644 --- a/crates/taskito-core/src/storage/sqlite/pubsub.rs +++ b/crates/taskito-core/src/storage/sqlite/pubsub.rs @@ -29,6 +29,9 @@ impl SqliteStorage { topic_subscriptions::queue.eq(sub.queue), topic_subscriptions::durable.eq(sub.durable), topic_subscriptions::owner_worker_id.eq(sub.owner_worker_id), + topic_subscriptions::priority.eq(sub.priority), + topic_subscriptions::max_retries.eq(sub.max_retries), + topic_subscriptions::timeout_ms.eq(sub.timeout_ms), )) .execute(&mut conn)?; diff --git a/crates/taskito-core/src/storage/sqlite/tests.rs b/crates/taskito-core/src/storage/sqlite/tests.rs index 29b58b51..975093fa 100644 --- a/crates/taskito-core/src/storage/sqlite/tests.rs +++ b/crates/taskito-core/src/storage/sqlite/tests.rs @@ -1385,6 +1385,9 @@ fn make_sub<'a>( durable: owner.is_none(), owner_worker_id: owner, created_at, + priority: None, + max_retries: None, + timeout_ms: None, } } diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 13793853..e1d2ed10 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -870,6 +870,9 @@ fn test_topic_subscriptions_crud(s: &impl Storage) { durable: owner.is_none(), owner_worker_id: owner, created_at, + priority: None, + max_retries: None, + timeout_ms: None, }; // Upsert idempotency: re-registering (topic, name) updates in place. diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index b93844af..54d4ae36 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -3,8 +3,6 @@ //! 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 std::collections::HashMap; - use serde::{Deserialize, Serialize}; use taskito_core::job::{now_millis, Job, NewJob}; use taskito_core::pubsub::{DeliveryDefaults, PublishRequest}; @@ -103,18 +101,6 @@ pub struct PublishOptions { pub timeout_ms: Option, pub expires_ms: Option, pub result_ttl_ms: Option, - /// Per-task delivery settings from the SDK's task registry, keyed by task name. - pub task_defaults: Option>, -} - -/// A subscriber task's own delivery settings. Absent fields mirror -/// [`build_new_job`]'s zero defaults so a delivery resolves like a plain enqueue. -#[derive(Deserialize, Default, Clone, Copy)] -#[serde(rename_all = "camelCase", default)] -pub struct TaskDeliveryDefaults { - pub priority: i32, - pub max_retries: i32, - pub timeout_ms: i64, } /// Build a core [`PublishRequest`] from a publish call. The queue-level @@ -146,21 +132,6 @@ pub fn build_publish_request( max_retries: 0, timeout_ms: 0, }, - task_defaults: options - .task_defaults - .unwrap_or_default() - .into_iter() - .map(|(name, task)| { - ( - name, - DeliveryDefaults { - priority: task.priority, - max_retries: task.max_retries, - timeout_ms: task.timeout_ms, - }, - ) - }) - .collect(), } } @@ -326,6 +297,8 @@ pub struct WorkerOptions { } /// One topic subscription declared by the SDK, registered at worker start. +/// Delivery settings (from the subscriber task's own config) are persisted on +/// the row so a producer-only process applies them without loading the task. #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct SubscriptionSpec { @@ -334,6 +307,12 @@ pub struct SubscriptionSpec { pub task_name: String, pub queue: String, pub durable: bool, + #[serde(default)] + pub priority: Option, + #[serde(default)] + pub max_retries: Option, + #[serde(default)] + pub timeout_ms: Option, } /// A task's retry-backoff curve. Fields left unset fall back to the core's diff --git a/crates/taskito-java/src/queue/pubsub.rs b/crates/taskito-java/src/queue/pubsub.rs index 0c11e59d..83f69253 100644 --- a/crates/taskito-java/src/queue/pubsub.rs +++ b/crates/taskito-java/src/queue/pubsub.rs @@ -5,7 +5,7 @@ //! module only marshals arguments and views. use jni::objects::{JByteArray, JClass, JString}; -use jni::sys::{jboolean, jlong, jstring, JNI_FALSE}; +use jni::sys::{jboolean, jint, jlong, jstring, JNI_FALSE}; use jni::JNIEnv; use taskito_core::job::now_millis; use taskito_core::pubsub::publish_to_topic; @@ -21,9 +21,13 @@ use crate::ffi::{guard, new_string, read_bytes, read_optional_string, read_strin use super::{borrow_queue, to_jboolean}; /// `void registerSubscription(long handle, String topic, String subscriptionName, -/// String taskName, String queue, boolean durable, String ownerWorkerIdOrNull)` -/// — insert or update a subscription (idempotent on topic + name). +/// 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 +/// numerics use `i32::MIN` / `i64::MIN` as the "unset → queue default" sentinel, +/// since JNI carries primitives, not boxed nullables. #[no_mangle] +#[allow(clippy::too_many_arguments)] pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSubscription< 'local, >( @@ -36,6 +40,9 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu queue: JString<'local>, durable: jboolean, owner_worker_id: JString<'local>, + priority: jint, + max_retries: jint, + timeout_ms: jlong, ) { guard(&mut env, (), |env| { let queue_handle = unsafe { borrow_queue(handle) }; @@ -62,6 +69,9 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerSu durable: durable != 0, owner_worker_id: owner_worker_id.as_deref(), created_at: now_millis(), + 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), }; queue_handle.storage.register_subscription(&row)?; Ok(()) diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 2429880e..a1868436 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -219,6 +219,9 @@ fn register_subscriptions( durable: spec.durable, owner_worker_id: (!spec.durable).then_some(worker_id), created_at, + priority: spec.priority, + max_retries: spec.max_retries, + timeout_ms: spec.timeout_ms, }; storage.register_subscription(&row)?; } diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index f38bc688..a185fafb 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -1,8 +1,6 @@ //! Plain option objects passed from JavaScript. napi maps snake_case Rust //! fields to camelCase JS keys (`maxRetries`, `timeoutMs`). -use std::collections::HashMap; - use napi::bindgen_prelude::Buffer; use napi_derive::napi; @@ -54,20 +52,9 @@ pub struct EnqueueJob { pub options: Option, } -/// A subscriber task's own delivery settings, keyed by task name in -/// [`PublishOptions::task_defaults`]. Unset fields fall back to the queue -/// defaults, so the shell never has to duplicate them. -#[napi(object)] -#[derive(Default)] -pub struct DeliveryDefaultsInput { - pub priority: Option, - pub max_retries: Option, - pub timeout_ms: Option, -} - /// Options for [`crate::queue::JsQueue::publish`]. All optional — omitted -/// delivery settings resolve per subscriber (task defaults, then queue -/// defaults) in the core. +/// delivery settings resolve per subscriber (the subscription row's persisted +/// settings, then queue defaults) in the core. #[napi(object)] #[derive(Default)] pub struct PublishOptions { @@ -87,9 +74,6 @@ pub struct PublishOptions { /// Expire undelivered jobs at `now + expiresMs`. pub expires_ms: Option, pub result_ttl_ms: Option, - /// Per-task delivery defaults from the publisher's task registry, so a - /// subscriber's own registration options are honored. - pub task_defaults: Option>, } /// Filter for [`crate::queue::JsQueue::list_jobs`]. All fields optional. diff --git a/crates/taskito-node/src/queue/pubsub.rs b/crates/taskito-node/src/queue/pubsub.rs index ef5a44da..c887202a 100644 --- a/crates/taskito-node/src/queue/pubsub.rs +++ b/crates/taskito-node/src/queue/pubsub.rs @@ -2,8 +2,6 @@ //! the core (`taskito_core::pubsub`); every method touches storage, so each is //! async and offloads the blocking I/O to the blocking pool. -use std::collections::HashMap; - use napi::bindgen_prelude::{spawn_blocking, Buffer, Result}; use napi_derive::napi; use taskito_core::job::now_millis; @@ -12,7 +10,7 @@ use taskito_core::storage::models::NewSubscriptionRow; use taskito_core::Storage; use super::JsQueue; -use crate::config::{DeliveryDefaultsInput, PublishOptions}; +use crate::config::PublishOptions; use crate::convert::{ job_to_js, subscription_to_js, JsJob, JsSubscription, DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS, @@ -25,6 +23,7 @@ impl JsQueue { /// Ephemeral subscriptions (`durable: false`) carry the owning worker's id /// so they can be reaped once that worker is gone. #[napi] + #[allow(clippy::too_many_arguments)] pub async fn register_subscription( &self, topic: String, @@ -33,6 +32,9 @@ impl JsQueue { queue: String, durable: bool, owner_worker_id: Option, + priority: Option, + max_retries: Option, + timeout_ms: Option, ) -> Result<()> { // An owner-less ephemeral row could never be reaped — reject it before // it reaches storage. @@ -52,6 +54,9 @@ impl JsQueue { durable, owner_worker_id: owner_worker_id.as_deref(), created_at: now_millis(), + priority, + max_retries, + timeout_ms, }; storage.register_subscription(&row).map_err(to_napi_err) }) @@ -205,38 +210,5 @@ fn build_publish_request( result_ttl_ms, namespace, queue_defaults, - task_defaults: task_defaults(opts.task_defaults.unwrap_or_default(), queue_defaults)?, }) } - -/// Resolve per-task delivery defaults, filling unset fields from the queue -/// defaults so the shell never duplicates them. Entries get the same -/// non-negative checks as the top-level options, naming the offending task. -fn task_defaults( - input: HashMap, - queue_defaults: DeliveryDefaults, -) -> Result> { - input - .into_iter() - .map(|(name, defaults)| { - let max_retries = match defaults.max_retries { - Some(n) => { - non_negative(n as i64, &format!("taskDefaults[\"{name}\"].maxRetries"))? as i32 - } - None => queue_defaults.max_retries, - }; - let timeout_ms = match defaults.timeout_ms { - Some(n) => non_negative(n, &format!("taskDefaults[\"{name}\"].timeoutMs"))?, - None => queue_defaults.timeout_ms, - }; - Ok(( - name, - DeliveryDefaults { - priority: defaults.priority.unwrap_or(queue_defaults.priority), - max_retries, - timeout_ms, - }, - )) - }) - .collect() -} diff --git a/crates/taskito-python/src/py_queue/pubsub.rs b/crates/taskito-python/src/py_queue/pubsub.rs index 98968774..a5d44041 100644 --- a/crates/taskito-python/src/py_queue/pubsub.rs +++ b/crates/taskito-python/src/py_queue/pubsub.rs @@ -1,5 +1,3 @@ -use std::collections::HashMap; - use pyo3::prelude::*; use taskito_core::job::now_millis; @@ -17,7 +15,12 @@ type SubscriptionTuple = (String, String, String, String, bool, bool); #[pymethods] impl PyQueue { /// Insert or update a topic subscription (idempotent on topic + name). - #[pyo3(signature = (topic, subscription_name, task_name, queue="default", durable=true, owner_worker_id=None))] + /// + /// `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))] + #[allow(clippy::too_many_arguments)] pub fn register_subscription( &self, topic: &str, @@ -26,6 +29,9 @@ impl PyQueue { queue: &str, durable: bool, owner_worker_id: Option<&str>, + priority: Option, + max_retries: Option, + timeout_ms: Option, ) -> PyResult<()> { // An unowned ephemeral row could never be reaped (cleanup keys off // live worker ids), so it would stay active forever. @@ -43,6 +49,9 @@ impl PyQueue { durable, owner_worker_id, created_at: now_millis(), + priority, + max_retries, + timeout_ms, }; self.storage .register_subscription(&row) @@ -108,7 +117,7 @@ impl PyQueue { /// Publish a payload to a topic: one job per active subscription. /// Returns the created jobs — empty when nothing is subscribed. - #[pyo3(signature = (topic, payload, idempotency_key=None, metadata=None, notes=None, priority=None, delay_seconds=None, max_retries=None, timeout=None, expires=None, result_ttl=None, task_defaults=None))] + #[pyo3(signature = (topic, payload, idempotency_key=None, metadata=None, notes=None, priority=None, delay_seconds=None, max_retries=None, timeout=None, expires=None, result_ttl=None))] #[allow(clippy::too_many_arguments)] pub fn publish( &self, @@ -123,7 +132,6 @@ impl PyQueue { timeout: Option, expires: Option, result_ttl: Option, - task_defaults: Option>, ) -> PyResult> { let now = now_millis(); let scheduled_at = match delay_seconds { @@ -185,20 +193,6 @@ impl PyQueue { max_retries: self.default_retry, timeout_ms: self.default_timeout.saturating_mul(1000), }, - task_defaults: task_defaults - .unwrap_or_default() - .into_iter() - .map(|(name, (priority, max_retries, timeout_ms))| { - ( - name, - DeliveryDefaults { - priority, - max_retries, - timeout_ms, - }, - ) - }) - .collect(), }; let jobs = publish_to_topic(&self.storage, &request) .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; 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 2bba3195..a3bfdb55 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -681,7 +681,16 @@ public Taskito subscribe(String topic, Task task, SubscriptionOptions opt // Register before recording locally, so a failed registration leaves no // orphaned local declaration behind. if (options.durable()) { - backend.registerSubscription(topic, name, task.name(), options.queue(), true, null); + backend.registerSubscription( + topic, + name, + task.name(), + options.queue(), + true, + null, + taskDefaults.priority(), + taskDefaults.maxRetries(), + taskDefaults.timeoutMs()); } // Re-declaring (topic, name) replaces the previous local entry, mirroring // the backend's upsert instead of accumulating duplicates. @@ -713,11 +722,11 @@ public List publish(String topic, Object payload, PublishOptions options) { } /** - * The wire shape of a publish: the caller's options plus each locally - * subscribed task's own delivery settings, so deliveries honor per-task - * defaults. Tasks subscribed elsewhere fall back to the core defaults. + * The wire shape of a publish: just the caller's own options. Per-subscriber + * delivery defaults are persisted on each subscription row (see + * {@link #subscribe}), so the core resolves them without the producer's help. */ - private Map publishRequest(PublishOptions options) { + private static Map publishRequest(PublishOptions options) { Map request = new LinkedHashMap<>(); putIfSet(request, "idempotencyKey", options.idempotencyKey()); putIfSet(request, "metadata", options.metadata()); @@ -728,17 +737,6 @@ private Map publishRequest(PublishOptions options) { putIfSet(request, "timeoutMs", options.timeoutMs()); putIfSet(request, "expiresMs", options.expiresMs()); putIfSet(request, "resultTtlMs", options.resultTtlMs()); - Map> taskDefaults = new LinkedHashMap<>(); - for (SubscriptionConfig config : subscriptions) { - Map defaults = new LinkedHashMap<>(); - defaults.put("priority", config.taskPriority() == null ? 0 : config.taskPriority()); - defaults.put("maxRetries", config.taskMaxRetries() == null ? 0 : config.taskMaxRetries()); - defaults.put("timeoutMs", config.taskTimeoutMs() == null ? 0L : config.taskTimeoutMs()); - taskDefaults.put(config.taskName(), defaults); - } - if (!taskDefaults.isEmpty()) { - request.put("taskDefaults", taskDefaults); - } return request; } 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 7690e07b..afdb67fe 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 @@ -294,9 +294,39 @@ public void registerSubscription( String queue, boolean durable, String ownerWorkerIdOrNull) { + // Route the settings-less form through the full one (nulls = queue + // defaults) so this backend is reachable via either overload. + registerSubscription(topic, subscriptionName, taskName, queue, durable, ownerWorkerIdOrNull, null, null, null); + } + + @Override + public void registerSubscription( + String topic, + String subscriptionName, + String taskName, + String queue, + boolean durable, + String ownerWorkerIdOrNull, + Integer priority, + Integer maxRetries, + Long timeoutMs) { + // 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; + int nativeMaxRetries = maxRetries == null ? Integer.MIN_VALUE : maxRetries; + long nativeTimeoutMs = timeoutMs == null ? Long.MIN_VALUE : timeoutMs; withOpenHandle(() -> { NativeQueue.registerSubscription( - handle, topic, subscriptionName, taskName, queue, durable, ownerWorkerIdOrNull); + handle, + topic, + subscriptionName, + taskName, + queue, + durable, + ownerWorkerIdOrNull, + nativePriority, + nativeMaxRetries, + nativeTimeoutMs); return null; }); } 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 d7f36855..97c1eccc 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 @@ -136,7 +136,12 @@ public static native long registerPeriodic( public static native boolean setPeriodicEnabled(long handle, String name, boolean enabled); // ── Pub/Sub ───────────────────────────────────────────────────── - /** Insert or update a topic subscription (idempotent on topic + name). */ + /** + * 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". + */ public static native void registerSubscription( long handle, String topic, @@ -144,7 +149,10 @@ public static native void registerSubscription( String taskName, String queue, boolean durable, - String ownerWorkerIdOrNull); + String ownerWorkerIdOrNull, + int priority, + int maxRetries, + long timeoutMs); /** A JSON array of subscriptions — all of them, or only a topic's active ones. */ public static native String listSubscriptions(long handle, String topicOrNull); 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 e5506773..b4bc52f7 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 @@ -167,10 +167,36 @@ default boolean setPeriodicEnabled(String name, boolean enabled) { // compiling and fail explicitly only when pub/sub is actually used. String PUBSUB_UNSUPPORTED = "pub/sub not supported by this backend"; + /** + * Insert or update a topic subscription with no per-subscriber delivery + * settings; deliveries take the queue defaults. This is the base method a + * backend overrides — the {@code Integer/Integer/Long} overload delegates + * here so a backend that only implements this form keeps working. + */ + default void registerSubscription( + String topic, + String subscriptionName, + String taskName, + String queue, + boolean durable, + String ownerWorkerIdOrNull) { + throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); + } + /** * Insert or update a topic subscription (idempotent on topic + name). * Re-registering updates routing but preserves a paused state; an ephemeral * subscription ({@code durable=false}) requires {@code ownerWorkerIdOrNull}. + * + *

{@code priority}, {@code maxRetries}, and {@code timeoutMs} are the + * subscriber task's own delivery settings, persisted on the row so a + * producer-only process applies them without loading the task; {@code null} + * means "take the queue default". + * + *

The default drops the three settings and delegates to the six-argument + * form, so a backend that predates delivery-setting persistence still + * registers the subscription (it just takes queue defaults) instead of + * throwing. */ default void registerSubscription( String topic, @@ -178,8 +204,11 @@ default void registerSubscription( String taskName, String queue, boolean durable, - String ownerWorkerIdOrNull) { - throw new UnsupportedOperationException(PUBSUB_UNSUPPORTED); + String ownerWorkerIdOrNull, + Integer priority, + Integer maxRetries, + Long timeoutMs) { + registerSubscription(topic, subscriptionName, taskName, queue, durable, ownerWorkerIdOrNull); } /** A JSON array of subscriptions — all of them, or only a topic's active ones. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 4630e979..239d3fe8 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -440,7 +440,11 @@ private String encodeOptions() { } } - /** Serialize the declared topic subscriptions into the wire shape the binding reads. */ + /** + * Serialize the declared topic subscriptions into the wire shape the binding reads. + * The subscriber task's delivery settings are emitted only when set, so an unset + * one is persisted as "take the queue default" rather than a spurious zero. + */ private List> encodeSubscriptions() { List> specs = new ArrayList<>(subscriptions.size()); for (SubscriptionConfig config : subscriptions) { @@ -450,6 +454,15 @@ private List> encodeSubscriptions() { spec.put("taskName", config.taskName()); spec.put("queue", config.queue()); spec.put("durable", config.durable()); + if (config.taskPriority() != null) { + spec.put("priority", config.taskPriority()); + } + if (config.taskMaxRetries() != null) { + spec.put("maxRetries", config.taskMaxRetries()); + } + if (config.taskTimeoutMs() != null) { + spec.put("timeoutMs", config.taskTimeoutMs()); + } specs.add(spec); } return specs; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubTest.java index 68d454ac..9fc28ece 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/PubSubTest.java @@ -183,6 +183,28 @@ void deliveriesHonorSubscriberTaskDefaultsAndPublishOverrides(@TempDir Path dir) } } + @Test + void producerOnlyPublishAppliesPersistedRowDeliverySettings(@TempDir Path dir) throws Exception { + String options = new ObjectMapper() + .writeValueAsString( + Map.of("backend", "sqlite", "dsn", dir.resolve("t.db").toString())); + JniQueueBackend backend = JniQueueBackend.open(options); + try (Taskito producer = Taskito.builder().open(backend)) { + // A subscriber elsewhere persisted its own delivery settings on the row. + backend.registerSubscription("orders", "sink", SEND_EMAIL.name(), "default", true, null, 5, 7, 30_000L); + + // A producer with no local knowledge of SEND_EMAIL still fans the + // delivery out with the subscriber's persisted settings. + List deliveries = producer.publish("orders", "o-1"); + assertEquals(1, deliveries.size()); + Job delivery = deliveries.get(0); + assertEquals(SEND_EMAIL.name(), delivery.taskName); + assertEquals(5, delivery.priority); + assertEquals(7, delivery.maxRetries); + assertEquals(30_000L, delivery.timeoutMs); + } + } + @Test @Timeout(30) void ephemeralSubscriptionRegistersWithARunningWorker(@TempDir Path dir) throws Exception { diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index 69d7036b..254fd383 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -472,14 +472,29 @@ public synchronized void registerSubscription( String queue, boolean durable, String ownerWorkerIdOrNull) { + // Reachable via either overload; nulls take the queue defaults. + registerSubscription(topic, subscriptionName, taskName, queue, durable, ownerWorkerIdOrNull, null, null, null); + } + + @Override + public synchronized void registerSubscription( + String topic, + String subscriptionName, + String taskName, + String queue, + boolean durable, + String ownerWorkerIdOrNull, + Integer priority, + Integer maxRetries, + Long timeoutMs) { // Mirrors the native guard: an ownerless ephemeral row would never be // reaped yet keep receiving deliveries. if (!durable && ownerWorkerIdOrNull == null) { throw new TaskitoException("an ephemeral subscription (durable=false) requires ownerWorkerId"); } - // Upsert mirrors the core: routing (task, queue, durable, owner) is - // replaced while active and creation time survive — re-declaring never - // resumes a paused subscription or refreshes the reap grace window. + // Upsert mirrors the core: routing (task, queue, durable, owner) and the + // persisted delivery settings are replaced while active and creation time + // survive — re-declaring never resumes a pause or refreshes the reap grace. subscriptions.compute(subscriptionKey(topic, subscriptionName), (key, existing) -> { SubscriptionRec rec = new SubscriptionRec( topic, @@ -488,6 +503,9 @@ public synchronized void registerSubscription( queue, durable, ownerWorkerIdOrNull, + priority, + maxRetries, + timeoutMs, existing == null ? seq.incrementAndGet() : existing.createdSeq, existing == null ? now() : existing.createdAt); if (existing != null) { @@ -591,18 +609,19 @@ private static Map subscriptionView(SubscriptionRec sub) { /** * One delivery's enqueue options: routing from the subscription, delivery - * settings resolving publish override → the subscriber task's defaults → - * zero, the idempotency key salted per subscriber (the unique-key scan is - * global, so a raw key would dedup away all but one delivery), and the - * caller's notes stamped with {@code topic}/{@code subscription}. + * settings resolving publish override → the subscription row's persisted + * setting → zero (so a producer-only publish applies the subscriber's own + * settings without knowing the task), the idempotency key salted per + * subscriber (the unique-key scan is global, so a raw key would dedup away + * all but one delivery), and the caller's notes stamped with + * {@code topic}/{@code subscription}. */ private static ObjectNode deliveryOptions(JsonNode opts, SubscriptionRec sub) { - JsonNode taskDefaults = opts == null ? null : opts.path("taskDefaults").get(sub.taskName); ObjectNode delivery = JSON.createObjectNode(); delivery.put("queue", sub.queue); - delivery.put("priority", resolveDeliveryInt(opts, taskDefaults, "priority")); - delivery.put("maxRetries", resolveDeliveryInt(opts, taskDefaults, "maxRetries")); - delivery.put("timeoutMs", resolveDeliveryLong(opts, taskDefaults, "timeoutMs")); + delivery.put("priority", resolveDeliveryInt(opts, "priority", sub.priority)); + delivery.put("maxRetries", resolveDeliveryInt(opts, "maxRetries", sub.maxRetries)); + delivery.put("timeoutMs", resolveDeliveryLong(opts, "timeoutMs", sub.timeoutMs)); delivery.put("delayMs", optLong(opts, "delayMs", 0)); // Publish-level expiry and result retention pass straight through; the // enqueue path resolves them (expiresMs → an absolute expiry). @@ -628,20 +647,20 @@ private static void copyIfSet(JsonNode opts, ObjectNode delivery, String field) } } - private static int resolveDeliveryInt(JsonNode opts, JsonNode taskDefaults, String field) { + private static int resolveDeliveryInt(JsonNode opts, String field, Integer rowSetting) { JsonNode override = opts == null ? null : opts.get(field); if (override != null && !override.isNull()) { return override.asInt(); } - return optInt(taskDefaults, field, 0); + return rowSetting == null ? 0 : rowSetting; } - private static long resolveDeliveryLong(JsonNode opts, JsonNode taskDefaults, String field) { + private static long resolveDeliveryLong(JsonNode opts, String field, Long rowSetting) { JsonNode override = opts == null ? null : opts.get(field); if (override != null && !override.isNull()) { return override.asLong(); } - return optLong(taskDefaults, field, 0); + return rowSetting == null ? 0L : rowSetting; } /** The caller's notes (if any) with {@code topic} and {@code subscription} stamped in. */ @@ -694,7 +713,10 @@ private void registerWorkerSubscriptions(String workerId, JsonNode options) { text(spec, "taskName"), optText(spec, "queue", DEFAULT_QUEUE), durable, - durable ? null : workerId); + durable ? null : workerId, + boxedInt(spec, "priority"), + boxedInt(spec, "maxRetries"), + boxedLong(spec, "timeoutMs")); } } @@ -965,6 +987,18 @@ private static long optLong(JsonNode node, String field, long fallback) { return value == null || value.isNull() ? fallback : value.asLong(); } + /** A nullable int field — absent/null stays {@code null} so it reads as "queue default". */ + private static Integer boxedInt(JsonNode node, String field) { + JsonNode value = node == null ? null : node.get(field); + return value == null || value.isNull() ? null : value.asInt(); + } + + /** A nullable long field — absent/null stays {@code null} so it reads as "queue default". */ + private static Long boxedLong(JsonNode node, String field) { + JsonNode value = node == null ? null : node.get(field); + return value == null || value.isNull() ? null : value.asLong(); + } + /** A stored job. */ private static final class JobRec { String id; @@ -1001,6 +1035,10 @@ private static final class SubscriptionRec { final String queue; final boolean durable; final String ownerWorkerId; + // The subscriber task's persisted delivery settings; null = queue default. + final Integer priority; + final Integer maxRetries; + final Long timeoutMs; final long createdSeq; volatile long createdAt; volatile boolean active = true; @@ -1012,6 +1050,9 @@ private static final class SubscriptionRec { String queue, boolean durable, String ownerWorkerId, + Integer priority, + Integer maxRetries, + Long timeoutMs, long createdSeq, long createdAt) { this.topic = topic; @@ -1020,6 +1061,9 @@ private static final class SubscriptionRec { this.queue = queue; this.durable = durable; this.ownerWorkerId = ownerWorkerId; + this.priority = priority; + this.maxRetries = maxRetries; + this.timeoutMs = timeoutMs; this.createdSeq = createdSeq; this.createdAt = createdAt; } diff --git a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryPubSubTest.java b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryPubSubTest.java index 4175ce54..3cf45fab 100644 --- a/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryPubSubTest.java +++ b/sdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryPubSubTest.java @@ -181,6 +181,26 @@ void deliveriesCarryStampedNotesAndTaskDefaults() { } } + @Test + void producerOnlyPublishAppliesPersistedRowDeliverySettings() { + // A subscriber (in another process) persisted its own delivery settings + // on the subscription row, as a durable subscribe() does. + InMemoryQueueBackend backend = new InMemoryQueueBackend(); + backend.registerSubscription("orders", "sink", AUDIT.name(), "default", true, null, 5, 7, 30_000L); + + // A producer that never registered AUDIT publishes with no local task + // knowledge; the delivery still carries the subscriber's own settings. + try (Taskito producer = Taskito.builder().open(backend)) { + List deliveries = producer.publish("orders", "o-1"); + assertEquals(1, deliveries.size()); + Job delivery = deliveries.get(0); + assertEquals(AUDIT.name(), delivery.taskName); + assertEquals(5, delivery.priority); + assertEquals(7, delivery.maxRetries); + assertEquals(30_000L, delivery.timeoutMs); + } + } + @Test @Timeout(20) void ephemeralSubscriptionBindsToAWorkerAndIsReapedWhenItStops() { diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index fe1bf35c..063065a3 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -17,7 +17,6 @@ export type NativeWorker = InstanceType; export type { CircuitBreakerInput, - DeliveryDefaultsInput, EnqueueOptions, JobFilter, JsCircuitBreaker, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index a022f4d7..5deeca86 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -25,7 +25,6 @@ import { type Interception, InterceptionMetrics, type Interceptor } from "./inte import { Lock, type LockOptions } from "./locks"; import type { EnqueueContext, Middleware } from "./middleware"; import { - type DeliveryDefaultsInput, JsQueue, type EnqueueOptions as NativeEnqueueOptions, type NativeQueue, @@ -464,7 +463,6 @@ export class Queue { return this.native.publish(topic, payload, { ...rest, notes: notes === undefined ? undefined : encodeNotes(notes), - taskDefaults: this.deliveryTaskDefaults(), }); } @@ -539,6 +537,11 @@ export class Queue { subscription: PendingSubscription, ownerWorkerId: string | undefined, ): Promise { + // Persist the subscriber's own delivery settings on the subscription row so + // a producer-only process — which never registers the task — still applies + // them when it publishes. There is no per-task priority option, so priority + // stays undefined and falls back to the queue default in the core. + const options = this.tasks.get(subscription.taskName)?.options; return this.native.registerSubscription( subscription.topic, subscription.subscriptionName, @@ -546,26 +549,12 @@ export class Queue { subscription.queue, subscription.durable, ownerWorkerId, + undefined, + options?.maxRetries, + options?.timeoutMs, ); } - /** - * Per-task delivery defaults from this process's task registry so a publish - * honors each subscriber's own registration options. Unset fields and tasks - * registered elsewhere fall back to queue defaults in the core. - */ - private deliveryTaskDefaults(): Record { - const defaults: Record = {}; - for (const [name, task] of this.tasks) { - const { maxRetries, timeoutMs } = task.options ?? {}; - if (maxRetries === undefined && timeoutMs === undefined) { - continue; - } - defaults[name] = { maxRetries, timeoutMs }; - } - return defaults; - } - /** * Run enqueue interceptors, merge per-task defaults, run the middleware * `onEnqueue` hooks, then serialize the args and encode the options — the diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index fc0bce23..eb3e2cdc 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -35,10 +35,10 @@ export interface EnqueueOptions extends Omit { /** * Per-publish options. Mirrors the native options, but `notes` is a structured - * object here (validated and JSON-encoded before it reaches the core) and - * `taskDefaults` is built internally from the queue's task registry. + * object here (validated and JSON-encoded before it reaches the core) rather + * than a pre-encoded string. */ -export interface PublishOptions extends Omit { +export interface PublishOptions extends Omit { /** Structured annotations stamped on every delivery (plus `topic`/`subscription`). */ notes?: Record; } diff --git a/sdks/node/test/core/pubsub.test.ts b/sdks/node/test/core/pubsub.test.ts index e1b7a1e3..56e4ad7c 100644 --- a/sdks/node/test/core/pubsub.test.ts +++ b/sdks/node/test/core/pubsub.test.ts @@ -98,10 +98,22 @@ it("registering an ephemeral subscription without an owner rejects", async () => ).rejects.toThrow(/ephemeral subscription \(durable=false\) requires ownerWorkerId/); }); -it("publish rejects negative per-task delivery defaults naming the task", async () => { - const queue = newQueue(); - queue.task("bad_task", () => undefined, { maxRetries: -1 }); - await expect(queue.publish("orders", [1])).rejects.toThrow('taskDefaults["bad_task"].maxRetries'); +it("persists a subscriber's delivery settings for a producer-only publisher", async () => { + // Two Queue instances on the same DB: the consumer registers the subscriber + // (persisting its maxRetries on the subscription row), the producer only + // publishes and never registers the task — yet the delivery must honor 7. + const dbPath = join(mkdtempSync(join(tmpdir(), "taskito-pubsub-")), "q.db"); + const consumer = new Queue({ dbPath }); + consumer.subscriber("orders", "send_email", () => undefined, { + subscriptionName: "email", + maxRetries: 7, + }); + await consumer.declareSubscriptions(); + + const producer = new Queue({ dbPath }); + const [job] = await producer.publish("orders", [1]); + expect(job).toBeDefined(); + expect(producer.getJob(job?.id ?? "")?.maxRetries).toBe(7); }); it("unsubscribe removes the subscription and reports a missing one", async () => { diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index d30b2799..c9aaf422 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -194,6 +194,9 @@ class PyQueue: queue: str = "default", durable: bool = True, owner_worker_id: str | None = None, + priority: int | None = None, + max_retries: int | None = None, + timeout_ms: int | None = None, ) -> None: ... def list_subscriptions( self, topic: str | None = None @@ -216,7 +219,6 @@ class PyQueue: timeout: int | None = None, expires: float | None = None, result_ttl: int | None = None, - task_defaults: dict[str, tuple[int, int, int]] | None = None, ) -> list[PyJob]: ... def run_worker( self, diff --git a/sdks/python/taskito/mixins/pubsub.py b/sdks/python/taskito/mixins/pubsub.py index fdd9ee4e..4d1dfb6e 100644 --- a/sdks/python/taskito/mixins/pubsub.py +++ b/sdks/python/taskito/mixins/pubsub.py @@ -142,7 +142,6 @@ def publish( timeout=timeout, expires=expires, result_ttl=result_ttl, - task_defaults=self._delivery_task_defaults(), ) return [JobResult(py_job=py_job, queue=self) for py_job in py_jobs] # type: ignore[arg-type] @@ -191,14 +190,18 @@ def list_topics(self) -> list[str]: def _delivery_task_defaults(self) -> dict[str, tuple[int, int, int]]: """Per-task ``(priority, max_retries, timeout_ms)`` from this process's - task registry, so deliveries honor each subscriber's own settings. - Tasks registered elsewhere fall back to queue defaults.""" + task registry. Persisted on the subscription row at registration so a + producer-only process (which never loads the task) still applies each + subscriber's own settings. Tasks not registered here get queue defaults.""" return { config.name: (config.priority, config.max_retries, config.timeout * 1000) for config in self._task_configs # type: ignore[attr-defined] } def _register_subscription(self, config: dict[str, Any], owner_worker_id: str | None) -> None: + priority, max_retries, timeout_ms = self._delivery_task_defaults().get( + config["task_name"], (None, None, None) + ) self._inner.register_subscription( topic=config["topic"], subscription_name=config["subscription_name"], @@ -206,6 +209,9 @@ def _register_subscription(self, config: dict[str, Any], owner_worker_id: str | queue=config["queue"], durable=config["durable"], owner_worker_id=owner_worker_id, + priority=priority, + max_retries=max_retries, + timeout_ms=timeout_ms, ) def _declare_worker_subscriptions(self, worker_id: str) -> None: diff --git a/sdks/python/tests/core/test_pubsub.py b/sdks/python/tests/core/test_pubsub.py index 5e8480f1..e5c9c9ee 100644 --- a/sdks/python/tests/core/test_pubsub.py +++ b/sdks/python/tests/core/test_pubsub.py @@ -178,6 +178,39 @@ def send_email(order_id: int) -> None: ... assert len(deliveries) == 2 +class TestCrossProcessDeliverySettings: + def test_producer_without_task_applies_persisted_settings( + self, tmp_path: Any, queue: Queue, run_worker: threading.Thread, poll_until: PollUntil + ) -> None: + """A subscriber's own @task settings must reach deliveries even when the + publisher never registered the task — proving the settings persist on the + subscription row, not just in the publishing process's registry.""" + + @queue.subscriber("orders", name="picky", max_retries=7, priority=4, timeout=42) + def handle(order_id: int) -> None: ... + + queue.declare_subscriptions() + + # A separate Queue on the same DB stands in for a producer-only process: + # it has no task registry at all. + producer = Queue(db_path=str(tmp_path / "test.db")) + (delivery,) = producer.publish("orders", 1) + + job = queue.get_job(delivery.id) + assert job is not None + assert job._py_job.max_retries == 7 + assert job._py_job.priority == 4 + + def test_publish_override_beats_persisted_setting(self, queue: Queue) -> None: + @queue.subscriber("orders", name="picky", max_retries=7) + def handle(order_id: int) -> None: ... + + queue.declare_subscriptions() + (delivery,) = queue.publish("orders", 1, max_retries=1) + job = queue.get_job(delivery.id) + assert job is not None and job._py_job.max_retries == 1 + + class TestEphemeralSubscriptions: def test_ephemeral_registers_only_with_worker(self, queue: Queue) -> None: @queue.subscriber("orders", name="debug-tail", durable=False)