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
commit d3963edfcb50b3a40e479e784f5e30e4b96739c4 Author: Serge Huber <[email protected]> AuthorDate: Mon Aug 17 10:26:21 2026 +0200 UNOMI-979: Judge scheduler lock expiry by the owner's recorded lease A lock's renewal cadence is derived from its owner's configured lock timeout (lockTimeout/3), but expiry was judged against the OBSERVER's timeout. A node configured with a shorter timeout than a peer's renewal cadence therefore saw every renewal gap as a dead lock: it marked the live execution CRASHED, cleared the lock, and the next peer tick re-dispatched the task while the original execution was still running. Reproduced deterministically with a 1s-timeout observer against 10s-timeout workers; in production the same double execution follows from configuration drift or a rolling upgrade. It was also the root cause of the CI flakes in SchedulerServiceImplTest ("expected: <1> but was: <2>"). Locks now record the lease the owner granted itself (ScheduledTask lockLeaseMillis, stamped on every acquire and renewal, cleared on release), and isLockExpired() judges against that lease. Documents written before lease recording carry no lease and fall back to the observer's timeout - the exact pre-change behaviour - so a rolling upgrade changes nothing for existing locks. Corrupt (negative) leases fall back the same way rather than widening or wedging the lock; an absurdly large lease is honoured, because the owner declared it and stealing early is what causes double execution. Also in this change: - ScheduledTask now tolerates unknown JSON properties. Jackson's default rejects the first unrecognized field, so during a rolling upgrade an older node would lose the ability to read ANY task document a newer node had written the moment a field is added - this field or any future one. - startLockRenewal() warns when the configured lock timeout is at or below the minimum renewal interval, the one configuration where a node cannot keep its own lease alive and peers may legitimately recover its live work. - ES and OpenSearch scheduledTask mappings gain the lockLeaseMillis field. The divergent-timeout steal is pinned end to end by testShortTimeoutObserverCannotRecoverLiveRenewedLock (fails as "expected: <RUNNING> but was: <CRASHED>" with the fix reverted), the recovery direction by testDeadOwnersShortLeaseDrivesPromptRecoveryByPatientSurvivor (a patient survivor recovers a dead owner's task as soon as the OWNER's lease expires - faster than before, and the case that proves failover still works). Unit coverage in TaskLockManagerTest exercises lease stamping on all three acquire paths, re-stamping on renewal after a runtime timeout change, clearing on release, both override directions, the legacy/corrupt fallbacks, boundary equality and overflow. ScheduledTaskLeaseSerializationTest pins the persistence format through both real read paths, the legacy-document fallback, and the newer-version-document case (fails with UnrecognizedPropertyException without the annotation). Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../org/apache/unomi/api/tasks/ScheduledTask.java | 36 +++- .../META-INF/cxs/mappings/scheduledTask.json | 3 + .../META-INF/cxs/mappings/scheduledTask.json | 3 + .../impl/scheduler/SchedulerServiceImpl.java | 3 + .../impl/scheduler/TaskExecutionManager.java | 12 ++ .../services/impl/scheduler/TaskLockManager.java | 29 +++- .../impl/scheduler/TaskRecoveryManager.java | 2 + .../services/impl/scheduler/TaskStateManager.java | 2 + .../ScheduledTaskLeaseSerializationTest.java | 124 ++++++++++++++ .../scheduler/SchedulerServiceClusterRaceTest.java | 144 ++++++++++++++++ .../impl/scheduler/TaskLockManagerTest.java | 182 +++++++++++++++++++++ 11 files changed, 535 insertions(+), 5 deletions(-) diff --git a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java index c5d698e77..5a2f52ace 100644 --- a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java +++ b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java @@ -16,6 +16,7 @@ */ package org.apache.unomi.api.tasks; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import org.apache.unomi.api.Item; import java.io.Serializable; @@ -40,6 +41,11 @@ import java.util.HashSet; * @see org.apache.unomi.api.services.SchedulerService * @see TaskExecutor */ +// Tolerate unknown properties so a node running THIS version can still deserialize task +// documents written by a NEWER version that has added fields (rolling upgrade window). +// Without this, Jackson's default rejects the first unrecognized field and the older node +// loses access to all scheduler state until it is upgraded. +@JsonIgnoreProperties(ignoreUnknown = true) public class ScheduledTask extends Item implements Serializable { /** @@ -86,6 +92,7 @@ public class ScheduledTask extends Item implements Serializable { private boolean enabled; private String lockOwner; private Date lockDate; + private long lockLeaseMillis; private boolean oneShot; private boolean allowParallelExecution; private TaskStatus status; @@ -343,13 +350,40 @@ public class ScheduledTask extends Item implements Serializable { /** * Sets the date when the current lock was acquired. - * + * * @param lockDate the lock acquisition date */ public void setLockDate(Date lockDate) { this.lockDate = lockDate; } + /** + * Duration in milliseconds for which the current lock is valid, as declared by the node that + * acquired or last renewed it. + * <p> + * A lock's lifetime is a lease granted by its <em>owner</em>: the owner renews it on a cadence + * derived from its own configured lock timeout, so only the owner's timeout describes when a + * missing renewal actually means the owner is dead. Observers must judge expiry against this + * recorded lease, never against their own configured timeout — a node configured with a shorter + * timeout than the owner's renewal cadence would otherwise "recover" a lock whose owner is alive + * and mid-execution, and the task would run twice. + * + * @return the lease duration in milliseconds, or {@code 0} when the lock predates lease + * recording (legacy documents) and the observer's own timeout is the only guide + */ + public long getLockLeaseMillis() { + return lockLeaseMillis; + } + + /** + * Sets the lease duration granted with the current lock. + * + * @param lockLeaseMillis the lease duration in milliseconds, {@code 0} when unlocked or unknown + */ + public void setLockLeaseMillis(long lockLeaseMillis) { + this.lockLeaseMillis = lockLeaseMillis; + } + /** * Determines whether this task should execute only once. * Tasks with period=0 are automatically marked as one-shot tasks. diff --git a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json index f36fc297c..030305e8e 100644 --- a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json +++ b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json @@ -75,6 +75,9 @@ "lockDate": { "type": "date" }, + "lockLeaseMillis": { + "type": "long" + }, "lastExecutionDate": { "type": "date" }, diff --git a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json index 9c1541d96..a251eebf4 100644 --- a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json +++ b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json @@ -78,6 +78,9 @@ "lockDate": { "type": "date" }, + "lockLeaseMillis": { + "type": "long" + }, "lastExecutionDate": { "type": "date" }, 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 3c98963fc..5e84267da 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 @@ -493,6 +493,7 @@ public class SchedulerServiceImpl implements SchedulerService { if (newStatus == TaskStatus.COMPLETED || newStatus == TaskStatus.FAILED) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); task.setWaitingForTaskType(null); task.setCurrentStep(null); // Update last execution date for completed/failed tasks @@ -511,6 +512,7 @@ public class SchedulerServiceImpl implements SchedulerService { } else if (newStatus == TaskStatus.WAITING) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); } else if (newStatus == TaskStatus.RUNNING) { // Update status details for running tasks Map<String, Object> details = task.getStatusDetails(); @@ -899,6 +901,7 @@ public class SchedulerServiceImpl implements SchedulerService { // and PersistenceSchedulerProvider.preDestroy need not unlock RUNNING. task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); if (task.isPersistent() && persistenceProvider != null) { if (!persistenceProvider.saveTask(task)) { LOGGER.warn("Failed to persist CRASHED state for task {} during shutdown; " 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 8f3bd41a4..cde83c97f 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 @@ -502,6 +502,7 @@ public class TaskExecutionManager { task.setExecutingNodeId(null); task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); schedulerService.saveTask(task, true); } catch (Exception e) { LOGGER.warn("Failed to abort prepared task {} during shutdown: {}", @@ -532,6 +533,16 @@ public class TaskExecutionManager { return; } long interval = Math.max(MIN_LOCK_RENEWAL_INTERVAL_MS, lockManager.getLockTimeout() / 3); + if (interval >= lockManager.getLockTimeout()) { + // The renewal floor exceeds the configured timeout, so this node cannot renew its own + // lease fast enough to keep it alive: peers may legitimately treat its live locks as + // expired between two renewals and recover mid-execution tasks. Surface the + // misconfiguration instead of leaving sporadic double executions to be diagnosed. + LOGGER.warn("Lock timeout {}ms is at or below the minimum renewal interval {}ms: " + + "this node's live locks can expire between renewals and be recovered by peers. " + + "Configure a lock timeout of at least {}ms.", + lockManager.getLockTimeout(), interval, MIN_LOCK_RENEWAL_INTERVAL_MS * 3); + } LockRenewalHandle handle = new LockRenewalHandle(); activeLockRenewals.put(task.getItemId(), handle); try { @@ -668,6 +679,7 @@ public class TaskExecutionManager { private boolean persistTerminalState(ScheduledTask task) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); if (!task.isPersistent()) { boolean saved = schedulerService.saveTask(task); if (!saved) { diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java index 978d97fa6..fbe1517c0 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java +++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java @@ -161,6 +161,7 @@ public class TaskLockManager { // Just set lock info but don't enforce exclusivity task.setLockOwner(nodeId); task.setLockDate(new Date()); + task.setLockLeaseMillis(lockTimeout); metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED); return true; } @@ -194,8 +195,10 @@ public class TaskLockManager { latest.setLockOwner(nodeId); latest.setLockDate(new Date()); + latest.setLockLeaseMillis(lockTimeout); task.setLockOwner(nodeId); task.setLockDate(latest.getLockDate()); + task.setLockLeaseMillis(lockTimeout); metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED); // For non-persistent tasks, we just update the in-memory map @@ -247,9 +250,12 @@ public class TaskLockManager { task.setSystemMetadata(SEQ_NO, latestTask.getSystemMetadata(SEQ_NO)); task.setSystemMetadata(PRIMARY_TERM, latestTask.getSystemMetadata(PRIMARY_TERM)); - // Step 6: Set lock information + // Step 6: Set lock information. The lease records THIS node's timeout with the lock: + // renewal cadence is derived from the owner's timeout, so only the owner's timeout says + // when a missing renewal means the owner is dead (see isLockExpired()). task.setLockOwner(nodeId); task.setLockDate(new Date()); + task.setLockLeaseMillis(lockTimeout); LOGGER.debug("LOCK-DIAG [{}] node {} : attempting CAS write - if_seq_no={}, if_primary_term={}, " + "writing lockOwner={}", @@ -391,6 +397,7 @@ public class TaskLockManager { if (latestOwner == null) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); LOGGER.debug("LOCK-DIAG [{}] node {} : releaseLock() no-op, store already unlocked", task.getItemId(), nodeId); return true; @@ -406,8 +413,10 @@ public class TaskLockManager { toSave.setLockOwner(null); toSave.setLockDate(null); + toSave.setLockLeaseMillis(0); task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); // Compare-and-set on the freshly loaded seq_no/primary_term, not a blind overwrite: // a peer may win a legitimate CAS-based lock acquisition in the window between our @@ -474,6 +483,7 @@ public class TaskLockManager { } latest.setLockDate(new Date()); + latest.setLockLeaseMillis(lockTimeout); // Compare-and-set on the fresh store view: if a peer stole the lock between the // read above and this write, renewal fails closed instead of resurrecting our lock. @@ -486,6 +496,7 @@ public class TaskLockManager { // the executing thread's later compare-and-set writes are checked against the // store's current version, not the pre-renewal one. task.setLockDate(latest.getLockDate()); + task.setLockLeaseMillis(latest.getLockLeaseMillis()); copyOccMetadata(latest, task); LOGGER.debug("LOCK-DIAG [{}] node {} : renewLock() succeeded, new lockDate={}", task.getItemId(), nodeId, latest.getLockDate()); @@ -537,12 +548,22 @@ public class TaskLockManager { return true; } + // Judge expiry against the lease the OWNER recorded with the lock, not this node's own + // configured timeout. The owner renews on a cadence derived from its own timeout + // (lockTimeout/3, see TaskExecutionManager#startLockRenewal), so a node configured with a + // shorter timeout than the owner's renewal cadence would otherwise declare a live, + // renewed lock dead in the gap between two renewals and "recover" a task that is still + // executing — observed as double execution under divergent per-node configuration. + // Locks written before lease recording carry no lease (0); only for those does this + // node's own timeout remain the best available guess. + long lease = task.getLockLeaseMillis() > 0 ? task.getLockLeaseMillis() : lockTimeout; long now = System.currentTimeMillis(); long lockAge = now - task.getLockDate().getTime(); - boolean expired = lockAge > lockTimeout; + boolean expired = lockAge > lease; LOGGER.debug("LOCK-DIAG isLockExpired() : task={}, lockDate={} ({}), now={}, lockAge={}ms, " - + "lockTimeout={}ms -> expired={}", - task.getItemId(), task.getLockDate(), task.getLockDate().getTime(), now, lockAge, lockTimeout, expired); + + "lease={}ms (recorded={}ms, own timeout={}ms) -> expired={}", + task.getItemId(), task.getLockDate(), task.getLockDate().getTime(), now, lockAge, + lease, task.getLockLeaseMillis(), lockTimeout, expired); return expired; } } 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 43b29e368..d41c34c30 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 @@ -255,6 +255,7 @@ public class TaskRecoveryManager { } latest.setLockOwner(null); latest.setLockDate(null); + latest.setLockLeaseMillis(0); // Record the crash in execution history recordCrash(latest, previousOwner); @@ -273,6 +274,7 @@ public class TaskRecoveryManager { task.setStatus(latest.getStatus()); task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); task.setStatusDetails(latest.getStatusDetails()); task.setCurrentStep(latest.getCurrentStep()); task.setLastError(latest.getLastError()); 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 c8877bd72..d8bf91266 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 @@ -148,6 +148,7 @@ public class TaskStateManager { private void clearTaskExecution(ScheduledTask task) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); task.setWaitingForTaskType(null); task.setCurrentStep(null); } @@ -162,6 +163,7 @@ public class TaskStateManager { private void clearLockInfo(ScheduledTask task) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); } private void updateRunningState(ScheduledTask task, String nodeId) { diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java new file mode 100644 index 000000000..126cee701 --- /dev/null +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.services.impl.scheduler; + +import org.apache.unomi.api.Item; +import org.apache.unomi.api.tasks.ScheduledTask; +import org.apache.unomi.persistence.spi.CustomObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Date; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Persistence-format coverage for {@link ScheduledTask#getLockLeaseMillis()}. + * <p> + * The lock lease is a cross-node security decision (it decides who may declare a peer dead), so + * its survival through the production serializer is not an implementation detail: a field that + * silently fails to round-trip would degrade every observer to the legacy observer-timeout + * fallback and quietly reintroduce the divergent-timeout double-execution bug. Both store read + * paths are exercised: direct class binding, and the {@code Item}-dispatched path the persistence + * services actually use ({@code readValue(json, Item.class)} via {@code ItemDeserializer}). + */ +public class ScheduledTaskLeaseSerializationTest { + + private CustomObjectMapper mapper; + + @BeforeEach + public void setUp() { + mapper = CustomObjectMapper.getCustomInstance(); + mapper.registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE, ScheduledTask.class); + } + + private ScheduledTask lockedTask() { + ScheduledTask task = new ScheduledTask(); + task.setItemId("lease-serialization-test"); + task.setTaskType("lease-serialization-test"); + task.setStatus(ScheduledTask.TaskStatus.RUNNING); + task.setLockOwner("node-a"); + task.setLockDate(new Date()); + task.setLockLeaseMillis(12345); + return task; + } + + @Test + public void leaseSurvivesRoundTripViaDirectClassBinding() throws Exception { + String json = mapper.writeValueAsString(lockedTask()); + assertTrue(json.contains("\"lockLeaseMillis\":12345"), "lease must be serialized: " + json); + + ScheduledTask back = mapper.readValue(json, ScheduledTask.class); + assertEquals(12345, back.getLockLeaseMillis()); + assertEquals("node-a", back.getLockOwner()); + } + + @Test + public void leaseSurvivesRoundTripViaItemDispatchedPath() throws Exception { + // This is the path the persistence services use when loading store documents. + String json = mapper.writeValueAsString(lockedTask()); + Item item = mapper.readValue(json, Item.class); + assertTrue(item instanceof ScheduledTask, "itemType dispatch should yield a ScheduledTask"); + assertEquals(12345, ((ScheduledTask) item).getLockLeaseMillis()); + } + + /** + * A document written BEFORE lease recording (no {@code lockLeaseMillis} field) must load with + * lease 0, which {@code TaskLockManager#isLockExpired} treats as "fall back to the observer's + * own timeout" — i.e. exactly the pre-lease behaviour, so a rolling upgrade cannot make old + * locks unexpirable or instantly expired. + */ + @Test + public void legacyDocumentWithoutLeaseLoadsAsZero() throws Exception { + String legacyJson = "{" + + "\"itemId\":\"legacy-task\"," + + "\"itemType\":\"scheduledTask\"," + + "\"taskType\":\"legacy-task\"," + + "\"status\":\"RUNNING\"," + + "\"lockOwner\":\"old-node\"," + + "\"lockDate\":\"2026-01-01T00:00:00Z\"" + + "}"; + Item item = mapper.readValue(legacyJson, Item.class); + ScheduledTask task = (ScheduledTask) item; + assertEquals(0, task.getLockLeaseMillis(), "missing lease must read as 0 (legacy fallback)"); + assertEquals("old-node", task.getLockOwner()); + } + + /** + * A document written by a NEWER version carrying a field this version does not know must + * still deserialize (rolling upgrade window: older binaries keep reading scheduler state + * written by upgraded peers). Pinned by {@code @JsonIgnoreProperties(ignoreUnknown = true)} + * on ScheduledTask — without it, Jackson's default rejects the first unknown field and the + * older node loses access to every task document the newer node has touched. + */ + @Test + public void documentFromNewerVersionWithUnknownFieldStillLoads() throws Exception { + String futureJson = "{" + + "\"itemId\":\"future-task\"," + + "\"itemType\":\"scheduledTask\"," + + "\"taskType\":\"future-task\"," + + "\"status\":\"SCHEDULED\"," + + "\"lockLeaseMillis\":5000," + + "\"someFieldAddedInAFutureVersion\":\"whatever\"" + + "}"; + Item item = mapper.readValue(futureJson, Item.class); + assertNotNull(item); + assertEquals(5000, ((ScheduledTask) item).getLockLeaseMillis()); + } +} diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java index 62ae062bd..c967de791 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java @@ -331,6 +331,150 @@ public class SchedulerServiceClusterRaceTest { assertEquals(1, executors.size(), "Exactly one node should have executed"); } + /** + * A node configured with a SHORTER lock timeout than a peer must not "recover" that peer's + * live, renewed lock. + * <p> + * The owner renews its lock every {@code lockTimeout/3} — a cadence derived from its OWN + * timeout. Before lock leases were recorded ({@link ScheduledTask#getLockLeaseMillis()}), + * expiry was judged against the <em>observer's</em> timeout, so an observer whose timeout was + * shorter than the owner's renewal cadence saw every renewal gap as an expired lock: it marked + * the live execution CRASHED and cleared the lock, and the next peer tick re-dispatched the + * task while the original execution was still running. This reproduced deterministically as + * {@code maxConcurrent=2} with a 1s-timeout observer against 10s-timeout workers, and is also a + * production hazard under config drift or rolling upgrades. The recorded lease makes expiry + * owner-relative, so the divergent observer becomes harmless. + */ + @Test + public void testShortTimeoutObserverCannotRecoverLiveRenewedLock() throws Exception { + SchedulerServiceImpl worker1 = createNode("lease-worker1", true, 10000); + SchedulerServiceImpl worker2 = createNode("lease-worker2", true, 10000); + // Divergent config: this node judges everything with a 500ms timeout. It registers no + // executor for the task type, so any double execution must come via a worker re-dispatch. + SchedulerServiceImpl watchdog = createNode("lease-watchdog", true, 500); + seedActiveNodes("lease-worker1", "lease-worker2", "lease-watchdog"); + + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger executions = new AtomicInteger(0); + + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "lease-liveness-test"; + } + + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) throws Exception { + executions.incrementAndGet(); + started.countDown(); + assertTrue(release.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)); + callback.complete(); + } + }; + worker1.registerTaskExecutor(executor); + worker2.registerTaskExecutor(executor); + + ScheduledTask task = worker1.newTask("lease-liveness-test") + .disallowParallelExecution() + .asOneShot() + .schedule(); + + assertTrue(started.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS), "One worker should start the task"); + + // Let the lock age past the watchdog's 500ms timeout while staying far inside the owner's + // 10s lease (the owner's renewal cadence is 10s/3, so the age check below cannot be + // satisfied by a renewal racing us — any observed age > 600ms is a genuine renewal gap). + long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS; + while (System.currentTimeMillis() < deadline) { + ScheduledTask stored = persistenceService.load(task.getItemId(), ScheduledTask.class); + if (stored != null && stored.getLockDate() != null + && System.currentTimeMillis() - stored.getLockDate().getTime() > 600) { + break; + } + Thread.sleep(50); + } + + // Force the divergent observer's recovery pass repeatedly — the deterministic version of + // the background tick that used to steal the lock. + for (int i = 0; i < 3; i++) { + watchdog.recoverCrashedTasks(); + } + + ScheduledTask observed = persistenceService.load(task.getItemId(), ScheduledTask.class); + assertEquals(ScheduledTask.TaskStatus.RUNNING, observed.getStatus(), + "A live, renewed lock must not be marked CRASHED by a shorter-timeout observer"); + assertNotNull(observed.getLockOwner(), "The owner's lock must not be cleared"); + + release.countDown(); + + ScheduledTask done = waitForStatus(worker1, task.getItemId(), ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS); + assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus()); + assertEquals(1, executions.get(), + "The task must execute exactly once despite the divergent-timeout observer"); + } + + /** + * The recovery-enabling direction of lease-based expiry: a genuinely DEAD owner must still be + * recovered, and the moment that happens is decided by the lease the dead owner recorded, not + * by the survivor's own (here much longer) timeout. This is the guarantee that keeps crash + * failover working after the lease change — and it is now faster when the dead node ran with + * a short timeout, because peers no longer wait out their own longer opinion. + */ + @Test + public void testDeadOwnersShortLeaseDrivesPromptRecoveryByPatientSurvivor() throws Exception { + SchedulerServiceImpl survivor = createNode("lease-survivor", true, 30_000); + seedActiveNodes("lease-survivor"); + + CountDownLatch recovered = new CountDownLatch(1); + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "dead-owner-lease-test"; + } + + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) { + recovered.countDown(); + callback.complete(); + } + }; + survivor.registerTaskExecutor(executor); + + // Manufacture what a crashed node leaves behind: RUNNING, locked, lease recorded from a + // short timeout, and silent (no renewal will ever come). lockDate is backdated past the + // lease so the very first recovery pass can act. + ScheduledTask ghost = new ScheduledTask(); + ghost.setItemId("ghost-owned-task"); + ghost.setTaskType("dead-owner-lease-test"); + ghost.setEnabled(true); + ghost.setPersistent(true); + ghost.setOneShot(true); + ghost.setStatus(ScheduledTask.TaskStatus.RUNNING); + ghost.setExecutingNodeId("ghost-node"); + ghost.setLockOwner("ghost-node"); + ghost.setLockDate(new Date(System.currentTimeMillis() - 2000)); + ghost.setLockLeaseMillis(500); + persistenceService.save(ghost); + persistenceService.refreshIndex(ScheduledTask.class); + persistenceService.refresh(); + + // Force recovery passes rather than waiting for background ticks. The survivor's own + // timeout is 30s: pre-lease it would have refused to touch this lock for 30s, and this + // latch (10s) would time out. The recorded 500ms lease is what lets it act now. + long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS; + while (recovered.getCount() > 0 && System.currentTimeMillis() < deadline) { + survivor.recoverCrashedTasks(); + recovered.await(250, TimeUnit.MILLISECONDS); + } + + assertTrue(recovered.getCount() == 0, + "a patient survivor must recover a dead owner's task as soon as the OWNER's lease expires"); + ScheduledTask done = waitForStatus(survivor, "ghost-owned-task", ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS); + assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus(), + "the recovered task must run to completion on the survivor"); + } + @Test public void testAffinityOpenFieldAfterBackupWindowsWhenPrimaryDead() throws Exception { SchedulerServiceImpl backup1 = createNode("aff-backup1", true, 10000); diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java index 249783884..25fe94bbc 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java @@ -216,6 +216,188 @@ public class TaskLockManagerTest { assertFalse(lockManager.isLockExpired(task)); } + /** + * Expiry must be judged against the lease the OWNER recorded with the lock, not this + * observer's own timeout. An observer configured shorter than the owner's renewal cadence + * would otherwise "recover" a live, renewed lock between two renewals and double-run the + * task (see SchedulerServiceClusterRaceTest#testShortTimeoutObserverCannotRecoverLiveRenewedLock + * for the end-to-end version). + */ + @Test + public void testIsLockExpiredHonoursRecordedLeaseOverObserverTimeout() { + ScheduledTask task = TaskTestFixtures.baseTask("lease"); + task.setLockDate(new Date(System.currentTimeMillis() - 5000)); + + // 5s-old lock, observer timeout 1s: expired by observer maths, but the owner granted 10s. + task.setLockLeaseMillis(10000); + assertFalse(lockManager.isLockExpired(task), + "a lock inside its owner-recorded lease must not expire under a shorter observer timeout"); + + // The reverse also holds: an owner that granted itself a SHORT lease is expired even + // when the observer's own timeout would still consider it live. + task.setLockDate(new Date(System.currentTimeMillis() - 500)); + task.setLockLeaseMillis(100); + assertTrue(lockManager.isLockExpired(task), + "a lock past its owner-recorded lease is expired regardless of the observer timeout"); + } + + /** Locks written before lease recording (lease 0) fall back to the observer's own timeout. */ + @Test + public void testIsLockExpiredFallsBackToObserverTimeoutForLegacyLocks() { + ScheduledTask task = TaskTestFixtures.baseTask("legacy"); + task.setLockDate(new Date(System.currentTimeMillis() - 5000)); + task.setLockLeaseMillis(0); + assertTrue(lockManager.isLockExpired(task)); + + task.setLockDate(new Date()); + assertFalse(lockManager.isLockExpired(task)); + } + + /** A corrupt negative lease must not wedge or widen the lock: treat it like a legacy lock. */ + @Test + public void testIsLockExpiredNegativeLeaseFallsBackToObserverTimeout() { + ScheduledTask task = TaskTestFixtures.baseTask("corrupt"); + task.setLockDate(new Date(System.currentTimeMillis() - 5000)); + task.setLockLeaseMillis(-1); + assertTrue(lockManager.isLockExpired(task), "negative lease + old lock: observer timeout applies"); + + task.setLockDate(new Date()); + assertFalse(lockManager.isLockExpired(task), "negative lease + fresh lock: observer timeout applies"); + } + + /** Boundary parity with the observer-timeout path: age == lease is NOT yet expired. */ + @Test + public void testIsLockExpiredFalseWhenAgeEqualsRecordedLease() { + ScheduledTask task = TaskTestFixtures.baseTask("edge"); + task.setLockLeaseMillis(2000); + task.setLockDate(new Date(System.currentTimeMillis() - 2000)); + assertFalse(lockManager.isLockExpired(task)); + } + + /** + * An absurd lease (misconfigured or corrupt owner) must not overflow the arithmetic. The lock + * is honoured as unexpired — the owner declared it, and stealing it risks double execution; + * a genuinely wedged task from a dead misconfigured node is an operator decision, not one a + * peer may take unilaterally with a shorter opinion. + */ + @Test + public void testIsLockExpiredHugeLeaseIsHonouredWithoutOverflow() { + ScheduledTask task = TaskTestFixtures.baseTask("huge"); + task.setLockLeaseMillis(Long.MAX_VALUE); + task.setLockDate(new Date(System.currentTimeMillis() - 100_000)); + assertFalse(lockManager.isLockExpired(task)); + } + + // ------------------------------------------------------------------ lease stamping + // Every path that writes a lock must record the owner's lease with it, and every path that + // clears a lock must clear the lease: a cleared owner with a leftover lease (or the reverse) + // would make expiry decisions against a lock that no longer exists. + + @Test + public void testParallelAcquireStampsLease() { + ScheduledTask task = TaskTestFixtures.baseTask("parallel-lease"); + task.setAllowParallelExecution(true); + assertTrue(lockManager.acquireLock(task)); + assertEquals(1000, task.getLockLeaseMillis(), "parallel marker must record the owner's lease"); + } + + @Test + public void testInMemoryAcquireStampsLease() { + ScheduledTask task = TaskTestFixtures.baseTask("mem-lease"); + task.setPersistent(false); + assertTrue(lockManager.acquireLock(task)); + assertEquals(1000, task.getLockLeaseMillis(), "in-memory lock must record the owner's lease"); + } + + @Test + public void testDistributedAcquireStampsLease() { + ScheduledTask task = TaskTestFixtures.baseTask("dist-lease"); + task.setNextScheduledExecution(new Date(System.currentTimeMillis() - 10_000)); + ScheduledTask latest = TaskTestFixtures.baseTask("dist-lease"); + latest.setItemId(task.getItemId()); + latest.setSystemMetadata("seq_no", 3L); + latest.setSystemMetadata("primary_term", 1L); + when(schedulerService.getTask(task.getItemId())).thenReturn(latest); + when(schedulerService.saveTaskWithRefresh(any(ScheduledTask.class))).thenReturn(true); + + assertTrue(lockManager.acquireLock(task)); + assertEquals(1000, task.getLockLeaseMillis(), "distributed lock must record the owner's lease"); + } + + /** + * Renewal re-stamps the lease from the owner's CURRENT timeout, so a runtime configuration + * change (ConfigAdmin update) propagates to the store within one renewal interval instead of + * peers judging against a stale grant for the rest of the execution. + */ + @Test + public void testRenewLockRestampsLeaseFromCurrentTimeout() { + ScheduledTask task = TaskTestFixtures.baseTask("renew-lease"); + task.setLockOwner(NODE); + task.setLockDate(new Date()); + task.setLockLeaseMillis(1000); + + ScheduledTask storeView = TaskTestFixtures.baseTask("renew-lease"); + storeView.setItemId(task.getItemId()); + storeView.setLockOwner(NODE); + storeView.setLockDate(task.getLockDate()); + storeView.setLockLeaseMillis(1000); + when(schedulerService.getTask(task.getItemId())).thenReturn(storeView); + when(schedulerService.saveTaskWithRefresh(storeView)).thenReturn(true); + + lockManager.setLockTimeout(5000); + assertTrue(lockManager.renewLock(task)); + assertEquals(5000, storeView.getLockLeaseMillis(), "store must carry the current lease"); + assertEquals(5000, task.getLockLeaseMillis(), "caller's view must be synced to the current lease"); + } + + @Test + public void testReleaseLockClearsLease() { + ScheduledTask task = TaskTestFixtures.baseTask("release-lease"); + task.setLockOwner(NODE); + task.setLockDate(new Date()); + task.setLockLeaseMillis(1000); + + ScheduledTask stored = TaskTestFixtures.baseTask("release-lease"); + stored.setItemId(task.getItemId()); + stored.setLockOwner(NODE); + stored.setLockDate(task.getLockDate()); + stored.setLockLeaseMillis(1000); + when(schedulerService.getTask(eq(task.getItemId()), eq(true))).thenReturn(stored); + + assertTrue(lockManager.releaseLock(task)); + assertEquals(0, task.getLockLeaseMillis(), "release must clear the caller's lease"); + assertEquals(0, stored.getLockLeaseMillis(), "release must clear the persisted lease"); + } + + /** + * The recovery-enabling direction: a dead owner that granted itself a SHORT lease is + * recoverable by an observer configured with a much longer timeout — the observer must not + * impose its own, slower opinion on a lock whose owner promised to renew far sooner. + */ + @Test + public void testNonOwnerCanReleaseLockPastItsShortRecordedLease() { + lockManager.setLockTimeout(60_000); // observer is very patient by its own config + + ScheduledTask stored = TaskTestFixtures.baseTask("dead-short-lease"); + stored.setLockOwner("dead-node"); + stored.setLockDate(new Date(System.currentTimeMillis() - 2000)); + stored.setLockLeaseMillis(500); // owner promised renewal every ~166ms and is silent for 2s + + ScheduledTask callerView = TaskTestFixtures.baseTask("dead-short-lease"); + callerView.setItemId(stored.getItemId()); + callerView.setLockOwner("dead-node"); + callerView.setLockDate(stored.getLockDate()); + callerView.setLockLeaseMillis(500); + when(schedulerService.getTask(eq(callerView.getItemId()), eq(true))).thenReturn(stored); + + assertTrue(lockManager.isLockExpired(callerView), + "a lock silent past its own lease is expired even for a patient observer"); + assertTrue(lockManager.releaseLock(callerView), + "recovery must be able to clear a dead owner's expired-by-lease lock"); + assertNull(stored.getLockOwner()); + assertEquals(0, stored.getLockLeaseMillis()); + } + @Test public void testAffinityBlocksBackupDuringPrimaryWindow() { List<String> nodes = Arrays.asList("aaa-node", NODE, "zzz-node");
