Skip to content

fix(reliability): reserve conversion job identifiers after current analytics #443

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
seonghobae wants to merge 2 commits into
base: fix/analytics-tenant-query-main-20260813-r2
Choose a base branch
from
Draft

fix(reliability): reserve conversion job identifiers after current analytics #443

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8dfab1a
test(repository): reserve conversion job identifiers
seonghobae Aug 14, 2026
d5a57a7
fix(repository): reject conversion job identifier reuse
seonghobae Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 77 additions & 37 deletions src/main/java/com/clearfolio/viewer/repository/InMemoryConversionJobRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -16,6 +16,11 @@

/**
* In-memory repository implementation for conversion job persistence.
*
* <p>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.</p>
*/
@Repository
public class InMemoryConversionJobRepository implements ConversionJobRepository, ConversionJobStateStore {
Expand All @@ -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<UUID, ConversionJob> jobs = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, UUID> jobsByTenantAndContentHash = new ConcurrentHashMap<>();
private final Set<UUID> reservedJobIdentifiers = ConcurrentHashMap.newKeySet();
private final ConcurrentLinkedQueue<ConversionJobLifecycleEvent> lifecycleEvents = new ConcurrentLinkedQueue<>();
private final Object jobIndexLock = new Object();

/**
* Creates an empty thread-safe process-local job repository and lifecycle ledger.
Expand All @@ -44,49 +52,63 @@ 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;
}

/**
* {@inheritDoc}
*/
@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<ConversionJob> 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());
}

/**
Expand Down Expand Up @@ -149,9 +171,14 @@ public List<ConversionJob> 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
);
}
}
}

Expand Down Expand Up @@ -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;
}
Expand Down
128 changes: 128 additions & 0 deletions ...learfolio/viewer/repository/InMemoryConversionJobRepositoryIdentifierReservationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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");
}
}
}