From c3a62835f9579dea206f573cac7663d62544bdf4 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 13:12:35 +0530 Subject: [PATCH 1/3] feat(java): proxy ref expiry + purpose binding --- .../org/byteveda/taskito/proxies/Proxies.java | 55 ++++++++++++++++--- .../byteveda/taskito/proxies/ProxyRef.java | 16 +++++- 2 files changed, 61 insertions(+), 10 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/proxies/Proxies.java b/sdks/java/src/main/java/org/byteveda/taskito/proxies/Proxies.java index 98a1f228..915840d6 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/proxies/Proxies.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/proxies/Proxies.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.SerializationFeature; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; +import java.time.Duration; import java.util.Base64; import java.util.LinkedHashMap; import java.util.Map; @@ -44,33 +45,63 @@ public Proxies register(ProxyHandler handler) { return this; } - /** Deconstruct {@code value} into a signed ref; throws if no handler accepts it. */ - @SuppressWarnings("unchecked") + /** Deconstruct {@code value} into a signed ref with no expiry or purpose. */ public ProxyRef deconstruct(Object value) { + return deconstruct(value, null, null); + } + + /** Deconstruct {@code value} into a ref that expires after {@code ttl}. */ + public ProxyRef deconstruct(Object value, Duration ttl) { + return deconstruct(value, ttl, null); + } + + /** + * Deconstruct {@code value} into a signed ref bound to {@code ttl} (nullable) + * and {@code purpose} (nullable); throws if no handler accepts it. + */ + @SuppressWarnings("unchecked") + public ProxyRef deconstruct(Object value, Duration ttl, String purpose) { if (value == null) { throw new ProxyException("cannot deconstruct null"); } + Long expiresAtMs = ttl == null ? null : System.currentTimeMillis() + ttl.toMillis(); for (ProxyHandler handler : handlers.values()) { if (handler.handles(value)) { Map reference = ((ProxyHandler) handler).deconstruct(value); - return new ProxyRef(handler.id(), reference, sign(handler.id(), reference)); + String signature = sign(handler.id(), reference, expiresAtMs, purpose); + return new ProxyRef(handler.id(), reference, signature, expiresAtMs, purpose); } } throw new ProxyException("no proxy handler for " + value.getClass().getName()); } - /** Verify a ref's signature and reconstruct the resource. */ - @SuppressWarnings("unchecked") + /** Verify a ref's signature and expiry, then reconstruct the resource. */ public Object reconstruct(ProxyRef ref) { + return reconstruct(ref, null); + } + + /** + * Verify a ref's signature, expiry, and (when {@code expectedPurpose} is + * non-null) its bound purpose, then reconstruct the resource. + */ + @SuppressWarnings("unchecked") + public Object reconstruct(ProxyRef ref, String expectedPurpose) { ProxyHandler handler = (ProxyHandler) handlers.get(ref.handler()); if (handler == null) { throw new ProxyException("unknown proxy handler '" + ref.handler() + "'"); } - byte[] expected = sign(ref.handler(), ref.reference()).getBytes(StandardCharsets.UTF_8); + byte[] expected = sign(ref.handler(), ref.reference(), ref.expiresAtMs(), ref.purpose()) + .getBytes(StandardCharsets.UTF_8); byte[] actual = (ref.signature() == null ? "" : ref.signature()).getBytes(StandardCharsets.UTF_8); if (!MessageDigest.isEqual(expected, actual)) { throw new ProxyException("proxy signature mismatch for handler '" + ref.handler() + "'"); } + if (ref.expiresAtMs() != null && System.currentTimeMillis() > ref.expiresAtMs()) { + throw new ProxyException("proxy ref expired for handler '" + ref.handler() + "'"); + } + if (expectedPurpose != null && !expectedPurpose.equals(ref.purpose())) { + throw new ProxyException("proxy purpose mismatch for handler '" + ref.handler() + "'"); + } return handler.reconstruct(ref.reference()); } @@ -80,13 +111,23 @@ public T resolve(ProxyRef ref) { return (T) reconstruct(ref); } - private String sign(String handlerId, Map reference) { + /** {@link #reconstruct(ProxyRef, String)} cast to the caller's type. */ + @SuppressWarnings("unchecked") + public T resolve(ProxyRef ref, String expectedPurpose) { + return (T) reconstruct(ref, expectedPurpose); + } + + private String sign(String handlerId, Map reference, Long expiresAtMs, String purpose) { try { Mac mac = Mac.getInstance(ALGORITHM); mac.init(new SecretKeySpec(key, ALGORITHM)); mac.update(handlerId.getBytes(StandardCharsets.UTF_8)); mac.update((byte) '\n'); mac.update(canonical.writeValueAsBytes(reference)); + mac.update((byte) '\n'); + mac.update(String.valueOf(expiresAtMs).getBytes(StandardCharsets.UTF_8)); + mac.update((byte) '\n'); + mac.update((purpose == null ? "" : purpose).getBytes(StandardCharsets.UTF_8)); return Base64.getEncoder().encodeToString(mac.doFinal()); } catch (Exception e) { throw new ProxyException("failed to sign proxy ref", e); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyRef.java b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyRef.java index 332e43fd..1eeb00d4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyRef.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyRef.java @@ -6,11 +6,21 @@ /** * A serializable, signed reference to a non-serializable resource. Carry it in a * job payload; the worker reconstructs the resource with - * {@link Proxies#reconstruct}. + * {@link Proxies#reconstruct}. The optional {@code expiresAtMs} and + * {@code purpose} are folded into the signature, so neither can be tampered with. * * @param handler the {@link ProxyHandler#id()} that produced (and reconstructs) it * @param reference the handler's serializable reference data - * @param signature an HMAC over {@code handler + reference}, verified on reconstruct + * @param signature an HMAC over {@code handler + reference + expiresAtMs + purpose} + * @param expiresAtMs wall-clock expiry (epoch ms), or null for no expiry + * @param purpose an optional binding label the worker can require on reconstruct */ @JsonIgnoreProperties(ignoreUnknown = true) -public record ProxyRef(String handler, Map reference, String signature) {} +public record ProxyRef( + String handler, Map reference, String signature, Long expiresAtMs, String purpose) { + + /** A ref with neither expiry nor purpose binding. */ + public ProxyRef(String handler, Map reference, String signature) { + this(handler, reference, signature, null, null); + } +} From 1f3185ddeeee7646053c4e42ac07c923177617ec Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 13:12:36 +0530 Subject: [PATCH 2/3] test(java): cover proxy expiry + purpose --- .../java/org/byteveda/taskito/ProxyTest.java | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.java index 067ce6c4..1d12e548 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.java @@ -7,6 +7,7 @@ import java.io.File; import java.nio.file.Files; import java.nio.file.Path; +import java.time.Duration; import java.util.List; import java.util.Map; import org.byteveda.taskito.errors.ProxyException; @@ -93,4 +94,45 @@ void rejectsDuplicateHandlerId() { Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); assertThrows(ProxyException.class, () -> proxies.register(new FileProxyHandler())); } + + @Test + void roundTripsWithinTtl(@TempDir Path dir) { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + File file = dir.resolve("data.txt").toFile(); + + ProxyRef ref = proxies.deconstruct(file, Duration.ofHours(1)); + File back = proxies.resolve(ref); + assertEquals(file.getAbsolutePath(), back.getAbsolutePath()); + } + + @Test + void rejectsExpiredRef(@TempDir Path dir) throws Exception { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + ProxyRef ref = proxies.deconstruct(dir.resolve("a").toFile(), Duration.ofMillis(10)); + + Thread.sleep(40); + assertThrows(ProxyException.class, () -> proxies.reconstruct(ref)); + } + + @Test + void rejectsTamperedExpiry(@TempDir Path dir) { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + ProxyRef ref = proxies.deconstruct(dir.resolve("a").toFile(), Duration.ofMillis(10)); + // Extend the expiry while keeping the original signature → signature must fail. + ProxyRef extended = + new ProxyRef(ref.handler(), ref.reference(), ref.signature(), ref.expiresAtMs() + 3_600_000L, null); + + assertThrows(ProxyException.class, () -> proxies.reconstruct(extended)); + } + + @Test + void enforcesPurposeWhenRequested(@TempDir Path dir) { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + File file = dir.resolve("a").toFile(); + ProxyRef ref = proxies.deconstruct(file, null, "emails"); + + assertEquals(file.getAbsolutePath(), ((File) proxies.reconstruct(ref, "emails")).getAbsolutePath()); + assertEquals(file.getAbsolutePath(), ((File) proxies.reconstruct(ref)).getAbsolutePath()); // unchecked + assertThrows(ProxyException.class, () -> proxies.reconstruct(ref, "billing")); + } } From ab171e8473f532b33b0c3959c46ddc11edeeac2f Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 13:13:11 +0530 Subject: [PATCH 3/3] docs(java): README gates + proxies reposition --- sdks/java/README.md | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/sdks/java/README.md b/sdks/java/README.md index c59e787f..fbfdd01c 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -236,8 +236,9 @@ Taskito secure = Taskito.builder() ### Resources (worker dependency injection) -Register a resource once; resolve it inside handlers. Scopes: `WORKER` (built -once, shared) and `TASK` (built + disposed per invocation). +The primary way to use a non-serializable dependency (pool, client, logger) in a +handler: register it once and resolve it inside the worker. Scopes: `WORKER` +(built once, shared) and `TASK` (built + disposed per invocation). ```java taskito.resource("db", ctx -> openPool()); // WORKER @@ -245,11 +246,28 @@ taskito.resource("tx", ResourceScope.TASK, ctx -> ctx.use("db").begin(), T taskito.worker().handle(save, p -> Resources.use("tx").save(p)).start(); ``` -### Enqueue predicates (gates) +When a handler takes `@Resource` parameters, the `@TaskHandler` processor wires +them from the runtime for you — no `Resources.use` call needed. + +### Cross-process references (proxies) + +Secondary to resources: when a *specific* resource identity must travel inside a +payload to another process, carry a signed `ProxyRef` and rebuild it on the +worker. Bind an optional TTL and purpose — both are folded into the HMAC. + +```java +Proxies proxies = new Proxies(hmacKey).register(new FileProxyHandler()); +ProxyRef ref = proxies.deconstruct(file, Duration.ofMinutes(5), "report"); // producer +File same = proxies.resolve(ref, "report"); // worker (expiry + purpose checked) +``` + +### Enqueue gates ```java -taskito.predicate("send_email", ctx -> isBusinessHours()); // rejects → PredicateRejectedException -// Predicates.allOf / anyOf / not compose them. +taskito.predicate("send_email", ctx -> payloadValid(ctx)); // boolean: false → PredicateRejectedException +taskito.gate("send_email", Recipes.businessHours(zone)); // allow / skip / defer / reject +Optional id = taskito.tryEnqueue(emailTask, msg); // empty when a gate skips +// Recipes: businessHours / timeWindow / dayOfWeek / payloadMatches / featureFlag. ``` ### Producer batching @@ -302,7 +320,9 @@ org.byteveda.taskito │ TaskMetric, WorkerInfo, TaskLog, JobFilter (read-only views) ├── worker/ Worker runtime (concurrency, autoscale) ├── resources/ worker DI — ResourceRuntime, Resources.use(name), scopes -├── predicates/ enqueue gates — Predicate, Predicates (allOf/anyOf/not) +├── proxies/ signed cross-process references — Proxies, ProxyRef, handlers +├── interception/ enqueue-time arg interception — Interceptor, Interception +├── predicates/ enqueue gates — Predicate, EnqueueGate, EnqueueDecision, Recipes ├── batch/ Batcher — producer-side batching ├── autoscale/ Autoscaler, AutoscaleOptions ├── scaler/ Scaler — KEDA HTTP endpoint