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,25 @@ 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:
+
+- `writeJob` — a `tx_writer` writes a batch of messages to the topic inside a
+  transaction (a transactional topic write); `seqno` advances only on a successful
+  commit (a chaos 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)` (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.
+
+The run must finish with `topic_lost_errors == 0`. The topic and the sink table are
+ref-scoped. See `TOPIC_TX_SCENARIO.md` for the full design; extra options:
+`--messages-per-tx`, `--sink-table`.
+
 ## Collected metrics
 - `oks`      - amount of OK requests
 - `not_oks`  - amount of not OK requests

tests/slo/TOPIC_TX_SCENARIO.md

@@ -0,0 +1,101 @@
+# Topic-tx SLO scenario — 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 with a `tx_writer` (a transactional topic write); 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 sink rows — so every produced message lands in the sink table exactly
+once, even while a chaos monkey keeps killing YDB nodes.
+
+## Topology
+
+```
+   producer tx (transactional topic write)          per-ref topic: /Root/testdb/slo_topic_tx_<ref>
+        │  tx_writer(tx).write(msg); commit          ┌──────────────────────────────────────────────┐
+   w0 ──┼──produce p0──▶ partition 0 ────────────────▶│ p0: m1 m2 m3 …   (server-ordered per producer) │
+   w1 ──┴──produce p1──▶ partition 1 ────────────────▶│ p1: m1 m2 m3 …                                 │
+                                                      └───────────────────────┬──────────────────────┘
+                                                                             │ one consumer group
+                                   ┌──────────────────────────────────────────▼──────────────────────┐
+                                   │ consumer tx: receive_batch_with_tx(tx) → decode → validate        │
+                                   │              → UPSERT sink(writer_id, seqno); commit               │
+                                   └───────────────────────────────────────────┬───────────────────────┘
+                                  ▲                                             ▼  sink(writer_id, seqno) — exactly once
+                   chaos-monkey: kills a YDB node every ~20-60s
+```
+
+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)
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant W as Producer tx (tx_writer)
+    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: tx_writer.write(msgs), commit
+    loop until --time
+        W->>S: tx_writer.write(payload)   (buffered, flushed on commit)
+        alt commit ok
+            S-->>W: topic write committed
+            Note right of W: seqno advances only on commit (no gap)
+        else 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->>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 Unavailable (chaos)
+            R->>M: 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 |
+| exactly-once into sink | sink keyed by `(writer_id, seqno)` — idempotent UPSERT | re-processing after a rollback creates no duplicate row |
+| `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
+
+- **The topic read and the table write share one transaction.** A consumer commit
+  advances the topic offset *and* writes the sink row atomically; an abort (chaos)
+  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
+  (chaos) re-reads the same messages from the last committed offset and
+  re-UPSERTs the same keys: idempotent, so the table stays exactly-once.
+- **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 table** — the `current` and `baseline` containers
+  run on the same cluster in parallel; ref-scoping keeps their validation from
+  mixing.
+
+## 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,17 @@ 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 a ref-scoped
+            # sink table, 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="--path ${DATABASE%/}/slo_topic_${SAFE_REF}"
+            ;;
+    esac
 fi
 
 # Schema prep is idempotent at the SDK level for topics; for tables, a parallel

tests/slo/src/core/metrics.py

@@ -68,6 +68,13 @@ def measure(self, labels):
         finally:
             self.stop(labels, start_ts, error=error)
 
+    def cancel(self, labels, start_time: float) -> None:
+        """Undo a ``start()`` for an operation that turned out to be a no-op
+        (e.g. a transactional reader that had no batch to process): balances the
+        pending counter without recording a success or a failure, so idle waits
+        do not count against availability."""
+        return None
+
     # --- Topic workload extensions (no-ops unless overridden) ---
     def record_e2e(self, seconds: float) -> None:
         """Record an end-to-end (write -> read) message latency, in seconds."""
@@ -252,6 +259,10 @@ def stop(
         with self._lock:
             self._get_hdr(op_type, op_status).record_value(duration_us)
 
+    def cancel(self, labels, start_time: float) -> None:
+        labels_t = _normalize_labels(labels)
+        self._pending.add(-1, attributes={"ref": REF, "operation_type": labels_t[0]})
+
     def push(self) -> None:
         with self._lock:
             for (op_type, op_status), hist in self._hdr.items():

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

tests/slo/src/jobs/async_topic_tx_jobs.py

@@ -0,0 +1,155 @@
+import asyncio
+import logging
+import time
+
+from core.metrics import OP_TYPE_READ, OP_TYPE_WRITE, REF
+
+import ydb
+import ydb.aio
+
+from .async_topic_jobs import AsyncTopicJobManager
+from .topic_payload import decode_payload, encode_payload
+from .topic_tx_jobs import SINK_UPSERT_TEMPLATE
+
+logger = logging.getLogger(__name__)
+
+
+class AsyncTopicTxJobManager(AsyncTopicJobManager):
+    """Async mirror of :class:`TopicTxJobManager` (transactional topic <-> table
+    pipeline). See that class for the scenario description."""
+
+    def __init__(self, driver, args, metrics, sink_table):
+        super().__init__(driver, args, metrics)
+        self.sink_table = sink_table
+        self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10)))
+        self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table)
+        self.retry_settings = ydb.RetrySettings(idempotent=True)
+
+    # --- shared query helper --------------------------------------------------
+
+    async def _upsert_sink(self, tx, decoded):
+        writer_id, seqno, write_ts_ns = decoded
+        async with await tx.execute(
+            self._sink_upsert_query,
+            {
+                "$writer_id": (writer_id, ydb.PrimitiveType.Uint64),
+                "$seqno": (seqno, ydb.PrimitiveType.Uint64),
+                "$write_ts_ns": (write_ts_ns, ydb.PrimitiveType.Uint64),
+            },
+            commit_tx=False,
+        ) as result_sets:
+            async for _ in result_sets:
+                pass
+
+    # --- producer (transactional topic write) --------------------------------
+
+    async def _run_topic_writes(self, writer_id, limiter):
+        start_time = time.time()
+        partition_id = writer_id % self.partitions
+        producer_id = f"{REF}-p{writer_id}"
+        logger.info("Start async topic tx writer %s (partition %s, producer %s)", writer_id, partition_id, producer_id)
+
+        # `seqno` lives across recreations and advances only after a successful
+        # commit, so the sequence the reader sees is gap-free.
+        seqno = 1
+        while time.time() - start_time < self.args.time:
+            try:
+                async with ydb.aio.QuerySessionPool(self.aio_driver) as pool:
+                    while time.time() - start_time < self.args.time:
+                        base_seqno = seqno
+                        payloads = []
+                        for k in range(self.messages_per_tx):
+                            if time.time() - start_time >= self.args.time:
+                                break
+                            async with limiter:
+                                payloads.append(
+                                    encode_payload(
+                                        writer_id, base_seqno + k, time.monotonic_ns(), self.args.message_size
+                                    )
+                                )
+                        if not payloads:
+                            break
+
+                        async def callee(tx):
+                            tx_writer = self.aio_driver.topic_client.tx_writer(
+                                tx,
+                                self.args.path,
+                                producer_id=producer_id,
+                                partition_id=partition_id,
+                                codec=ydb.TopicCodec.RAW,
+                            )
+                            for payload in payloads:
+                                await tx_writer.write(ydb.TopicWriterMessage(data=payload))
+
+                        ts = self.metrics.start((OP_TYPE_WRITE,))
+                        try:
+                            await pool.retry_tx_async(callee, retry_settings=self.retry_settings)
+                            self.metrics.stop((OP_TYPE_WRITE,), ts)
+                            seqno = base_seqno + len(payloads)  # advance only on commit
+                        except Exception as e:
+                            self.metrics.stop((OP_TYPE_WRITE,), ts, error=e)
+                            logger.error("Tx write error (recreating writer): %s", e)
+                            break
+            except Exception as e:
+                logger.error("Topic tx writer %s recreate: %s", writer_id, e)
+                await asyncio.sleep(0.2)
+
+        logger.info("Stop async topic tx writer %s", writer_id)
+
+    # --- consumer (topic -> table) -------------------------------------------
+
+    async def _run_topic_reads(self, reader_id):
+        start_time = time.time()
+        read_timeout = self.args.read_timeout / 1000
+        logger.info("Start async topic tx reader %s", reader_id)
+
+        while time.time() - start_time < self.args.time:
+            try:
+                async with ydb.aio.QuerySessionPool(self.aio_driver) as pool:
+                    async with self.aio_driver.topic_client.reader(self.args.path, self.args.consumer) as reader:
+                        while time.time() - start_time < self.args.time:
+                            committed = []
+
+                            async def callee(tx):
+                                nonlocal committed
+                                committed = []
+                                batch = await asyncio.wait_for(
+                                    reader.receive_batch_with_tx(tx, max_messages=self.messages_per_tx), read_timeout
+                                )
+                                if batch is None:
+                                    return
+                                for msg in batch.messages:
+                                    decoded = decode_payload(msg.data)
+                                    if decoded is not None:
+                                        await self._upsert_sink(tx, decoded)
+                                    committed.append(decoded)
+
+                            ts = self.metrics.start((OP_TYPE_READ,))
+                            try:
+                                await pool.retry_tx_async(callee, retry_settings=self.retry_settings)
+                            except asyncio.TimeoutError:
+                                # No batch within read_timeout: an idle reader or
+                                # the SDK transparently reconnecting under chaos,
+                                # not a read failure. Don't count the op (real
+                                # errors surface as exceptions the tx retries).
+                                self.metrics.cancel((OP_TYPE_READ,), ts)
+                                continue
+                            except Exception as e:
+                                self.metrics.stop((OP_TYPE_READ,), ts, error=e)
+                                logger.error("Tx read error (recreating reader): %s", e)
+                                break
+
+                            if not committed:
+                                self.metrics.cancel((OP_TYPE_READ,), ts)
+                                continue
+
+                            # A batch committed: the offsets/rows are durable, so
+                            # the seqnos are exactly-once delivered.
+                            self.metrics.stop((OP_TYPE_READ,), ts)
+                            for decoded in committed:
+                                self._record(decoded)
+            except Exception as e:
+                logger.error("Topic tx reader %s recreate: %s", reader_id, e)
+                await asyncio.sleep(0.2)
+
+        logger.info("Stop async topic tx reader %s", reader_id)