feat(java): thread and request resource scopes

sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java

@@ -1,9 +1,11 @@
 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.
+ * Handed to a resource factory so it can depend on other resources. A factory
+ * may only depend on same-or-longer-lived resources: a {@code WORKER} factory
+ * may {@link #use} only worker resources, a {@code THREAD} factory may use
+ * worker or thread resources, and {@code TASK}/{@code REQUEST} factories may
+ * use any scope.
  */
 public interface ResourceContext {
     /** The scope the factory is building for. */

sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java

@@ -6,6 +6,7 @@
 import java.util.Deque;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -34,6 +35,13 @@ public final class ResourceRuntime {
     private final ConcurrentMap<String, Counter> counters;
     private final ConcurrentMap<String, Object> workerCache = new ConcurrentHashMap<>();
     private final ConcurrentMap<String, ReentrantLock> workerLocks = new ConcurrentHashMap<>();
+    /**
+     * Per-name, per-thread instances for {@code THREAD}-scoped resources. Only the
+     * owning thread reads or writes its own entry (so no per-name lock is needed);
+     * the concurrent maps exist for cross-thread visibility at worker teardown.
+     */
+    private final ConcurrentMap<String, ConcurrentMap<Thread, Object>> threadCache = new ConcurrentHashMap<>();
+
     private final Deque<Runnable> workerTeardown = new ArrayDeque<>();
     /** Names being resolved on the current thread, so a dependency cycle fails fast instead of recursing. */
     private final ThreadLocal<Set<String>> resolving = ThreadLocal.withInitial(LinkedHashSet::new);
@@ -53,6 +61,19 @@ public <T> T use(String name) {
         }
     };
 
+    /** Context handed to a thread factory: it may use worker or thread resources. */
+    private final ResourceContext threadContext = new ResourceContext() {
+        @Override
+        public ResourceScope scope() {
+            return ResourceScope.THREAD;
+        }
+
+        @Override
+        public <T> T use(String name) {
+            return cast(resolveThread(name));
+        }
+    };
+
     /** A client-level runtime: holds definitions + counters, hands each worker a child via {@link #forWorker()}. */
     public ResourceRuntime() {
         this.definitions = new ConcurrentHashMap<>();
@@ -122,7 +143,9 @@ public Map<String, ResourceStat> metrics() {
     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");
+            throw new ResourceException("resource '" + name + "' is "
+                    + scopeWord(definition.scope())
+                    + "-scoped; a worker resource may only use worker resources");
         }
         Object cached = workerCache.get(name);
         if (cached != null) {
@@ -149,11 +172,23 @@ Object resolveWorker(String name) {
         }
     }
 
-    /** Resolve for a task: worker-scoped hits the shared cache; task-scoped builds once per invocation. */
+    /**
+     * Resolve for a task, dispatching by the resource's scope: worker hits the
+     * shared cache, thread hits the current thread's cache, request builds fresh
+     * on every use, task builds once per invocation.
+     */
     Object resolveForTask(TaskScope scope, String name) {
         ResourceDefinition definition = definition(name);
-        if (definition.scope() == ResourceScope.WORKER) {
-            return resolveWorker(name);
+        switch (definition.scope()) {
+            case WORKER:
+                return resolveWorker(name);
+            case THREAD:
+                return resolveThread(name);
+            case REQUEST:
+                return buildRequest(scope, name, definition);
+            case TASK:
+            default:
+                break;
         }
         Map<String, Object> cache = scope.cache();
         Object cached = cache.get(name);
@@ -169,6 +204,63 @@ Object resolveForTask(TaskScope scope, String name) {
         return value;
     }
 
+    /**
+     * Resolve a thread-scoped resource for the current worker thread, building it
+     * lazily. Only the owning thread touches its own map entry, so a plain
+     * get-build-put is race-free; the shared {@code workerTeardown} deque keeps
+     * disposal globally LIFO across worker and thread instances.
+     */
+    Object resolveThread(String name) {
+        ResourceDefinition definition = definition(name);
+        if (definition.scope() == ResourceScope.WORKER) {
+            return resolveWorker(name);
+        }
+        if (definition.scope() != ResourceScope.THREAD) {
+            throw new ResourceException("resource '" + name + "' is "
+                    + scopeWord(definition.scope())
+                    + "-scoped; a thread resource may only use worker or thread resources");
+        }
+        ConcurrentMap<Thread, Object> perThread = threadCache.computeIfAbsent(name, key -> new ConcurrentHashMap<>());
+        Object cached = perThread.get(Thread.currentThread());
+        if (cached != null) {
+            return unwrap(cached);
+        }
+        Object value = build(name, definition, threadContext);
+        perThread.put(Thread.currentThread(), value == null ? NULL : value);
+        counter(name).created.incrementAndGet();
+        if (definition.dispose() != null) {
+            synchronized (workerTeardown) {
+                workerTeardown.push(() -> dispose(name, value, definition.dispose()));
+            }
+        }
+        return value;
+    }
+
+    /** Build a request-scoped resource: fresh on every use, disposed with the task (LIFO). */
+    private Object buildRequest(TaskScope scope, String name, ResourceDefinition definition) {
+        Object value = build(name, definition, requestContext(scope));
+        counter(name).created.incrementAndGet();
+        if (definition.dispose() != null) {
+            scope.pushTeardown(() -> dispose(name, value, definition.dispose()));
+        }
+        return value;
+    }
+
+    /** Context handed to a request factory: dependencies resolve through the active task scope. */
+    private ResourceContext requestContext(TaskScope scope) {
+        return new ResourceContext() {
+            @Override
+            public ResourceScope scope() {
+                return ResourceScope.REQUEST;
+            }
+
+            @Override
+            public <T> T use(String name) {
+                return scope.use(name);
+            }
+        };
+    }
+
     private Object build(String name, ResourceDefinition definition, ResourceContext context) {
         Set<String> chain = resolving.get();
         if (!chain.add(name)) {
@@ -192,6 +284,11 @@ private void disposeWorker() {
             }
         }
         workerCache.clear();
+        threadCache.clear();
+    }
+
+    private static String scopeWord(ResourceScope scope) {
+        return scope.name().toLowerCase(Locale.ROOT);
     }
 
     private void dispose(String name, Object value, Consumer<Object> disposer) {

sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java

@@ -4,6 +4,18 @@
 public enum ResourceScope {
     /** Built once, lazily, and shared across every task on the worker. */
     WORKER,
+    /**
+     * Built lazily once per worker thread and shared by every task that runs on
+     * that thread; disposed when the worker shuts down. If the pool retires a
+     * thread early (cached pools, autoscale), its instance is retained until
+     * worker teardown — bounded by the number of threads the pool ever created.
+     */
+    THREAD,
     /** Built lazily per task invocation and disposed when the task ends. */
-    TASK
+    TASK,
+    /**
+     * Built fresh on every {@code use()} and disposed when the task ends —
+     * never cached, so N uses inside one task yield N instances.
+     */
+    REQUEST
 }

sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java

@@ -2,10 +2,14 @@
  * Worker-side dependency injection: register a resource once, resolve it inside
  * task handlers.
  *
- * <p>Two scopes (mirroring the Node SDK): {@link org.byteveda.taskito.resources.ResourceScope#WORKER}
+ * <p>Four scopes: {@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#THREAD} resources are
+ * built lazily once per worker thread and disposed at worker shutdown;
  * {@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)}.
+ * lazily per task invocation and disposed (LIFO) when it ends; and
+ * {@link org.byteveda.taskito.resources.ResourceScope#REQUEST} resources are
+ * built fresh on every use and disposed with the task. Handlers resolve them
+ * with {@link org.byteveda.taskito.resources.Resources#use(String)}.
  */
 package org.byteveda.taskito.resources;

sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java

@@ -1,6 +1,8 @@
 package org.byteveda.taskito;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -103,6 +105,123 @@ void circularResourceDependencyIsRejected(@TempDir Path dir) throws Exception {
         }
     }
 
+    @Test
+    @Timeout(30)
+    void threadResourceReusedAcrossSequentialTasks(@TempDir Path dir) throws Exception {
+        int jobs = 2;
+        AtomicInteger disposed = new AtomicInteger();
+        try (Taskito queue =
+                Taskito.builder().url(dir.resolve("rth.db").toString()).open()) {
+            queue.resource("perThread", ResourceScope.THREAD, ctx -> new Object(), value -> disposed.incrementAndGet());
+            AtomicReference<Object> first = new AtomicReference<>();
+            AtomicReference<Object> second = new AtomicReference<>();
+            CountDownLatch ran = new CountDownLatch(jobs);
+            try (Worker worker = queue.worker()
+                    .concurrency(1) // one thread → both jobs share its instance
+                    .handle(TASK, p -> {
+                        Object instance = Resources.use("perThread");
+                        if (!first.compareAndSet(null, instance)) {
+                            second.set(instance);
+                        }
+                        ran.countDown();
+                        return p;
+                    })
+                    .start()) {
+                for (int i = 0; i < jobs; i++) {
+                    queue.enqueue(TASK, i);
+                }
+                assertTrue(ran.await(20, TimeUnit.SECONDS));
+            } // worker close disposes thread-scoped instances
+
+            assertSame(first.get(), second.get(), "same thread reuses its instance");
+            assertEquals(1, queue.resourceMetrics().get("perThread").created());
+            assertEquals(1, disposed.get(), "disposed once at worker shutdown");
+        }
+    }
+
+    @Test
+    @Timeout(30)
+    void requestResourceFreshPerUse(@TempDir Path dir) throws Exception {
+        AtomicInteger disposed = new AtomicInteger();
+        try (Taskito queue =
+                Taskito.builder().url(dir.resolve("rrq.db").toString()).open()) {
+            queue.resource("perUse", ResourceScope.REQUEST, ctx -> new Object(), value -> disposed.incrementAndGet());
+            AtomicReference<Object> a = new AtomicReference<>();
+            AtomicReference<Object> b = new AtomicReference<>();
+            CountDownLatch ran = new CountDownLatch(1);
+            try (Worker worker = queue.worker()
+                    .handle(TASK, p -> {
+                        a.set(Resources.use("perUse"));
+                        b.set(Resources.use("perUse"));
+                        ran.countDown();
+                        return p;
+                    })
+                    .start()) {
+                queue.enqueue(TASK, 1);
+                assertTrue(ran.await(20, TimeUnit.SECONDS));
+            }
+            assertNotSame(a.get(), b.get(), "every use builds a fresh instance");
+            assertEquals(2, queue.resourceMetrics().get("perUse").created());
+            assertEquals(2, disposed.get(), "both instances disposed at task end");
+        }
+    }
+
+    @Test
+    @Timeout(30)
+    void threadFactoryCannotUseTaskScopedResource(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                Taskito.builder().url(dir.resolve("rtg.db").toString()).open()) {
+            queue.resource("perTask", ResourceScope.TASK, ctx -> new Object());
+            queue.resource("perThread", ResourceScope.THREAD, ctx -> ctx.use("perTask"));
+            AtomicReference<Class<?>> thrown = new AtomicReference<>();
+            CountDownLatch ran = new CountDownLatch(1);
+            try (Worker worker = queue.worker()
+                    .handle(TASK, p -> {
+                        try {
+                            Resources.use("perThread");
+                        } 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 workerFactoryCannotUseThreadResource(@TempDir Path dir) throws Exception {
+        try (Taskito queue =
+                Taskito.builder().url(dir.resolve("rwg.db").toString()).open()) {
+            queue.resource("perThread", ResourceScope.THREAD, ctx -> new Object());
+            queue.resource("shared", ctx -> ctx.use("perThread")); // WORKER factory
+            AtomicReference<Class<?>> thrown = new AtomicReference<>();
+            CountDownLatch ran = new CountDownLatch(1);
+            try (Worker worker = queue.worker()
+                    .handle(TASK, p -> {
+                        try {
+                            Resources.use("shared");
+                        } 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 {