SLO: transactional topic<->table workloads (sync/async), exactly-once under TLI

.github/workflows/slo.yml

@@ -41,6 +41,14 @@ jobs:
             command: "--write-rps 200 --write-threads 8 --read-threads 8"
             metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml
             thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml
+          - name: sync-topic-tx
+            command: "--write-rps 50 --write-threads 8 --read-threads 8 --messages-per-tx 10"
+            metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml
+            thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml
+          - name: async-topic-tx
+            command: "--write-rps 50 --write-threads 8 --read-threads 8 --messages-per-tx 10"
+            metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml
+            thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml
 
     concurrency:
       group: slo-${{ github.ref }}-${{ matrix.sdk.name }}

tests/slo/README.md

@@ -17,13 +17,15 @@ The workload is selected by a single label (`WORKLOAD_NAME` env var, also the
 execution mode is derived from the label itself — an `async-*` label runs the
 async (`ydb.aio`) path:
 
-| Label          | Service       | Mode  |
-|----------------|---------------|-------|
-| `sync-table`   | Table service | sync  |
-| `sync-query`   | Query service | sync  |
-| `async-query`  | Query service | async |
-| `sync-topic`   | Topic service | sync  |
-| `async-topic`  | Topic service | async |
+| Label            | Service               | Mode  |
+|------------------|-----------------------|-------|
+| `sync-table`     | Table service         | sync  |
+| `sync-query`     | Query service         | sync  |
+| `async-query`    | Query service         | async |
+| `sync-topic`     | Topic service         | sync  |
+| `async-topic`    | Topic service         | async |
+| `sync-topic-tx`  | Topic + Query (tx)    | sync  |
+| `async-topic-tx` | Topic + Query (tx)    | async |
 
 > The `--async` CLI flag is kept as a manual override for `*-run` commands.
 > The bare `topic` label is still accepted as an alias for `sync-topic`.
@@ -269,6 +271,28 @@ When running `topic-run` (`sync-topic` / `async-topic`), the program creates `re
 
 Each message carries `writer_id:seqno:write_ts_ns:` followed by padding to the configured size. Topics are scoped per ref so the current and baseline containers (same cluster, run in parallel) don't share a topic.
 
+### Transactional topic ↔ table workload
+
+When running `topic-run` under `sync-topic-tx` / `async-topic-tx`, **both** ends of
+the pipeline run inside a YDB transaction, validating **exactly-once** delivery of
+a topic into a table under chaos **and** under TLI (transaction locks invalidated):
+
+- `writeJob` — a `tx_writer` writes a batch of messages to the topic while the same
+  transaction reads-and-bumps a shared hot counter row (table → topic). The commit
+  persists the topic write and the table update atomically; `seqno` advances only on
+  a successful commit (a chaos/TLI abort leaves no gap).
+- `readJob` — `receive_batch_with_tx` reads a batch while the same transaction
+  UPSERTs each message into a sink table keyed by `(writer_id, seqno)` and bumps a
+  hot row (topic → table). The commit advances the topic offset and writes the sink
+  rows atomically, so a rolled-back tx re-reads and re-UPSERTs the same keys —
+  idempotent, exactly-once.
+
+TLI is induced on purpose via a few shared hot rows (`--tli-hot-keys`); the tx retry
+loop absorbs it (`topic_tx_tli` counts the aborts, informational) and the run must
+still finish with `topic_lost_errors == 0`. The topic and the sink/hot tables are
+ref-scoped. See `TOPIC_TX_SCENARIO.md` for the full design; extra options:
+`--messages-per-tx`, `--tli-hot-keys`, `--sink-table`, `--hot-table`.
+
 ## Collected metrics
 - `oks`      - amount of OK requests
 - `not_oks`  - amount of not OK requests

tests/slo/TOPIC_TX_SCENARIO.md

@@ -0,0 +1,124 @@
+# Topic-tx SLO scenario — both-ends transactional topic ↔ table pipeline
+
+What the `sync-topic-tx` / `async-topic-tx` workloads do: an **exactly-once**
+processing pipeline where **both** ends run inside a YDB transaction. Producers
+write to the topic in a `tx_writer` transaction that also touches a table;
+consumers read from the topic with `receive_batch_with_tx` and, in the *same*
+transaction, UPSERT the message into a sink table. The commit atomically advances
+the topic offset and persists the table rows — so every produced message lands in
+the sink table exactly once, even while a chaos monkey keeps killing YDB nodes.
+
+On top of the connection breaks, we **deliberately induce TLI**
+(`ydb.Aborted`, "transaction locks invalidated" — the штатный optimistic-lock
+conflict) with a small set of shared hot rows. The tx retry loop must absorb both
+failure modes and still deliver exactly once.
+
+## Topology
+
+```
+   contention table (few hot rows)                      per-ref topic: /Root/testdb/slo_topic_tx_<ref>
+   ┌─────────────────────────────┐                      ┌──────────────────────────────────────────────┐
+   │ hot[0] hot[1] … hot[K-1]     │◀──read-modify-write──│ p0: m1 m2 m3 …   (server-ordered per producer) │
+   └─────────────────────────────┘        (TLI)         │ p1: m1 m2 m3 …                                 │
+        ▲ producer tx (table → topic)                   └───────────────────────┬──────────────────────┘
+        │  tx_writer(tx).write(msg) + bump hot[k]                                │ one consumer group
+        │  commit ⇒ topic write + table update atomic                           │
+   w0 ──┘                                          consumer tx (topic → table)   │
+                                     ┌──────────────────────────────────────────▼──────────────────────┐
+                                     │ receive_batch_with_tx(tx) → decode → validate → UPSERT sink       │
+                                     │ + bump hot[k];  commit ⇒ offset advance + sink rows atomic         │
+                                     └───────────────────────────────────────────────────────────────────┘
+                                  ▲                                    │
+                   chaos-monkey: kills a YDB node every ~20-60s        ▼  sink(writer_id, seqno) — exactly once
+```
+
+Each message is self-describing: `writer_id : seqno : write_ts_ns : <padding>`
+(same codec as the plain topic scenario, `jobs/topic_payload.py`).
+
+## Flow (happy path + chaos + TLI)
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant W as Producer tx (tx_writer)
+    participant H as hot[k] (contention row)
+    participant S as YDB topic · partition i
+    participant R as Consumer tx (receive_batch_with_tx)
+    participant T as sink table
+    participant M as Metrics (OTLP)
+
+    Note over W: retry_tx: read+bump hot[k], write msgs, commit
+    loop until --time
+        W->>H: SELECT val / UPSERT val+1  (takes optimistic lock)
+        W->>S: tx_writer.write(payload)   (buffered, flushed on commit)
+        alt commit ok
+            S-->>W: topic write + table update committed atomically
+            Note right of W: seqno advances only on commit (no gap)
+        else Aborted (TLI) or Unavailable (chaos)
+            Note right of W: retry_tx re-runs callee; nothing persisted (no gap)
+        end
+    end
+
+    loop drain as fast as messages arrive
+        S-->>R: receive_batch_with_tx(tx)
+        R->>H: SELECT val / UPSERT val+1  (takes optimistic lock)
+        R->>T: UPSERT sink(writer_id, seqno) — keyed, idempotent
+        alt commit ok
+            R->>M: for each msg: delivered++, e2e = now - write_ts, validate seqno
+            Note right of R: offset advance + sink rows committed atomically
+        else Aborted (TLI) or Unavailable (chaos)
+            R->>M: tli++ (if Aborted); retry_tx re-reads from last committed offset
+            Note right of R: rolled-back tx advanced no offset ⇒ re-processed, upsert idempotent
+        end
+    end
+```
+
+## Invariants we assert
+
+| Signal | Meaning | Expectation |
+|--------|---------|-------------|
+| `topic_lost_errors` | forward gap in a producer's seqno (real loss) | **0** — hard-fails the run, under chaos **and** TLI |
+| exactly-once into sink | sink keyed by `(writer_id, seqno)` — idempotent UPSERT | re-processing after a rollback creates no duplicate row |
+| `topic_tx_tli` | TLI aborts absorbed by the retry loop | > 0 (proves the mechanism fires) — informational |
+| `topic_e2e_latency_p50/p99` | write-commit → read-commit latency | reported |
+| `topic_duplicates` | redelivery after a rolled-back tx | informational (at-least-once at the app level) |
+| `read/write_availability` | liveness under chaos | ~100% |
+
+## Why it stays correct under chaos + TLI
+
+- **The topic op and the table op share one transaction.** A commit advances the
+  topic offset *and* writes the table row atomically; an abort (chaos or TLI)
+  rolls back both. There is no window where one happened without the other.
+- **seqno advances only after a successful commit** — an aborted/retried producer
+  tx leaves no gap. Validation reads the seqno embedded in the payload, so a
+  retried commit is at worst an app-level duplicate, never a false loss.
+- **sink is keyed by `(writer_id, seqno)`** — a consumer tx that rolls back
+  (TLI/chaos) re-reads the same messages from the last committed offset and
+  re-UPSERTs the same keys: idempotent, so the table stays exactly-once.
+- **TLI is retriable** (`ydb/_errors.py`, slow backoff) and counted via
+  `RetrySettings.on_ydb_error_callback`; `retry_tx_*` absorbs it transparently.
+- **validation runs only after the commit returns** — only durable offsets/rows
+  feed the per-producer next-expected-seqno check (shared across readers, since a
+  partition can move between them on rebalance).
+- **per-ref topic + per-ref sink/hot tables** — the `current` and `baseline`
+  containers run on the same cluster in parallel; ref-scoping keeps their
+  validation and their hot-row contention from mixing.
+
+## Inducing TLI on purpose
+
+Both the producer and consumer transactions do a read-modify-write
+(`SELECT val` / `UPSERT val+1`) on one of `--tli-hot-keys` shared hot rows
+(chosen by `writer_id % hot_keys` / `reader_id % hot_keys`). The read takes an
+optimistic lock, so concurrent transactions on the same key conflict on commit and
+one gets `ydb.Aborted` (TLI). Keep the hot-key count small for frequent conflicts,
+larger to make them rare. The hot table doubles as the producer's "table side" of
+the table → topic transaction.
+
+## Not covered here
+
+- Reading the sink table back for an independent full-table exactly-once audit
+  (validation is the in-memory per-producer next-expected-seqno check, fed only by
+  committed batches; the atomic offset+row commit makes that a faithful proxy).
+- Snapshot/read-only tx modes — the pipeline runs serializable read-write.
+- Deliberate non-retriable failures (SchemeError/BadRequest) — those are code/config
+  bugs, out of scope for a liveness SLO.

tests/slo/docker-entrypoint.sh

@@ -6,6 +6,7 @@
 #
 # Inputs come from the env vars injected by the action:
 #   WORKLOAD_NAME       sync-table | sync-query | async-query | sync-topic | async-topic
+#                       | sync-topic-tx | async-topic-tx
 #   WORKLOAD_DURATION   run duration in seconds
 #   YDB_ENDPOINT        grpc://ydb:2136
 #   YDB_DATABASE        /Root/testdb
@@ -21,7 +22,7 @@ set -e
 
 case "${WORKLOAD_NAME:-sync-query}" in
     sync-table|sync-query|async-query) PREFIX=table ;;
-    topic|sync-topic|async-topic) PREFIX=topic ;;
+    topic|sync-topic|async-topic|sync-topic-tx|async-topic-tx) PREFIX=topic ;;
     *)
         echo "Unknown WORKLOAD_NAME: ${WORKLOAD_NAME}" >&2
         exit 1
@@ -41,7 +42,18 @@ EXTRA_ARGS=""
 if [ "$PREFIX" = "topic" ]; then
     REF_RAW="${WORKLOAD_REF:-${REF:-main}}"
     SAFE_REF=$(printf '%s' "$REF_RAW" | tr -c 'a-zA-Z0-9_' '_')
-    EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}"
+    case "${WORKLOAD_NAME:-}" in
+        *topic-tx)
+            # The transactional pipeline gets its own topic plus ref-scoped sink
+            # and hot (contention) tables, so current/baseline don't collide.
+            EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_tx_${SAFE_REF}"
+            EXTRA_ARGS="${EXTRA_ARGS} --sink-table slo_topic_tx_sink_${SAFE_REF}"
+            EXTRA_ARGS="${EXTRA_ARGS} --hot-table slo_topic_tx_hot_${SAFE_REF}"
+            ;;
+        *)
+            EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}"
+            ;;
+    esac
 fi
 
 # Schema prep is idempotent at the SDK level for topics; for tables, a parallel

tests/slo/metrics-topic.yaml

@@ -49,3 +49,15 @@ metrics:
         increase(sdk_topic_messages_duplicated_total[5s])
       )
     round: 1
+
+  # TLI (transaction locks invalidated) aborts absorbed by the tx retry loop in
+  # the *-topic-tx workloads. Informational: it is the штатный optimistic-lock
+  # mechanism firing under the hot-key contention we induce on purpose — the run
+  # must still complete with topic_lost_errors == 0.
+  - name: topic_tx_tli
+    unit: aborts
+    query: |
+      sum by(ref) (
+        increase(sdk_topic_tx_tli_total[5s])
+      )
+    round: 1

tests/slo/src/core/metrics.py

@@ -85,6 +85,10 @@ def inc_duplicated(self, n: int = 1) -> None:
         """Count messages detected as duplicates (redelivery of an already-seen seqno)."""
         return None
 
+    def inc_tli(self, n: int = 1) -> None:
+        """Count transaction-locks-invalidated (TLI) aborts absorbed by the tx retry loop."""
+        return None
+
 
 class DummyMetrics(BaseMetrics):
     def start(self, labels) -> float:
@@ -202,6 +206,10 @@ def __init__(self, otlp_metrics_endpoint: str):
             name="sdk.topic.messages.duplicated.total",
             description="Total number of messages detected as duplicates (redelivered seqno).",
         )
+        self._topic_tx_tli = self._meter.create_counter(
+            name="sdk.topic.tx.tli.total",
+            description="Total number of TLI (transaction locks invalidated) aborts absorbed by the tx retry loop.",
+        )
 
         self._lock = threading.Lock()
         self._hdr: dict = {}
@@ -293,6 +301,9 @@ def inc_lost(self, n: int = 1) -> None:
     def inc_duplicated(self, n: int = 1) -> None:
         self._topic_duplicated.add(int(n), attributes={"ref": REF})
 
+    def inc_tli(self, n: int = 1) -> None:
+        self._topic_tx_tli.add(int(n), attributes={"ref": REF})
+
 
 def _resolve_metrics_endpoint(cli_endpoint: Optional[str]) -> str:
     """

tests/slo/src/jobs/async_topic_jobs.py

@@ -150,7 +150,9 @@ async def _run_topic_reads(self, reader_id: int):
         logger.info("Stop async topic reader %s", reader_id)
 
     def _validate(self, msg) -> None:
-        decoded = decode_payload(msg.data)
+        self._record(decode_payload(msg.data))
+
+    def _record(self, decoded) -> None:
         if decoded is None:
             self.metrics.inc_delivered()
             return