diff --git a/src/main/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistry.java b/src/main/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistry.java new file mode 100644 index 000000000..20f09bd10 --- /dev/null +++ b/src/main/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistry.java @@ -0,0 +1,71 @@ +package com.clearfolio.viewer.lifecycle; + +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; + +/** + * Provides fixed-memory, process-local serialization for one document lifecycle. + * + *

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.

+ * + *

This reference registry protects one JVM. Multi-instance deployments must + * provide an equivalent distributed generation fence, object-version + * precondition, or transactional outbox consumer guarantee.

+ */ +public final class ArtifactLifecycleLockRegistry { + + private static final int DEFAULT_STRIPE_COUNT = 256; + private static final ArtifactLifecycleLockRegistry SHARED = + new ArtifactLifecycleLockRegistry(); + + private final ReentrantLock[] locks; + + private ArtifactLifecycleLockRegistry() { + this.locks = new ReentrantLock[DEFAULT_STRIPE_COUNT]; + for (int index = 0; index < locks.length; index++) { + locks[index] = new ReentrantLock(); + } + } + + /** + * Returns the sole process-wide registry. + * + *

A single construction path prevents Spring injection and standalone + * callers from creating disjoint lock domains for the same job identifier.

+ * + * @return shared fixed-memory lifecycle lock registry + */ + public static ArtifactLifecycleLockRegistry shared() { + return SHARED; + } + + /** + * Executes one action while holding the stripe for a conversion job. + * + *

The lock is always released, including when the supplied action throws. + * The action's return value or runtime exception is passed through unchanged.

+ * + * @param jobId permanently reserved conversion-job identifier + * @param action work that reads or mutates metadata and artifact bytes + * @param action return type + * @return value returned by the action + * @throws NullPointerException when the identifier or action is absent + */ + public T withJobLock(UUID jobId, Supplier action) { + UUID requiredJobId = Objects.requireNonNull(jobId, "jobId"); + Supplier requiredAction = Objects.requireNonNull(action, "action"); + ReentrantLock lock = locks[Math.floorMod(requiredJobId.hashCode(), locks.length)]; + lock.lock(); + try { + return requiredAction.get(); + } finally { + lock.unlock(); + } + } +} diff --git a/src/test/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistryTest.java b/src/test/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistryTest.java new file mode 100644 index 000000000..8e9f00dd4 --- /dev/null +++ b/src/test/java/com/clearfolio/viewer/lifecycle/ArtifactLifecycleLockRegistryTest.java @@ -0,0 +1,117 @@ +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 secondThread = new AtomicReference<>(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future first = executor.submit(() -> registry.withJobLock(jobId, () -> { + firstEntered.countDown(); + await(releaseFirst); + return "first"; + })); + assertTrue(firstEntered.await(5, TimeUnit.SECONDS)); + + Future 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); + assertFalse(ArtifactLifecycleLockRegistry.class.isAnnotationPresent( + org.springframework.stereotype.Component.class)); + 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); + } + } +}