From 1c7d5f0fbb559f604c25b8c202dd5e44b032e00e Mon Sep 17 00:00:00 2001 From: kartikeya <67143288+kartikeya-27@users.noreply.github.com> Date: Thu, 23 Jul 2026 03:36:39 +0530 Subject: [PATCH 1/5] feat(java): signal retry intent by exception type A handler throwing RetryableException or NonRetryableException overrides the task retryOn predicate. --- .../taskito/errors/NonRetryableException.java | 21 +++ .../taskito/errors/RetryableException.java | 22 ++++ .../byteveda/taskito/errors/package-info.java | 5 + .../java/org/byteveda/taskito/task/Task.java | 4 + .../taskito/worker/RetryDecision.java | 57 ++++++++ .../taskito/worker/WorkerDispatchBridge.java | 19 +-- .../taskito/core/ExceptionHierarchyTest.java | 4 + .../taskito/worker/RetryDecisionTest.java | 69 ++++++++++ .../worker/TypedRetryExceptionTest.java | 122 ++++++++++++++++++ 9 files changed, 305 insertions(+), 18 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/NonRetryableException.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/RetryableException.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/worker/RetryDecision.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/worker/RetryDecisionTest.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/worker/TypedRetryExceptionTest.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/NonRetryableException.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/NonRetryableException.java new file mode 100644 index 00000000..1cdca36d --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/NonRetryableException.java @@ -0,0 +1,21 @@ +package org.byteveda.taskito.errors; + +import org.byteveda.taskito.TaskitoException; + +/** + * Thrown by a task handler to mark the failure as permanent: the job + * dead-letters at once, whatever retry budget is left. For failures no amount + * of retrying fixes — a malformed payload, a 4xx response, a rejected charge. + * + *

Overrides the task's {@code retryOn} predicate, so the throw site decides + * even when the task classifies its failures by type. + */ +public class NonRetryableException extends TaskitoException { + public NonRetryableException(String message) { + super(message); + } + + public NonRetryableException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/RetryableException.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/RetryableException.java new file mode 100644 index 00000000..e8347a8e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/RetryableException.java @@ -0,0 +1,22 @@ +package org.byteveda.taskito.errors; + +import org.byteveda.taskito.TaskitoException; + +/** + * Thrown by a task handler to mark the failure as transient: the job retries on + * the task's backoff curve until its budget is spent. Overrides the task's + * {@code retryOn} predicate, so a handler can retry one failure a whitelist + * would otherwise dead-letter. + * + *

Retrying is already the default — reach for this only to override a + * predicate, or to say so explicitly at the throw site. + */ +public class RetryableException extends TaskitoException { + public RetryableException(String message) { + super(message); + } + + public RetryableException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java index 24eccaa6..9de98ea0 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/package-info.java @@ -7,6 +7,11 @@ * {@link org.byteveda.taskito.errors.WorkflowException}, etc.) to react to one * category. Native (JNI) errors surface as the base {@code TaskitoException}. * + *

{@link org.byteveda.taskito.errors.RetryableException} and + * {@link org.byteveda.taskito.errors.NonRetryableException} run the other way: + * a task handler throws them to tell the worker whether the failure is worth + * retrying. + * *

Also home to {@link org.byteveda.taskito.errors.TaskErrors}, the codec for * the structured task-error JSON stored in job and dead-letter {@code error} * fields, and its decoded view {@link org.byteveda.taskito.errors.TaskError}. diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java index 74eaca90..16c9e7e4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java @@ -133,6 +133,10 @@ public Task retryPolicy(RetryPolicy retryPolicy) { * payload decoding and result serialization can fail too, so a whitelist * predicate dead-letters those as well. A timeout is detected outside the * handler and always consumes a retry. + * + *

A handler that throws {@link org.byteveda.taskito.errors.RetryableException} + * or {@link org.byteveda.taskito.errors.NonRetryableException} decides for itself + * — those bypass this predicate. */ public Task retryOn(Predicate retryOn) { return new Task<>( diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/RetryDecision.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/RetryDecision.java new file mode 100644 index 00000000..40034b09 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/RetryDecision.java @@ -0,0 +1,57 @@ +package org.byteveda.taskito.worker; + +import java.util.function.Predicate; +import org.byteveda.taskito.errors.NonRetryableException; +import org.byteveda.taskito.errors.RetryableException; +import org.byteveda.taskito.logging.TaskitoLogger; + +/** + * Classifies a failed attempt as retryable or permanent. A typed signal thrown + * by the handler ({@link RetryableException} / {@link NonRetryableException}) + * wins over the task's {@code retryOn} predicate; with neither, every failure + * retries. + */ +final class RetryDecision { + private static final TaskitoLogger LOG = TaskitoLogger.create("worker"); + /** Bounds the cause walk so a self-referential chain can't spin the worker thread. */ + private static final int MAX_CAUSE_DEPTH = 16; + + private RetryDecision() {} + + /** Whether {@code error} should be retried under {@code retryOn} ({@code null} = no predicate). */ + static boolean isRetryable(Predicate retryOn, Throwable error) { + Boolean signalled = signalledIntent(error); + if (signalled != null) { + return signalled; + } + if (retryOn == null) { + return true; + } + try { + return retryOn.test(error); + } catch (RuntimeException e) { + // A broken classifier must not silently turn transient failures into dead letters. + LOG.warn("retryOn predicate threw; retrying the failure", e); + return true; + } + } + + /** + * The handler's explicit retry intent, or {@code null} when it threw neither + * typed exception. Walks the cause chain so a signal wrapped by framework + * code still counts; the outermost signal wins. + */ + private static Boolean signalledIntent(Throwable error) { + Throwable cause = error; + for (int depth = 0; cause != null && depth < MAX_CAUSE_DEPTH; depth++) { + if (cause instanceof NonRetryableException) { + return false; + } + if (cause instanceof RetryableException) { + return true; + } + cause = cause.getCause(); + } + return null; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index a465bacd..b6e4549f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -118,7 +118,7 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { m.onError(context, t); } // Canonical structured error (middleware above saw the live Throwable). - bound.failJob(token, TaskErrors.encode(t), isRetryable(task, t)); + bound.failJob(token, TaskErrors.encode(t), RetryDecision.isRetryable(task.retryOn, t)); } finally { if (scope != null) { Resources.exit(scope); // unbind the thread + dispose task-scoped resources (LIFO) @@ -126,23 +126,6 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { } } - /** - * Ask a task's {@code retryOn} predicate whether {@code error} is worth retrying. - * No predicate means retry, and so does one that throws — a broken classifier must - * not silently turn transient failures into dead letters. - */ - private static boolean isRetryable(RegisteredTask task, Throwable error) { - if (task.retryOn == null) { - return true; - } - try { - return task.retryOn.test(error); - } catch (RuntimeException e) { - LOG.warn("retryOn predicate threw; retrying the failure", e); - return true; - } - } - /** Reverse a task's payload codecs (last applied, first undone). */ private byte[] decodePayload(byte[] payload, List codecNames) { byte[] bytes = payload; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/ExceptionHierarchyTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/ExceptionHierarchyTest.java index e8e2716f..dc96322c 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/core/ExceptionHierarchyTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/ExceptionHierarchyTest.java @@ -8,6 +8,8 @@ import org.byteveda.taskito.errors.ConfigurationException; import org.byteveda.taskito.errors.CryptoException; import org.byteveda.taskito.errors.LockException; +import org.byteveda.taskito.errors.NonRetryableException; +import org.byteveda.taskito.errors.RetryableException; import org.byteveda.taskito.errors.SerializationException; import org.byteveda.taskito.errors.WebhookException; import org.byteveda.taskito.errors.WorkflowException; @@ -25,6 +27,8 @@ void everySpecificExceptionExtendsTheBase() { assertTrue(TaskitoException.class.isAssignableFrom(LockException.class)); assertTrue(TaskitoException.class.isAssignableFrom(ConfigurationException.class)); assertTrue(TaskitoException.class.isAssignableFrom(WebhookException.class)); + assertTrue(TaskitoException.class.isAssignableFrom(RetryableException.class)); + assertTrue(TaskitoException.class.isAssignableFrom(NonRetryableException.class)); // CryptoException is a kind of SerializationException. assertTrue(SerializationException.class.isAssignableFrom(CryptoException.class)); } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryDecisionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryDecisionTest.java new file mode 100644 index 00000000..920ab148 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/worker/RetryDecisionTest.java @@ -0,0 +1,69 @@ +package org.byteveda.taskito.worker; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.function.Predicate; +import org.byteveda.taskito.errors.NonRetryableException; +import org.byteveda.taskito.errors.RetryableException; +import org.junit.jupiter.api.Test; + +/** Unit-level classification rules behind a failed attempt's retry decision. */ +class RetryDecisionTest { + + private static final Predicate REJECT_ALL = error -> false; + + @Test + void everyFailureRetriesWithoutAPredicate() { + assertTrue(RetryDecision.isRetryable(null, new IllegalStateException("boom"))); + } + + @Test + void nonRetryableExceptionOverridesAnAcceptingPredicate() { + assertFalse(RetryDecision.isRetryable(error -> true, new NonRetryableException("malformed input"))); + } + + @Test + void retryableExceptionOverridesARejectingPredicate() { + assertTrue(RetryDecision.isRetryable(REJECT_ALL, new RetryableException("connection reset"))); + } + + @Test + void aWrappedSignalStillCounts() { + Throwable wrapped = new IllegalStateException("handler failed", new NonRetryableException("bad request")); + assertFalse(RetryDecision.isRetryable(null, wrapped)); + } + + @Test + void theOutermostSignalWins() { + Throwable wrapped = new RetryableException("retry the batch", new NonRetryableException("bad row")); + assertTrue(RetryDecision.isRetryable(null, wrapped)); + } + + @Test + void aCyclicCauseChainTerminates() { + CyclicException error = new CyclicException(); + assertTrue(RetryDecision.isRetryable(null, error)); + } + + @Test + void anUnsignalledFailureFallsBackToThePredicate() { + assertFalse(RetryDecision.isRetryable(REJECT_ALL, new IllegalArgumentException("malformed input"))); + } + + @Test + void aThrowingPredicateRetries() { + Predicate broken = error -> { + throw new IllegalStateException("classifier bug"); + }; + assertTrue(RetryDecision.isRetryable(broken, new IllegalStateException("boom"))); + } + + /** Its own cause — the JDK forbids this via initCause, an override does not. */ + private static final class CyclicException extends RuntimeException { + @Override + public synchronized Throwable getCause() { + return this; + } + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/worker/TypedRetryExceptionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/worker/TypedRetryExceptionTest.java new file mode 100644 index 00000000..c350e22e --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/worker/TypedRetryExceptionTest.java @@ -0,0 +1,122 @@ +package org.byteveda.taskito.worker; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.errors.NonRetryableException; +import org.byteveda.taskito.errors.RetryableException; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.task.RetryPolicy; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * A handler signals its retry intent by exception type: {@link NonRetryableException} + * dead-letters at once, {@link RetryableException} spends the budget — both over the + * task's {@code retryOn} predicate. + */ +class TypedRetryExceptionTest { + + private static final RetryPolicy FAST = RetryPolicy.delays(Duration.ofMillis(10), Duration.ofMillis(10)); + + @Test + @Timeout(30) + void nonRetryableExceptionDeadLettersWithoutSpendingTheBudget(@TempDir Path dir) throws Exception { + Task charge = Task.of("charge", String.class).maxRetries(3).retryPolicy(FAST); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(charge, "go"); + + AtomicInteger attempts = new AtomicInteger(); + AtomicInteger retries = new AtomicInteger(); + CountDownLatch dead = new CountDownLatch(1); + + try (Worker worker = queue.worker() + .handle(charge, (String payload) -> { + attempts.incrementAndGet(); + throw new NonRetryableException("card declined"); + }) + .on(EventName.RETRY, event -> retries.incrementAndGet()) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(25, TimeUnit.SECONDS), "task should dead-letter"); + + assertEquals(1, attempts.get(), "a permanent failure must not be retried"); + assertEquals(0, retries.get(), "no RETRY outcome should be emitted"); + } + } + } + + @Test + @Timeout(30) + void retryableExceptionOverridesARejectingPredicate(@TempDir Path dir) throws Exception { + Task sync = Task.of("sync", String.class) + .maxRetries(2) + .retryPolicy(FAST) + .retryOn(error -> false); // would dead-letter everything + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(sync, "go"); + + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch dead = new CountDownLatch(1); + + try (Worker worker = queue.worker() + .handle(sync, (String payload) -> { + attempts.incrementAndGet(); + throw new RetryableException("upstream 503"); + }) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(25, TimeUnit.SECONDS), "task should exhaust its budget"); + + assertEquals(3, attempts.get(), "first run plus both retries"); + } + } + } + + @Test + @Timeout(30) + void aWrappedNonRetryableSignalStillDeadLetters(@TempDir Path dir) throws Exception { + Task parse = Task.of("parse", String.class) + .maxRetries(3) + .retryPolicy(FAST) + .retryOn(error -> true); // would retry everything + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(parse, "go"); + + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch dead = new CountDownLatch(1); + + try (Worker worker = queue.worker() + .handle(parse, (String payload) -> { + attempts.incrementAndGet(); + throw new IllegalStateException("row 3", new NonRetryableException("malformed input")); + }) + .on(EventName.DEAD, event -> dead.countDown()) + .start()) { + assertTrue(dead.await(25, TimeUnit.SECONDS), "task should dead-letter"); + + assertEquals(1, attempts.get(), "a wrapped permanent failure must not be retried"); + } + } + } +} From a8aebb146f1f300d80c072a8fb5d48f998c82d9c Mon Sep 17 00:00:00 2001 From: kartikeya <67143288+kartikeya-27@users.noreply.github.com> Date: Thu, 23 Jul 2026 03:36:45 +0530 Subject: [PATCH 2/5] docs: document typed retry signals for the Java SDK --- .../docs/java/api-reference/errors.mdx | 26 ++++++++++++++ docs/content/docs/java/api-reference/task.mdx | 2 +- .../guides/reliability/error-handling.mdx | 23 ++++++++++-- .../shared/guides/reliability/retries.mdx | 36 +++++++++++++++++-- 4 files changed, 80 insertions(+), 7 deletions(-) diff --git a/docs/content/docs/java/api-reference/errors.mdx b/docs/content/docs/java/api-reference/errors.mdx index e4ce7b90..fb761662 100644 --- a/docs/content/docs/java/api-reference/errors.mdx +++ b/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. diff --git a/docs/content/docs/java/api-reference/task.mdx b/docs/content/docs/java/api-reference/task.mdx index 6984c473..53e0375c 100644 --- a/docs/content/docs/java/api-reference/task.mdx +++ b/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)` | Classifies a thrown exception; `false` dead-letters it immediately. | +| `retryOn(Predicate)` | 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. | diff --git a/docs/content/docs/shared/guides/reliability/error-handling.mdx b/docs/content/docs/shared/guides/reliability/error-handling.mdx index ea1863c2..f56c0e1d 100644 --- a/docs/content/docs/shared/guides/reliability/error-handling.mdx +++ b/docs/content/docs/shared/guides/reliability/error-handling.mdx @@ -55,9 +55,21 @@ Task FETCH = Task.of("fetch", String.class) -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. + + + +There's no in-task "stop retrying" signal from inside a failing attempt — set +retries to `0` for a fire-once task. + + + + + +There's no in-task "stop retrying" signal from inside a failing attempt — set +retries to `0` for a fire-once task. + + @@ -65,6 +77,11 @@ to `0` for a fire-once task. 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). + ## Timeouts diff --git a/docs/content/docs/shared/guides/reliability/retries.mdx b/docs/content/docs/shared/guides/reliability/retries.mdx index f9c51f68..698687c7 100644 --- a/docs/content/docs/shared/guides/reliability/retries.mdx +++ b/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. + ## Per-job overrides @@ -338,9 +367,10 @@ retry-budget check. -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. From 2255e49a1fd69a0f64b7b93959063c130a259854 Mon Sep 17 00:00:00 2001 From: kartikeya <67143288+kartikeya-27@users.noreply.github.com> Date: Thu, 23 Jul 2026 03:36:51 +0530 Subject: [PATCH 3/5] fix(core): log an unparseable retention document --- crates/taskito-core/src/scheduler/retention.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/taskito-core/src/scheduler/retention.rs b/crates/taskito-core/src/scheduler/retention.rs index 1f3143ad..08777a12 100644 --- a/crates/taskito-core/src/scheduler/retention.rs +++ b/crates/taskito-core/src/scheduler/retention.rs @@ -149,7 +149,16 @@ pub fn read_effective_retention( 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 From 43c5843f973f452baffffa3f6a622c09cccbdea0 Mon Sep 17 00:00:00 2001 From: kartikeya <67143288+kartikeya-27@users.noreply.github.com> Date: Thu, 23 Jul 2026 03:37:07 +0530 Subject: [PATCH 4/5] feat(core): own the reserved settings prefixes Every shell now derives its hidden-key list from the core instead of hardcoding one, so a missed prefix cannot reopen the hole in one binding. --- crates/taskito-core/BINDING_CONTRACT.md | 18 +++++-- crates/taskito-core/src/lib.rs | 3 ++ crates/taskito-core/src/settings.rs | 50 +++++++++++++++++++ crates/taskito-java/src/queue/admin.rs | 23 ++++++++- crates/taskito-node/src/lib.rs | 10 ++++ crates/taskito-python/src/lib.rs | 11 ++++ .../dashboard/api/SettingsHandlers.java | 18 +++---- .../taskito/internal/NativeQueue.java | 3 ++ .../dashboard/api/SettingsHandlersTest.java | 5 ++ sdks/node/src/dashboard/handlers/settings.ts | 4 +- sdks/node/src/native.ts | 2 +- sdks/node/test/dashboard/opsSettings.test.ts | 5 ++ sdks/python/taskito/_taskito.pyi | 4 ++ .../taskito/dashboard/handlers/settings.py | 6 ++- .../dashboard/test_dashboard_settings.py | 15 ++++++ 15 files changed, 159 insertions(+), 18 deletions(-) create mode 100644 crates/taskito-core/src/settings.rs diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 633d9b52..1681717a 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/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:` (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` (opaque), `status`, `priority`, `retry_count`, `max_retries`, `timeout_ms`, `unique_key`, diff --git a/crates/taskito-core/src/lib.rs b/crates/taskito-core/src/lib.rs index beae8c0f..66126f7d 100644 --- a/crates/taskito-core/src/lib.rs +++ b/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; diff --git a/crates/taskito-core/src/settings.rs b/crates/taskito-core/src/settings.rs new file mode 100644 index 00000000..39010104 --- /dev/null +++ b/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")); + } +} diff --git a/crates/taskito-java/src/queue/admin.rs b/crates/taskito-java/src/queue/admin.rs index f4f8120b..ee4de2f8 100644 --- a/crates/taskito-java/src/queue/admin.rs +++ b/crates/taskito-java/src/queue/admin.rs @@ -1,7 +1,7 @@ //! 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; @@ -9,7 +9,7 @@ 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 = 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>( diff --git a/crates/taskito-node/src/lib.rs b/crates/taskito-node/src/lib.rs index 5fd001d5..fdf8a0a2 100644 --- a/crates/taskito-node/src/lib.rs +++ b/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 { + taskito_core::RESERVED_SETTING_PREFIXES + .iter() + .map(|prefix| (*prefix).to_string()) + .collect() +} diff --git a/crates/taskito-python/src/lib.rs b/crates/taskito-python/src/lib.rs index 7bec9cb1..90d4f3d2 100644 --- a/crates/taskito-python/src/lib.rs +++ b/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 { + 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::()?; m.add_class::()?; m.add_class::()?; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java index ec09c801..46482beb 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java @@ -6,21 +6,21 @@ import org.byteveda.taskito.dashboard.store.SettingsAccess; import org.byteveda.taskito.dashboard.support.DashboardError; import org.byteveda.taskito.dashboard.support.Json; +import org.byteveda.taskito.internal.NativeQueue; /** - * 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 PROTECTED_PREFIXES = - List.of("auth:", "webhooks:", "taskito.webhooks", "retention:"); + // Auth state, the webhook store, and the retention windows the cleaner + // publishes. The list comes from the core so every shell hides the same keys. + private static final List PROTECTED_PREFIXES = List.of(NativeQueue.reservedSettingPrefixes()); private final SettingsAccess settings; 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 7344cfd7..6e9dcb83 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 @@ -99,6 +99,9 @@ private NativeQueue() {} public static native String listSettings(long handle); + /** Settings-key prefixes the dashboard's generic KV surface must hide. */ + public static native String[] reservedSettingPrefixes(); + /** The published retention windows as JSON, or {@code null} if unreported. */ public static native String effectiveRetention(long handle); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java index c1bf4ec3..0777b56b 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/api/SettingsHandlersTest.java @@ -37,14 +37,19 @@ void nonStringValueIsJsonEncoded() { void protectedKeysAreInvisible() { settings.setSetting("auth:users", "{}"); settings.setSetting("webhooks:subscriptions", "[]"); + settings.setSetting("retention:effective:default", "{}"); handlers.put("visible", Map.of("value", "1")); Map listed = asMap(handlers.list()); assertTrue(listed.containsKey("visible")); assertTrue(!listed.containsKey("auth:users")); assertTrue(!listed.containsKey("webhooks:subscriptions")); + // The published retention document is a report, not an editable row. + assertTrue(!listed.containsKey("retention:effective:default")); assertNull(handlers.get("auth:users")); + assertNull(handlers.get("retention:effective:default")); assertThrows(DashboardError.class, () -> handlers.put("auth:x", Map.of("value", "1"))); assertThrows(DashboardError.class, () -> handlers.delete("webhooks:subscriptions")); + assertThrows(DashboardError.class, () -> handlers.put("retention:effective:default", Map.of("value", "{}"))); } @Test diff --git a/sdks/node/src/dashboard/handlers/settings.ts b/sdks/node/src/dashboard/handlers/settings.ts index f7673a6f..3b4217d5 100644 --- a/sdks/node/src/dashboard/handlers/settings.ts +++ b/sdks/node/src/dashboard/handlers/settings.ts @@ -3,6 +3,7 @@ // integrations, external links). import type { Queue } from "../../index"; +import { reservedSettingPrefixes } from "../../native"; import { BadRequestError, NotFoundError } from "../errors"; const MAX_KEY_LENGTH = 256; @@ -14,7 +15,8 @@ const MAX_VALUE_LENGTH = 64 * 1024; // 64 KiB — enough for any dashboard confi // keys hold subscription rows with plaintext HMAC secrets plus delivery // logs. `retention:` holds the windows the cleaner publishes — a report of // what the worker does, not a knob. Protected keys are treated as absent. -const PROTECTED_PREFIXES = ["auth:", "webhooks:", "webhook:", "retention:"]; +// The list comes from the core so every shell hides the same keys. +const PROTECTED_PREFIXES = reservedSettingPrefixes(); const isProtected = (key: string): boolean => PROTECTED_PREFIXES.some((p) => key.startsWith(p)); diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index 8d90406f..342dcb75 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -9,7 +9,7 @@ const require = createRequire(import.meta.url); const bindingPath = fileURLToPath(new URL("../native/index.js", import.meta.url)); const binding = require(bindingPath) as typeof import("../native/index"); -export const { JsQueue, JsWorker } = binding; +export const { JsQueue, JsWorker, reservedSettingPrefixes } = binding; /** Instance types of the native classes, for typing fields and parameters. */ export type NativeQueue = InstanceType; diff --git a/sdks/node/test/dashboard/opsSettings.test.ts b/sdks/node/test/dashboard/opsSettings.test.ts index 3ef60fec..55773885 100644 --- a/sdks/node/test/dashboard/opsSettings.test.ts +++ b/sdks/node/test/dashboard/opsSettings.test.ts @@ -86,6 +86,11 @@ describe("settings api", () => { expect( (await fetch(`${base}/api/settings/auth:users`, { method: "DELETE", headers })).status, ).toBe(404); + + // The published retention document is a report, not an editable row. + expect((await put("/api/settings/retention:effective:default", { value: "{}" })).status).toBe( + 400, + ); }); it("deletes settings and 404s unknown keys", async () => { diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index 39b6e6b4..0735b615 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -573,3 +573,7 @@ class PyResultSender: def _init_rust_logging() -> None: """Activate the Rust → Python `logging` bridge (idempotent).""" ... + +def reserved_setting_prefixes() -> list[str]: + """Settings-key prefixes the dashboard's generic KV surface must hide.""" + ... diff --git a/sdks/python/taskito/dashboard/handlers/settings.py b/sdks/python/taskito/dashboard/handlers/settings.py index 5c91d411..75a9ff33 100644 --- a/sdks/python/taskito/dashboard/handlers/settings.py +++ b/sdks/python/taskito/dashboard/handlers/settings.py @@ -12,6 +12,7 @@ import json from typing import TYPE_CHECKING, Any +from taskito._taskito import reserved_setting_prefixes from taskito.dashboard.errors import _BadRequest, _NotFound if TYPE_CHECKING: @@ -30,8 +31,9 @@ # these must stay out of the generic settings endpoints too. ``retention:`` # holds the windows the cleaner publishes — a report of what the worker does, # not a knob, and writable only by the cleaner itself. The settings handlers -# treat every protected key as if absent. -_PROTECTED_PREFIXES = ("auth:", "webhooks:", "retention:") +# treat every protected key as if absent. The list comes from the core so every +# shell hides the same keys. +_PROTECTED_PREFIXES = tuple(reserved_setting_prefixes()) def _is_protected(key: str) -> bool: diff --git a/sdks/python/tests/dashboard/test_dashboard_settings.py b/sdks/python/tests/dashboard/test_dashboard_settings.py index cebb2473..fee4a946 100644 --- a/sdks/python/tests/dashboard/test_dashboard_settings.py +++ b/sdks/python/tests/dashboard/test_dashboard_settings.py @@ -157,6 +157,21 @@ def test_get_settings_hides_webhook_keys(dashboard_server: tuple[AuthedClient, Q assert exc_info.value.code == 404 +def test_settings_api_hides_published_retention( + dashboard_server: tuple[AuthedClient, Queue], +) -> None: + """``retention:`` is a report of what the cleaner applies, not an editable row.""" + client, queue = dashboard_server + queue.set_setting("retention:effective:default", json.dumps({"enabled": True})) + + snapshot = client.get("/api/settings") + assert not any(k.startswith("retention:") for k in snapshot) + + with pytest.raises(urllib.error.HTTPError) as exc_info: + client.put("/api/settings/retention:effective:default", {"value": "{}"}) + assert exc_info.value.code == 400 + + def test_put_setting_with_missing_value_field_returns_400( dashboard_server: tuple[AuthedClient, Queue], ) -> None: From 7ad8ef17a84b7937218abb55be9b0ba9a180527b Mon Sep 17 00:00:00 2001 From: kartikeya <67143288+kartikeya-27@users.noreply.github.com> Date: Thu, 23 Jul 2026 03:50:31 +0530 Subject: [PATCH 5/5] fix(java): resolve reserved prefixes through the settings store Keeps dashboard class loading off the native library; the in-memory test store mirrors the core list, locked by a test. --- .../dashboard/api/SettingsHandlers.java | 15 ++++++++------- .../dashboard/store/SettingsAccess.java | 12 ++++++++++++ .../taskito/dashboard/InMemorySettings.java | 10 ++++++++++ .../ReservedSettingPrefixesTest.java | 19 +++++++++++++++++++ 4 files changed, 49 insertions(+), 7 deletions(-) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/dashboard/ReservedSettingPrefixesTest.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java index 46482beb..5cb66b85 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/SettingsHandlers.java @@ -6,7 +6,6 @@ import org.byteveda.taskito.dashboard.store.SettingsAccess; import org.byteveda.taskito.dashboard.support.DashboardError; import org.byteveda.taskito.dashboard.support.Json; -import org.byteveda.taskito.internal.NativeQueue; /** * Generic settings KV API. Keys under the core's reserved prefixes ({@code auth:}, @@ -18,14 +17,16 @@ public final class SettingsHandlers { static final int MAX_KEY_LENGTH = 256; static final int MAX_VALUE_LENGTH = 64 * 1024; - // Auth state, the webhook store, and the retention windows the cleaner - // publishes. The list comes from the core so every shell hides the same keys. - private static final List PROTECTED_PREFIXES = List.of(NativeQueue.reservedSettingPrefixes()); 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 protectedPrefixes; public SettingsHandlers(SettingsAccess settings) { this.settings = settings; + this.protectedPrefixes = settings.reservedPrefixes(); } public Object list() { @@ -70,11 +71,11 @@ private static Map 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"); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/SettingsAccess.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/SettingsAccess.java index 4e5a4466..0ec827f7 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/SettingsAccess.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/store/SettingsAccess.java @@ -1,8 +1,10 @@ package org.byteveda.taskito.dashboard.store; +import java.util.List; import java.util.Map; import java.util.Optional; import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.internal.NativeQueue; /** * Narrow view of the {@code dashboard_settings} KV store — the single @@ -23,6 +25,16 @@ public interface SettingsAccess { /** All settings; callers filter by key prefix. */ Map listSettings(); + /** + * Key prefixes the generic settings API must treat as absent (auth state, + * webhooks, runtime-published documents). Comes from the core so every shell + * hides the same keys; resolved on call, not on class load, so an in-memory + * store can answer without the native library. + */ + default List reservedPrefixes() { + return List.of(NativeQueue.reservedSettingPrefixes()); + } + static SettingsAccess of(Taskito queue) { return new SettingsAccess() { @Override diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java index e1c415e7..34344764 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/InMemorySettings.java @@ -1,6 +1,7 @@ package org.byteveda.taskito.dashboard; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import org.byteveda.taskito.dashboard.store.SettingsAccess; @@ -28,4 +29,13 @@ public boolean deleteSetting(String key) { public Map listSettings() { return new HashMap<>(map); } + + /** + * Mirrors the core list so unit tests need no native library. + * {@code ReservedSettingPrefixesTest} fails if the two drift. + */ + @Override + public List reservedPrefixes() { + return List.of("auth:", "retention:", "taskito.webhooks", "webhook:", "webhooks:"); + } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/ReservedSettingPrefixesTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/ReservedSettingPrefixesTest.java new file mode 100644 index 00000000..a8048a77 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/ReservedSettingPrefixesTest.java @@ -0,0 +1,19 @@ +package org.byteveda.taskito.dashboard; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import org.byteveda.taskito.internal.NativeQueue; +import org.junit.jupiter.api.Test; + +/** The in-memory test store mirrors the core's reserved prefixes; this locks the two together. */ +class ReservedSettingPrefixesTest { + + @Test + void inMemoryStoreMirrorsTheCoreList() { + List core = List.of(NativeQueue.reservedSettingPrefixes()); + assertTrue(core.contains("retention:"), "core must reserve the published retention document"); + assertEquals(core, new InMemorySettings().reservedPrefixes()); + } +}