fix(reliability): serialize same-job artifact lifecycle work

src/main/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistry.java

@@ -0,0 +1,81 @@
+package com.clearfolio.viewer.lifecycle;
+
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Provides fixed-memory, process-local serialization for one document lifecycle.
+ *
+ * <p>Conversion and deletion must not write or remove bytes for the same
+ * permanently reserved job identifier at the same time. The registry hashes
+ * identifiers onto a fixed number of lock stripes, so memory use remains
+ * bounded while unrelated jobs may proceed concurrently unless they share a
+ * stripe.</p>
+ *
+ * <p>This reference registry protects one JVM. Multi-instance deployments must
+ * provide an equivalent distributed generation fence, object-version
+ * precondition, or transactional outbox consumer guarantee.</p>
+ */
+@Component
+public final class ArtifactLifecycleLockRegistry {
+
+    private static final int DEFAULT_STRIPE_COUNT = 256;
+    private static final ArtifactLifecycleLockRegistry SHARED =
+            new ArtifactLifecycleLockRegistry(DEFAULT_STRIPE_COUNT);
+
+    private final ReentrantLock[] locks;
+
+    /**
+     * Creates a registry with the default fixed number of lock stripes.
+     */
+    public ArtifactLifecycleLockRegistry() {
+        this(DEFAULT_STRIPE_COUNT);
+    }
+
+    ArtifactLifecycleLockRegistry(int stripeCount) {
+        if (stripeCount <= 0) {
+            throw new IllegalArgumentException("stripeCount must be positive");
+        }
+        this.locks = new ReentrantLock[stripeCount];
+        for (int index = 0; index < stripeCount; index++) {
+            locks[index] = new ReentrantLock();
+        }
+    }
+
+    /**
+     * Returns the process-wide registry used by standalone constructors.
+     *
+     * @return shared fixed-memory lifecycle lock registry
+     */
+    public static ArtifactLifecycleLockRegistry shared() {
+        return SHARED;
+    }
+
+    /**
+     * Executes one action while holding the stripe for a conversion job.
+     *
+     * <p>The lock is always released, including when the supplied action throws.
+     * The action's return value or runtime exception is passed through unchanged.</p>
+     *
+     * @param jobId permanently reserved conversion-job identifier
+     * @param action work that reads or mutates metadata and artifact bytes
+     * @param <T> action return type
+     * @return value returned by the action
+     * @throws NullPointerException when the identifier or action is absent
+     */
+    public <T> T withJobLock(UUID jobId, Supplier<T> action) {
+        UUID requiredJobId = Objects.requireNonNull(jobId, "jobId");
+        Supplier<T> requiredAction = Objects.requireNonNull(action, "action");
+        ReentrantLock lock = locks[Math.floorMod(requiredJobId.hashCode(), locks.length)];
+        lock.lock();
+        try {
+            return requiredAction.get();
+        } finally {
+            lock.unlock();
+        }
+    }
+}

src/test/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistryTest.java

@@ -0,0 +1,115 @@
+package com.clearfolio.viewer.lifecycle;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies fixed-memory process-local serialization for artifact lifecycles.
+ */
+class ArtifactLifecycleLockRegistryTest {
+
+    @Test
+    void sameJobCannotEnterTwoArtifactLifecyclesConcurrently() throws Exception {
+        ArtifactLifecycleLockRegistry registry = ArtifactLifecycleLockRegistry.shared();
+        UUID jobId = UUID.fromString("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee");
+        CountDownLatch firstEntered = new CountDownLatch(1);
+        CountDownLatch releaseFirst = new CountDownLatch(1);
+        CountDownLatch secondStarted = new CountDownLatch(1);
+        CountDownLatch secondEntered = new CountDownLatch(1);
+        AtomicReference<Thread> secondThread = new AtomicReference<>();
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            Future<String> first = executor.submit(() -> registry.withJobLock(jobId, () -> {
+                firstEntered.countDown();
+                await(releaseFirst);
+                return "first";
+            }));
+            assertTrue(firstEntered.await(5, TimeUnit.SECONDS));
+
+            Future<String> second = executor.submit(() -> {
+                secondThread.set(Thread.currentThread());
+                secondStarted.countDown();
+                return registry.withJobLock(jobId, () -> {
+                    secondEntered.countDown();
+                    return "second";
+                });
+            });
+
+            assertTrue(secondStarted.await(5, TimeUnit.SECONDS));
+            assertTrue(awaitBlockedOnLock(secondThread.get()));
+            assertFalse(secondEntered.await(200, TimeUnit.MILLISECONDS));
+            releaseFirst.countDown();
+
+            assertEquals("first", first.get(5, TimeUnit.SECONDS));
+            assertEquals("second", second.get(5, TimeUnit.SECONDS));
+            assertTrue(secondEntered.await(5, TimeUnit.SECONDS));
+        } finally {
+            releaseFirst.countDown();
+            executor.shutdownNow();
+            assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test
+    void actionFailureReleasesStripeForFollowingWork() {
+        ArtifactLifecycleLockRegistry registry = ArtifactLifecycleLockRegistry.shared();
+        UUID jobId = UUID.randomUUID();
+        IllegalStateException expected = new IllegalStateException("controlled failure");
+
+        IllegalStateException actual = assertThrows(
+                IllegalStateException.class,
+                () -> registry.withJobLock(jobId, () -> {
+                    throw expected;
+                })
+        );
+
+        assertSame(expected, actual);
+        assertEquals("recovered", registry.withJobLock(jobId, () -> "recovered"));
+    }
+
+    @Test
+    void constructionAndNullInputsFailClosed() {
+        ArtifactLifecycleLockRegistry registry = ArtifactLifecycleLockRegistry.shared();
+
+        assertEquals(0, ArtifactLifecycleLockRegistry.class.getConstructors().length);
+        assertSame(registry, ArtifactLifecycleLockRegistry.shared());
+        assertThrows(NullPointerException.class, () -> registry.withJobLock(null, () -> "unused"));
+        assertThrows(NullPointerException.class, () -> registry.withJobLock(UUID.randomUUID(), null));
+    }
+
+    private static boolean awaitBlockedOnLock(Thread thread) {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (System.nanoTime() < deadline) {
+            Thread.State state = thread.getState();
+            if (state == Thread.State.WAITING || state == Thread.State.BLOCKED) {
+                return true;
+            }
+            Thread.onSpinWait();
+        }
+        return false;
+    }
+
+    private static void await(CountDownLatch latch) {
+        try {
+            if (!latch.await(5, TimeUnit.SECONDS)) {
+                throw new IllegalStateException("latch timeout");
+            }
+        } catch (InterruptedException exception) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("interrupted", exception);
+        }
+    }
+}