diff --git a/.github/workflows/unomi-ci-build-tests.yml b/.github/workflows/unomi-ci-build-tests.yml index 076ab448e2..07f2f58583 100644 --- a/.github/workflows/unomi-ci-build-tests.yml +++ b/.github/workflows/unomi-ci-build-tests.yml @@ -21,7 +21,9 @@ jobs: unit-tests: name: Execute unit tests runs-on: ubuntu-latest - timeout-minutes: 15 + # Slightly above the previous 15: retried failures (rerunFailingTestsCount below) add time + # on a red build, and a timeout is a much worse signal than a clean failure. + timeout-minutes: 20 steps: - uses: actions/checkout@v5 - name: Set up JDK 17 @@ -36,12 +38,79 @@ jobs: sudo apt-get install -y graphviz dot -V - name: Build and Unit tests + env: + # Retry a failing test twice before calling the build red. Several suites (notably the + # scheduler ones) are timing-sensitive and this runner has 2 vCPU, so a single unlucky + # scheduling hiccup should not fail a whole build. A test that only passes on retry is + # NOT silently forgiven: Surefire records it as a flake, and the step below surfaces + # every one in the job summary so the flake rate stays visible instead of becoming + # invisible green. + MAVEN_EXTRA_OPTS: -Dsurefire.rerunFailingTestsCount=2 run: ./build.sh --ci # Keep only third-party dependencies in the post-job Maven cache: Unomi's own # snapshots are rebuilt every run and would only bloat the cache / risk staleness - name: Clean Unomi artifacts from Maven cache if: always() run: rm -rf ~/.m2/repository/org/apache/unomi + # A flake is a test that failed and then passed on retry. The build is green, so without + # this the signal is lost entirely — which is how the scheduler suites stayed unreliable + # for as long as they did. + - name: Detect flaky tests + id: flakes + if: always() + run: | + python3 - <<'PY' >> "$GITHUB_STEP_SUMMARY" + import glob, os, xml.etree.ElementTree as ET + flaky = [] + for path in glob.glob('**/target/surefire-reports/TEST-*.xml', recursive=True): + try: + root = ET.parse(path).getroot() + except ET.ParseError: + continue + for case in root.iter('testcase'): + reruns = case.findall('flakyFailure') + case.findall('flakyError') + if reruns: + msg = (reruns[0].get('message') or '').strip().replace('\n', ' ') + flaky.append((case.get('classname', '?'), case.get('name', '?'), + len(reruns), msg[:160])) + if flaky: + print('### :warning: Flaky tests detected\n') + print('These failed and then passed on retry. The build is green, but each one is') + print('a real intermittent failure worth investigating.\n') + print('| Test | Retries | First failure |') + print('| --- | --- | --- |') + for cls, name, n, msg in sorted(flaky): + print(f'| `{cls}.{name}` | {n} | {msg or "—"} |') + else: + print('### No flaky tests detected\n') + with open(os.environ['GITHUB_OUTPUT'], 'a') as out: + out.write(f'found={"true" if flaky else "false"}\n') + out.write(f'count={len(flaky)}\n') + PY + # Uploaded when the build failed OR when something only passed on retry: those are exactly + # the runs where the reports (and the scheduler diagnostics dumped into them) are worth + # keeping. Skipped on a clean green run so this does not accumulate on every push. + - name: Archive unit test reports + uses: actions/upload-artifact@v6 + if: always() && (job.status == 'failure' || steps.flakes.outputs.found == 'true') + with: + name: unit-test-reports-jdk17-${{ github.run_number }} + path: | + **/target/surefire-reports/** + if-no-files-found: ignore + retention-days: 14 + # Always publish so a later "re-run failed jobs" pass updates the check to green, matching + # the integration-test job's behaviour. + - name: Publish Test Report + uses: mikepenz/action-junit-report@v3 + if: always() + continue-on-error: true + with: + report_paths: '**/target/surefire-reports/TEST-*.xml' + check_name: 'JUnit Test Report (unit tests)' + update_check: true + fail_on_failure: false + require_tests: false integration-tests: name: Execute integration tests @@ -74,11 +143,19 @@ jobs: MAVEN_EXTRA_OPTS: >- -Dopensearch.port=${{ matrix.port }} -Delasticsearch.port=${{ matrix.port }} + # This job is gated on `needs: unit-tests`, so the unit suite and the Javadoc/checkstyle + # validation have already passed on this exact commit. Re-running either here is pure + # duplication before the integration tests this job exists for, and the legs run + # sequentially (max-parallel: 1), so it costs twice over. + # --skip-unit-tests activates the skip-unit-tests profile, which sets surefire's skip + # only: failsafe, and therefore the ITs, still run. + # --no-javadoc drops the two extra full-reactor invocations --ci adds + # (javadoc:javadoc and javadoc-tags-warn checkstyle:check). run: | if [ "${{ matrix.search-engine }}" = "opensearch" ]; then - ./build.sh --ci --integration-tests --use-opensearch + ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc --use-opensearch else - ./build.sh --ci --integration-tests + ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc fi # Keep only third-party dependencies in the post-job Maven cache: Unomi's own # snapshots are rebuilt every run and would only bloat the cache / risk staleness 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 c5d698e771..5a2f52acef 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 @@ * @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 enum TaskStatus { 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 Date getLockDate() { /** * 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. + *

+ * A lock's lifetime is a lease granted by its owner: 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/build.sh b/build.sh index c74fd5eb80..408768e320 100755 --- a/build.sh +++ b/build.sh @@ -279,8 +279,11 @@ RESOLVER_DEBUG=false KEEP_CONTAINER=false IT_SEARCH_ENGINE_LOGS=false IT_MEMORY_SAMPLER=true -IT_MEMORY_INTERVAL=30 +# 10s, matching itests/sample-it-memory.sh: at 30s a single sample spans several ITs, so a +# stall cannot be attributed to the test that caused it. +IT_MEMORY_INTERVAL=10 JAVADOC=false +NO_JAVADOC=false LOG_FILE="" LOG_FILE_ONLY=false @@ -327,8 +330,9 @@ EOF echo -e " ${CYAN}--keep-container${NC} Keep search engine container running after tests (for post-failure inspection)" echo -e " ${CYAN}--search-engine-logs${NC} Stream search engine Docker logs to the Maven console during integration tests" echo -e " ${CYAN}--no-memory-sampler${NC} Disable JVM/system memory sampling during integration tests" - echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval in seconds (default: 30)" + echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval in seconds (default: 10)" echo -e " ${CYAN}--javadoc${NC} Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" + echo -e " ${CYAN}--no-javadoc${NC} Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo -e " ${CYAN}--ci${NC} CI mode: no Karaf, non-interactive, includes Javadoc" echo -e " ${CYAN}--log-file PATH${NC} Tee all output to PATH (console + file)" echo -e " ${CYAN}--log-file-only${NC} With --log-file: write to file only, suppress console" @@ -371,8 +375,9 @@ EOF echo " --keep-container Keep search engine container running after tests (for post-failure inspection)" echo " --search-engine-logs Stream search engine Docker logs to the Maven console during integration tests" echo " --no-memory-sampler Disable JVM/system memory sampling during integration tests" - echo " --memory-interval SEC Memory sample interval in seconds (default: 30)" + echo " --memory-interval SEC Memory sample interval in seconds (default: 10)" echo " --javadoc Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" + echo " --no-javadoc Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo " --ci CI mode: no Karaf, non-interactive, includes Javadoc" echo " --log-file PATH Tee all output to PATH (console + file)" echo " --log-file-only With --log-file: write to file only, suppress console" @@ -549,6 +554,11 @@ while [ "$1" != "" ]; do --javadoc) JAVADOC=true ;; + --no-javadoc) + # Explicit veto, applied after argument parsing so it wins regardless of whether it + # appears before or after --ci (which turns Javadoc on). + NO_JAVADOC=true + ;; --log-file) shift LOG_FILE="$1" @@ -1167,6 +1177,12 @@ echo "Estimated time: 3-5 minutes for build, 50-60 minutes with integration test start_timer # Build phases with enhanced output +# Apply the --no-javadoc veto now that all arguments are parsed, so it wins over --ci +# regardless of flag order. +if [ "$NO_JAVADOC" = true ]; then + JAVADOC=false +fi + [ "$JAVADOC" = true ] && total_steps=4 || total_steps=2 current_step=0 diff --git a/itests/lib/it-run-memory.sh b/itests/lib/it-run-memory.sh index 1c8f434668..8c61cdd5b5 100644 --- a/itests/lib/it-run-memory.sh +++ b/itests/lib/it-run-memory.sh @@ -27,7 +27,7 @@ IT_MEMORY_SAMPLER_LOG="memory-sampler.log" IT_MEMORY_SAMPLER_CACHE="memory-sampler.cache" IT_MEMORY_SWAP_PRESSURE_MB=2048 -IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m' +IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m\tkaraf_cpu_pct\tkaraf_io_read_mb_s\tkaraf_io_write_mb_s\tsearch_cpu_pct\tsearch_io_read_mb_s\tsearch_io_write_mb_s' _IT_MEMORY_OS="" @@ -350,7 +350,12 @@ it_memory_parse_docker_mem_to_mb() { } it_memory_find_karaf_pid() { - pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1 + # `|| true`: pgrep exits non-zero when nothing matches, and with `set -euo pipefail` that + # aborted the whole sample. The sampler starts before Karaf does, so every sample taken + # during startup was discarded -- exactly the window where the search engine is booting and + # its resource use is most interesting. No match now yields an empty pid, which the callers + # and the summarizer already treat as "no Karaf yet" (guarded by `if ($2+0 > 0)`). + pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1 || true } it_memory_karaf_max_mb_cached() { @@ -446,21 +451,139 @@ it_memory_search_engine_stats() { echo -e "$(it_memory_mb_from_bytes "${used_bytes:-0}")\t$(it_memory_mb_from_bytes "${max_bytes:-0}")" } -it_memory_docker_rss_mb() { +# --- CPU and disk I/O sampling ------------------------------------------------- +# +# Added to answer "is the run CPU-bound, I/O-bound, or waiting?". The memory columns alone +# could not distinguish a busy run from an idle one blocked on a remote call, which is exactly +# the question raised by the Elasticsearch/OpenSearch IT duration gap: system load was near +# idle on the slower engine, so the extra time was spent waiting rather than computing. +# +# CPU is measured as a TRUE INTERVAL PERCENTAGE, not ps(1)'s %cpu -- that is an average over the +# whole process lifetime, so a JVM that was busy at startup reads as busy forever and the number +# is useless for spotting a stall. On Linux -- which is what CI runs, and the only place these +# numbers are compared across runs -- we delta /proc//stat between samples for a true +# interval figure. macOS has no procfs, so it falls back to ps(1)'s lifetime average: good enough +# to see that a process is alive and roughly how hard it has worked, but NOT comparable with a +# Linux sample and not to be read as "CPU right now". Everything here is best-effort: a missing +# file, a dead pid or an absent docker CLI yields 0 and never fails a run. + +# Stores "value timestamp" pairs so the next sample can compute a delta. +_it_memory_counter_cache() { + local target_dir="$1" key="$2" + echo "$target_dir/.it-memory-counter-$key" +} + +# Echoes the per-second rate between this reading and the previous one, or 0 on the first call. +_it_memory_rate_per_sec() { + local target_dir="$1" key="$2" value="$3" + local cache prev_value prev_ts now delta_v delta_t + cache="$(_it_memory_counter_cache "$target_dir" "$key")" + now="$(date +%s)" + + if [ -r "$cache" ]; then + read -r prev_value prev_ts < "$cache" 2>/dev/null || true + fi + printf '%s %s\n' "$value" "$now" > "$cache" 2>/dev/null || true + + if [ -z "${prev_value:-}" ] || [ -z "${prev_ts:-}" ]; then + echo "0" + return + fi + delta_t=$((now - prev_ts)) + [ "$delta_t" -le 0 ] && { echo "0"; return; } + delta_v="$(awk -v a="$value" -v b="$prev_value" 'BEGIN { d = a - b; print (d > 0 ? d : 0) }')" + awk -v d="$delta_v" -v t="$delta_t" 'BEGIN { printf "%.2f", d / t }' +} + +# Interval CPU% for a pid. >100 is legitimate on multi-core (sum across threads). +it_memory_process_cpu_pct() { + local target_dir="$1" pid="${2:-}" + local ticks hz cpu_s rate + + if [ -z "$pid" ] || [ "$pid" = "0" ] || ! kill -0 "$pid" 2>/dev/null; then + echo "0" + return + fi + + if it_memory_is_linux && [ -r "/proc/$pid/stat" ]; then + # Fields 14 (utime) and 15 (stime), in clock ticks. comm (field 2) is parenthesised and + # may itself contain spaces AND parentheses, so split after the LAST ')' rather than the + # first: a process named e.g. "java (worker)" otherwise shifts every subsequent index. + ticks="$(awk '{ + i = length($0) + while (i > 0 && substr($0, i, 1) != ")") i-- + n = split(substr($0, i + 2), f, " ") + if (n >= 13) print f[12] + f[13]; else print 0 + }' "/proc/$pid/stat" 2>/dev/null)" + [ -z "$ticks" ] && { echo "0"; return; } + hz="$(getconf CLK_TCK 2>/dev/null || echo 100)" + cpu_s="$(awk -v t="$ticks" -v hz="$hz" 'BEGIN { printf "%.4f", t / hz }')" + rate="$(_it_memory_rate_per_sec "$target_dir" "cpu-$pid" "$cpu_s")" + awk -v r="$rate" 'BEGIN { printf "%.1f", r * 100 }' + return + fi + + # macOS / no procfs: lifetime average, better than nothing for a local run. + ps -o %cpu= -p "$pid" 2>/dev/null | tr -d ' ' | awk 'NF { printf "%.1f", $1; found = 1 } END { if (!found) print 0 }' +} + +# Interval disk read/write in MB/s for a pid (Linux only; /proc//io). +it_memory_process_io_mb_s() { + local target_dir="$1" pid="${2:-}" + local read_bytes write_bytes read_rate write_rate + + if [ -z "$pid" ] || [ "$pid" = "0" ] || ! it_memory_is_linux || [ ! -r "/proc/$pid/io" ]; then + echo -e "0\t0" + return + fi + + read_bytes="$(awk '/^read_bytes:/ { print $2 }' "/proc/$pid/io" 2>/dev/null)" + write_bytes="$(awk '/^write_bytes:/ { print $2 }' "/proc/$pid/io" 2>/dev/null)" + read_rate="$(_it_memory_rate_per_sec "$target_dir" "ior-$pid" "${read_bytes:-0}")" + write_rate="$(_it_memory_rate_per_sec "$target_dir" "iow-$pid" "${write_bytes:-0}")" + awk -v r="$read_rate" -v w="$write_rate" 'BEGIN { printf "%.2f\t%.2f", r / 1048576, w / 1048576 }' +} + +# One docker stats call per sample, returning RSS, CPU% and block I/O together. +# +# Deliberately a single invocation: `docker stats --no-stream` costs ~1s and briefly loads the +# daemon, and the sampler now runs 3x more often (10s rather than 30s). Two calls per sample +# would have meant six times the docker traffic of the original sampler, perturbing the very +# run being measured and tripling the exposure to a hung daemon. CPUPerc is already an interval +# measurement; BlockIO is cumulative and is deltaed here. +# +# Echoes: rss_mb \t cpu_pct \t io_read_mb_s \t io_write_mb_s +it_memory_docker_sample() { local target_dir="$1" - local container rss + local container stats mem cpu blockio read_raw write_raw read_b write_b read_rate write_rate if ! command -v docker >/dev/null 2>&1; then - echo "0" + echo -e "0\t0\t0\t0" return fi container="$(it_memory_resolve_docker_container "$target_dir")" - rss="$(docker stats --no-stream --format '{{.MemUsage}}' "$container" 2>/dev/null | head -1 | cut -d/ -f1 | tr -d ' ')" + stats="$(docker stats --no-stream --format '{{.MemUsage}}|{{.CPUPerc}}|{{.BlockIO}}' "$container" 2>/dev/null | head -1)" + if [ -z "$stats" ]; then + echo -e "0\t0\t0\t0" + return + fi + + mem="$(echo "$stats" | cut -d'|' -f1 | cut -d/ -f1 | tr -d ' ')" + cpu="$(echo "$stats" | cut -d'|' -f2 | tr -d ' %')" + blockio="$(echo "$stats" | cut -d'|' -f3)" + read_raw="$(echo "$blockio" | cut -d/ -f1 | tr -d ' ')" + write_raw="$(echo "$blockio" | cut -d/ -f2 | tr -d ' ')" + read_b="$(it_memory_parse_docker_mem_to_mb "$read_raw")" + write_b="$(it_memory_parse_docker_mem_to_mb "$write_raw")" + read_rate="$(_it_memory_rate_per_sec "$target_dir" "dior" "${read_b:-0}")" + write_rate="$(_it_memory_rate_per_sec "$target_dir" "diow" "${write_b:-0}")" - it_memory_parse_docker_mem_to_mb "$rss" + printf '%s\t%.1f\t%.2f\t%.2f\n' \ + "$(it_memory_parse_docker_mem_to_mb "$mem")" "${cpu:-0}" "${read_rate:-0}" "${write_rate:-0}" } + it_memory_system_stats() { local mem_available swap_used load_1m @@ -482,13 +605,22 @@ it_memory_sample_once() { es_line="$(it_memory_search_engine_stats "$port")" sys_line="$(it_memory_system_stats)" - printf '%s\t%s\t%s\t%s\t%s\t%s\n' \ + # One docker call per sample; split into the RSS column (8) and the CPU/IO columns (15-17). + local docker_line docker_rss docker_cpu_io + docker_line="$(it_memory_docker_sample "$target_dir")" + docker_rss="$(printf '%s' "$docker_line" | cut -f1)" + docker_cpu_io="$(printf '%s' "$docker_line" | cut -f2-4)" + + printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \ "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ "${karaf_pid:-0}" \ "$karaf_line" \ "$es_line" \ - "$(it_memory_docker_rss_mb "$target_dir")" \ - "$sys_line" + "${docker_rss:-0}" \ + "$sys_line" \ + "$(it_memory_process_cpu_pct "$target_dir" "$karaf_pid")" \ + "$(it_memory_process_io_mb_s "$target_dir" "$karaf_pid")" \ + "${docker_cpu_io:-$(printf '0\t0\t0')}" } it_memory_write_samples_header() { @@ -508,7 +640,7 @@ it_memory_summarize_samples() { awk -F'\t' -v summary="$summary_file" -v swap_pressure_mb="$IT_MEMORY_SWAP_PRESSURE_MB" ' NR == 1 { next } - NF < 11 { next } + NF < 11 { next } # pre-CPU/IO samples still summarize { samples++ if ($2+0 > 0) { @@ -524,6 +656,18 @@ it_memory_summarize_samples() { if ($11+0 > peak_load) peak_load = $11+0 if (samples == 1) first_swap = $10+0 last_swap = $10+0 + # CPU / IO columns are absent in samples written before they were added. + if (NF >= 17) { + cpu_samples++ + karaf_cpu_sum += $12+0; if ($12+0 > peak_karaf_cpu) peak_karaf_cpu = $12+0 + search_cpu_sum += $15+0; if ($15+0 > peak_search_cpu) peak_search_cpu = $15+0 + io_sum += $13+0 + $14+0 + $16+0 + $17+0 + if ($13+0 + $14+0 > peak_karaf_io) peak_karaf_io = $13+0 + $14+0 + if ($16+0 + $17+0 > peak_search_io) peak_search_io = $16+0 + $17+0 + # "Idle" = neither process using meaningful CPU: the signature of a run that is + # waiting on latency rather than doing work. + if ($12+0 < 10 && $15+0 < 10) idle_samples++ + } } END { if (samples == 0) exit 1 @@ -541,6 +685,24 @@ it_memory_summarize_samples() { printf("memory.min.system.mem.available.mb=%d\n", min_mem_avail+0) >> summary printf("memory.peak.system.swap.used.mb=%d\n", peak_swap+0) >> summary printf("memory.peak.system.load.1m=%.2f\n", peak_load+0) >> summary + if (cpu_samples > 0) { + printf("cpu.samples.count=%d\n", cpu_samples) >> summary + printf("cpu.mean.karaf.pct=%.1f\n", karaf_cpu_sum / cpu_samples) >> summary + printf("cpu.peak.karaf.pct=%.1f\n", peak_karaf_cpu+0) >> summary + printf("cpu.mean.search.pct=%.1f\n", search_cpu_sum / cpu_samples) >> summary + printf("cpu.peak.search.pct=%.1f\n", peak_search_cpu+0) >> summary + printf("cpu.idle.samples.pct=%d\n", idle_samples * 100 / cpu_samples) >> summary + printf("io.peak.karaf.mb.s=%.2f\n", peak_karaf_io+0) >> summary + printf("io.peak.search.mb.s=%.2f\n", peak_search_io+0) >> summary + printf("io.mean.total.mb.s=%.2f\n", io_sum / cpu_samples) >> summary + # Mostly-idle CPU with negligible I/O means the run is latency-bound: time is + # going on waiting (polls, refresh intervals, timeouts), not on work. + if (idle_samples * 100 / cpu_samples >= 70 && io_sum / cpu_samples < 5) { + printf("cpu.warning.mostly.idle=true\n") >> summary + } else { + printf("cpu.warning.mostly.idle=false\n") >> summary + } + } printf("memory.karaf.headroom.pct=%d\n", karaf_headroom+0) >> summary printf("memory.search.headroom.pct=%d\n", es_headroom+0) >> summary if (swap_pressure) { diff --git a/itests/sample-it-memory.sh b/itests/sample-it-memory.sh index 19e0d5d5c0..e0469bd02d 100755 --- a/itests/sample-it-memory.sh +++ b/itests/sample-it-memory.sh @@ -39,7 +39,10 @@ source "$SCRIPT_DIR/lib/it-run.sh" source "$SCRIPT_DIR/lib/it-run-memory.sh" TARGET_DIR="$SCRIPT_DIR/target" -INTERVAL=30 +# 10s, down from 30s: at 30s a sample covers several ITs at once, so a stall cannot be +# attributed to the test that caused it. Each sample is a few cheap reads plus one +# `docker stats --no-stream`, and a 50-minute run produces ~300 rows (a few tens of KB). +INTERVAL=10 SEARCH_PORT="" PRINT_ONLY=false COMMAND="" @@ -58,7 +61,7 @@ Commands: Options: --target-dir DIR IT target directory (default: itests/target) - --interval SEC Sample interval in seconds for start (default: 30) + --interval SEC Sample interval in seconds for start (default: 10) --port PORT Search engine HTTP port override --print-only With operator-note: print to stdout instead of writing file -h, --help Show this help @@ -78,7 +81,7 @@ parse_args() { ;; --interval) shift - INTERVAL="${1:-30}" + INTERVAL="${1:-10}" ;; --port) shift 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 f36fc297c2..030305e8e8 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 9c1541d968..a251eebf47 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 3c98963fc3..5e84267dac 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 @@ private void updateTaskState(ScheduledTask task, TaskStatus newStatus, String er 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 @@ private void updateTaskState(ScheduledTask task, TaskStatus newStatus, String er } 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 details = task.getStatusDetails(); @@ -899,6 +901,7 @@ public void preDestroy() { // 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 8f3bd41a48..98b98a2e30 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 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.*; @@ -502,6 +504,7 @@ private void abortPreparedExecution(ScheduledTask task) { 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 +535,16 @@ private void startLockRenewal(ScheduledTask task) { 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 { @@ -657,9 +670,57 @@ private boolean canCommitTerminalTransition(ScheduledTask task) { // 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. + *

+ * 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 version, 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. + *

+ * 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 latestDetails = latest.getStatusDetails(); + if (latestDetails == null) { + return; + } + Object latestHistory = latestDetails.get("executionHistory"); + if (!(latestHistory instanceof List)) { + return; + } + Map 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 @@ -668,6 +729,7 @@ private boolean canCommitTerminalTransition(ScheduledTask task) { 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 978d97fa64..fbe1517c07 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 boolean acquireLock(ScheduledTask task) { // 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 @@ private boolean acquireInMemoryLock(ScheduledTask task) { 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 @@ private boolean acquireDistributedLock(ScheduledTask task) { 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 boolean releaseLock(ScheduledTask task) { 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 boolean releaseLock(ScheduledTask task) { 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 boolean renewLock(ScheduledTask task) { } 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 boolean renewLock(ScheduledTask task) { // 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 boolean isLockExpired(ScheduledTask task) { 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 43b29e3685..d41c34c309 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 @@ private void recoverCrashedTask(ScheduledTask task) { } latest.setLockOwner(null); latest.setLockDate(null); + latest.setLockLeaseMillis(0); // Record the crash in execution history recordCrash(latest, previousOwner); @@ -273,6 +274,7 @@ private void recoverCrashedTask(ScheduledTask task) { 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 c8877bd725..d8bf91266c 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 @@ private void updateStateSpecificFields(ScheduledTask task, TaskStatus newStatus, private void clearTaskExecution(ScheduledTask task) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); task.setWaitingForTaskType(null); task.setCurrentStep(null); } @@ -162,6 +163,7 @@ private void preserveCrashState(ScheduledTask task, String nodeId) { 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 0000000000..126cee7013 --- /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()}. + *

+ * 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/SchedulerDiagnosticsExtension.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java new file mode 100644 index 0000000000..73bf1eb5d3 --- /dev/null +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java @@ -0,0 +1,271 @@ +/* + * 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 ch.qos.logback.classic.Level; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.CyclicBufferAppender; +import org.apache.unomi.api.tasks.ScheduledTask; +import org.apache.unomi.persistence.spi.PersistenceService; +import org.junit.jupiter.api.extension.AfterEachCallback; +import org.junit.jupiter.api.extension.BeforeEachCallback; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.api.extension.TestExecutionExceptionHandler; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Field; +import java.text.SimpleDateFormat; +import java.util.Collection; +import java.util.Date; +import java.util.List; +import java.util.Map; + +/** + * Makes an intermittent scheduler test failure diagnosable from the CI log it failed in. + *

+ * Scheduler failures are almost always about state and ordering: which node held a lock, when it + * was renewed, who decided it had expired, what the store actually contained. The scheduler already + * logs all of that as {@code LOCK-DIAG} lines, but only at DEBUG, and CI does not run at DEBUG -- + * so every intermittent failure historically arrived as a bare assertion message with the evidence + * discarded. Re-running with {@code -DTEST_LOG_LEVEL=DEBUG} rarely helps, because an intermittent + * failure usually does not recur on demand. + *

+ * This extension therefore captures DEBUG for the scheduler packages into a bounded in-memory ring + * buffer that costs nothing on a passing test, and dumps it -- together with a snapshot of every + * task document in the store -- at the moment a test fails. The snapshot is taken from + * {@link TestExecutionExceptionHandler}, which runs before the test's own {@code @AfterEach} + * teardown, so the store is still alive and holds the state that caused the failure rather than + * whatever cleanup left behind. + *

+ * When {@code -DTEST_LOG_LEVEL} is set explicitly, the extension stays out of the way and leaves + * logback's configured behaviour alone: an explicit request for console output should get console + * output. + */ +public class SchedulerDiagnosticsExtension + implements BeforeEachCallback, AfterEachCallback, TestExecutionExceptionHandler { + + /** Enough lines to cover several checker ticks across a handful of nodes. */ + private static final int BUFFER_SIZE = 4000; + + /** Packages whose DEBUG output explains scheduler behaviour. */ + private static final String[] CAPTURED_LOGGERS = { + "org.apache.unomi.services.impl.scheduler", + "org.apache.unomi.services.impl.cluster" + }; + + private static final String APPENDER_NAME = "scheduler-diagnostics-ring-buffer"; + private static final ExtensionContext.Namespace NAMESPACE = + ExtensionContext.Namespace.create(SchedulerDiagnosticsExtension.class); + + private static boolean explicitLogLevelRequested() { + String requested = System.getProperty("TEST_LOG_LEVEL"); + return requested != null && !requested.trim().isEmpty(); + } + + @Override + public void beforeEach(ExtensionContext context) { + if (explicitLogLevelRequested()) { + return; + } + if (!(LoggerFactory.getILoggerFactory() instanceof LoggerContext)) { + return; // not logback (shaded/OSGi runs); nothing to attach to + } + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + + CyclicBufferAppender buffer = new CyclicBufferAppender<>(); + buffer.setContext(loggerContext); + buffer.setName(APPENDER_NAME); + buffer.setMaxSize(BUFFER_SIZE); + buffer.start(); + + for (String name : CAPTURED_LOGGERS) { + ch.qos.logback.classic.Logger logger = loggerContext.getLogger(name); + // additive=false keeps the captured DEBUG out of the console on passing runs; the + // buffer dump below is the only consumer, and it only fires on failure. + logger.setLevel(Level.DEBUG); + logger.setAdditive(false); + logger.addAppender(buffer); + } + context.getStore(NAMESPACE).put(APPENDER_NAME, buffer); + } + + @Override + public void afterEach(ExtensionContext context) { + @SuppressWarnings("unchecked") + CyclicBufferAppender buffer = + context.getStore(NAMESPACE).remove(APPENDER_NAME, CyclicBufferAppender.class); + if (buffer == null || !(LoggerFactory.getILoggerFactory() instanceof LoggerContext)) { + return; + } + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + for (String name : CAPTURED_LOGGERS) { + ch.qos.logback.classic.Logger logger = loggerContext.getLogger(name); + logger.detachAppender(buffer); + logger.setAdditive(true); + logger.setLevel(null); // inherit from root again + } + buffer.stop(); + } + + @Override + public void handleTestExecutionException(ExtensionContext context, Throwable throwable) + throws Throwable { + StringBuilder report = new StringBuilder(); + report.append("\n================ SCHEDULER DIAGNOSTICS for ") + .append(context.getRequiredTestClass().getSimpleName()).append('.') + .append(context.getRequiredTestMethod().getName()) + .append(" ================\n") + .append("Failure: ").append(throwable).append('\n'); + + appendTaskSnapshot(report, context); + appendBufferedLog(report, context); + + report.append("================ END SCHEDULER DIAGNOSTICS ================\n"); + // stdout, not a logger: this must survive whatever logging configuration is in force, and + // Surefire captures stdout into the report the CI log shows. + System.out.println(report); + + throw throwable; + } + + /** + * Dumps every task document the test's persistence service can see. Taken before teardown, so + * this is the state that produced the failure. + */ + private void appendTaskSnapshot(StringBuilder report, ExtensionContext context) { + report.append("\n-- task documents in the store at failure time --\n"); + PersistenceService persistenceService = findPersistenceService(context); + if (persistenceService == null) { + report.append(" (no PersistenceService field found on the test instance)\n"); + return; + } + try { + // getAllItems is search-based, and both the in-memory harness and a real cluster hold a + // refresh interval behind the store. Force visibility first: the test has already + // failed, so there is no state left worth preserving, and a snapshot that silently + // reports "(none)" because of refresh lag is worse than useless. + try { + persistenceService.refreshIndex(ScheduledTask.class); + persistenceService.refresh(); + } catch (Exception ignored) { + report.append(" (refresh before snapshot failed; list may lag the store)\n"); + } + List tasks = + persistenceService.getAllItems(ScheduledTask.class, 0, -1, null).getList(); + if (tasks.isEmpty()) { + report.append(" (none)\n"); + return; + } + SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS"); + for (ScheduledTask task : tasks) { + report.append(" ").append(task.getItemId()) + .append(" type=").append(task.getTaskType()) + .append(" status=").append(task.getStatus()) + .append(" enabled=").append(task.isEnabled()) + .append(" execNode=").append(task.getExecutingNodeId()) + .append(" lockOwner=").append(task.getLockOwner()) + .append(" lockDate=") + .append(task.getLockDate() == null ? "null" : fmt.format(task.getLockDate())) + .append(" lease=").append(task.getLockLeaseMillis()).append("ms") + .append(" success=").append(task.getSuccessCount()) + .append(" failure=").append(task.getFailureCount()) + .append(" nextExec=") + .append(task.getNextScheduledExecution() == null + ? "null" : fmt.format(task.getNextScheduledExecution())) + .append(" history=").append(historySize(task)) + .append(" lastError=").append(task.getLastError()) + .append('\n'); + } + } catch (Exception e) { + report.append(" (failed to read tasks: ").append(e).append(")\n"); + } + } + + private static int historySize(ScheduledTask task) { + Map details = task.getStatusDetails(); + if (details == null) { + return 0; + } + Object history = details.get("executionHistory"); + return history instanceof Collection ? ((Collection) history).size() : 0; + } + + /** + * Finds a {@link PersistenceService} on the test instance. Reflection rather than an interface + * the tests must implement: the point is that adding this extension to a test class costs one + * annotation and no other change. + */ + private PersistenceService findPersistenceService(ExtensionContext context) { + Object testInstance = context.getTestInstance().orElse(null); + if (testInstance == null) { + return null; + } + for (Class type = testInstance.getClass(); type != null; type = type.getSuperclass()) { + for (Field field : type.getDeclaredFields()) { + if (!PersistenceService.class.isAssignableFrom(field.getType())) { + continue; + } + try { + field.setAccessible(true); + PersistenceService value = (PersistenceService) field.get(testInstance); + if (value != null) { + return value; + } + } catch (ReflectiveOperationException | RuntimeException ignored) { + // Not readable; keep looking. + } + } + } + return null; + } + + private void appendBufferedLog(StringBuilder report, ExtensionContext context) { + @SuppressWarnings("unchecked") + CyclicBufferAppender buffer = + context.getStore(NAMESPACE).get(APPENDER_NAME, CyclicBufferAppender.class); + if (buffer == null) { + report.append("\n-- captured scheduler DEBUG log --\n") + .append(" (not captured; -DTEST_LOG_LEVEL was set, so the log went to the console)\n"); + return; + } + int count = buffer.getLength(); + report.append("\n-- captured scheduler DEBUG log (last ").append(count) + .append(" events, newest last) --\n"); + if (count == 0) { + report.append(" (empty)\n"); + return; + } + SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS"); + for (int i = 0; i < count; i++) { + ILoggingEvent event = buffer.get(i); + if (event == null) { + continue; + } + report.append(" ").append(fmt.format(new Date(event.getTimeStamp()))) + .append(" [").append(event.getThreadName()).append("] ") + .append(event.getLevel()).append(' ') + .append(shortLoggerName(event.getLoggerName())).append(" - ") + .append(event.getFormattedMessage()).append('\n'); + } + } + + private static String shortLoggerName(String loggerName) { + int lastDot = loggerName.lastIndexOf('.'); + return lastDot < 0 ? loggerName : loggerName.substring(lastDot + 1); + } +} 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 62ae062bd8..d2ce88f7f6 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 @@ -75,6 +75,7 @@ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) @Tag("ClusterTests") +@ExtendWith(SchedulerDiagnosticsExtension.class) public class SchedulerServiceClusterRaceTest { private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerServiceClusterRaceTest.class); @@ -331,6 +332,150 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce 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. + *

+ * 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 observer's 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/SchedulerServiceImplTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java index d3e91d56ce..0521f476bd 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 @@ -74,9 +74,30 @@ * - RetryTests: Task retry behavior and delay * - MaintenanceTests: Task cleanup and maintenance * - QueryTests: Task querying and filtering + * + *

Debugging

+ * Logging is configured by {@code src/test/resources/logback-test.xml}; run with + * {@code -DTEST_LOG_LEVEL=DEBUG} to see the scheduler's {@code LOCK-DIAG} traces, which record + * every lock acquisition, renewal, expiry verdict and recovery decision. A bare assertion + * failure from this suite is rarely diagnosable without them. + * + *

Timing rules for this suite

+ * The {@code setUp()} scheduler keeps polling in the background for the whole test, so: + * */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class SchedulerServiceImplTest { private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerServiceImplTest.class); @@ -89,7 +110,17 @@ public class SchedulerServiceImplTest { private static final long TEST_TIMEOUT = 15000; // 15 seconds — extra margin for loaded CI runners /** Time unit for test timeouts */ private static final TimeUnit TEST_TIME_UNIT = TimeUnit.MILLISECONDS; - /** Lock timeout for testing lock expiration */ + /** + * Lock timeout for the setUp scheduler and for multi-node tests, matching + * {@code TaskLockManager}'s production default. Deliberately NOT short: the setUp scheduler + * keeps polling in the background during every test, and a node whose lock timeout is shorter + * than a peer's renewal cadence (peer timeout / 3) declares that peer's live locks expired in + * the gap between renewals — before lock leases this stole locks from mid-execution tasks and + * double-ran them (the CI flake in testConcurrentLockAcquisition). All nodes sharing one store + * must agree on this value unless lock expiry itself is the behaviour under test. + */ + private static final long DEFAULT_LOCK_TIMEOUT = 10000; // 10 seconds + /** Short lock timeout for tests that exercise lock expiration; set it explicitly per test. */ private static final long TEST_LOCK_TIMEOUT = 1000; // 1 second /** Thread pool size for parallel execution */ private static final int TEST_THREAD_POOL_SIZE = 4; @@ -113,18 +144,12 @@ public class SchedulerServiceImplTest { // Test categories with documentation // JUnit 5 provides tags; marker interfaces removed - - private static void configureDebugLogging() { - // Enable debug logging for scheduler package - System.setProperty("org.slf4j.simpleLogger.log.org.apache.unomi.services.impl.scheduler", "DEBUG"); - System.setProperty("org.slf4j.simpleLogger.showDateTime", "true"); - System.setProperty("org.slf4j.simpleLogger.dateTimeFormat", "yyyy-MM-dd HH:mm:ss.SSS"); - System.setProperty("org.slf4j.simpleLogger.showThreadName", "true"); - } + // (An earlier configureDebugLogging() helper set org.slf4j.simpleLogger.* properties here; + // it was dead code — logback-test.xml binds logback, which ignores those. Use + // -DTEST_LOG_LEVEL=DEBUG instead, see the class javadoc.) @BeforeEach public void setUp() throws IOException { - configureDebugLogging(); CustomObjectMapper.getCustomInstance().registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE, ScheduledTask.class); securityService = TestHelper.createSecurityService(); @@ -155,9 +180,10 @@ public void setUp() throws IOException { false, 0); // Set TTL to 0 for immediate purging in tests - // Configure scheduler for testing + // Configure scheduler for testing. The lock timeout matches the production default and + // the multi-node tests' nodes; tests exercising expiry shorten it themselves. schedulerService.setThreadPoolSize(TEST_THREAD_POOL_SIZE); - schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT); + schedulerService.setLockTimeout(DEFAULT_LOCK_TIMEOUT); schedulerService.postConstruct(); } @@ -284,7 +310,10 @@ public void testFixedDelayExecution() throws Exception { .schedule(); assertTrue(executionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should execute three times"); - assertEquals(3, executionCount.get(), "Task should execute exactly three times"); + // Lower bound, not equality: the periodic task keeps firing between the latch release + // and this line, so an exact count is a race against the next period (cf. the fixed-rate + // test above, which already asserts >= for the same reason). + assertTrue(executionCount.get() >= 3, "Task should execute at least three times"); if (workerError.get() != null) { throw new AssertionError("Assertion failed in worker thread", workerError.get()); } @@ -452,7 +481,10 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce @Test @Tag("ClusterTests") public void testClusteringSupport() throws Exception { - // Test clustering behavior with multiple nodes + // Test clustering behavior with multiple nodes. The setUp scheduler is not part of this + // cluster: stop it so it cannot interfere with the three nodes' tasks or the node + // detection markers below (testNodeFailure pattern). + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl nonExecutorNode = TestHelper.createSchedulerService("node3", persistenceService, executionContextManager, bundleContext, clusterService, -1, false, true); @@ -470,7 +502,10 @@ public void testClusteringSupport() throws Exception { persistenceService.refresh(); CountDownLatch exclusiveLatch = new CountDownLatch(1); - CountDownLatch allNodesLatch = new CountDownLatch(3); // one execution observed per node + // Opens on the FIRST runOnAllNodes execution, on whichever node wins the first + // round; allNodesNodes records the distinct winners (see the comment below on why + // "all three nodes" is not a property the implementation promises). + CountDownLatch allNodesLatch = new CountDownLatch(1); Set exclusiveNodes = ConcurrentHashMap.newKeySet(); Set allNodesNodes = ConcurrentHashMap.newKeySet(); @@ -496,9 +531,8 @@ public String getTaskType() { @Override public void execute(ScheduledTask task, TaskStatusCallback callback) { - if (allNodesNodes.add(task.getExecutingNodeId())) { - allNodesLatch.countDown(); - } + allNodesNodes.add(task.getExecutingNodeId()); + allNodesLatch.countDown(); callback.complete(); } }; @@ -531,13 +565,22 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { exclusiveNodes.contains("node3"), "Exclusive task must not execute on a non-executor node"); + // What runOnAllNodes actually promises, as implemented: ANY node - including a + // non-executor - may poll and run the task. It does NOT promise that every node runs + // it: all nodes share the task's single schedule (one lastExecutionDate / + // nextScheduledExecution on one document), so each period has ONE phase-dependent + // winner and there is no fairness across nodes. This test used to demand an execution + // from all three nodes within the timeout, which made it a lottery over checker-tick + // phases - the "runOnAllNodes task should execute on every node" CI flake. The + // non-executor half of the guarantee is pinned deterministically in + // testRunOnAllNodesExecutesOnNonExecutorNode, where the non-executor is the only node. assertTrue( allNodesLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), - "runOnAllNodes task should execute on every node including non-executors"); - assertTrue(allNodesNodes.contains("node1"), "runOnAllNodes should run on node1"); - assertTrue(allNodesNodes.contains("node2"), "runOnAllNodes should run on node2"); - assertTrue(allNodesNodes.contains("node3"), "runOnAllNodes should run on non-executor node3"); + "runOnAllNodes task should execute on at least one node"); + // Keep the lock-inspection task's execution alive until this test has finished + // inspecting its lock, instead of betting on a fixed sleep outlasting the checks. + CountDownLatch lockTaskRelease = new CountDownLatch(1); TaskExecutor clusterLockTestExecutor = new TaskExecutor() { @Override public String getTaskType() { @@ -546,7 +589,7 @@ public String getTaskType() { @Override public void execute(ScheduledTask task, TaskStatusCallback callback) { try { - Thread.sleep(5000); + lockTaskRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT); callback.complete(); } catch (InterruptedException e) { callback.fail(e.getMessage()); @@ -554,39 +597,103 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { } }; - schedulerService.registerTaskExecutor(clusterLockTestExecutor); + // Register on the cluster's own executor nodes (the setUp scheduler is stopped). + node1.registerTaskExecutor(clusterLockTestExecutor); + node2.registerTaskExecutor(clusterLockTestExecutor); - // Test lock management - ScheduledTask lockTask = node1.newTask("cluster-lock-test") - .disallowParallelExecution() - .schedule(); + try { + // Test lock management + ScheduledTask lockTask = node1.newTask("cluster-lock-test") + .disallowParallelExecution() + .schedule(); + + // Refresh persistence to ensure task updates are available (handles refresh delay) + persistenceService.refresh(); + // Wait until the task has a lock owner. Deadline-based rather than + // TestHelper.retryUntil's fixed 20x100ms budget, which a loaded runner exceeds + // (dispatch needs a checker tick plus the simulated refresh delay). + ScheduledTask lockedTask = null; + long lockDeadline = System.currentTimeMillis() + TEST_TIMEOUT; + while (System.currentTimeMillis() < lockDeadline) { + lockedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); + if (lockedTask != null && lockedTask.getLockOwner() != null) { + break; + } + Thread.sleep(100); + } + assertNotNull(lockedTask, "Lock task should be persisted"); + assertNotNull(lockedTask.getLockOwner(), "Task should have lock owner"); + assertNotNull(lockedTask.getLockDate(), "Task should have lock date"); + + // Test lock release - directly update task in persistence + lockedTask.setLockOwner(null); + lockedTask.setLockDate(null); + lockedTask.setLockLeaseMillis(0); + persistenceService.save(lockedTask); + + // Refresh index to ensure changes are visible + persistenceService.refreshIndex(ScheduledTask.class); + + // Get latest state and verify lock release + ScheduledTask releasedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); + assertNull(releasedTask.getLockOwner(), "Lock should be released"); + } finally { + lockTaskRelease.countDown(); + } - // Refresh persistence to ensure task updates are available (handles refresh delay) - persistenceService.refresh(); - // Retry until task has lock owner (handles refresh delay for updates) - ScheduledTask lockedTask = TestHelper.retryUntil( - () -> persistenceService.load(lockTask.getItemId(), ScheduledTask.class), - t -> t != null && t.getLockOwner() != null - ); - assertNotNull(lockedTask.getLockOwner(), "Task should have lock owner"); - assertNotNull(lockedTask.getLockDate(), "Task should have lock date"); + } finally { + node1.preDestroy(); + node2.preDestroy(); + nonExecutorNode.preDestroy(); + } + } - // Test lock release - directly update task in persistence - lockedTask.setLockOwner(null); - lockedTask.setLockDate(null); - persistenceService.save(lockedTask); + /** + * The non-executor half of the runOnAllNodes guarantee, pinned deterministically: a node + * with {@code executorNode=false} must still poll for and execute runOnAllNodes tasks. + *

+ * testClusteringSupport cannot assert this reliably — with executor nodes present, all nodes + * race on the task's single shared schedule and there is no fairness, so whether the + * non-executor ever wins a round within the timeout is checker-phase luck. Here the + * non-executor is the ONLY node, so if it does not poll runOnAllNodes work (the regression + * this pins), nothing executes and the latch times out. + */ + @Test + @Tag("ClusterTests") + public void testRunOnAllNodesExecutesOnNonExecutorNode() throws Exception { + schedulerService.preDestroy(); + SchedulerServiceImpl nonExecutorOnly = TestHelper.createSchedulerService( + "solo-non-executor", persistenceService, executionContextManager, bundleContext, clusterService, -1, false, true); - // Refresh index to ensure changes are visible - persistenceService.refreshIndex(ScheduledTask.class); + try { + CountDownLatch executed = new CountDownLatch(1); + AtomicReference executingNode = new AtomicReference<>(); - // Get latest state and verify lock release - ScheduledTask releasedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); - assertNull(releasedTask.getLockOwner(), "Lock should be released"); + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "all-nodes-solo-test"; + } + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) { + executingNode.set(task.getExecutingNodeId()); + executed.countDown(); + callback.complete(); + } + }; + nonExecutorOnly.registerTaskExecutor(executor); + + nonExecutorOnly.newTask("all-nodes-solo-test") + .runOnAllNodes() + .withPeriod(100, TimeUnit.MILLISECONDS) + .schedule(); + + assertTrue(executed.await(TEST_TIMEOUT, TEST_TIME_UNIT), + "a non-executor node must poll for and run runOnAllNodes tasks"); + assertEquals("solo-non-executor", executingNode.get()); } finally { - node1.preDestroy(); - node2.preDestroy(); - nonExecutorNode.preDestroy(); + nonExecutorOnly.preDestroy(); } } @@ -748,17 +855,29 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { failureLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should fail once"); - // Verify metrics and history + // The 100ms-period task keeps executing (and failing) after the latches fire, so exact + // counts are a race against the next period. Cancel it and wait for the cancellation to + // land before reading anything. + schedulerService.cancelTask(task.getItemId()); + TestHelper.retryUntil( + () -> schedulerService.getTask(task.getItemId()), + t -> t != null && t.getStatus() != ScheduledTask.TaskStatus.RUNNING + && t.getStatus() != ScheduledTask.TaskStatus.SCHEDULED); + + // Verify metrics and history. Successes are exact (the executor only ever completes the + // first two); failures are a lower bound (every later period failed until the cancel won). ScheduledTask finalTask = schedulerService.getTask(task.getItemId()); @SuppressWarnings("unchecked") List> history = (List>) finalTask.getStatusDetails().get("executionHistory"); assertNotNull(history, "Should have execution history"); - assertEquals(3, history.size(), "Should have 3 history entries"); + assertTrue(history.size() >= 3, "Should have at least 3 history entries, had " + history.size()); assertEquals(2, finalTask.getSuccessCount(), "Should have 2 successful executions"); - assertEquals(1, finalTask.getFailureCount(), "Should have 1 failed execution"); - assertEquals(3, finalTask.getSuccessCount() + finalTask.getFailureCount(), "Total executions should be 3"); + assertTrue(finalTask.getFailureCount() >= 1, + "Should have at least 1 failed execution, had " + finalTask.getFailureCount()); + // No history-size == successCount+failureCount equality here: an execution in flight + // while the cancel lands may or may not get its failure recorded, by design. // Verify history entries int successEntries = 0; @@ -774,7 +893,7 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { } assertEquals(2, successEntries, "Should have 2 successful executions"); - assertEquals(1, failureEntries, "Should have 1 failed execution"); + assertTrue(failureEntries >= 1, "Should have at least 1 failed execution"); // Verify metrics assertTrue(schedulerService.getMetric("tasks.completed") > 0, "Should have completed tasks metric"); @@ -932,12 +1051,34 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { 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 @@ -1023,13 +1164,24 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce * tasks that already executed, stranding the task in CRASHED state forever. The * execution manager must recognize that the execution it owns is still alive, reclaim * the task and process the failure (and its retry) normally. + * + *

NOTE: since lock renewal was introduced (the lock is re-stamped every lockTimeout/3 + * while the executor runs), a stalled-but-live execution's lock no longer expires from + * natural timing, so the CRASH-mark this test was written around does not fire anymore - + * verified from the LOCK-DIAG traces: renewal succeeds throughout the stall and no expiry + * verdict ever triggers. The test remains valuable as a pin on the surviving behaviour + * (a failure reported after a stall longer than the lock timeout still schedules its + * retries and completes), and its assertions were already written to tolerate both worlds + * (>= 3 executions). The reclaim path itself is now only reachable when renewal genuinely + * stops (e.g. a GC pause longer than the full lease) and is covered at unit level in + * TaskExecutionManagerTest. */ @Test @Tag("RetryTests") public void testOneShotRetryAfterRecoveryMarksLiveExecutionCrashed() throws Exception { - // setUp() only sets the lock timeout on the scheduler service; the lock manager - // created by TestHelper keeps its 10s default. Shorten it here so a stalled - // execution's lock actually expires within this test's stall window. + // Shorten the lock timeout so the stall below dwarfs it. (setLockTimeout on the service + // propagates to the lock manager as well; setting the lock manager directly is + // equivalent and kept for clarity about what the timeout is FOR here.) schedulerService.getLockManager().setLockTimeout(TEST_LOCK_TIMEOUT); CountDownLatch completionLatch = new CountDownLatch(1); @@ -1320,6 +1472,7 @@ public void testLockTimeout() throws Exception { schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT); CountDownLatch executionLatch = new CountDownLatch(1); + CountDownLatch holdRelease = new CountDownLatch(1); AtomicBoolean taskStarted = new AtomicBoolean(false); TaskExecutor executor = new TaskExecutor() { @@ -1335,8 +1488,9 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { taskStarted.set(true); executionLatch.countDown(); - // Hold the lock longer than timeout - Thread.sleep(TEST_LOCK_TIMEOUT * 2); + // Hold the lock until the test has finished inspecting it - a latch the + // test releases, not a fixed sleep the test hopes is long enough. + holdRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT); callback.complete(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -1359,12 +1513,15 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { // Directly update task to simulate lock expiration runningTask.setLockOwner(null); runningTask.setLockDate(null); + runningTask.setLockLeaseMillis(0); persistenceService.save(runningTask); persistenceService.refreshIndex(ScheduledTask.class); // Check lock status after manual release ScheduledTask updatedTask = persistenceService.load(task.getItemId(), ScheduledTask.class); assertNull(updatedTask.getLockOwner(), "Lock should be released after manual update"); + + holdRelease.countDown(); } /** @@ -1631,6 +1788,12 @@ public boolean canResume(ScheduledTask task) { @Test @Tag("ClusterTests") public void testConcurrentLockAcquisition() throws Exception { + // This test is about node1/node2 only: stop the setUp scheduler so the "two-node" cluster + // really has two nodes (testNodeFailure pattern). It used to stay up with a 1s lock + // timeout against these nodes' 10s, declare their live locks expired between renewals, + // and mark the running task CRASHED — which a peer then re-dispatched concurrently + // (the "expected: <1> but was: <2>" CI flake). + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -1738,6 +1901,8 @@ private static class ExecutionInfo { @Test @Tag("ClusterTests") public void testTaskRebalancing() throws Exception { + // Two-node test: stop the setUp scheduler so it is not a hidden third participant. + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = null; try { @@ -1861,6 +2026,8 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { @Test @Tag("ClusterTests") public void testLockStealing() throws Exception { + // Two-node test: stop the setUp scheduler so it is not a hidden third participant. + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -1946,6 +2113,10 @@ public void testNewTaskDefaultsToExclusiveLocking() { @Test public void testNodeAffinity() throws Exception { + // Three-node test. Stop the setUp scheduler: getActiveNodes() falls back to scanning + // tasks with recent locks, and a foreign recovery pass that clears the detection tasks' + // locks below would silently shrink the cluster this test asserts on. + schedulerService.preDestroy(); // Create test nodes with cluster service SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -2425,7 +2596,9 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { newSchedulerService.preDestroy(); assertTrue(executed, "Task should execute after scheduler restart"); - assertEquals(2, executionCount.get(), "Task should have executed twice"); + // Lower bound: the 500ms-period task may legitimately fire again between the latch + // release and preDestroy() completing on a slow runner. + assertTrue(executionCount.get() >= 2, "Task should have executed at least twice"); // Verify the reloaded task has same ID ScheduledTask reloadedTask = persistenceService.load(persistentTask.getItemId(), ScheduledTask.class); @@ -2757,10 +2930,13 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { assertTrue( secondExecutionLatch.await(TEST_TIMEOUT * 2, TEST_TIME_UNIT), "Task should execute after restart with dedicated executor"); - assertEquals(2, executionCount.get(), "Task should execute twice"); - // Clean up + // Stop the scheduler before asserting the count, exactly like the first assertion above: + // the task runs at 100ms fixed rate, so a third tick can fire between the latch release + // and the assert. After preDestroy() the count is stable; >= tolerates a tick that + // squeezed in before shutdown took effect. newSchedulerService.preDestroy(); + assertTrue(executionCount.get() >= 2, "Task should execute at least twice"); } /** 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 bafcc53565..8aa3105b04 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.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; @@ -45,6 +49,7 @@ */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class TaskExecutionManagerTest { private static final String NODE = "exec-node"; @@ -91,6 +96,62 @@ public void tearDown() { executionManager.shutdown(); } + /** + * A terminal handler must increment counters from the STORE's values, not from the possibly + * stale copy the wrapper is carrying. + *

+ * 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 storeDetails = new HashMap<>(); + List> 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> history = + (List>) 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"); @@ -128,6 +189,19 @@ public void testPrepareForExecutionFailsWhenLockDenied() { assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); } + /** + * Waits for the wrapper's asynchronous terminal transition to land on the shared task object. + * The executor's callback returns before the wrapper finishes its bookkeeping, so asserting + * the final status right after the latch (or after a fixed sleep) races the wrapper thread. + */ + private static void awaitStatus(ScheduledTask task, ScheduledTask.TaskStatus expected, long timeoutMs) + throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (task.getStatus() != expected && System.currentTimeMillis() < deadline) { + Thread.sleep(20); + } + } + @Test public void testExecuteTaskDuplicateDispatchIsSkipped() throws Exception { CountDownLatch started = new CountDownLatch(1); @@ -148,6 +222,9 @@ public void testExecuteTaskDuplicateDispatchIsSkipped() throws Exception { // Second dispatch while claim held executionManager.executeTask(task, executor); release.countDown(); + // Deliberate quiet window for a NEGATIVE assertion: a wrongly accepted duplicate + // dispatch would start within milliseconds. Too short can only miss a violation + // (false green), never fail a healthy run. Thread.sleep(200); assertEquals(1, runs.get()); } @@ -273,7 +350,7 @@ public void testHandleTaskErrorOneShotExhaustsRetriesStaysFailed() throws Except executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.FAILED, 5000); assertEquals(ScheduledTask.TaskStatus.FAILED, task.getStatus()); assertEquals(1, task.getFailureCount()); assertTrue(task.isEnabled()); @@ -298,7 +375,7 @@ public void testHandleTaskErrorPeriodicResetsFailureCountAfterExhaustion() throw executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(0, task.getFailureCount()); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertNotNull(task.getNextScheduledExecution()); @@ -340,7 +417,7 @@ public void testHandleTaskErrorSkipsRetryScheduleDuringShutdown() throws Excepti releaser.start(); executionManager.shutdown(); releaser.join(2000); - Thread.sleep(200); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertEquals(1, task.getFailureCount()); // No second attempt — retry schedule skipped after scheduler shutdown @@ -362,7 +439,7 @@ public void testHandleTaskCompletionOneShotDisablesAndClearsNext() throws Except executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); assertFalse(task.isEnabled()); assertNull(task.getNextScheduledExecution()); @@ -384,7 +461,7 @@ public void testHandleTaskCompletionPeriodicReschedules() throws Exception { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertNotNull(task.getNextScheduledExecution()); assertTrue(task.getNextScheduledExecution().getTime() >= before + 5_000); @@ -406,7 +483,7 @@ public void testHandleTaskCompletionPeriodZeroDoesNotReschedule() throws Excepti executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); } @@ -427,6 +504,8 @@ public void testCompletionAndErrorIgnoredWhenNotRunning() throws Exception { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); + // Deliberate quiet window for a NEGATIVE assertion (callbacks must have been ignored); + // a poll cannot confirm that nothing happened. Thread.sleep(100); assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); assertEquals(completedBefore, metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED)); @@ -448,7 +527,7 @@ public void testReclaimPrematureCrashBeforeCompletion() throws Exception { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); assertFalse(task.isEnabled()); } @@ -545,12 +624,12 @@ public void testTerminalCompleteSkippedWhenCancelledInStore() throws Exception { ScheduledTask task = TaskTestFixtures.baseTask("cancel-race"); executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); - assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); // persistTerminalState() is skipped (terminal transition correctly bailed out above), but // the wrapper's cleanup still CAS-clears executingNodeId once; that write is expected to // fail harmlessly against a real store since the document moved on to CANCELLED. - verify(schedulerService, times(1)).saveTaskWithRefresh(any()); + // timeout() waits for the asynchronous cleanup instead of betting a fixed sleep on it. + verify(schedulerService, timeout(5000).times(1)).saveTaskWithRefresh(any()); + assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); assertEquals(0, metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED)); } @@ -572,12 +651,12 @@ public void testTerminalCompleteSkippedWhenPeerHoldsLock() throws Exception { ScheduledTask task = TaskTestFixtures.baseTask("peer-lock"); executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); - assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus()); // persistTerminalState() is skipped (peer holds the lock), but the wrapper's cleanup still // CAS-clears executingNodeId once; that write is expected to fail harmlessly against a real // store since the peer is the authoritative owner. - verify(schedulerService, times(1)).saveTaskWithRefresh(any()); + // timeout() waits for the asynchronous cleanup instead of betting a fixed sleep on it. + verify(schedulerService, timeout(5000).times(1)).saveTaskWithRefresh(any()); + assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus()); } @Test @@ -600,9 +679,11 @@ public void testAbortPreparedExecutionOnShutdownReleasesLockAndMarksCrashed() th }; ScheduledTask task = TaskTestFixtures.baseTask("abort-prep"); executionManager.executeTask(task, executor); - Thread.sleep(300); - assertEquals(1, executed.getCount(), "executor must not run after shutdown-abort"); + // Positive half: wait for the asynchronous abort to land instead of a fixed sleep. + awaitStatus(task, ScheduledTask.TaskStatus.CRASHED, 5000); assertEquals(ScheduledTask.TaskStatus.CRASHED, task.getStatus()); + // Negative half: the executor must never have run (green-direction check). + assertEquals(1, executed.getCount(), "executor must not run after shutdown-abort"); assertNull(task.getLockOwner()); verify(schedulerService, atLeastOnce()).saveTask(any(ScheduledTask.class), eq(true)); } 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 2497838845..57cec8d58e 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 @@ -44,6 +44,7 @@ */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class TaskLockManagerTest { private static final String NODE = "lock-node"; @@ -216,6 +217,188 @@ public void testIsLockExpiredFalseWhenAgeEqualsTimeout() { 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 nodes = Arrays.asList("aaa-node", NODE, "zzz-node");