SLO: async-query + topic (sync/async) workloads, label-driven, with delivery/ordering validation (#851)

Author: vgvoleg

.github/workflows/slo-report.yml

@@ -12,11 +12,15 @@ on:
 
 jobs:
   publish-slo-report:
-    # SLO workflow runs on every PR event, but the job inside is gated
-    # on the "SLO" label. Without the label the matrix is skipped and the
-    # workflow conclusion is "skipped" — in that case there is nothing
-    # to publish and no label to remove.
-    if: github.event.workflow_run.conclusion != 'skipped'
+    # Only act on runs that actually executed to completion. A "skipped"
+    # conclusion means the matrix was gated out (no "SLO" label) — nothing to
+    # publish. A "cancelled" conclusion means the run was superseded mid-flight
+    # (e.g. by `cancel-in-progress` when a new run started) — there is no real
+    # report and the label must NOT be stripped, otherwise it gets removed long
+    # before the actual workload finishes. Success/failure are genuine endings.
+    if: >-
+      github.event.workflow_run.conclusion == 'success' ||
+      github.event.workflow_run.conclusion == 'failure'
     name: Publish YDB SLO Report
     runs-on: ubuntu-latest
     permissions:
@@ -31,7 +35,12 @@ jobs:
           github_run_id: ${{ github.event.workflow_run.id }}
 
   remove-slo-label:
-    if: github.event.workflow_run.conclusion != 'skipped'
+    # Same gating as the report job: only consume the "SLO" label once a run has
+    # genuinely finished (success/failure). Removing it on a "cancelled" run
+    # would strip the label mid-flight when a run is superseded.
+    if: >-
+      github.event.workflow_run.conclusion == 'success' ||
+      github.event.workflow_run.conclusion == 'failure'
     name: Remove SLO Label
     runs-on: ubuntu-latest
     permissions:

.github/workflows/slo.yml

@@ -11,7 +11,14 @@ permissions:
 
 jobs:
   ydb-slo-action:
-    if: contains(github.event.pull_request.labels.*.name, 'SLO')
+    # On `labeled` events run only when the `SLO` label itself was just added —
+    # otherwise unrelated label changes (e.g. the AI-review bot toggling
+    # `ai_review_in_process` / `ai_reviewed`) would spawn a fresh run that
+    # cancels the in-progress one via `cancel-in-progress`. For the other
+    # trigger types keep gating on the `SLO` label being present.
+    if: >-
+      (github.event.action == 'labeled' && github.event.label.name == 'SLO') ||
+      (github.event.action != 'labeled' && contains(github.event.pull_request.labels.*.name, 'SLO'))
 
     name: Run YDB SLO Tests
     runs-on: "large-runner-python-sdk"
@@ -24,6 +31,16 @@ jobs:
             command: "--read-rps 1000 --write-rps 100"
           - name: sync-query
             command: "--read-rps 1000 --write-rps 100"
+          - name: async-query
+            command: "--read-rps 1000 --write-rps 100"
+          - name: sync-topic
+            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: async-topic
+            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
 
     concurrency:
       group: slo-${{ github.ref }}-${{ matrix.sdk.name }}
@@ -125,3 +142,10 @@ jobs:
           workload_baseline_ref: ${{ steps.baseline.outputs.ref }}
           workload_baseline_image: ydb-app-baseline
           workload_baseline_command: ${{ matrix.sdk.command }}
+          # Custom metrics (e.g. topic e2e latency / loss) merged on top of the
+          # action defaults; empty for workloads that don't set it.
+          metrics_yaml_path: ${{ matrix.sdk.metrics_yaml_path }}
+          # Per-scenario threshold overrides (topics make read_latency neutral).
+          # Ignored by action versions without ydb-slo-action#57; self-activates
+          # once per-scenario thresholds land in the consumed tag.
+          thresholds_yaml_path: ${{ matrix.sdk.thresholds_yaml_path }}

tests/slo/README.md

@@ -10,10 +10,23 @@ There are two workload types:
 - **Table SLO** - tests table operations (read/write)
 - **Topic SLO** - tests topic operations (publish/consume)
 
-### Implementations:
+### Implementations / labels:
 
-- `sync`
-- `async` (now unimplemented)
+The workload is selected by a single label (`WORKLOAD_NAME` env var, also the
+`sdk.name` matrix value in `.github/workflows/slo.yml`). The sync/async
+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 |
+
+> 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`.
 
 ### Usage:
 
@@ -246,16 +259,15 @@ Table have these fields:
 Primary key: `("object_hash", "object_id")`
 
 ### Topic workload
-When running `topic-run` command, the program creates three jobs: `readJob`, `writeJob`, `metricsJob`.
+When running `topic-run` (`sync-topic` / `async-topic`), the program creates `readJob`, `writeJob` and `metricsJob`, and additionally **validates end-to-end delivery and per-producer ordering** under chaos.
 
-- `readJob`    reads messages from topic using TopicReader and commits offsets
-- `writeJob`   generates and publishes messages to topic using TopicWriter
-- `metricsJob` periodically sends metrics to Prometheus
+- `writeJob` — each writer is pinned to a partition (`partition_id = i % partitions`) with a stable, ref-scoped `producer_id`, and publishes with `write_with_ack`. The seqno advances only on a successful ack (a failed write leaves no gap).
+- `readJob` — reads with a consumer, commits offsets, and demultiplexes messages by `writer_id`, tracking the next expected seqno per producer (shared across readers, since a partition can move between them on rebalance):
+  - a **forward gap** (a seqno past the expected one) is counted as **lost** — partition order is server-guaranteed, so a gap is real loss (fails the run via the `*_error*` threshold);
+  - a **backward** seqno (already seen) is a **duplicate** — reconnect redelivery; with producer-id dedup it should stay near zero (informational);
+  - **end-to-end latency** is `read_ts − write_ts` for the first delivery of each message (writer and reader share the process, so the timestamps are comparable).
 
-Messages contain:
-- Sequential message ID
-- Thread identifier
-- Configurable payload size (padded with 'x' characters)
+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.
 
 ## Collected metrics
 - `oks`      - amount of OK requests
@@ -266,6 +278,12 @@ Messages contain:
 
 Metrics are collected for both table operations (`read`, `write`) and topic operations (`read`, `write`).
 
+Topic workloads additionally emit (surfaced through `tests/slo/metrics-topic.yaml`, merged into the action metrics via `metrics_yaml_path`):
+- `topic_e2e_latency_p50_ms` / `_p99_ms` — write → read latency of a delivered message (the meaningful topic latency; the generic `read_latency` mostly reflects `receive_message` wait time, not read cost, so it is kept **informational** for topics via `tests/slo/thresholds-topic.yaml` — `direction: neutral` — while `write_latency` stays gated)
+- `topic_delivered_rps` — unique messages read back per second
+- `topic_lost_errors` — messages detected as lost (must stay 0)
+- `topic_duplicates` — redelivered messages (informational)
+
 > Note: with Prometheus OTLP receiver (no Pushgateway) counters/histograms are cumulative and cannot be reset to `0`.
 > If you need clean separation between runs, use distinct `REF`/`WORKLOAD` (and/or `SLO_INSTANCE_ID`) so each run writes into separate time series.
 

tests/slo/docker-entrypoint.sh

@@ -5,20 +5,23 @@
 # in parallel; both must be able to schema-prepare and then run.
 #
 # Inputs come from the env vars injected by the action:
-#   WORKLOAD_NAME       sync-table | sync-query | topic
+#   WORKLOAD_NAME       sync-table | sync-query | async-query | sync-topic | async-topic
 #   WORKLOAD_DURATION   run duration in seconds
 #   YDB_ENDPOINT        grpc://ydb:2136
 #   YDB_DATABASE        /Root/testdb
 #
+# The sync/async execution mode is derived from the WORKLOAD_NAME label itself
+# (an `async-*` label runs the async path); no extra flag is needed here.
+#
 # Anything passed after the script name is appended to the `*-run` command —
 # this is how tuning flags from `workload_current_command` (e.g. --read-rps)
 # reach the workload.
 
 set -e
 
 case "${WORKLOAD_NAME:-sync-query}" in
-    sync-table|sync-query) PREFIX=table ;;
-    topic) PREFIX=topic ;;
+    sync-table|sync-query|async-query) PREFIX=table ;;
+    topic|sync-topic|async-topic) PREFIX=topic ;;
     *)
         echo "Unknown WORKLOAD_NAME: ${WORKLOAD_NAME}" >&2
         exit 1
@@ -29,12 +32,25 @@ ENDPOINT="${YDB_ENDPOINT:-grpc://localhost:2136}"
 DATABASE="${YDB_DATABASE:-/local}"
 DURATION="${WORKLOAD_DURATION:-600}"
 
+# Topic paths must live under the database; derive one from $DATABASE so the
+# same image works against both local (/local) and CI (/Root/testdb) databases.
+# Scope the topic by ref so the current and baseline containers (same cluster,
+# run in parallel) don't share a topic — otherwise their readers/producers would
+# cross-contaminate delivery/ordering validation.
+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}"
+fi
+
 # Schema prep is idempotent at the SDK level for topics; for tables, a parallel
 # baseline container may race and fail with "already exists" — tolerate it.
-python ./tests/slo/src "${PREFIX}-create" "$ENDPOINT" "$DATABASE" \
+python ./tests/slo/src "${PREFIX}-create" "$ENDPOINT" "$DATABASE" $EXTRA_ARGS \
     || echo "WARN: ${PREFIX}-create exited non-zero (treated as already-prepared)" >&2
 
 exec python ./tests/slo/src \
     "${PREFIX}-run" "$ENDPOINT" "$DATABASE" \
     --time "$DURATION" \
+    $EXTRA_ARGS \
     "$@"

tests/slo/metrics-topic.yaml

@@ -0,0 +1,51 @@
+# Custom topic-workload metrics, merged on top of the action's default
+# deploy/metrics.yaml (see ydb-slo-action shared/metrics.ts mergeMetricConfigs).
+# Passed to the action via `metrics_yaml_path` for topic workloads only.
+#
+# Metric names are chosen to match the default threshold patterns
+# (deploy/thresholds.yaml):
+#   *_latency_*  -> lower_is_better
+#   *_rps        -> higher_is_better
+#   *_error*     -> warning_max 0.1 / critical_max 1.0  (fails the run if > 0)
+
+metrics:
+  - name: topic_e2e_latency_p50_ms
+    unit: ms
+    query: |
+      1000 * max by(ref) (
+        sdk_topic_e2e_latency_p50_seconds
+      )
+
+  - name: topic_e2e_latency_p99_ms
+    unit: ms
+    query: |
+      1000 * max by(ref) (
+        sdk_topic_e2e_latency_p99_seconds
+      )
+
+  - name: topic_delivered_rps
+    unit: msgs/s
+    query: |
+      sum by(ref) (
+        rate(sdk_topic_messages_delivered_total[5s])
+      )
+
+  # Lost messages must stay at zero even under chaos (at-least-once delivery).
+  # Named *_error* so the default threshold fails the run on any loss.
+  - name: topic_lost_errors
+    unit: msgs
+    query: |
+      sum by(ref) (
+        increase(sdk_topic_messages_lost_total[5s])
+      )
+    round: 1
+
+  # Duplicates are informational: reconnect redelivery can produce them; with
+  # producer-id dedup they should stay near zero.
+  - name: topic_duplicates
+    unit: msgs
+    query: |
+      sum by(ref) (
+        increase(sdk_topic_messages_duplicated_total[5s])
+      )
+    round: 1

tests/slo/src/core/metrics.py

@@ -68,6 +68,23 @@ def measure(self, labels):
         finally:
             self.stop(labels, start_ts, error=error)
 
+    # --- 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."""
+        return None
+
+    def inc_delivered(self, n: int = 1) -> None:
+        """Count messages successfully read back."""
+        return None
+
+    def inc_lost(self, n: int = 1) -> None:
+        """Count messages detected as lost (a forward gap in a producer's seqno)."""
+        return None
+
+    def inc_duplicated(self, n: int = 1) -> None:
+        """Count messages detected as duplicates (redelivery of an already-seen seqno)."""
+        return None
+
 
 class DummyMetrics(BaseMetrics):
     def start(self, labels) -> float:
@@ -164,8 +181,31 @@ def __init__(self, otlp_metrics_endpoint: str):
             for name, _ in self._PERCENTILES
         }
 
+        # Topic-workload metrics: end-to-end (write -> read) latency + delivery counters.
+        self._topic_e2e_gauges = {
+            name: self._meter.create_gauge(
+                name=f"sdk.topic.e2e.latency.{name}.seconds",
+                unit="s",
+                description=f"Topic end-to-end latency {name} computed over the last push window.",
+            )
+            for name, _ in self._PERCENTILES
+        }
+        self._topic_delivered = self._meter.create_counter(
+            name="sdk.topic.messages.delivered.total",
+            description="Total number of messages successfully read back from the topic.",
+        )
+        self._topic_lost = self._meter.create_counter(
+            name="sdk.topic.messages.lost.total",
+            description="Total number of messages detected as lost (forward gap in a producer's seqno).",
+        )
+        self._topic_duplicated = self._meter.create_counter(
+            name="sdk.topic.messages.duplicated.total",
+            description="Total number of messages detected as duplicates (redelivered seqno).",
+        )
+
         self._lock = threading.Lock()
         self._hdr: dict = {}
+        self._e2e_hdr = self._HdrHistogram(self._HDR_MIN_US, self._HDR_MAX_US, self._HDR_SIG_FIGS)
 
     def _get_hdr(self, op_type: str, op_status: str):
         key = (op_type, op_status)
@@ -223,14 +263,36 @@ def push(self) -> None:
                     self._latency_gauges[name].set(value_s, attributes=attrs)
             for hist in self._hdr.values():
                 hist.reset()
+
+            if self._e2e_hdr.get_total_count() > 0:
+                attrs = {"ref": REF}
+                for name, percentile in self._PERCENTILES:
+                    value_s = self._e2e_hdr.get_value_at_percentile(percentile) / 1_000_000
+                    self._topic_e2e_gauges[name].set(value_s, attributes=attrs)
+                self._e2e_hdr.reset()
         self._provider.force_flush()
 
     def reset(self) -> None:
         with self._lock:
             for hist in self._hdr.values():
                 hist.reset()
+            self._e2e_hdr.reset()
         self._provider.force_flush()
 
+    def record_e2e(self, seconds: float) -> None:
+        duration_us = min(max(int(seconds * 1_000_000), self._HDR_MIN_US), self._HDR_MAX_US)
+        with self._lock:
+            self._e2e_hdr.record_value(duration_us)
+
+    def inc_delivered(self, n: int = 1) -> None:
+        self._topic_delivered.add(int(n), attributes={"ref": REF})
+
+    def inc_lost(self, n: int = 1) -> None:
+        self._topic_lost.add(int(n), attributes={"ref": REF})
+
+    def inc_duplicated(self, n: int = 1) -> None:
+        self._topic_duplicated.add(int(n), attributes={"ref": REF})
+
 
 def _resolve_metrics_endpoint(cli_endpoint: Optional[str]) -> str:
     """