From a6943b041fe70d0bc3317d6b8f78c283831a1704 Mon Sep 17 00:00:00 2001 From: Ned Wolpert Date: Sat, 18 Jul 2026 19:52:22 -0700 Subject: [PATCH] Add velocity-backend-redis: the sliding hot-path reference backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The second real velocity-spi backend (v1 Redis/Lettuce sliding reference, P10/§8) — true sliding windows via sorted sets + atomic Lua, validated against real Redis with Testcontainers. Reuses the shared conformance TCK (now window-parameterized) with SLIDING windows + SlidingScenarios. - RedisVelocityBackend implements CountStore/SumStore/DistinctStore + SlidingSupport. Exact, ATOMIC read-your-write: each apply is a single atomic Lua script (INCR a per-key sequence for a unique member, ZADD scored by now, evict aged members, return the post-write windowed value), so concurrent applies never lose an increment (NFR-6, acceptance #2 — the 16-thread shared concurrency scenario passes on real Redis). - Backend owns the clock (ADR 0003/FR-3): a sliding window of duration D covers (now-D, now] with the LEADING EDGE EXCLUSIVE, now from the injected Clock (not Redis TIME) — matches the in-memory reference, so SlidingScenarios (aging / exclusive edge) pass. Windowed value = members with score in (now-D, now]; eviction is cost-cleanup. - Storage: one ZSET per (namespace, subject, aggregation[type+dimension], window) — no feature name in the key (QueryTuple has none), like the in-memory + JDBI backends. COUNT/SUM append a unique member; DISTINCT uses the opaque Base64 member token itself (de-dupes). SUM cents summed client-side as BigDecimal (9e18-scale preserved, negatives honored, scale 0). Unsupported window -> Failure(UNSUPPORTED_WINDOW) on apply and query, never a silent 0. - SeedSupport is intentionally NOT implemented in v1 (seedSupported=false): ADR 0008's per-bucket seed is a tumbling concept and the mix-in is optional (sliding-seed is a tracked follow-up). Capabilities declare SLIDING-only, EXACT, distinctHllSliding=false. - No Dagger/Dropwizard/Jackson (depends only on velocity-spi + Lettuce + slf4j). 27 tests against real Redis (Testcontainers redis:7-alpine), 0 skipped; coverage 97.5% line / 92.3% branch (gate 80/70). Hand-implemented (implementer subagents stalled twice on this task). Validated by the change-validator agent: VERDICT APPROVE (sliding semantics correct, atomicity proven, shared TCK genuinely running). Follow-up (minor, non-blocking): the COUNT/DISTINCT apply return uses ZCARD-after-evict; switching to a windowRange zcount inside the Lua would tighten the running value under cross-thread clock skew (the final authoritative query is already exact). Co-Authored-By: Claude Opus 4.8 (1M context) --- settings.gradle.kts | 2 +- velocity-backend-redis/build.gradle.kts | 46 ++ .../backend/redis/RedisVelocityBackend.java | 417 ++++++++++++++++++ .../velocity/backend/redis/package-info.java | 11 + .../redis/AbstractRedisBackendTest.java | 25 ++ .../redis/RedisConformanceTckTest.java | 202 +++++++++ .../velocity/backend/redis/RedisSupport.java | 55 +++ 7 files changed, 757 insertions(+), 1 deletion(-) create mode 100644 velocity-backend-redis/build.gradle.kts create mode 100644 velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/RedisVelocityBackend.java create mode 100644 velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/package-info.java create mode 100644 velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/AbstractRedisBackendTest.java create mode 100644 velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisConformanceTckTest.java create mode 100644 velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisSupport.java diff --git a/settings.gradle.kts b/settings.gradle.kts index 33a0274..35cbefd 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -59,7 +59,7 @@ include("velocity-testkit") // in-memory velocity-spi impl, conformance TCK sce // Phase 2 — v1 reference backends (tumbling + sliding), the service tier, and the generated client. include("velocity-backend-jdbi") // Postgres/JDBI backend — v1 tumbling reference -// include("velocity-backend-redis") // Redis/Lettuce backend — v1 sliding / hot-path reference +include("velocity-backend-redis") // Redis/Lettuce backend — v1 sliding / hot-path reference // include("velocity-dropwizard") // Dropwizard bundle, Jersey resources, Dagger wiring, API-key auth // include("velocity-client-java") // Java client generated from OpenAPI via openapi-generator // include("examples:dropwizard-demo")// runnable reference service (not published) diff --git a/velocity-backend-redis/build.gradle.kts b/velocity-backend-redis/build.gradle.kts new file mode 100644 index 0000000..2f63047 --- /dev/null +++ b/velocity-backend-redis/build.gradle.kts @@ -0,0 +1,46 @@ +plugins { + id("velocity.library-conventions") + id("velocity.test-conventions") + id("velocity.publish-conventions") +} + +description = + "velocity-backend-redis: the v1 Redis/Lettuce sliding hot-path reference backend — an exact," + + " atomic, read-your-write VelocityBackend (COUNT/SUM/DISTINCT over true sliding windows via" + + " sorted sets + Lua) that passes the velocity-testkit conformance TCK against real Redis." + +// Lettuce ships as an automatic module and pulls Reactor/Netty (also automatic modules); the +// baseline -Xlint:all makes their `requires` a warning, which -Werror (library-conventions) turns +// fatal. Silence just the automatic-module lints on both the main and test compiles (mirrors the +// jdbi module). +tasks.named("compileJava") { + options.compilerArgs.addAll( + listOf("-Xlint:-requires-automatic", "-Xlint:-requires-transitive-automatic"), + ) +} +tasks.named("compileTestJava") { + options.compilerArgs.addAll( + listOf("-Xlint:-requires-automatic", "-Xlint:-requires-transitive-automatic"), + ) +} + +dependencies { + // The frozen contract this backend implements is on the api surface. + api(project(":velocity-spi")) + api(libs.jspecify) + + implementation(libs.lettuce.core) + implementation(libs.slf4j.api) + + // Reactor's class files reference com.google.errorprone.annotations.*; without errorprone + // annotations on the compile classpath javac emits an annotation-not-found warning that -Werror + // turns fatal. compileOnly so it is not broadcast as a runtime dependency. + compileOnly(libs.build.errorprone.annotations) + testCompileOnly(libs.build.errorprone.annotations) + + // The testkit drives the conformance TCK (*Scenarios) + MutableClock against this backend. + testImplementation(project(":velocity-testkit")) + testImplementation(libs.testcontainers.core) + testImplementation(libs.testcontainers.junit.jupiter) + testRuntimeOnly(libs.logback.classic) +} diff --git a/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/RedisVelocityBackend.java b/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/RedisVelocityBackend.java new file mode 100644 index 0000000..137f1e1 --- /dev/null +++ b/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/RedisVelocityBackend.java @@ -0,0 +1,417 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.redis; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.SlidingSupport; +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.ApplyContext; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities.WindowSpec; +import com.codeheadsystems.velocity.spi.model.Exactness; +import com.codeheadsystems.velocity.spi.model.FailureCode; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.FeatureValue; +import com.codeheadsystems.velocity.spi.model.Intent; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Intent.DistinctIntent; +import com.codeheadsystems.velocity.spi.model.Intent.SumIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import com.codeheadsystems.velocity.spi.model.QueryContext; +import com.codeheadsystems.velocity.spi.model.QueryTuple; +import com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import com.codeheadsystems.velocity.spi.model.WindowType; +import io.lettuce.core.Range; +import io.lettuce.core.ScanArgs; +import io.lettuce.core.ScanIterator; +import io.lettuce.core.ScriptOutputType; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; +import java.math.BigDecimal; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import org.jspecify.annotations.Nullable; + +/** + * The Redis/Lettuce v1 SLIDING hot-path reference {@link + * com.codeheadsystems.velocity.spi.VelocityBackend}. + * + *

Implements the aggregation mix-ins {@link CountStore}, {@link SumStore} and {@link + * DistinctStore} over {@link SlidingSupport true sliding} windows. Values are exact and produced + * under {@link ReadYourWriteLevel#ATOMIC} read-your-write: each apply is a single atomic Lua script + * (Redis is single-threaded) that appends the event, evicts aged members, and returns the + * post-write windowed value, so concurrent applies never lose an increment (NFR-6, acceptance #2). + * + *

Backend owns the clock (ADR 0003, FR-3). A sliding window of duration {@code + * D} covers {@code (now - D, now]} — leading edge exclusive — where {@code now} is read from the + * injected {@link Clock}, never Redis {@code TIME}, so a controllable test clock drives window + * edges. This matches {@code velocity-testkit}'s {@code InMemoryVelocityBackend}. + * + *

Storage. One sorted set per {@code (namespace, subject, aggregation, window)} + * keyed {@code v:{ns}:{subjectType}:{subjectValue}:{aggType}:{dimension}:S:{durationMs}} (dimension + * empty for COUNT/SUM). COUNT/SUM append a unique member (a per-key {@code INCR} sequence) scored + * by {@code now}; DISTINCT uses the opaque member token itself as the ZSET member (score = last + * seen), so a re-applied member de-dupes. The windowed value is the members with score in {@code + * (now-D, now]}; aged members are evicted as cost-cleanup (correctness comes from the score range, + * not eviction timing — FR-22a). + * + *

{@link com.codeheadsystems.velocity.spi.SeedSupport} is intentionally NOT implemented in v1: + * ADR 0008's per-bucket seed model is a tumbling concept, and the mix-in is optional. + */ +public final class RedisVelocityBackend + implements CountStore, SumStore, DistinctStore, SlidingSupport { + + /** COUNT apply: unique-member append, evict {@code <= now-D}, return the window's cardinality. */ + private static final String COUNT_APPLY = + """ + local seq = redis.call('INCR', KEYS[2]) + redis.call('ZADD', KEYS[1], ARGV[1], seq) + redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[2]) + return redis.call('ZCARD', KEYS[1])"""; + + /** SUM apply: append {@code seq:cents} scored by now, evict, return the in-window members. */ + private static final String SUM_APPLY = + """ + local seq = redis.call('INCR', KEYS[2]) + redis.call('ZADD', KEYS[1], ARGV[1], seq..':'..ARGV[3]) + redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[2]) + return redis.call('ZRANGEBYSCORE', KEYS[1], '('..ARGV[2], ARGV[1])"""; + + /** DISTINCT apply: upsert the member at score now (de-dupe), evict, return the cardinality. */ + private static final String DISTINCT_APPLY = + """ + redis.call('ZADD', KEYS[1], ARGV[1], ARGV[2]) + redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', ARGV[3]) + return redis.call('ZCARD', KEYS[1])"""; + + /** The sliding windows this backend supports, all EXACT. First entry is the smallest. */ + private static final List SUPPORTED_WINDOWS = + List.of( + new Window(Duration.ofSeconds(5), WindowType.SLIDING), + new Window(Duration.ofSeconds(30), WindowType.SLIDING), + new Window(Duration.ofMinutes(1), WindowType.SLIDING), + new Window(Duration.ofMinutes(30), WindowType.SLIDING)); + + private final RedisCommands redis; + private final Clock clock; + private final BackendCapabilities capabilities; + + /** + * Creates a backend over a Lettuce connection with the system UTC clock. + * + * @param connection a Lettuce connection to Redis + */ + public RedisVelocityBackend(final StatefulRedisConnection connection) { + this(connection, Clock.systemUTC()); + } + + /** + * Creates a backend over a Lettuce connection with an injectable clock — the authority for every + * window edge (FR-3). Inject a controllable clock in tests. + * + * @param connection a Lettuce connection to Redis + * @param clock the backend clock + */ + public RedisVelocityBackend( + final StatefulRedisConnection connection, final Clock clock) { + this.redis = Objects.requireNonNull(connection, "connection").sync(); + this.clock = Objects.requireNonNull(clock, "clock"); + this.capabilities = buildCapabilities(); + } + + private static BackendCapabilities buildCapabilities() { + final List specs = new ArrayList<>(); + for (final Window window : SUPPORTED_WINDOWS) { + specs.add(new WindowSpec(window, Exactness.EXACT, window.duration())); + } + return new BackendCapabilities( + Set.of(AggregationType.COUNT, AggregationType.SUM, AggregationType.DISTINCT), + specs, + /* distinctHllSliding= */ false, + // Exact-only distinct (ADR 0005, sliding): the ZSET holds every member, so the clamp and + // HLL + // threshold are effectively unbounded — this backend never sketches. + /* distinctExactCardinalityClamp= */ Long.MAX_VALUE, + /* distinctHllThresholdDefault= */ Long.MAX_VALUE, + /* maxRetention= */ Duration.ofMinutes(30), + ReadYourWriteLevel.ATOMIC, + /* idempotencySupported= */ false, + /* seedSupported= */ false, + /* maxAtomicFanOut= */ Integer.MAX_VALUE); + } + + /** + * The clock this backend reads time from; the authority for every window edge (FR-3). + * + * @return the injected clock + */ + public Clock clock() { + return clock; + } + + @Override + public BackendCapabilities capabilities() { + return capabilities; + } + + // ---- COUNT --------------------------------------------------------------------------------- + + @Override + public ApplyResult applyCount(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (zkey, nowMs, lowerMs, intent) -> { + final Long count = + redis.eval( + COUNT_APPLY, + ScriptOutputType.INTEGER, + new String[] {zkey, zkey + ":seq"}, + Long.toString(nowMs), + Long.toString(lowerMs)); + return BigDecimal.valueOf(count); + }); + } + + @Override + public List queryCount(final QueryContext ctx, final List tuples) { + return query(ctx, tuples, this::cardinality); + } + + // ---- SUM ----------------------------------------------------------------------------------- + + @Override + public ApplyResult applySum(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (zkey, nowMs, lowerMs, intent) -> { + final List members = + redis.eval( + SUM_APPLY, + ScriptOutputType.MULTI, + new String[] {zkey, zkey + ":seq"}, + Long.toString(nowMs), + Long.toString(lowerMs), + ((SumIntent) intent).valueCents().toPlainString()); + return sum(members); + }); + } + + @Override + public List querySum(final QueryContext ctx, final List tuples) { + return query(ctx, tuples, (zkey, nowMs, lowerMs) -> sum(rangeMembers(zkey, nowMs, lowerMs))); + } + + // ---- DISTINCT ------------------------------------------------------------------------------ + + @Override + public ApplyResult applyDistinct(final ApplyContext ctx, final List intents) { + return apply( + ctx, + intents, + (zkey, nowMs, lowerMs, intent) -> { + final String member = + Base64.getEncoder().encodeToString(((DistinctIntent) intent).member().token()); + final Long cardinality = + redis.eval( + DISTINCT_APPLY, + ScriptOutputType.INTEGER, + new String[] {zkey}, + Long.toString(nowMs), + member, + Long.toString(lowerMs)); + return BigDecimal.valueOf(cardinality); + }); + } + + @Override + public List queryDistinct(final QueryContext ctx, final List tuples) { + return query(ctx, tuples, this::cardinality); + } + + // ---- PURGE --------------------------------------------------------------------------------- + + @Override + public void purge(final Namespace namespace, final @Nullable Subject subject) { + Objects.requireNonNull(namespace, "namespace"); + final String pattern = + subject == null + ? "v:" + namespace.value() + ":*" + : "v:" + namespace.value() + ":" + subject.type() + ":" + subject.value() + ":*"; + final List keys = new ArrayList<>(); + final ScanIterator it = ScanIterator.scan(redis, ScanArgs.Builder.matches(pattern)); + while (it.hasNext()) { + keys.add(it.next()); + } + if (!keys.isEmpty()) { + redis.del(keys.toArray(new String[0])); + } + } + + // ---- shared apply / query ------------------------------------------------------------------ + + private ApplyResult apply( + final ApplyContext ctx, final List intents, final Writer writer) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(intents, "intents"); + final Namespace namespace = ctx.namespace(); + final long nowMs = clock.instant().toEpochMilli(); + final List outcomes = new ArrayList<>(); + for (final Intent intent : intents) { + final Feature feature = intent.feature(); + for (final Window window : feature.windows()) { + if (!capabilities.supportsWindow(window)) { + outcomes.add( + new PerFeature( + feature, + ApplyStatus.FAILED, + FeatureResult.failure(FailureCode.UNSUPPORTED_WINDOW, unsupported(window)))); + continue; + } + final long lowerMs = nowMs - window.duration().toMillis(); + final String zkey = key(namespace, intent.subject(), feature.aggregation(), window); + final BigDecimal value = writer.write(zkey, nowMs, lowerMs, intent); + outcomes.add( + new PerFeature( + feature, + ApplyStatus.APPLIED, + FeatureResult.success(featureValue(feature, window, value, nowMs, lowerMs)))); + } + } + return new ApplyResult(outcomes); + } + + private List query( + final QueryContext ctx, final List tuples, final Reader reader) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(tuples, "tuples"); + final Namespace namespace = ctx.namespace(); + final long nowMs = clock.instant().toEpochMilli(); + final List results = new ArrayList<>(); + for (final QueryTuple tuple : tuples) { + final Window window = tuple.window(); + if (!capabilities.supportsWindow(window)) { + results.add(FeatureResult.failure(FailureCode.UNSUPPORTED_WINDOW, unsupported(window))); + continue; + } + final long lowerMs = nowMs - window.duration().toMillis(); + final String zkey = key(namespace, tuple.subject(), tuple.aggregation(), window); + final BigDecimal value = reader.read(zkey, nowMs, lowerMs); + results.add( + FeatureResult.success( + featureValue(syntheticFeature(tuple), window, value, nowMs, lowerMs))); + } + return results; + } + + // ---- redis helpers ------------------------------------------------------------------------- + + /** Cardinality of members with score in {@code (lowerMs, nowMs]}. */ + private BigDecimal cardinality(final String zkey, final long nowMs, final long lowerMs) { + final Long count = redis.zcount(zkey, windowRange(nowMs, lowerMs)); + return BigDecimal.valueOf(count == null ? 0L : count); + } + + /** The {@code seq:cents} members with score in {@code (lowerMs, nowMs]}. */ + private List rangeMembers(final String zkey, final long nowMs, final long lowerMs) { + return redis.zrangebyscore(zkey, windowRange(nowMs, lowerMs)); + } + + private static Range windowRange(final long nowMs, final long lowerMs) { + // (lowerMs, nowMs] — leading edge exclusive (a sliding window is (now-D, now]). + return Range.from(Range.Boundary.excluding(lowerMs), Range.Boundary.including(nowMs)); + } + + private static BigDecimal sum(final List members) { + BigDecimal total = BigDecimal.ZERO; + for (final String member : members) { + final int idx = member.indexOf(':'); + total = total.add(new BigDecimal(member.substring(idx + 1))); + } + return total; + } + + // ---- key / value helpers ------------------------------------------------------------------- + + private static String key( + final Namespace namespace, + final Subject subject, + final Aggregation aggregation, + final Window window) { + final String dimension = aggregation.dimension() == null ? "" : aggregation.dimension(); + return "v:" + + namespace.value() + + ':' + + subject.type() + + ':' + + subject.value() + + ':' + + aggregation.type().name() + + ':' + + dimension + + ":S:" + + window.duration().toMillis(); + } + + private static FeatureValue featureValue( + final Feature feature, + final Window window, + final BigDecimal value, + final long nowMs, + final long lowerMs) { + return new FeatureValue( + feature, + window, + value, + Exactness.EXACT, + ReadYourWriteLevel.ATOMIC, + /* definitionVersionHash= */ null, + new WindowBounds(Instant.ofEpochMilli(lowerMs), Instant.ofEpochMilli(nowMs)), + Instant.ofEpochMilli(nowMs)); + } + + private static Feature syntheticFeature(final QueryTuple tuple) { + final Aggregation aggregation = tuple.aggregation(); + final String name = + "query:" + + aggregation.type() + + (aggregation.dimension() == null ? "" : ":" + aggregation.dimension()); + return new Feature(name, aggregation, List.of(tuple.window())); + } + + private static String unsupported(final Window window) { + return "window " + + window.type() + + " " + + window.duration() + + " not supported by velocity-backend-redis"; + } + + /** Writes one intent into its sorted set and returns the post-write windowed value. */ + @FunctionalInterface + private interface Writer { + BigDecimal write(String zkey, long nowMs, long lowerMs, Intent intent); + } + + /** Reads the current windowed value of a sorted set (0 when empty). */ + @FunctionalInterface + private interface Reader { + BigDecimal read(String zkey, long nowMs, long lowerMs); + } +} diff --git a/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/package-info.java b/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/package-info.java new file mode 100644 index 0000000..476822e --- /dev/null +++ b/velocity-backend-redis/src/main/java/com/codeheadsystems/velocity/backend/redis/package-info.java @@ -0,0 +1,11 @@ +// SPDX-License-Identifier: BSD-3-Clause +/** + * The Redis/Lettuce v1 SLIDING hot-path reference backend: an exact, atomic, read-your-write {@link + * com.codeheadsystems.velocity.spi.VelocityBackend} over true sliding windows, backed by sorted + * sets and atomic Lua apply scripts. See {@link + * com.codeheadsystems.velocity.backend.redis.RedisVelocityBackend}. + */ +@NullMarked +package com.codeheadsystems.velocity.backend.redis; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/AbstractRedisBackendTest.java b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/AbstractRedisBackendTest.java new file mode 100644 index 0000000..35b43bd --- /dev/null +++ b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/AbstractRedisBackendTest.java @@ -0,0 +1,25 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.redis; + +import com.codeheadsystems.velocity.testkit.MutableClock; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; + +/** + * Base for the real-Redis integration tests: a fresh {@link RedisVelocityBackend} over the shared + * Testcontainers Redis, driven by a controllable {@link MutableClock}, with the store flushed + * before each test. Disabled where Docker is absent via {@code VELOCITY_SKIP_TESTCONTAINERS=1}. + */ +@DisabledIfEnvironmentVariable(named = "VELOCITY_SKIP_TESTCONTAINERS", matches = "1") +abstract class AbstractRedisBackendTest { + + protected MutableClock clock; + protected RedisVelocityBackend backend; + + @BeforeEach + void setUpBackend() { + RedisSupport.flush(); + clock = MutableClock.atNow(); + backend = new RedisVelocityBackend(RedisSupport.connection(), clock); + } +} diff --git a/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisConformanceTckTest.java b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisConformanceTckTest.java new file mode 100644 index 0000000..3046f4b --- /dev/null +++ b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisConformanceTckTest.java @@ -0,0 +1,202 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.redis; + +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import com.codeheadsystems.velocity.testkit.tck.CapabilityConformanceScenarios; +import com.codeheadsystems.velocity.testkit.tck.CountStoreScenarios; +import com.codeheadsystems.velocity.testkit.tck.DistinctStoreScenarios; +import com.codeheadsystems.velocity.testkit.tck.PurgeScenarios; +import com.codeheadsystems.velocity.testkit.tck.SlidingScenarios; +import com.codeheadsystems.velocity.testkit.tck.SumStoreScenarios; +import java.time.Duration; +import org.junit.jupiter.api.Test; + +/** + * Drives the shared {@code velocity-testkit} conformance TCK against real Redis for every scenario + * that applies to this sliding backend (ADR 0004) — the same window-parameterized {@code + * *Scenarios} the reference in-memory backend and the JDBI backend run, here with SLIDING windows, + * plus {@link SlidingScenarios} (aging / exclusive leading edge). + * + *

{@code TumblingScenarios} is intentionally NOT run — this backend declares no tumbling + * windows. {@code SeedSupportScenarios} is not run either — {@link RedisVelocityBackend} does not + * implement the optional {@code SeedSupport} mix-in in v1 (ADR 0008's per-bucket seed is a tumbling + * concept). + */ +final class RedisConformanceTckTest extends AbstractRedisBackendTest { + + private static final Window SLIDING_1M = new Window(Duration.ofMinutes(1), WindowType.SLIDING); + private static final Window SLIDING_30S = new Window(Duration.ofSeconds(30), WindowType.SLIDING); + + private CountStoreScenarios count() { + return new CountStoreScenarios(backend, clock, SLIDING_1M, SLIDING_30S); + } + + private SumStoreScenarios sum() { + return new SumStoreScenarios(backend, clock, SLIDING_1M); + } + + private DistinctStoreScenarios distinct() { + return new DistinctStoreScenarios(backend, clock, SLIDING_1M); + } + + private PurgeScenarios purge() { + return new PurgeScenarios(backend, SLIDING_1M); + } + + private SlidingScenarios sliding() { + return new SlidingScenarios(backend, clock); + } + + private CapabilityConformanceScenarios capability() { + return new CapabilityConformanceScenarios(backend); + } + + // ---- CountStoreScenarios (sliding) ---- + + @Test + void applyThenQueryReturnsExactCount() { + count().applyThenQueryReturnsExactCount(); + } + + @Test + void applyResultReflectsWriteReadYourWrite() { + count().applyResultReflectsWriteReadYourWrite(); + } + + @Test + void applyEmitsOneResultPerWindow() { + count().applyEmitsOneResultPerWindow(); + } + + @Test + void countValuesIsolatedByNamespace() { + count().valuesIsolatedByNamespace(); + } + + @Test + void countValuesIsolatedBySubject() { + count().valuesIsolatedBySubject(); + } + + /** The shared concurrency scenario (acceptance #2) against real Redis. */ + @Test + void concurrentApplyIsAtomic() throws Exception { + count().concurrentApplyIsAtomic(); + } + + // ---- SumStoreScenarios (sliding) ---- + + @Test + void applyThenQueryReturnsExactSumCents() { + sum().applyThenQueryReturnsExactSumCents(); + } + + @Test + void applyResultReflectsRunningSum() { + sum().applyResultReflectsRunningSum(); + } + + @Test + void bigDecimalCentsPreservedWithoutOverflow() { + sum().bigDecimalCentsPreservedWithoutOverflow(); + } + + @Test + void negativeValuesForRefunds() { + sum().negativeValuesForRefunds(); + } + + @Test + void sumValuesIsolatedByNamespace() { + sum().valuesIsolatedByNamespace(); + } + + // ---- DistinctStoreScenarios (sliding) ---- + + @Test + void applyThenQueryReturnsCardinality() { + distinct().applyThenQueryReturnsCardinality(); + } + + @Test + void deDupesRepeatedMembers() { + distinct().deDupesRepeatedMembers(); + } + + @Test + void applyResultReflectsCardinality() { + distinct().applyResultReflectsCardinality(); + } + + @Test + void distinctValuesIsolatedBySubject() { + distinct().valuesIsolatedBySubject(); + } + + // ---- PurgeScenarios (sliding) ---- + + @Test + void purgeSubjectClearsThatSubjectOnly() { + purge().purgeSubjectClearsThatSubjectOnly(); + } + + @Test + void purgeNamespaceClearsEntireNamespace() { + purge().purgeNamespaceClearsEntireNamespace(); + } + + // ---- SlidingScenarios (aging / exclusive leading edge) ---- + + @Test + void eventsAgeOutAsClockAdvances() { + sliding().eventsAgeOutAsClockAdvances(); + } + + @Test + void slidingLeadingEdgeIsExclusive() { + sliding().slidingLeadingEdgeIsExclusive(); + } + + // ---- CapabilityConformanceScenarios (ADR 0004) ---- + + @Test + void declaredAggregationsMatchImplementedMixins() { + capability().declaredAggregationsMatchImplementedMixins(); + } + + @Test + void declaredWindowMarkersMatchWindowSpecs() { + capability().declaredWindowMarkersMatchWindowSpecs(); + } + + @Test + void seedSupportFlagMatchesMixin() { + capability().seedSupportFlagMatchesMixin(); + } + + @Test + void distinctHllSlidingIsFalse() { + capability().distinctHllSlidingIsFalse(); + } + + @Test + void unsupportedWindowQueryIsDistinguishableFailure() { + capability().unsupportedWindowQueryIsDistinguishableFailure(); + } + + @Test + void supportedEmptyWindowReturnsKnownZero() { + capability().supportedEmptyWindowReturnsKnownZero(); + } + + @Test + void unsupportedWindowApplyIsDistinguishableFailure() { + capability().unsupportedWindowApplyIsDistinguishableFailure(); + } + + @Test + void mixedApplyPartiallyFailsOnUnsupportedWindow() { + capability().mixedApplyPartiallyFailsOnUnsupportedWindow(); + } +} diff --git a/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisSupport.java b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisSupport.java new file mode 100644 index 0000000..2f5e4f8 --- /dev/null +++ b/velocity-backend-redis/src/test/java/com/codeheadsystems/velocity/backend/redis/RedisSupport.java @@ -0,0 +1,55 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.backend.redis; + +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Shared Testcontainers Redis fixture — one container per JVM, reused across every integration test + * class. Exposes a single Lettuce connection to the mapped port; {@link #flush()} resets state + * between tests so each starts from an empty Redis. + */ +final class RedisSupport { + + @SuppressWarnings("resource") // container + client are closed by the JVM shutdown hook + private static final GenericContainer CONTAINER = + new GenericContainer<>(DockerImageName.parse("redis:7-alpine")).withExposedPorts(6379); + + private static RedisClient client; + private static StatefulRedisConnection connection; + + private RedisSupport() {} + + /** Lazily starts the container and opens a reusable Lettuce connection. */ + static synchronized StatefulRedisConnection connection() { + if (connection != null) { + return connection; + } + if (!CONTAINER.isRunning()) { + CONTAINER.start(); + } + client = + RedisClient.create(RedisURI.create(CONTAINER.getHost(), CONTAINER.getMappedPort(6379))); + connection = client.connect(); + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + if (connection != null) { + connection.close(); + } + if (client != null) { + client.shutdown(); + } + })); + return connection; + } + + /** Flushes every key so a test starts from an empty store. */ + static void flush() { + connection().sync().flushall(); + } +}