From 20d912f67eec406a1ca7f36bee0dceb87c21be61 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Tue, 30 Jun 2026 16:02:14 +0200 Subject: [PATCH 1/4] UNOMI-939: Fix scheduler duplicate dispatch and cancel-from-completed races --- .../impl/scheduler/SchedulerServiceImpl.java | 8 +- .../impl/scheduler/TaskExecutionManager.java | 84 ++++++++++++------- .../impl/scheduler/TaskRecoveryManager.java | 9 ++ .../impl/scheduler/TaskStateManager.java | 5 +- 4 files changed, 72 insertions(+), 34 deletions(-) diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java index 74f3cbe71..646eb3561 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java +++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java @@ -1172,10 +1172,14 @@ private void cancelTaskInternal(String taskId) { } ScheduledTask task = getTask(taskId); if (task != null) { - // Only cancel if in a cancellable state + // Only cancel if in a cancellable state. COMPLETED is included because a periodic task + // can be observed in that transient state by a caller racing handleTaskCompletion() + // (status briefly goes RUNNING -> COMPLETED -> SCHEDULED); without it, a cancelTask() + // call landing in that window would silently no-op and leave the task enabled. if (task.getStatus() == ScheduledTask.TaskStatus.SCHEDULED || task.getStatus() == ScheduledTask.TaskStatus.WAITING || - task.getStatus() == ScheduledTask.TaskStatus.RUNNING) { + task.getStatus() == ScheduledTask.TaskStatus.RUNNING || + task.getStatus() == ScheduledTask.TaskStatus.COMPLETED) { task.setEnabled(false); stateManager.updateTaskState(task, ScheduledTask.TaskStatus.CANCELLED, null, nodeId); diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java index f787ac563..206e2eb00 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java +++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java @@ -176,13 +176,28 @@ public void executeTask(ScheduledTask task, TaskExecutor executor) { // Ensure the executing set exists even under concurrent clears during shutdown Set executingSet = executingTasksByType.computeIfAbsent(taskType, k -> ConcurrentHashMap.newKeySet()); + // Atomically claim this task before dispatching. The persisted/cached status check above + // is only updated once prepareForExecution() actually runs inside the async taskWrapper below, + // so a second poll tick landing in that gap could otherwise see a stale non-RUNNING status and + // dispatch the same task again. Set.add() is atomic, so only one caller can win this race. + if (!executingSet.add(task.getItemId())) { + LOGGER.debug("Node {} : Task {} is already dispatched for execution, skipping duplicate dispatch", nodeId, task.getItemId()); + return; + } + TaskExecutor.TaskStatusCallback statusCallback = createStatusCallback(task); Runnable taskWrapper = createTaskWrapper(task, executor, statusCallback); // Execute task immediately using the scheduler - ScheduledFuture future = scheduler.schedule(taskWrapper, 0, TimeUnit.MILLISECONDS); - scheduledTasks.put(task.getItemId(), future); - executingSet.add(task.getItemId()); + try { + ScheduledFuture future = scheduler.schedule(taskWrapper, 0, TimeUnit.MILLISECONDS); + scheduledTasks.put(task.getItemId(), future); + } catch (Exception e) { + // Scheduling failed (e.g. rejected during shutdown): release the claim so a later + // attempt isn't permanently blocked, since the wrapper that would normally do so never ran. + executingSet.remove(task.getItemId()); + throw e; + } } catch (Exception e) { LOGGER.error("Node "+nodeId+", Error executing task: " + task.getItemId(), e); handleTaskError(task, e.getMessage(), System.currentTimeMillis()); @@ -281,35 +296,34 @@ private Runnable createTaskWrapper(ScheduledTask task, TaskExecutor executor, return; } - // Check shutdown again before preparing for execution - if (schedulerService != null && schedulerService.isShutdownNow()) { - LOGGER.debug("Node {} : Skipping task {} execution as scheduler is shutting down", nodeId, taskId); - return; - } - - // Prepare task for execution (both persistent and in-memory) - if (!prepareForExecution(task)) { - return; - } - - // Final shutdown check before executing - if (schedulerService != null && schedulerService.isShutdownNow()) { - LOGGER.debug("Node {} : Skipping task {} execution as scheduler is shutting down", nodeId, taskId); - return; - } - + // From here on, executeTask() has already atomically claimed taskId in executingTasksByType + // before scheduling this wrapper. Every exit path below - including the early returns for + // shutdown and prepareForExecution() failing (e.g. lock already held by the caller, as + // happens when TaskRecoveryManager.attemptTaskResumption()/attemptTaskRestart() acquire the + // lock themselves before calling executeTask()) - must release that claim in the outer + // finally below, or the task is permanently blocked from ever being dispatched again. + boolean executingNodeIdSet = false; try { - // Get or create the executing tasks set - Set executingTasks = executingTasksByType.computeIfAbsent(taskType, - k -> ConcurrentHashMap.newKeySet()); + // Check shutdown again before preparing for execution + if (schedulerService != null && schedulerService.isShutdownNow()) { + LOGGER.debug("Node {} : Skipping task {} execution as scheduler is shutting down", nodeId, taskId); + return; + } + + // Prepare task for execution (both persistent and in-memory) + if (!prepareForExecution(task)) { + return; + } - // Only add to executing set if not already there - if (taskId != null) { - executingTasks.add(taskId); + // Final shutdown check before executing + if (schedulerService != null && schedulerService.isShutdownNow()) { + LOGGER.debug("Node {} : Skipping task {} execution as scheduler is shutting down", nodeId, taskId); + return; } // Set the executing node ID task.setExecutingNodeId(nodeId); + executingNodeIdSet = true; schedulerService.saveTask(task); long startTime = System.currentTimeMillis(); @@ -329,11 +343,15 @@ private Runnable createTaskWrapper(ScheduledTask task, TaskExecutor executor, LOGGER.error("Unexpected error while executing task: " + taskId, e); statusCallback.fail("Unexpected error: " + e.getMessage()); } finally { - // Clear executing node ID - task.setExecutingNodeId(null); - schedulerService.saveTask(task); + // Only clear/save executingNodeId if we actually set it above; otherwise we never + // touched the task and a redundant save here could race a concurrent legitimate holder. + if (executingNodeIdSet) { + task.setExecutingNodeId(null); + schedulerService.saveTask(task); + } - // Remove task from executing set + // Always release the dispatch claim taken by executeTask(), regardless of which path + // above was taken. try { Set executingTasks = executingTasksByType.get(taskType); if (executingTasks != null && taskId != null) { @@ -424,7 +442,11 @@ private void handleTaskError(ScheduledTask task, String error, long startTime) { executeTask(task, executor); } }; - long retryDelay = task.getNextScheduledExecution().getTime() - System.currentTimeMillis(); + // Use the configured retry delay directly rather than re-deriving it from + // nextScheduledExecution: that target was computed before the state/history/metrics + // bookkeeping above ran, so subtracting "now" here would silently erode the delay + // by however long that bookkeeping took (worse under slower/contended runners). + long retryDelay = task.getRetryDelay(); scheduler.schedule(retryTask, retryDelay, TimeUnit.MILLISECONDS); LOGGER.debug("Scheduled retry #{} for task {} in {} ms", task.getFailureCount(), task.getItemId(), retryDelay); diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java index 03691ba62..56c771317 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java +++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java @@ -134,6 +134,15 @@ private void recoverCrashedTask(ScheduledTask task) { return; } + // Re-check shutdown right before writing: preDestroy() marks RUNNING tasks owned by this + // node as crashed with a more specific "Interrupted by scheduler shutdown" cause, and that + // write must win over the generic "Node failure detected" cause below if both race. + if (shutdownNow) { + LOGGER.debug("Node {} Skipping recovery of task {} : {} as scheduler is shutting down", + nodeId, task.getTaskType(), task.getItemId()); + return; + } + // First mark as crashed and release lock String previousOwner = task.getLockOwner(); if (task.getStatus() != ScheduledTask.TaskStatus.CRASHED) { diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java index cce41b29b..1e909bd90 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java +++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java @@ -39,7 +39,10 @@ public enum TaskTransition { EXECUTE(TaskStatus.RUNNING, EnumSet.of(TaskStatus.SCHEDULED, TaskStatus.CRASHED, TaskStatus.WAITING)), COMPLETE(TaskStatus.COMPLETED, EnumSet.of(TaskStatus.RUNNING)), FAIL(TaskStatus.FAILED, EnumSet.of(TaskStatus.RUNNING)), - CANCEL(TaskStatus.CANCELLED, EnumSet.of(TaskStatus.RUNNING, TaskStatus.SCHEDULED, TaskStatus.WAITING)), + // COMPLETED is included because a periodic task can be observed in that transient state + // (briefly, between finishing one run and being rescheduled for the next) by a concurrent + // cancelTask() call; cancellation should still apply rather than silently no-op. + CANCEL(TaskStatus.CANCELLED, EnumSet.of(TaskStatus.RUNNING, TaskStatus.SCHEDULED, TaskStatus.WAITING, TaskStatus.COMPLETED)), CRASH(TaskStatus.CRASHED, EnumSet.of(TaskStatus.RUNNING, TaskStatus.SCHEDULED)), WAIT(TaskStatus.WAITING, EnumSet.of(TaskStatus.SCHEDULED, TaskStatus.RUNNING)); From 934bc927a25a7339129a4f4df955aa8b7f5948c6 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Tue, 30 Jun 2026 16:02:14 +0200 Subject: [PATCH 2/4] UNOMI-940: Serialize cache refresh with direct writes per item type --- .../AbstractMultiTypeCachingService.java | 59 +++++++++++++++++-- 1 file changed, 54 insertions(+), 5 deletions(-) diff --git a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java index 499c7aa7d..4891fc347 100644 --- a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java +++ b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java @@ -43,6 +43,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Function; import java.util.stream.Collectors; @@ -68,6 +71,25 @@ public abstract class AbstractMultiTypeCachingService extends AbstractContextAwa protected TenantService tenantService; protected AuditService auditService; + /** + * Per-item-type locks ensuring saveItem()/removeItem() (direct writes) and refreshTypeCache()'s + * persistence-read-then-cache-write cycle can never interleave for the same item type. Without this, + * refreshTypeCache() could read a snapshot from persistence just before a concurrent direct write + * commits, then overwrite the cache with that stale snapshot after the direct write's own cache.put() + * already applied the fresh value. Direct writes only need to exclude a concurrent refresh of the + * same type, not each other, so they take the read lock (shared, concurrent) while refreshTypeCache() + * takes the write lock (exclusive) for its whole read-then-cache cycle, including the "remove items no + * longer in persistence" step, which reads from the same persistence snapshot. + * + * Cardinality is bounded by the number of distinct item types a service registers (getTypeConfigs()), + * typically one to a handful per service, so this map needs no eviction. + */ + private final Map typeRefreshLocks = new ConcurrentHashMap<>(); + + private ReadWriteLock typeRefreshLock(String itemType) { + return typeRefreshLocks.computeIfAbsent(itemType, k -> new ReentrantReadWriteLock()); + } + /** * Map tracking which plugin/bundle contributed which items. * Key is the bundle ID, value is the list of items contributed by that bundle. @@ -261,12 +283,25 @@ protected void shutdownTimers() { scheduledRefreshTasks.clear(); } - @SuppressWarnings("unchecked") protected void refreshTypeCache(CacheableTypeConfig config) { if (!config.isRequiresRefresh()) { return; } + // Excludes any concurrent saveItem()/removeItem() for this item type for the duration of this + // refresh, so its persistence read can never be overwritten by - or overwrite - a direct write. + // See typeRefreshLocks for the full rationale. + Lock refreshLock = typeRefreshLock(config.getItemType()).writeLock(); + refreshLock.lock(); + try { + refreshTypeCacheLocked(config); + } finally { + refreshLock.unlock(); + } + } + + @SuppressWarnings("unchecked") + private void refreshTypeCacheLocked(CacheableTypeConfig config) { // Only create the global state maps if we need them Map> oldGlobalState = null; Map> newGlobalState = null; @@ -858,8 +893,15 @@ protected void saveItem(T item, Function void saveItem(T item, Function void removeItem(String id, Class itemClass, String itemType) { String currentTenant = contextManager.getCurrentContext().getTenantId(); - persistenceService.remove(id, itemClass); - cacheService.remove(itemType, id, currentTenant, itemClass); + // Excludes a concurrent refreshTypeCache() for this item type: see typeRefreshLocks. + Lock writeGuard = typeRefreshLock(itemType).readLock(); + writeGuard.lock(); + try { + persistenceService.remove(id, itemClass); + cacheService.remove(itemType, id, currentTenant, itemClass); + } finally { + writeGuard.unlock(); + } } } From 2de436808ab30e09adaa2b06b181ce4eea3e02ae Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Tue, 30 Jun 2026 16:02:14 +0200 Subject: [PATCH 3/4] test: Harden unit tests for slow CI runners --- .../impl/events/EventServiceImplTest.java | 7 +- .../scheduler/SchedulerServiceImplTest.java | 104 ++++++++++++++++++ .../impl/DefaultTracerServiceTest.java | 4 +- 3 files changed, 111 insertions(+), 4 deletions(-) diff --git a/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java b/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java index 109d5ef97..a5092fd22 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java @@ -417,10 +417,13 @@ public void testSearch_WithFullTextAndCondition() { condition.setParameter("propertyValue", "test"); query.setCondition(condition); - // Test - retry until event is available (handles refresh delay) + // Test - retry until event is available (handles refresh delay). + // Full-text indexing can take longer than the default 2000ms budget on loaded CI runners, + // so use a more generous timeout for this specific query. PartialList result = TestHelper.retryQueryUntilAvailable( () -> eventService.search(query), - 1 + 1, + 10000L ); // Verify diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java index 593867e2a..74bb68f12 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java @@ -622,6 +622,56 @@ public boolean canResume(ScheduledTask task) { assertTrue(recovered.get(), "Task should be recovered and executed"); } + @Test + @Tag("ClusterTests") + public void testNoDuplicateDispatchUnderSlowExecution() throws Exception { + // Validates TaskExecutionManager.executeTask(): the dispatch claim it takes in + // executingTasksByType before scheduling a task wrapper must hold for the entire run, or a + // poll tick landing while the first run is still in flight (status not yet flipped to RUNNING, + // or - as observed before this fix - never released because prepareForExecution() returned + // false on an early-return path) could dispatch the same task a second time. The execution + // below deliberately runs past several poll ticks (TASK_CHECK_INTERVAL = 1000ms) to exercise + // that window. + AtomicInteger concurrentExecutions = new AtomicInteger(0); + AtomicInteger maxConcurrentExecutions = new AtomicInteger(0); + AtomicInteger totalExecutions = new AtomicInteger(0); + CountDownLatch completionLatch = new CountDownLatch(1); + + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "slow-dispatch-test"; + } + + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) { + int concurrent = concurrentExecutions.incrementAndGet(); + maxConcurrentExecutions.updateAndGet(max -> Math.max(max, concurrent)); + totalExecutions.incrementAndGet(); + try { + Thread.sleep(2500); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + concurrentExecutions.decrementAndGet(); + } + callback.complete(); + completionLatch.countDown(); + } + }; + + schedulerService.registerTaskExecutor(executor); + schedulerService.newTask("slow-dispatch-test") + .disallowParallelExecution() + .schedule(); + + assertTrue(completionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should complete"); + + assertEquals(1, maxConcurrentExecutions.get(), "Task must never run concurrently with itself"); + assertEquals(1, totalExecutions.get(), + "Task must execute exactly once despite a slow run spanning multiple poll ticks"); + } + // Metrics and history tests @Test @Tag("MetricsTests") @@ -1223,6 +1273,60 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { assertFalse(completed.get(), "Task should not complete after cancellation"); } + /** + * Validates the fix broadening the CANCEL transition to accept COMPLETED. A periodic task only + * observes COMPLETED transiently (RUNNING -> COMPLETED -> SCHEDULED), but a one-shot task stays + * COMPLETED indefinitely once it finishes, giving a fully deterministic way to exercise + * cancelTask() against a task that is genuinely COMPLETED rather than racing the transient window. + * Before the fix, cancelTaskInternal() only handled SCHEDULED/WAITING/RUNNING and silently left a + * COMPLETED task enabled. + */ + @Test + @Tag("StateTests") + public void testCancelTaskWhileCompleted() throws Exception { + CountDownLatch executionLatch = new CountDownLatch(1); + + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "cancel-completed-test"; + } + + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) { + callback.complete(); + executionLatch.countDown(); + } + }; + + schedulerService.registerTaskExecutor(executor); + + ScheduledTask task = schedulerService.newTask("cancel-completed-test") + .asOneShot() + .schedule(); + + assertTrue(executionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should execute"); + + TestHelper.retryUntil( + () -> schedulerService.getTask(task.getItemId()), + t -> t != null && t.getStatus() == ScheduledTask.TaskStatus.COMPLETED + ); + + schedulerService.cancelTask(task.getItemId()); + + TestHelper.retryUntil( + () -> schedulerService.getTask(task.getItemId()), + t -> t != null && t.getStatus() == ScheduledTask.TaskStatus.CANCELLED + ); + + ScheduledTask cancelledTask = schedulerService.getTask(task.getItemId()); + assertEquals( + ScheduledTask.TaskStatus.CANCELLED, + cancelledTask.getStatus(), + "cancelTask() on a COMPLETED task must actually cancel it, not silently no-op"); + assertFalse(cancelledTask.isEnabled(), "Cancelled task should be disabled"); + } + /** * Tests task querying and filtering capabilities. */ diff --git a/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java b/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java index 87577ae7a..15771f9b7 100644 --- a/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java +++ b/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java @@ -86,9 +86,9 @@ public void testTracingOperations() { RequestTracer tracer = tracerService.getCurrentTracer(); // Start a root operation - tracer.startOperation("test", "Root operation", null); long startTime = System.currentTimeMillis(); - + tracer.startOperation("test", "Root operation", null); + // Add some traces tracer.trace("Test message", null); tracer.trace("Test with context", "context-data"); From 269729571c0bd86c3ee0d65c629b5ef8f8bd0fb6 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Tue, 30 Jun 2026 20:40:30 +0200 Subject: [PATCH 4/4] test: Add deterministic concurrency test for cache refresh lock Forces saveItem() and refreshTypeCache() to interleave via controlled latches instead of relying on timing-dependent integration tests to occasionally catch a regression in the per-item-type ReadWriteLock. --- .../AbstractMultiTypeCachingServiceTest.java | 90 +++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java index 73a1684a2..57923a771 100644 --- a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java +++ b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java @@ -16,6 +16,7 @@ */ package org.apache.unomi.services.common.cache; +import org.apache.unomi.api.ExecutionContext; import org.apache.unomi.api.Item; import org.apache.unomi.api.services.ExecutionContextManager; import org.apache.unomi.api.services.cache.CacheableTypeConfig; @@ -32,6 +33,9 @@ import java.io.Serializable; import java.util.*; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import static org.junit.Assert.*; @@ -110,6 +114,18 @@ public String toString() { } } + // Minimal Item subclass: saveItem()/removeItem() require , + // which the plain-Serializable TestSerializable above does not satisfy. + // Public (not just the ITEM_TYPE field): Item's reflective ITEM_TYPE lookup needs the + // declaring class itself to be accessible, not just the field's own modifiers. + public static class TestItem extends Item { + public static final String ITEM_TYPE = TEST_ITEM_TYPE; + + public TestItem(String itemId) { + super(itemId); + } + } + private static class TestCachingServiceImpl extends AbstractMultiTypeCachingService { private final Set> typeConfigs = new HashSet<>(); @@ -136,6 +152,11 @@ protected Set> getTypeConfigs() { return typeConfigs; } + // Exposes the protected saveItem() for the concurrency test below. + void callSaveItem(TestItem item) { + saveItem(item, Item::getItemId, TEST_ITEM_TYPE); + } + // Helper method to set a config as persistable for testing void makeConfigPersistable() { try { @@ -194,6 +215,9 @@ public void setUp() { runnable.run(); return null; }).when(contextManager).executeAsSystem(any(Runnable.class)); + + // saveItem() reads the current tenant from contextManager.getCurrentContext().getTenantId(). + when(contextManager.getCurrentContext()).thenReturn(new ExecutionContext(TEST_TENANT, null, null)); } @Test @@ -377,4 +401,70 @@ public void testRefreshCacheHandlesMultipleTenants() { // Verify items were removed from system tenant verify(cacheService).remove(eq(TEST_ITEM_TYPE), eq("item3"), eq(SYSTEM_TENANT), eq(TestSerializable.class)); } + + /** + * Validates the ReadWriteLock fix in AbstractMultiTypeCachingService: saveItem() must block while + * a refreshTypeCache() for the same item type holds the write lock, so a refresh's persistence read + * can never be followed by it overwriting a cache entry that a concurrent direct write just applied. + * Without the fix, saveItem() and refreshTypeCache() could interleave freely and the two race + * scenarios this test would otherwise need to get lucky on (read-before-write, write-after-read) + * would only manifest occasionally - here we force the interleaving deterministically instead. + */ + @Test + public void testSaveItemExcludedByConcurrentRefresh() throws Exception { + when(cacheService.getTenantCache(eq(TEST_TENANT), eq(TestSerializable.class))).thenReturn(new HashMap<>()); + when(cacheService.getTenantCache(eq(SYSTEM_TENANT), eq(TestSerializable.class))).thenReturn(new HashMap<>()); + doReturn(Collections.singleton(TEST_TENANT)).when(testCachingService).getTenants(); + + CacheableTypeConfig config = null; + for (CacheableTypeConfig typeConfig : testCachingService.getTypeConfigs()) { + if (typeConfig.getType().equals(TestSerializable.class)) { + @SuppressWarnings("unchecked") + CacheableTypeConfig typedConfig = (CacheableTypeConfig) typeConfig; + config = typedConfig; + break; + } + } + assertNotNull("Should find config for TestSerializable", config); + final CacheableTypeConfig finalConfig = config; + + CountDownLatch refreshHoldingLock = new CountDownLatch(1); + CountDownLatch releaseRefresh = new CountDownLatch(1); + AtomicBoolean refreshStillRunningWhenSaveReturned = new AtomicBoolean(true); + + // Stand-in for the persistence read inside refreshTypeCache(): signal that the refresh has + // entered its critical section (and therefore holds the write lock), then block until the + // test releases it. + doAnswer(invocation -> { + refreshHoldingLock.countDown(); + assertTrue("Test setup: refresh should be released within timeout", + releaseRefresh.await(5, TimeUnit.SECONDS)); + return Collections.emptyList(); + }).when(testCachingService).loadItemsForTenant(eq(TEST_TENANT), eq(finalConfig)); + + Thread refreshThread = new Thread(() -> testCachingService.refreshTypeCache(finalConfig)); + refreshThread.start(); + + assertTrue("Refresh should reach loadItemsForTenant while holding the write lock", + refreshHoldingLock.await(5, TimeUnit.SECONDS)); + + Thread saveThread = new Thread(() -> { + testCachingService.callSaveItem(new TestItem("item1")); + refreshStillRunningWhenSaveReturned.set(refreshThread.isAlive()); + }); + saveThread.start(); + + // saveItem()'s read lock should be blocked behind the refresh's write lock: give it a moment, + // then confirm it has NOT returned yet. + saveThread.join(500); + assertTrue("saveItem() should still be blocked while refresh holds the write lock", saveThread.isAlive()); + + releaseRefresh.countDown(); + refreshThread.join(5000); + saveThread.join(5000); + + assertFalse("saveThread should have finished", saveThread.isAlive()); + assertFalse("saveItem() must not proceed until the concurrent refresh released its write lock", + refreshStillRunningWhenSaveReturned.get()); + } }