feat(java): signal retry intent by exception type

crates/taskito-core/BINDING_CONTRACT.md

@@ -236,9 +236,8 @@ what it applies to the settings KV on every cleanup sweep, and every shell's
 dashboard echoes that document instead of guessing at the defaults.
 
 - **Key** `retention:effective:<namespace>` (unnamespaced queues use `default`).
-  The `retention:` prefix is **reserved**: a shell's settings API must treat it
-  the way it treats `auth:` — never listed, written, or deleted through the
-  generic KV endpoints, so the published policy cannot be spoofed.
+  The `retention:` prefix is **reserved** (see below), so the published policy
+  cannot be spoofed through a dashboard's generic KV endpoints.
 - **Document** (snake_case; windows in **milliseconds**, `null` = keep forever):
   `enabled`, `defaulted`, `namespace`, `reported_at` (Unix ms), and `windows` with
   `archived_jobs_ttl_ms`, `dead_letter_ttl_ms`, `task_logs_ttl_ms`,
@@ -251,6 +250,19 @@ dashboard echoes that document instead of guessing at the defaults.
 - Shells read it through `scheduler::retention::read_effective_retention_json`
   rather than parsing the key themselves.
 
+## Reserved settings prefixes (cross-SDK)
+The settings KV also backs auth state, webhook subscriptions, and the retention
+document above. None of it belongs on the dashboard's generic key/value surface:
+reads leak credentials, writes spoof a published policy.
+
+- `settings::RESERVED_SETTING_PREFIXES` is the canonical list, exported by every
+  binding (`reserved_setting_prefixes()` / `NativeQueue.reservedSettingPrefixes()`).
+  A shell's settings API MUST derive its hide list from it rather than
+  hardcoding one — a prefix missed in one shell reopens the hole for all of them.
+- A key under a reserved prefix is **absent** to that API: never listed, read,
+  written, or deleted through it. The runtime's own readers and writers use the
+  `Storage` settings methods, which stay unrestricted.
+
 ## 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`,

crates/taskito-core/src/lib.rs

@@ -12,6 +12,8 @@ pub mod pubsub;
 pub mod resilience;
 /// The [`Scheduler`]: job dispatch, retries, maintenance, retention.
 pub mod scheduler;
+/// Reserved settings-key prefixes: the namespaces the runtime owns.
+pub mod settings;
 /// The [`Storage`] trait, backend implementations, and shared records.
 pub mod storage;
 /// Native worker: task registry, dispatcher trait, worker runner.
@@ -29,6 +31,7 @@ pub use scheduler::retention::{EffectiveRetention, RetentionConfig};
 pub use scheduler::{
     JobResult, QueueConfig, ResultOutcome, Scheduler, SchedulerConfig, TaskConfig,
 };
+pub use settings::{is_reserved_setting_key, RESERVED_SETTING_PREFIXES};
 pub use storage::cursor::Page;
 #[cfg(feature = "postgres")]
 pub use storage::postgres::PostgresStorage;

crates/taskito-core/src/scheduler/retention.rs

@@ -149,7 +149,16 @@ pub fn read_effective_retention<S: Storage>(
     let Some(raw) = storage.get_setting(&retention_setting_key(namespace))? else {
         return Ok(None);
     };
-    Ok(serde_json::from_str(&raw).ok())
+    // Unparseable reads as unreported, but say so: during a rolling upgrade that
+    // is a schema mismatch, not "no leader has swept yet".
+    Ok(serde_json::from_str(&raw)
+        .inspect_err(|error| {
+            log::warn!(
+                "published retention document for {} is unparseable: {error}",
+                namespace.unwrap_or(DEFAULT_NAMESPACE)
+            )
+        })
+        .ok())
 }
 
 /// The published document for `namespace` as JSON, re-encoded from the parsed

crates/taskito-core/src/settings.rs

@@ -0,0 +1,50 @@
+//! Reserved settings-key prefixes — the settings namespaces the runtime owns.
+//!
+//! The settings KV doubles as the store for auth state, webhook subscriptions,
+//! and the retention windows a cleanup leader publishes. None of those belong on
+//! a dashboard's generic key/value surface: reading them leaks credentials and
+//! writing them spoofs a published policy. The canonical list lives here so that
+//! every binding hides the same keys instead of re-deriving it — a prefix missed
+//! in one shell reopens the hole for all of them.
+
+/// Key prefixes a shell's generic settings API must treat as absent — never
+/// listed, read, written, or deleted through it. The runtime's own readers and
+/// writers use the [`Storage`](crate::storage::Storage) settings methods, which
+/// stay unrestricted.
+pub const RESERVED_SETTING_PREFIXES: &[&str] = &[
+    "auth:",            // dashboard sessions, OAuth state, API tokens
+    "retention:",       // the windows a cleanup leader publishes
+    "taskito.webhooks", // webhook store
+    "webhook:",         // webhook store
+    "webhooks:",        // webhook subscriptions and delivery log
+];
+
+/// Whether `key` falls under a reserved prefix.
+pub fn is_reserved_setting_key(key: &str) -> bool {
+    RESERVED_SETTING_PREFIXES
+        .iter()
+        .any(|prefix| key.starts_with(prefix))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::scheduler::retention::retention_setting_key;
+
+    #[test]
+    fn runtime_owned_namespaces_are_reserved() {
+        assert!(is_reserved_setting_key(&retention_setting_key(None)));
+        assert!(is_reserved_setting_key(&retention_setting_key(Some(
+            "billing"
+        ))));
+        assert!(is_reserved_setting_key("auth:session:abc"));
+        assert!(is_reserved_setting_key("webhooks:subscriptions"));
+    }
+
+    #[test]
+    fn ordinary_keys_are_not() {
+        assert!(!is_reserved_setting_key("dashboard.theme"));
+        assert!(!is_reserved_setting_key("retentio"));
+        assert!(!is_reserved_setting_key("authors"));
+    }
+}

crates/taskito-java/src/queue/admin.rs

@@ -1,15 +1,15 @@
 //! Mutating administration entry points for `NativeQueue`.
 
 use jni::objects::{JClass, JString};
-use jni::sys::{jboolean, jlong, jstring, JNI_FALSE};
+use jni::sys::{jboolean, jlong, jobjectArray, jstring, JNI_FALSE};
 use jni::JNIEnv;
 use taskito_core::job::{now_millis, NewJob};
 use taskito_core::Storage;
 
 use super::borrow_queue;
 use crate::convert::{to_json, DeadJobView, ReplayEntryView};
 use crate::error::BindingError;
-use crate::ffi::{guard, new_string, read_string};
+use crate::ffi::{guard, new_string, new_string_array, read_string};
 
 /// `String listDead(long handle, long limit, long offset)` — dead-letter entries.
 #[no_mangle]
@@ -168,6 +168,25 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listPaused
     })
 }
 
+/// `String[] reservedSettingPrefixes()` — settings-key prefixes a dashboard's
+/// generic KV surface must hide. Sourced from the core so every shell hides the
+/// same keys.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_reservedSettingPrefixes<
+    'local,
+>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+) -> jobjectArray {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let prefixes: Vec<String> = taskito_core::RESERVED_SETTING_PREFIXES
+            .iter()
+            .map(|prefix| (*prefix).to_string())
+            .collect();
+        new_string_array(env, &prefixes)
+    })
+}
+
 /// `String getSetting(long handle, String key)` — value, or `null` if unset.
 #[no_mangle]
 pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getSetting<'local>(

crates/taskito-node/src/lib.rs

@@ -15,3 +15,13 @@ mod worker;
 
 pub use queue::JsQueue;
 pub use worker::JsWorker;
+
+/// Settings-key prefixes the dashboard's generic KV surface must hide. Sourced
+/// from the core so every shell hides the same keys.
+#[napi_derive::napi]
+pub fn reserved_setting_prefixes() -> Vec<String> {
+    taskito_core::RESERVED_SETTING_PREFIXES
+        .iter()
+        .map(|prefix| (*prefix).to_string())
+        .collect()
+}

crates/taskito-python/src/lib.rs

@@ -26,12 +26,23 @@ fn _init_rust_logging() {
     let _ = pyo3_log::try_init();
 }
 
+/// Settings-key prefixes the dashboard's generic KV surface must hide. Sourced
+/// from the core so every shell hides the same keys.
+#[pyfunction]
+fn reserved_setting_prefixes() -> Vec<String> {
+    taskito_core::RESERVED_SETTING_PREFIXES
+        .iter()
+        .map(|prefix| (*prefix).to_string())
+        .collect()
+}
+
 // `gil_used = true`: this extension relies on the GIL for its shared mutable
 // state (scheduler, workflow tracker). Until that state is audited for the
 // free-threaded build, advertise GIL dependence so 3.13t/3.14t fall back safely.
 #[pymodule(gil_used = true)]
 fn _taskito(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(_init_rust_logging, m)?)?;
+    m.add_function(wrap_pyfunction!(reserved_setting_prefixes, m)?)?;
     m.add_class::<PyQueue>()?;
     m.add_class::<PyJob>()?;
     m.add_class::<PyTaskConfig>()?;

docs/content/docs/java/api-reference/errors.mdx

@@ -33,8 +33,34 @@ try {
 | `ProxyException` | `TaskitoException` | A proxy reference couldn't be created or reconstructed — no handler, signature mismatch, allowlist violation. |
 | `WebhookException` | `TaskitoException` | A webhook couldn't be stored, loaded, signed, or encoded. |
 | `WorkflowException` | `TaskitoException` | Workflow definition, submission, await-timeout, or query error. |
+| `RetryableException` | `TaskitoException` | *You* throw it from a handler: this failure is transient, retry it. |
+| `NonRetryableException` | `TaskitoException` | *You* throw it from a handler: this failure is permanent, dead-letter it. |
 
 Handler exceptions are different: an exception thrown *inside* a
 `TaskFunction` isn't rethrown to you — it fails that attempt, and the core
 retries with the task's backoff until the retry budget is spent, then
 dead-letters the job. Inspect those via `jobErrors(id)` and `listDead`.
+
+## Signalling retry intent
+
+The last two rows above travel the other way — a handler throws them to
+classify its own failure, overriding the task's `retryOn` predicate:
+
+```java
+queue.worker().handle(CHARGE, (Order order) -> {
+    Response response = gateway.charge(order);
+    if (response.status() == 402) {
+        throw new NonRetryableException("card declined"); // dead-letters now
+    }
+    if (response.status() >= 500) {
+        throw new RetryableException("gateway " + response.status()); // spends the budget
+    }
+    return response.body();
+});
+```
+
+Both are honoured through the cause chain, so a signal wrapped by framework
+code still counts; when a chain carries both, the outermost wins. Any other
+exception is classified by the task's [`retryOn`
+predicate](/java/guides/reliability/retries#exception-filtering), and retries
+when there is none.

docs/content/docs/java/api-reference/task.mdx

@@ -29,7 +29,7 @@ enqueue options. The fluent option methods each return a new descriptor.
 | `timeoutMs(long)` / `timeout(Duration)` | Per-attempt timeout. |
 | `delayMs(long)` / `delay(Duration)` | Schedule after a delay. |
 | `retryPolicy(RetryPolicy)` | Backoff curve — registered with the worker on `start()`. |
-| `retryOn(Predicate<Throwable>)` | Classifies a thrown exception; `false` dead-letters it immediately. |
+| `retryOn(Predicate<Throwable>)` | Classifies a thrown exception; `false` dead-letters it immediately. A handler throwing [`RetryableException` / `NonRetryableException`](/java/api-reference/errors#signalling-retry-intent) overrides it. |
 | `codecs(String...)` | Named [payload codecs](/java/api-reference/serializers) applied to this task. |
 | `circuitBreaker(CircuitBreakerConfig)` | Trip the task after repeated failures; the worker registers the breaker on `start()`. |
 | `withOptions(EnqueueOptions)` | Replace the default options wholesale. |

docs/content/docs/shared/guides/reliability/error-handling.mdx

@@ -55,16 +55,33 @@ Task<String> FETCH = Task.of("fetch", String.class)
   </Tab>
 </CodeTabs>
 
-Returning normally is success; an uncaught throw is failure. There's no
-in-task "stop retrying" signal from inside a failing attempt — set retries
-to `0` for a fire-once task.
+Returning normally is success; an uncaught throw is failure.
+
+<SdkOnly sdk="python">
+
+There's no in-task "stop retrying" signal from inside a failing attempt — set
+retries to `0` for a fire-once task.
+
+</SdkOnly>
+
+<SdkOnly sdk="node">
+
+There's no in-task "stop retrying" signal from inside a failing attempt — set
+retries to `0` for a fire-once task.
+
+</SdkOnly>
 
 <SdkOnly sdk="java">
 
 `TaskFunction` declares `throws Exception`, so checked exceptions propagate
 without wrapping. `maxRetries` defaults to `0` — never retry — so a fresh
 task is fire-once until you opt into retries.
 
+A failing attempt can also decide for itself: throwing `NonRetryableException`
+dead-letters the job immediately, `RetryableException` insists on a retry. See
+[Typed signals from the
+handler](/java/guides/reliability/retries#typed-signals-from-the-handler).
+
 </SdkOnly>
 
 ## Timeouts

docs/content/docs/shared/guides/reliability/retries.mdx

@@ -269,6 +269,35 @@ It sees every exception raised while running the task, not only the handler's:
 too, and a whitelist like the first example above dead-letters those as well. A
 timeout is detected outside the handler and always retries.
 
+### Typed signals from the handler
+
+A predicate classifies a task's failures by type up front. When only the
+handler knows — the same `IOException` is transient on a 503 and permanent on
+a 422 — throw the intent instead:
+
+```java
+queue.worker().handle(CHARGE, (Order order) -> {
+    Response response = gateway.charge(order);
+    if (response.status() == 402) {
+        throw new NonRetryableException("card declined");  // dead-letters now
+    }
+    if (response.status() >= 500) {
+        throw new RetryableException("gateway " + response.status());  // spends the budget
+    }
+    return response.body();
+});
+```
+
+| Exception | Effect |
+|---|---|
+| `NonRetryableException` | Dead-letters the job at once, whatever budget is left. |
+| `RetryableException` | Retries on the task's backoff curve until the budget is spent. |
+
+Both live in `org.byteveda.taskito.errors` and both beat `retryOn` — the throw
+site is more specific than the task-wide predicate, so it wins. They are
+honoured through the cause chain too, so a signal wrapped by framework code
+still counts; if a chain carries both, the outermost wins.
+
 </SdkOnly>
 
 ## Per-job overrides
@@ -338,9 +367,10 @@ retry-budget check.
 
 <SdkOnly sdk="java">
 
-The "Exception passes filter?" step only applies filtering when `retryOn` is
-set on the task — otherwise every exception passes straight through to the
-retry-budget check.
+The "Exception passes filter?" step is answered by `RetryableException` /
+`NonRetryableException` when the handler threw one; otherwise it applies
+`retryOn`, and with no predicate every exception passes straight through to
+the retry-budget check.
 
 </SdkOnly>
 

sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java

@@ -8,24 +8,25 @@
 import org.byteveda.taskito.dashboard.support.Json;
 
 /**
- * Generic settings KV API. Keys under the reserved prefixes ({@code auth:},
- * {@code webhooks:}, {@code retention:}) are treated as absent everywhere — never listed, read,
- * written, or deleted through this surface — so auth and webhook state cannot be
- * exposed or clobbered. Keys are capped at 256 chars, values at 64 KiB.
+ * Generic settings KV API. Keys under the core's reserved prefixes ({@code auth:},
+ * {@code webhooks:}, {@code retention:}, …) are treated as absent everywhere — never
+ * listed, read, written, or deleted through this surface — so auth state, webhooks, and
+ * published runtime documents cannot be exposed or clobbered. Keys are capped at 256
+ * chars, values at 64 KiB.
  */
 public final class SettingsHandlers {
     static final int MAX_KEY_LENGTH = 256;
     static final int MAX_VALUE_LENGTH = 64 * 1024;
-    // Hide auth state, the webhook store (persisted under the "taskito.webhooks"
-    // key), and the retention windows the cleaner publishes — a report of what
-    // the worker does, not a knob.
-    private static final List<String> PROTECTED_PREFIXES =
-            List.of("auth:", "webhooks:", "taskito.webhooks", "retention:");
 
     private final SettingsAccess settings;
+    // Auth state, the webhook store, and the retention windows the cleaner
+    // publishes. The store hands over the core's list, so every shell hides the
+    // same keys and this class never touches the native library itself.
+    private final List<String> protectedPrefixes;
 
     public SettingsHandlers(SettingsAccess settings) {
         this.settings = settings;
+        this.protectedPrefixes = settings.reservedPrefixes();
     }
 
     public Object list() {
@@ -70,11 +71,11 @@ private static Map<String, Object> entry(String key, String value) {
         return m;
     }
 
-    private static boolean isProtected(String key) {
-        return PROTECTED_PREFIXES.stream().anyMatch(key::startsWith);
+    private boolean isProtected(String key) {
+        return protectedPrefixes.stream().anyMatch(key::startsWith);
     }
 
-    private static void validateKey(String key) {
+    private void validateKey(String key) {
         if (key == null || key.isEmpty() || key.length() > MAX_KEY_LENGTH) {
             throw DashboardError.badRequest("invalid setting key");
         }