Compare commits
4
Commits
f68e661296
...
9de2800a83
| Author | SHA1 | Date | |
|---|---|---|---|
 Zakaria El Orche
|
9de2800a83 | ||
|
Zakaria El Orche
|
dbf057f493 | ||
 Zakaria El OrcheandClaude Opus 5
|
24bb10175d | ||
|
Zakaria El OrcheandClaude Opus 5
|
5c163b7f8d |
@@ -38,7 +38,8 @@ Format: `<type>(<scope>): <short description>`
|
||||
|
||||
Allowed scopes: `core`, `testing`, `ext-jackson`, `ext-openapi`, `ext-oidc`, `ext-routeviewer`,
|
||||
`ext-view-core`, `ext-view-jte`, `ext-view-thymeleaf`, `ext-limiter`, `ext-web-bundler`,
|
||||
`ext-mcp`, `ext-validation`, `ext-data-core`, `ext-data-jdbc`, `ext-data-hibernate`, `release`, `deps`, `ci`.
|
||||
`ext-mcp`, `ext-validation`, `ext-scheduler`, `ext-data-core`, `ext-data-jdbc`,
|
||||
`ext-data-hibernate`, `ext-cache-core`, `ext-cache-caffeine`, `release`, `deps`, `ci`.
|
||||
|
||||
Examples:
|
||||
```
|
||||
|
||||
@@ -16,6 +16,10 @@ a zero-allocation FSM router, bounded protocol state, and one shared request/res
|
||||
| `flash-extensions/flash-ext-view-core` | Minimal shared SSR runtime primitives |
|
||||
| `flash-extensions/flash-ext-view-jte` | Opinionated jte SSR extension |
|
||||
| `flash-extensions/flash-ext-view-thymeleaf` | Opinionated Thymeleaf SSR extension |
|
||||
| `flash-extensions/flash-ext-validation` | Request validation — jakarta constraints, compiled once per type |
|
||||
| `flash-extensions/flash-ext-scheduler` | Interval and cron background jobs on virtual threads |
|
||||
| `flash-extensions/flash-ext-cache-core` | Caching contract — `Cache`, `CacheManager`, `CacheSpec` |
|
||||
| `flash-extensions/flash-ext-cache-caffeine` | In-process cache backed by Caffeine |
|
||||
|
||||
## Requirements
|
||||
|
||||
@@ -146,6 +150,9 @@ See extension-specific READMEs for full details:
|
||||
- [`flash-ext-mcp`](flash-extensions/flash-ext-mcp/docs/README.md)
|
||||
- [`flash-ext-view-jte`](flash-extensions/flash-ext-view-jte/README.md)
|
||||
- [`flash-ext-view-thymeleaf`](flash-extensions/flash-ext-view-thymeleaf/README.md)
|
||||
- [`flash-ext-validation`](flash-extensions/flash-ext-validation/docs/README.md)
|
||||
- [`flash-ext-scheduler`](flash-extensions/flash-ext-scheduler/docs/README.md)
|
||||
- [`flash-ext-cache-caffeine`](flash-extensions/flash-ext-cache-caffeine/docs/README.md)
|
||||
- [`flash-testing`](flash-testing/docs/README.md)
|
||||
|
||||
## Error handlers
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
# flash-ext-cache-caffeine
|
||||
|
||||
In-process caching backed by [Caffeine](https://github.com/ben-manes/caffeine). Implements
|
||||
[`flash-ext-cache-core`](../../flash-ext-cache-core/docs/README.md).
|
||||
|
||||
## Dependency
|
||||
|
||||
```xml
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-caffeine</artifactId>
|
||||
<version>${flash.version}</version>
|
||||
</dependency>
|
||||
```
|
||||
|
||||
## Quick start
|
||||
|
||||
```java
|
||||
FlashApp.create(8080)
|
||||
.install(new CaffeineCacheExtension())
|
||||
.scan("dev.example.api");
|
||||
```
|
||||
|
||||
```java
|
||||
@GET("/api/users/{id}")
|
||||
public final class GetUser extends RequestHandler {
|
||||
|
||||
private Cache<String, User> users;
|
||||
private UserRepository repo;
|
||||
|
||||
@Override protected void onInit() {
|
||||
repo = require(UserRepository.class);
|
||||
users = require(CacheManager.class).build("users", spec -> spec
|
||||
.maxSize(10_000)
|
||||
.ttl(Duration.ofMinutes(10)));
|
||||
}
|
||||
|
||||
@Override public Object handle(Request req, Response res) {
|
||||
return users.get(req.param("id"), repo::findById);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The extension takes no configuration. Each cache declares its own size and TTL where it is built.
|
||||
|
||||
## Why Caffeine and not a `LinkedHashMap`
|
||||
|
||||
An LRU on top of `LinkedHashMap` is about sixty lines, and for a cache that is genuinely
|
||||
low-traffic it is the right answer — `ConcurrentHashMap::computeIfAbsent` is one line and has no
|
||||
hit rate to get wrong.
|
||||
|
||||
This module exists for the case where that stops being true. Caffeine's W-TinyLFU admission,
|
||||
striped frequency counters and amortised eviction are not a weekend's work to reproduce, and the
|
||||
failure mode of getting them wrong is a cache that is *slower* than no cache — lock contention on
|
||||
every lookup, or an eviction policy that throws away exactly the entries you were about to want.
|
||||
|
||||
## Lifecycle
|
||||
|
||||
Caches are released through `FlashContext.onClose`, so `app.stop()` drops every entry. That is
|
||||
invisible in production with one app per process and matters immediately under test, where many
|
||||
apps start and stop in one JVM.
|
||||
|
||||
## Statistics
|
||||
|
||||
```java
|
||||
CacheStats stats = users.stats();
|
||||
stats.hitRate(); // 0.0 until something is looked up
|
||||
```
|
||||
|
||||
Requires `recordStats()` on the spec. Without it you get `CacheStats.DISABLED`, which is honest
|
||||
about being unmeasured rather than reporting zeroes that look like a cold cache.
|
||||
|
||||
`manager.names()` lists every cache built so far, for an ops endpoint.
|
||||
|
||||
## What this is not
|
||||
|
||||
**HTTP caching.** If what you want is for the *client* to stop asking — `Cache-Control`, `ETag`,
|
||||
`304 Not Modified` — that is a middleware, not an object cache, and it saves the whole request
|
||||
rather than the lookup inside it. Reach for that first: it is cheaper, and the two solve different
|
||||
problems.
|
||||
|
||||
**A shared cache.** Every replica has its own. Two instances will hold different values for the
|
||||
same key, and an invalidation on one does not reach the other. When that becomes a problem the
|
||||
answer is a networked backend — see the note on `flash-ext-cache-redis` — and the semantics change
|
||||
with it: a cache that can fail is no longer transparent.
|
||||
@@ -0,0 +1,39 @@
|
||||
<?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-cache-caffeine</artifactId>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-core</artifactId>
|
||||
</dependency>
|
||||
<!--
|
||||
Caffeine rather than a hand-rolled LRU: W-TinyLFU admission, striped counters and
|
||||
amortised eviction are not a weekend's work to get right, and getting them wrong is a
|
||||
cache that is slower than no cache.
|
||||
-->
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</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>
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
package dev.relism.flash.ext.cache.caffeine;
|
||||
|
||||
import dev.relism.flash.ext.cache.Cache;
|
||||
import dev.relism.flash.ext.cache.CacheStats;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* {@link Cache} over a Caffeine cache. A thin adapter by design: every method delegates directly,
|
||||
* adding no wrapper object, no copy and no synchronisation of its own.
|
||||
*/
|
||||
final class CaffeineCache<K, V> implements Cache<K, V> {
|
||||
|
||||
private final com.github.benmanes.caffeine.cache.Cache<K, V> delegate;
|
||||
private final boolean statsRecorded;
|
||||
|
||||
CaffeineCache(com.github.benmanes.caffeine.cache.Cache<K, V> delegate, boolean statsRecorded) {
|
||||
this.delegate = delegate;
|
||||
this.statsRecorded = statsRecorded;
|
||||
}
|
||||
|
||||
@Override
|
||||
public V get(K key, Function<? super K, ? extends V> loader) {
|
||||
// Caffeine's own get(key, mappingFunction) already guarantees the loader runs once per key
|
||||
// across concurrent callers; wrapping it in anything of ours would only add a race.
|
||||
return delegate.get(key, loader);
|
||||
}
|
||||
|
||||
@Override public V getIfPresent(K key) { return delegate.getIfPresent(key); }
|
||||
@Override public void put(K key, V value) { delegate.put(key, value); }
|
||||
@Override public void invalidate(K key) { delegate.invalidate(key); }
|
||||
@Override public void invalidateAll() { delegate.invalidateAll(); }
|
||||
@Override public long estimatedSize() { return delegate.estimatedSize(); }
|
||||
|
||||
@Override
|
||||
public CacheStats stats() {
|
||||
if (!statsRecorded) return CacheStats.DISABLED;
|
||||
com.github.benmanes.caffeine.cache.stats.CacheStats snapshot = delegate.stats();
|
||||
return new CacheStats(snapshot.hitCount(), snapshot.missCount(), snapshot.evictionCount());
|
||||
}
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
package dev.relism.flash.ext.cache.caffeine;
|
||||
|
||||
import dev.relism.flash.ext.cache.CacheManager;
|
||||
import dev.relism.flash.extension.FlashContext;
|
||||
import dev.relism.flash.extension.FlashExtension;
|
||||
import dev.relism.flash.extension.FlashRegistrar;
|
||||
|
||||
/**
|
||||
* Installs an in-process {@link CacheManager} backed by Caffeine.
|
||||
*
|
||||
* <pre>{@code
|
||||
* FlashApp.create(8080)
|
||||
* .install(new CaffeineCacheExtension())
|
||||
* .scan("dev.example.api");
|
||||
* }</pre>
|
||||
*
|
||||
* <p>No configuration. Each cache declares its own size and TTL where it is built, because those
|
||||
* are properties of what is being cached, not of the process caching it.
|
||||
*
|
||||
* <p>Caches are dropped through {@link FlashContext#onClose}, so a stopped app does not keep its
|
||||
* values alive — which matters when many apps start and stop in one JVM, as they do under test.
|
||||
*/
|
||||
public final class CaffeineCacheExtension implements FlashExtension {
|
||||
|
||||
@Override
|
||||
public void configure(FlashRegistrar<?> app, FlashContext ctx) {
|
||||
ctx.supply(CacheManager.class, services -> {
|
||||
CaffeineCacheManager manager = new CaffeineCacheManager();
|
||||
services.onClose(manager::clear);
|
||||
return manager;
|
||||
});
|
||||
}
|
||||
}
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
package dev.relism.flash.ext.cache.caffeine;
|
||||
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import dev.relism.flash.ext.cache.Cache;
|
||||
import dev.relism.flash.ext.cache.CacheManager;
|
||||
import dev.relism.flash.ext.cache.CacheSpec;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/** In-process {@link CacheManager} backed by Caffeine. */
|
||||
final class CaffeineCacheManager implements CacheManager {
|
||||
|
||||
private final Map<String, Entry> caches = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public <K, V> Cache<K, V> build(String name, java.util.function.Consumer<CacheSpec> configure) {
|
||||
CacheSpec spec = CacheSpec.of();
|
||||
configure.accept(spec);
|
||||
|
||||
Entry entry = caches.computeIfAbsent(name, key -> new Entry(describe(spec), create(spec)));
|
||||
// Two handlers sharing a cache is the point; two handlers disagreeing about its size or
|
||||
// TTL is a bug that would otherwise resolve to whichever one ran first.
|
||||
String requested = describe(spec);
|
||||
if (!entry.signature.equals(requested))
|
||||
throw new IllegalStateException("Cache '" + name + "' already exists as " + entry.signature
|
||||
+ " but was requested as " + requested);
|
||||
return (Cache<K, V>) entry.cache;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public <K, V> Cache<K, V> cache(String name) {
|
||||
Entry entry = caches.get(name);
|
||||
return entry == null ? null : (Cache<K, V>) entry.cache;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Set<String> names() {
|
||||
return Set.copyOf(caches.keySet());
|
||||
}
|
||||
|
||||
/** Releases every entry so a stopped app does not keep its values alive. */
|
||||
void clear() {
|
||||
caches.values().forEach(entry -> entry.cache.invalidateAll());
|
||||
caches.clear();
|
||||
}
|
||||
|
||||
private static CaffeineCache<Object, Object> create(CacheSpec spec) {
|
||||
Caffeine<Object, Object> builder = Caffeine.newBuilder();
|
||||
if (spec.bounded()) builder.maximumSize(spec.maxSize());
|
||||
if (spec.ttl() != null) builder.expireAfterWrite(spec.ttl());
|
||||
if (spec.ttlAfterAccess() != null) builder.expireAfterAccess(spec.ttlAfterAccess());
|
||||
if (spec.statsRecorded()) builder.recordStats();
|
||||
return new CaffeineCache<>(builder.build(), spec.statsRecorded());
|
||||
}
|
||||
|
||||
private static String describe(CacheSpec spec) {
|
||||
return "maxSize=" + spec.maxSize() + " ttl=" + spec.ttl()
|
||||
+ " ttlAfterAccess=" + spec.ttlAfterAccess() + " stats=" + spec.statsRecorded();
|
||||
}
|
||||
|
||||
private record Entry(String signature, CaffeineCache<Object, Object> cache) {}
|
||||
}
|
||||
+160
@@ -0,0 +1,160 @@
|
||||
package dev.relism.flash.ext.cache.caffeine;
|
||||
|
||||
import dev.relism.flash.ext.cache.Cache;
|
||||
import dev.relism.flash.ext.cache.CacheManager;
|
||||
import dev.relism.flash.ext.cache.CacheStats;
|
||||
import dev.relism.flash.extension.FlashApp;
|
||||
import dev.relism.flash.extension.FlashConfiguration;
|
||||
import dev.relism.flash.testing.FlashTest;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.RegisterExtension;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class CaffeineCacheTest {
|
||||
|
||||
private static final AtomicInteger loads = new AtomicInteger();
|
||||
|
||||
@RegisterExtension
|
||||
static FlashTest app = FlashTest.of(configured -> {
|
||||
configured.install(new CaffeineCacheExtension());
|
||||
configured.ctx().onReady(() -> {
|
||||
Cache<String, String> users = configured.ctx().require(CacheManager.class)
|
||||
.build("users", spec -> spec.maxSize(100).ttl(Duration.ofMinutes(5)).recordStats());
|
||||
configured.get("/users/{id}", (req, res) ->
|
||||
users.get(req.param("id"), id -> "loaded:" + id + ":" + loads.incrementAndGet()));
|
||||
});
|
||||
});
|
||||
|
||||
private static CacheManager manager() {
|
||||
return app.app().ctx().require(CacheManager.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aRepeatedRequestIsServedFromCache() {
|
||||
String first = app.get("/users/alice").expectStatus(200).body();
|
||||
String second = app.get("/users/alice").expectStatus(200).body();
|
||||
|
||||
assertEquals(first, second);
|
||||
assertTrue(first.startsWith("loaded:alice:"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void distinctKeysLoadSeparately() {
|
||||
assertNotEquals(app.get("/users/bob").body(), app.get("/users/carol").body());
|
||||
}
|
||||
|
||||
@Test
|
||||
void statsCountHitsAndMisses() {
|
||||
Cache<String, String> cache = manager().build("stats-probe", spec -> spec.maxSize(10).recordStats());
|
||||
cache.get("k", key -> "v");
|
||||
cache.get("k", key -> "v");
|
||||
|
||||
CacheStats stats = cache.stats();
|
||||
assertEquals(1, stats.misses());
|
||||
assertEquals(1, stats.hits());
|
||||
assertEquals(0.5, stats.hitRate());
|
||||
}
|
||||
|
||||
@Test
|
||||
void statsAreDisabledUnlessAskedFor() {
|
||||
Cache<String, String> cache = manager().build("no-stats", spec -> spec.maxSize(10));
|
||||
cache.get("k", key -> "v");
|
||||
|
||||
assertEquals(CacheStats.DISABLED, cache.stats());
|
||||
}
|
||||
|
||||
@Test
|
||||
void theLoaderRunsOncePerKeyUnderConcurrency() throws Exception {
|
||||
Cache<String, String> cache = manager().build("single-flight", spec -> spec.maxSize(10));
|
||||
AtomicInteger invocations = new AtomicInteger();
|
||||
int threads = 16;
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
CountDownLatch done = new CountDownLatch(threads);
|
||||
|
||||
for (int i = 0; i < threads; i++) {
|
||||
Thread.ofVirtual().start(() -> {
|
||||
try {
|
||||
start.await();
|
||||
cache.get("hot", key -> {
|
||||
invocations.incrementAndGet();
|
||||
return "value";
|
||||
});
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
} finally {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
}
|
||||
start.countDown();
|
||||
assertTrue(done.await(5, java.util.concurrent.TimeUnit.SECONDS));
|
||||
|
||||
assertEquals(1, invocations.get(), "concurrent callers must share one load, not race");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aNullLoaderResultStoresNothing() {
|
||||
Cache<String, String> cache = manager().build("nulls", spec -> spec.maxSize(10));
|
||||
|
||||
assertNull(cache.get("missing", key -> null));
|
||||
assertNull(cache.getIfPresent("missing"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void invalidateDropsOneKeyAndInvalidateAllDropsEverything() {
|
||||
Cache<String, String> cache = manager().build("invalidation", spec -> spec.maxSize(10));
|
||||
cache.put("a", "1");
|
||||
cache.put("b", "2");
|
||||
|
||||
cache.invalidate("a");
|
||||
assertNull(cache.getIfPresent("a"));
|
||||
assertEquals("2", cache.getIfPresent("b"));
|
||||
|
||||
cache.invalidateAll();
|
||||
assertNull(cache.getIfPresent("b"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildIsIdempotentPerName() {
|
||||
Cache<String, String> first = manager().build("shared", spec -> spec.maxSize(10));
|
||||
Cache<String, String> second = manager().build("shared", spec -> spec.maxSize(10));
|
||||
|
||||
assertSame(first, second, "two handlers asking for one cache must get one cache");
|
||||
assertSame(first, manager().cache("shared"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void disagreeingOnASharedCacheIsARejectedMistakeNotASilentWinner() {
|
||||
manager().build("contested", spec -> spec.maxSize(10));
|
||||
|
||||
IllegalStateException conflict = assertThrows(IllegalStateException.class,
|
||||
() -> manager().build("contested", spec -> spec.maxSize(999)));
|
||||
assertTrue(conflict.getMessage().contains("contested"), conflict.getMessage());
|
||||
}
|
||||
|
||||
@Test
|
||||
void unknownNameReturnsNullRatherThanBuildingOne() {
|
||||
assertNull(manager().cache("never-built"));
|
||||
}
|
||||
|
||||
/** Why CaffeineCacheExtension registers onClose: values must not outlive the app holding them. */
|
||||
@Test
|
||||
void stoppingTheAppReleasesEveryCache() {
|
||||
FlashApp standalone = FlashApp.create(FlashConfiguration.builder()
|
||||
.port(0).host("127.0.0.1").shutdownDrainTimeoutMs(250).build())
|
||||
.install(new CaffeineCacheExtension());
|
||||
standalone.start();
|
||||
CacheManager manager = standalone.ctx().require(CacheManager.class);
|
||||
manager.build("scoped", spec -> spec.maxSize(10)).put("k", "v");
|
||||
assertEquals(1, manager.names().size());
|
||||
|
||||
standalone.stop().join();
|
||||
|
||||
assertTrue(manager.names().isEmpty(), "caches must be released when the app stops");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
# flash-ext-cache-core
|
||||
|
||||
The caching contract, shared across backends. Like `flash-ext-data-core`, this module talks to
|
||||
nothing: it defines the abstractions and a backend implements them.
|
||||
|
||||
## Components
|
||||
|
||||
- `Cache<K, V>` — a named cache. `get(key, loader)` is the method that matters.
|
||||
- `CacheManager` — creates and hands back named caches.
|
||||
- `CacheSpec` — size and expiry for one cache.
|
||||
- `CacheStats` — hit/miss/eviction counters.
|
||||
|
||||
Install a backend, not this module: [`flash-ext-cache-caffeine`](../../flash-ext-cache-caffeine/docs/README.md)
|
||||
for in-process caching.
|
||||
|
||||
## The one shape that matters
|
||||
|
||||
```java
|
||||
User user = users.get(id, repo::findById);
|
||||
```
|
||||
|
||||
Compute-if-absent is the only cache operation most code needs, and the only one that is hard to
|
||||
get right — the loader runs **once per key** across concurrent callers, and the rest wait rather
|
||||
than each computing their own. `getIfPresent`, `put`, `invalidate` and `invalidateAll` exist for
|
||||
what it cannot express.
|
||||
|
||||
A loader returning `null` stores nothing and returns `null`. Caching absence is a decision, not a
|
||||
default; wrap it in an `Optional` or a sentinel if you want it.
|
||||
|
||||
## Naming and sharing
|
||||
|
||||
`CacheManager.build(name, spec)` is idempotent per name: two handlers asking for `"users"` get one
|
||||
cache, not two, so nobody has to coordinate who creates it first.
|
||||
|
||||
If they disagree about the spec, that throws. The alternative is a cache whose size depends on
|
||||
which handler happened to initialise first, which is the kind of bug that only shows up under
|
||||
load.
|
||||
|
||||
## Specs
|
||||
|
||||
```java
|
||||
CacheSpec.of()
|
||||
.maxSize(10_000)
|
||||
.ttl(Duration.ofMinutes(10))
|
||||
.recordStats();
|
||||
```
|
||||
|
||||
Every field is optional, but a spec that sets neither `maxSize` nor `ttl` is an unbounded cache
|
||||
that never expires — a memory leak wearing a hat. Set at least one.
|
||||
|
||||
`recordStats()` is off by default: counting costs a pair of atomic increments on every lookup, and
|
||||
a cache nobody is measuring should not pay for numbers nobody reads. Without it, `stats()` returns
|
||||
`CacheStats.DISABLED` rather than silently zero.
|
||||
|
||||
## Where the spec lives
|
||||
|
||||
On the cache, at the point it is built — not in application config. Size and TTL are properties of
|
||||
*what is being cached*, not of the process doing the caching, and a TTL in a config file is a TTL
|
||||
nobody can relate back to the data it governs.
|
||||
|
||||
## Writing a backend
|
||||
|
||||
Implement `CacheManager` and `Cache`, provide the manager from a `FlashExtension`, and register
|
||||
cleanup with `FlashContext.onClose` so a stopped app does not keep its values alive.
|
||||
@@ -0,0 +1,25 @@
|
||||
<?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-cache-core</artifactId>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
package dev.relism.flash.ext.cache;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* A named cache. Obtained from a {@link CacheManager}, safe to hold in a handler field and share
|
||||
* across threads.
|
||||
*
|
||||
* <pre>{@code
|
||||
* User user = users.get(id, repo::findById);
|
||||
* }</pre>
|
||||
*
|
||||
* <p>{@link #get} is the only method most code needs. The rest exist for the cases it cannot
|
||||
* express: reading without populating, writing a value computed elsewhere, and invalidating.
|
||||
*
|
||||
* @param <K> key type — must have a stable {@code hashCode}/{@code equals}
|
||||
* @param <V> value type
|
||||
*/
|
||||
public interface Cache<K, V> {
|
||||
|
||||
/**
|
||||
* Returns the cached value, computing and storing it with {@code loader} if absent.
|
||||
*
|
||||
* <p>The loader runs at most once per key across concurrent callers; the others wait for it
|
||||
* rather than each computing their own. A loader returning {@code null} stores nothing and
|
||||
* {@code null} is returned.
|
||||
*/
|
||||
V get(K key, Function<? super K, ? extends V> loader);
|
||||
|
||||
/** The cached value, or {@code null} if absent. Never invokes a loader. */
|
||||
V getIfPresent(K key);
|
||||
|
||||
/** Stores {@code value}, replacing any existing entry. */
|
||||
void put(K key, V value);
|
||||
|
||||
/** Drops {@code key}. Does nothing if it was absent. */
|
||||
void invalidate(K key);
|
||||
|
||||
/** Drops every entry. */
|
||||
void invalidateAll();
|
||||
|
||||
/** Approximate entry count. Approximate because eviction is asynchronous in most backends. */
|
||||
long estimatedSize();
|
||||
|
||||
/**
|
||||
* Hit/miss counters since this cache was built, or {@link CacheStats#DISABLED} when the
|
||||
* backend was not asked to record them.
|
||||
*/
|
||||
CacheStats stats();
|
||||
}
|
||||
Vendored
+31
@@ -0,0 +1,31 @@
|
||||
package dev.relism.flash.ext.cache;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* Creates and hands back named caches. Resolve it with {@code require(CacheManager.class)}; a
|
||||
* backend extension such as {@code flash-ext-cache-caffeine} provides the implementation.
|
||||
*
|
||||
* <pre>{@code
|
||||
* @Override protected void onInit() {
|
||||
* users = require(CacheManager.class).build("users", spec -> spec
|
||||
* .maxSize(10_000)
|
||||
* .ttl(Duration.ofMinutes(10)));
|
||||
* }
|
||||
* }</pre>
|
||||
*
|
||||
* <p>{@link #build} is idempotent per name: calling it twice returns the same cache rather than
|
||||
* two, so several handlers can share one without coordinating who creates it. The spec of the
|
||||
* first call wins; a later call with a different spec is a configuration mistake and throws.
|
||||
*/
|
||||
public interface CacheManager {
|
||||
|
||||
/** Creates the named cache, or returns the existing one. */
|
||||
<K, V> Cache<K, V> build(String name, Consumer<CacheSpec> spec);
|
||||
|
||||
/** The named cache, or {@code null} if {@link #build} has not been called for it. */
|
||||
<K, V> Cache<K, V> cache(String name);
|
||||
|
||||
/** Every cache name built so far, for an ops endpoint. */
|
||||
java.util.Set<String> names();
|
||||
}
|
||||
Vendored
+64
@@ -0,0 +1,64 @@
|
||||
package dev.relism.flash.ext.cache;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* How one cache should behave. Every field is optional — a spec that sets nothing gives an
|
||||
* unbounded cache that never expires, which is a memory leak wearing a hat, so set at least one
|
||||
* of {@link #maxSize} or {@link #ttl}.
|
||||
*
|
||||
* <pre>{@code
|
||||
* CacheSpec.of().maxSize(10_000).ttl(Duration.ofMinutes(10))
|
||||
* }</pre>
|
||||
*
|
||||
* <p>Mutable builder rather than a record with {@code withX} copies: it is constructed once at
|
||||
* boot inside a lambda and never shared.
|
||||
*/
|
||||
public final class CacheSpec {
|
||||
|
||||
private long maxSize = -1;
|
||||
private Duration ttl;
|
||||
private Duration ttlAfterAccess;
|
||||
private boolean recordStats;
|
||||
|
||||
private CacheSpec() {}
|
||||
|
||||
public static CacheSpec of() {
|
||||
return new CacheSpec();
|
||||
}
|
||||
|
||||
/** Maximum entries before the backend starts evicting. Negative means unbounded. */
|
||||
public CacheSpec maxSize(long maxSize) {
|
||||
this.maxSize = maxSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Entries expire this long after they were written. */
|
||||
public CacheSpec ttl(Duration ttl) {
|
||||
this.ttl = ttl;
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Entries expire this long after they were last read or written. */
|
||||
public CacheSpec ttlAfterAccess(Duration ttlAfterAccess) {
|
||||
this.ttlAfterAccess = ttlAfterAccess;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Records hit/miss counters for {@link Cache#stats()}.
|
||||
*
|
||||
* <p>Off by default: counting costs a pair of atomic increments on every lookup, and a cache
|
||||
* nobody is measuring should not pay for numbers nobody reads.
|
||||
*/
|
||||
public CacheSpec recordStats() {
|
||||
this.recordStats = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
public long maxSize() { return maxSize; }
|
||||
public Duration ttl() { return ttl; }
|
||||
public Duration ttlAfterAccess() { return ttlAfterAccess; }
|
||||
public boolean statsRecorded() { return recordStats; }
|
||||
public boolean bounded() { return maxSize >= 0; }
|
||||
}
|
||||
Vendored
+20
@@ -0,0 +1,20 @@
|
||||
package dev.relism.flash.ext.cache;
|
||||
|
||||
/**
|
||||
* Hit/miss counters for one cache.
|
||||
*
|
||||
* @param hits lookups that found a value
|
||||
* @param misses lookups that had to load
|
||||
* @param evictions entries dropped to respect {@link CacheSpec#maxSize()}
|
||||
*/
|
||||
public record CacheStats(long hits, long misses, long evictions) {
|
||||
|
||||
/** Returned when {@link CacheSpec#recordStats()} was not set — all zero, and says so. */
|
||||
public static final CacheStats DISABLED = new CacheStats(0, 0, 0);
|
||||
|
||||
/** Hits divided by lookups, or 0 when nothing has been looked up yet. */
|
||||
public double hitRate() {
|
||||
long total = hits + misses;
|
||||
return total == 0 ? 0 : (double) hits / total;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
# flash-ext-cache-redis — planned
|
||||
|
||||
Not implemented. This directory holds the design so the decision is written down rather than
|
||||
rediscovered; there is deliberately **no module, no pom and no source**, because an empty module
|
||||
that builds an empty jar is dead weight in the reactor and in everyone's dependency tree.
|
||||
|
||||
Add it when there is a second replica that actually needs shared state.
|
||||
|
||||
## What it would implement
|
||||
|
||||
`CacheManager` and `Cache` from [`flash-ext-cache-core`](../../flash-ext-cache-core/docs/README.md),
|
||||
so switching backend is an install-line change:
|
||||
|
||||
```java
|
||||
.install(new RedisCacheExtension(RedisConfig.of("redis://localhost:6379")))
|
||||
```
|
||||
|
||||
## The part that is not a drop-in
|
||||
|
||||
`flash-ext-cache-caffeine` cannot fail. A networked cache can, and that changes the contract in
|
||||
ways an adapter cannot hide:
|
||||
|
||||
- **`get(key, loader)` can fail before reaching the loader.** The honest default is to fall
|
||||
through to the loader and serve the value uncached, so Redis being down degrades throughput
|
||||
rather than taking the application with it. That has to be a decision, not an accident.
|
||||
- **Values must be serialized.** Caffeine stores references. A `byte[]` codec belongs in the spec,
|
||||
and the natural default is whatever `flash-ext-jackson` is already configured with.
|
||||
- **`invalidateAll()` is not free.** Against a shared keyspace it is either a scan or a key
|
||||
prefix per cache name. The prefix is the right answer, and it means cache names become part of
|
||||
the wire contract.
|
||||
- **Stats are per-client, not per-cache.** Hit rate stays meaningful; eviction count does not,
|
||||
because Redis evicts on its own policy.
|
||||
|
||||
## Why it is not built yet
|
||||
|
||||
Nothing in the codebase has two replicas sharing cache state. Building it now would mean choosing
|
||||
a client library, a serialization format and a failure policy with no real usage to check them
|
||||
against — and the failure policy in particular is the kind of decision that is wrong until a
|
||||
production incident tells you otherwise.
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -25,9 +25,12 @@
|
||||
<module>flash-ext-web-bundler</module>
|
||||
<module>flash-ext-mcp</module>
|
||||
<module>flash-ext-validation</module>
|
||||
<module>flash-ext-scheduler</module>
|
||||
<module>flash-ext-data-core</module>
|
||||
<module>flash-ext-data-jdbc</module>
|
||||
<module>flash-ext-data-hibernate</module>
|
||||
<module>flash-ext-cache-core</module>
|
||||
<module>flash-ext-cache-caffeine</module>
|
||||
</modules>
|
||||
|
||||
<dependencyManagement>
|
||||
@@ -37,6 +40,26 @@
|
||||
<artifactId>flash-ext-validation</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-scheduler</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-caffeine</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</artifactId>
|
||||
<version>${caffeine.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-testing</artifactId>
|
||||
|
||||
@@ -39,6 +39,7 @@
|
||||
<jmh.version>1.37</jmh.version>
|
||||
<junit.version>5.11.0</junit.version>
|
||||
<jakarta.validation.version>3.1.1</jakarta.validation.version>
|
||||
<caffeine.version>3.1.8</caffeine.version>
|
||||
<build.helper.plugin.version>3.6.0</build.helper.plugin.version>
|
||||
</properties>
|
||||
|
||||
@@ -73,6 +74,21 @@
|
||||
<artifactId>flash-ext-validation</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-scheduler</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>dev.relism</groupId>
|
||||
<artifactId>flash-ext-cache-caffeine</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>jakarta.validation</groupId>
|
||||
<artifactId>jakarta.validation-api</artifactId>
|
||||
|
||||
Reference in New Issue
Block a user