diff --git a/README.md b/README.md index 1b522f555..6cab53e5e 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,7 @@ It works seamlessly across AEM on-premise, AMS, and AEMaaCS environments. - [Repo example](#repo-example) - [Abortable example](#abortable-example) - [Script documentation](#script-documentation) + - [Locking](#locking) - [History](#history) - [Extension scripts](#extension-scripts) - [Example extension script](#example-extension-script) @@ -669,6 +670,38 @@ For complete examples, see the [example scripts directory](ui.content.example/sr ACM Script Metadata +#### Locking + +While a script runs, ACM holds a repository lock scoped to that script, so the same script cannot run twice at the same time (e.g. a scheduled run starting before the previous one finished, two cluster nodes firing together, or the script triggered manually twice). A second attempt while the lock is held ends with the `LOCKED` status instead of running the code again. + +To avoid a lock being held forever when an instance is killed abruptly (e.g. pod recycling, `kill -9`, a crash) — leaving a *stale lock* that would block every future run — the lock carries an expiration time. Once it expires, the next run recreates it and proceeds normally. The default expiration comes from the global `Lock Timeout` OSGi configuration of the *Code Executor* (24 hours). + +A single global value rarely fits every script, so it can be overridden per script by setting `context.lockTimeout` in `describeRun()`. This is handy when: + +- **The script runs longer than the global default** — the timeout must always exceed the maximum expected execution time, otherwise the lock could expire mid-run and let a second run start concurrently. Long-running migrations or reports should raise it. +- **The script is short and should recover fast** — lowering it means a stale lock left by a crash is cleared sooner, so the script becomes runnable again quickly. +- **Expiration should be disabled entirely** — for a critical singleton where accidental concurrency is worse than a stuck lock, set `0` or a negative value (the lock then only clears on a normal finish or a manual reset). + +```groovy +import java.time.Duration + +void describeRun() { + context.lockTimeout = "2h" // human-readable: 'ms', 's', 'm', 'h', 'd' (e.g. '2h30m', '500ms') + // context.lockTimeout = "PT2H30M" // ISO-8601 duration + // context.lockTimeout = Duration.ofHours(2) + // context.lockTimeout = 7_200_000 // milliseconds (0 or negative disables lock expiration) +} + +boolean canRun() { + return conditions.always() +} + +void doRun() { + // long-running work protected by the extended lock timeout +} +``` + +The effective timeout is resolved by precedence: value set in the script's `describeRun()` → value set by an extension's `prepareRun()` → global OSGi `Lock Timeout` configuration. ### History diff --git a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionContext.java b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionContext.java index c4a1fd7fc..7a19538a8 100644 --- a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionContext.java +++ b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionContext.java @@ -3,8 +3,10 @@ import dev.vml.es.acm.core.AcmConstants; import dev.vml.es.acm.core.code.log.LogInterceptorManager; import dev.vml.es.acm.core.repo.Repo; +import dev.vml.es.acm.core.util.DurationUtils; import dev.vml.es.acm.core.util.ResolverUtils; import groovy.lang.Binding; +import java.time.Duration; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; import org.apache.sling.api.resource.ResourceResolverFactory; @@ -42,6 +44,8 @@ public static String varPath(String executionId) { private boolean skipped = false; + private Duration lockTimeout = null; + private final Inputs inputs; private InputValues inputValues; @@ -171,6 +175,14 @@ public void setSkipped(boolean skipped) { this.skipped = skipped; } + public Duration getLockTimeout() { + return lockTimeout; + } + + public void setLockTimeout(Object lockTimeout) { + this.lockTimeout = DurationUtils.toDuration(lockTimeout); + } + public boolean isAborted() { return getCodeContext() .getOsgiContext() diff --git a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionQueue.java b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionQueue.java index 10ca5b090..6de938c9e 100644 --- a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionQueue.java +++ b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionQueue.java @@ -277,7 +277,12 @@ public JobExecutionResult process(Job job, JobExecutionContext context) { try { Execution immediateExecution = future.get(); - if (immediateExecution.getStatus() == ExecutionStatus.SKIPPED) { + if (immediateExecution.getStatus() == ExecutionStatus.LOCKED) { + LOG.debug("Execution locked '{}'", immediateExecution); + return context.result() + .message(QueuedMessage.of(ExecutionStatus.LOCKED, null).toJson()) + .cancelled(); + } else if (immediateExecution.getStatus() == ExecutionStatus.SKIPPED) { LOG.debug("Execution skipped '{}'", immediateExecution); return context.result() .message(QueuedMessage.of(ExecutionStatus.SKIPPED, null).toJson()) diff --git a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionStatus.java b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionStatus.java index 46707d1c8..496b25d75 100644 --- a/core/src/main/java/dev/vml/es/acm/core/code/ExecutionStatus.java +++ b/core/src/main/java/dev/vml/es/acm/core/code/ExecutionStatus.java @@ -17,6 +17,7 @@ public enum ExecutionStatus { STOPPING, // queued & immediate execution statuses SKIPPED, + LOCKED, STOPPED, ABORTED, FAILED, @@ -66,4 +67,12 @@ public static ExecutionStatus of(Job job, Executor executor) { public static List completed() { return Arrays.asList(ExecutionStatus.SUCCEEDED, ExecutionStatus.FAILED); } + + /** + * Statuses indicating the executable code was never actually run (conditions not met or lock held), + * so there is no meaningful output/input to persist - such executions are kept out of history by default. + */ + public static List notRun() { + return Arrays.asList(ExecutionStatus.SKIPPED, ExecutionStatus.LOCKED); + } } diff --git a/core/src/main/java/dev/vml/es/acm/core/code/Executor.java b/core/src/main/java/dev/vml/es/acm/core/code/Executor.java index ed751d651..1aeaaf7fe 100644 --- a/core/src/main/java/dev/vml/es/acm/core/code/Executor.java +++ b/core/src/main/java/dev/vml/es/acm/core/code/Executor.java @@ -12,13 +12,17 @@ import dev.vml.es.acm.core.notification.NotificationManager; import dev.vml.es.acm.core.osgi.InstanceInfo; import dev.vml.es.acm.core.osgi.OsgiContext; +import dev.vml.es.acm.core.repo.LockInfo; import dev.vml.es.acm.core.repo.Locker; import dev.vml.es.acm.core.script.ScriptRepository; import dev.vml.es.acm.core.state.Permissions; import dev.vml.es.acm.core.util.DateUtils; +import dev.vml.es.acm.core.util.DurationUtils; import dev.vml.es.acm.core.util.ResolverUtils; import dev.vml.es.acm.core.util.StringUtil; +import java.time.Duration; import java.util.Arrays; +import java.util.Calendar; import java.util.Date; import java.util.LinkedHashMap; import java.util.Map; @@ -102,6 +106,12 @@ public class Executor implements EventListener { name = "Notification Details Length", description = "Max length of the output and error. Use negative value to skip abbreviation.") int notificationDetailsLength() default 512; + + @AttributeDefinition( + name = "Lock Timeout", + description = + "Expiration time (in milliseconds) of executable locks, protecting against stale locks left behind when an instance is killed abruptly (e.g. pod recycling). Must exceed the max expected execution time. Use a zero or negative value to disable expiration.") + long lockTimeout() default 24L * 60L * 60L * 1000L; } @Reference @@ -239,13 +249,22 @@ private ContextualExecution executeInternal(ExecutionContext context) { boolean locking = !healthChecking; String lockName = executableLockName(context); - if (locking && queryLocker(resolverFactory, l -> l.isLocked(lockName))) { - return execution.end(ExecutionStatus.SKIPPED); + if (locking) { + LockInfo lockInfo = queryLocker(resolverFactory, l -> l.readLock(lockName)); + if (lockInfo != null && !lockInfo.isExpired()) { + LOG.warn( + "Execution locked '{}' by another run of '{}' since {} (expires {})", + context.getId(), + context.getExecutable().getId(), + formatLockTime(lockInfo.getLockedAt()), + formatLockTime(lockInfo.getLockedUntil())); + return execution.end(ExecutionStatus.LOCKED); + } } try { if (locking) { - useLocker(resolverFactory, l -> l.lock(lockName)); + useLocker(resolverFactory, l -> l.lock(lockName, lockTimeout(context))); } context.notifyStatus(ExecutionStatus.RUNNING); if (!healthChecking && config.logPrintingEnabled()) { @@ -289,9 +308,21 @@ private String executableLockName(ExecutionContext context) { LOCK_DIR, StringUtils.removeStart(context.getExecutable().getId(), AcmConstants.SETTINGS_ROOT + "/")); } + private Duration lockTimeout(ExecutionContext context) { + if (context.getLockTimeout() != null) { + return context.getLockTimeout(); + } + return DurationUtils.toDuration(config.lockTimeout()); + } + + private String formatLockTime(Calendar calendar) { + return calendar != null ? DateUtils.humanFormat().format(calendar.getTime()) : "never"; + } + private void handleHistory(ContextualExecution execution) { if (execution.getContext().isHistory() - && (execution.getContext().isDebug() || (execution.getStatus() != ExecutionStatus.SKIPPED))) { + && (execution.getContext().isDebug() + || !ExecutionStatus.notRun().contains(execution.getStatus()))) { useHistory(resolverFactory, h -> h.save(execution)); } } diff --git a/core/src/main/java/dev/vml/es/acm/core/repo/LockInfo.java b/core/src/main/java/dev/vml/es/acm/core/repo/LockInfo.java new file mode 100644 index 000000000..1c1a80142 --- /dev/null +++ b/core/src/main/java/dev/vml/es/acm/core/repo/LockInfo.java @@ -0,0 +1,27 @@ +package dev.vml.es.acm.core.repo; + +import java.util.Calendar; + +public class LockInfo { + + private final Calendar lockedAt; + + private final Calendar lockedUntil; + + public LockInfo(Calendar lockedAt, Calendar lockedUntil) { + this.lockedAt = lockedAt; + this.lockedUntil = lockedUntil; + } + + public Calendar getLockedAt() { + return lockedAt; + } + + public Calendar getLockedUntil() { + return lockedUntil; + } + + public boolean isExpired() { + return lockedUntil != null && Calendar.getInstance().after(lockedUntil); + } +} diff --git a/core/src/main/java/dev/vml/es/acm/core/repo/Locker.java b/core/src/main/java/dev/vml/es/acm/core/repo/Locker.java index ef8158a5e..67a9b15af 100644 --- a/core/src/main/java/dev/vml/es/acm/core/repo/Locker.java +++ b/core/src/main/java/dev/vml/es/acm/core/repo/Locker.java @@ -3,6 +3,7 @@ import dev.vml.es.acm.core.AcmConstants; import dev.vml.es.acm.core.AcmException; import dev.vml.es.acm.core.util.ResourceSpliterator; +import java.time.Duration; import java.util.Calendar; import java.util.HashMap; import java.util.Map; @@ -27,6 +28,8 @@ public class Locker { private static final String LOCKED_PROP = "locked"; + private static final String LOCKED_UNTIL_PROP = "lockedUntil"; + private static final int RETRY_COUNT = 5; private final ResourceResolver resolver; @@ -45,10 +48,29 @@ public Locker(ResourceResolver resolver, Supplier autoCommit) { public boolean isLocked(String lockName) { String name = normalizeName(lockName); Resource lock = getLock(name); - return isLock(lock); + return isLockActive(lock); + } + + public LockInfo readLock(String lockName) { + String name = normalizeName(lockName); + Resource lock = getLock(name); + if (!isLockPresent(lock)) { + return null; + } + return readLock(lock); + } + + private LockInfo readLock(Resource lock) { + return new LockInfo( + lock.getValueMap().get(LOCKED_PROP, Calendar.class), + lock.getValueMap().get(LOCKED_UNTIL_PROP, Calendar.class)); } public void lock(String lockName) { + lock(lockName, null); + } + + public void lock(String lockName, Duration ttl) { String name = normalizeName(lockName); PersistenceException exceptionLast = null; @@ -56,8 +78,13 @@ public void lock(String lockName) { try { Resource lockCurrent = getLock(name); if (lockCurrent != null) { - LOG.warn("Cannot create lock '{}' as it already exists!", name); - return; + if (readLock(lockCurrent).isExpired()) { + LOG.warn("Cannot create lock '{}' as it is stale - removing.", name); + resolver.delete(lockCurrent); + } else { + LOG.warn("Cannot create lock '{}' as it already exists!", name); + return; + } } Resource dirResource; @@ -73,6 +100,11 @@ public void lock(String lockName) { Map props = new HashMap<>(); props.put(JcrConstants.JCR_PRIMARYTYPE, RESOURCE_TYPE); props.put(LOCKED_PROP, Calendar.getInstance()); + if (ttl != null && !ttl.isNegative() && !ttl.isZero()) { + Calendar lockedUntil = Calendar.getInstance(); + lockedUntil.setTimeInMillis(lockedUntil.getTimeInMillis() + ttl.toMillis()); + props.put(LOCKED_UNTIL_PROP, lockedUntil); + } resolver.create(dirResource, nodeName, props); if (autoCommit.get()) { resolver.commit(); @@ -163,16 +195,20 @@ private Stream locks() { if (resource == null) { return Stream.empty(); } - return ResourceSpliterator.stream(resource).skip(1).filter(this::isLock); + return ResourceSpliterator.stream(resource).skip(1).filter(this::isLockPresent); + } + + private boolean isLockActive(Resource lock) { + return isLockPresent(lock) && !readLock(lock).isExpired(); } - private boolean isLock(Resource lock) { + private boolean isLockPresent(Resource lock) { return lock != null && lock.isResourceType(RESOURCE_TYPE) && lock.getValueMap().containsKey(LOCKED_PROP); } public boolean anyLocked() { - return locks().findAny().isPresent(); + return locks().anyMatch(lock -> !readLock(lock).isExpired()); } } diff --git a/core/src/main/java/dev/vml/es/acm/core/util/DurationUtils.java b/core/src/main/java/dev/vml/es/acm/core/util/DurationUtils.java new file mode 100644 index 000000000..525e82f7a --- /dev/null +++ b/core/src/main/java/dev/vml/es/acm/core/util/DurationUtils.java @@ -0,0 +1,108 @@ +package dev.vml.es.acm.core.util; + +import java.time.Duration; +import java.time.format.DateTimeParseException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.commons.lang3.StringUtils; + +public final class DurationUtils { + + private DurationUtils() { + // intentionally empty + } + + private static final Pattern HUMAN_PATTERN = Pattern.compile("(\\d+)\\s*(ms|s|m|h|d)", Pattern.CASE_INSENSITIVE); + + /** + * Parses a duration from a human-readable string (e.g. {@code '10m'}, {@code '2h30m'}, {@code '1d'}, + * {@code '500ms'}) or an ISO-8601 duration (e.g. {@code 'PT10M'}, {@code 'PT2H30M'}, {@code 'P1D'}). + */ + public static Duration toDuration(String text) { + if (StringUtils.isBlank(text)) { + return null; + } + String normalized = text.trim(); + + if (normalized.regionMatches(true, 0, "P", 0, 1)) { + try { + return Duration.parse(normalized); + } catch (DateTimeParseException e) { + throw new IllegalArgumentException(formatError(text), e); + } + } + + Matcher matcher = HUMAN_PATTERN.matcher(normalized); + Duration result = Duration.ZERO; + int matchedEnd = 0; + boolean matched = false; + while (matcher.find()) { + if (matcher.start() != matchedEnd) { + break; // gap means unexpected characters + } + long amount; + try { + amount = Long.parseLong(matcher.group(1)); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(formatError(text), e); + } + result = result.plus(toUnitDuration(amount, matcher.group(2).toLowerCase())); + matchedEnd = matcher.end(); + matched = true; + } + if (!matched || matchedEnd != normalized.length()) { + throw new IllegalArgumentException(formatError(text)); + } + return result; + } + + private static Duration toUnitDuration(long amount, String unit) { + switch (unit) { + case "ms": + return Duration.ofMillis(amount); + case "s": + return Duration.ofSeconds(amount); + case "m": + return Duration.ofMinutes(amount); + case "h": + return Duration.ofHours(amount); + case "d": + return Duration.ofDays(amount); + default: + throw new IllegalArgumentException(String.format("Unsupported duration unit '%s'!", unit)); + } + } + + private static String formatError(String text) { + return String.format("Cannot parse duration '%s'! Expected ISO-8601 (e.g. 'PT10M') or human-readable format (e.g. '10m', '2h30m', '1d', '500ms').", text); + } + + /** + * Converts milliseconds to a duration. A non-positive value yields a non-positive duration, + * which downstream consumers (e.g. the locker) interpret as "no expiration". + */ + public static Duration toDuration(long millis) { + return Duration.ofMillis(millis); + } + + public static Long toMillis(Duration duration) { + return duration == null ? null : duration.toMillis(); + } + + public static Duration toDuration(Object value) { + if (value == null) { + return null; + } + if (value instanceof Duration) { + return (Duration) value; + } + if (value instanceof Number) { + return toDuration(((Number) value).longValue()); + } + if (value instanceof String) { + return toDuration((String) value); + } + throw new IllegalArgumentException( + String.format("Cannot convert value of type '%s' to duration!", value.getClass().getName())); + } +} diff --git a/ui.frontend/src/components/ExecutionProgressBar.tsx b/ui.frontend/src/components/ExecutionProgressBar.tsx index 10b127dc3..dbfa1bbca 100644 --- a/ui.frontend/src/components/ExecutionProgressBar.tsx +++ b/ui.frontend/src/components/ExecutionProgressBar.tsx @@ -86,6 +86,8 @@ const ExecutionProgressBar: React.FC = ({ execution, return 'positive'; case ExecutionStatus.SKIPPED: return 'warning'; + case ExecutionStatus.LOCKED: + return 'warning'; case ExecutionStatus.ABORTED: case ExecutionStatus.FAILED: return 'critical'; diff --git a/ui.frontend/src/components/ExecutionStatusBadge.tsx b/ui.frontend/src/components/ExecutionStatusBadge.tsx index b9bf1cde1..9dc475ab7 100644 --- a/ui.frontend/src/components/ExecutionStatusBadge.tsx +++ b/ui.frontend/src/components/ExecutionStatusBadge.tsx @@ -3,6 +3,7 @@ import Alert from '@spectrum-icons/workflow/Alert'; import Cancel from '@spectrum-icons/workflow/Cancel'; import Checkmark from '@spectrum-icons/workflow/Checkmark'; import Clock from '@spectrum-icons/workflow/Clock'; +import LockClosed from '@spectrum-icons/workflow/LockClosed'; import Pause from '@spectrum-icons/workflow/Pause'; import React from 'react'; import { ExecutionStatus } from '../types/execution'; @@ -28,6 +29,8 @@ const getVariant = (status: ExecutionStatus): 'positive' | 'negative' | 'neutral case ExecutionStatus.STOPPED: case ExecutionStatus.SKIPPED: return 'neutral'; + case ExecutionStatus.LOCKED: + return 'yellow'; case ExecutionStatus.ABORTED: return 'negative'; default: @@ -53,6 +56,8 @@ const getIcon = (status: ExecutionStatus) => { return ; case ExecutionStatus.SKIPPED: return ; + case ExecutionStatus.LOCKED: + return ; case ExecutionStatus.ABORTED: return ; default: diff --git a/ui.frontend/src/components/ScriptsAutomaticHelpButton.tsx b/ui.frontend/src/components/ScriptsAutomaticHelpButton.tsx index 25c170a43..24c51d5f0 100644 --- a/ui.frontend/src/components/ScriptsAutomaticHelpButton.tsx +++ b/ui.frontend/src/components/ScriptsAutomaticHelpButton.tsx @@ -6,6 +6,7 @@ import Close from '@spectrum-icons/workflow/Close'; import Code from '@spectrum-icons/workflow/Code'; import Heart from '@spectrum-icons/workflow/Heart'; import Help from '@spectrum-icons/workflow/Help'; +import LockClosed from '@spectrum-icons/workflow/LockClosed'; import Replay from '@spectrum-icons/workflow/Replay'; import React from 'react'; @@ -30,6 +31,10 @@ const ScriptsAutomaticHelpButton: React.FC = () => (

Scripts that cannot run are skipped. Skipped executions are saved in the history only if debug mode is enabled. All other script executions are always saved for auditing purposes.

+

+ While a script runs, it is locked so the same script is not executed concurrently on another instance; such attempts are marked as locked. Locks expire automatically (configurable via the code executor + settings) to recover from instances killed abruptly, e.g. pod recycling. +

Script executor is active only when the instance is healthy, meaning all OSGi bundles are active, and no recent core OSGi events have occurred. Some bundles may be ignored, which may be useful to address known issues and still consider the instance as healthy. diff --git a/ui.frontend/src/hooks/execution.ts b/ui.frontend/src/hooks/execution.ts index 1c32aa7ac..9cf411dfd 100644 --- a/ui.frontend/src/hooks/execution.ts +++ b/ui.frontend/src/hooks/execution.ts @@ -40,7 +40,9 @@ export const useExecutionPolling = (executionId: string | undefined | null, poll if (queuedExecution.status === ExecutionStatus.FAILED) { ToastQueue.negative('Code execution failed!', { timeout: ToastTimeoutQuick }); } else if (queuedExecution.status === ExecutionStatus.SKIPPED) { - ToastQueue.neutral('Code execution cannot run!', { timeout: ToastTimeoutQuick }); + ToastQueue.neutral('Code execution skipped — conditions not met.', { timeout: ToastTimeoutQuick }); + } else if (queuedExecution.status === ExecutionStatus.LOCKED) { + ToastQueue.neutral('Code execution locked — already running.', { timeout: ToastTimeoutQuick }); } else if (queuedExecution.status === ExecutionStatus.SUCCEEDED) { ToastQueue.positive('Code execution succeeded!', { timeout: ToastTimeoutQuick }); } diff --git a/ui.frontend/src/pages/HistoryPage.tsx b/ui.frontend/src/pages/HistoryPage.tsx index d87d42fbc..0110d061a 100644 --- a/ui.frontend/src/pages/HistoryPage.tsx +++ b/ui.frontend/src/pages/HistoryPage.tsx @@ -5,6 +5,7 @@ import NotFound from '@spectrum-icons/illustrations/NotFound'; import Alert from '@spectrum-icons/workflow/Alert'; import Cancel from '@spectrum-icons/workflow/Cancel'; import Checkmark from '@spectrum-icons/workflow/Checkmark'; +import LockClosed from '@spectrum-icons/workflow/LockClosed'; import Pause from '@spectrum-icons/workflow/Pause'; import Search from '@spectrum-icons/workflow/Search'; import Star from '@spectrum-icons/workflow/Star'; @@ -119,6 +120,10 @@ const HistoryPage = () => { Skipped + + + Locked + Aborted diff --git a/ui.frontend/src/types/execution.ts b/ui.frontend/src/types/execution.ts index bb0b2f48e..4e5c28978 100644 --- a/ui.frontend/src/types/execution.ts +++ b/ui.frontend/src/types/execution.ts @@ -25,6 +25,7 @@ export enum ExecutionStatus { STOPPED = 'STOPPED', FAILED = 'FAILED', SKIPPED = 'SKIPPED', + LOCKED = 'LOCKED', ABORTED = 'ABORTED', SUCCEEDED = 'SUCCEEDED', }