feat(ext-scheduler): add interval and cron background jobs
jobs.every(Duration.ofMinutes(5), reports::refresh);
jobs.cron("0 0 3 * * *", archive::sweep);
One daemon platform thread keeps time; every job body runs on a virtual thread,
so a slow job delays nothing but its own next run and cannot occupy the timer.
Cron expressions compile once into a bitmask per field — a long for seconds and
minutes, an int for the rest — so matching an instant is a shift and a mask
rather than a parse or a set lookup. Next-fire advances by the largest unit
that cannot match instead of ticking second by second, so a yearly expression
resolves in a few dozen iterations rather than thirty million. Five or six
fields, ranges, steps, lists, named months and weekdays, both Sunday encodings,
and the standard union semantics when both day fields are restricted. A
malformed expression throws when the job is registered, not when it would have
fired.
Overlapping runs are skipped, and deliberately not configurable: two copies of
one job at once is a bug in every case anyone has needed, and a flag would only
let it be set wrongly. A skipped run logs how long the previous one has been
going.
A throwing job is logged and keeps its schedule. scheduleAtFixedRate cancels
the task on first exception, silently — a job that dies at 3am and is never
heard from again is the failure this avoids.
Shutdown goes through FlashContext.onClose, so app.stop() drains HTTP first,
then gives running jobs a grace period before forcing them down. Nothing runs
after the app stops. The grace period is the only knob.
Known ceiling, marked in the source: schedules are per-instance, so two
replicas run every job twice. A distributed lock should not exist until there
is a second replica.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Zakaria El Orche
co-authored by
Claude Opus 5
Claude Opus 5
parent
58bae41f7a
commit
5c163b7f8d
@@ -0,0 +1,125 @@
|
||||
# flash-ext-scheduler
|
||||
|
||||
Background jobs on an interval or a cron schedule. One platform thread keeps time, every job body
|
||||
runs on a virtual thread, and the whole thing stops with the app.
|
||||
|
||||
## What it provides
|
||||
|
||||
| Component | Description |
|
||||
|---|---|
|
||||
| `Scheduler` | `every(interval, work)` and `cron(expression, work)` |
|
||||
| `Cron` | A cron expression compiled to bitmasks; usable on its own |
|
||||
|
||||
## Dependency
|
||||
|
||||
```xml
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-scheduler</artifactId>
|
||||
<version>${flash.version}</version>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
## Quick start
|
||||
|
||||
```java
|
||||
FlashApp.create(8080)
|
||||
.install(new SchedulerExtension())
|
||||
.apply(new BlogApp())
|
||||
.start();
|
||||
```
|
||||
|
||||
```java
|
||||
// inside BlogApp.configure(app)
|
||||
app.ctx().onReady(() -> {
|
||||
Scheduler jobs = app.ctx().require(Scheduler.class);
|
||||
|
||||
jobs.every(Duration.ofMinutes(5), reports::refresh);
|
||||
jobs.cron("0 0 3 * * *", archive::sweep);
|
||||
});
|
||||
```
|
||||
|
||||
`onReady` is the right place: the service graph is resolved by then, so a job can close over the
|
||||
services it needs.
|
||||
|
||||
Give a job an explicit name when the logs should be readable — a method reference names itself, a
|
||||
lambda cannot:
|
||||
|
||||
```java
|
||||
jobs.every("refresh-reports", Duration.ofMinutes(5), () -> reports.refresh());
|
||||
jobs.cron("nightly-archive", "0 0 3 * * *", () -> archive.sweep());
|
||||
```
|
||||
|
||||
`jobNames()` returns them in registration order, for an ops or health endpoint.
|
||||
|
||||
## Cron expressions
|
||||
|
||||
Five fields (`min hour dom mon dow`) or six with a leading seconds field.
|
||||
|
||||
| Expression | Fires |
|
||||
|---|---|
|
||||
| `0 0 3 * * *` | 03:00:00 daily |
|
||||
| `*/15 * * * *` | every 15 minutes |
|
||||
| `0 9 * * MON-FRI` | 09:00 on weekdays |
|
||||
| `0 0 1 MAR *` | midnight on 1 March |
|
||||
| `0,30 * * * *` | on the hour and the half hour |
|
||||
|
||||
Supports `*`, `?`, single values, `a-b` ranges, `a/n` steps, comma lists, and three-letter month
|
||||
and day names. Both `0` and `7` mean Sunday.
|
||||
|
||||
Standard cron semantics for the two day fields: when **both** are restricted, a match is their
|
||||
**union** — `0 0 1 * MON` means "the 1st, or any Monday", not "a Monday that is the 1st".
|
||||
|
||||
A malformed expression throws when you register the job, not the first time it would have fired.
|
||||
|
||||
## Overlapping runs are skipped
|
||||
|
||||
Not configurable. If a run is still going when the next is due, the next is skipped and a WARN
|
||||
records how long the previous one has been running.
|
||||
|
||||
Two copies of the same job running at once is a bug in every case anyone has needed so far, and a
|
||||
flag would only let it be configured wrongly. If you genuinely want concurrent runs, register the
|
||||
job twice under different names.
|
||||
|
||||
## Failure
|
||||
|
||||
A throwing job is logged at ERROR and keeps its schedule. A raw `scheduleAtFixedRate` cancels the
|
||||
task on the first exception, silently — a job that dies at 3am and is never heard from again is
|
||||
the failure mode this avoids.
|
||||
|
||||
## Shutdown
|
||||
|
||||
The scheduler is registered with `FlashContext.onClose`, so `app.stop()` drains in-flight HTTP
|
||||
requests first, then gives running jobs their grace period (10s by default) before forcing them
|
||||
down. Nothing runs after the app has stopped.
|
||||
|
||||
```java
|
||||
new SchedulerExtension(Duration.ofSeconds(30)) // longer grace for slow jobs
|
||||
```
|
||||
|
||||
That is the only knob.
|
||||
|
||||
## Threading
|
||||
|
||||
`Scheduler` holds one daemon platform thread for timing and dispatches every job body to a virtual
|
||||
thread. A slow job delays nothing but its own next run; it cannot occupy the timer or starve other
|
||||
jobs.
|
||||
|
||||
## Performance
|
||||
|
||||
A cron expression is parsed **once**, into a bitmask per field — a `long` for seconds and minutes,
|
||||
an `int` for the rest. Matching a candidate instant is a shift and a mask, not a parse or a set
|
||||
lookup. Computing the next fire time advances by the largest unit that cannot match rather than
|
||||
ticking second by second, so a yearly expression resolves in a few dozen iterations rather than
|
||||
thirty million.
|
||||
|
||||
To be precise about where that matters: this runs once per fire, not once per request. The
|
||||
allocation discipline elsewhere in Flash is about the request hot path, and a scheduler that
|
||||
computes a `ZonedDateTime` a few times an hour is not on it. The bitmasks are here because parsing
|
||||
a string on every candidate instant would be genuinely wasteful, not to shave an allocation.
|
||||
|
||||
## Known ceiling
|
||||
|
||||
**Schedules are per-instance.** Two replicas run every job twice. The fix is a distributed lock,
|
||||
and it should not exist until there is a second replica — it is marked with a `ponytail:` comment
|
||||
in `SchedulerExtension`.
|
||||
@@ -0,0 +1,38 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-extensions</artifactId>
|
||||
<version>2.1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>flash-ext-scheduler</artifactId>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-testing</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
+182
@@ -0,0 +1,182 @@
|
||||
package dev.relism.flash.ext.scheduler;
|
||||
|
||||
import java.time.DayOfWeek;
|
||||
import java.time.ZonedDateTime;
|
||||
|
||||
/**
|
||||
* A cron expression compiled to bitmasks.
|
||||
*
|
||||
* <p>Accepts five fields ({@code min hour dom mon dow}) or six with a leading seconds field.
|
||||
* Each is parsed once into a bitmask — a {@code long} for seconds and minutes, an {@code int} for
|
||||
* the rest — so matching a candidate instant is a shift and a mask rather than a parse, a list
|
||||
* walk or a set lookup.
|
||||
*
|
||||
* <pre>{@code
|
||||
* Cron.parse("0 0 3 * * *") // 03:00:00 every day
|
||||
* Cron.parse("*}{@code /15 * * * *") // every 15 minutes
|
||||
* Cron.parse("0 9 * * MON-FRI") // 09:00 on weekdays
|
||||
* }</pre>
|
||||
*/
|
||||
public final class Cron {
|
||||
|
||||
private final long seconds;
|
||||
private final long minutes;
|
||||
private final int hours;
|
||||
private final int daysOfMonth;
|
||||
private final int months;
|
||||
private final int daysOfWeek;
|
||||
private final boolean anyDayOfMonth;
|
||||
private final boolean anyDayOfWeek;
|
||||
private final String source;
|
||||
|
||||
private Cron(long seconds, long minutes, int hours, int daysOfMonth, int months, int daysOfWeek,
|
||||
boolean anyDayOfMonth, boolean anyDayOfWeek, String source) {
|
||||
this.seconds = seconds;
|
||||
this.minutes = minutes;
|
||||
this.hours = hours;
|
||||
this.daysOfMonth = daysOfMonth;
|
||||
this.months = months;
|
||||
this.daysOfWeek = daysOfWeek;
|
||||
this.anyDayOfMonth = anyDayOfMonth;
|
||||
this.anyDayOfWeek = anyDayOfWeek;
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
/** @throws IllegalArgumentException if the expression is malformed — at boot, not at fire time */
|
||||
public static Cron parse(String expression) {
|
||||
String[] fields = expression.trim().split("\\s+");
|
||||
if (fields.length != 5 && fields.length != 6)
|
||||
throw new IllegalArgumentException(
|
||||
"Cron needs 5 fields (min hour dom mon dow) or 6 with seconds, got " + fields.length
|
||||
+ ": " + expression);
|
||||
|
||||
int offset = fields.length == 6 ? 1 : 0;
|
||||
long secondMask = offset == 1 ? mask(fields[0], 0, 59, null) : 1L; // no seconds field: fire at :00
|
||||
return new Cron(
|
||||
secondMask,
|
||||
mask(fields[offset], 0, 59, null),
|
||||
(int) mask(fields[offset + 1], 0, 23, null),
|
||||
(int) mask(fields[offset + 2], 1, 31, null),
|
||||
(int) mask(fields[offset + 3], 1, 12, MONTHS),
|
||||
(int) mask(fields[offset + 4], 0, 6, DAYS),
|
||||
isWildcard(fields[offset + 2]),
|
||||
isWildcard(fields[offset + 4]),
|
||||
expression);
|
||||
}
|
||||
|
||||
/**
|
||||
* The first instant strictly after {@code from} that matches.
|
||||
*
|
||||
* <p>Advances by the largest unit that cannot match rather than ticking second by second, so a
|
||||
* yearly expression resolves in a few dozen iterations instead of thirty million.
|
||||
*/
|
||||
public ZonedDateTime nextAfter(ZonedDateTime from) {
|
||||
ZonedDateTime candidate = from.plusSeconds(1).withNano(0);
|
||||
// Four years covers the longest gap a valid expression can produce (29 February).
|
||||
for (int guard = 0; guard < 4 * 366 * 24; guard++) {
|
||||
if (!isSet(months, candidate.getMonthValue())) {
|
||||
candidate = candidate.plusMonths(1).withDayOfMonth(1).toLocalDate()
|
||||
.atStartOfDay(candidate.getZone());
|
||||
continue;
|
||||
}
|
||||
if (!dayMatches(candidate)) {
|
||||
candidate = candidate.plusDays(1).toLocalDate().atStartOfDay(candidate.getZone());
|
||||
continue;
|
||||
}
|
||||
if (!isSet(hours, candidate.getHour())) {
|
||||
candidate = candidate.plusHours(1).withMinute(0).withSecond(0);
|
||||
continue;
|
||||
}
|
||||
if (!isSet(minutes, candidate.getMinute())) {
|
||||
candidate = candidate.plusMinutes(1).withSecond(0);
|
||||
continue;
|
||||
}
|
||||
if (!isSet(seconds, candidate.getSecond())) {
|
||||
candidate = candidate.plusSeconds(1);
|
||||
continue;
|
||||
}
|
||||
return candidate;
|
||||
}
|
||||
throw new IllegalStateException("Cron expression never fires: " + source);
|
||||
}
|
||||
|
||||
@Override public String toString() { return source; }
|
||||
|
||||
/**
|
||||
* Standard cron semantics: when both day fields are restricted the match is a union, not an
|
||||
* intersection — "1st of the month, or any Monday".
|
||||
*/
|
||||
private boolean dayMatches(ZonedDateTime candidate) {
|
||||
boolean dom = isSet(daysOfMonth, candidate.getDayOfMonth());
|
||||
boolean dow = isSet(daysOfWeek, candidate.getDayOfWeek() == DayOfWeek.SUNDAY
|
||||
? 0 : candidate.getDayOfWeek().getValue());
|
||||
if (anyDayOfMonth && anyDayOfWeek) return true;
|
||||
if (anyDayOfMonth) return dow;
|
||||
if (anyDayOfWeek) return dom;
|
||||
return dom || dow;
|
||||
}
|
||||
|
||||
private static boolean isSet(long mask, int value) { return (mask & (1L << value)) != 0; }
|
||||
private static boolean isSet(int mask, int value) { return (mask & (1 << value)) != 0; }
|
||||
|
||||
private static boolean isWildcard(String field) { return "*".equals(field) || "?".equals(field); }
|
||||
|
||||
private static final String[] MONTHS =
|
||||
{"JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC"};
|
||||
private static final String[] DAYS = {"SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"};
|
||||
|
||||
/** Parses one field into a bitmask: {@code *}, {@code a}, {@code a-b}, {@code a/n}, and lists. */
|
||||
private static long mask(String field, int min, int max, String[] names) {
|
||||
long bits = 0;
|
||||
for (String part : field.split(",")) {
|
||||
int step = 1;
|
||||
int slash = part.indexOf('/');
|
||||
if (slash >= 0) {
|
||||
step = Integer.parseInt(part.substring(slash + 1));
|
||||
if (step <= 0) throw new IllegalArgumentException("Cron step must be positive: " + field);
|
||||
part = part.substring(0, slash);
|
||||
}
|
||||
int from;
|
||||
int to;
|
||||
if (isWildcard(part)) {
|
||||
from = min;
|
||||
to = max;
|
||||
} else {
|
||||
int dash = part.indexOf('-');
|
||||
if (dash > 0) {
|
||||
from = value(part.substring(0, dash), names, min, max, field);
|
||||
to = value(part.substring(dash + 1), names, min, max, field);
|
||||
} else {
|
||||
from = value(part, names, min, max, field);
|
||||
to = slash >= 0 ? max : from;
|
||||
}
|
||||
}
|
||||
if (from > to) throw new IllegalArgumentException("Cron range is inverted: " + field);
|
||||
for (int v = from; v <= to; v += step) bits |= 1L << v;
|
||||
}
|
||||
if (bits == 0) throw new IllegalArgumentException("Cron field matches nothing: " + field);
|
||||
return bits;
|
||||
}
|
||||
|
||||
private static int value(String token, String[] names, int min, int max, String field) {
|
||||
int parsed = -1;
|
||||
if (names != null) {
|
||||
String upper = token.toUpperCase();
|
||||
for (int i = 0; i < names.length; i++) {
|
||||
if (names[i].equals(upper)) { parsed = i + (names == MONTHS ? 1 : 0); break; }
|
||||
}
|
||||
}
|
||||
if (parsed < 0) {
|
||||
try {
|
||||
parsed = Integer.parseInt(token);
|
||||
} catch (NumberFormatException malformed) {
|
||||
throw new IllegalArgumentException("Cron field is not a number: " + field, malformed);
|
||||
}
|
||||
}
|
||||
if (parsed == 7 && names == DAYS) parsed = 0; // both 0 and 7 mean Sunday
|
||||
if (parsed < min || parsed > max)
|
||||
throw new IllegalArgumentException("Cron value " + parsed + " out of range " + min + "-" + max
|
||||
+ " in: " + field);
|
||||
return parsed;
|
||||
}
|
||||
}
|
||||
+149
@@ -0,0 +1,149 @@
|
||||
package dev.relism.flash.ext.scheduler;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.ZonedDateTime;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* Runs background jobs on an interval or a cron expression.
|
||||
*
|
||||
* <pre>{@code
|
||||
* Scheduler jobs = ctx.require(Scheduler.class);
|
||||
*
|
||||
* jobs.every(Duration.ofMinutes(5), reports::refresh);
|
||||
* jobs.cron("0 0 3 * * *", archive::sweep);
|
||||
* }</pre>
|
||||
*
|
||||
* <p>One platform thread keeps time; every job body runs on a virtual thread, so a slow job
|
||||
* blocks nothing but itself.
|
||||
*
|
||||
* <p><b>Overlapping runs are skipped, always.</b> Not an option — two copies of the same job
|
||||
* running at once is a bug in every case anyone has needed so far, and a flag would only let it
|
||||
* be configured wrongly. A skipped run is logged at WARN with how long the previous one has been
|
||||
* going.
|
||||
*/
|
||||
@Slf4j
|
||||
public final class Scheduler {
|
||||
|
||||
private final ScheduledExecutorService clock;
|
||||
private final ScheduledExecutorService workers;
|
||||
private final List<Job> jobs = new CopyOnWriteArrayList<>();
|
||||
private volatile boolean stopped;
|
||||
|
||||
Scheduler() {
|
||||
ThreadFactory timing = runnable -> {
|
||||
Thread thread = new Thread(runnable, "flash-scheduler");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
};
|
||||
this.clock = Executors.newSingleThreadScheduledExecutor(timing);
|
||||
this.workers = Executors.newScheduledThreadPool(0, Thread.ofVirtual().factory());
|
||||
}
|
||||
|
||||
/** Runs {@code work} every {@code interval}, starting one interval from now. */
|
||||
public void every(Duration interval, Runnable work) {
|
||||
every(callerName(work), interval, work);
|
||||
}
|
||||
|
||||
/** Runs {@code work} every {@code interval} under an explicit name, used in logs. */
|
||||
public void every(String name, Duration interval, Runnable work) {
|
||||
if (interval.isZero() || interval.isNegative())
|
||||
throw new IllegalArgumentException("Interval must be positive for job '" + name + "'");
|
||||
Job job = new Job(name, work);
|
||||
jobs.add(job);
|
||||
long millis = interval.toMillis();
|
||||
clock.scheduleAtFixedRate(() -> dispatch(job), millis, millis, TimeUnit.MILLISECONDS);
|
||||
log.info("Scheduled '{}' every {}", name, interval);
|
||||
}
|
||||
|
||||
/** Runs {@code work} on a cron schedule. The expression is validated now, not at fire time. */
|
||||
public void cron(String expression, Runnable work) {
|
||||
cron(callerName(work), expression, work);
|
||||
}
|
||||
|
||||
/** Runs {@code work} on a cron schedule under an explicit name. */
|
||||
public void cron(String name, String expression, Runnable work) {
|
||||
Cron cron = Cron.parse(expression);
|
||||
Job job = new Job(name, work);
|
||||
jobs.add(job);
|
||||
scheduleNext(job, cron);
|
||||
log.info("Scheduled '{}' at cron [{}]", name, expression);
|
||||
}
|
||||
|
||||
/** Job names in registration order — for a health or ops endpoint. */
|
||||
public List<String> jobNames() {
|
||||
return jobs.stream().map(job -> job.name).toList();
|
||||
}
|
||||
|
||||
private void scheduleNext(Job job, Cron cron) {
|
||||
if (stopped) return;
|
||||
ZonedDateTime now = ZonedDateTime.now();
|
||||
long delay = Math.max(1, Duration.between(now, cron.nextAfter(now)).toMillis());
|
||||
clock.schedule(() -> {
|
||||
dispatch(job);
|
||||
scheduleNext(job, cron);
|
||||
}, delay, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
private void dispatch(Job job) {
|
||||
if (stopped) return;
|
||||
if (!job.running.compareAndSet(false, true)) {
|
||||
log.warn("Skipping '{}': previous run started {} ago and is still going",
|
||||
job.name, Duration.ofNanos(System.nanoTime() - job.startedAtNanos));
|
||||
return;
|
||||
}
|
||||
job.startedAtNanos = System.nanoTime();
|
||||
workers.execute(() -> {
|
||||
try {
|
||||
job.work.run();
|
||||
} catch (RuntimeException failure) {
|
||||
// A throwing job must not kill its schedule the way a raw scheduleAtFixedRate would.
|
||||
log.error("Job '{}' failed", job.name, failure);
|
||||
} finally {
|
||||
job.running.set(false);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Called by {@link SchedulerExtension} through {@code FlashContext.onClose}. */
|
||||
void shutdown(Duration grace) {
|
||||
stopped = true;
|
||||
clock.shutdownNow();
|
||||
workers.shutdown();
|
||||
try {
|
||||
if (!workers.awaitTermination(grace.toMillis(), TimeUnit.MILLISECONDS)) {
|
||||
log.warn("Scheduler jobs still running after {} — forcing shutdown", grace);
|
||||
workers.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException interrupted) {
|
||||
Thread.currentThread().interrupt();
|
||||
workers.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
||||
/** Method references carry a usable name; lambdas do not, so those get a positional one. */
|
||||
private String callerName(Runnable work) {
|
||||
String simple = work.getClass().getSimpleName();
|
||||
return simple.isEmpty() || simple.contains("$$Lambda") ? "job-" + (jobs.size() + 1) : simple;
|
||||
}
|
||||
|
||||
private static final class Job {
|
||||
final String name;
|
||||
final Runnable work;
|
||||
final AtomicBoolean running = new AtomicBoolean();
|
||||
volatile long startedAtNanos;
|
||||
|
||||
Job(String name, Runnable work) {
|
||||
this.name = name;
|
||||
this.work = work;
|
||||
}
|
||||
}
|
||||
}
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
package dev.relism.flash.ext.scheduler;
|
||||
|
||||
import dev.relism.flash.extension.FlashContext;
|
||||
import dev.relism.flash.extension.FlashExtension;
|
||||
import dev.relism.flash.extension.FlashRegistrar;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Installs the {@link Scheduler}.
|
||||
*
|
||||
* <pre>{@code
|
||||
* FlashApp.create(8080)
|
||||
* .install(new SchedulerExtension())
|
||||
* .apply(new BlogApp())
|
||||
* .start();
|
||||
* }</pre>
|
||||
*
|
||||
* <p>No configuration beyond an optional shutdown grace period. The scheduler starts with the app
|
||||
* and stops with it: shutdown is registered through {@link FlashContext#onClose}, so
|
||||
* {@code app.stop()} drains in-flight requests first, then gives running jobs their grace period
|
||||
* before forcing them down. Without that a job would outlive the app that owns it.
|
||||
*
|
||||
* <p>ponytail: schedules are per-instance. Two replicas run every job twice. The fix is a
|
||||
* distributed lock, and it should not exist until there is a second replica.
|
||||
*/
|
||||
public final class SchedulerExtension implements FlashExtension {
|
||||
|
||||
private static final Duration DEFAULT_GRACE = Duration.ofSeconds(10);
|
||||
|
||||
private final Duration shutdownGrace;
|
||||
|
||||
public SchedulerExtension() {
|
||||
this(DEFAULT_GRACE);
|
||||
}
|
||||
|
||||
/** @param shutdownGrace how long {@code stop()} waits for running jobs before forcing them */
|
||||
public SchedulerExtension(Duration shutdownGrace) {
|
||||
this.shutdownGrace = shutdownGrace;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void configure(FlashRegistrar<?> app, FlashContext ctx) {
|
||||
ctx.supply(Scheduler.class, services -> {
|
||||
Scheduler scheduler = new Scheduler();
|
||||
services.onClose(() -> scheduler.shutdown(shutdownGrace));
|
||||
return scheduler;
|
||||
});
|
||||
}
|
||||
}
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
package dev.relism.flash.ext.scheduler;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.ZoneOffset;
|
||||
import java.time.ZonedDateTime;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class CronTest {
|
||||
|
||||
private static ZonedDateTime at(String isoLocal) {
|
||||
return ZonedDateTime.parse(isoLocal + "Z", java.time.format.DateTimeFormatter.ISO_ZONED_DATE_TIME);
|
||||
}
|
||||
|
||||
private static String next(String expression, String from) {
|
||||
return Cron.parse(expression).nextAfter(at(from)).withZoneSameInstant(ZoneOffset.UTC).toString();
|
||||
}
|
||||
|
||||
@Test
|
||||
void dailyAtAFixedTime() {
|
||||
assertEquals("2026-01-01T03:00Z", next("0 0 3 * * *", "2026-01-01T02:59:59"));
|
||||
assertEquals("2026-01-02T03:00Z", next("0 0 3 * * *", "2026-01-01T03:00:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void fiveFieldExpressionsFireOnTheMinute() {
|
||||
assertEquals("2026-01-01T03:00Z", next("0 3 * * *", "2026-01-01T02:59:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void stepsAndLists() {
|
||||
assertEquals("2026-01-01T00:15Z", next("*/15 * * * *", "2026-01-01T00:01:00"));
|
||||
assertEquals("2026-01-01T00:30Z", next("0,30 * * * *", "2026-01-01T00:15:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void namedMonthsAndWeekdays() {
|
||||
assertEquals("2026-01-05T09:00Z", next("0 9 * * MON-FRI", "2026-01-03T10:00:00"));
|
||||
assertEquals("2026-03-01T00:00Z", next("0 0 1 MAR *", "2026-01-31T00:00:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothSundayEncodingsWork() {
|
||||
assertEquals(next("0 0 * * 0", "2026-01-01T00:00:00"), next("0 0 * * 7", "2026-01-01T00:00:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void restrictedDayFieldsAreAUnionNotAnIntersection() {
|
||||
// "1st of the month, or any Monday" — standard cron semantics.
|
||||
assertEquals("2026-01-05T00:00Z", next("0 0 1 * MON", "2026-01-02T00:00:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void leapDayResolvesWithinTheGuard() {
|
||||
assertEquals("2028-02-29T00:00Z", next("0 0 29 FEB *", "2026-01-01T00:00:00"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void malformedExpressionsFailAtParseTimeNotFireTime() {
|
||||
for (String bad : List.of("* * *", "99 * * * *", "0 0 * * NOPE", "*/0 * * * *", "5-1 * * * *")) {
|
||||
assertThrows(IllegalArgumentException.class, () -> Cron.parse(bad), bad);
|
||||
}
|
||||
}
|
||||
}
|
||||
+116
@@ -0,0 +1,116 @@
|
||||
package dev.relism.flash.ext.scheduler;
|
||||
|
||||
import dev.relism.flash.extension.FlashApp;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class SchedulerTest {
|
||||
|
||||
private static FlashApp app() {
|
||||
return FlashApp.create(FlashConfiguration.builder()
|
||||
.port(0).host("127.0.0.1").shutdownDrainTimeoutMs(250).build());
|
||||
}
|
||||
|
||||
@Test
|
||||
void intervalJobsRunRepeatedly() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
CountDownLatch ran = new CountDownLatch(3);
|
||||
try {
|
||||
app.start();
|
||||
app.ctx().require(Scheduler.class).every(Duration.ofMillis(30), ran::countDown);
|
||||
|
||||
assertTrue(ran.await(3, TimeUnit.SECONDS), "interval job should have run three times");
|
||||
} finally {
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void overlappingRunsAreSkipped() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
AtomicInteger started = new AtomicInteger();
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
try {
|
||||
app.start();
|
||||
app.ctx().require(Scheduler.class).every(Duration.ofMillis(20), () -> {
|
||||
started.incrementAndGet();
|
||||
try { release.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
|
||||
});
|
||||
|
||||
Thread.sleep(300);
|
||||
assertEquals(1, started.get(), "a job still running must not be started again");
|
||||
} finally {
|
||||
release.countDown();
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void aThrowingJobKeepsItsSchedule() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
CountDownLatch ran = new CountDownLatch(3);
|
||||
try {
|
||||
app.start();
|
||||
app.ctx().require(Scheduler.class).every(Duration.ofMillis(30), () -> {
|
||||
ran.countDown();
|
||||
throw new IllegalStateException("boom");
|
||||
});
|
||||
|
||||
assertTrue(ran.await(3, TimeUnit.SECONDS), "a throwing job must not cancel its own schedule");
|
||||
} finally {
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
|
||||
/** The reason SchedulerExtension needs FlashContext.onClose: a job must not outlive its app. */
|
||||
@Test
|
||||
void stoppingTheAppStopsTheScheduler() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
AtomicInteger runs = new AtomicInteger();
|
||||
app.start();
|
||||
app.ctx().require(Scheduler.class).every(Duration.ofMillis(20), runs::incrementAndGet);
|
||||
Thread.sleep(120);
|
||||
assertTrue(runs.get() > 0, "job should have run while the app was up");
|
||||
|
||||
app.stop().join();
|
||||
int afterStop = runs.get();
|
||||
Thread.sleep(200);
|
||||
|
||||
assertEquals(afterStop, runs.get(), "no job may run after the app stopped");
|
||||
}
|
||||
|
||||
@Test
|
||||
void badCronFailsWhenRegisteredNotWhenItWouldFire() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
try {
|
||||
app.start();
|
||||
Scheduler scheduler = app.ctx().require(Scheduler.class);
|
||||
|
||||
assertThrows(IllegalArgumentException.class, () -> scheduler.cron("not a cron", () -> {}));
|
||||
} finally {
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void jobNamesAreReportedForOps() throws Exception {
|
||||
FlashApp app = app().install(new SchedulerExtension());
|
||||
try {
|
||||
app.start();
|
||||
Scheduler scheduler = app.ctx().require(Scheduler.class);
|
||||
scheduler.every("refresh-reports", Duration.ofMinutes(5), () -> {});
|
||||
scheduler.cron("nightly-archive", "0 0 3 * * *", () -> {});
|
||||
|
||||
assertEquals(java.util.List.of("refresh-reports", "nightly-archive"), scheduler.jobNames());
|
||||
} finally {
|
||||
app.stop().join();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user