SLO: transactional topic<->table workloads (sync/async), exactly-once under TLI#854
Open
vgvoleg wants to merge 3 commits into
Open
SLO: transactional topic<->table workloads (sync/async), exactly-once under TLI#854vgvoleg wants to merge 3 commits into
vgvoleg wants to merge 3 commits into
Conversation
… 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.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #854 +/- ##
=======================================
Coverage 81.22% 81.22%
=======================================
Files 94 94
Lines 12111 12111
Branches 1184 1184
=======================================
Hits 9837 9837
+ Misses 1809 1808 -1
- Partials 465 466 +1
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
Adds new SLO workloads (sync-topic-tx / async-topic-tx) to validate an exactly-once transactional pipeline between YDB Topic and Table under chaos + deliberate TLI contention, integrating the new metric (topic_tx_tli) into SLO reporting and CI.
Changes:
- Implement transactional topic↔table producer/consumer job managers (sync + async) using
tx_writer/receive_batch_with_tx, plus runner/CLI wiring. - Add
topic_tx_tlimetric emission and expose it in metrics/threshold configs. - Document the scenario and update SLO README / workflow matrix / entrypoint ref-scoping for tx topics + tables.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
tests/slo/TOPIC_TX_SCENARIO.md |
New design doc describing the transactional topic↔table SLO scenario and invariants. |
tests/slo/thresholds-topic.yaml |
Adds topic_tx_tli as a neutral (informational) threshold metric. |
tests/slo/src/runners/topic_runner.py |
Creates/drops tx sink+hot tables for *-topic-tx workloads; selects tx vs non-tx job managers. |
tests/slo/src/options.py |
Adds CLI options for sink/hot tables, hot-key count, and messages-per-tx. |
tests/slo/src/jobs/topic_tx_jobs.py |
New sync transactional workload implementation (topic→table + table→topic in one tx). |
tests/slo/src/jobs/topic_jobs.py |
Refactors validation to reuse _record() for both classic and tx workloads. |
tests/slo/src/jobs/async_topic_tx_jobs.py |
New async mirror of the transactional workload. |
tests/slo/src/jobs/async_topic_jobs.py |
Refactors validation to reuse _record() for both classic and tx workloads. |
tests/slo/src/core/metrics.py |
Adds inc_tli() and OTLP counter sdk.topic.tx.tli.total. |
tests/slo/README.md |
Documents new *-topic-tx workload labels and describes behavior/options. |
tests/slo/metrics-topic.yaml |
Adds PromQL query/series for topic_tx_tli. |
tests/slo/docker-entrypoint.sh |
Adds ref-scoped topic + sink/hot table naming for *-topic-tx workloads. |
.github/workflows/slo.yml |
Runs new sync-topic-tx / async-topic-tx workloads in the SLO workflow matrix. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![robot-vibe-db[bot]](https://avatars.githubusercontent.com/in/3319454?s=60&v=4){kind=small}
There was a problem hiding this comment.
AI Review Summary
Verdict: ✅ No critical issues found
Critical issues
No critical issues found.
Other findings
- Nit | High: Non-English word "штатный" in documentation and config comments —
TOPIC_TX_SCENARIO.md:12,thresholds-topic.yaml:33,metrics-topic.yaml:55 - Minor | Low: Table names interpolated into SQL via
.format()(standard for YDB, but worth noting as a code smell for maintainability) —topic_tx_jobs.py:56-58,topic_runner.py:59-67
Overall assessment: This is a well-designed PR. The transactional semantics are carefully handled:
seqnoadvances only after successful commit (no gap on abort)- The
committedlist is reset on retry vianonlocal(correct for retry_tx) - Sink UPSERT is keyed by
(writer_id, seqno)making re-processing idempotent - Hot rows are correctly seeded and
_bump_hotgracefully handles missing rows - The
_recordrefactor in parent classes is clean and preserves existing behavior - Thread safety is correct: sync uses
_expected_lock, async relies on single event loop - The
_validate→_recordsplit correctly separates message decoding from delivery tracking, allowing tx subclasses to call_recordonly after commit - Ref-scoping of topics and tables prevents cross-contamination between current/baseline containers
- The entrypoint correctly sanitizes ref names and scopes all resources
This review was generated automatically. Critical issues require attention; other findings are advisory.
If this comment was useful, please give it a 👍 — it helps us improve the review bot.
| 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) |
There was a problem hiding this comment.
Severity: Minor
Confidence: Low
Table names are interpolated into SQL strings via .format(). This is the standard pattern in the existing SLO codebase (same approach in table_runner.py) and YDB doesn't support parameterized table names, so it's not exploitable here — especially since the inputs come from CLI args sanitized in docker-entrypoint.sh (via tr -c 'a-zA-Z0-9_' '_'). Just flagging for awareness since it's a new set of format-injected queries.
|
Analysis performed by claude, claude-opus-4-6. |
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.
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.
🌋 SLO Test Results🔴 7 workload(s) tested — 2 workload(s) exceeded failure thresholds
Threshold violations: async-query:
async-topic-tx:
sync-topic:
Generated by ydb-slo-action |
|
Analysis performed by claude, claude-opus-4-6. |
Comment on lines
+23
to
+31
| """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. |
Comment on lines
+71
to
+76
| 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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds two SLO workloads —
sync-topic-tx/async-topic-tx— that validate the both-ends transactional topic ↔ table pipeline (exactly-once) under chaos and TLI.tx_writer(tx)writes a batch to the topic while the same tx does a read-modify-write on a shared hot row. Commit persists the topic write + table update atomically;seqnoadvances only on commit (no gap on abort).receive_batch_with_tx(tx)reads a batch while the same tx UPSERTs each message into a sink table keyed by(writer_id, seqno)and bumps a hot row. Commit advances the offset + writes rows atomically → a rolled-back tx re-reads and re-UPSERTs the same keys (idempotent, exactly-once).--tli-hot-keys) →ydb.Aborted(locks invalidated).retry_tx_*absorbs it (counted viaon_ydb_error_callback→topic_tx_tli, neutral). Even retry-exhaustion recovers with no loss.Reuses the existing topic writer/reader machinery, payload codec, metrics, and reconnect/recreate loops. Design doc:
tests/slo/TOPIC_TX_SCENARIO.md.Verified (local
/local, 30s each)black/flake8/py_compileclean. No CHANGELOG (internal tests/CI).