From aceb6556c222701804324859f2cfeaad64505121 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:04:57 +0530 Subject: [PATCH 1/6] feat(java): worker resources (dependency injection) Register a resource once, resolve it in handlers via Resources.use(name). Two scopes: WORKER (built once, shared, disposed at worker close) and TASK (built + disposed per invocation). Injected at the single dispatch point via a ThreadLocal ScopeContext seam; zero overhead when unused. --- .../org/byteveda/taskito/DefaultTaskito.java | 40 +++- .../java/org/byteveda/taskito/Taskito.java | 19 ++ .../taskito/errors/ResourceException.java | 18 ++ .../taskito/internal/ScopeContext.java | 31 +++ .../taskito/resources/ResourceContext.java | 14 ++ .../taskito/resources/ResourceDefinition.java | 34 +++ .../taskito/resources/ResourceRuntime.java | 194 ++++++++++++++++++ .../taskito/resources/ResourceScope.java | 9 + .../taskito/resources/ResourceStat.java | 10 + .../byteveda/taskito/resources/Resources.java | 40 ++++ .../byteveda/taskito/resources/TaskScope.java | 48 +++++ .../taskito/resources/package-info.java | 11 + .../org/byteveda/taskito/worker/Worker.java | 22 +- .../taskito/worker/WorkerDispatchBridge.java | 18 +- 14 files changed, 500 insertions(+), 8 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/errors/ResourceException.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/internal/ScopeContext.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceStat.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/Resources.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/TaskScope.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java 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 b459e73f..05819762 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -13,6 +13,8 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; +import java.util.function.Function; import org.byteveda.taskito.core.CoreFacade; import org.byteveda.taskito.errors.SerializationException; import org.byteveda.taskito.errors.WorkflowException; @@ -29,6 +31,11 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.resources.ResourceContext; +import org.byteveda.taskito.resources.ResourceDefinition; +import org.byteveda.taskito.resources.ResourceRuntime; +import org.byteveda.taskito.resources.ResourceScope; +import org.byteveda.taskito.resources.ResourceStat; import org.byteveda.taskito.scheduling.PeriodicTask; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; @@ -55,6 +62,7 @@ final class DefaultTaskito implements Taskito { private final CoreFacade facade; private final Serializer serializer; private final List middleware = new CopyOnWriteArrayList<>(); + private final ResourceRuntime resources = new ResourceRuntime(); DefaultTaskito(QueueBackend backend, Serializer serializer) { this.backend = backend; @@ -73,6 +81,36 @@ public Taskito use(Middleware middleware) { return this; } + // ── Resources ──────────────────────────────────────────────────── + + @Override + public Taskito resource(String name, Function factory) { + return resource(name, ResourceScope.WORKER, factory); + } + + @Override + public Taskito resource(String name, ResourceScope scope, Function factory) { + resources.register(name, new ResourceDefinition(factory::apply, scope, null)); + return this; + } + + @Override + public Taskito resource( + String name, ResourceScope scope, Function factory, Consumer dispose) { + resources.register(name, new ResourceDefinition(factory::apply, scope, value -> dispose.accept(cast(value)))); + return this; + } + + @Override + public Map resourceMetrics() { + return resources.metrics(); + } + + @SuppressWarnings("unchecked") + private static T cast(Object value) { + return (T) value; + } + // ── Producer ──────────────────────────────────────────────────── @Override @@ -576,7 +614,7 @@ private static String encodeGate(GateConfig gate) { @Override public Worker.Builder worker() { - return Worker.builder(backend, serializer, middleware); + return Worker.builder(backend, serializer, middleware, resources); } @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 1b474f68..1ea5a7d9 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -11,6 +11,8 @@ import java.util.Locale; import java.util.Map; import java.util.Optional; +import java.util.function.Consumer; +import java.util.function.Function; import org.byteveda.taskito.errors.ConfigurationException; import org.byteveda.taskito.internal.JniQueueBackend; import org.byteveda.taskito.locks.Lock; @@ -25,6 +27,9 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.resources.ResourceContext; +import org.byteveda.taskito.resources.ResourceScope; +import org.byteveda.taskito.resources.ResourceStat; import org.byteveda.taskito.scheduling.PeriodicTask; import org.byteveda.taskito.serialization.JsonSerializer; import org.byteveda.taskito.serialization.Serializer; @@ -55,6 +60,20 @@ static Builder builder() { /** Register cross-cutting middleware (enqueue + worker hooks); returns {@code this}. */ Taskito use(Middleware middleware); + // ── Resources (worker-side dependency injection) ───────────────── + + /** Register a worker-scoped resource resolved in handlers via {@code Resources.use(name)}. */ + Taskito resource(String name, Function factory); + + /** Register a resource with an explicit {@link ResourceScope}. */ + Taskito resource(String name, ResourceScope scope, Function factory); + + /** Register a resource with a scope and a disposer run when the scope ends. */ + Taskito resource(String name, ResourceScope scope, Function factory, Consumer dispose); + + /** Per-resource counters (created / disposed / active). */ + Map resourceMetrics(); + // ── Producer ──────────────────────────────────────────────────── /** Enqueue a typed payload using the task's default options; returns the job id. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/errors/ResourceException.java b/sdks/java/src/main/java/org/byteveda/taskito/errors/ResourceException.java new file mode 100644 index 00000000..e6fdf2dc --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/errors/ResourceException.java @@ -0,0 +1,18 @@ +package org.byteveda.taskito.errors; + +import org.byteveda.taskito.TaskitoException; + +/** + * A worker resource could not be built, resolved, or disposed — e.g. a factory + * threw, an unknown resource name was requested, or {@code Resources.use} was + * called outside a task handler. + */ +public class ResourceException extends TaskitoException { + public ResourceException(String message) { + super(message); + } + + public ResourceException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/ScopeContext.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/ScopeContext.java new file mode 100644 index 00000000..b134a12f --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/ScopeContext.java @@ -0,0 +1,31 @@ +package org.byteveda.taskito.internal; + +/** + * A single seam for per-task context propagation. Backed by a {@link ThreadLocal} + * today (the SDK targets Java 17); if the floor later rises to a JDK with a + * stable {@code ScopedValue}, only this class changes — callers are unaffected. + * + *

Each task runs on a pooled worker thread, so callers must {@link #set} on + * entry and {@link #clear} in a {@code finally} on exit to avoid leaking context + * into the next task scheduled on that thread. + * + * @param the value carried for the duration of a task + */ +public final class ScopeContext { + private final ThreadLocal holder = new ThreadLocal<>(); + + /** The value bound on the current thread, or {@code null} if none. */ + public T get() { + return holder.get(); + } + + /** Bind {@code value} on the current thread. */ + public void set(T value) { + holder.set(value); + } + + /** Unbind any value on the current thread. */ + public void clear() { + holder.remove(); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java new file mode 100644 index 00000000..70633516 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java @@ -0,0 +1,14 @@ +package org.byteveda.taskito.resources; + +/** + * Handed to a resource factory so it can depend on other resources. A + * {@code WORKER} factory may only {@link #use} other worker resources; a + * {@code TASK} factory may use either. + */ +public interface ResourceContext { + /** The scope the factory is building for. */ + ResourceScope scope(); + + /** Resolve another resource by name (building it if needed). */ + T use(String name); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java new file mode 100644 index 00000000..b32f5dbb --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java @@ -0,0 +1,34 @@ +package org.byteveda.taskito.resources; + +import java.util.function.Consumer; +import java.util.function.Function; + +/** + * How to build (and optionally dispose) a resource. + * + * @param factory builds the resource, possibly using others via the context + * @param scope the resource's lifetime (defaults to {@link ResourceScope#WORKER}) + * @param dispose optional cleanup run when the scope ends ({@code null} for none) + */ +public record ResourceDefinition( + Function factory, ResourceScope scope, Consumer dispose) { + + public ResourceDefinition { + if (factory == null) { + throw new IllegalArgumentException("resource factory must not be null"); + } + if (scope == null) { + scope = ResourceScope.WORKER; + } + } + + /** A worker-scoped resource with no disposer. */ + public static ResourceDefinition worker(Function factory) { + return new ResourceDefinition(factory, ResourceScope.WORKER, null); + } + + /** A task-scoped resource with no disposer. */ + public static ResourceDefinition task(Function factory) { + return new ResourceDefinition(factory, ResourceScope.TASK, null); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java new file mode 100644 index 00000000..a878e560 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java @@ -0,0 +1,194 @@ +package org.byteveda.taskito.resources; + +import java.lang.System.Logger; +import java.lang.System.Logger.Level; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; +import org.byteveda.taskito.errors.ResourceException; + +/** + * Registry and lifecycle for worker resources. Holds the definitions, the + * single shared cache of worker-scoped instances, and per-resource counters. + * Worker-scoped resources are built at most once (under a per-name lock, so a + * concurrent first-use does not double-build) and disposed LIFO when the last + * worker lease is released. + */ +public final class ResourceRuntime { + private static final Logger LOG = System.getLogger(ResourceRuntime.class.getName()); + /** Cache sentinel for a factory that legitimately returned {@code null}. */ + private static final Object NULL = new Object(); + + private final ConcurrentMap definitions = new ConcurrentHashMap<>(); + private final ConcurrentMap workerCache = new ConcurrentHashMap<>(); + private final ConcurrentMap workerLocks = new ConcurrentHashMap<>(); + private final ConcurrentMap counters = new ConcurrentHashMap<>(); + private final Deque workerTeardown = new ArrayDeque<>(); + private int leases; // guarded by this + + /** Context handed to a worker factory: it may only use other worker resources. */ + private final ResourceContext workerContext = new ResourceContext() { + @Override + public ResourceScope scope() { + return ResourceScope.WORKER; + } + + @Override + public T use(String name) { + return cast(resolveWorker(name)); + } + }; + + /** Register a resource under {@code name}, replacing any prior definition. */ + public void register(String name, ResourceDefinition definition) { + definitions.put(name, definition); + counters.computeIfAbsent(name, key -> new Counter()); + } + + /** Whether any resource is registered (lets the worker skip all wiring when unused). */ + public boolean isEmpty() { + return definitions.isEmpty(); + } + + /** A fresh per-invocation scope for one task. */ + public TaskScope createTaskScope() { + return new TaskScope(this); + } + + /** Lease the worker resources (paired with {@link #teardownWorker}). */ + public synchronized void acquireWorker() { + leases++; + } + + /** Release a worker lease; when the last one drops, dispose worker resources LIFO. */ + public synchronized void teardownWorker() { + if (leases > 0) { + leases--; + } + if (leases == 0) { + disposeWorker(); + } + } + + /** Per-resource counters snapshot. */ + public Map metrics() { + Map out = new LinkedHashMap<>(); + counters.forEach((name, counter) -> { + long created = counter.created.get(); + long disposed = counter.disposed.get(); + out.put(name, new ResourceStat(created, disposed, created - disposed)); + }); + return out; + } + + /** Resolve a worker-scoped resource, building it once under a per-name lock. */ + Object resolveWorker(String name) { + ResourceDefinition definition = definition(name); + if (definition.scope() != ResourceScope.WORKER) { + throw new ResourceException("resource '" + name + "' is task-scoped; a worker resource cannot use it"); + } + Object cached = workerCache.get(name); + if (cached != null) { + return unwrap(cached); + } + ReentrantLock lock = workerLocks.computeIfAbsent(name, key -> new ReentrantLock()); + lock.lock(); + try { + cached = workerCache.get(name); + if (cached != null) { + return unwrap(cached); + } + Object value = build(name, definition, workerContext); + workerCache.put(name, value == null ? NULL : value); + counter(name).created.incrementAndGet(); + if (definition.dispose() != null) { + synchronized (workerTeardown) { + workerTeardown.push(() -> dispose(name, value, definition.dispose())); + } + } + return value; + } finally { + lock.unlock(); + } + } + + /** Resolve for a task: worker-scoped hits the shared cache; task-scoped builds once per invocation. */ + Object resolveForTask(TaskScope scope, String name) { + ResourceDefinition definition = definition(name); + if (definition.scope() == ResourceScope.WORKER) { + return resolveWorker(name); + } + Map cache = scope.cache(); + Object cached = cache.get(name); + if (cached != null) { + return unwrap(cached); + } + Object value = build(name, definition, scope); + cache.put(name, value == null ? NULL : value); + counter(name).created.incrementAndGet(); + if (definition.dispose() != null) { + scope.pushTeardown(() -> dispose(name, value, definition.dispose())); + } + return value; + } + + private Object build(String name, ResourceDefinition definition, ResourceContext context) { + try { + return definition.factory().apply(context); + } catch (RuntimeException e) { + throw new ResourceException("failed to build resource '" + name + "'", e); + } + } + + private void disposeWorker() { + synchronized (workerTeardown) { + while (!workerTeardown.isEmpty()) { + workerTeardown.pop().run(); + } + } + workerCache.clear(); + } + + private void dispose(String name, Object value, Consumer disposer) { + try { + disposer.accept(unwrap(value)); + counter(name).disposed.incrementAndGet(); + } catch (RuntimeException e) { + // A disposer must never fail teardown — record and continue. + LOG.log(Level.WARNING, "disposing resource '" + name + "' failed", e); + } + } + + private ResourceDefinition definition(String name) { + ResourceDefinition definition = definitions.get(name); + if (definition == null) { + throw new ResourceException("unknown resource '" + name + "'"); + } + return definition; + } + + private Counter counter(String name) { + return counters.computeIfAbsent(name, key -> new Counter()); + } + + private static Object unwrap(Object value) { + return value == NULL ? null : value; + } + + @SuppressWarnings("unchecked") + private static T cast(Object value) { + return (T) value; + } + + /** Mutable per-resource counters. */ + private static final class Counter { + final AtomicLong created = new AtomicLong(); + final AtomicLong disposed = new AtomicLong(); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java new file mode 100644 index 00000000..973f1cf0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java @@ -0,0 +1,9 @@ +package org.byteveda.taskito.resources; + +/** Lifetime of a worker resource. */ +public enum ResourceScope { + /** Built once, lazily, and shared across every task on the worker. */ + WORKER, + /** Built lazily per task invocation and disposed when the task ends. */ + TASK +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceStat.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceStat.java new file mode 100644 index 00000000..932be726 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceStat.java @@ -0,0 +1,10 @@ +package org.byteveda.taskito.resources; + +/** + * Per-resource counters. + * + * @param created how many instances have been built + * @param disposed how many have been disposed + * @param active currently-live instances ({@code created - disposed}) + */ +public record ResourceStat(long created, long disposed, long active) {} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/Resources.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/Resources.java new file mode 100644 index 00000000..74012823 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/Resources.java @@ -0,0 +1,40 @@ +package org.byteveda.taskito.resources; + +import org.byteveda.taskito.errors.ResourceException; +import org.byteveda.taskito.internal.ScopeContext; + +/** + * Resolve worker resources from inside a task handler: {@code Resources.use("db")}. + * Valid only while a task runs on this worker; the worker binds the task's scope + * around the handler call. + */ +public final class Resources { + private static final ScopeContext ACTIVE = new ScopeContext<>(); + + private Resources() {} + + /** Resolve the named resource for the current task. */ + public static T use(String name) { + TaskScope scope = ACTIVE.get(); + if (scope == null) { + throw new ResourceException("Resources.use(\"" + name + "\") called outside a task handler"); + } + return scope.use(name); + } + + /** Bind {@code scope} for the current thread (called by the worker before the handler). */ + public static void enter(TaskScope scope) { + ACTIVE.set(scope); + } + + /** + * Unbind the current thread's scope and dispose its task-scoped resources + * (called by the worker after the handler, in a {@code finally}). + */ + public static void exit(TaskScope scope) { + ACTIVE.clear(); + if (scope != null) { + scope.teardown(); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/TaskScope.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/TaskScope.java new file mode 100644 index 00000000..0125b801 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/TaskScope.java @@ -0,0 +1,48 @@ +package org.byteveda.taskito.resources; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashMap; +import java.util.Map; + +/** + * Per-invocation resource scope. Caches task-scoped resources for one task and + * disposes them (LIFO) when the task ends. Also serves as the + * {@link ResourceContext} for a task-scoped factory, which may use worker or task + * resources. Not thread-safe — confined to the single thread running the task. + */ +public final class TaskScope implements ResourceContext { + private final ResourceRuntime runtime; + private final Map cache = new HashMap<>(); + private final Deque teardown = new ArrayDeque<>(); + + TaskScope(ResourceRuntime runtime) { + this.runtime = runtime; + } + + @Override + public ResourceScope scope() { + return ResourceScope.TASK; + } + + @Override + @SuppressWarnings("unchecked") + public T use(String name) { + return (T) runtime.resolveForTask(this, name); + } + + Map cache() { + return cache; + } + + void pushTeardown(Runnable disposer) { + teardown.push(disposer); + } + + /** Dispose this invocation's task-scoped resources in reverse build order. */ + void teardown() { + while (!teardown.isEmpty()) { + teardown.pop().run(); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java new file mode 100644 index 00000000..ae4a6bca --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java @@ -0,0 +1,11 @@ +/** + * Worker-side dependency injection: register a resource once, resolve it inside + * task handlers. + * + *

Two scopes (mirroring the Node SDK): {@link org.byteveda.taskito.resources.ResourceScope#WORKER} + * resources are built lazily once and shared across every task on the worker; + * {@link org.byteveda.taskito.resources.ResourceScope#TASK} resources are built + * lazily per task invocation and disposed (LIFO) when it ends. Handlers resolve + * them with {@link org.byteveda.taskito.resources.Resources#use(String)}. + */ +package org.byteveda.taskito.resources; 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 9f2fd59e..73777223 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 @@ -19,6 +19,7 @@ import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.resources.ResourceRuntime; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerControl; @@ -39,17 +40,21 @@ public final class Worker implements AutoCloseable { private final WorkerControl control; private final ExecutorService executor; private final WorkflowTracker tracker; + private final ResourceRuntime resources; private final CountDownLatch shutdown = new CountDownLatch(1); private boolean closed; - private Worker(WorkerControl control, ExecutorService executor, WorkflowTracker tracker) { + private Worker( + WorkerControl control, ExecutorService executor, WorkflowTracker tracker, ResourceRuntime resources) { this.control = control; this.executor = executor; this.tracker = tracker; + this.resources = resources; } - public static Builder builder(QueueBackend backend, Serializer serializer, List middleware) { - return new Builder(backend, serializer, middleware); + public static Builder builder( + QueueBackend backend, Serializer serializer, List middleware, ResourceRuntime resources) { + return new Builder(backend, serializer, middleware, resources); } /** Stop dispatching; in-flight jobs continue to drain. */ @@ -109,6 +114,7 @@ public synchronized void close() { if (tracker != null) { tracker.close(); // stop the gate-timeout scheduler } + resources.teardownWorker(); // dispose worker-scoped resources when the last lease drops shutdown.countDown(); } @@ -119,6 +125,7 @@ public static final class Builder { private final QueueBackend backend; private final Serializer serializer; private final List middleware; + private final ResourceRuntime resources; private final Map handlers = new HashMap<>(); private final Map taskPolicies = new HashMap<>(); private final Map>> listeners = new EnumMap<>(EventName.class); @@ -128,10 +135,11 @@ public static final class Builder { private Integer batchSize; private WorkflowTracker tracker; - Builder(QueueBackend backend, Serializer serializer, List middleware) { + Builder(QueueBackend backend, Serializer serializer, List middleware, ResourceRuntime resources) { this.backend = backend; this.serializer = serializer; this.middleware = middleware; + this.resources = resources; } public Builder handle(String taskName, Class payloadType, TaskFunction handler) { @@ -233,10 +241,12 @@ public Worker start() { Emitter emitter = new Emitter(); listeners.forEach((name, bound) -> bound.forEach(listener -> emitter.on(name, listener))); WorkerDispatchBridge bridge = - new WorkerDispatchBridge(backend, handlers, serializer, executor, emitter, middleware); + new WorkerDispatchBridge(backend, handlers, serializer, executor, emitter, middleware, resources); WorkerControl control = backend.startWorker(bridge, encodeOptions()); bridge.bind(control); - return new Worker(control, executor, tracker); + // Lease worker resources only after the native worker started cleanly. + resources.acquireWorker(); + return new Worker(control, executor, tracker, resources); } private String encodeOptions() { 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 da5ec6b2..f040019c 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 @@ -14,6 +14,9 @@ import org.byteveda.taskito.middleware.JobInfo; import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.middleware.TaskContext; +import org.byteveda.taskito.resources.ResourceRuntime; +import org.byteveda.taskito.resources.Resources; +import org.byteveda.taskito.resources.TaskScope; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerBridge; @@ -35,6 +38,7 @@ final class WorkerDispatchBridge implements WorkerBridge { private final ExecutorService executor; private final Emitter emitter; private final List middleware; + private final ResourceRuntime resources; // Resolved once startWorker returns; job tasks await it before completing. private final CompletableFuture control = new CompletableFuture<>(); @@ -44,13 +48,15 @@ final class WorkerDispatchBridge implements WorkerBridge { Serializer serializer, ExecutorService executor, Emitter emitter, - List middleware) { + List middleware, + ResourceRuntime resources) { this.backend = backend; this.handlers = handlers; this.serializer = serializer; this.executor = executor; this.emitter = emitter; this.middleware = middleware; + this.resources = resources; } void bind(WorkerControl control) { @@ -71,6 +77,12 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { } JobInfo job = new JobInfo(jobId, taskName, () -> loadMetadata(jobId)); TaskContext context = new TaskContext(jobId, taskName, job); + // Bind a per-task resource scope around the handler; skip all wiring when + // no resources are registered (zero overhead for the common case). + TaskScope scope = resources.isEmpty() ? null : resources.createTaskScope(); + if (scope != null) { + Resources.enter(scope); + } try { for (Middleware m : middleware) { m.before(context); @@ -86,6 +98,10 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { m.onError(context, t); } bound.failJob(token, describe(t)); + } finally { + if (scope != null) { + Resources.exit(scope); // unbind the thread + dispose task-scoped resources (LIFO) + } } } From ccf9c32b5d95347b97f759f78fd9bdbdd9c405a2 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:05:04 +0530 Subject: [PATCH 2/6] test(java): cover worker resources Worker-scoped sharing, task-scoped per-invocation build + dispose via metrics, and use-outside-task rejection. --- .../org/byteveda/taskito/ResourceTest.java | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java new file mode 100644 index 00000000..590bcb22 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -0,0 +1,63 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.errors.ResourceException; +import org.byteveda.taskito.resources.ResourceScope; +import org.byteveda.taskito.resources.ResourceStat; +import org.byteveda.taskito.resources.Resources; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class ResourceTest { + + private static final Task TASK = Task.of("res.task", Integer.class); + + @Test + @Timeout(30) + void workerResourceSharedTaskResourcePerInvocation(@TempDir Path dir) throws Exception { + int jobs = 3; + AtomicInteger taskDisposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("r.db").toString()).open()) { + queue.resource("shared", ctx -> new Object()); // WORKER scope (default): built once + queue.resource("perTask", ResourceScope.TASK, ctx -> new Object(), value -> taskDisposed.incrementAndGet()); + + CountDownLatch ran = new CountDownLatch(jobs); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + Resources.use("shared"); + Resources.use("perTask"); + ran.countDown(); + return p; + }) + .start()) { + for (int i = 0; i < jobs; i++) { + queue.enqueue(TASK, i); + } + assertTrue(ran.await(20, TimeUnit.SECONDS), "handlers did not all run"); + } // worker close drains in-flight tasks, so every teardown has run + + Map metrics = queue.resourceMetrics(); + assertEquals(1, metrics.get("shared").created(), "worker resource built once"); + assertEquals(jobs, metrics.get("perTask").created(), "task resource built per invocation"); + assertEquals(jobs, metrics.get("perTask").disposed(), "task resource disposed per invocation"); + assertEquals(jobs, taskDisposed.get()); + } + } + + @Test + void useOutsideTaskThrows() { + assertThrows(ResourceException.class, () -> Resources.use("anything")); + } +} From 52b308abd31326e16a6c86db44aeac786c6396a7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:40:51 +0530 Subject: [PATCH 3/6] fix(java): keep the 3-arg Worker.builder overload --- .../src/main/java/org/byteveda/taskito/worker/Worker.java | 4 ++++ 1 file changed, 4 insertions(+) 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 73777223..57255504 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 @@ -52,6 +52,10 @@ private Worker( this.resources = resources; } + public static Builder builder(QueueBackend backend, Serializer serializer, List middleware) { + return new Builder(backend, serializer, middleware, new ResourceRuntime()); + } + public static Builder builder( QueueBackend backend, Serializer serializer, List middleware, ResourceRuntime resources) { return new Builder(backend, serializer, middleware, resources); From 8daed756f1ee10db8a0d685314fc07d6bd2847c7 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:42:35 +0530 Subject: [PATCH 4/6] fix(java): give each worker its own resource runtime --- .../org/byteveda/taskito/DefaultTaskito.java | 3 +- .../taskito/resources/ResourceRuntime.java | 38 +++++++++++++++---- .../org/byteveda/taskito/ResourceTest.java | 29 ++++++++++++++ 3 files changed, 62 insertions(+), 8 deletions(-) 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 05819762..01e78aed 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -614,7 +614,8 @@ private static String encodeGate(GateConfig gate) { @Override public Worker.Builder worker() { - return Worker.builder(backend, serializer, middleware, resources); + // Each worker gets its own runtime (own WORKER-scoped cache) over shared definitions. + return Worker.builder(backend, serializer, middleware, resources.forWorker()); } @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java index a878e560..e9c3a967 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java @@ -14,21 +14,24 @@ import org.byteveda.taskito.errors.ResourceException; /** - * Registry and lifecycle for worker resources. Holds the definitions, the - * single shared cache of worker-scoped instances, and per-resource counters. - * Worker-scoped resources are built at most once (under a per-name lock, so a - * concurrent first-use does not double-build) and disposed LIFO when the last - * worker lease is released. + * Registry and lifecycle for worker resources. The client-level instance holds + * the definitions and per-resource counters; each worker gets its own live + * runtime via {@link #forWorker()}, sharing those definitions/counters but with + * its own cache of worker-scoped instances — so a {@code WORKER}-scoped resource + * is one instance per worker, not one per client. Worker-scoped + * resources are built at most once per worker (under a per-name lock, so a + * concurrent first-use does not double-build) and disposed LIFO when that + * worker's last lease is released. */ public final class ResourceRuntime { private static final Logger LOG = System.getLogger(ResourceRuntime.class.getName()); /** Cache sentinel for a factory that legitimately returned {@code null}. */ private static final Object NULL = new Object(); - private final ConcurrentMap definitions = new ConcurrentHashMap<>(); + private final ConcurrentMap definitions; + private final ConcurrentMap counters; private final ConcurrentMap workerCache = new ConcurrentHashMap<>(); private final ConcurrentMap workerLocks = new ConcurrentHashMap<>(); - private final ConcurrentMap counters = new ConcurrentHashMap<>(); private final Deque workerTeardown = new ArrayDeque<>(); private int leases; // guarded by this @@ -45,6 +48,27 @@ public T use(String name) { } }; + /** A client-level runtime: holds definitions + counters, hands each worker a child via {@link #forWorker()}. */ + public ResourceRuntime() { + this.definitions = new ConcurrentHashMap<>(); + this.counters = new ConcurrentHashMap<>(); + } + + private ResourceRuntime( + ConcurrentMap definitions, ConcurrentMap counters) { + this.definitions = definitions; + this.counters = counters; + } + + /** + * A per-worker runtime sharing this runtime's definitions and counters but with + * its own worker-scoped cache, teardown stack, and lease count — so each worker + * builds and disposes its own {@code WORKER}-scoped instances. + */ + public ResourceRuntime forWorker() { + return new ResourceRuntime(definitions, counters); + } + /** Register a resource under {@code name}, replacing any prior definition. */ public void register(String name, ResourceDefinition definition) { definitions.put(name, definition); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java index 590bcb22..0ce7cbfd 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -7,6 +7,7 @@ import java.nio.file.Path; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.byteveda.taskito.errors.ResourceException; @@ -14,6 +15,7 @@ import org.byteveda.taskito.resources.ResourceStat; import org.byteveda.taskito.resources.Resources; import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.task.TaskFunction; import org.byteveda.taskito.worker.Worker; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -60,4 +62,31 @@ void workerResourceSharedTaskResourcePerInvocation(@TempDir Path dir) throws Exc void useOutsideTaskThrows() { assertThrows(ResourceException.class, () -> Resources.use("anything")); } + + @Test + @Timeout(30) + void workerScopedResourceIsBuiltPerWorker(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rpw.db").toString()).open()) { + queue.resource("perWorker", ctx -> new Object()); + // Force exactly one job per worker so both live workers resolve the resource. + CyclicBarrier bothBusy = new CyclicBarrier(2); + CountDownLatch ran = new CountDownLatch(2); + TaskFunction handler = p -> { + Resources.use("perWorker"); + bothBusy.await(20, TimeUnit.SECONDS); + ran.countDown(); + return p; + }; + try (Worker w1 = queue.worker().concurrency(1).handle(TASK, handler).start(); + Worker w2 = + queue.worker().concurrency(1).handle(TASK, handler).start()) { + queue.enqueue(TASK, 1); + queue.enqueue(TASK, 2); + assertTrue(ran.await(25, TimeUnit.SECONDS)); + } + // Shared-runtime would build once (created==1); per-worker builds one each. + assertEquals(2, queue.resourceMetrics().get("perWorker").created()); + } + } } From 882c718a98efc0a0722ed6621f3a419aba8c17ba Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:43:27 +0530 Subject: [PATCH 5/6] fix(java): reject duplicate resource registration --- .../org/byteveda/taskito/resources/ResourceRuntime.java | 6 ++++-- .../src/test/java/org/byteveda/taskito/ResourceTest.java | 9 +++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java index e9c3a967..1b1281dc 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java @@ -69,9 +69,11 @@ public ResourceRuntime forWorker() { return new ResourceRuntime(definitions, counters); } - /** Register a resource under {@code name}, replacing any prior definition. */ + /** Register a resource under {@code name}. A name may be registered only once. */ public void register(String name, ResourceDefinition definition) { - definitions.put(name, definition); + if (definitions.putIfAbsent(name, definition) != null) { + throw new ResourceException("resource '" + name + "' is already registered"); + } counters.computeIfAbsent(name, key -> new Counter()); } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java index 0ce7cbfd..c2af1c7d 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -63,6 +63,15 @@ void useOutsideTaskThrows() { assertThrows(ResourceException.class, () -> Resources.use("anything")); } + @Test + void duplicateRegistrationThrows(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rdup.db").toString()).open()) { + queue.resource("db", ctx -> new Object()); + assertThrows(ResourceException.class, () -> queue.resource("db", ctx -> new Object())); + } + } + @Test @Timeout(30) void workerScopedResourceIsBuiltPerWorker(@TempDir Path dir) throws Exception { From 33e0dea04ea6a8c83f4192b73a7fb39b68324b0c Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 21:45:13 +0530 Subject: [PATCH 6/6] fix(java): fail fast on circular resource dependencies --- .../taskito/resources/ResourceRuntime.java | 13 ++++++++ .../org/byteveda/taskito/ResourceTest.java | 31 +++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java index 1b1281dc..19e4dfdd 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java @@ -5,7 +5,9 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; @@ -33,6 +35,9 @@ public final class ResourceRuntime { private final ConcurrentMap workerCache = new ConcurrentHashMap<>(); private final ConcurrentMap workerLocks = new ConcurrentHashMap<>(); private final Deque workerTeardown = new ArrayDeque<>(); + /** Names being resolved on the current thread, so a dependency cycle fails fast instead of recursing. */ + private final ThreadLocal> resolving = ThreadLocal.withInitial(LinkedHashSet::new); + private int leases; // guarded by this /** Context handed to a worker factory: it may only use other worker resources. */ @@ -165,10 +170,18 @@ Object resolveForTask(TaskScope scope, String name) { } private Object build(String name, ResourceDefinition definition, ResourceContext context) { + Set chain = resolving.get(); + if (!chain.add(name)) { + throw new ResourceException("circular resource dependency at '" + name + "' (chain: " + chain + ")"); + } try { return definition.factory().apply(context); + } catch (ResourceException e) { + throw e; // a nested unknown/cycle/build failure already carries its message } catch (RuntimeException e) { throw new ResourceException("failed to build resource '" + name + "'", e); + } finally { + chain.remove(name); } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java index c2af1c7d..1bb08715 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -10,6 +10,7 @@ import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.byteveda.taskito.errors.ResourceException; import org.byteveda.taskito.resources.ResourceScope; import org.byteveda.taskito.resources.ResourceStat; @@ -72,6 +73,36 @@ void duplicateRegistrationThrows(@TempDir Path dir) throws Exception { } } + @Test + @Timeout(30) + void circularResourceDependencyIsRejected(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rcyc.db").toString()).open()) { + queue.resource("self", ctx -> { + ctx.use("self"); // self-reference — must fail fast, not StackOverflow + return new Object(); + }); + AtomicReference> thrown = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + try { + Resources.use("self"); + } catch (RuntimeException e) { + thrown.set(e.getClass()); + } finally { + ran.countDown(); + } + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } + assertEquals(ResourceException.class, thrown.get()); + } + } + @Test @Timeout(30) void workerScopedResourceIsBuiltPerWorker(@TempDir Path dir) throws Exception {