feat(java): pooled resource scope

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -41,6 +41,7 @@
 import org.byteveda.taskito.predicates.EnqueueGate;
 import org.byteveda.taskito.predicates.Predicate;
 import org.byteveda.taskito.predicates.PredicateContext;
+import org.byteveda.taskito.resources.PoolConfig;
 import org.byteveda.taskito.resources.ResourceContext;
 import org.byteveda.taskito.resources.ResourceDefinition;
 import org.byteveda.taskito.resources.ResourceRuntime;
@@ -116,6 +117,16 @@ public <T> Taskito resource(
         return this;
     }
 
+    @Override
+    public <T> Taskito resource(
+            String name, PoolConfig pool, Function<ResourceContext, T> factory, Consumer<T> dispose) {
+        resources.register(
+                name,
+                new ResourceDefinition(
+                        factory::apply, ResourceScope.POOLED, value -> dispose.accept(cast(value)), pool));
+        return this;
+    }
+
     @Override
     public Map<String, ResourceStat> resourceMetrics() {
         return resources.metrics();

sdks/java/src/main/java/org/byteveda/taskito/Taskito.java

@@ -32,6 +32,7 @@
 import org.byteveda.taskito.model.WorkerInfo;
 import org.byteveda.taskito.predicates.EnqueueGate;
 import org.byteveda.taskito.predicates.Predicate;
+import org.byteveda.taskito.resources.PoolConfig;
 import org.byteveda.taskito.resources.ResourceContext;
 import org.byteveda.taskito.resources.ResourceScope;
 import org.byteveda.taskito.resources.ResourceStat;
@@ -78,6 +79,15 @@ static Builder builder() {
     /** Register a resource with a scope and a disposer run when the scope ends. */
     <T> Taskito resource(String name, ResourceScope scope, Function<ResourceContext, T> factory, Consumer<T> dispose);
 
+    /**
+     * Register a {@link ResourceScope#POOLED} resource: a bounded pool of
+     * instances shared across tasks. Each task checks one instance out for its
+     * duration and returns it at task end; {@code pool} bounds capacity and
+     * {@code dispose} runs when the pool retires an instance (worker shutdown or
+     * {@link PoolConfig#maxLifetime()} expiry).
+     */
+    <T> Taskito resource(String name, PoolConfig pool, Function<ResourceContext, T> factory, Consumer<T> dispose);
+
     /** Per-resource counters (created / disposed / active). */
     Map<String, ResourceStat> resourceMetrics();
 

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

@@ -0,0 +1,60 @@
+package org.byteveda.taskito.resources;
+
+import java.time.Duration;
+
+/**
+ * Sizing and timing for a {@link ResourceScope#POOLED} resource (part of the
+ * cross-SDK contract for pooled resources).
+ *
+ * @param poolSize the maximum number of concurrently checked-out instances (must be > 0)
+ * @param poolMin how many instances to build eagerly at worker start (defaults to 0, at most
+ *     {@code poolSize})
+ * @param acquireTimeout how long a checkout may wait for capacity (defaults to 10s)
+ * @param maxLifetime an idle instance older than this is disposed instead of reused
+ *     ({@code null} keeps instances forever)
+ */
+public record PoolConfig(int poolSize, int poolMin, Duration acquireTimeout, Duration maxLifetime) {
+
+    private static final Duration DEFAULT_ACQUIRE_TIMEOUT = Duration.ofSeconds(10);
+
+    public PoolConfig {
+        if (poolSize <= 0) {
+            throw new IllegalArgumentException("poolSize must be > 0");
+        }
+        if (poolMin < 0) {
+            throw new IllegalArgumentException("poolMin must be >= 0");
+        }
+        if (poolMin > poolSize) {
+            throw new IllegalArgumentException("poolMin must be <= poolSize");
+        }
+        if (acquireTimeout == null) {
+            acquireTimeout = DEFAULT_ACQUIRE_TIMEOUT;
+        }
+        if (acquireTimeout.isNegative() || acquireTimeout.isZero()) {
+            throw new IllegalArgumentException("acquireTimeout must be positive");
+        }
+        if (maxLifetime != null && (maxLifetime.isNegative() || maxLifetime.isZero())) {
+            throw new IllegalArgumentException("maxLifetime must be positive");
+        }
+    }
+
+    /** A pool of {@code poolSize} with no prewarm, a 10s acquire timeout, and unlimited lifetime. */
+    public static PoolConfig of(int poolSize) {
+        return new PoolConfig(poolSize, 0, null, null);
+    }
+
+    /** This configuration with {@code poolMin} instances built eagerly at worker start. */
+    public PoolConfig withPoolMin(int poolMin) {
+        return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime);
+    }
+
+    /** This configuration with a different checkout wait limit. */
+    public PoolConfig withAcquireTimeout(Duration acquireTimeout) {
+        return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime);
+    }
+
+    /** This configuration with a maximum instance lifetime ({@code null} for unlimited). */
+    public PoolConfig withMaxLifetime(Duration maxLifetime) {
+        return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime);
+    }
+}

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

@@ -2,8 +2,9 @@
 
 /**
  * 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
+ * may only depend on same-or-longer-lived resources: {@code WORKER} and
+ * {@code POOLED} factories may {@link #use} only worker resources (a pooled
+ * instance outlives the task that built it), a {@code THREAD} factory may use
  * worker or thread resources, and {@code TASK}/{@code REQUEST} factories may
  * use any scope.
  */

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

@@ -9,9 +9,11 @@
  * @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)
+ * @param pool bounded-pool sizing, required for {@link ResourceScope#POOLED} and
+ *     {@code null} for every other scope
  */
 public record ResourceDefinition(
-        Function<ResourceContext, Object> factory, ResourceScope scope, Consumer<Object> dispose) {
+        Function<ResourceContext, Object> factory, ResourceScope scope, Consumer<Object> dispose, PoolConfig pool) {
 
     public ResourceDefinition {
         if (factory == null) {
@@ -20,6 +22,18 @@ public record ResourceDefinition(
         if (scope == null) {
             scope = ResourceScope.WORKER;
         }
+        if (scope == ResourceScope.POOLED && pool == null) {
+            throw new IllegalArgumentException("a pooled resource requires a PoolConfig");
+        }
+        if (scope != ResourceScope.POOLED && pool != null) {
+            throw new IllegalArgumentException("a PoolConfig is only valid for a pooled resource");
+        }
+    }
+
+    /** A definition for any non-pooled scope. */
+    public ResourceDefinition(
+            Function<ResourceContext, Object> factory, ResourceScope scope, Consumer<Object> dispose) {
+        this(factory, scope, dispose, null);
     }
 
     /** A worker-scoped resource with no disposer. */

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

@@ -0,0 +1,231 @@
+package org.byteveda.taskito.resources;
+
+import java.lang.System.Logger;
+import java.lang.System.Logger.Level;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import org.byteveda.taskito.errors.ResourceException;
+
+/**
+ * A bounded checkout/return pool backing {@link ResourceScope#POOLED} resources
+ * (part of the cross-SDK contract for pooled resources). A semaphore caps
+ * concurrent checkouts at {@link PoolConfig#poolSize()}; released instances park
+ * on an idle deque and are reused unless {@link PoolConfig#maxLifetime()} has
+ * elapsed, in which case they are disposed and a fresh instance is built. The
+ * factory always runs on the acquiring thread (so per-thread bookkeeping such as
+ * dependency-cycle detection keeps working). Thread-safe.
+ */
+public final class ResourcePool {
+    private static final Logger LOG = System.getLogger(ResourcePool.class.getName());
+    /** Sentinel distinguishing "no idle instance" from a factory that returned {@code null}. */
+    private static final Object NO_IDLE = new Object();
+
+    private final String name;
+    private final PoolConfig config;
+    private final Supplier<Object> factory;
+    private final Consumer<Object> dispose;
+
+    private final Semaphore capacity;
+    private final Deque<IdleInstance> idle = new ArrayDeque<>();
+    private final ReentrantLock lock = new ReentrantLock();
+    private int active; // guarded by lock
+    private long totalAcquisitions; // guarded by lock
+    private long totalTimeouts; // guarded by lock
+    private boolean closed; // guarded by lock
+
+    /** An instance parked in the pool, stamped so {@code maxLifetime} can expire it. */
+    private record IdleInstance(Object instance, long createdAtNanos) {
+        static IdleInstance of(Object instance) {
+            return new IdleInstance(instance, System.nanoTime());
+        }
+    }
+
+    /**
+     * A point-in-time snapshot of the pool.
+     *
+     * @param size the configured capacity
+     * @param active instances currently checked out
+     * @param idle instances parked and ready for reuse
+     * @param totalAcquisitions successful checkouts so far
+     * @param totalTimeouts checkouts that gave up waiting for capacity
+     */
+    public record Stats(int size, int active, int idle, long totalAcquisitions, long totalTimeouts) {}
+
+    /**
+     * @param name the resource name (used in error messages and logs)
+     * @param config pool sizing and timing
+     * @param factory builds one instance; invoked on the acquiring thread
+     * @param dispose tears one instance down, or {@code null} for none
+     */
+    public ResourcePool(String name, PoolConfig config, Supplier<Object> factory, Consumer<Object> dispose) {
+        this.name = name;
+        this.config = config;
+        this.factory = factory;
+        this.dispose = dispose;
+        this.capacity = new Semaphore(config.poolSize());
+    }
+
+    /** Build {@link PoolConfig#poolMin()} instances upfront, best-effort: log and stop on the first failure. */
+    public void prewarm() {
+        for (int i = 0; i < config.poolMin(); i++) {
+            Object instance;
+            try {
+                instance = factory.get();
+            } catch (RuntimeException e) {
+                LOG.log(Level.WARNING, "failed to prewarm resource '" + name + "'", e);
+                return;
+            }
+            lock.lock();
+            try {
+                if (closed) {
+                    // A shutdown raced the prewarm — don't park instances on a
+                    // drained pool; dispose the fresh build and stop.
+                    disposeQuietly(instance);
+                    return;
+                }
+                idle.addLast(IdleInstance.of(instance));
+            } finally {
+                lock.unlock();
+            }
+        }
+    }
+
+    /**
+     * Check an instance out, blocking up to {@link PoolConfig#acquireTimeout()}
+     * for capacity. Reuses an idle instance when one is still within its
+     * lifetime; otherwise builds a fresh one on the calling thread.
+     *
+     * @throws ResourceException when the pool stays exhausted past the timeout
+     */
+    public Object acquire() {
+        awaitCapacity();
+        Object reused = pollIdle();
+        if (reused != NO_IDLE) {
+            return reused;
+        }
+        // No idle instance — build one, holding the permit throughout. The active
+        // count moves only after the factory succeeds, so a failing factory can
+        // neither leak capacity nor underflow the counter.
+        Object instance;
+        try {
+            instance = factory.get();
+        } catch (RuntimeException | Error e) {
+            capacity.release();
+            throw e;
+        }
+        lock.lock();
+        try {
+            recordAcquired();
+        } finally {
+            lock.unlock();
+        }
+        return instance;
+    }
+
+    /** Return a checked-out instance: park it for reuse, or dispose it if the pool has shut down. */
+    public void release(Object instance) {
+        lock.lock();
+        try {
+            active--;
+            if (closed) {
+                disposeQuietly(instance);
+            } else {
+                idle.addLast(IdleInstance.of(instance));
+            }
+        } finally {
+            lock.unlock();
+        }
+        capacity.release();
+    }
+
+    /** Dispose every idle instance and mark the pool shut down (later releases dispose immediately). */
+    public void shutdown() {
+        lock.lock();
+        try {
+            closed = true;
+            while (!idle.isEmpty()) {
+                disposeQuietly(idle.pollFirst().instance());
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    /** A snapshot of the pool's counters. */
+    public Stats stats() {
+        lock.lock();
+        try {
+            return new Stats(config.poolSize(), active, idle.size(), totalAcquisitions, totalTimeouts);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    /** Take a capacity permit or fail with a timeout error. */
+    private void awaitCapacity() {
+        boolean acquired;
+        try {
+            acquired = capacity.tryAcquire(config.acquireTimeout().toNanos(), TimeUnit.NANOSECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new ResourceException("interrupted while acquiring resource '" + name + "' from its pool", e);
+        }
+        if (!acquired) {
+            lock.lock();
+            try {
+                totalTimeouts++;
+            } finally {
+                lock.unlock();
+            }
+            throw new ResourceException("resource '" + name + "' pool timed out after "
+                    + config.acquireTimeout().toMillis() + "ms");
+        }
+    }
+
+    /** Pop a live idle instance, disposing any that outlived {@code maxLifetime}; {@link #NO_IDLE} if none. */
+    private Object pollIdle() {
+        lock.lock();
+        try {
+            while (!idle.isEmpty()) {
+                IdleInstance entry = idle.pollFirst();
+                if (!expired(entry)) {
+                    recordAcquired();
+                    return entry.instance();
+                }
+                disposeQuietly(entry.instance());
+            }
+            return NO_IDLE;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private boolean expired(IdleInstance entry) {
+        Duration lifetime = config.maxLifetime();
+        return lifetime != null && System.nanoTime() - entry.createdAtNanos() >= lifetime.toNanos();
+    }
+
+    /** Update counters for a successful checkout. Caller holds {@code lock}. */
+    private void recordAcquired() {
+        totalAcquisitions++;
+        active++;
+    }
+
+    /** Dispose one instance; a failing disposer is logged, never propagated. */
+    private void disposeQuietly(Object instance) {
+        if (dispose == null) {
+            return;
+        }
+        try {
+            dispose.accept(instance);
+        } catch (RuntimeException e) {
+            LOG.log(Level.WARNING, "disposing pooled resource '" + name + "' failed", e);
+        }
+    }
+}