diff --git a/src/main/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepository.java b/src/main/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepository.java
index 03fd355bc..4d50673ea 100644
--- a/src/main/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepository.java
+++ b/src/main/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepository.java
@@ -2,12 +2,12 @@
import java.time.Instant;
import java.util.List;
+import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
import org.springframework.stereotype.Repository;
@@ -16,6 +16,11 @@
/**
* In-memory repository implementation for conversion job persistence.
+ *
+ * Once stored, a job identifier remains reserved for the lifetime of this
+ * repository instance. Deletion removes the live job and secondary index but
+ * does not make the identifier reusable. That prevents delayed asynchronous
+ * work from resolving an old identifier to a different job or tenant.
*/
@Repository
public class InMemoryConversionJobRepository implements ConversionJobRepository, ConversionJobStateStore {
@@ -27,10 +32,13 @@ public class InMemoryConversionJobRepository implements ConversionJobRepository,
private static final String EVENT_JOB_SUCCEEDED = "conversion.job.succeeded";
private static final String EVENT_JOB_FAILED = "conversion.job.failed";
private static final String EVENT_RETRY_ACCEPTED = "conversion.retry.accepted";
+ private static final String IDENTIFIER_COLLISION_MESSAGE = "Conversion job identifier collision.";
private final ConcurrentHashMap jobs = new ConcurrentHashMap<>();
private final ConcurrentHashMap jobsByTenantAndContentHash = new ConcurrentHashMap<>();
+ private final Set reservedJobIdentifiers = ConcurrentHashMap.newKeySet();
private final ConcurrentLinkedQueue lifecycleEvents = new ConcurrentLinkedQueue<>();
+ private final Object jobIndexLock = new Object();
/**
* Creates an empty thread-safe process-local job repository and lifecycle ledger.
@@ -44,11 +52,26 @@ public InMemoryConversionJobRepository() {
*/
@Override
public ConversionJob save(ConversionJob job) {
- jobs.put(job.getJobId(), job);
- if (job.getContentHash() != null && !job.getContentHash().isBlank()) {
- jobsByTenantAndContentHash.putIfAbsent(contentKey(job.getTenantId(), job.getContentHash()), job.getJobId());
+ Objects.requireNonNull(job, "job");
+ synchronized (jobIndexLock) {
+ UUID jobId = Objects.requireNonNull(job.getJobId(), "job.jobId");
+ ConversionJob existing = jobs.get(jobId);
+ if (existing == job) {
+ return existing;
+ }
+ rejectReservedIdentifier(jobId, existing);
+
+ String contentHash = job.getContentHash();
+ reservedJobIdentifiers.add(jobId);
+ jobs.put(jobId, job);
+ if (contentHash != null && !contentHash.isBlank()) {
+ jobsByTenantAndContentHash.putIfAbsent(
+ contentKey(job.getTenantId(), contentHash),
+ jobId
+ );
+ }
+ return job;
}
- return job;
}
/**
@@ -56,37 +79,36 @@ public ConversionJob save(ConversionJob job) {
*/
@Override
public ConversionJobRepository.FindOrStoreResult findOrStoreByContentHash(ConversionJob candidate) {
- String contentHash = candidate.getContentHash();
- if (contentHash == null || contentHash.isBlank()) {
- save(candidate);
+ Objects.requireNonNull(candidate, "candidate");
+ synchronized (jobIndexLock) {
+ UUID candidateJobId = Objects.requireNonNull(candidate.getJobId(), "candidate.jobId");
+ ConversionJob existingById = jobs.get(candidateJobId);
+ if (existingById == candidate) {
+ appendLifecycleEvent(existingById, EVENT_DEDUPE_HIT, null);
+ return new ConversionJobRepository.FindOrStoreResult(existingById, false);
+ }
+ rejectReservedIdentifier(candidateJobId, existingById);
+
+ String contentHash = candidate.getContentHash();
+ if (contentHash == null || contentHash.isBlank()) {
+ storeNewCandidate(candidate);
+ appendLifecycleEvent(candidate, EVENT_SUBMITTED, null);
+ return new ConversionJobRepository.FindOrStoreResult(candidate, true);
+ }
+
+ String contentIndexKey = contentKey(candidate.getTenantId(), contentHash);
+ UUID existingJobId = jobsByTenantAndContentHash.get(contentIndexKey);
+ ConversionJob existingByContent = existingJobId == null ? null : jobs.get(existingJobId);
+ if (existingByContent != null) {
+ appendLifecycleEvent(existingByContent, EVENT_DEDUPE_HIT, null);
+ return new ConversionJobRepository.FindOrStoreResult(existingByContent, false);
+ }
+
+ storeNewCandidate(candidate);
+ jobsByTenantAndContentHash.put(contentIndexKey, candidateJobId);
appendLifecycleEvent(candidate, EVENT_SUBMITTED, null);
return new ConversionJobRepository.FindOrStoreResult(candidate, true);
}
-
- String contentKey = contentKey(candidate.getTenantId(), contentHash);
- AtomicBoolean created = new AtomicBoolean(false);
- AtomicReference canonical = new AtomicReference<>();
- jobsByTenantAndContentHash.compute(
- contentKey,
- (key, existingJobId) -> {
- if (existingJobId != null) {
- ConversionJob existing = jobs.get(existingJobId);
- if (existing != null) {
- canonical.set(existing);
- return existingJobId;
- }
- }
-
- jobs.put(candidate.getJobId(), candidate);
- created.set(true);
- canonical.set(candidate);
- return candidate.getJobId();
- }
- );
-
- ConversionJob storedJob = canonical.get();
- appendLifecycleEvent(storedJob, created.get() ? EVENT_SUBMITTED : EVENT_DEDUPE_HIT, null);
- return new ConversionJobRepository.FindOrStoreResult(canonical.get(), created.get());
}
/**
@@ -149,9 +171,14 @@ public List findAllByTenantId(String tenantId) {
*/
@Override
public void deleteById(UUID jobId) {
- ConversionJob removed = jobs.remove(jobId);
- if (removed != null && removed.getContentHash() != null && !removed.getContentHash().isBlank()) {
- jobsByTenantAndContentHash.remove(contentKey(removed.getTenantId(), removed.getContentHash()), jobId);
+ synchronized (jobIndexLock) {
+ ConversionJob removed = jobs.remove(jobId);
+ if (removed != null && removed.getContentHash() != null && !removed.getContentHash().isBlank()) {
+ jobsByTenantAndContentHash.remove(
+ contentKey(removed.getTenantId(), removed.getContentHash()),
+ jobId
+ );
+ }
}
}
@@ -257,6 +284,19 @@ public boolean retryDeadLettered(UUID jobId, String operatorId) {
return true;
}
+ private void storeNewCandidate(ConversionJob candidate) {
+ UUID jobId = Objects.requireNonNull(candidate.getJobId(), "candidate.jobId");
+ rejectReservedIdentifier(jobId, jobs.get(jobId));
+ reservedJobIdentifiers.add(jobId);
+ jobs.put(jobId, candidate);
+ }
+
+ private void rejectReservedIdentifier(UUID jobId, ConversionJob existing) {
+ if (existing != null || reservedJobIdentifiers.contains(jobId)) {
+ throw new IllegalStateException(IDENTIFIER_COLLISION_MESSAGE);
+ }
+ }
+
private String contentKey(String tenantId, String contentHash) {
return normalizeTenantId(tenantId) + "\u001f" + contentHash;
}
diff --git a/src/test/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepositoryIdentifierReservationTest.java b/src/test/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepositoryIdentifierReservationTest.java
new file mode 100644
index 000000000..4f0088d98
--- /dev/null
+++ b/src/test/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepositoryIdentifierReservationTest.java
@@ -0,0 +1,128 @@
+package com.clearfolio.viewer.repository;
+
+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 org.junit.jupiter.api.Test;
+
+import com.clearfolio.viewer.model.ConversionJob;
+
+/**
+ * Proves that process-local job identifiers have immutable ownership and cannot
+ * be rebound to a different job object, including after deletion.
+ */
+class InMemoryConversionJobRepositoryIdentifierReservationTest {
+
+ private static final String COLLISION_MESSAGE = "Conversion job identifier collision.";
+
+ @Test
+ void exactLiveObjectIsAnIdempotentFindOrStoreHit() {
+ InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
+ ConversionJob job = job(UUID.randomUUID(), "tenant-north", "shared-content-hash");
+ repository.save(job);
+
+ ConversionJobRepository.FindOrStoreResult result = repository.findOrStoreByContentHash(job);
+
+ assertFalse(result.created());
+ assertSame(job, result.canonicalJob());
+ }
+
+ @Test
+ void distinctObjectCannotReplaceLiveIdentifier() {
+ InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
+ UUID sharedJobId = UUID.randomUUID();
+ ConversionJob existing = job(sharedJobId, "tenant-north", "existing-content-hash");
+ ConversionJob collision = job(sharedJobId, "tenant-south", "replacement-content-hash");
+ repository.save(existing);
+
+ IllegalStateException exception = assertThrows(
+ IllegalStateException.class,
+ () -> repository.save(collision)
+ );
+
+ assertEquals(COLLISION_MESSAGE, exception.getMessage());
+ assertSame(existing, repository.findById(sharedJobId).orElseThrow());
+ assertTrue(repository.findByTenantAndContentHash(
+ "tenant-south",
+ "replacement-content-hash"
+ ).isEmpty());
+ }
+
+ @Test
+ void collisionIsRejectedBeforeCandidateContentHashAccess() {
+ InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
+ UUID sharedJobId = UUID.randomUUID();
+ ConversionJob existing = job(sharedJobId, "tenant-north", "existing-content-hash");
+ repository.save(existing);
+
+ IllegalStateException exception = assertThrows(
+ IllegalStateException.class,
+ () -> repository.findOrStoreByContentHash(new HashAccessFailureJob(sharedJobId))
+ );
+
+ assertEquals(COLLISION_MESSAGE, exception.getMessage());
+ assertSame(existing, repository.findById(sharedJobId).orElseThrow());
+ }
+
+ @Test
+ void deletedIdentifierRemainsReservedAgainstLaterReplacement() {
+ InMemoryConversionJobRepository repository = new InMemoryConversionJobRepository();
+ UUID sharedJobId = UUID.randomUUID();
+ ConversionJob original = job(sharedJobId, "tenant-north", "original-content-hash");
+ repository.save(original);
+ repository.deleteById(sharedJobId);
+
+ ConversionJob replacement = job(sharedJobId, "tenant-south", "replacement-content-hash");
+ IllegalStateException exception = assertThrows(
+ IllegalStateException.class,
+ () -> repository.findOrStoreByContentHash(replacement)
+ );
+
+ assertEquals(COLLISION_MESSAGE, exception.getMessage());
+ assertTrue(repository.findById(sharedJobId).isEmpty());
+ assertTrue(repository.findByTenantAndContentHash(
+ "tenant-south",
+ "replacement-content-hash"
+ ).isEmpty());
+ }
+
+ private static ConversionJob job(UUID jobId, String tenantId, String contentHash) {
+ return new ConversionJob(
+ jobId,
+ tenantId,
+ "owner",
+ "document.pdf",
+ "application/pdf",
+ contentHash,
+ 100L,
+ 3
+ );
+ }
+
+ /** Candidate whose content hash must not be read after a UUID collision. */
+ private static final class HashAccessFailureJob extends ConversionJob {
+
+ private HashAccessFailureJob(UUID jobId) {
+ super(
+ jobId,
+ "tenant-south",
+ "owner",
+ "collision.pdf",
+ "application/pdf",
+ "constructor-content-hash",
+ 100L,
+ 3
+ );
+ }
+
+ @Override
+ public String getContentHash() {
+ throw new IllegalStateException("candidate content hash was accessed first");
+ }
+ }
+}