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 915840d6..fd7b8a03 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 @@ -84,12 +84,32 @@ public Object reconstruct(ProxyRef ref) { * 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()); + ProxyHandler handler = handlerFor(ref.handler()); + verify(ref, expectedPurpose); + return handler.reconstruct(ref.reference()); + } + + /** + * A new {@link ProxySession}: deconstruct/reconstruct with per-instance + * identity dedup and close-time cleanup. Confine a session to one thread. + */ + public ProxySession session() { + return new ProxySession(this); + } + + /** Look up a handler by id, failing on an unknown id. */ + @SuppressWarnings("unchecked") + ProxyHandler handlerFor(String handlerId) { + ProxyHandler handler = (ProxyHandler) handlers.get(handlerId); if (handler == null) { - throw new ProxyException("unknown proxy handler '" + ref.handler() + "'"); + throw new ProxyException("unknown proxy handler '" + handlerId + "'"); } + return handler; + } + + /** Verify a ref's signature, expiry, and (when requested) bound purpose. */ + void verify(ProxyRef ref, String expectedPurpose) { 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); @@ -102,7 +122,6 @@ public Object reconstruct(ProxyRef ref, String expectedPurpose) { if (expectedPurpose != null && !expectedPurpose.equals(ref.purpose())) { throw new ProxyException("proxy purpose mismatch for handler '" + ref.handler() + "'"); } - return handler.reconstruct(ref.reference()); } /** {@link #reconstruct(ProxyRef)} cast to the caller's type. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyHandler.java b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyHandler.java index 89fde023..e21d1a4f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyHandler.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxyHandler.java @@ -21,4 +21,12 @@ public interface ProxyHandler { /** Rebuild the resource from a reference produced by {@link #deconstruct}. */ T reconstruct(Map reference); + + /** + * Release a resource this handler reconstructed, invoked (LIFO, once per + * unique instance) when a {@link ProxySession} that produced it closes. + * Direct {@link Proxies#reconstruct(ProxyRef)} calls have no lifecycle and + * never trigger this. Default: no-op. + */ + default void cleanup(T value) {} } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxySession.java b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxySession.java new file mode 100644 index 00000000..235cd118 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/proxies/ProxySession.java @@ -0,0 +1,151 @@ +package org.byteveda.taskito.proxies; + +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.HashMap; +import java.util.IdentityHashMap; +import java.util.Map; +import org.byteveda.taskito.errors.ProxyException; + +/** + * A unit-of-work wrapper over {@link Proxies} adding identity dedup and a + * cleanup lifecycle. Within one session: + * + *
    + *
  • {@link #deconstruct(Object)} memoizes by (instance identity, purpose) — + * deconstructing the same object again returns the same {@link ProxyRef} + * without calling the handler twice. The first call's TTL wins per + * (instance, purpose); use a new session to refresh expiry. + *
  • {@link #reconstruct(ProxyRef)} memoizes by the ref's signature — every + * ref to the same underlying resource resolves to the same instance, and + * the handler reconstructs it once. Signature, expiry, and purpose are + * re-verified on every call, memo hit or not. + *
  • {@link #close()} runs {@link ProxyHandler#cleanup} once per unique + * reconstructed instance, in reverse reconstruction order (LIFO). + *
+ * + *

Not thread-safe — confine a session to the thread that created it. A + * session models one producer batch or one task invocation. + */ +public final class ProxySession implements AutoCloseable { + private static final Logger LOG = System.getLogger(ProxySession.class.getName()); + + private final Proxies proxies; + private final IdentityHashMap> deconstructed = new IdentityHashMap<>(); + private final Map reconstructed = new HashMap<>(); + private final Deque cleanups = new ArrayDeque<>(); + private boolean closed; + + ProxySession(Proxies proxies) { + this.proxies = proxies; + } + + /** Deconstruct {@code value} (no expiry or purpose), deduped by instance identity. */ + public ProxyRef deconstruct(Object value) { + return deconstruct(value, null, null); + } + + /** Deconstruct {@code value} with a TTL, deduped by instance identity. */ + public ProxyRef deconstruct(Object value, Duration ttl) { + return deconstruct(value, ttl, null); + } + + /** + * Deconstruct {@code value} bound to {@code ttl}/{@code purpose} (both + * nullable), deduped by (instance identity, purpose). + */ + public ProxyRef deconstruct(Object value, Duration ttl, String purpose) { + ensureOpen(); + if (value == null) { + throw new ProxyException("cannot deconstruct null"); + } + Map byPurpose = deconstructed.computeIfAbsent(value, key -> new HashMap<>()); + ProxyRef cached = byPurpose.get(purpose); + if (cached != null) { + return cached; + } + ProxyRef ref = proxies.deconstruct(value, ttl, purpose); + byPurpose.put(purpose, ref); + return ref; + } + + /** Verify and reconstruct {@code ref}, deduped by its signature. */ + public Object reconstruct(ProxyRef ref) { + return reconstruct(ref, null); + } + + /** + * Verify (always — including memo hits, so a ref that expires mid-session + * stops resolving) and reconstruct {@code ref}, deduped by its signature. + */ + public Object reconstruct(ProxyRef ref, String expectedPurpose) { + ensureOpen(); + ProxyHandler handler = proxies.handlerFor(ref.handler()); + proxies.verify(ref, expectedPurpose); + String signature = ref.signature(); + if (reconstructed.containsKey(signature)) { + return reconstructed.get(signature); + } + Object value = handler.reconstruct(ref.reference()); + reconstructed.put(signature, value); + cleanups.push(() -> handler.cleanup(value)); + return value; + } + + /** {@link #reconstruct(ProxyRef)} cast to the caller's type. */ + @SuppressWarnings("unchecked") + public T resolve(ProxyRef ref) { + return (T) reconstruct(ref); + } + + /** {@link #reconstruct(ProxyRef, String)} cast to the caller's type. */ + @SuppressWarnings("unchecked") + public T resolve(ProxyRef ref, String expectedPurpose) { + return (T) reconstruct(ref, expectedPurpose); + } + + /** + * Run {@link ProxyHandler#cleanup} for every reconstructed instance in + * reverse order (LIFO), once each. A cleanup failure is logged and never + * skips the rest; a JVM-level {@link Error} is rethrown only after the + * remaining cleanups have drained. Idempotent. + */ + @Override + public void close() { + if (closed) { + return; + } + closed = true; + Error fatal = null; + while (!cleanups.isEmpty()) { + Runnable cleanup = cleanups.pop(); + try { + cleanup.run(); + } catch (RuntimeException e) { + // Cleanup must never fail the rest of the teardown — record and continue. + LOG.log(Level.WARNING, "proxy cleanup failed", e); + } catch (Error e) { + // Drain the remaining cleanups first, then let the error propagate — + // aborting here would abandon them for good (closed is already set). + if (fatal == null) { + fatal = e; + } + LOG.log(Level.WARNING, "proxy cleanup threw an error", e); + } + } + deconstructed.clear(); + reconstructed.clear(); + if (fatal != null) { + throw fatal; + } + } + + private void ensureOpen() { + if (closed) { + throw new ProxyException("proxy session is closed"); + } + } +} 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 1d12e548..44d69669 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ProxyTest.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 com.fasterxml.jackson.databind.ObjectMapper; @@ -8,12 +10,16 @@ import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import org.byteveda.taskito.errors.ProxyException; import org.byteveda.taskito.proxies.FileProxyHandler; import org.byteveda.taskito.proxies.Proxies; +import org.byteveda.taskito.proxies.ProxyHandler; import org.byteveda.taskito.proxies.ProxyRef; +import org.byteveda.taskito.proxies.ProxySession; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -135,4 +141,174 @@ void enforcesPurposeWhenRequested(@TempDir Path dir) { assertEquals(file.getAbsolutePath(), ((File) proxies.reconstruct(ref)).getAbsolutePath()); // unchecked assertThrows(ProxyException.class, () -> proxies.reconstruct(ref, "billing")); } + + /** Counts handler invocations and records cleanups, for session tests. */ + private static final class CountingFileHandler implements ProxyHandler { + private final FileProxyHandler delegate = new FileProxyHandler(); + final AtomicInteger deconstructs = new AtomicInteger(); + final AtomicInteger reconstructs = new AtomicInteger(); + final List cleaned = new ArrayList<>(); + + @Override + public String id() { + return delegate.id(); + } + + @Override + public boolean handles(Object value) { + return delegate.handles(value); + } + + @Override + public Map deconstruct(File value) { + deconstructs.incrementAndGet(); + return delegate.deconstruct(value); + } + + @Override + public File reconstruct(Map reference) { + reconstructs.incrementAndGet(); + return delegate.reconstruct(reference); + } + + @Override + public void cleanup(File value) { + cleaned.add(value); + } + } + + @Test + void sessionDedupsSameInstanceOnDeconstruct(@TempDir Path dir) { + CountingFileHandler handler = new CountingFileHandler(); + Proxies proxies = new Proxies(KEY).register(handler); + File file = dir.resolve("dedup.txt").toFile(); + try (ProxySession session = proxies.session()) { + ProxyRef first = session.deconstruct(file); + ProxyRef second = session.deconstruct(file); + assertSame(first, second); + } + assertEquals(1, handler.deconstructs.get()); + } + + @Test + void sessionKeepsPurposesDistinct(@TempDir Path dir) { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + File file = dir.resolve("purpose.txt").toFile(); + try (ProxySession session = proxies.session()) { + ProxyRef emails = session.deconstruct(file, null, "emails"); + ProxyRef billing = session.deconstruct(file, null, "billing"); + assertNotSame(emails, billing); + assertEquals(file.getAbsolutePath(), ((File) proxies.reconstruct(emails, "emails")).getAbsolutePath()); + assertEquals(file.getAbsolutePath(), ((File) proxies.reconstruct(billing, "billing")).getAbsolutePath()); + } + } + + @Test + void sessionMemoizesReconstructBySignature(@TempDir Path dir) { + CountingFileHandler handler = new CountingFileHandler(); + Proxies proxies = new Proxies(KEY).register(handler); + ProxyRef ref = proxies.deconstruct(dir.resolve("memo.txt").toFile()); + try (ProxySession session = proxies.session()) { + Object first = session.reconstruct(ref); + Object second = session.reconstruct(ref); + assertSame(first, second); + } + assertEquals(1, handler.reconstructs.get()); + } + + @Test + void sessionCleanupRunsOnceLifo(@TempDir Path dir) { + CountingFileHandler handler = new CountingFileHandler(); + Proxies proxies = new Proxies(KEY).register(handler); + ProxyRef refA = proxies.deconstruct(dir.resolve("a.txt").toFile()); + ProxyRef refB = proxies.deconstruct(dir.resolve("b.txt").toFile()); + ProxySession session = proxies.session(); + File a = session.resolve(refA); + File b = session.resolve(refB); + session.resolve(refA); // memo hit — must not add a second cleanup + session.close(); + session.close(); // idempotent + assertEquals(List.of(b, a), handler.cleaned, "cleanup runs once per instance, LIFO"); + } + + @Test + void sessionsAreIsolated(@TempDir Path dir) { + CountingFileHandler handler = new CountingFileHandler(); + Proxies proxies = new Proxies(KEY).register(handler); + ProxyRef ref = proxies.deconstruct(dir.resolve("iso.txt").toFile()); + try (ProxySession one = proxies.session(); + ProxySession two = proxies.session()) { + File fromOne = one.resolve(ref); + File fromTwo = two.resolve(ref); + assertNotSame(fromOne, fromTwo); + one.close(); + assertEquals(1, handler.cleaned.size(), "closing one session leaves the other live"); + assertEquals(fromOne, handler.cleaned.get(0)); + } + } + + @Test + void sessionReverifiesOnMemoHit(@TempDir Path dir) throws Exception { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + ProxyRef ref = proxies.deconstruct(dir.resolve("ttl.txt").toFile(), Duration.ofMillis(30)); + try (ProxySession session = proxies.session()) { + session.resolve(ref); + Thread.sleep(80); + assertThrows(ProxyException.class, () -> session.resolve(ref), "expired ref must stop resolving"); + } + } + + @Test + void sessionDrainsRemainingCleanupsWhenOneThrowsError(@TempDir Path dir) { + List cleaned = new ArrayList<>(); + Proxies proxies = new Proxies(KEY).register(new ProxyHandler() { + private final FileProxyHandler delegate = new FileProxyHandler(); + + @Override + public String id() { + return delegate.id(); + } + + @Override + public boolean handles(Object value) { + return delegate.handles(value); + } + + @Override + public Map deconstruct(File value) { + return delegate.deconstruct(value); + } + + @Override + public File reconstruct(Map reference) { + return delegate.reconstruct(reference); + } + + @Override + public void cleanup(File value) { + cleaned.add(value); + if (value.getName().equals("boom.txt")) { + throw new AssertionError("cleanup error"); + } + } + }); + ProxyRef refA = proxies.deconstruct(dir.resolve("a.txt").toFile()); + ProxyRef refBoom = proxies.deconstruct(dir.resolve("boom.txt").toFile()); + ProxySession session = proxies.session(); + session.resolve(refA); + session.resolve(refBoom); // cleaned last-in-first-out → throws first + assertThrows(AssertionError.class, session::close); + assertEquals(2, cleaned.size(), "the error must not abandon the remaining cleanups"); + } + + @Test + void sessionRejectsUseAfterClose(@TempDir Path dir) { + Proxies proxies = new Proxies(KEY).register(new FileProxyHandler()); + File file = dir.resolve("closed.txt").toFile(); + ProxyRef ref = proxies.deconstruct(file); + ProxySession session = proxies.session(); + session.close(); + assertThrows(ProxyException.class, () -> session.deconstruct(file)); + assertThrows(ProxyException.class, () -> session.reconstruct(ref)); + } }