This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch UNOMI-979-scheduler-lock-lease
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/UNOMI-979-scheduler-lock-lease
by this push:
new 4d0ba8be5 UNOMI-979: Rebase task counters on the store before a
terminal transition
4d0ba8be5 is described below
commit 4d0ba8be5715a7fcf9da36f504ec468a708ae607
Author: Serge Huber <[email protected]>
AuthorDate: Mon Aug 17 13:11:11 2026 +0200
UNOMI-979: Rebase task counters on the store before a terminal transition
CI reported SchedulerServiceImplTest.testMetricsAndHistory failing with
"Should
have 2 successful executions ==> expected: <2> but was: <1>": a periodic
task
recorded one success after two successful executions.
canCommitTerminalTransition() loads the authoritative document by id and
carries
only the OCC tokens onto the executing task instance. The counters and
execution
history stay as the dispatched copy had them - and that copy comes from
findEnabledScheduledOrWaitingTasks(), a search query that lags the store by
up to
the index refresh interval. The compare-and-set in persistTerminalState()
then
protects only the document version, not those values, so incrementing a
stale
base and CAS-writing it succeeds and silently discards the newer count. Two
successful executions dispatched from the same lagged view therefore both
write
successCount=1.
Accumulators are now rebased on the fresh read before the terminal handler
increments them: successCount, failureCount, and the append-only execution
history (taken from the store when it is ahead, so other statusDetails keys
such
as checkpoint and crash markers survive). Status, lock fields and
scheduling are
untouched - those belong to the execution's own outcome.
This is a data-integrity fix independent of the lock-lease change: success
and
failure counts, and the execution history a UI or operator reads, were
under-reported whenever a dispatch raced the refresh interval. It affects
Elasticsearch and OpenSearch equally, both having real refresh lag.
testTerminalCompletionRebasesCountersOnStoreValues pins it and is
mutation-validated: with the rebase removed it fails "expected: <2> but was:
<1>", the exact CI symptom, deterministically.
Also: testOneShotRetryBehavior's retry-delay assertion now reports the full
gap
sequence, execution count and persistence mode on failure instead of a bare
boolean. It is deliberately not relaxed - an execution landing sooner than
the
retry delay is a real contract violation. The suspected mechanism is the
same
staleness family on the dispatch side (prepareForExecution() checks due-ness
against the instance it is handed, so a lagged copy still carrying an
already
past nextScheduledExecution passes the check), but that is unproven and
fixing
it means changing the dispatch path, so the next occurrence is made decisive
rather than silenced.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---
.../impl/scheduler/TaskExecutionManager.java | 50 ++++++++++++++++++
.../impl/scheduler/SchedulerServiceImplTest.java | 32 ++++++++++--
.../impl/scheduler/TaskExecutionManagerTest.java | 60 ++++++++++++++++++++++
3 files changed, 137 insertions(+), 5 deletions(-)
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 cde83c97f..98b98a2e3 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
@@ -23,6 +23,8 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
@@ -668,9 +670,57 @@ public class TaskExecutionManager {
// Carry OCC tokens from the fresh load so persistTerminalState can
CAS.
TaskLockManager.copyOccMetadata(latest, task);
+ rebaseAccumulatorsFromStore(task, latest);
return true;
}
+ /**
+ * Rebases the running task's accumulating fields on the authoritative
store document before a
+ * terminal handler increments them.
+ * <p>
+ * The task instance a wrapper carries comes from the dispatch path, whose
discovery query
+ * ({@code findEnabledScheduledOrWaitingTasks}) is search-based and
therefore lags the store by
+ * up to the index refresh interval. Its {@code successCount}, {@code
failureCount} and
+ * execution history can predate writes that have already landed. The
compare-and-set in
+ * {@link #persistTerminalState} protects only the document
<em>version</em>, not these values:
+ * incrementing a stale base and then CAS-writing it succeeds and silently
loses the newer
+ * count. Observed as a periodic task reporting one success after two
successful executions.
+ * <p>
+ * Only accumulators are taken from the store. Status, lock fields and
scheduling are the
+ * terminal handler's business and are set from the execution's own
outcome.
+ *
+ * @param task the executing task instance about to be mutated by a
terminal handler
+ * @param latest the authoritative document, freshly loaded by id
+ */
+ private static void rebaseAccumulatorsFromStore(ScheduledTask task,
ScheduledTask latest) {
+ task.setSuccessCount(latest.getSuccessCount());
+ task.setFailureCount(latest.getFailureCount());
+
+ // Execution history is append-only, so the longer list is the more
current one. Other
+ // statusDetails keys stay as the execution left them (checkpoint
markers, crash details).
+ Map<String, Object> latestDetails = latest.getStatusDetails();
+ if (latestDetails == null) {
+ return;
+ }
+ Object latestHistory = latestDetails.get("executionHistory");
+ if (!(latestHistory instanceof List)) {
+ return;
+ }
+ Map<String, Object> details = task.getStatusDetails();
+ if (details == null) {
+ details = new HashMap<>();
+ task.setStatusDetails(details);
+ } else if (!(details instanceof HashMap)) {
+ details = new HashMap<>(details);
+ task.setStatusDetails(details);
+ }
+ Object ourHistory = details.get("executionHistory");
+ int ourSize = ourHistory instanceof List ? ((List<?>)
ourHistory).size() : 0;
+ if (((List<?>) latestHistory).size() > ourSize) {
+ details.put("executionHistory", new ArrayList<>((List<?>)
latestHistory));
+ }
+ }
+
/**
* Persists a terminal task state. Persistent tasks use compare-and-set so
a late
* complete/fail cannot clobber CANCELLED or a peer's RUNNING document.
Lock fields are
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 5f9e8ee38..1be4eea88 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
@@ -1050,12 +1050,34 @@ public class SchedulerServiceImplTest {
executionLatch.await(TEST_TIMEOUT, TimeUnit.MILLISECONDS),
"Task should complete all executions");
- // Verify retry delays
+ // Verify retry delays. Deliberately NOT relaxed: an execution landing
sooner than the
+ // retry delay means a retry attempt was dispatched early, which is a
real contract
+ // violation worth failing on. Suspected mechanism if this fires on CI
and not locally:
+ // prepareForExecution() checks due-ness against the task instance it
was handed, and the
+ // checker discovers tasks with a search query that lags the store, so
a stale copy still
+ // carrying the pre-retry (already past) nextScheduledExecution passes
the due check and
+ // executes immediately. Unproven - hence the diagnostics below rather
than a weakened
+ // assertion, so the next occurrence is decisive instead of just a
boolean.
for (int i = 1; i < executionTimes.size(); i++) {
- long delay = executionTimes.get(i) - executionTimes.get(i-1);
- assertTrue(
- delay >= TEST_RETRY_DELAY,
- "Retry delay should be at least " + TEST_RETRY_DELAY + "ms");
+ long delay = executionTimes.get(i) - executionTimes.get(i - 1);
+ if (delay < TEST_RETRY_DELAY) {
+ StringBuilder detail = new StringBuilder();
+ detail.append("Retry delay should be at least
").append(TEST_RETRY_DELAY)
+ .append("ms but execution #").append(i + 1).append(" came
").append(delay)
+ .append("ms after #").append(i)
+ .append(". persistent=").append(persistent)
+ .append(", executions=").append(executionTimes.size())
+ .append(" (expected ").append(TEST_MAX_RETRIES +
1).append("), gaps=[");
+ for (int j = 1; j < executionTimes.size(); j++) {
+ if (j > 1) {
+ detail.append(", ");
+ }
+ detail.append(executionTimes.get(j) - executionTimes.get(j
- 1)).append("ms");
+ }
+ detail.append("]. More executions than expected points at a
duplicate dispatch; "
+ + "the right count with a short gap points at an early
retry schedule.");
+ fail(detail.toString());
+ }
}
// Wait for the task to transition from RUNNING to COMPLETED state
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
index 9c045bf45..f81b9c6e4 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java
@@ -28,8 +28,12 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -91,6 +95,62 @@ public class TaskExecutionManagerTest {
executionManager.shutdown();
}
+ /**
+ * A terminal handler must increment counters from the STORE's values, not
from the possibly
+ * stale copy the wrapper is carrying.
+ * <p>
+ * The dispatch path discovers tasks with a search query, which lags the
store by up to the
+ * index refresh interval, so the executing instance can hold counters
that predate writes
+ * already committed. {@code persistTerminalState}'s compare-and-set
protects only the document
+ * version, so incrementing a stale base then CAS-writing it succeeds and
silently loses the
+ * newer count. Observed in CI as a periodic task reporting one success
after two successful
+ * executions ({@code SchedulerServiceImplTest.testMetricsAndHistory}).
+ */
+ @Test
+ public void testTerminalCompletionRebasesCountersOnStoreValues() throws
Exception {
+ CountDownLatch done = new CountDownLatch(1);
+ TaskExecutor executor = new TaskExecutor() {
+ @Override public String getTaskType() { return "stale-counters"; }
+ @Override public void execute(ScheduledTask task,
TaskStatusCallback callback) {
+ callback.complete();
+ done.countDown();
+ }
+ };
+
+ // What the wrapper carries: a search-lagged view that has not seen
the first success.
+ ScheduledTask stale = TaskTestFixtures.baseTask("stale-counters");
+ stale.setOneShot(false);
+ stale.setPeriod(60_000);
+ stale.setSuccessCount(0);
+ stale.setFailureCount(0);
+
+ // What the store actually holds: one success already recorded, with
its history entry.
+ ScheduledTask store = TaskTestFixtures.baseTask("stale-counters");
+ store.setItemId(stale.getItemId());
+ store.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ store.setExecutingNodeId(NODE);
+ store.setSuccessCount(1);
+ Map<String, Object> storeDetails = new HashMap<>();
+ List<Map<String, Object>> storeHistory = new ArrayList<>();
+ storeHistory.add(Collections.singletonMap("status", "SUCCESS"));
+ storeDetails.put("executionHistory", storeHistory);
+ store.setStatusDetails(storeDetails);
+ when(schedulerService.getTask(eq(stale.getItemId()),
eq(true))).thenReturn(store);
+
+ executionManager.executeTask(stale, executor);
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ awaitStatus(stale, ScheduledTask.TaskStatus.SCHEDULED, 5000);
+
+ assertEquals(2, stale.getSuccessCount(),
+ "the second success must count from the store's value (1), not the
stale copy's (0)");
+
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> history =
+ (List<Map<String, Object>>)
stale.getStatusDetails().get("executionHistory");
+ assertEquals(2, history.size(),
+ "history must extend the store's entries rather than restart from
the stale copy's");
+ }
+
@Test
public void testPrepareForExecutionRejectsDisabledAndWrongStatus() {
ScheduledTask disabled = TaskTestFixtures.baseTask("p");
