Add isolated transactional Query RMW workload

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.
 - `<module>/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

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<Void> 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<Throwable> 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<String, RmwState.Operation> expected = new LinkedHashMap<>();
+        try {
+            client.create();
+            long started = clock.getAsLong();
+            long round = 0;
+            while (!stop.get() && (duration.isZero()
+                    || clock.getAsLong() - started < duration.toNanos())) {
+                List<CompletableFuture<Void>> 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);
+    }
+}

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<Counter> counters;
+        private final List<Operation> operations;
+
+        public Snapshot(List<Counter> counters, List<Operation> operations) {
+            this.counters = List.copyOf(counters);
+            this.operations = List.copyOf(operations);
+        }
+
+        public List<Counter> counters() { return counters; }
+        public List<Operation> operations() { return operations; }
+    }
+
+    public static void validate(Map<String, Operation> expected, Snapshot actual) {
+        Map<String, Operation> 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<Long, Counter> 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));
+            }
+        }
+    }
+}