diff --git a/slo-workload/AGENTS.md b/slo-workload/AGENTS.md index cf82adb..45aadf0 100644 --- a/slo-workload/AGENTS.md +++ b/slo-workload/AGENTS.md @@ -19,6 +19,9 @@ Concretely: - `query/`, `jdbc/`, `spring-data-jdbc/`, `spring-data-jpa/` — one workload per client under test. Each module is ~thin: a `Main` that wires a `KvClient` implementation into the shared `Launcher`. +- `query-rmw/` — separate native Query transactional scenario using `core/rmw`. + It reuses Config/Metrics and the namespace helper, not the basic KV schema or + checker. JDBC and JPA do not implement this scenario. - `docker/Dockerfile.sdk` — single shared Dockerfile used by the **SDK** CI (`ydb-java-sdk`). Picks the workload module via build args. - `/Dockerfile` — per-workload Dockerfile used by the **JDBC driver** @@ -119,6 +122,18 @@ availability still computes correctly. `ydb-jdbc-driver/.github/workflows/slo.yaml`). This repo does **not** run SLO CI by itself — there's no YDB cluster here. +The separate Query RMW scenario is an exception to the KV adapter shape. Its +runner retains each logical ID/counter/delta before admission, admits bounded +concurrent rounds and settles every admitted transaction before independent +validation and cleanup. Expected totals and versions come from that retained +image, never the observed ledger. Retry the complete SDK transaction with the +same ID; the transactional ledger justifies SDK idempotent retry. Its scoped +QueryClient observation counts outer SDK session acquisitions, including failed +acquisition and retries after commit, rather than callback entries. Use the +existing source Dockerfile with the RMW module/jar build arguments; see +`query-rmw/README.md`. Do not imply support in other adapters or change the +basic KV checker to accommodate this scenario. + ## Local sanity checks ```bash diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwRunner.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwRunner.java new file mode 100644 index 0000000..3e3a6b5 --- /dev/null +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwRunner.java @@ -0,0 +1,107 @@ +package tech.ydb.slo.core.rmw; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.LongSupplier; + +/** Bounded concurrent rounds: a fast writer cannot starve a retrying peer. */ +public final class RmwRunner { + public interface Client extends AutoCloseable { + void create(); + CompletableFuture apply(RmwState.Operation operation); + RmwState.Snapshot read(); + void drop(); + @Override + void close(); + } + + private final Client client; + private final int writers; + private final LongSupplier clock; + private final AtomicBoolean stop = new AtomicBoolean(); + private final AtomicReference failure = new AtomicReference<>(); + + public RmwRunner(Client client, int writers) { + this(client, writers, System::nanoTime); + } + + RmwRunner(Client client, int writers, LongSupplier clock) { + if (writers < 2) { + throw new IllegalArgumentException("At least two contending writers are required"); + } + this.client = client; + this.writers = writers; + this.clock = clock; + } + + public void requestStop() { + stop.set(true); + } + + public Throwable failure() { + return failure.get(); + } + + public int run(Duration duration) { + Map expected = new LinkedHashMap<>(); + try { + client.create(); + long started = clock.getAsLong(); + long round = 0; + while (!stop.get() && (duration.isZero() + || clock.getAsLong() - started < duration.toNanos())) { + List> pending = new ArrayList<>(); + for (int writer = 0; writer < writers && !stop.get(); writer++) { + RmwState.Operation operation = new RmwState.Operation( + writer + ":" + round, round % 2, writer + 1L); + // Retain the independent expected image BEFORE invoking the adapter. + expected.put(operation.id(), operation); + try { + pending.add(client.apply(operation).handle((ignored, error) -> { + if (error != null) { + fail(error); + } + return null; + })); + } catch (Throwable error) { + fail(error); + } + } + // Planned stop and final errors stop admissions, not admitted SDK operations. + CompletableFuture.allOf(pending.toArray(new CompletableFuture[0])).join(); + round++; + } + // Independent read after every admitted operation has settled, even after failure. + RmwState.validate(expected, client.read()); + } catch (Throwable error) { + fail(error); + } finally { + try { + client.drop(); + } catch (Throwable error) { + fail(error); + } + try { + client.close(); + } catch (Throwable error) { + fail(error); + } + } + return failure.get() == null ? 0 : 1; + } + + private void fail(Throwable error) { + while (error instanceof CompletionException && error.getCause() != null) { + error = error.getCause(); + } + failure.compareAndSet(null, error); + stop.set(true); + } +} diff --git a/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwState.java b/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwState.java new file mode 100644 index 0000000..b71087b --- /dev/null +++ b/slo-workload/core/src/main/java/tech/ydb/slo/core/rmw/RmwState.java @@ -0,0 +1,119 @@ +package tech.ydb.slo.core.rmw; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** The expected image is retained by the admission loop, never reconstructed from storage. */ +public final class RmwState { + private RmwState() { } + + public static final class Operation { + private final String id; + private final long counter; + private final long delta; + + public Operation(String id, long counter, long delta) { + this.id = id; + this.counter = counter; + this.delta = delta; + } + + public String id() { return id; } + public long counter() { return counter; } + public long delta() { return delta; } + + @Override + public boolean equals(Object other) { + if (!(other instanceof Operation)) { + return false; + } + Operation that = (Operation) other; + return id.equals(that.id) && counter == that.counter && delta == that.delta; + } + + @Override + public int hashCode() { return Objects.hash(id, counter, delta); } + } + + public static final class Counter { + private final long id; + private final long value; + private final long version; + + public Counter(long id, long value, long version) { + this.id = id; + this.value = value; + this.version = version; + } + + public long id() { return id; } + public long value() { return value; } + public long version() { return version; } + + @Override + public boolean equals(Object other) { + if (!(other instanceof Counter)) { + return false; + } + Counter that = (Counter) other; + return id == that.id && value == that.value && version == that.version; + } + + @Override + public int hashCode() { return Objects.hash(id, value, version); } + + @Override + public String toString() { return "Counter(" + id + ", " + value + ", " + version + ")"; } + } + + public static final class Snapshot { + private final List counters; + private final List operations; + + public Snapshot(List counters, List operations) { + this.counters = List.copyOf(counters); + this.operations = List.copyOf(operations); + } + + public List counters() { return counters; } + public List operations() { return operations; } + } + + public static void validate(Map expected, Snapshot actual) { + Map ledger = new HashMap<>(); + for (Operation operation : actual.operations()) { + if (ledger.put(operation.id(), operation) != null) { + throw new IllegalStateException("Duplicate logical ID: " + operation.id()); + } + } + if (!expected.equals(ledger)) { + throw new IllegalStateException("Ledger does not match admitted IDs and payloads"); + } + + long[] totals = new long[2]; + long[] versions = new long[2]; + for (Operation operation : expected.values()) { + int counter = Math.toIntExact(operation.counter()); + totals[counter] = Math.addExact(totals[counter], operation.delta()); + versions[counter] = Math.incrementExact(versions[counter]); + } + Map counters = new HashMap<>(); + for (Counter counter : actual.counters()) { + if (counters.put(counter.id(), counter) != null) { + throw new IllegalStateException("Duplicate counter: " + counter.id()); + } + } + if (counters.size() != 2) { + throw new IllegalStateException("Expected exactly two counters"); + } + for (int id = 0; id < 2; id++) { + Counter wanted = new Counter(id, totals[id], versions[id]); + if (!wanted.equals(counters.get((long) id))) { + throw new IllegalStateException("Counter differs from admitted operations: expected " + + wanted + ", got " + counters.get((long) id)); + } + } + } +} diff --git a/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwRunnerTest.java b/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwRunnerTest.java new file mode 100644 index 0000000..77848cb --- /dev/null +++ b/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwRunnerTest.java @@ -0,0 +1,197 @@ +package tech.ydb.slo.core.rmw; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +class RmwRunnerTest { + @Test + void plannedStopSettlesInflightAndValidatesBeforeDropping() throws Exception { + ControlledClient client = new ControlledClient(); + RmwRunner runner = new RmwRunner(client, 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> runner.run(Duration.ZERO)); + client.rounds.get(0).get(10, TimeUnit.SECONDS); + runner.requestStop(); + assertFalse(result.isDone()); + assertFalse(client.read); + client.succeed(0); + assertFalse(result.isDone()); + client.succeed(1); + assertEquals(0, result.get(10, TimeUnit.SECONDS)); + assertNull(runner.failure()); + assertTrue(client.read); + assertTrue(client.dropped); + assertTrue(client.closed); + assertEquals(List.of(new RmwState.Operation("0:0", 0, 1), + new RmwState.Operation("1:0", 0, 2)), client.admitted); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + void boundedRoundsGiveEveryWriterOneAdmissionBeforeAnyNextRound() throws Exception { + ControlledClient client = new ControlledClient(); + RmwRunner runner = new RmwRunner(client, 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> runner.run(Duration.ZERO)); + client.rounds.get(0).get(10, TimeUnit.SECONDS); + client.succeed(0); + assertEquals(2, client.admitted.size()); + assertFalse(client.rounds.get(1).isDone()); + client.succeed(1); + client.rounds.get(1).get(10, TimeUnit.SECONDS); + runner.requestStop(); + client.succeed(3); + client.succeed(2); + assertEquals(0, result.get(10, TimeUnit.SECONDS)); + assertEquals(List.of("0:0", "1:0", "0:1", "1:1"), + client.admitted.stream().map(RmwState.Operation::id).toList()); + assertEquals(List.of(0L, 0L, 1L, 1L), + client.admitted.stream().map(RmwState.Operation::counter).toList()); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + void firstFinalCauseSurvivesPeerSettlementValidationAndCleanupFailures() throws Exception { + ControlledClient client = new ControlledClient() { + @Override + public void drop() { + super.drop(); + throw new IllegalStateException("cleanup"); + } + }; + RuntimeException nativeCause = new RuntimeException("native final cause"); + RmwRunner runner = new RmwRunner(client, 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> runner.run(Duration.ZERO)); + client.rounds.get(0).get(10, TimeUnit.SECONDS); + client.pending.get(0).completeExceptionally(nativeCause); + assertFalse(result.isDone()); + client.succeed(1); + assertEquals(1, result.get(10, TimeUnit.SECONDS)); + assertSame(nativeCause, runner.failure()); + assertTrue(client.read); + assertTrue(client.dropped); + assertTrue(client.closed); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + void ordinaryDurationCompletesHealthyRound() { + AtomicLong time = new AtomicLong(); + ControlledClient client = new ControlledClient() { + @Override + public CompletableFuture apply(RmwState.Operation operation) { + applied.add(operation); + return CompletableFuture.completedFuture(null); + } + }; + RmwRunner runner = new RmwRunner(client, 2, () -> time.getAndAdd(1_000_000_000L)); + assertEquals(0, runner.run(Duration.ofSeconds(2))); + assertEquals(2, client.applied.size()); + assertTrue(client.read); + assertTrue(client.dropped); + } + + @Test + void coherentLossFailsAgainstPreAdmissionImage() { + AtomicLong time = new AtomicLong(); + ControlledClient client = new ControlledClient() { + @Override + public CompletableFuture apply(RmwState.Operation operation) { + // Report success but lose BOTH ledger and matching counter update. + return CompletableFuture.completedFuture(null); + } + }; + RmwRunner runner = new RmwRunner(client, 2, () -> time.getAndAdd(1_000_000_000L)); + assertEquals(1, runner.run(Duration.ofSeconds(2))); + assertInstanceOf(IllegalStateException.class, runner.failure()); + assertTrue(client.dropped); + } + + @Test + void setupFailureStillDropsAndClosesWithOriginalCause() { + RuntimeException original = new RuntimeException("create"); + ControlledClient client = new ControlledClient() { + @Override + public void create() { + throw original; + } + }; + RmwRunner runner = new RmwRunner(client, 2); + assertEquals(1, runner.run(Duration.ofSeconds(1))); + assertSame(original, runner.failure()); + assertTrue(client.dropped); + assertTrue(client.closed); + } + + private static class ControlledClient implements RmwRunner.Client { + final List admitted = new ArrayList<>(); + final List applied = new ArrayList<>(); + final List> pending = new ArrayList<>(); + final List> rounds = List.of(new CompletableFuture<>(), new CompletableFuture<>()); + boolean read; + boolean dropped; + boolean closed; + + public void create() { } + + public CompletableFuture apply(RmwState.Operation operation) { + admitted.add(operation); + CompletableFuture future = new CompletableFuture<>(); + pending.add(future); + if (pending.size() % 2 == 0) { + rounds.get(pending.size() / 2 - 1).complete(null); + } + return future; + } + + void succeed(int index) { + applied.add(admitted.get(index)); + pending.get(index).complete(null); + } + + public RmwState.Snapshot read() { + read = true; + long[] values = new long[2]; + long[] versions = new long[2]; + for (RmwState.Operation operation : applied) { + values[(int) operation.counter()] += operation.delta(); + versions[(int) operation.counter()]++; + } + return new RmwState.Snapshot(List.of(new RmwState.Counter(0, values[0], versions[0]), + new RmwState.Counter(1, values[1], versions[1])), List.copyOf(applied)); + } + + public void drop() { + assertTrue(pending.stream().allMatch(CompletableFuture::isDone)); + dropped = true; + } + + public void close() { + closed = true; + } + } +} diff --git a/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwStateTest.java b/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwStateTest.java new file mode 100644 index 0000000..a9623b2 --- /dev/null +++ b/slo-workload/core/src/test/java/tech/ydb/slo/core/rmw/RmwStateTest.java @@ -0,0 +1,64 @@ +package tech.ydb.slo.core.rmw; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +class RmwStateTest { + private final RmwState.Operation first = new RmwState.Operation("0:0", 0, 3); + private final RmwState.Operation second = new RmwState.Operation("1:0", 1, 5); + private final Map expected = Map.of(first.id(), first, second.id(), second); + private final List correct = List.of( + new RmwState.Counter(0, 3, 1), new RmwState.Counter(1, 5, 1)); + + @Test + void exactRowsPassIncludingEmptyRun() { + assertDoesNotThrow(() -> RmwState.validate(expected, + new RmwState.Snapshot(correct, List.of(second, first)))); + assertDoesNotThrow(() -> RmwState.validate(Map.of(), new RmwState.Snapshot(List.of( + new RmwState.Counter(0, 0, 0), new RmwState.Counter(1, 0, 0)), List.of()))); + } + + @Test + void sumPreservingCounterCorruptionFails() { + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(List.of(new RmwState.Counter(0, 2, 1), + new RmwState.Counter(1, 6, 1)), List.of(first, second)))); + } + + @Test + void coherentlyMissingLedgerAndCounterFails() { + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(List.of(new RmwState.Counter(0, 0, 0), correct.get(1)), List.of(second)))); + } + + @Test + void wrongVersionFailsIndependentlyOfValue() { + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(List.of(new RmwState.Counter(0, 3, 0), correct.get(1)), + List.of(first, second)))); + } + + @Test + void wrongPayloadOrCounterFailsEvenWithSameIds() { + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(correct, List.of(new RmwState.Operation(first.id(), 0, 4), second)))); + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(correct, List.of(new RmwState.Operation(first.id(), 1, 3), second)))); + } + + @Test + void duplicateExtraAndMissingRowsFail() { + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(correct, List.of(first, first, second)))); + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(correct, List.of(first, second, new RmwState.Operation("9:9", 0, 1))))); + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(List.of(correct.get(0)), List.of(first, second)))); + assertThrows(IllegalStateException.class, () -> RmwState.validate(expected, + new RmwState.Snapshot(List.of(correct.get(0), correct.get(0)), List.of(first, second)))); + } +} diff --git a/slo-workload/pom.xml b/slo-workload/pom.xml index a43b80f..7f790d6 100644 --- a/slo-workload/pom.xml +++ b/slo-workload/pom.xml @@ -121,6 +121,7 @@ core query + query-rmw jdbc spring-data-jdbc spring-data-jpa diff --git a/slo-workload/query-rmw/README.md b/slo-workload/query-rmw/README.md new file mode 100644 index 0000000..baec1e5 --- /dev/null +++ b/slo-workload/query-rmw/README.md @@ -0,0 +1,85 @@ +# Native Query transactional RMW + +This is a separate native Query workload, not an adapter for JDBC, Spring Data +JDBC or JPA. It reuses the existing `Config`, `Metrics`, process namespace helper +and source-built SDK Dockerfile. The basic KV workloads are independent. + +```sh +mvn -pl slo-workload/query-rmw -am package +java -jar slo-workload/query-rmw/target/ydb-slo-query-rmw-workload.jar \ + --duration 60 --writers 4 +``` + +Connection, authentication, workload name/ref and OTLP configuration use the +same environment variables as the basic workloads. `WORKLOAD_DURATION` supplies +the default duration. `--duration 0` means unlimited. At least two writers are +required. The default is four. + +Use `slo-workload/docker/Dockerfile.sdk` with +`WORKLOAD_MODULE=slo-workload/query-rmw`, +`WORKLOAD_JAR=ydb-slo-query-rmw-workload.jar`, and `COPY_LIBS=true` in the existing +two-repository source-build context. No shell wrapper or alternate SDK build +helper is required. The consuming workflow must explicitly select this module; +selecting the basic Query workload does not run RMW. + +## Lifecycle and correctness + +One invocation owns two tables named from the existing process-private UUID +namespace: `_counters` and `_operations`. It acknowledges +creation, seeds counters 0 and 1 at value/version 0, runs writers, waits for all +admitted operations, performs a fresh snapshot read, and acknowledges dropping +both tables. Cleanup is attempted after setup or run failure as well. A failed +drop fails an otherwise healthy invocation. + +Each writer gets one admission in a bounded concurrent round. All writers in a +round target the same counter; consecutive rounds alternate counters. The next +round waits for all current operations, including retries, so a fast writer +cannot repeatedly overtake a contending peer. IDs are stable `writer:sequence` +strings. A writer's delta is `writer + 1`. Before each admission the runner retains +that ID, counter and delta in its own expected map. + +`SessionRetryContext.supplyResult` encloses the entire interactive +`QueryTransaction` created with `SERIALIZABLE_RW`: read operation ID; if present, +verify its payload and commit without incrementing; otherwise read the counter, +update its value and version, insert the operation record, and commit. A retry +creates a new transaction but retains the same immutable operation. The SDK's +`idempotent(true)` declaration permits its existing ambiguous-result retries; +the transactional ledger makes that declaration valid. Retry budgets, backoff, +session acquisition timeout and per-query timeouts use the basic Query defaults. +No application retry loop or policy tuning is added. + +The final check requires the exact admitted ID set and payloads, exactly two +counters, and each counter's exact total and version computed from the independent +expected map. It does not derive expected IDs or totals from the observed ledger. +This catches both sum-preserving corruption and a coherently missing ledger row +plus its counter update. Stream/status failures retain their native SDK cause. +The first final failure survives later validation and cleanup failures and exits +1. Recovered retries, normal duration and a requested runner stop exit 0 only +after in-flight settlement, validation and cleanup. Stop never cancels an +admitted SDK transaction. + +## Attempt metrics + +The legacy `sdk_retry_attempts_total` family counts technical attempts including +the first, not callback invocations or individual RPCs. `AttemptClient` is a +per-invocation, non-owning observation view of `QueryClient`. It increments on +every `SessionRetryContext` session acquisition, including acquisition failures +before the callback. The callback includes reads, writes and commit, so a failed +commit retried by the SDK starts another counted outer attempt. Scheduler, +tracer, pool and session behavior delegate unchanged. Counts cannot leak between +concurrent logical operations. One complete RMW is one logical write metric; +its component statements are not separate logical operations. Setup, final read +and cleanup also use the existing read/write metric families. + +## Focused tests + +```sh +JAVA_HOME=/path/to/jdk17 mvn -pl slo-workload/query-rmw -am package +``` + +Core tests exercise independent validation, duration, stop/settlement, fairness +and first-cause retention. Query tests retain the real SDK retry context, Result +composition and QueryReader, replacing only the Query transport boundary. They +exercise acquisition retry, commit retry, a committed-but-UNDETERMINED response, +deduplication and native failures. They are not a real database contention or +chaos receipt; that requires the consuming SDK workflow against a real YDB. diff --git a/slo-workload/query-rmw/pom.xml b/slo-workload/query-rmw/pom.xml new file mode 100644 index 0000000..0aaaeb6 --- /dev/null +++ b/slo-workload/query-rmw/pom.xml @@ -0,0 +1,60 @@ + + + 4.0.0 + + tech.ydb.examples + slo-workload + 1.1.0-SNAPSHOT + ../pom.xml + + slo-workload-query-rmw + YDB SLO native Query transactional RMW workload + + + tech.ydb.examples + slo-workload-core + + + tech.ydb + ydb-sdk-query + + + org.apache.logging.log4j + log4j-slf4j2-impl + + + org.junit.jupiter + junit-jupiter + 5.10.1 + test + + + + ydb-slo-query-rmw-workload + + + org.apache.maven.plugins + maven-compiler-plugin + + + org.apache.maven.plugins + maven-dependency-plugin + + + org.apache.maven.plugins + maven-jar-plugin + + + + true + libs/ + tech.ydb.slo.Main + + + + + + + diff --git a/slo-workload/query-rmw/src/main/java/tech/ydb/slo/Main.java b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/Main.java new file mode 100644 index 0000000..8609911 --- /dev/null +++ b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/Main.java @@ -0,0 +1,67 @@ +package tech.ydb.slo; + +import java.time.Duration; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import tech.ydb.slo.core.Config; +import tech.ydb.slo.core.Launcher; +import tech.ydb.slo.core.Metrics; +import tech.ydb.slo.core.rmw.RmwRunner; +import tech.ydb.slo.queryrmw.QueryRmwClient; + +public final class Main { + private static final Logger logger = LoggerFactory.getLogger(Main.class); + + private static final class Options { + @Parameter(names = {"--duration", "--time"}, description = "Seconds; 0 means unlimited") + private int duration = -1; + + @Parameter(names = "--writers", description = "Concurrent writers per bounded round (at least 2)") + private int writers = 4; + } + + private Main() { } + + public static void main(String[] args) { + System.exit(run(args)); + } + + static int run(String[] args) { + Options options = new Options(); + Config config; + try { + JCommander.newBuilder().programName("query-rmw").acceptUnknownOptions(false) + .addObject(options).build().parse(args); + config = Config.fromEnv("java-slo-query-rmw"); + if (options.duration < 0) { + options.duration = config.durationSeconds(); + } + if (options.duration < 0 || options.writers < 2) { + throw new IllegalArgumentException("Require duration >= 0 and writers >= 2"); + } + } catch (RuntimeException error) { + logger.error("Invalid RMW configuration", error); + return 2; + } + + try (Metrics metrics = Metrics.create(config)) { + String namespace = Launcher.tablePathFor(config); + logger.info("Starting Query RMW: namespace={}, writers={}, duration={}s", + namespace, options.writers, options.duration); + RmwRunner runner = new RmwRunner(new QueryRmwClient(config, namespace, metrics), options.writers); + int code = runner.run(Duration.ofSeconds(options.duration)); + if (runner.failure() != null) { + logger.error("Query RMW failed", runner.failure()); + } + metrics.flush(); + return code; + } catch (Throwable error) { + logger.error("Query RMW failed", error); + return 1; + } + } +} diff --git a/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/AttemptClient.java b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/AttemptClient.java new file mode 100644 index 0000000..90af603 --- /dev/null +++ b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/AttemptClient.java @@ -0,0 +1,52 @@ +package tech.ydb.slo.queryrmw; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +import tech.ydb.core.Result; +import tech.ydb.core.tracing.Tracer; +import tech.ydb.query.QueryClient; +import tech.ydb.query.QuerySession; +import tech.ydb.table.SessionPoolStats; + +/** Scoped to ONE retry invocation, including failed session acquisition and commit retries. */ +final class AttemptClient implements QueryClient { + private final QueryClient delegate; + private final AtomicInteger attempts = new AtomicInteger(); + + AttemptClient(QueryClient delegate) { + this.delegate = delegate; + } + + int attempts() { + return attempts.get(); + } + + @Override + public CompletableFuture> createSession(Duration timeout) { + attempts.incrementAndGet(); + return delegate.createSession(timeout); + } + + @Override + public ScheduledExecutorService getScheduler() { + return delegate.getScheduler(); + } + + @Override + public Tracer getTracer() { + return delegate.getTracer(); + } + + @Override + public SessionPoolStats getSessionPoolStats() { + return delegate.getSessionPoolStats(); + } + + @Override + public void close() { + // This observation view does not own the shared client. + } +} diff --git a/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/QueryRmwClient.java b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/QueryRmwClient.java new file mode 100644 index 0000000..56cfc69 --- /dev/null +++ b/slo-workload/query-rmw/src/main/java/tech/ydb/slo/queryrmw/QueryRmwClient.java @@ -0,0 +1,224 @@ +package tech.ydb.slo.queryrmw; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.Function; + +import tech.ydb.auth.AuthProvider; +import tech.ydb.auth.NopAuthProvider; +import tech.ydb.auth.TokenAuthProvider; +import tech.ydb.common.transaction.TxMode; +import tech.ydb.core.Result; +import tech.ydb.core.grpc.GrpcTransport; +import tech.ydb.query.QueryClient; +import tech.ydb.query.QuerySession; +import tech.ydb.query.QueryTransaction; +import tech.ydb.query.result.QueryInfo; +import tech.ydb.query.settings.ExecuteQuerySettings; +import tech.ydb.query.tools.QueryReader; +import tech.ydb.query.tools.SessionRetryContext; +import tech.ydb.slo.core.Config; +import tech.ydb.slo.core.Metrics; +import tech.ydb.slo.core.kv.KvWorkloadParams; +import tech.ydb.slo.core.rmw.RmwRunner; +import tech.ydb.slo.core.rmw.RmwState; +import tech.ydb.table.query.Params; +import tech.ydb.table.result.ResultSetReader; +import tech.ydb.table.values.PrimitiveValue; + +public final class QueryRmwClient implements RmwRunner.Client { + private final QueryClient client; + private final GrpcTransport transport; + private final Metrics metrics; + private final String counters; + private final String ledger; + private final KvWorkloadParams defaults = new KvWorkloadParams(); + private boolean countersCreated; + private boolean ledgerCreated; + + public QueryRmwClient(Config config, String namespace, Metrics metrics) { + AuthProvider auth = config.token().isEmpty() + ? NopAuthProvider.INSTANCE : new TokenAuthProvider(config.token()); + this.transport = GrpcTransport.forConnectionString(config.connectionString()) + .withAuthProvider(auth).build(); + this.client = QueryClient.newClient(transport).build(); + this.metrics = metrics; + this.counters = namespace + "_counters"; + this.ledger = namespace + "_operations"; + } + + QueryRmwClient(QueryClient client, String namespace, Metrics metrics) { + this.client = client; + this.transport = null; + this.metrics = metrics; + this.counters = namespace + "_counters"; + this.ledger = namespace + "_operations"; + } + + private CompletableFuture execute(Metrics.OperationType type, boolean idempotent, + Function>> action) { + AttemptClient observed = new AttemptClient(client); + Metrics.Span span = metrics.startOperation(type); + // Same retry budget/backoff/session timeout as the basic Query workload. + // Only the semantic idempotence declaration differs: the ledger makes RMW replay safe. + return SessionRetryContext.create(observed).maxRetries(Math.max(1, defaults.maxAttempts())) + .idempotent(idempotent).build().supplyResult(action) + .thenApply(Result::getValue).whenComplete((value, error) -> { + if (error == null) { + span.finishSuccess(observed.attempts()); + } else { + span.finishError(observed.attempts(), error.getClass().getSimpleName()); + } + }); + } + + private ExecuteQuerySettings settings(boolean write) { + return ExecuteQuerySettings.newBuilder().withRequestTimeout(Duration.ofMillis( + write ? defaults.writeTimeoutMs() : defaults.readTimeoutMs())).build(); + } + + private void schema(String query) { + execute(Metrics.OperationType.WRITE, false, + session -> session.createQuery(query, TxMode.NONE, Params.empty(), settings(true)).execute()).join(); + } + + @Override + public void create() { + schema("CREATE TABLE `" + counters + "` (id Uint64 NOT NULL, value Int64 NOT NULL, " + + "version Uint64 NOT NULL, PRIMARY KEY (id));"); + countersCreated = true; + schema("CREATE TABLE `" + ledger + "` (op_id Utf8 NOT NULL, counter_id Uint64 NOT NULL, " + + "delta Int64 NOT NULL, PRIMARY KEY (op_id));"); + ledgerCreated = true; + execute(Metrics.OperationType.WRITE, true, session -> session.createQuery( + "UPSERT INTO `" + counters + "` (id, value, version) VALUES (0u, 0, 0u), (1u, 0, 0u);", + TxMode.SERIALIZABLE_RW, Params.empty(), settings(true)).execute()).join(); + } + + @Override + public CompletableFuture apply(RmwState.Operation operation) { + return execute(Metrics.OperationType.WRITE, true, session -> { + QueryTransaction tx = session.createNewTransaction(TxMode.SERIALIZABLE_RW); + return QueryReader.readFrom(tx.createQuery( + "DECLARE $op AS Utf8; SELECT op_id, counter_id, delta FROM `" + ledger + "` WHERE op_id = $op;", + false, Params.of("$op", PrimitiveValue.newText(operation.id())), settings(true))) + .thenCompose(Result.compose(reader -> { + List found = operations(reader); + if (!found.isEmpty()) { + if (!found.equals(List.of(operation))) { + throw new IllegalStateException("Deduplication payload mismatch for " + operation.id()); + } + return tx.commit(); + } + return increment(tx, operation); + })); + }).thenApply(ignored -> null); + } + + private CompletableFuture> increment(QueryTransaction tx, RmwState.Operation operation) { + return QueryReader.readFrom(tx.createQuery( + "DECLARE $counter AS Uint64; SELECT id, value, version FROM `" + counters + "` WHERE id = $counter;", + false, Params.of("$counter", PrimitiveValue.newUint64(operation.counter())), settings(true))) + .thenCompose(Result.compose(reader -> { + List found = counters(reader); + if (found.size() != 1 || found.get(0).id() != operation.counter()) { + throw new IllegalStateException("Missing or unexpected counter " + operation.counter()); + } + RmwState.Counter before = found.get(0); + Params params = Params.of( + "$op", PrimitiveValue.newText(operation.id()), + "$counter", PrimitiveValue.newUint64(operation.counter()), + "$delta", PrimitiveValue.newInt64(operation.delta()), + "$value", PrimitiveValue.newInt64(Math.addExact(before.value(), operation.delta())), + "$version", PrimitiveValue.newUint64(Math.incrementExact(before.version()))); + return tx.createQuery("DECLARE $op AS Utf8; DECLARE $counter AS Uint64; " + + "DECLARE $delta AS Int64; DECLARE $value AS Int64; DECLARE $version AS Uint64; " + + "UPDATE `" + counters + "` SET value = $value, version = $version WHERE id = $counter; " + + "INSERT INTO `" + ledger + "` (op_id, counter_id, delta) VALUES ($op, $counter, $delta);", + false, params, settings(true)).execute() + .thenCompose(Result.compose(ignored -> tx.commit())); + })); + } + + @Override + public RmwState.Snapshot read() { + QueryReader reader = execute(Metrics.OperationType.READ, true, + session -> QueryReader.readFrom(session.createQuery( + "SELECT id, value, version FROM `" + counters + "`; " + + "SELECT op_id, counter_id, delta FROM `" + ledger + "`;", + TxMode.SNAPSHOT_RO, Params.empty(), settings(false)))).join(); + if (reader.getResultSetCount() != 2) { + throw new IllegalStateException("Expected counter and ledger result sets"); + } + return new RmwState.Snapshot(counters(reader.getResultSet(0)), operations(reader.getResultSet(1))); + } + + private static List counters(QueryReader reader) { + if (reader.getResultSetCount() != 1) { + throw new IllegalStateException("Expected one counter result set"); + } + return counters(reader.getResultSet(0)); + } + + private static List counters(ResultSetReader rows) { + List result = new ArrayList<>(); + while (rows.next()) { + result.add(new RmwState.Counter(rows.getColumn("id").getUint64(), + rows.getColumn("value").getInt64(), rows.getColumn("version").getUint64())); + } + return result; + } + + private static List operations(QueryReader reader) { + if (reader.getResultSetCount() != 1) { + throw new IllegalStateException("Expected one ledger result set"); + } + return operations(reader.getResultSet(0)); + } + + private static List operations(ResultSetReader rows) { + List result = new ArrayList<>(); + while (rows.next()) { + result.add(new RmwState.Operation(rows.getColumn("op_id").getText(), + rows.getColumn("counter_id").getUint64(), rows.getColumn("delta").getInt64())); + } + return result; + } + + @Override + public void drop() { + RuntimeException first = null; + if (ledgerCreated) { + try { + schema("DROP TABLE `" + ledger + "`;"); + } catch (RuntimeException error) { + first = error; + } + } + if (countersCreated) { + try { + schema("DROP TABLE `" + counters + "`;"); + } catch (RuntimeException error) { + if (first == null) { + first = error; + } + } + } + if (first != null) { + throw first; + } + } + + @Override + public void close() { + try { + client.close(); + } finally { + if (transport != null) { + transport.close(); + } + } + } +} diff --git a/slo-workload/query-rmw/src/main/resources/log4j2.xml b/slo-workload/query-rmw/src/main/resources/log4j2.xml new file mode 100644 index 0000000..0ebaed0 --- /dev/null +++ b/slo-workload/query-rmw/src/main/resources/log4j2.xml @@ -0,0 +1,34 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/slo-workload/query-rmw/src/test/java/tech/ydb/slo/core/TestMetrics.java b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/core/TestMetrics.java new file mode 100644 index 0000000..1c86326 --- /dev/null +++ b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/core/TestMetrics.java @@ -0,0 +1,10 @@ +package tech.ydb.slo.core; + +/** In-process metrics only: no receiver, transport or environment dependency. */ +public final class TestMetrics { + private TestMetrics() { } + + public static Metrics create() { + return Metrics.create(new Config("grpc://localhost:1/local", "", "", "unit", "query-rmw-test", 0, "")); + } +} diff --git a/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryBoundary.java b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryBoundary.java new file mode 100644 index 0000000..41177f6 --- /dev/null +++ b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryBoundary.java @@ -0,0 +1,220 @@ +package tech.ydb.slo.queryrmw; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +import tech.ydb.common.transaction.TxMode; +import tech.ydb.core.Result; +import tech.ydb.core.Status; +import tech.ydb.core.StatusCode; +import tech.ydb.proto.ValueProtos; +import tech.ydb.query.QueryClient; +import tech.ydb.query.QuerySession; +import tech.ydb.query.QueryStream; +import tech.ydb.query.QueryTransaction; +import tech.ydb.query.result.QueryInfo; +import tech.ydb.query.result.QueryResultPart; +import tech.ydb.query.settings.BeginTransactionSettings; +import tech.ydb.query.settings.CommitTransactionSettings; +import tech.ydb.query.settings.ExecuteQuerySettings; +import tech.ydb.query.settings.RollbackTransactionSettings; +import tech.ydb.slo.core.rmw.RmwState; +import tech.ydb.table.SessionPoolStats; +import tech.ydb.table.query.Params; +import tech.ydb.table.values.PrimitiveType; +import tech.ydb.table.values.PrimitiveValue; + +import static org.junit.jupiter.api.Assertions.*; + +/** SDK boundary only. QueryReader, Result composition and SessionRetryContext remain real. */ +final class QueryBoundary implements QueryClient { + final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + final Map ledger = new LinkedHashMap<>(); + final RmwState.Counter[] counters = {new RmwState.Counter(0, 0, 0), new RmwState.Counter(1, 0, 0)}; + final List idsRead = new ArrayList<>(); + final List schema = new ArrayList<>(); + int acquisitions; + int sessionsClosed; + int commits; + int writes; + int snapshots; + StatusCode acquisitionStatus; + StatusCode commitStatus; + RuntimeException commitException; + RuntimeException readException; + String failedSchema; + RuntimeException schemaException; + boolean closed; + + public CompletableFuture> createSession(Duration timeout) { + assertEquals(Duration.ofSeconds(5), timeout); + acquisitions++; + if (acquisitionStatus != null) { + StatusCode status = acquisitionStatus; + acquisitionStatus = null; + return CompletableFuture.completedFuture(Result.fail(Status.of(status))); + } + return CompletableFuture.completedFuture(Result.success(new Session())); + } + + public ScheduledExecutorService getScheduler() { + return scheduler; + } + + public SessionPoolStats getSessionPoolStats() { + return null; + } + + public void close() { + closed = true; + scheduler.shutdownNow(); + } + + private final class Session implements QuerySession { + private QueryTransaction current; + public String getId() { return "test-session"; } + public QueryTransaction currentTransaction() { return current; } + public QueryTransaction createNewTransaction(TxMode mode) { + assertEquals(TxMode.SERIALIZABLE_RW, mode); + current = new Transaction(this); + return current; + } + public CompletableFuture> beginTransaction(TxMode mode, BeginTransactionSettings settings) { + throw new AssertionError("This adapter uses the SDK implicit transaction API"); + } + public QueryStream createQuery(String sql, TxMode mode, Params params, ExecuteQuerySettings settings) { + if (mode == TxMode.SNAPSHOT_RO) { + snapshots++; + assertTrue(sql.contains("FROM `private_counters`")); + assertTrue(sql.contains("FROM `private_operations`")); + return stream(List.of(counterRows(List.of(counters)), operationRows(new ArrayList<>(ledger.values())))); + } + schema.add(sql); + if (failedSchema != null && sql.startsWith(failedSchema)) { + return failing(schemaException); + } + return stream(List.of()); + } + public void close() { sessionsClosed++; } + } + + private final class Transaction implements QueryTransaction { + private final Session session; + private RmwState.Operation pending; + private RmwState.Counter updated; + Transaction(Session session) { this.session = session; } + public QuerySession getSession() { return session; } + public String getId() { return "test-tx"; } + public String getSessionId() { return session.getId(); } + public TxMode getTxMode() { return TxMode.SERIALIZABLE_RW; } + public CompletableFuture getStatusFuture() { return CompletableFuture.completedFuture(Status.SUCCESS); } + public CompletableFuture rollback(RollbackTransactionSettings settings) { + pending = null; + updated = null; + return CompletableFuture.completedFuture(Status.SUCCESS); + } + public QueryStream createQuery(String sql, boolean commitAtEnd, Params params, ExecuteQuerySettings settings) { + assertFalse(commitAtEnd); + if (sql.contains("SELECT op_id")) { + assertTrue(sql.contains("FROM `private_operations` WHERE op_id = $op")); + String id = params.toPb().get("$op").getValue().getTextValue(); + idsRead.add(id); + if (readException != null) { + return failing(readException); + } + RmwState.Operation operation = ledger.get(id); + return stream(List.of(operationRows(operation == null ? List.of() : List.of(operation)))); + } + if (sql.contains("SELECT id")) { + assertTrue(sql.contains("FROM `private_counters` WHERE id = $counter")); + int id = (int) params.toPb().get("$counter").getValue().getUint64Value(); + return stream(List.of(counterRows(List.of(counters[id])))); + } + assertTrue(sql.contains("UPDATE `private_counters` SET value = $value, version = $version WHERE id = $counter")); + assertTrue(sql.contains("INSERT INTO `private_operations`")); + String id = params.toPb().get("$op").getValue().getTextValue(); + long counter = params.toPb().get("$counter").getValue().getUint64Value(); + pending = new RmwState.Operation(id, counter, params.toPb().get("$delta").getValue().getInt64Value()); + updated = new RmwState.Counter(counter, params.toPb().get("$value").getValue().getInt64Value(), + params.toPb().get("$version").getValue().getUint64Value()); + writes++; + return stream(List.of()); + } + public CompletableFuture> commit(CommitTransactionSettings settings) { + commits++; + if (commitException != null) { + return CompletableFuture.failedFuture(commitException); + } + StatusCode status = commitStatus; + commitStatus = null; + if (status == null || status == StatusCode.UNDETERMINED) { + if (pending != null) { + assertFalse(ledger.containsKey(pending.id())); + ledger.put(pending.id(), pending); + counters[(int) updated.id()] = updated; + } + } + return CompletableFuture.completedFuture(status == null + ? Result.success(new QueryInfo(null)) : Result.fail(Status.of(status))); + } + } + + private static QueryStream stream(List rows) { + return new QueryStream() { + public CompletableFuture> execute(PartsHandler handler) { + if (handler != null) { + for (int index = 0; index < rows.size(); index++) { + handler.onNextPart(new QueryResultPart(index, rows.get(index))); + } + } + return CompletableFuture.completedFuture(Result.success(new QueryInfo(null))); + } + public void cancel() { throw new AssertionError("No admitted operation may be cancelled"); } + }; + } + + private static QueryStream failing(RuntimeException error) { + return new QueryStream() { + public CompletableFuture> execute(PartsHandler handler) { + return CompletableFuture.failedFuture(error); + } + public void cancel() { throw new AssertionError("Unexpected cancellation"); } + }; + } + + private static ValueProtos.ResultSet counterRows(List counters) { + ValueProtos.ResultSet.Builder rows = ValueProtos.ResultSet.newBuilder() + .addColumns(column("id", PrimitiveType.Uint64)) + .addColumns(column("value", PrimitiveType.Int64)) + .addColumns(column("version", PrimitiveType.Uint64)); + for (RmwState.Counter counter : counters) { + rows.addRows(ValueProtos.Value.newBuilder().addItems(PrimitiveValue.newUint64(counter.id()).toPb()) + .addItems(PrimitiveValue.newInt64(counter.value()).toPb()) + .addItems(PrimitiveValue.newUint64(counter.version()).toPb())); + } + return rows.build(); + } + + private static ValueProtos.ResultSet operationRows(List operations) { + ValueProtos.ResultSet.Builder rows = ValueProtos.ResultSet.newBuilder() + .addColumns(column("op_id", PrimitiveType.Text)) + .addColumns(column("counter_id", PrimitiveType.Uint64)) + .addColumns(column("delta", PrimitiveType.Int64)); + for (RmwState.Operation operation : operations) { + rows.addRows(ValueProtos.Value.newBuilder().addItems(PrimitiveValue.newText(operation.id()).toPb()) + .addItems(PrimitiveValue.newUint64(operation.counter()).toPb()) + .addItems(PrimitiveValue.newInt64(operation.delta()).toPb())); + } + return rows.build(); + } + + private static ValueProtos.Column column(String name, PrimitiveType type) { + return ValueProtos.Column.newBuilder().setName(name).setType(type.toPb()).build(); + } +} diff --git a/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryRmwClientTest.java b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryRmwClientTest.java new file mode 100644 index 0000000..500ffa2 --- /dev/null +++ b/slo-workload/query-rmw/src/test/java/tech/ydb/slo/queryrmw/QueryRmwClientTest.java @@ -0,0 +1,175 @@ +package tech.ydb.slo.queryrmw; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import tech.ydb.core.Status; +import tech.ydb.core.StatusCode; +import tech.ydb.core.UnexpectedResultException; +import tech.ydb.slo.core.Metrics; +import tech.ydb.slo.core.TestMetrics; +import tech.ydb.slo.core.rmw.RmwRunner; +import tech.ydb.slo.core.rmw.RmwState; + +import static org.junit.jupiter.api.Assertions.*; + +class QueryRmwClientTest { + private final RmwState.Operation operation = new RmwState.Operation("0:0", 0, 3); + + @Test + void actualQueryAdapterCreatesAppliesReadsAndDropsPrivateTables() throws Exception { + QueryBoundary boundary = new QueryBoundary(); + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(boundary, "private", metrics)) { + client.create(); + client.apply(operation).get(10, TimeUnit.SECONDS); + RmwState.Operation other = new RmwState.Operation("1:0", 1, 7); + client.apply(other).get(10, TimeUnit.SECONDS); + RmwState.Snapshot snapshot = client.read(); + RmwState.validate(Map.of(operation.id(), operation, other.id(), other), snapshot); + assertEquals(List.of(new RmwState.Counter(0, 3, 1), new RmwState.Counter(1, 7, 1)), snapshot.counters()); + client.drop(); + assertEquals(5, boundary.schema.size()); + assertTrue(boundary.schema.get(0).startsWith("CREATE TABLE `private_counters`")); + assertTrue(boundary.schema.get(1).startsWith("CREATE TABLE `private_operations`")); + assertEquals("DROP TABLE `private_operations`;", boundary.schema.get(3)); + assertEquals("DROP TABLE `private_counters`;", boundary.schema.get(4)); + assertEquals(boundary.acquisitions, boundary.sessionsClosed); + } + assertTrue(boundary.closed); + } + + @Test + void failedCreateDropsOnlyAcknowledgedTables() { + for (String table : List.of("private_operations", "private_counters")) { + QueryBoundary boundary = new QueryBoundary(); + UnexpectedResultException cause = new UnexpectedResultException("create rejected", + Status.of(StatusCode.SCHEME_ERROR)); + boundary.failedSchema = "CREATE TABLE `" + table + "`"; + boundary.schemaException = cause; + try (Metrics metrics = TestMetrics.create()) { + RmwRunner runner = new RmwRunner(new QueryRmwClient(boundary, "private", metrics), 2); + assertEquals(1, runner.run(Duration.ofSeconds(1))); + assertSame(cause, runner.failure()); + assertFalse(boundary.schema.contains("DROP TABLE `" + table + "`;")); + assertEquals(table.equals("private_operations"), + boundary.schema.contains("DROP TABLE `private_counters`;")); + assertTrue(boundary.closed); + } + } + } + + @Test + void realSdkRetriesAcquisitionAndCommitAndObservationCountsBoth() throws Exception { + QueryBoundary boundary = new QueryBoundary(); + boundary.acquisitionStatus = StatusCode.BAD_SESSION; + boundary.commitStatus = StatusCode.BAD_SESSION; + AttemptClient observed = new AttemptClient(boundary); + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(observed, "private", metrics)) { + client.apply(operation).get(10, TimeUnit.SECONDS); + assertEquals(3, observed.attempts()); + assertEquals(3, boundary.acquisitions); + assertEquals(2, boundary.sessionsClosed); + assertEquals(List.of(operation.id(), operation.id()), boundary.idsRead); + assertEquals(2, boundary.writes); + assertEquals(2, boundary.commits); + assertEquals(Map.of(operation.id(), operation), boundary.ledger); + assertEquals(new RmwState.Counter(0, 3, 1), boundary.counters[0]); + assertSame(boundary.getScheduler(), observed.getScheduler()); + assertSame(boundary.getTracer(), observed.getTracer()); + } finally { + boundary.close(); + } + } + + @Test + void ambiguousCommitUsesSameIdAndDeduplicatesThroughActualSdkRetry() throws Exception { + QueryBoundary boundary = new QueryBoundary(); + boundary.commitStatus = StatusCode.UNDETERMINED; + AttemptClient observed = new AttemptClient(boundary); + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(observed, "private", metrics)) { + client.apply(operation).get(10, TimeUnit.SECONDS); + assertEquals(2, observed.attempts()); + assertEquals(List.of(operation.id(), operation.id()), boundary.idsRead); + assertEquals(1, boundary.writes); + assertEquals(2, boundary.commits); + assertEquals(new RmwState.Counter(0, 3, 1), boundary.counters[0]); + client.apply(operation).get(10, TimeUnit.SECONDS); + assertEquals(1, boundary.writes); + assertEquals(1, boundary.ledger.size()); + RmwState.validate(Map.of(operation.id(), operation), client.read()); + } finally { + boundary.close(); + } + } + + @Test + void finalNativeCommitCauseSurvivesLifecycleAndMissingLedger() { + QueryBoundary boundary = new QueryBoundary(); + UnexpectedResultException nativeCause = new UnexpectedResultException("commit rejected", Status.of(StatusCode.BAD_REQUEST)); + boundary.commitException = nativeCause; + try (Metrics metrics = TestMetrics.create()) { + QueryRmwClient client = new QueryRmwClient(boundary, "private", metrics); + RmwRunner runner = new RmwRunner(client, 2); + assertEquals(1, runner.run(Duration.ofSeconds(1))); + assertSame(nativeCause, runner.failure()); + assertEquals(1, boundary.snapshots); + assertEquals(1, boundary.commits); + assertTrue(boundary.closed); + assertTrue(boundary.ledger.isEmpty()); + assertEquals("DROP TABLE `private_counters`;", boundary.schema.get(boundary.schema.size() - 1)); + } + } + + @Test + void nativeStreamFailureIsNotReplacedByMissingRows() throws Exception { + QueryBoundary boundary = new QueryBoundary(); + UnexpectedResultException nativeCause = new UnexpectedResultException("stream rejected", Status.of(StatusCode.BAD_REQUEST)); + boundary.readException = nativeCause; + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(boundary, "private", metrics)) { + ExecutionException failure = assertThrows(ExecutionException.class, + () -> client.apply(operation).get(10, TimeUnit.SECONDS)); + assertSame(nativeCause, failure.getCause()); + assertEquals(1, boundary.acquisitions); + assertEquals(0, boundary.commits); + } + } + + @Test + void dedupPayloadMismatchIsTerminalAndDoesNotRetryOrApply() { + QueryBoundary boundary = new QueryBoundary(); + boundary.ledger.put(operation.id(), new RmwState.Operation(operation.id(), 1, operation.delta())); + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(boundary, "private", metrics)) { + ExecutionException failure = assertThrows(ExecutionException.class, + () -> client.apply(operation).get(10, TimeUnit.SECONDS)); + assertInstanceOf(IllegalStateException.class, failure.getCause()); + assertEquals(1, boundary.acquisitions); + assertEquals(0, boundary.writes); + assertEquals(0, boundary.commits); + } + } + + @Test + void successfulNativeSnapshotWithWrongVersionFailsWithoutRetryingRead() { + QueryBoundary boundary = new QueryBoundary(); + boundary.ledger.put(operation.id(), operation); + boundary.counters[0] = new RmwState.Counter(0, 3, 0); + try (Metrics metrics = TestMetrics.create(); + QueryRmwClient client = new QueryRmwClient(boundary, "private", metrics)) { + RmwState.Snapshot snapshot = client.read(); + assertThrows(IllegalStateException.class, + () -> RmwState.validate(Map.of(operation.id(), operation), snapshot)); + assertEquals(1, boundary.acquisitions); + assertEquals(1, boundary.snapshots); + } + } +}