Fail KV workloads on final errors and wrong data

slo-workload/AGENTS.md

@@ -147,9 +147,12 @@ pushing to CI.
   Locally that's fine; in CI it would merge current and baseline series. The
   action always sets it, but watch out when reproducing CI behaviour
   manually.
-- **Prefill failure threshold is 50%**. If half the writes during prefill
-  fail, the runner refuses to start the read loop — empty key-space means
-  meaningless read latency. Don't lower the threshold to "fix" a flaky test.
+- **Any final operation failure fails the workload**, including prefill.
+  Recovered retries stay successful. The first terminal cause survives peer
+  shutdown and cleanup; successful later operations cannot erase it.
+- **Each launcher owns a process-private table for its whole lifecycle.**
+  One owner checks a reserved control key's exact generation and payload before,
+  during and after load; random load reads do not assert committed-key presence.
 - **`acceptUnknownOptions(false)`** on JCommander is intentional. A typo in
   the ydb-slo-action `workload_*_command` input should fail loudly, not
   silently fall back to defaults.

slo-workload/core/pom.xml

@@ -21,6 +21,12 @@
     </description>
 
     <dependencies>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.10.1</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.beust</groupId>
             <artifactId>jcommander</artifactId>

slo-workload/core/src/main/java/tech/ydb/slo/core/Config.java

@@ -9,7 +9,7 @@ public final class Config {
     private final int durationSeconds;
     private final String otlpEndpoint;
 
-    private Config(
+    Config(
             String connectionString,
             String jdbcUrl,
             String token,

slo-workload/core/src/main/java/tech/ydb/slo/core/Launcher.java

@@ -111,7 +111,8 @@ public static int run(
     }
 
     public static String tablePathFor(Config config) {
-        return sanitize(config.workloadName()) + "_" + sanitize(config.ref());
+        return sanitize(config.workloadName()) + "_" + sanitize(config.ref())
+                + "_" + java.util.UUID.randomUUID().toString().replace("-", "");
     }
 
     private static void closeQuietly(AutoCloseable closeable, String name) {

slo-workload/core/src/main/java/tech/ydb/slo/core/kv/ControlRows.java

@@ -0,0 +1,42 @@
+package tech.ydb.slo.core.kv;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+
+/** One runner owns this key in its process-private table; load IDs start at zero. */
+final class ControlRows {
+    private Row published;
+    private long generation;
+
+    void verify(KvSession session, KvWorkloadParams params) {
+        if (published != null) {
+            requireRow(session.read(published.id(), params.readTimeoutMs()), published);
+        }
+        Row next = new Row(Long.MAX_VALUE, "control/" + generation, generation + 0.25,
+                Instant.now().truncatedTo(ChronoUnit.MICROS), generation);
+        requireSuccess(session.write(next, params.writeTimeoutMs()));
+        published = next;
+        requireRow(session.read(next.id(), params.readTimeoutMs()), next);
+        generation++;
+    }
+
+    private static void requireSuccess(OpOutcome outcome) {
+        if (!outcome.isSuccess()) {
+            throw new IllegalStateException("Control operation failed: " + outcome.errorKind(), outcome.cause());
+        }
+    }
+
+    static void requireRow(OpOutcome outcome, Row expected) {
+        requireSuccess(outcome);
+        if (outcome.rows().size() != 1) {
+            throw new IllegalStateException("Control row cardinality: " + outcome.rows().size());
+        }
+        Row actual = outcome.rows().get(0);
+        if (actual.id() != expected.id() || actual.payloadHash() != expected.payloadHash()
+                || !expected.payloadStr().equals(actual.payloadStr())
+                || Double.compare(expected.payloadDouble(), actual.payloadDouble()) != 0
+                || !expected.payloadTimestamp().equals(actual.payloadTimestamp())) {
+            throw new IllegalStateException("Control row mismatch at generation " + expected.payloadHash());
+        }
+    }
+}

slo-workload/core/src/main/java/tech/ydb/slo/core/kv/OpOutcome.java

@@ -4,19 +4,40 @@ public final class OpOutcome {
     private final boolean success;
     private final int retryAttempts;
     private final String errorKind;
+    private final Throwable cause;
+    private final java.util.List<Row> rows;
 
-    private OpOutcome(boolean success, int retryAttempts, String errorKind) {
+    private OpOutcome(boolean success, int retryAttempts, String errorKind, Throwable cause,
+            java.util.List<Row> rows) {
         this.success = success;
         this.retryAttempts = Math.max(0, retryAttempts);
         this.errorKind = errorKind;
+        this.cause = cause;
+        this.rows = java.util.List.copyOf(rows);
     }
 
     public static OpOutcome success(int retryAttempts) {
-        return new OpOutcome(true, retryAttempts, null);
+        return success(retryAttempts, java.util.List.of());
+    }
+
+    public static OpOutcome success(int retryAttempts, java.util.List<Row> rows) {
+        return new OpOutcome(true, retryAttempts, null, null, rows);
+    }
+
+    public java.util.List<Row> rows() {
+        return rows;
     }
 
     public static OpOutcome error(int retryAttempts, String errorKind) {
-        return new OpOutcome(false, retryAttempts, errorKind);
+        return error(retryAttempts, errorKind, null);
+    }
+
+    public static OpOutcome error(int retryAttempts, String errorKind, Throwable cause) {
+        return new OpOutcome(false, retryAttempts, errorKind, cause, java.util.List.of());
+    }
+
+    public Throwable cause() {
+        return cause;
     }
 
     public boolean isSuccess() {

slo-workload/core/src/main/java/tech/ydb/slo/core/kv/RowGenerator.java

@@ -21,10 +21,13 @@ public Row generate() {
     }
 
     public static Row generate(long id) {
+        if (id == Long.MAX_VALUE) {
+            throw new IllegalStateException("Load writer reached the reserved control key");
+        }
         long payloadHash = ThreadLocalRandom.current().nextLong();
         double payloadDouble = ThreadLocalRandom.current().nextDouble();
         String payloadStr = randomPayloadString();
-        Instant payloadTimestamp = Instant.now();
+        Instant payloadTimestamp = Instant.now().truncatedTo(java.time.temporal.ChronoUnit.MICROS);
 
         return new Row(id, payloadStr, payloadDouble, payloadTimestamp, payloadHash);
     }