diff --git a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/CountStore.java b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/CountStore.java index 5f860a4..6978390 100644 --- a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/CountStore.java +++ b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/CountStore.java @@ -25,9 +25,17 @@ public interface CountStore extends VelocityBackend { * APPLIED|FAILED|SKIPPED} — a {@code SKIPPED} being a backend-owned outcome such as an idempotent * replay) alongside the {@link FeatureResult} (ADR 0009, FR-34). * + *

The result cardinality is one {@link + * com.codeheadsystems.velocity.spi.model.PerFeature PerFeature} per {@code (intent × each of the + * intent's feature's windows)}, not one per intent: a feature tracked over several + * windows contributes one entry per window, each carrying that window's post-apply value. The + * entries are therefore not positionally aligned with {@code intents}; disambiguate by the + * entry's feature and its {@link com.codeheadsystems.velocity.spi.model.FeatureValue#window() + * window}. + * * @param ctx the namespace + optional deadline context * @param intents the count intents to apply - * @return the per-intent apply outcomes, positionally aligned with {@code intents} + * @return one apply outcome per {@code (intent × the intent's feature's windows)} */ ApplyResult applyCount(ApplyContext ctx, List intents); diff --git a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/DistinctStore.java b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/DistinctStore.java index e1717fa..2fa4608 100644 --- a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/DistinctStore.java +++ b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/DistinctStore.java @@ -30,9 +30,17 @@ public interface DistinctStore extends VelocityBackend { * com.codeheadsystems.velocity.spi.model.ApplyStatus apply status} ({@code * APPLIED|FAILED|SKIPPED}) alongside the {@link FeatureResult} (ADR 0009, FR-34). * + *

The result cardinality is one {@link + * com.codeheadsystems.velocity.spi.model.PerFeature PerFeature} per {@code (intent × each of the + * intent's feature's windows)}, not one per intent: a feature tracked over several + * windows contributes one entry per window, each carrying that window's post-apply value. The + * entries are therefore not positionally aligned with {@code intents}; disambiguate by the + * entry's feature and its {@link com.codeheadsystems.velocity.spi.model.FeatureValue#window() + * window}. + * * @param ctx the namespace + optional deadline context * @param intents the distinct intents to apply - * @return the per-intent apply outcomes, positionally aligned with {@code intents} + * @return one apply outcome per {@code (intent × the intent's feature's windows)} */ ApplyResult applyDistinct(ApplyContext ctx, List intents); diff --git a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/SumStore.java b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/SumStore.java index b70b35b..0e5b2c2 100644 --- a/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/SumStore.java +++ b/velocity-spi/src/main/java/com/codeheadsystems/velocity/spi/SumStore.java @@ -24,9 +24,17 @@ public interface SumStore extends VelocityBackend { * com.codeheadsystems.velocity.spi.model.ApplyStatus apply status} ({@code * APPLIED|FAILED|SKIPPED}) alongside the {@link FeatureResult} (ADR 0009, FR-34). * + *

The result cardinality is one {@link + * com.codeheadsystems.velocity.spi.model.PerFeature PerFeature} per {@code (intent × each of the + * intent's feature's windows)}, not one per intent: a feature tracked over several + * windows contributes one entry per window, each carrying that window's post-apply value. The + * entries are therefore not positionally aligned with {@code intents}; disambiguate by the + * entry's feature and its {@link com.codeheadsystems.velocity.spi.model.FeatureValue#window() + * window}. + * * @param ctx the namespace + optional deadline context * @param intents the sum intents to apply - * @return the per-intent apply outcomes, positionally aligned with {@code intents} + * @return one apply outcome per {@code (intent × the intent's feature's windows)} */ ApplyResult applySum(ApplyContext ctx, List intents); diff --git a/velocity-testkit/build.gradle.kts b/velocity-testkit/build.gradle.kts index 7647898..8921177 100644 --- a/velocity-testkit/build.gradle.kts +++ b/velocity-testkit/build.gradle.kts @@ -13,13 +13,9 @@ dependencies { // contract; both are on the api surface so backend modules inherit them from their tests. api(project(":velocity-spi")) api(libs.jspecify) - api(libs.bundles.jackson) // AssertJ is on the api surface because the *Scenarios TCK asserts with `assertThat`; backend // modules drive these scenarios from their integration tests and inherit the dependency. api(libs.assertj.core) - implementation(libs.caffeine) - implementation(libs.slf4j.api) - testRuntimeOnly(libs.logback.classic) } diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackend.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackend.java new file mode 100644 index 0000000..6a1fc88 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackend.java @@ -0,0 +1,412 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.SeedSupport; +import com.codeheadsystems.velocity.spi.SlidingSupport; +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.TumblingSupport; +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.BucketValue; +import com.codeheadsystems.velocity.spi.model.DistinctMember; +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.SeedAggregate.CountValue; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.ExactDistinct; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.HllDistinct; +import com.codeheadsystems.velocity.spi.model.SeedAggregate.SumValue; +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 java.math.BigDecimal; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import org.jspecify.annotations.Nullable; + +/** + * The reference in-memory {@code velocity-spi} backend (ADR 0004, NFR-13/18). + * + *

Two jobs: it proves the frozen SPI contract is implementable as an exact, + * atomic, thread-safe store, and it is the primary subject the conformance TCK ({@code + * com.codeheadsystems.velocity.testkit.tck}) runs against. It implements every data-plane mix-in — + * {@link CountStore}, {@link SumStore}, {@link DistinctStore} — over both window markers ({@link + * SlidingSupport}, {@link TumblingSupport}) plus {@link SeedSupport}, and declares {@link + * ReadYourWriteLevel#ATOMIC} because an apply's returned value already reflects that write. + * + *

Model. Windowing lives in the backend (ADR 0003). The store keeps raw + * events per {@code (namespace, subject, aggregation)} — a list of instants for COUNT, of + * {@code (instant, cents)} for SUM, of {@code (instant, member)} for DISTINCT — and every window a + * feature declares is computed from that one stored event stream at read time (so one + * apply records a single underlying event, not one per window). This is exactly the property that + * lets a seeded bucket and an organically recorded bucket merge identically (ADR 0008). Window + * arithmetic is {@link WindowMath}. The backend is the clock authority (FR-3): inject a {@link + * MutableClock} to drive window aging deterministically. + * + *

Distinct is exact-only here. The clamp/threshold are set effectively + * infinite, so this backend never degrades to HLL; it therefore rejects an {@link HllDistinct} seed + * (ADR 0005/0006). + */ +public final class InMemoryVelocityBackend + implements CountStore, SumStore, DistinctStore, SlidingSupport, TumblingSupport, SeedSupport { + + private final Clock clock; + private final BackendCapabilities capabilities; + private final ConcurrentMap store = new ConcurrentHashMap<>(); + + /** Creates a backend whose clock is {@link Clock#systemUTC()}. */ + public InMemoryVelocityBackend() { + this(Clock.systemUTC()); + } + + /** + * Creates a backend driven by the given clock — the authority for all window edges (FR-3). + * + * @param clock the backend clock; inject a {@link MutableClock} to control time in tests + */ + public InMemoryVelocityBackend(final Clock clock) { + this.clock = Objects.requireNonNull(clock, "clock"); + this.capabilities = buildCapabilities(); + } + + private static BackendCapabilities buildCapabilities() { + final List windows = + List.of( + windowSpec(WindowType.SLIDING, Duration.ofSeconds(5)), + windowSpec(WindowType.SLIDING, Duration.ofMinutes(1)), + windowSpec(WindowType.TUMBLING, Duration.ofMinutes(1)), + windowSpec(WindowType.TUMBLING, Duration.ofHours(1))); + return new BackendCapabilities( + Set.of(AggregationType.COUNT, AggregationType.SUM, AggregationType.DISTINCT), + windows, + /* distinctHllSliding= */ false, + /* distinctExactCardinalityClamp= */ Long.MAX_VALUE, + /* distinctHllThresholdDefault= */ Long.MAX_VALUE, + /* maxRetention= */ Duration.ofDays(30), + ReadYourWriteLevel.ATOMIC, + /* idempotencySupported= */ false, + /* seedSupported= */ true, + /* maxAtomicFanOut= */ Integer.MAX_VALUE); + } + + private static WindowSpec windowSpec(final WindowType type, final Duration duration) { + // In-memory keeps raw events, so "granularity" is nominal; report it as the window duration. + return new WindowSpec(new Window(duration, type), Exactness.EXACT, duration); + } + + /** + * 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; + } + + @Override + public ApplyResult applyCount(final ApplyContext ctx, final List intents) { + return applyIntents(ctx, intents); + } + + @Override + public ApplyResult applySum(final ApplyContext ctx, final List intents) { + return applyIntents(ctx, intents); + } + + @Override + public ApplyResult applyDistinct(final ApplyContext ctx, final List intents) { + return applyIntents(ctx, intents); + } + + private ApplyResult applyIntents(final ApplyContext ctx, final List intents) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(intents, "intents"); + final Namespace namespace = ctx.namespace(); + final Instant now = clock.instant(); + final List outcomes = new ArrayList<>(); + for (final Intent intent : intents) { + final Feature feature = intent.feature(); + final StoreKey key = new StoreKey(namespace, intent.subject(), feature.aggregation()); + final EventLog log = store.computeIfAbsent(key, unused -> new EventLog()); + final Event event = toEvent(intent, now); + synchronized (log) { + final boolean anySupported = + feature.windows().stream().anyMatch(capabilities::supportsWindow); + if (anySupported) { + log.add(event); + } + final AggregationType type = feature.aggregation().type(); + for (final Window window : feature.windows()) { + if (capabilities.supportsWindow(window)) { + final FeatureValue value = log.valueOf(feature, window, type, now); + outcomes.add( + new PerFeature(feature, ApplyStatus.APPLIED, FeatureResult.success(value))); + } else { + outcomes.add( + new PerFeature( + feature, + ApplyStatus.FAILED, + FeatureResult.failure(FailureCode.UNSUPPORTED_WINDOW, unsupported(window)))); + } + } + } + } + return new ApplyResult(outcomes); + } + + private static Event toEvent(final Intent intent, final Instant now) { + return switch (intent) { + case CountIntent ignored -> new Event(now, null, null); + case SumIntent sum -> new Event(now, sum.valueCents(), null); + case DistinctIntent distinct -> new Event(now, null, distinct.member()); + }; + } + + @Override + public List queryCount(final QueryContext ctx, final List tuples) { + return queryTuples(ctx, tuples); + } + + @Override + public List querySum(final QueryContext ctx, final List tuples) { + return queryTuples(ctx, tuples); + } + + @Override + public List queryDistinct(final QueryContext ctx, final List tuples) { + return queryTuples(ctx, tuples); + } + + private List queryTuples(final QueryContext ctx, final List tuples) { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(tuples, "tuples"); + final Namespace namespace = ctx.namespace(); + final Instant now = clock.instant(); + 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 AggregationType type = tuple.aggregation().type(); + final StoreKey key = new StoreKey(namespace, tuple.subject(), tuple.aggregation()); + final EventLog log = store.get(key); + final Feature synthetic = syntheticFeature(tuple); + final BigDecimal value; + if (log == null) { + // A supported window with genuinely no data is a KNOWN zero — Success(0), never a failure. + value = BigDecimal.ZERO; + } else { + synchronized (log) { + value = log.compute(window, WindowMath.boundsAt(window, now), type); + } + } + results.add( + FeatureResult.success( + new FeatureValue( + synthetic, + window, + value, + Exactness.EXACT, + ReadYourWriteLevel.ATOMIC, + /* definitionVersionHash= */ null, + WindowMath.boundsAt(window, now), + now))); + } + return results; + } + + @Override + public void purge(final Namespace namespace, final @Nullable Subject subject) { + Objects.requireNonNull(namespace, "namespace"); + if (subject == null) { + store.keySet().removeIf(key -> key.namespace().equals(namespace)); + } else { + store + .keySet() + .removeIf(key -> key.namespace().equals(namespace) && key.subject().equals(subject)); + } + } + + @Override + public void seed( + final Namespace namespace, + final Subject subject, + final Feature feature, + final List buckets) { + Objects.requireNonNull(namespace, "namespace"); + Objects.requireNonNull(subject, "subject"); + Objects.requireNonNull(feature, "feature"); + Objects.requireNonNull(buckets, "buckets"); + final StoreKey key = new StoreKey(namespace, subject, feature.aggregation()); + for (final BucketValue bucket : buckets) { + final Duration bucketDuration = + Duration.between(bucket.bounds().start(), bucket.bounds().end()); + final Window matched = matchingSupportedWindow(feature, bucketDuration); + if (matched == null) { + throw new IllegalArgumentException( + "seed bucket of " + + bucketDuration + + " matches no supported window of feature '" + + feature.name() + + "'"); + } + final Instant at = midpoint(bucket.bounds()); + final EventLog log = store.computeIfAbsent(key, unused -> new EventLog()); + synchronized (log) { + switch (bucket.aggregate()) { + case CountValue count -> { + for (long i = 0; i < count.count(); i++) { + log.add(new Event(at, null, null)); + } + } + case SumValue sum -> log.add(new Event(at, sum.cents(), null)); + case ExactDistinct exact -> { + for (final DistinctMember member : exact.members()) { + log.add(new Event(at, null, member)); + } + } + case HllDistinct hll -> + throw new IllegalArgumentException( + "in-memory backend is exact-only distinct; rejecting HLL seed " + + hll + + " (ADR 0005/0006)"); + } + } + } + } + + private @Nullable Window matchingSupportedWindow( + final Feature feature, final Duration bucketDuration) { + for (final Window window : feature.windows()) { + if (window.duration().equals(bucketDuration) && capabilities.supportsWindow(window)) { + return window; + } + } + return null; + } + + private static Instant midpoint(final WindowBounds bounds) { + return bounds.start().plus(Duration.between(bounds.start(), bounds.end()).dividedBy(2)); + } + + 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 in-memory backend"; + } + + /** + * The join key both apply and query resolve to; the aggregation carries the DISTINCT dimension. + */ + private record StoreKey(Namespace namespace, Subject subject, Aggregation aggregation) {} + + /** One raw event; {@code cents} is set only for SUM, {@code member} only for DISTINCT. */ + private record Event( + Instant timestamp, @Nullable BigDecimal cents, @Nullable DistinctMember member) {} + + /** + * The raw event stream for one {@link StoreKey}. Not thread-safe on its own; the backend + * synchronizes on the instance so an apply's add-then-read is atomic (read-your-write, ADR 0007). + */ + private static final class EventLog { + + private final List events = new ArrayList<>(); + + void add(final Event event) { + events.add(event); + } + + FeatureValue valueOf( + final Feature feature, final Window window, final AggregationType type, final Instant now) { + final WindowBounds bounds = WindowMath.boundsAt(window, now); + return new FeatureValue( + feature, + window, + compute(window, bounds, type), + Exactness.EXACT, + ReadYourWriteLevel.ATOMIC, + /* definitionVersionHash= */ null, + bounds, + now); + } + + BigDecimal compute(final Window window, final WindowBounds bounds, final AggregationType type) { + return switch (type) { + case COUNT -> { + long count = 0; + for (final Event event : events) { + if (WindowMath.contains(window, bounds, event.timestamp())) { + count++; + } + } + yield BigDecimal.valueOf(count); + } + case SUM -> { + BigDecimal sum = BigDecimal.ZERO; + for (final Event event : events) { + final BigDecimal cents = event.cents(); + if (cents != null && WindowMath.contains(window, bounds, event.timestamp())) { + sum = sum.add(cents); + } + } + yield sum; + } + case DISTINCT -> { + final Set distinct = new HashSet<>(); + for (final Event event : events) { + final DistinctMember member = event.member(); + if (member != null && WindowMath.contains(window, bounds, event.timestamp())) { + distinct.add(member); + } + } + yield BigDecimal.valueOf(distinct.size()); + } + }; + } + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/MutableClock.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/MutableClock.java new file mode 100644 index 0000000..8ef2643 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/MutableClock.java @@ -0,0 +1,90 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Objects; + +/** + * A test-controllable {@link Clock} whose instant can be set or advanced by hand. + * + *

The backend is the clock authority for windowing (FR-3, ADR 0003), so the conformance TCK + * needs to move time deterministically — age an event out of a sliding window, cross a tumbling + * bucket boundary — without sleeping. Inject one of these into any backend that accepts a {@link + * Clock} (the {@link InMemoryVelocityBackend} does) and the same instance drives both the backend's + * bucketing and the scenario's expectations. + * + *

The stored instant is {@code volatile} so an advance on one thread is visible to a concurrent + * apply/query on another; it is not a substitute for the backend's own per-key locking. + */ +public final class MutableClock extends Clock { + + private final ZoneId zone; + private volatile Instant instant; + + /** + * Creates a clock fixed at {@code start} in UTC. + * + * @param start the initial instant + */ + public MutableClock(final Instant start) { + this(start, ZoneOffset.UTC); + } + + private MutableClock(final Instant start, final ZoneId zone) { + this.instant = Objects.requireNonNull(start, "start"); + this.zone = Objects.requireNonNull(zone, "zone"); + } + + /** + * A clock fixed at the current system instant, truncated to milliseconds so bucket-boundary math + * is free of sub-millisecond noise. + * + * @return a new mutable clock at "now" + */ + public static MutableClock atNow() { + return new MutableClock(Instant.ofEpochMilli(System.currentTimeMillis())); + } + + /** + * Sets the clock to an explicit instant. + * + * @param newInstant the instant the clock should report from now on + */ + public void setInstant(final Instant newInstant) { + this.instant = Objects.requireNonNull(newInstant, "newInstant"); + } + + /** + * Advances the clock forward by the given amount. + * + * @param amount the (non-negative) duration to move forward + */ + public void advance(final Duration amount) { + Objects.requireNonNull(amount, "amount"); + this.instant = this.instant.plus(amount); + } + + @Override + public ZoneId getZone() { + return zone; + } + + @Override + public Clock withZone(final ZoneId newZone) { + return new MutableClock(instant, newZone); + } + + @Override + public Instant instant() { + return instant; + } + + @Override + public long millis() { + return instant.toEpochMilli(); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/WindowMath.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/WindowMath.java new file mode 100644 index 0000000..24f17bc --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/WindowMath.java @@ -0,0 +1,86 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import java.time.Instant; +import java.util.Objects; + +/** + * The window-bounds arithmetic the {@link InMemoryVelocityBackend} computes at read time (ADR 0003: + * the backend owns time-shaping). Exposed as a small pure utility so both the backend and the + * conformance TCK compute the same bounds — a scenario can seed exactly the bucket a query + * will later merge. + * + *

Semantics (FR-14): + * + *

+ */ +public final class WindowMath { + + private WindowMath() {} + + /** + * The concrete {@link WindowBounds} a value over {@code window} covers as of {@code now}. + * + * @param window the window (duration + type) + * @param now the backend-clock instant the read is as-of + * @return the reported bounds for the window at {@code now} + */ + public static WindowBounds boundsAt(final Window window, final Instant now) { + Objects.requireNonNull(window, "window"); + Objects.requireNonNull(now, "now"); + return switch (window.type()) { + case SLIDING -> new WindowBounds(now.minus(window.duration()), now); + case TUMBLING -> tumblingBucket(now, window.duration().toMillis()); + }; + } + + /** + * The aligned tumbling bucket {@code [floor(nowMs/durationMs)·durationMs, +durationMs)} covering + * {@code now}. + * + * @param now the instant to align + * @param durationMs the bucket width in milliseconds; must be strictly positive + * @return the bucket's bounds + */ + public static WindowBounds tumblingBucket(final Instant now, final long durationMs) { + Objects.requireNonNull(now, "now"); + if (durationMs <= 0) { + throw new IllegalArgumentException("durationMs must be positive, was " + durationMs); + } + final long nowMs = now.toEpochMilli(); + final long startMs = Math.floorDiv(nowMs, durationMs) * durationMs; + return new WindowBounds( + Instant.ofEpochMilli(startMs), Instant.ofEpochMilli(startMs + durationMs)); + } + + /** + * Whether an event at {@code timestamp} falls inside {@code window} whose reported {@code bounds} + * were computed by {@link #boundsAt}. Sliding is {@code (start, end]}; tumbling is {@code [start, + * end)}. + * + * @param window the window whose membership rule to apply + * @param bounds the bounds produced by {@link #boundsAt} for the same read + * @param timestamp the event timestamp to test + * @return {@code true} if the event contributes to the window's value + */ + public static boolean contains( + final Window window, final WindowBounds bounds, final Instant timestamp) { + Objects.requireNonNull(window, "window"); + Objects.requireNonNull(bounds, "bounds"); + Objects.requireNonNull(timestamp, "timestamp"); + return switch (window.type()) { + case SLIDING -> timestamp.isAfter(bounds.start()) && !timestamp.isAfter(bounds.end()); + case TUMBLING -> !timestamp.isBefore(bounds.start()) && timestamp.isBefore(bounds.end()); + }; + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/package-info.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/package-info.java index e47bc79..f0f1f22 100644 --- a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/package-info.java +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/package-info.java @@ -1,6 +1,16 @@ // SPDX-License-Identifier: BSD-3-Clause /** - * The velocity-testkit: in-memory velocity-spi backend, conformance TCK scenarios, and fixtures. + * The velocity-testkit: the in-memory reference {@code velocity-spi} backend and its time helpers. + * + *

{@link com.codeheadsystems.velocity.testkit.InMemoryVelocityBackend} is the exact, atomic, + * thread-safe reference implementation that proves the frozen SPI contract is implementable (ADR + * 0004, NFR-13/18); {@link com.codeheadsystems.velocity.testkit.MutableClock} and {@link + * com.codeheadsystems.velocity.testkit.WindowMath} let the conformance TCK ({@link + * com.codeheadsystems.velocity.testkit.tck}) drive window aging deterministically. Every {@code + * velocity-backend-*} must pass that TCK for its declared capabilities. */ +@NullMarked package com.codeheadsystems.velocity.testkit; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CapabilityConformanceScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CapabilityConformanceScenarios.java new file mode 100644 index 0000000..76e23c1 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CapabilityConformanceScenarios.java @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.DIMENSION; +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.UNSUPPORTED; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertFailure; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.SeedSupport; +import com.codeheadsystems.velocity.spi.SlidingSupport; +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.TumblingSupport; +import com.codeheadsystems.velocity.spi.VelocityBackend; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.AggregationType; +import com.codeheadsystems.velocity.spi.model.BackendCapabilities; +import com.codeheadsystems.velocity.spi.model.FailureCode; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.QueryContext; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.util.List; +import java.util.Objects; + +/** + * The ADR 0004 coherence suite: what a backend declares in {@link BackendCapabilities} + * must match what it implements, and the negatives are contract. + * Declaration↔implementation agreement, {@code distinctHllSliding == false}, an unsupported window + * that fast-rejects with a distinguishable {@link FailureCode#UNSUPPORTED_WINDOW} (never a silent + * {@code 0}), and a supported empty window that returns a known {@code Success(0)} rather than a + * failure. + */ +public final class CapabilityConformanceScenarios { + + private final VelocityBackend backend; + + /** + * @param backend the backend under test + */ + public CapabilityConformanceScenarios(final VelocityBackend backend) { + this.backend = Objects.requireNonNull(backend, "backend"); + } + + /** Each declared aggregation is implemented as its mix-in, and each undeclared one is not. */ + public void declaredAggregationsMatchImplementedMixins() { + final BackendCapabilities caps = backend.capabilities(); + assertThat(caps.supportsAggregation(AggregationType.COUNT)) + .isEqualTo(backend instanceof CountStore); + assertThat(caps.supportsAggregation(AggregationType.SUM)) + .isEqualTo(backend instanceof SumStore); + assertThat(caps.supportsAggregation(AggregationType.DISTINCT)) + .isEqualTo(backend instanceof DistinctStore); + } + + /** The window markers agree with the declared window specs' types. */ + public void declaredWindowMarkersMatchWindowSpecs() { + final BackendCapabilities caps = backend.capabilities(); + final boolean anySliding = + caps.windows().stream().anyMatch(spec -> spec.window().type() == WindowType.SLIDING); + final boolean anyTumbling = + caps.windows().stream().anyMatch(spec -> spec.window().type() == WindowType.TUMBLING); + assertThat(backend instanceof SlidingSupport).isEqualTo(anySliding); + assertThat(backend instanceof TumblingSupport).isEqualTo(anyTumbling); + } + + /** The seed capability flag agrees with the presence of the {@link SeedSupport} mix-in. */ + public void seedSupportFlagMatchesMixin() { + assertThat(backend.capabilities().seedSupported()).isEqualTo(backend instanceof SeedSupport); + } + + /** + * HLL-distinct on a sliding window is structurally impossible: the flag must be false (ADR 0005). + */ + public void distinctHllSlidingIsFalse() { + assertThat(backend.capabilities().distinctHllSliding()).isFalse(); + } + + /** + * Querying an unsupported window fast-rejects with a distinguishable {@link + * FailureCode#UNSUPPORTED_WINDOW} — never a silent {@code 0} (ADR 0009 rule 1, FR-13). + */ + public void unsupportedWindowQueryIsDistinguishableFailure() { + final FeatureResult result = queryOne(UNSUPPORTED); + assertThat(result.isSuccess()).isFalse(); + assertFailure(result, FailureCode.UNSUPPORTED_WINDOW); + } + + /** + * A supported window with no data returns a known {@code Success(0)}, not a failure — the backend + * knows the answer is zero (distinct from "I could not answer"). + */ + public void supportedEmptyWindowReturnsKnownZero() { + final Window supported = backend.capabilities().windows().get(0).window(); + assertValue(queryOne(supported), 0); + } + + private FeatureResult queryOne(final Window window) { + final QueryContext ctx = Tck.query(NS_A); + if (backend instanceof CountStore countStore) { + return countStore + .queryCount(ctx, List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), window))) + .get(0); + } + if (backend instanceof SumStore sumStore) { + return sumStore + .querySum(ctx, List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), window))) + .get(0); + } + if (backend instanceof DistinctStore distinctStore) { + return distinctStore + .queryDistinct( + ctx, List.of(Tck.tuple(SUBJECT_A, Aggregation.distinct(DIMENSION), window))) + .get(0); + } + throw new IllegalStateException("backend implements no aggregation store to query"); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CountStoreScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CountStoreScenarios.java new file mode 100644 index 0000000..d8d4385 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/CountStoreScenarios.java @@ -0,0 +1,184 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_5S; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.TUMBLING_1H; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.countFeature; +import static com.codeheadsystems.velocity.testkit.tck.Tck.successValue; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.ApplyStatus; +import com.codeheadsystems.velocity.spi.model.Exactness; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.PerFeature; +import com.codeheadsystems.velocity.spi.model.ReadYourWriteLevel; +import com.codeheadsystems.velocity.testkit.MutableClock; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * The {@link CountStore} contract: apply records events, query reads back the exact windowed count, + * the write is reflected read-your-write, and values are isolated per namespace and subject. + */ +public final class CountStoreScenarios { + + private final CountStore store; + private final MutableClock clock; + + /** + * @param store a fresh count-capable backend under test + * @param clock the backend's controllable clock + */ + public CountStoreScenarios(final CountStore store, final MutableClock clock) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** Apply N events then query the window returns exactly N, flagged EXACT/ATOMIC. */ + public void applyThenQueryReturnsExactCount() { + final Feature feature = countFeature(SLIDING_1M); + store.applyCount(Tck.apply(NS_A), intents(feature, 3)); + + final var results = + store.queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_1M))); + assertThat(results).hasSize(1); + assertValue(results.get(0), 3); + assertThat(successValue(results.get(0)).exactness()).isEqualTo(Exactness.EXACT); + assertThat(successValue(results.get(0)).readYourWriteLevel()) + .isEqualTo(ReadYourWriteLevel.ATOMIC); + // The value is as-of the backend clock (FR-3/FR-7), not the caller's. + assertThat(successValue(results.get(0)).asOf()).isEqualTo(clock.instant()); + } + + /** The value returned by apply already reflects the caller's own write (ADR 0007, NFR-7). */ + public void applyResultReflectsWriteReadYourWrite() { + final Feature feature = countFeature(SLIDING_1M); + + final ApplyResult first = store.applyCount(Tck.apply(NS_A), intents(feature, 1)); + assertThat(first.perFeature()).hasSize(1); + assertThat(first.perFeature().get(0).status()).isEqualTo(ApplyStatus.APPLIED); + assertValue(first.perFeature().get(0).result(), 1); + + final ApplyResult second = store.applyCount(Tck.apply(NS_A), intents(feature, 1)); + assertValue(second.perFeature().get(0).result(), 2); + } + + /** A multi-window feature yields one {@link PerFeature} per window (apply cardinality). */ + public void applyEmitsOneResultPerWindow() { + final Feature feature = countFeature(SLIDING_5S, TUMBLING_1H); + final ApplyResult result = store.applyCount(Tck.apply(NS_A), intents(feature, 1)); + + assertThat(result.perFeature()).hasSize(2); + assertThat(result.perFeature()) + .allSatisfy(pf -> assertThat(pf.status()).isEqualTo(ApplyStatus.APPLIED)); + assertThat(result.perFeature().stream().map(pf -> successValue(pf.result()).window()).toList()) + .containsExactlyInAnyOrder(SLIDING_5S, TUMBLING_1H); + // One underlying event, not one per window: each window sees exactly the single apply. + assertThat(result.perFeature()).allSatisfy(pf -> assertValue(pf.result(), 1)); + } + + /** Counts in one namespace are invisible in another; the other reads a known zero. */ + public void valuesIsolatedByNamespace() { + store.applyCount(Tck.apply(NS_A), intents(countFeature(SLIDING_1M), 2)); + + assertValue( + store + .queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_1M))) + .get(0), + 2); + assertValue( + store + .queryCount( + Tck.query(NS_B), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_1M))) + .get(0), + 0); + } + + /** Counts for one subject are invisible for another subject in the same namespace. */ + public void valuesIsolatedBySubject() { + store.applyCount( + Tck.apply(NS_A), List.of(new CountIntent(countFeature(SLIDING_1M), SUBJECT_A))); + + assertValue( + store + .queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_1M))) + .get(0), + 1); + assertValue( + store + .queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_B, Aggregation.count(), SLIDING_1M))) + .get(0), + 0); + } + + /** + * Concurrent applies to the same subject are atomic: N threads each record once and the final + * count is exactly N, never lost to a read-modify-write race (NFR-6). + */ + public void concurrentApplyIsAtomic() throws Exception { + final Feature feature = countFeature(SLIDING_1M); + final int threads = 16; + final CountDownLatch ready = new CountDownLatch(threads); + final CountDownLatch fire = new CountDownLatch(1); + try (ExecutorService pool = Executors.newFixedThreadPool(threads)) { + final List> futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + futures.add( + pool.submit( + () -> { + ready.countDown(); + await(fire); + store.applyCount(Tck.apply(NS_A), intents(feature, 1)); + })); + } + assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue(); + fire.countDown(); + for (final Future future : futures) { + future.get(10, TimeUnit.SECONDS); + } + } + assertValue( + store + .queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_1M))) + .get(0), + threads); + } + + private static void await(final CountDownLatch latch) { + try { + latch.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + + private static List intents(final Feature feature, final int count) { + final List intents = new ArrayList<>(); + for (int i = 0; i < count; i++) { + intents.add(new CountIntent(feature, SUBJECT_A)); + } + return intents; + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/DistinctStoreScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/DistinctStoreScenarios.java new file mode 100644 index 0000000..da8fbb1 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/DistinctStoreScenarios.java @@ -0,0 +1,112 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.DIMENSION; +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.distinctFeature; +import static com.codeheadsystems.velocity.testkit.tck.Tck.member; +import static com.codeheadsystems.velocity.testkit.tck.Tck.successValue; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.DistinctStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.Exactness; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.Intent.DistinctIntent; +import com.codeheadsystems.velocity.testkit.MutableClock; +import java.util.List; +import java.util.Objects; + +/** + * The {@link DistinctStore} contract: cardinality counts distinct members, repeated members + * de-dupe, values are subject-isolated, and this exact backend flags every value {@link + * Exactness#EXACT}. + */ +public final class DistinctStoreScenarios { + + private final DistinctStore store; + private final MutableClock clock; + + /** + * @param store a fresh distinct-capable backend under test + * @param clock the backend's controllable clock + */ + public DistinctStoreScenarios(final DistinctStore store, final MutableClock clock) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** Apply distinct members then query returns the exact cardinality of the distinct set. */ + public void applyThenQueryReturnsCardinality() { + final Feature feature = distinctFeature(SLIDING_1M); + store.applyDistinct( + Tck.apply(NS_A), + List.of( + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("bob")), + new DistinctIntent(feature, SUBJECT_A, member("carol")))); + + final var result = + store + .queryDistinct( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.distinct(DIMENSION), SLIDING_1M))) + .get(0); + assertValue(result, 3); + assertThat(successValue(result).exactness()).isEqualTo(Exactness.EXACT); + assertThat(successValue(result).asOf()).isEqualTo(clock.instant()); + } + + /** Re-applying the same member does not increase cardinality. */ + public void deDupesRepeatedMembers() { + final Feature feature = distinctFeature(SLIDING_1M); + store.applyDistinct( + Tck.apply(NS_A), + List.of( + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("alice")), + new DistinctIntent(feature, SUBJECT_A, member("bob")))); + + assertValue( + store + .queryDistinct( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.distinct(DIMENSION), SLIDING_1M))) + .get(0), + 2); + } + + /** The value returned by apply already reflects the post-apply cardinality (read-your-write). */ + public void applyResultReflectsCardinality() { + final Feature feature = distinctFeature(SLIDING_1M); + final ApplyResult first = + store.applyDistinct( + Tck.apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("alice")))); + assertValue(first.perFeature().get(0).result(), 1); + + final ApplyResult second = + store.applyDistinct( + Tck.apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("bob")))); + assertValue(second.perFeature().get(0).result(), 2); + } + + /** Distinct members for one subject do not leak into another subject's cardinality. */ + public void valuesIsolatedBySubject() { + final Feature feature = distinctFeature(SLIDING_1M); + store.applyDistinct( + Tck.apply(NS_A), List.of(new DistinctIntent(feature, SUBJECT_A, member("alice")))); + + assertValue( + store + .queryDistinct( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_B, Aggregation.distinct(DIMENSION), SLIDING_1M))) + .get(0), + 0); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/PurgeScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/PurgeScenarios.java new file mode 100644 index 0000000..45ac585 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/PurgeScenarios.java @@ -0,0 +1,70 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.countFeature; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +import com.codeheadsystems.velocity.spi.model.Subject; +import java.util.List; +import java.util.Objects; + +/** + * The {@link com.codeheadsystems.velocity.spi.VelocityBackend#purge purge} contract (FR-23): {@code + * purge(ns, subject)} erases only that subject within the namespace, and {@code purge(ns, null)} + * erases the whole namespace — neither touches another namespace. + */ +public final class PurgeScenarios { + + private final CountStore store; + + /** + * @param store a fresh count-capable backend under test + */ + public PurgeScenarios(final CountStore store) { + this.store = Objects.requireNonNull(store, "store"); + } + + /** {@code purge(ns, subject)} clears that subject only; other subjects survive. */ + public void purgeSubjectClearsThatSubjectOnly() { + final Feature feature = countFeature(SLIDING_1M); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_B))); + + store.purge(NS_A, SUBJECT_A); + + assertValue(count(NS_A, SUBJECT_A), 0); + assertValue(count(NS_A, SUBJECT_B), 1); + } + + /** {@code purge(ns, null)} clears the whole namespace; another namespace is untouched. */ + public void purgeNamespaceClearsEntireNamespace() { + final Feature feature = countFeature(SLIDING_1M); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_B))); + store.applyCount(Tck.apply(NS_B), List.of(new CountIntent(feature, SUBJECT_A))); + + store.purge(NS_A, null); + + assertValue(count(NS_A, SUBJECT_A), 0); + assertValue(count(NS_A, SUBJECT_B), 0); + assertValue(count(NS_B, SUBJECT_A), 1); + } + + private FeatureResult count(final Namespace namespace, final Subject subject) { + return store + .queryCount( + Tck.query(namespace), List.of(Tck.tuple(subject, Aggregation.count(), SLIDING_1M))) + .get(0); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SeedSupportScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SeedSupportScenarios.java new file mode 100644 index 0000000..b38d33a --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SeedSupportScenarios.java @@ -0,0 +1,113 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.TUMBLING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.countFeature; +import static com.codeheadsystems.velocity.testkit.tck.Tck.distinctFeature; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.SeedSupport; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.BucketValue; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.SeedAggregate; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import com.codeheadsystems.velocity.testkit.MutableClock; +import com.codeheadsystems.velocity.testkit.WindowMath; +import java.util.List; +import java.util.Objects; + +/** + * The {@link SeedSupport} contract (ADR 0008): a seeded per-bucket aggregate and an organically + * recorded one merge identically through the same windowed read path (acceptance #16), an + * unrepresentable seed is rejected, and — by type — a single total can never be seeded (a {@link + * BucketValue} always carries {@link WindowBounds}, so there is no total-only variant to pass). + * + * @param a backend that both records/queries counts and supports seeding + */ +public final class SeedSupportScenarios { + + private final B backend; + private final MutableClock clock; + + /** + * @param backend a fresh seed-capable, count-capable backend under test + * @param clock the backend's controllable clock + */ + public SeedSupportScenarios(final B backend, final MutableClock clock) { + this.backend = Objects.requireNonNull(backend, "backend"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** + * A bucket seeded with {@code CountValue(n)} answers a windowed query exactly as {@code n} events + * recorded organically into that same bucket do — the seed/record equivalence of ADR 0008. + */ + public void seededBucketMergesIdenticallyWithRecorded() { + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + final Feature feature = countFeature(TUMBLING_1M); + + // Seeded path: subject A gets a pre-computed bucket count of 3. + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of(new BucketValue(bucket, new SeedAggregate.CountValue(3)))); + + // Recorded path: subject B gets 3 organically recorded events in the same current bucket. + backend.applyCount( + Tck.apply(NS_A), + List.of( + new CountIntent(feature, SUBJECT_B), + new CountIntent(feature, SUBJECT_B), + new CountIntent(feature, SUBJECT_B))); + + assertValue(query(SUBJECT_A), 3); + assertValue(query(SUBJECT_B), 3); + } + + /** An exact-only backend rejects an HLL-sketch distinct seed (ADR 0005/0006). */ + public void hllDistinctSeedRejected() { + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + final Feature feature = distinctFeature(TUMBLING_1M); + assertThatThrownBy( + () -> + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of( + new BucketValue( + bucket, new SeedAggregate.HllDistinct(new byte[] {1, 2, 3, 4}))))) + .isInstanceOf(IllegalArgumentException.class); + } + + /** A bucket whose span matches no supported window of the feature is rejected. */ + public void seedWithUnsupportedWindowRejected() { + final WindowBounds start = WindowMath.tumblingBucket(clock.instant(), 60_000); + // A 90-second bucket: the feature declares only a 1-minute window, so nothing matches. + final WindowBounds badBucket = new WindowBounds(start.start(), start.start().plusSeconds(90)); + final Feature feature = countFeature(TUMBLING_1M); + assertThatThrownBy( + () -> + backend.seed( + NS_A, + SUBJECT_A, + feature, + List.of(new BucketValue(badBucket, new SeedAggregate.CountValue(1))))) + .isInstanceOf(IllegalArgumentException.class); + } + + private com.codeheadsystems.velocity.spi.model.FeatureResult query( + final com.codeheadsystems.velocity.spi.model.Subject subject) { + return backend + .queryCount(Tck.query(NS_A), List.of(Tck.tuple(subject, Aggregation.count(), TUMBLING_1M))) + .get(0); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SlidingScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SlidingScenarios.java new file mode 100644 index 0000000..ed2a7bb --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SlidingScenarios.java @@ -0,0 +1,69 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_5S; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.countFeature; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.testkit.MutableClock; +import java.time.Duration; +import java.util.List; +import java.util.Objects; + +/** + * The {@link com.codeheadsystems.velocity.spi.SlidingSupport} contract: a sliding window covers + * {@code (now - duration, now]} against the backend clock (FR-3, ADR 0005). Events age out as the + * clock advances past the duration, and the leading edge is exclusive at exactly the duration + * boundary. + */ +public final class SlidingScenarios { + + private final CountStore store; + private final MutableClock clock; + + /** + * @param store a fresh count-capable, sliding backend under test + * @param clock the backend's controllable clock + */ + public SlidingScenarios(final CountStore store, final MutableClock clock) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** + * An event stays in the window until the clock advances past the window duration, then drops out. + */ + public void eventsAgeOutAsClockAdvances() { + final Feature feature = countFeature(SLIDING_5S); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + + assertValue(currentCount(), 1); + clock.advance(Duration.ofSeconds(3)); + assertValue(currentCount(), 1); // 3s < 5s, still inside + clock.advance(Duration.ofSeconds(3)); // now +6s, past the 5s window + assertValue(currentCount(), 0); + } + + /** At exactly {@code now == eventTime + duration} the event is on the exclusive leading edge. */ + public void slidingLeadingEdgeIsExclusive() { + final Feature feature = countFeature(SLIDING_5S); + store.applyCount(Tck.apply(NS_A), List.of(new CountIntent(feature, SUBJECT_A))); + + assertValue(currentCount(), 1); + clock.advance( + Duration.ofSeconds(5)); // window is now (eventTime, eventTime+5s]; event sits at start + assertValue(currentCount(), 0); + } + + private com.codeheadsystems.velocity.spi.model.FeatureResult currentCount() { + return store + .queryCount(Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), SLIDING_5S))) + .get(0); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SumStoreScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SumStoreScenarios.java new file mode 100644 index 0000000..075a717 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/SumStoreScenarios.java @@ -0,0 +1,141 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_B; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SLIDING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.cents; +import static com.codeheadsystems.velocity.testkit.tck.Tck.successValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.sumFeature; +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.SumStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyResult; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.Intent.SumIntent; +import com.codeheadsystems.velocity.testkit.MutableClock; +import java.math.BigDecimal; +import java.util.List; +import java.util.Objects; + +/** + * The {@link SumStore} contract: sums are exact {@code BigDecimal} cents (P3/FR-10) — no binary + * float, no silent overflow — negatives are honored for refunds, and values are namespace-isolated. + */ +public final class SumStoreScenarios { + + private final SumStore store; + private final MutableClock clock; + + /** + * @param store a fresh sum-capable backend under test + * @param clock the backend's controllable clock + */ + public SumStoreScenarios(final SumStore store, final MutableClock clock) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** Apply cents values then query returns their exact sum with scale 0. */ + public void applyThenQueryReturnsExactSumCents() { + final Feature feature = sumFeature(SLIDING_1M); + store.applySum( + Tck.apply(NS_A), + List.of( + new SumIntent(feature, SUBJECT_A, cents(150)), + new SumIntent(feature, SUBJECT_A, cents(250)))); + + final BigDecimal value = + successValue( + store + .querySum( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), SLIDING_1M))) + .get(0)) + .value(); + assertThat(value).isEqualByComparingTo(cents(400)); + assertThat(value.scale()).isZero(); + assertThat( + successValue( + store + .querySum( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), SLIDING_1M))) + .get(0)) + .asOf()) + .isEqualTo(clock.instant()); + } + + /** The value returned by apply already reflects the running sum (read-your-write). */ + public void applyResultReflectsRunningSum() { + final Feature feature = sumFeature(SLIDING_1M); + final ApplyResult first = + store.applySum(Tck.apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, cents(500)))); + assertThat(successValue(first.perFeature().get(0).result()).value()) + .isEqualByComparingTo(cents(500)); + + final ApplyResult second = + store.applySum(Tck.apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, cents(125)))); + assertThat(successValue(second.perFeature().get(0).result()).value()) + .isEqualByComparingTo(cents(625)); + } + + /** + * Large cents sums are preserved exactly — well past a {@code long}-cents / double's safe range. + */ + public void bigDecimalCentsPreservedWithoutOverflow() { + final Feature feature = sumFeature(SLIDING_1M); + final BigDecimal big = new BigDecimal("9000000000000000000"); // > Long.MAX_VALUE / 2 + store.applySum( + Tck.apply(NS_A), + List.of(new SumIntent(feature, SUBJECT_A, big), new SumIntent(feature, SUBJECT_A, big))); + + final BigDecimal value = + successValue( + store + .querySum( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), SLIDING_1M))) + .get(0)) + .value(); + assertThat(value).isEqualByComparingTo(big.add(big)); + } + + /** A negative value (refund) reduces the sum. */ + public void negativeValuesForRefunds() { + final Feature feature = sumFeature(SLIDING_1M); + store.applySum( + Tck.apply(NS_A), + List.of( + new SumIntent(feature, SUBJECT_A, cents(500)), + new SumIntent(feature, SUBJECT_A, cents(-200)))); + + assertThat( + successValue( + store + .querySum( + Tck.query(NS_A), + List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), SLIDING_1M))) + .get(0)) + .value()) + .isEqualByComparingTo(cents(300)); + } + + /** Sums in one namespace are invisible in another; the other reads a known zero. */ + public void valuesIsolatedByNamespace() { + final Feature feature = sumFeature(SLIDING_1M); + store.applySum(Tck.apply(NS_A), List.of(new SumIntent(feature, SUBJECT_A, cents(999)))); + + assertThat( + successValue( + store + .querySum( + Tck.query(NS_B), + List.of(Tck.tuple(SUBJECT_A, Aggregation.sum(), SLIDING_1M))) + .get(0)) + .value()) + .isEqualByComparingTo(BigDecimal.ZERO); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/Tck.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/Tck.java new file mode 100644 index 0000000..49d1f49 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/Tck.java @@ -0,0 +1,95 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.ApplyContext; +import com.codeheadsystems.velocity.spi.model.DistinctMember; +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.Namespace; +import com.codeheadsystems.velocity.spi.model.QueryContext; +import com.codeheadsystems.velocity.spi.model.QueryTuple; +import com.codeheadsystems.velocity.spi.model.Subject; +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; + +/** Shared fixtures and assertion helpers for the conformance scenarios. */ +final class Tck { + + static final Namespace NS_A = new Namespace("tck-ns-a"); + static final Namespace NS_B = new Namespace("tck-ns-b"); + static final Subject SUBJECT_A = new Subject("card", "subject-a"); + static final Subject SUBJECT_B = new Subject("card", "subject-b"); + + static final Window SLIDING_5S = new Window(Duration.ofSeconds(5), WindowType.SLIDING); + static final Window SLIDING_1M = new Window(Duration.ofMinutes(1), WindowType.SLIDING); + static final Window TUMBLING_1M = new Window(Duration.ofMinutes(1), WindowType.TUMBLING); + static final Window TUMBLING_1H = new Window(Duration.ofHours(1), WindowType.TUMBLING); + + /** A window no reference-backend window spec covers — used to drive the negative paths. */ + static final Window UNSUPPORTED = new Window(Duration.ofDays(7), WindowType.SLIDING); + + static final String DIMENSION = "ip"; + + private Tck() {} + + static Feature countFeature(final Window... windows) { + return new Feature("tck.count", Aggregation.count(), List.of(windows)); + } + + static Feature sumFeature(final Window... windows) { + return new Feature("tck.sum", Aggregation.sum(), List.of(windows)); + } + + static Feature distinctFeature(final Window... windows) { + return new Feature("tck.distinct", Aggregation.distinct(DIMENSION), List.of(windows)); + } + + static ApplyContext apply(final Namespace namespace) { + return new ApplyContext(namespace, null); + } + + static QueryContext query(final Namespace namespace) { + return new QueryContext(namespace, null); + } + + static QueryTuple tuple( + final Subject subject, final Aggregation aggregation, final Window window) { + return new QueryTuple(subject, aggregation, window); + } + + static DistinctMember member(final String token) { + return new DistinctMember(token.getBytes(StandardCharsets.UTF_8)); + } + + static BigDecimal cents(final long amount) { + return BigDecimal.valueOf(amount); + } + + /** Asserts the result is a {@link FeatureResult.Success} and returns its value. */ + static FeatureValue successValue(final FeatureResult result) { + assertThat(result).isInstanceOf(FeatureResult.Success.class); + return ((FeatureResult.Success) result).value(); + } + + /** Asserts the result is a success whose numeric value equals {@code expected}. */ + static void assertValue(final FeatureResult result, final long expected) { + assertThat(successValue(result).value()).isEqualByComparingTo(BigDecimal.valueOf(expected)); + } + + /** Asserts the result is a {@link FeatureResult.Failure} carrying {@code code}. */ + static FeatureResult.Failure assertFailure( + final FeatureResult result, final com.codeheadsystems.velocity.spi.model.FailureCode code) { + assertThat(result).isInstanceOf(FeatureResult.Failure.class); + final FeatureResult.Failure failure = (FeatureResult.Failure) result; + assertThat(failure.code()).isEqualTo(code); + return failure; + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/TumblingScenarios.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/TumblingScenarios.java new file mode 100644 index 0000000..97baacd --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/TumblingScenarios.java @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit.tck; + +import static com.codeheadsystems.velocity.testkit.tck.Tck.NS_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.SUBJECT_A; +import static com.codeheadsystems.velocity.testkit.tck.Tck.TUMBLING_1M; +import static com.codeheadsystems.velocity.testkit.tck.Tck.assertValue; +import static com.codeheadsystems.velocity.testkit.tck.Tck.countFeature; + +import com.codeheadsystems.velocity.spi.CountStore; +import com.codeheadsystems.velocity.spi.model.Aggregation; +import com.codeheadsystems.velocity.spi.model.Feature; +import com.codeheadsystems.velocity.spi.model.FeatureResult; +import com.codeheadsystems.velocity.spi.model.Intent.CountIntent; +import com.codeheadsystems.velocity.testkit.MutableClock; +import com.codeheadsystems.velocity.testkit.WindowMath; +import java.time.Instant; +import java.util.List; +import java.util.Objects; + +/** + * The {@link com.codeheadsystems.velocity.spi.TumblingSupport} contract: a tumbling window is the + * aligned current bucket (FR-14). Events accumulate within a bucket and the value resets to zero + * the instant the clock crosses into the next aligned bucket — edge-approximate at the boundary is + * intended. + */ +public final class TumblingScenarios { + + private final CountStore store; + private final MutableClock clock; + + /** + * @param store a fresh count-capable, tumbling backend under test + * @param clock the backend's controllable clock + */ + public TumblingScenarios(final CountStore store, final MutableClock clock) { + this.store = Objects.requireNonNull(store, "store"); + this.clock = Objects.requireNonNull(clock, "clock"); + } + + /** + * Events accumulate inside a bucket, then the value resets when the clock crosses the boundary. + */ + public void tumblingAccumulatesThenResetsAtBoundary() { + final Instant bucketStart = WindowMath.tumblingBucket(clock.instant(), 60_000).start(); + clock.setInstant(bucketStart.plusSeconds(10)); // 10s into the current 1m bucket + + final Feature feature = countFeature(TUMBLING_1M); + store.applyCount( + Tck.apply(NS_A), + List.of(new CountIntent(feature, SUBJECT_A), new CountIntent(feature, SUBJECT_A))); + assertValue(currentCount(), 2); + + clock.setInstant(bucketStart.plusSeconds(65)); // 5s into the NEXT aligned bucket + assertValue(currentCount(), 0); + } + + private FeatureResult currentCount() { + return store + .queryCount( + Tck.query(NS_A), List.of(Tck.tuple(SUBJECT_A, Aggregation.count(), TUMBLING_1M))) + .get(0); + } +} diff --git a/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/package-info.java b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/package-info.java new file mode 100644 index 0000000..4cd0c23 --- /dev/null +++ b/velocity-testkit/src/main/java/com/codeheadsystems/velocity/testkit/tck/package-info.java @@ -0,0 +1,17 @@ +// SPDX-License-Identifier: BSD-3-Clause + +/** + * The mandatory {@code velocity-spi} conformance TCK (ADR 0004, NFR-18). + * + *

Each {@code *Scenarios} class is a backend-neutral contract suite: constructed with a backend + * (the specific mix-in type) plus a {@link com.codeheadsystems.velocity.testkit.MutableClock}, it + * exposes public methods that assert one facet of the frozen contract with AssertJ. A backend + * module drives them from its own JUnit test against a fresh store and a controllable clock, so the + * same contract — exact aggregates, read-your-write, window aging, seed/record equivalence, + * distinguishable failures instead of a silent {@code 0} — is verified identically for every + * backend. The suites deliberately have no JUnit dependency so a backend picks its own test runner. + */ +@NullMarked +package com.codeheadsystems.velocity.testkit.tck; + +import org.jspecify.annotations.NullMarked; diff --git a/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTck.java b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTck.java new file mode 100644 index 0000000..5b433af --- /dev/null +++ b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTck.java @@ -0,0 +1,204 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +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.SeedSupportScenarios; +import com.codeheadsystems.velocity.testkit.tck.SlidingScenarios; +import com.codeheadsystems.velocity.testkit.tck.SumStoreScenarios; +import com.codeheadsystems.velocity.testkit.tck.TumblingScenarios; +import java.time.Instant; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +/** + * Drives every conformance {@code *Scenarios} suite against the {@link InMemoryVelocityBackend}, so + * the reference backend passes its own TCK (ADR 0004). This is both the correctness proof for the + * reference implementation and the coverage driver for the testkit's own gate. + */ +class InMemoryVelocityBackendTck { + + /** A deterministic, mid-minute/mid-hour instant so time-independent scenarios are stable. */ + private static final Instant FIXED = Instant.parse("2026-07-18T12:00:30Z"); + + private InMemoryVelocityBackend backend; + private MutableClock clock; + + @BeforeEach + void freshBackend() { + clock = new MutableClock(FIXED); + backend = new InMemoryVelocityBackend(clock); + } + + @Nested + class Count { + @Test + void applyThenQueryReturnsExactCount() { + new CountStoreScenarios(backend, clock).applyThenQueryReturnsExactCount(); + } + + @Test + void applyResultReflectsWriteReadYourWrite() { + new CountStoreScenarios(backend, clock).applyResultReflectsWriteReadYourWrite(); + } + + @Test + void applyEmitsOneResultPerWindow() { + new CountStoreScenarios(backend, clock).applyEmitsOneResultPerWindow(); + } + + @Test + void valuesIsolatedByNamespace() { + new CountStoreScenarios(backend, clock).valuesIsolatedByNamespace(); + } + + @Test + void valuesIsolatedBySubject() { + new CountStoreScenarios(backend, clock).valuesIsolatedBySubject(); + } + + @Test + void concurrentApplyIsAtomic() throws Exception { + new CountStoreScenarios(backend, clock).concurrentApplyIsAtomic(); + } + } + + @Nested + class Sum { + @Test + void applyThenQueryReturnsExactSumCents() { + new SumStoreScenarios(backend, clock).applyThenQueryReturnsExactSumCents(); + } + + @Test + void applyResultReflectsRunningSum() { + new SumStoreScenarios(backend, clock).applyResultReflectsRunningSum(); + } + + @Test + void bigDecimalCentsPreservedWithoutOverflow() { + new SumStoreScenarios(backend, clock).bigDecimalCentsPreservedWithoutOverflow(); + } + + @Test + void negativeValuesForRefunds() { + new SumStoreScenarios(backend, clock).negativeValuesForRefunds(); + } + + @Test + void valuesIsolatedByNamespace() { + new SumStoreScenarios(backend, clock).valuesIsolatedByNamespace(); + } + } + + @Nested + class Distinct { + @Test + void applyThenQueryReturnsCardinality() { + new DistinctStoreScenarios(backend, clock).applyThenQueryReturnsCardinality(); + } + + @Test + void deDupesRepeatedMembers() { + new DistinctStoreScenarios(backend, clock).deDupesRepeatedMembers(); + } + + @Test + void applyResultReflectsCardinality() { + new DistinctStoreScenarios(backend, clock).applyResultReflectsCardinality(); + } + + @Test + void valuesIsolatedBySubject() { + new DistinctStoreScenarios(backend, clock).valuesIsolatedBySubject(); + } + } + + @Nested + class Sliding { + @Test + void eventsAgeOutAsClockAdvances() { + new SlidingScenarios(backend, clock).eventsAgeOutAsClockAdvances(); + } + + @Test + void slidingLeadingEdgeIsExclusive() { + new SlidingScenarios(backend, clock).slidingLeadingEdgeIsExclusive(); + } + } + + @Nested + class Tumbling { + @Test + void tumblingAccumulatesThenResetsAtBoundary() { + new TumblingScenarios(backend, clock).tumblingAccumulatesThenResetsAtBoundary(); + } + } + + @Nested + class Seed { + @Test + void seededBucketMergesIdenticallyWithRecorded() { + new SeedSupportScenarios<>(backend, clock).seededBucketMergesIdenticallyWithRecorded(); + } + + @Test + void hllDistinctSeedRejected() { + new SeedSupportScenarios<>(backend, clock).hllDistinctSeedRejected(); + } + + @Test + void seedWithUnsupportedWindowRejected() { + new SeedSupportScenarios<>(backend, clock).seedWithUnsupportedWindowRejected(); + } + } + + @Nested + class Capability { + @Test + void declaredAggregationsMatchImplementedMixins() { + new CapabilityConformanceScenarios(backend).declaredAggregationsMatchImplementedMixins(); + } + + @Test + void declaredWindowMarkersMatchWindowSpecs() { + new CapabilityConformanceScenarios(backend).declaredWindowMarkersMatchWindowSpecs(); + } + + @Test + void seedSupportFlagMatchesMixin() { + new CapabilityConformanceScenarios(backend).seedSupportFlagMatchesMixin(); + } + + @Test + void distinctHllSlidingIsFalse() { + new CapabilityConformanceScenarios(backend).distinctHllSlidingIsFalse(); + } + + @Test + void unsupportedWindowQueryIsDistinguishableFailure() { + new CapabilityConformanceScenarios(backend).unsupportedWindowQueryIsDistinguishableFailure(); + } + + @Test + void supportedEmptyWindowReturnsKnownZero() { + new CapabilityConformanceScenarios(backend).supportedEmptyWindowReturnsKnownZero(); + } + } + + @Nested + class Purge { + @Test + void purgeSubjectClearsThatSubjectOnly() { + new PurgeScenarios(backend).purgeSubjectClearsThatSubjectOnly(); + } + + @Test + void purgeNamespaceClearsEntireNamespace() { + new PurgeScenarios(backend).purgeNamespaceClearsEntireNamespace(); + } + } +} diff --git a/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTest.java b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTest.java new file mode 100644 index 0000000..98b6437 --- /dev/null +++ b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/InMemoryVelocityBackendTest.java @@ -0,0 +1,193 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +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.BucketValue; +import com.codeheadsystems.velocity.spi.model.DistinctMember; +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.Intent.CountIntent; +import com.codeheadsystems.velocity.spi.model.Namespace; +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.SeedAggregate; +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 java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** Focused tests for backend logic beyond what the shared TCK scenarios exercise. */ +class InMemoryVelocityBackendTest { + + private static final Instant FIXED = Instant.parse("2026-07-18T12:00:30Z"); + private static final Namespace NS = new Namespace("ns"); + private static final Subject SUBJECT = new Subject("card", "s1"); + private static final Window SLIDING_1M = new Window(Duration.ofMinutes(1), WindowType.SLIDING); + private static final Window TUMBLING_1M = new Window(Duration.ofMinutes(1), WindowType.TUMBLING); + private static final Window UNSUPPORTED = new Window(Duration.ofDays(7), WindowType.SLIDING); + + private MutableClock clock; + private InMemoryVelocityBackend backend; + + @BeforeEach + void setUp() { + clock = new MutableClock(FIXED); + backend = new InMemoryVelocityBackend(clock); + } + + @Test + void noArgConstructorUsesSystemUtc() { + assertThat(new InMemoryVelocityBackend().clock().getZone()).isEqualTo(ZoneOffset.UTC); + } + + @Test + void capabilitiesDeclareTheReferenceProfile() { + final BackendCapabilities caps = backend.capabilities(); + assertThat(caps.supportsAggregation(AggregationType.COUNT)).isTrue(); + assertThat(caps.supportsAggregation(AggregationType.SUM)).isTrue(); + assertThat(caps.supportsAggregation(AggregationType.DISTINCT)).isTrue(); + assertThat(caps.windows()).hasSize(4); + assertThat(caps.distinctHllSliding()).isFalse(); + assertThat(caps.readYourWriteLevel()).isEqualTo(ReadYourWriteLevel.ATOMIC); + assertThat(caps.idempotencySupported()).isFalse(); + assertThat(caps.seedSupported()).isTrue(); + assertThat(caps.maxAtomicFanOut()).isEqualTo(Integer.MAX_VALUE); + assertThat(caps.maxRetention()).isEqualTo(Duration.ofDays(30)); + assertThat(caps.distinctExactCardinalityClamp()).isEqualTo(Long.MAX_VALUE); + } + + @Test + void applyWithMixedWindowsFailsOnlyTheUnsupportedWindowButStillRecords() { + final Feature feature = + new Feature("mixed", Aggregation.count(), List.of(SLIDING_1M, UNSUPPORTED)); + final ApplyResult result = + backend.applyCount(new ApplyContext(NS, null), List.of(new CountIntent(feature, SUBJECT))); + + assertThat(result.perFeature()).hasSize(2); + final FeatureResult supported = + result.perFeature().stream() + .filter(pf -> pf.status() == ApplyStatus.APPLIED) + .findFirst() + .orElseThrow() + .result(); + assertThat(((FeatureResult.Success) supported).value().value()) + .isEqualByComparingTo(BigDecimal.ONE); + + final FeatureResult failed = + result.perFeature().stream() + .filter(pf -> pf.status() == ApplyStatus.FAILED) + .findFirst() + .orElseThrow() + .result(); + assertThat(((FeatureResult.Failure) failed).code()).isEqualTo(FailureCode.UNSUPPORTED_WINDOW); + + // The supported window still recorded the underlying event. + assertThat(countOver(SLIDING_1M)).isEqualByComparingTo(BigDecimal.ONE); + } + + @Test + void applyWithOnlyUnsupportedWindowsRecordsNothing() { + final Feature feature = new Feature("only-bad", Aggregation.count(), List.of(UNSUPPORTED)); + final ApplyResult result = + backend.applyCount(new ApplyContext(NS, null), List.of(new CountIntent(feature, SUBJECT))); + + assertThat(result.perFeature()).hasSize(1); + assertThat(result.perFeature().get(0).status()).isEqualTo(ApplyStatus.FAILED); + // Nothing was recorded, so a supported window still reads a known zero. + assertThat(countOver(SLIDING_1M)).isEqualByComparingTo(BigDecimal.ZERO); + } + + @Test + void queryUnsupportedWindowIsFailureNotSilentZero() { + final FeatureResult result = + backend + .queryCount( + new QueryContext(NS, null), + List.of(new QueryTuple(SUBJECT, Aggregation.count(), UNSUPPORTED))) + .get(0); + assertThat(result.isSuccess()).isFalse(); + assertThat(((FeatureResult.Failure) result).code()).isEqualTo(FailureCode.UNSUPPORTED_WINDOW); + } + + @Test + void seedExactDistinctIsQueryable() { + final Feature feature = + new Feature("distinct", Aggregation.distinct("ip"), List.of(TUMBLING_1M)); + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + backend.seed( + NS, + SUBJECT, + feature, + List.of( + new BucketValue( + bucket, + new SeedAggregate.ExactDistinct(List.of(member("a"), member("b"), member("a")))))); + + final FeatureResult result = + backend + .queryDistinct( + new QueryContext(NS, null), + List.of(new QueryTuple(SUBJECT, Aggregation.distinct("ip"), TUMBLING_1M))) + .get(0); + assertThat(((FeatureResult.Success) result).value().value()) + .isEqualByComparingTo(BigDecimal.valueOf(2)); + } + + @Test + void seedSumBucketIsQueryable() { + final Feature feature = new Feature("sum", Aggregation.sum(), List.of(TUMBLING_1M)); + final WindowBounds bucket = WindowMath.tumblingBucket(clock.instant(), 60_000); + backend.seed( + NS, + SUBJECT, + feature, + List.of(new BucketValue(bucket, new SeedAggregate.SumValue(BigDecimal.valueOf(750))))); + + final FeatureResult result = + backend + .querySum( + new QueryContext(NS, null), + List.of(new QueryTuple(SUBJECT, Aggregation.sum(), TUMBLING_1M))) + .get(0); + assertThat(((FeatureResult.Success) result).value().value()) + .isEqualByComparingTo(BigDecimal.valueOf(750)); + } + + @Test + void purgeMissingScopeIsNoop() { + assertThatCode(() -> backend.purge(NS, SUBJECT)).doesNotThrowAnyException(); + assertThatCode(() -> backend.purge(NS, null)).doesNotThrowAnyException(); + } + + private BigDecimal countOver(final Window window) { + final FeatureResult result = + backend + .queryCount( + new QueryContext(NS, null), + List.of(new QueryTuple(SUBJECT, Aggregation.count(), window))) + .get(0); + return ((FeatureResult.Success) result).value().value(); + } + + private static DistinctMember member(final String token) { + return new DistinctMember(token.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/MutableClockTest.java b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/MutableClockTest.java new file mode 100644 index 0000000..bd395c0 --- /dev/null +++ b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/MutableClockTest.java @@ -0,0 +1,54 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import org.junit.jupiter.api.Test; + +/** Unit tests for the test-controllable {@link MutableClock}. */ +class MutableClockTest { + + private static final Instant START = Instant.parse("2026-07-18T12:00:00Z"); + + @Test + void reportsTheStartInstantAndMillis() { + final MutableClock clock = new MutableClock(START); + assertThat(clock.instant()).isEqualTo(START); + assertThat(clock.millis()).isEqualTo(START.toEpochMilli()); + assertThat(clock.getZone()).isEqualTo(ZoneOffset.UTC); + } + + @Test + void advanceMovesForward() { + final MutableClock clock = new MutableClock(START); + clock.advance(Duration.ofSeconds(90)); + assertThat(clock.instant()).isEqualTo(START.plusSeconds(90)); + } + + @Test + void setInstantReplacesTime() { + final MutableClock clock = new MutableClock(START); + clock.setInstant(START.minusSeconds(10)); + assertThat(clock.instant()).isEqualTo(START.minusSeconds(10)); + } + + @Test + void withZoneKeepsInstantButChangesZone() { + final MutableClock clock = new MutableClock(START); + final Clock tokyo = clock.withZone(ZoneId.of("Asia/Tokyo")); + assertThat(tokyo.instant()).isEqualTo(START); + assertThat(tokyo.getZone()).isEqualTo(ZoneId.of("Asia/Tokyo")); + } + + @Test + void atNowIsCloseToSystemTime() { + final long before = System.currentTimeMillis(); + final MutableClock clock = MutableClock.atNow(); + assertThat(clock.millis()).isGreaterThanOrEqualTo(before); + } +} diff --git a/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/WindowMathTest.java b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/WindowMathTest.java new file mode 100644 index 0000000..a6d1824 --- /dev/null +++ b/velocity-testkit/src/test/java/com/codeheadsystems/velocity/testkit/WindowMathTest.java @@ -0,0 +1,71 @@ +// SPDX-License-Identifier: BSD-3-Clause +package com.codeheadsystems.velocity.testkit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.codeheadsystems.velocity.spi.model.Window; +import com.codeheadsystems.velocity.spi.model.WindowBounds; +import com.codeheadsystems.velocity.spi.model.WindowType; +import java.time.Duration; +import java.time.Instant; +import org.junit.jupiter.api.Test; + +/** Unit tests for the window-bounds arithmetic the backend relies on. */ +class WindowMathTest { + + private static final Window SLIDING_5S = new Window(Duration.ofSeconds(5), WindowType.SLIDING); + private static final Window TUMBLING_1M = new Window(Duration.ofMinutes(1), WindowType.TUMBLING); + + @Test + void slidingBoundsAreNowMinusDurationToNow() { + final Instant now = Instant.ofEpochMilli(100_000); + final WindowBounds bounds = WindowMath.boundsAt(SLIDING_5S, now); + assertThat(bounds.start()).isEqualTo(Instant.ofEpochMilli(95_000)); + assertThat(bounds.end()).isEqualTo(now); + } + + @Test + void tumblingBoundsAreTheAlignedBucket() { + final Instant now = Instant.ofEpochMilli(125_000); + final WindowBounds bounds = WindowMath.boundsAt(TUMBLING_1M, now); + assertThat(bounds.start()).isEqualTo(Instant.ofEpochMilli(120_000)); + assertThat(bounds.end()).isEqualTo(Instant.ofEpochMilli(180_000)); + } + + @Test + void tumblingBucketAlignsToFloor() { + final WindowBounds bounds = WindowMath.tumblingBucket(Instant.ofEpochMilli(179_999), 60_000); + assertThat(bounds.start()).isEqualTo(Instant.ofEpochMilli(120_000)); + assertThat(bounds.end()).isEqualTo(Instant.ofEpochMilli(180_000)); + } + + @Test + void tumblingBucketRejectsNonPositiveDuration() { + assertThatThrownBy(() -> WindowMath.tumblingBucket(Instant.EPOCH, 0)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void slidingContainsIsStartExclusiveEndInclusive() { + final Instant now = Instant.ofEpochMilli(100_000); + final WindowBounds bounds = WindowMath.boundsAt(SLIDING_5S, now); + assertThat(WindowMath.contains(SLIDING_5S, bounds, bounds.start())) + .isFalse(); // exclusive start + assertThat(WindowMath.contains(SLIDING_5S, bounds, Instant.ofEpochMilli(95_001))).isTrue(); + assertThat(WindowMath.contains(SLIDING_5S, bounds, bounds.end())).isTrue(); // inclusive end + assertThat(WindowMath.contains(SLIDING_5S, bounds, Instant.ofEpochMilli(100_001))).isFalse(); + assertThat(WindowMath.contains(SLIDING_5S, bounds, Instant.ofEpochMilli(94_999))).isFalse(); + } + + @Test + void tumblingContainsIsStartInclusiveEndExclusive() { + final Instant now = Instant.ofEpochMilli(125_000); + final WindowBounds bounds = WindowMath.boundsAt(TUMBLING_1M, now); + assertThat(WindowMath.contains(TUMBLING_1M, bounds, bounds.start())) + .isTrue(); // inclusive start + assertThat(WindowMath.contains(TUMBLING_1M, bounds, bounds.end())).isFalse(); // exclusive end + assertThat(WindowMath.contains(TUMBLING_1M, bounds, Instant.ofEpochMilli(179_999))).isTrue(); + assertThat(WindowMath.contains(TUMBLING_1M, bounds, Instant.ofEpochMilli(119_999))).isFalse(); + } +}