From 5754f2ccd67a3a4ab9245703422c43cf6e16ed23 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Fri, 3 Jul 2026 11:50:49 +0300 Subject: [PATCH 1/3] SLO: transactional topic<->table workloads (sync/async), exactly-once under TLI Both-ends transactional pipeline: tx_writer (table->topic) and receive_batch_with_tx + sink upsert (topic->table), commit makes topic offset and table write atomic. TLI is induced via hot-row contention and absorbed by retry_tx; run stays exactly-once and loss-free. --- .github/workflows/slo.yml | 8 + tests/slo/README.md | 38 +++- tests/slo/TOPIC_TX_SCENARIO.md | 124 +++++++++++++ tests/slo/docker-entrypoint.sh | 16 +- tests/slo/metrics-topic.yaml | 12 ++ tests/slo/src/core/metrics.py | 11 ++ tests/slo/src/jobs/async_topic_jobs.py | 4 +- tests/slo/src/jobs/async_topic_tx_jobs.py | 186 +++++++++++++++++++ tests/slo/src/jobs/topic_jobs.py | 4 +- tests/slo/src/jobs/topic_tx_jobs.py | 215 ++++++++++++++++++++++ tests/slo/src/options.py | 32 ++++ tests/slo/src/runners/topic_runner.py | 76 +++++++- tests/slo/thresholds-topic.yaml | 4 + 13 files changed, 716 insertions(+), 14 deletions(-) create mode 100644 tests/slo/TOPIC_TX_SCENARIO.md create mode 100644 tests/slo/src/jobs/async_topic_tx_jobs.py create mode 100644 tests/slo/src/jobs/topic_tx_jobs.py diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml index 8c066a3d4..67231712a 100644 --- a/.github/workflows/slo.yml +++ b/.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 }} diff --git a/tests/slo/README.md b/tests/slo/README.md index 88d46e724..42fa194c5 100644 --- a/tests/slo/README.md +++ b/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 diff --git a/tests/slo/TOPIC_TX_SCENARIO.md b/tests/slo/TOPIC_TX_SCENARIO.md new file mode 100644 index 000000000..5a4ffeb31 --- /dev/null +++ b/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_ + ┌─────────────────────────────┐ ┌──────────────────────────────────────────────┐ + │ 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 : ` +(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. diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh index 55d933ec3..30961e2ed 100755 --- a/tests/slo/docker-entrypoint.sh +++ b/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 diff --git a/tests/slo/metrics-topic.yaml b/tests/slo/metrics-topic.yaml index f1fd913d2..3795a195e 100644 --- a/tests/slo/metrics-topic.yaml +++ b/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 diff --git a/tests/slo/src/core/metrics.py b/tests/slo/src/core/metrics.py index 313e1683d..59dc193b3 100644 --- a/tests/slo/src/core/metrics.py +++ b/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: """ diff --git a/tests/slo/src/jobs/async_topic_jobs.py b/tests/slo/src/jobs/async_topic_jobs.py index c4cfe691b..e39df011a 100644 --- a/tests/slo/src/jobs/async_topic_jobs.py +++ b/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 diff --git a/tests/slo/src/jobs/async_topic_tx_jobs.py b/tests/slo/src/jobs/async_topic_tx_jobs.py new file mode 100644 index 000000000..6a2dde3b1 --- /dev/null +++ b/tests/slo/src/jobs/async_topic_tx_jobs.py @@ -0,0 +1,186 @@ +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 HOT_SELECT_TEMPLATE, HOT_UPSERT_TEMPLATE, SINK_UPSERT_TEMPLATE + +logger = logging.getLogger(__name__) + + +class AsyncTopicTxJobManager(AsyncTopicJobManager): + """Async mirror of :class:`TopicTxJobManager` (both-ends transactional + topic <-> table pipeline). See that class for the scenario description.""" + + def __init__(self, driver, args, metrics, sink_table, hot_table): + super().__init__(driver, args, metrics) + self.sink_table = sink_table + self.hot_table = hot_table + self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) + self.hot_keys = max(1, int(getattr(self.args, "tli_hot_keys", 4))) + + self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) + self._hot_select_query = HOT_SELECT_TEMPLATE.format(hot_table) + self._hot_upsert_query = HOT_UPSERT_TEMPLATE.format(hot_table) + + self.retry_settings = ydb.RetrySettings(idempotent=True, on_ydb_error_callback=self._on_ydb_error) + + async def run_tests(self): + # Emit the TLI counter at 0 up front so it shows as 0 (not a missing + # series) on a clean run; the delivery counters are emitted by super(). + self.metrics.inc_tli(0) + await super().run_tests() + + def _on_ydb_error(self, e): + if isinstance(e, ydb.Aborted): + self.metrics.inc_tli() + logger.debug("TLI abort (will retry): %s", e) + + # --- shared query helpers ------------------------------------------------- + + async def _exec(self, tx, query, params): + rows = [] + async with await tx.execute(query, params, commit_tx=False) as result_sets: + async for result_set in result_sets: + rows.extend(result_set.rows) + return rows + + async def _bump_hot(self, tx, hot_id): + # Read-modify-write on a shared hot row: the read takes an optimistic + # lock, so concurrent writers/readers on the same key conflict -> TLI. + rows = await self._exec(tx, self._hot_select_query, {"$id": (hot_id, ydb.PrimitiveType.Uint64)}) + cur = rows[0]["val"] if rows and rows[0]["val"] is not None else 0 + await self._exec( + tx, + self._hot_upsert_query, + {"$id": (hot_id, ydb.PrimitiveType.Uint64), "$val": (cur + 1, ydb.PrimitiveType.Uint64)}, + ) + + async def _upsert_sink(self, tx, decoded): + writer_id, seqno, write_ts_ns = decoded + await self._exec( + tx, + 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), + }, + ) + + # --- producer (table -> topic) ------------------------------------------- + + 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}" + hot_id = writer_id % self.hot_keys + 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): + await self._bump_hot(tx, hot_id) + 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 + hot_id = reader_id % self.hot_keys + 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 + await self._bump_hot(tx, hot_id) + 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) + self.metrics.stop((OP_TYPE_READ,), ts) + except asyncio.TimeoutError as e: + # Starved reader (outage/stall): a visible read + # failure, not a silent wait. + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + continue + except Exception as e: + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + logger.error("Tx read error (recreating reader): %s", e) + break + + # Validate only after commit: these offsets/rows are + # durable, so the seqnos are exactly-once delivered. + 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) diff --git a/tests/slo/src/jobs/topic_jobs.py b/tests/slo/src/jobs/topic_jobs.py index 216458578..325ef1646 100644 --- a/tests/slo/src/jobs/topic_jobs.py +++ b/tests/slo/src/jobs/topic_jobs.py @@ -156,7 +156,9 @@ def _run_topic_reads(self, reader_id): logger.info("Stop topic reader %s", reader_id) def _validate(self, msg): - decoded = decode_payload(msg.data) + self._record(decode_payload(msg.data)) + + def _record(self, decoded): if decoded is None: self.metrics.inc_delivered() return diff --git a/tests/slo/src/jobs/topic_tx_jobs.py b/tests/slo/src/jobs/topic_tx_jobs.py new file mode 100644 index 000000000..36dcb13ae --- /dev/null +++ b/tests/slo/src/jobs/topic_tx_jobs.py @@ -0,0 +1,215 @@ +import logging +import time + +from core.metrics import OP_TYPE_READ, OP_TYPE_WRITE, REF + +import ydb + +from .topic_jobs import TopicJobManager +from .topic_payload import decode_payload, encode_payload + +logger = logging.getLogger(__name__) + +SINK_UPSERT_TEMPLATE = """ +DECLARE $writer_id AS Uint64; +DECLARE $seqno AS Uint64; +DECLARE $write_ts_ns AS Uint64; +UPSERT INTO `{}` (writer_id, seqno, write_ts_ns, received_at) +VALUES ($writer_id, $seqno, $write_ts_ns, CurrentUtcTimestamp()); +""" + +HOT_SELECT_TEMPLATE = """ +DECLARE $id AS Uint64; +SELECT val FROM `{}` WHERE id = $id; +""" + +HOT_UPSERT_TEMPLATE = """ +DECLARE $id AS Uint64; +DECLARE $val AS Uint64; +UPSERT INTO `{}` (id, val) VALUES ($id, $val); +""" + + +class TopicTxJobManager(TopicJobManager): + """Both-ends transactional topic <-> table pipeline (sync). + + Producer: a ``tx_writer`` writes a batch of messages to the topic while the + same transaction reads-and-bumps a hot counter row (table -> topic). + Consumer: ``receive_batch_with_tx`` reads a batch while the same transaction + UPSERTs each message into a sink table and bumps a hot counter row + (topic -> table). The commit advances the topic offset and persists the + table rows atomically -> exactly-once. + + The shared hot rows (``--tli-hot-keys`` of them) create optimistic-lock + contention, so transactions occasionally fail with TLI (``ydb.Aborted``); + the tx retry loop absorbs them and delivery stays exactly-once. TLI aborts + are counted via ``RetrySettings.on_ydb_error_callback``. + """ + + def __init__(self, driver, args, metrics, sink_table, hot_table): + super().__init__(driver, args, metrics) + self.sink_table = sink_table + self.hot_table = hot_table + self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) + self.hot_keys = max(1, int(getattr(self.args, "tli_hot_keys", 4))) + + self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) + self._hot_select_query = HOT_SELECT_TEMPLATE.format(hot_table) + self._hot_upsert_query = HOT_UPSERT_TEMPLATE.format(hot_table) + + self.retry_settings = ydb.RetrySettings(idempotent=True, on_ydb_error_callback=self._on_ydb_error) + + def run_tests(self): + # Emit the TLI counter at 0 up front so it shows as 0 (not a missing + # series) on a clean run; the delivery counters are emitted by super(). + self.metrics.inc_tli(0) + super().run_tests() + + def _on_ydb_error(self, e): + if isinstance(e, ydb.Aborted): + self.metrics.inc_tli() + logger.debug("TLI abort (will retry): %s", e) + + # --- shared query helpers ------------------------------------------------- + + def _exec(self, tx, query, params): + rows = [] + with tx.execute(query, params, commit_tx=False) as result_sets: + for result_set in result_sets: + rows.extend(result_set.rows) + return rows + + def _bump_hot(self, tx, hot_id): + # Read-modify-write on a shared hot row: the read takes an optimistic + # lock, so concurrent writers/readers on the same key conflict -> TLI. + rows = self._exec(tx, self._hot_select_query, {"$id": (hot_id, ydb.PrimitiveType.Uint64)}) + cur = rows[0]["val"] if rows and rows[0]["val"] is not None else 0 + self._exec( + tx, + self._hot_upsert_query, + {"$id": (hot_id, ydb.PrimitiveType.Uint64), "$val": (cur + 1, ydb.PrimitiveType.Uint64)}, + ) + + def _upsert_sink(self, tx, decoded): + writer_id, seqno, write_ts_ns = decoded + self._exec( + tx, + 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), + }, + ) + + # --- producer (table -> topic) ------------------------------------------- + + 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}" + hot_id = writer_id % self.hot_keys + logger.info("Start 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 (a failed/aborted + # tx leaves no gap; a retried commit is at worst an app-level duplicate). + seqno = 1 + while time.time() - start_time < self.args.time: + try: + with ydb.QuerySessionPool(self.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 + with limiter: + payloads.append( + encode_payload( + writer_id, base_seqno + k, time.monotonic_ns(), self.args.message_size + ) + ) + if not payloads: + break + + def callee(tx): + self._bump_hot(tx, hot_id) + tx_writer = self.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: + tx_writer.write(ydb.TopicWriterMessage(data=payload)) + + ts = self.metrics.start((OP_TYPE_WRITE,)) + try: + pool.retry_tx_sync(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) + time.sleep(0.2) + + logger.info("Stop topic tx writer %s", writer_id) + + # --- consumer (topic -> table) ------------------------------------------- + + def _run_topic_reads(self, reader_id): + start_time = time.time() + read_timeout = self.args.read_timeout / 1000 + hot_id = reader_id % self.hot_keys + logger.info("Start topic tx reader %s", reader_id) + + while time.time() - start_time < self.args.time: + try: + with ydb.QuerySessionPool(self.driver) as pool: + with self.driver.topic_client.reader(self.args.path, self.args.consumer) as reader: + while time.time() - start_time < self.args.time: + committed = [] + + def callee(tx): + nonlocal committed + committed = [] + batch = reader.receive_batch_with_tx( + tx, max_messages=self.messages_per_tx, timeout=read_timeout + ) + if batch is None: + return + self._bump_hot(tx, hot_id) + for msg in batch.messages: + decoded = decode_payload(msg.data) + if decoded is not None: + self._upsert_sink(tx, decoded) + committed.append(decoded) + + ts = self.metrics.start((OP_TYPE_READ,)) + try: + pool.retry_tx_sync(callee, retry_settings=self.retry_settings) + self.metrics.stop((OP_TYPE_READ,), ts) + except TimeoutError as e: + # Starved reader (outage/stall): a visible read + # failure, not a silent wait. + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + continue + except Exception as e: + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + logger.error("Tx read error (recreating reader): %s", e) + break + + # Validate only after commit: these offsets/rows are + # durable, so the seqnos are exactly-once delivered. + for decoded in committed: + self._record(decoded) + except Exception as e: + logger.error("Topic tx reader %s recreate: %s", reader_id, e) + time.sleep(0.2) + + logger.info("Stop topic tx reader %s", reader_id) diff --git a/tests/slo/src/options.py b/tests/slo/src/options.py index 200c3c556..7186dc977 100644 --- a/tests/slo/src/options.py +++ b/tests/slo/src/options.py @@ -97,6 +97,28 @@ def make_table_cleanup_parser(subparsers): add_common_options(table_cleanup_parser) +def add_topic_tx_options(parser): + """Options for the transactional topic <-> table workloads (``*-topic-tx``).""" + parser.add_argument( + "--sink-table", + default="slo_topic_tx_sink", + type=str, + help="Sink table for the transactional consumer (exactly-once target)", + ) + parser.add_argument( + "--hot-table", + default="slo_topic_tx_hot", + type=str, + help="Hot-key contention table used to induce TLI", + ) + parser.add_argument( + "--tli-hot-keys", + default=4, + type=int, + help="Number of shared hot rows the transactions contend on (induces TLI)", + ) + + def make_topic_create_parser(subparsers): topic_create_parser = subparsers.add_parser("topic-create", help="Create topic with consumer") add_common_options(topic_create_parser) @@ -104,6 +126,7 @@ def make_topic_create_parser(subparsers): topic_create_parser.add_argument("--path", default="/local/slo_topic", type=str, help="Topic path") topic_create_parser.add_argument("--consumer", default="slo_consumer", type=str, help="Topic consumer name") topic_create_parser.add_argument("--partitions-count", default=8, type=int, help="Partition count") + add_topic_tx_options(topic_create_parser) def make_topic_run_parser(subparsers): @@ -131,6 +154,14 @@ def make_topic_run_parser(subparsers): ) topic_parser.add_argument("--message-size", default=100, type=int, help="Topic message size in bytes") + add_topic_tx_options(topic_parser) + topic_parser.add_argument( + "--messages-per-tx", + default=10, + type=int, + help="Messages written/read per transaction (amortizes tx overhead) in the *-topic-tx workloads", + ) + topic_parser.add_argument("--time", default=10, type=int, help="Time to run in seconds") topic_parser.add_argument( "--shutdown-time", @@ -152,6 +183,7 @@ def make_topic_cleanup_parser(subparsers): add_common_options(topic_cleanup_parser) topic_cleanup_parser.add_argument("--path", default="/local/slo_topic", type=str, help="Topic path") + add_topic_tx_options(topic_cleanup_parser) def get_root_parser(): diff --git a/tests/slo/src/runners/topic_runner.py b/tests/slo/src/runners/topic_runner.py index c9a8bdaa0..0a43ef3b5 100644 --- a/tests/slo/src/runners/topic_runner.py +++ b/tests/slo/src/runners/topic_runner.py @@ -1,21 +1,73 @@ import time +from os import path -from core.metrics import create_metrics +from core.metrics import WORKLOAD, create_metrics from jobs.async_topic_jobs import AsyncTopicJobManager +from jobs.async_topic_tx_jobs import AsyncTopicTxJobManager from jobs.topic_jobs import TopicJobManager +from jobs.topic_tx_jobs import TopicTxJobManager import ydb import ydb.aio from .base import BaseRunner +SINK_TABLE_DDL = """ +CREATE TABLE IF NOT EXISTS `{}` ( + writer_id Uint64, + seqno Uint64, + write_ts_ns Uint64, + received_at Timestamp, + PRIMARY KEY (writer_id, seqno) +); +""" + +HOT_TABLE_DDL = """ +CREATE TABLE IF NOT EXISTS `{}` ( + id Uint64, + val Uint64, + PRIMARY KEY (id) +); +""" + class TopicRunner(BaseRunner): @property def prefix(self) -> str: return "topic" + @staticmethod + def _is_tx_workload() -> bool: + return str(WORKLOAD).endswith("topic-tx") + + @staticmethod + def _tx_table_names(args): + return path.join(args.db, args.sink_table), path.join(args.db, args.hot_table) + def create(self, args): + self._create_topic(args) + if self._is_tx_workload(): + self._create_tx_tables(args) + + def _create_tx_tables(self, args): + sink_table, hot_table = self._tx_table_names(args) + hot_keys = max(1, int(getattr(args, "tli_hot_keys", 4))) + self.logger.info("Creating tx tables: sink=%s hot=%s (hot_keys=%d)", sink_table, hot_table, hot_keys) + + assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." + with ydb.QuerySessionPool(self.driver) as pool: + pool.execute_with_retries(SINK_TABLE_DDL.format(sink_table)) + pool.execute_with_retries(HOT_TABLE_DDL.format(hot_table)) + # Seed the hot rows so the producer/consumer read-modify-write always + # hits an existing row (a strong, deterministic optimistic lock). + for i in range(hot_keys): + pool.execute_with_retries( + "DECLARE $id AS Uint64; UPSERT INTO `{}` (id, val) VALUES ($id, 0);".format(hot_table), + {"$id": (i, ydb.PrimitiveType.Uint64)}, + ) + self.logger.info("Tx tables ready") + + def _create_topic(self, args): assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." retry_no = 0 while retry_no < 3: @@ -74,7 +126,11 @@ def run(self, args): self.logger.info("Starting topic SLO tests") - job_manager = TopicJobManager(self.driver, args, metrics) + if self._is_tx_workload(): + sink_table, hot_table = self._tx_table_names(args) + job_manager = TopicTxJobManager(self.driver, args, metrics, sink_table, hot_table) + else: + job_manager = TopicJobManager(self.driver, args, metrics) job_manager.run_tests() self.logger.info("Topic SLO tests completed") @@ -90,7 +146,11 @@ async def run_async(self, args): self.logger.info("Starting async topic SLO tests") # Use async driver for topic operations - job_manager = AsyncTopicJobManager(self.driver, args, metrics) + if self._is_tx_workload(): + sink_table, hot_table = self._tx_table_names(args) + job_manager = AsyncTopicTxJobManager(self.driver, args, metrics, sink_table, hot_table) + else: + job_manager = AsyncTopicJobManager(self.driver, args, metrics) await job_manager.run_tests() self.logger.info("Async topic SLO tests completed") @@ -111,3 +171,13 @@ def cleanup(self, args): else: self.logger.error("Failed to drop topic: %s", e) raise + + if self._is_tx_workload(): + sink_table, hot_table = self._tx_table_names(args) + with ydb.QuerySessionPool(self.driver) as pool: + for table in (sink_table, hot_table): + try: + pool.execute_with_retries("DROP TABLE `{}`;".format(table)) + self.logger.info("Table dropped: %s", table) + except ydb.Error as e: + self.logger.info("Table not dropped (%s): %s", table, e) diff --git a/tests/slo/thresholds-topic.yaml b/tests/slo/thresholds-topic.yaml index c90edaead..71098088b 100644 --- a/tests/slo/thresholds-topic.yaml +++ b/tests/slo/thresholds-topic.yaml @@ -30,3 +30,7 @@ metrics: direction: neutral - name: topic_e2e_latency_p99_ms direction: neutral + # TLI aborts are the штатный optimistic-lock mechanism we induce on purpose; + # visible in the report but never a regression signal. + - name: topic_tx_tli + direction: neutral From 2098ac7ef0eed63cc13b78d71a8b82556a081966 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Fri, 3 Jul 2026 12:43:16 +0300 Subject: [PATCH 2/3] SLO: drop artificial TLI from topic-tx workloads Remove the hot-row contention table, the tli_hot_keys/hot_table options, and the topic_tx_tli metric. The scenario keeps its point: a transactional topic write (tx_writer) and a transactional topic->sink-table consumer (receive_batch_with_tx + upsert), exactly-once under chaos. Single sink table. --- tests/slo/README.md | 31 ++++---- tests/slo/TOPIC_TX_SCENARIO.md | 89 +++++++++------------- tests/slo/docker-entrypoint.sh | 5 +- tests/slo/metrics-topic.yaml | 12 --- tests/slo/src/core/metrics.py | 11 --- tests/slo/src/jobs/async_topic_tx_jobs.py | 63 ++++------------ tests/slo/src/jobs/topic_tx_jobs.py | 90 +++++------------------ tests/slo/src/options.py | 12 --- tests/slo/src/runners/topic_runner.py | 46 ++++-------- tests/slo/thresholds-topic.yaml | 4 - 10 files changed, 94 insertions(+), 269 deletions(-) diff --git a/tests/slo/README.md b/tests/slo/README.md index 42fa194c5..5aa1e6cdc 100644 --- a/tests/slo/README.md +++ b/tests/slo/README.md @@ -273,25 +273,22 @@ Each message carries `writer_id:seqno:write_ts_ns:` followed by padding to the c ### 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). +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)` 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 + 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`, `--tli-hot-keys`, `--sink-table`, `--hot-table`. +`--messages-per-tx`, `--sink-table`. ## Collected metrics - `oks` - amount of OK requests diff --git a/tests/slo/TOPIC_TX_SCENARIO.md b/tests/slo/TOPIC_TX_SCENARIO.md index 5a4ffeb31..43ba9e672 100644 --- a/tests/slo/TOPIC_TX_SCENARIO.md +++ b/tests/slo/TOPIC_TX_SCENARIO.md @@ -1,73 +1,63 @@ -# Topic-tx SLO scenario — both-ends transactional topic ↔ table pipeline +# 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 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. +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 ``` - contention table (few hot rows) per-ref topic: /Root/testdb/slo_topic_tx_ - ┌─────────────────────────────┐ ┌──────────────────────────────────────────────┐ - │ 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 + producer tx (transactional topic write) per-ref topic: /Root/testdb/slo_topic_tx_ + │ 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 : ` (same codec as the plain topic scenario, `jobs/topic_payload.py`). -## Flow (happy path + chaos + TLI) +## Flow (happy path + chaos) ```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 + Note over W: retry_tx: tx_writer.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 + S-->>W: topic write committed Note right of W: seqno advances only on commit (no gap) - else Aborted (TLI) or Unavailable (chaos) + 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->>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 + 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 @@ -77,42 +67,29 @@ sequenceDiagram | Signal | Meaning | Expectation | |--------|---------|-------------| -| `topic_lost_errors` | forward gap in a producer's seqno (real loss) | **0** — hard-fails the run, under chaos **and** TLI | +| `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_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 +## Why it stays correct under chaos -- **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) +- **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 - (TLI/chaos) re-reads the same messages from the last committed offset and + (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. +- **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 diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh index 30961e2ed..cff86205f 100755 --- a/tests/slo/docker-entrypoint.sh +++ b/tests/slo/docker-entrypoint.sh @@ -44,11 +44,10 @@ if [ "$PREFIX" = "topic" ]; then SAFE_REF=$(printf '%s' "$REF_RAW" | tr -c 'a-zA-Z0-9_' '_') 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. + # 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="${EXTRA_ARGS} --hot-table slo_topic_tx_hot_${SAFE_REF}" ;; *) EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}" diff --git a/tests/slo/metrics-topic.yaml b/tests/slo/metrics-topic.yaml index 3795a195e..f1fd913d2 100644 --- a/tests/slo/metrics-topic.yaml +++ b/tests/slo/metrics-topic.yaml @@ -49,15 +49,3 @@ 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 diff --git a/tests/slo/src/core/metrics.py b/tests/slo/src/core/metrics.py index 59dc193b3..313e1683d 100644 --- a/tests/slo/src/core/metrics.py +++ b/tests/slo/src/core/metrics.py @@ -85,10 +85,6 @@ 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: @@ -206,10 +202,6 @@ 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 = {} @@ -301,9 +293,6 @@ 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: """ diff --git a/tests/slo/src/jobs/async_topic_tx_jobs.py b/tests/slo/src/jobs/async_topic_tx_jobs.py index 6a2dde3b1..5d38ac816 100644 --- a/tests/slo/src/jobs/async_topic_tx_jobs.py +++ b/tests/slo/src/jobs/async_topic_tx_jobs.py @@ -9,78 +9,44 @@ from .async_topic_jobs import AsyncTopicJobManager from .topic_payload import decode_payload, encode_payload -from .topic_tx_jobs import HOT_SELECT_TEMPLATE, HOT_UPSERT_TEMPLATE, SINK_UPSERT_TEMPLATE +from .topic_tx_jobs import SINK_UPSERT_TEMPLATE logger = logging.getLogger(__name__) class AsyncTopicTxJobManager(AsyncTopicJobManager): - """Async mirror of :class:`TopicTxJobManager` (both-ends transactional - topic <-> table pipeline). See that class for the scenario description.""" + """Async mirror of :class:`TopicTxJobManager` (transactional topic <-> table + pipeline). See that class for the scenario description.""" - def __init__(self, driver, args, metrics, sink_table, hot_table): + def __init__(self, driver, args, metrics, sink_table): super().__init__(driver, args, metrics) self.sink_table = sink_table - self.hot_table = hot_table self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) - self.hot_keys = max(1, int(getattr(self.args, "tli_hot_keys", 4))) - self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) - self._hot_select_query = HOT_SELECT_TEMPLATE.format(hot_table) - self._hot_upsert_query = HOT_UPSERT_TEMPLATE.format(hot_table) - - self.retry_settings = ydb.RetrySettings(idempotent=True, on_ydb_error_callback=self._on_ydb_error) - - async def run_tests(self): - # Emit the TLI counter at 0 up front so it shows as 0 (not a missing - # series) on a clean run; the delivery counters are emitted by super(). - self.metrics.inc_tli(0) - await super().run_tests() - - def _on_ydb_error(self, e): - if isinstance(e, ydb.Aborted): - self.metrics.inc_tli() - logger.debug("TLI abort (will retry): %s", e) - - # --- shared query helpers ------------------------------------------------- - - async def _exec(self, tx, query, params): - rows = [] - async with await tx.execute(query, params, commit_tx=False) as result_sets: - async for result_set in result_sets: - rows.extend(result_set.rows) - return rows - - async def _bump_hot(self, tx, hot_id): - # Read-modify-write on a shared hot row: the read takes an optimistic - # lock, so concurrent writers/readers on the same key conflict -> TLI. - rows = await self._exec(tx, self._hot_select_query, {"$id": (hot_id, ydb.PrimitiveType.Uint64)}) - cur = rows[0]["val"] if rows and rows[0]["val"] is not None else 0 - await self._exec( - tx, - self._hot_upsert_query, - {"$id": (hot_id, ydb.PrimitiveType.Uint64), "$val": (cur + 1, ydb.PrimitiveType.Uint64)}, - ) + self.retry_settings = ydb.RetrySettings(idempotent=True) + + # --- shared query helper -------------------------------------------------- async def _upsert_sink(self, tx, decoded): writer_id, seqno, write_ts_ns = decoded - await self._exec( - tx, + 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 (table -> topic) ------------------------------------------- + # --- 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}" - hot_id = writer_id % self.hot_keys 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 @@ -105,7 +71,6 @@ async def _run_topic_writes(self, writer_id, limiter): break async def callee(tx): - await self._bump_hot(tx, hot_id) tx_writer = self.aio_driver.topic_client.tx_writer( tx, self.args.path, @@ -136,7 +101,6 @@ async def callee(tx): async def _run_topic_reads(self, reader_id): start_time = time.time() read_timeout = self.args.read_timeout / 1000 - hot_id = reader_id % self.hot_keys logger.info("Start async topic tx reader %s", reader_id) while time.time() - start_time < self.args.time: @@ -154,7 +118,6 @@ async def callee(tx): ) if batch is None: return - await self._bump_hot(tx, hot_id) for msg in batch.messages: decoded = decode_payload(msg.data) if decoded is not None: diff --git a/tests/slo/src/jobs/topic_tx_jobs.py b/tests/slo/src/jobs/topic_tx_jobs.py index 36dcb13ae..175969832 100644 --- a/tests/slo/src/jobs/topic_tx_jobs.py +++ b/tests/slo/src/jobs/topic_tx_jobs.py @@ -18,97 +18,48 @@ VALUES ($writer_id, $seqno, $write_ts_ns, CurrentUtcTimestamp()); """ -HOT_SELECT_TEMPLATE = """ -DECLARE $id AS Uint64; -SELECT val FROM `{}` WHERE id = $id; -""" - -HOT_UPSERT_TEMPLATE = """ -DECLARE $id AS Uint64; -DECLARE $val AS Uint64; -UPSERT INTO `{}` (id, val) VALUES ($id, $val); -""" - class TopicTxJobManager(TopicJobManager): - """Both-ends transactional topic <-> table pipeline (sync). - - Producer: a ``tx_writer`` writes a batch of messages to the topic while the - same transaction reads-and-bumps a hot counter row (table -> topic). - Consumer: ``receive_batch_with_tx`` reads a batch while the same transaction - UPSERTs each message into a sink table and bumps a hot counter row - (topic -> table). The commit advances the topic offset and persists the - table rows atomically -> exactly-once. - - The shared hot rows (``--tli-hot-keys`` of them) create optimistic-lock - contention, so transactions occasionally fail with TLI (``ydb.Aborted``); - the tx retry loop absorbs them and delivery stays exactly-once. TLI aborts - are counted via ``RetrySettings.on_ydb_error_callback``. + """Transactional topic <-> table pipeline (sync). + + Producer: a ``tx_writer`` writes a batch of messages to the topic inside a + transaction (a transactional topic write). Consumer: + ``receive_batch_with_tx`` reads a batch while the same transaction UPSERTs + each message into a sink table keyed by ``(writer_id, seqno)``. The commit + advances the topic offset and persists the sink rows atomically -> a + rolled-back tx re-reads and re-UPSERTs the same keys (idempotent), so the + table receives every message exactly once, even under chaos. """ - def __init__(self, driver, args, metrics, sink_table, hot_table): + def __init__(self, driver, args, metrics, sink_table): super().__init__(driver, args, metrics) self.sink_table = sink_table - self.hot_table = hot_table self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) - self.hot_keys = max(1, int(getattr(self.args, "tli_hot_keys", 4))) - self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) - self._hot_select_query = HOT_SELECT_TEMPLATE.format(hot_table) - self._hot_upsert_query = HOT_UPSERT_TEMPLATE.format(hot_table) - - self.retry_settings = ydb.RetrySettings(idempotent=True, on_ydb_error_callback=self._on_ydb_error) - - def run_tests(self): - # Emit the TLI counter at 0 up front so it shows as 0 (not a missing - # series) on a clean run; the delivery counters are emitted by super(). - self.metrics.inc_tli(0) - super().run_tests() - - def _on_ydb_error(self, e): - if isinstance(e, ydb.Aborted): - self.metrics.inc_tli() - logger.debug("TLI abort (will retry): %s", e) - - # --- shared query helpers ------------------------------------------------- - - def _exec(self, tx, query, params): - rows = [] - with tx.execute(query, params, commit_tx=False) as result_sets: - for result_set in result_sets: - rows.extend(result_set.rows) - return rows - - def _bump_hot(self, tx, hot_id): - # Read-modify-write on a shared hot row: the read takes an optimistic - # lock, so concurrent writers/readers on the same key conflict -> TLI. - rows = self._exec(tx, self._hot_select_query, {"$id": (hot_id, ydb.PrimitiveType.Uint64)}) - cur = rows[0]["val"] if rows and rows[0]["val"] is not None else 0 - self._exec( - tx, - self._hot_upsert_query, - {"$id": (hot_id, ydb.PrimitiveType.Uint64), "$val": (cur + 1, ydb.PrimitiveType.Uint64)}, - ) + self.retry_settings = ydb.RetrySettings(idempotent=True) + + # --- shared query helper -------------------------------------------------- def _upsert_sink(self, tx, decoded): writer_id, seqno, write_ts_ns = decoded - self._exec( - tx, + with 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: + for _ in result_sets: + pass - # --- producer (table -> topic) ------------------------------------------- + # --- producer (transactional topic write) -------------------------------- 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}" - hot_id = writer_id % self.hot_keys logger.info("Start topic tx writer %s (partition %s, producer %s)", writer_id, partition_id, producer_id) # `seqno` lives across recreations and advances only after a successful @@ -134,7 +85,6 @@ def _run_topic_writes(self, writer_id, limiter): break def callee(tx): - self._bump_hot(tx, hot_id) tx_writer = self.driver.topic_client.tx_writer( tx, self.args.path, @@ -165,7 +115,6 @@ def callee(tx): def _run_topic_reads(self, reader_id): start_time = time.time() read_timeout = self.args.read_timeout / 1000 - hot_id = reader_id % self.hot_keys logger.info("Start topic tx reader %s", reader_id) while time.time() - start_time < self.args.time: @@ -183,7 +132,6 @@ def callee(tx): ) if batch is None: return - self._bump_hot(tx, hot_id) for msg in batch.messages: decoded = decode_payload(msg.data) if decoded is not None: diff --git a/tests/slo/src/options.py b/tests/slo/src/options.py index 7186dc977..7cd5e0347 100644 --- a/tests/slo/src/options.py +++ b/tests/slo/src/options.py @@ -105,18 +105,6 @@ def add_topic_tx_options(parser): type=str, help="Sink table for the transactional consumer (exactly-once target)", ) - parser.add_argument( - "--hot-table", - default="slo_topic_tx_hot", - type=str, - help="Hot-key contention table used to induce TLI", - ) - parser.add_argument( - "--tli-hot-keys", - default=4, - type=int, - help="Number of shared hot rows the transactions contend on (induces TLI)", - ) def make_topic_create_parser(subparsers): diff --git a/tests/slo/src/runners/topic_runner.py b/tests/slo/src/runners/topic_runner.py index 0a43ef3b5..5d717f786 100644 --- a/tests/slo/src/runners/topic_runner.py +++ b/tests/slo/src/runners/topic_runner.py @@ -22,14 +22,6 @@ ); """ -HOT_TABLE_DDL = """ -CREATE TABLE IF NOT EXISTS `{}` ( - id Uint64, - val Uint64, - PRIMARY KEY (id) -); -""" - class TopicRunner(BaseRunner): @property @@ -41,8 +33,8 @@ def _is_tx_workload() -> bool: return str(WORKLOAD).endswith("topic-tx") @staticmethod - def _tx_table_names(args): - return path.join(args.db, args.sink_table), path.join(args.db, args.hot_table) + def _tx_sink_table(args): + return path.join(args.db, args.sink_table) def create(self, args): self._create_topic(args) @@ -50,22 +42,13 @@ def create(self, args): self._create_tx_tables(args) def _create_tx_tables(self, args): - sink_table, hot_table = self._tx_table_names(args) - hot_keys = max(1, int(getattr(args, "tli_hot_keys", 4))) - self.logger.info("Creating tx tables: sink=%s hot=%s (hot_keys=%d)", sink_table, hot_table, hot_keys) + sink_table = self._tx_sink_table(args) + self.logger.info("Creating tx sink table: %s", sink_table) assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." with ydb.QuerySessionPool(self.driver) as pool: pool.execute_with_retries(SINK_TABLE_DDL.format(sink_table)) - pool.execute_with_retries(HOT_TABLE_DDL.format(hot_table)) - # Seed the hot rows so the producer/consumer read-modify-write always - # hits an existing row (a strong, deterministic optimistic lock). - for i in range(hot_keys): - pool.execute_with_retries( - "DECLARE $id AS Uint64; UPSERT INTO `{}` (id, val) VALUES ($id, 0);".format(hot_table), - {"$id": (i, ydb.PrimitiveType.Uint64)}, - ) - self.logger.info("Tx tables ready") + self.logger.info("Tx sink table ready") def _create_topic(self, args): assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." @@ -127,8 +110,7 @@ def run(self, args): self.logger.info("Starting topic SLO tests") if self._is_tx_workload(): - sink_table, hot_table = self._tx_table_names(args) - job_manager = TopicTxJobManager(self.driver, args, metrics, sink_table, hot_table) + job_manager = TopicTxJobManager(self.driver, args, metrics, self._tx_sink_table(args)) else: job_manager = TopicJobManager(self.driver, args, metrics) job_manager.run_tests() @@ -147,8 +129,7 @@ async def run_async(self, args): # Use async driver for topic operations if self._is_tx_workload(): - sink_table, hot_table = self._tx_table_names(args) - job_manager = AsyncTopicTxJobManager(self.driver, args, metrics, sink_table, hot_table) + job_manager = AsyncTopicTxJobManager(self.driver, args, metrics, self._tx_sink_table(args)) else: job_manager = AsyncTopicJobManager(self.driver, args, metrics) await job_manager.run_tests() @@ -173,11 +154,10 @@ def cleanup(self, args): raise if self._is_tx_workload(): - sink_table, hot_table = self._tx_table_names(args) + sink_table = self._tx_sink_table(args) with ydb.QuerySessionPool(self.driver) as pool: - for table in (sink_table, hot_table): - try: - pool.execute_with_retries("DROP TABLE `{}`;".format(table)) - self.logger.info("Table dropped: %s", table) - except ydb.Error as e: - self.logger.info("Table not dropped (%s): %s", table, e) + try: + pool.execute_with_retries("DROP TABLE `{}`;".format(sink_table)) + self.logger.info("Table dropped: %s", sink_table) + except ydb.Error as e: + self.logger.info("Table not dropped (%s): %s", sink_table, e) diff --git a/tests/slo/thresholds-topic.yaml b/tests/slo/thresholds-topic.yaml index 71098088b..c90edaead 100644 --- a/tests/slo/thresholds-topic.yaml +++ b/tests/slo/thresholds-topic.yaml @@ -30,7 +30,3 @@ metrics: direction: neutral - name: topic_e2e_latency_p99_ms direction: neutral - # TLI aborts are the штатный optimistic-lock mechanism we induce on purpose; - # visible in the report but never a regression signal. - - name: topic_tx_tli - direction: neutral From d82edb6ead66265eaa535363d1106fdf0b94e303 Mon Sep 17 00:00:00 2001 From: Oleg Ovcharuk Date: Fri, 3 Jul 2026 14:03:02 +0300 Subject: [PATCH 3/3] SLO: don't count idle receive-timeouts as tx read failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit read_availability was ~90% (async) not because reads failed but because a receive_batch_with_tx that blocks (idle reader, or the SDK transparently reconnecting under chaos) hit our wait_for timeout, which is not a ydb.Error so the tx retrier never retried it — and we wrongly recorded it as a failed read. Add metrics.cancel(): a timeout or an empty batch now balances the pending counter without a success/failure, so read_availability reflects the success rate of actual transactional reads (which the retrier drives ~100%). Real outages remain visible via write_availability / topic_delivered_rps. --- tests/slo/src/core/metrics.py | 11 +++++++++++ tests/slo/src/jobs/async_topic_tx_jobs.py | 20 +++++++++++++------- tests/slo/src/jobs/topic_tx_jobs.py | 20 +++++++++++++------- 3 files changed, 37 insertions(+), 14 deletions(-) diff --git a/tests/slo/src/core/metrics.py b/tests/slo/src/core/metrics.py index 313e1683d..58da8ff9a 100644 --- a/tests/slo/src/core/metrics.py +++ b/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(): diff --git a/tests/slo/src/jobs/async_topic_tx_jobs.py b/tests/slo/src/jobs/async_topic_tx_jobs.py index 5d38ac816..b127e096d 100644 --- a/tests/slo/src/jobs/async_topic_tx_jobs.py +++ b/tests/slo/src/jobs/async_topic_tx_jobs.py @@ -127,19 +127,25 @@ async def callee(tx): ts = self.metrics.start((OP_TYPE_READ,)) try: await pool.retry_tx_async(callee, retry_settings=self.retry_settings) - self.metrics.stop((OP_TYPE_READ,), ts) - except asyncio.TimeoutError as e: - # Starved reader (outage/stall): a visible read - # failure, not a silent wait. - self.metrics.stop((OP_TYPE_READ,), ts, error=e) + 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 - # Validate only after commit: these offsets/rows are - # durable, so the seqnos are exactly-once delivered. + 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: diff --git a/tests/slo/src/jobs/topic_tx_jobs.py b/tests/slo/src/jobs/topic_tx_jobs.py index 175969832..06313fcb8 100644 --- a/tests/slo/src/jobs/topic_tx_jobs.py +++ b/tests/slo/src/jobs/topic_tx_jobs.py @@ -141,19 +141,25 @@ def callee(tx): ts = self.metrics.start((OP_TYPE_READ,)) try: pool.retry_tx_sync(callee, retry_settings=self.retry_settings) - self.metrics.stop((OP_TYPE_READ,), ts) - except TimeoutError as e: - # Starved reader (outage/stall): a visible read - # failure, not a silent wait. - self.metrics.stop((OP_TYPE_READ,), ts, error=e) + except 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 - # Validate only after commit: these offsets/rows are - # durable, so the seqnos are exactly-once delivered. + 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: