From 8356da41f2f66e0983213878574af5f737270931 Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Wed, 14 Oct 2026 19:48:55 +0300 Subject: [PATCH 1/3] Fix KV SLO failure propagation and data checks --- .github/workflows/slo.yml | 8 +- test-requirements.txt | 1 + tests/slo/docker-entrypoint.sh | 31 ++++- tests/slo/src/jobs/async_table_jobs.py | 77 +++++++++-- tests/slo/src/jobs/base.py | 48 ++++++- tests/slo/src/jobs/table_jobs.py | 159 ++++++++++++++++----- tests/slo/src/root_runner.py | 70 ++++++---- tests/table/test_tx.py | 183 +++++++++++++++++++++++++ 8 files changed, 496 insertions(+), 81 deletions(-) mode change 100755 => 100644 tests/slo/docker-entrypoint.sh diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml index 8c066a3d4..0037e3dc1 100644 --- a/.github/workflows/slo.yml +++ b/.github/workflows/slo.yml @@ -118,11 +118,14 @@ jobs: "$GITHUB_WORKSPACE/sdk-current" # Build baseline: baseline SDK with the current workload runner - # (Dockerfile + tests/slo/), so the runner-side contract changes - # (entrypoint, metrics format) apply uniformly to both images. + # Keep its Dockerfile while copying runner-side contract changes. + cp "$GITHUB_WORKSPACE/sdk-baseline/tests/slo/Dockerfile" \ + "$RUNNER_TEMP/slo-baseline.Dockerfile" rm -rf "$GITHUB_WORKSPACE/sdk-baseline/tests/slo" cp -r "$GITHUB_WORKSPACE/sdk-current/tests/slo" \ "$GITHUB_WORKSPACE/sdk-baseline/tests/slo" + cp "$RUNNER_TEMP/slo-baseline.Dockerfile" \ + "$GITHUB_WORKSPACE/sdk-baseline/tests/slo/Dockerfile" docker build \ -f "$GITHUB_WORKSPACE/sdk-baseline/tests/slo/Dockerfile" \ -t "ydb-app-baseline" \ @@ -136,6 +139,7 @@ jobs: github_token: ${{ secrets.GITHUB_TOKEN }} workload_name: ${{ matrix.sdk.name }} workload_duration: "600" + fail_on_workload_error: true workload_current_ref: ${{ github.head_ref || github.ref_name }} workload_current_image: ydb-app-current workload_current_command: ${{ matrix.sdk.command }} diff --git a/test-requirements.txt b/test-requirements.txt index 304cee362..4fb6d19c6 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,4 +1,5 @@ attrs==21.2.0 +aiolimiter==1.1.0 bcrypt==3.2.0 black==25.11.0 build diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh old mode 100755 new mode 100644 index 55d933ec3..5dcfeb83d --- a/tests/slo/docker-entrypoint.sh +++ b/tests/slo/docker-entrypoint.sh @@ -44,8 +44,35 @@ if [ "$PREFIX" = "topic" ]; then 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. +if [ "$PREFIX" = "table" ]; then + TABLE_NAME="key_value_$(python -c 'import uuid; print(uuid.uuid4().hex)')" + python ./tests/slo/src table-create "$ENDPOINT" "$DATABASE" --table-name "$TABLE_NAME" + exec python -c ' +import signal +import subprocess +import sys + +table, endpoint, database, duration = sys.argv[1:5] +common = [sys.executable, "./tests/slo/src"] +workload = subprocess.Popen([ + *common, "table-run", endpoint, database, "--time", duration, + *sys.argv[5:], "--table-name", table, +]) +for signum in (signal.SIGTERM, signal.SIGINT, signal.SIGQUIT): + signal.signal(signum, lambda received, _frame: workload.send_signal(received)) +try: + status = workload.wait() +finally: + cleanup = subprocess.run([ + *common, "table-cleanup", endpoint, database, "--table-name", table, + ], check=False) + if cleanup.returncode: + print("WARN: table cleanup failed", file=sys.stderr) +sys.exit(status if status >= 0 else 128 - status) +' "$TABLE_NAME" "$ENDPOINT" "$DATABASE" "$DURATION" "$@" +fi + +# Topic schema preparation retains its existing behavior. python ./tests/slo/src "${PREFIX}-create" "$ENDPOINT" "$DATABASE" $EXTRA_ARGS \ || echo "WARN: ${PREFIX}-create exited non-zero (treated as already-prepared)" >&2 diff --git a/tests/slo/src/jobs/async_table_jobs.py b/tests/slo/src/jobs/async_table_jobs.py index 8c977cb9c..21ab5e0e0 100644 --- a/tests/slo/src/jobs/async_table_jobs.py +++ b/tests/slo/src/jobs/async_table_jobs.py @@ -2,6 +2,7 @@ import logging import time from random import randint +from uuid import uuid4 # `aiolimiter` is a runtime dependency (see `tests/slo/requirements.txt`). from aiolimiter import AsyncLimiter # pyright: ignore[reportMissingImports] @@ -12,7 +13,8 @@ import ydb.aio from .base import AsyncBaseJobManager -from .table_jobs import READ_QUERY_TEMPLATE, WRITE_QUERY_TEMPLATE +from .table_jobs import CONTROL_READ_QUERY, CONTROL_SCHEMA, READ_QUERY_TEMPLATE, WRITE_QUERY_TEMPLATE +from .table_jobs import check_control_result, control_row logger = logging.getLogger(__name__) @@ -33,12 +35,60 @@ async def run_tests(self): if self.workload_type != "async-query": raise ValueError(f"Unsupported async workload type: {self.workload_type}") + await self._run_control_checks() tasks = [ *self._run_query_read_jobs(), *self._run_query_write_jobs(), *self._run_metric_job(), ] - await asyncio.gather(*tasks) + await self._wait_jobs(tasks) + + async def _run_control_checks(self): + table = f"{self.table_name}_control_{uuid4().hex}" + write = WRITE_QUERY_TEMPLATE.format(table).replace("Digest::NumericHash($object_id)", "0") + types = ( + ydb.PrimitiveType.Uint64, + ydb.PrimitiveType.Utf8, + ydb.PrimitiveType.Double, + ydb.PrimitiveType.Timestamp, + ) + failure = None + async with ydb.aio.QuerySessionPool(self.aio_driver) as pool: + await pool.execute_with_retries(CONTROL_SCHEMA.format(table)) + try: + for generation in (1, 2): + expected = control_row(generation) + values = {f"${key}": value for key, value in expected.items() if key != "object_hash"} + values = {key: (value, kind) for (key, value), kind in zip(values.items(), types)} + for query, params, operation, timeout, checker in ( + (write, values, OP_TYPE_WRITE, self.args.write_timeout, None), + ( + CONTROL_READ_QUERY.format(table), + {}, + OP_TYPE_READ, + self.args.read_timeout, + lambda rows: check_control_result(rows, expected), + ), + ): + await self._execute_query( + pool, + query, + params, + ydb.BaseRequestSettings().with_timeout(timeout / 1000), + ydb.RetrySettings(idempotent=True, max_session_acquire_timeout=timeout / 1000), + (operation,), + check_result_cb=checker, + ) + except BaseException as error: + failure = error + raise + finally: + try: + await pool.execute_with_retries(f"DROP TABLE `{table}`;") + except BaseException: + if failure is None: + raise + logger.exception("Cannot clean up control table %s", table) def _run_query_read_jobs(self): logger.info("Start async query read jobs") @@ -118,7 +168,7 @@ async def _run_query_writes(self, query, limiter): logger.info("Stop async query write workload") - async def _execute_query(self, pool, query, params, request_settings, retry_setting, labels): + async def _execute_query(self, pool, query, params, request_settings, retry_setting, labels, check_result_cb=None): attempt = 0 error = None @@ -127,18 +177,29 @@ async def callee(session): attempt += 1 result_stream = await session.execute(query, parameters=params, settings=request_settings) - async for _ in result_stream: - pass + rows = [] + async for part in result_stream: + rows.extend(part.rows) + return rows ts = self.metrics.start(labels) try: - await pool.retry_operation_async(callee, retry_setting) + rows = await pool.retry_operation_async(callee, retry_setting) + if check_result_cb: + check_result_cb(rows) + return rows + except asyncio.CancelledError: + raise except ydb.Error as err: error = err + self.record_failure(err) logger.exception("[labels: %s] Cannot retry error:", labels) + raise except BaseException as err: error = err + self.record_failure(err) logger.exception("[labels: %s] Unexpected error:", labels) - - self.metrics.stop(labels, ts, attempts=attempt, error=error) + raise + finally: + self.metrics.stop(labels, ts, attempts=attempt, error=error) diff --git a/tests/slo/src/jobs/base.py b/tests/slo/src/jobs/base.py index 5c34443ec..694c3796e 100644 --- a/tests/slo/src/jobs/base.py +++ b/tests/slo/src/jobs/base.py @@ -17,8 +17,9 @@ class SyncRateLimiter: threads sharing the same instance. """ - def __init__(self, min_interval_s: float) -> None: + def __init__(self, min_interval_s: float, stop_event: threading.Event | None = None) -> None: self._min_interval_s = max(0.0, float(min_interval_s)) + self._stop_event = stop_event self._lock = threading.Lock() self._next_allowed_ts = 0.0 # monotonic timestamp @@ -37,7 +38,10 @@ def __enter__(self): # Sleep outside the lock so other threads can observe the schedule. if sleep_for > 0: - time.sleep(sleep_for) + if self._stop_event is None: + time.sleep(sleep_for) + elif self._stop_event.wait(sleep_for): + return self def __exit__(self, exc_type, exc, tb) -> bool: # Do not suppress exceptions. @@ -49,6 +53,26 @@ def __init__(self, driver, args, metrics): self.driver: ydb.Driver = driver self.args = args self.metrics = metrics + self._failure = None + self._failure_lock = threading.Lock() + self._stop = threading.Event() + + def record_failure(self, error): + with self._failure_lock: + if self._failure is None: + self._failure = error + self._stop.set() + + def raise_if_failed(self): + if self._failure is not None: + raise self._failure + + def _run_worker(self, worker, *args): + try: + worker(*args) + except BaseException as error: + self.record_failure(error) + logger.exception("Workload worker failed") @abstractmethod def run_tests(self): @@ -59,12 +83,12 @@ def _run_metric_job(self): return [] report_period_ms = max(1, int(self.args.report_period)) - limiter = SyncRateLimiter(min_interval_s=report_period_ms / 1000.0) + limiter = SyncRateLimiter(min_interval_s=report_period_ms / 1000.0, stop_event=self._stop) future = threading.Thread( name="slo_metrics_sender", - target=self._metric_sender, - args=(limiter, self.args.time), + target=self._run_worker, + args=(self._metric_sender, limiter, self.args.time), ) future.start() return [future] @@ -73,8 +97,10 @@ def _metric_sender(self, limiter, runtime): start_time = time.time() logger.info("Start push metrics") - while time.time() - start_time < runtime: + while not self._stop.is_set() and time.time() - start_time < runtime: with limiter: + if self._stop.is_set(): + break self.metrics.push() logger.info("Stop push metrics") @@ -91,6 +117,16 @@ class AsyncBaseJobManager(BaseJobManager): async def run_tests(self): pass + async def _wait_jobs(self, tasks): + try: + await asyncio.gather(*tasks) + except BaseException as error: + self.record_failure(error) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + self.raise_if_failed() + def _run_metric_job(self): if not getattr(self.args, "otlp_endpoint", None): return [] diff --git a/tests/slo/src/jobs/table_jobs.py b/tests/slo/src/jobs/table_jobs.py index 2452c7b78..e3aafb195 100644 --- a/tests/slo/src/jobs/table_jobs.py +++ b/tests/slo/src/jobs/table_jobs.py @@ -1,7 +1,9 @@ import logging import threading import time +from contextlib import nullcontext from random import randint +from uuid import uuid4 from core.generator import RowGenerator from core.metrics import OP_TYPE_READ, OP_TYPE_WRITE @@ -42,6 +44,7 @@ def __init__( request_settings, retry_settings, check_result_cb=None, + on_error=None, ): self.pool = pool self.query = query @@ -51,6 +54,36 @@ def __init__( self.request_settings = request_settings self.retry_settings = retry_settings self.check_result_cb = check_result_cb + self.on_error = on_error + + +CONTROL_SCHEMA = """ +CREATE TABLE `{}` ( + object_hash Uint64, object_id Uint64, payload_str Utf8, + payload_double Double, payload_timestamp Timestamp, + PRIMARY KEY (object_hash, object_id) +); +""" +CONTROL_READ_QUERY = """ +SELECT object_hash, object_id, payload_str, payload_double, + CAST(payload_timestamp AS Uint64) AS payload_timestamp +FROM `{}` WHERE object_hash = 0 AND object_id = 0; +""" + + +def control_row(generation): + return dict( + object_hash=0, + object_id=0, + payload_str=f"generation:{generation}", + payload_double=generation + 0.25, + payload_timestamp=946684800000000 + generation, + ) + + +def check_control_result(rows, expected): + if [dict(row) for row in rows] != [expected]: + raise RuntimeError(f"KV control data mismatch: expected {expected!r}, got {rows!r}") def execute_query(params: RequestParams): @@ -67,23 +100,30 @@ def transaction(session): commit_tx=True, settings=params.request_settings, ) - if params.check_result_cb: - params.check_result_cb(result) - - return result + with result if hasattr(result, "__enter__") else nullcontext(result): + return [row for part in result for row in part.rows] ts = params.metrics.start(params.labels) try: - params.pool.retry_operation_sync(transaction, retry_settings=params.retry_settings) + result = params.pool.retry_operation_sync(transaction, retry_settings=params.retry_settings) + if params.check_result_cb: + params.check_result_cb(result) + return result except ydb.Error as err: error = err + if params.on_error is not None: + params.on_error(err) logger.exception("[labels: %s] Cannot retry error:", params.labels) + raise except BaseException as err: error = err + if params.on_error is not None: + params.on_error(err) logger.exception("[labels: %s] Unexpected error:", params.labels) - - params.metrics.stop(params.labels, ts, attempts=attempt, error=error) + raise + finally: + params.metrics.stop(params.labels, ts, attempts=attempt, error=error) class TableJobManager(BaseJobManager): @@ -97,6 +137,7 @@ def __init__(self, driver, args, metrics, table_name, max_id): self.workload_type = WORKLOAD def run_tests(self): + self._run_control_checks() if self.workload_type == "sync-table": futures = [ *self._run_table_read_jobs(), @@ -114,6 +155,66 @@ def run_tests(self): for future in futures: future.join() + self.raise_if_failed() + + def _run_control_checks(self): + table = f"{self.table_name}_control_{uuid4().hex}" + is_table = self.workload_type == "sync-table" + pool_type = ydb.SessionPool if is_table else ydb.QuerySessionPool + write = WRITE_QUERY_TEMPLATE.format(table).replace("Digest::NumericHash($object_id)", "0") + read = CONTROL_READ_QUERY.format(table) + failure = None + with ydb.SessionPool(self.driver) as schema_pool: + schema_pool.retry_operation_sync(lambda session: session.execute_scheme(CONTROL_SCHEMA.format(table))) + try: + if is_table: + write = schema_pool.retry_operation_sync(lambda session: session.prepare(write)) + read = schema_pool.retry_operation_sync(lambda session: session.prepare(read)) + with pool_type(self.driver) as pool: + for generation in (1, 2): + expected = control_row(generation) + values = {f"${key}": value for key, value in expected.items() if key != "object_hash"} + if not is_table: + types = ( + ydb.PrimitiveType.Uint64, + ydb.PrimitiveType.Utf8, + ydb.PrimitiveType.Double, + ydb.PrimitiveType.Timestamp, + ) + values = {key: (value, kind) for (key, value), kind in zip(values.items(), types)} + for query, params, operation, timeout, checker in ( + (write, values, OP_TYPE_WRITE, self.args.write_timeout, None), + ( + read, + {}, + OP_TYPE_READ, + self.args.read_timeout, + lambda rows: check_control_result(rows, expected), + ), + ): + execute_query( + RequestParams( + pool, + query, + params, + self.metrics, + (operation,), + ydb.BaseRequestSettings().with_timeout(timeout / 1000), + ydb.RetrySettings(idempotent=True, max_session_acquire_timeout=timeout / 1000), + check_result_cb=checker, + on_error=self.record_failure, + ) + ) + except BaseException as error: + failure = error + raise + finally: + try: + schema_pool.retry_operation_sync(lambda session: session.drop_table(table)) + except BaseException: + if failure is None: + raise + logger.exception("Cannot clean up control table %s", table) def _run_table_read_jobs(self): logger.info("Start table read jobs") @@ -128,8 +229,8 @@ def _run_table_read_jobs(self): for i in range(self.args.read_threads): future = threading.Thread( name=f"slo_table_read_{i}", - target=self._run_table_reads, - args=(read_query, read_limiter), + target=self._run_worker, + args=(self._run_table_reads, read_query, read_limiter), ) future.start() futures.append(future) @@ -149,8 +250,8 @@ def _run_table_write_jobs(self): for i in range(self.args.write_threads): future = threading.Thread( name=f"slo_table_write_{i}", - target=self._run_table_writes, - args=(write_query, write_limiter), + target=self._run_worker, + args=(self._run_table_writes, write_query, write_limiter), ) future.start() futures.append(future) @@ -168,8 +269,8 @@ def _run_query_read_jobs(self): for i in range(self.args.read_threads): future = threading.Thread( name=f"slo_query_read_{i}", - target=self._run_query_reads, - args=(read_query, read_limiter), + target=self._run_worker, + args=(self._run_query_reads, read_query, read_limiter), ) future.start() futures.append(future) @@ -187,8 +288,8 @@ def _run_query_write_jobs(self): for i in range(self.args.write_threads): future = threading.Thread( name=f"slo_query_write_{i}", - target=self._run_query_writes, - args=(write_query, write_limiter), + target=self._run_worker, + args=(self._run_query_writes, write_query, write_limiter), ) future.start() futures.append(future) @@ -206,14 +307,10 @@ def _run_table_reads(self, query, limiter): ) with ydb.SessionPool(self.driver) as pool: - while time.time() - start_time < self.args.time: + while not self._stop.is_set() and time.time() - start_time < self.args.time: params = {"$object_id": randint(1, self.max_id)} with limiter: - - def check_result(result): - assert result[0].rows[0] - request_params = RequestParams( pool=pool, query=query, @@ -222,7 +319,7 @@ def check_result(result): labels=(OP_TYPE_READ,), request_settings=request_settings, retry_settings=retry_setting, - check_result_cb=check_result, + on_error=self.record_failure, ) execute_query(request_params) @@ -241,7 +338,7 @@ def _run_table_writes(self, query, limiter): row_generator = RowGenerator(self.max_id) with ydb.SessionPool(self.driver) as pool: - while time.time() - start_time < self.args.time: + while not self._stop.is_set() and time.time() - start_time < self.args.time: row = row_generator.get() params = { "$object_id": row.object_id, @@ -259,6 +356,7 @@ def _run_table_writes(self, query, limiter): labels=(OP_TYPE_WRITE,), request_settings=request_settings, retry_settings=retry_setting, + on_error=self.record_failure, ) execute_query(request_params) @@ -275,15 +373,10 @@ def _run_query_reads(self, query, limiter): ) with ydb.QuerySessionPool(self.driver) as pool: - while time.time() - start_time < self.args.time: + while not self._stop.is_set() and time.time() - start_time < self.args.time: params = {"$object_id": (randint(1, self.max_id), ydb.PrimitiveType.Uint64)} with limiter: - - def check_result(result): - with result: - pass - request_params = RequestParams( pool=pool, query=query, @@ -292,7 +385,7 @@ def check_result(result): labels=(OP_TYPE_READ,), request_settings=request_settings, retry_settings=retry_setting, - check_result_cb=check_result, + on_error=self.record_failure, ) execute_query(request_params) @@ -311,7 +404,7 @@ def _run_query_writes(self, query, limiter): row_generator = RowGenerator(self.max_id) with ydb.QuerySessionPool(self.driver) as pool: - while time.time() - start_time < self.args.time: + while not self._stop.is_set() and time.time() - start_time < self.args.time: row = row_generator.get() params = { "$object_id": (row.object_id, ydb.PrimitiveType.Uint64), @@ -323,10 +416,6 @@ def _run_query_writes(self, query, limiter): ), } - def check_result(result): - with result: - pass - with limiter: request_params = RequestParams( pool=pool, @@ -336,7 +425,7 @@ def check_result(result): labels=(OP_TYPE_WRITE,), request_settings=request_settings, retry_settings=retry_setting, - check_result_cb=check_result, + on_error=self.record_failure, ) execute_query(request_params) diff --git a/tests/slo/src/root_runner.py b/tests/slo/src/root_runner.py index 2c8c2e75a..fb7117c45 100644 --- a/tests/slo/src/root_runner.py +++ b/tests/slo/src/root_runner.py @@ -51,23 +51,30 @@ def _run_sync_command(self, args, runner_instance, command): grpc_keep_alive_timeout=5000, ) - with ydb.Driver(driver_config) as driver: - driver.wait(timeout=300) - try: - runner_instance.set_driver(driver) - if command == "create": - runner_instance.create(args) - elif command == "run": - runner_instance.run(args) - elif command == "cleanup": - runner_instance.cleanup(args) - else: - raise RuntimeError(f"Unknown command {command} for prefix {runner_instance.prefix}") - except BaseException: - logger.exception("Something went wrong") - raise - finally: - driver.stop(timeout=getattr(args, "shutdown_time", 10)) + failure = None + try: + with ydb.Driver(driver_config) as driver: + try: + driver.wait(timeout=300) + runner_instance.set_driver(driver) + if command == "create": + runner_instance.create(args) + elif command == "run": + runner_instance.run(args) + elif command == "cleanup": + runner_instance.cleanup(args) + else: + raise RuntimeError(f"Unknown command {command} for prefix {runner_instance.prefix}") + except BaseException as error: + failure = error + logger.exception("Something went wrong") + raise + finally: + driver.stop(timeout=getattr(args, "shutdown_time", 10)) + except BaseException as error: + if failure is not None and error is not failure: + raise failure from error + raise async def _run_async_command(self, args, runner_instance, command): """Run command in asynchronous mode""" @@ -77,17 +84,24 @@ async def _run_async_command(self, args, runner_instance, command): grpc_keep_alive_timeout=5000, ) - async with ydb.aio.Driver(driver_config) as driver: - await driver.wait(timeout=300) - try: - runner_instance.set_driver(driver) - if command == "run": - await runner_instance.run_async(args) - else: - raise RuntimeError(f"Async mode only supports 'run' command, got '{command}'") - except BaseException: - logger.exception("Something went wrong in async mode") - raise + failure = None + try: + async with ydb.aio.Driver(driver_config) as driver: + try: + await driver.wait(timeout=300) + runner_instance.set_driver(driver) + if command == "run": + await runner_instance.run_async(args) + else: + raise RuntimeError(f"Async mode only supports 'run' command, got '{command}'") + except BaseException as error: + failure = error + logger.exception("Something went wrong in async mode") + raise + except BaseException as error: + if failure is not None and error is not failure: + raise failure from error + raise def create_runner() -> SLORunner: diff --git a/tests/table/test_tx.py b/tests/table/test_tx.py index 0a8ca24ae..ffdbc4ebc 100644 --- a/tests/table/test_tx.py +++ b/tests/table/test_tx.py @@ -196,3 +196,186 @@ def test_truncated_response_allow(driver_sync, table_name, table_path): result = t.execute("SELECT * FROM %s" % table_name) assert result[0].truncated assert len(result[0].rows) == 1000 + + +class TestSloControlData: + def test_sync_result_checked_after_retry_and_stream_errors_preserved(self, monkeypatch): + from pathlib import Path + from types import SimpleNamespace + from unittest.mock import Mock + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "tests/slo/src")) + from jobs.table_jobs import RequestParams, check_control_result, control_row, execute_query + + expected = control_row(2) + session = Mock() + active = False + + def retry(callee, retry_settings): + nonlocal active + active = True + try: + return ydb.retry_operation_sync(lambda: callee(session), retry_settings=retry_settings) + finally: + active = False + + def check(rows): + assert not active + check_control_result(rows, expected) + + pool = SimpleNamespace(retry_operation_sync=retry) + metrics = Mock() + errors = [] + request = RequestParams( + pool, "query", {}, metrics, ("read",), None, ydb.RetrySettings(max_retries=1), check, errors.append + ) + for changes in ( + {"payload_str": "generation:1"}, + {"payload_str": "corrupt"}, + {"payload_double": 2.5}, + {"payload_timestamp": 1}, + {"object_id": 1}, + ): + session.transaction.return_value.execute.return_value = [SimpleNamespace(rows=[{**expected, **changes}])] + with pytest.raises(RuntimeError): + execute_query(request) + for rows in ([], [expected, expected]): + session.transaction.return_value.execute.return_value = [SimpleNamespace(rows=rows)] + with pytest.raises(RuntimeError): + execute_query(request) + session.transaction.return_value.execute.side_effect = [ + ydb.Aborted("recovered"), + [SimpleNamespace(rows=[dict(expected)])], + ] + assert execute_query(request) == [expected] + assert metrics.stop.call_args.kwargs["attempts"] == 2 + terminal = ydb.BadRequest("original") + + class BrokenStream: + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def __iter__(self): + yield SimpleNamespace(rows=[expected]) + raise terminal + + session.transaction.return_value.execute.side_effect = None + session.transaction.return_value.execute.return_value = BrokenStream() + with pytest.raises(ydb.BadRequest) as failure: + execute_query(request) + assert failure.value is terminal + assert errors[-1] is terminal + + @pytest.mark.asyncio + async def test_async_result_checked_after_retry_and_stream_errors_preserved(self, monkeypatch): + from pathlib import Path + from types import SimpleNamespace + from unittest.mock import AsyncMock, Mock + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "tests/slo/src")) + from jobs.async_table_jobs import AsyncTableJobManager + from jobs.table_jobs import check_control_result, control_row + from ydb.retries import retry_operation_async + + expected = control_row(2) + session = SimpleNamespace(execute=AsyncMock()) + manager = AsyncTableJobManager(Mock(), SimpleNamespace(), Mock(), "/local/load", 1000) + active = False + + async def retry(callee, retry_settings): + nonlocal active + active = True + try: + return await retry_operation_async(lambda: callee(session), retry_settings=retry_settings) + finally: + active = False + + def check(rows): + assert not active + check_control_result(rows, expected) + + async def stream(rows, error=None): + yield SimpleNamespace(rows=rows) + if error is not None: + raise error + + pool = SimpleNamespace(retry_operation_async=retry) + settings = ydb.RetrySettings(max_retries=1) + for rows in ( + [], + [{**expected, "payload_str": "generation:1"}], + [{**expected, "payload_str": "corrupt"}], + [{**expected, "payload_double": 3.25}], + [{**expected, "payload_timestamp": 1}], + [{**expected, "object_id": 1}], + [expected, expected], + ): + session.execute.return_value = stream(rows) + with pytest.raises(RuntimeError): + await manager._execute_query(pool, "query", {}, None, settings, ("read",), check) + session.execute.side_effect = [ydb.Aborted("recovered"), stream([dict(expected)])] + assert await manager._execute_query(pool, "query", {}, None, settings, ("read",), check) == [expected] + terminal = ydb.BadRequest("original") + session.execute.side_effect = None + session.execute.return_value = stream([expected], terminal) + with pytest.raises(ydb.BadRequest) as failure: + await manager._execute_query(pool, "query", {}, None, settings, ("read",), check) + assert failure.value is terminal + + def test_owned_control_tables_and_sequential_generations(self, monkeypatch): + import threading + from concurrent.futures import ThreadPoolExecutor + from pathlib import Path + from types import SimpleNamespace + from unittest.mock import Mock + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "tests/slo/src")) + from jobs import table_jobs + + barrier = threading.Barrier(2, timeout=5) + tables = [] + checked = threading.local() + + def execute(request): + if request.check_result_cb is None: + generation = int(request.params["$payload_str"].split(":")[1]) + assert generation == getattr(checked, "generation", 0) + 1 + checked.expected = {key[1:]: value for key, value in request.params.items()} + checked.expected["object_hash"] = 0 + if generation == 1: + barrier.wait() + else: + request.check_result_cb([dict(checked.expected)]) + checked.generation = int(checked.expected["payload_str"].split(":")[1]) + + def scheme(query): + tables.append(query.split("`")[1]) + + session = Mock() + session.execute_scheme.side_effect = scheme + session.prepare.side_effect = lambda query: query + pool = Mock() + pool.__enter__ = Mock(return_value=pool) + pool.__exit__ = Mock(return_value=False) + pool.retry_operation_sync.side_effect = lambda callee: callee(session) + monkeypatch.setattr(table_jobs.ydb, "SessionPool", lambda driver: pool) + owners = [SimpleNamespace(hex=f"owner_{i}") for i in (1, 2)] + monkeypatch.setattr(table_jobs, "uuid4", Mock(side_effect=owners)) + monkeypatch.setattr(table_jobs, "execute_query", execute) + args = SimpleNamespace(read_timeout=10000, write_timeout=20000) + + def owner(): + manager = table_jobs.TableJobManager(Mock(), args, Mock(), "/local/load", 1000) + manager.workload_type = "sync-table" + manager._run_control_checks() + assert checked.generation == 2 + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [executor.submit(owner) for _ in range(2)] + for future in futures: + future.result(timeout=10) + assert len(tables) == len(set(tables)) == 2 + assert sorted(call.args[0] for call in session.drop_table.call_args_list) == sorted(tables) From e3040d165f61b517e391325c684506e780e6beae Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Wed, 14 Oct 2026 19:49:17 +0300 Subject: [PATCH 2/3] Keep the SLO entrypoint executable --- tests/slo/docker-entrypoint.sh | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 tests/slo/docker-entrypoint.sh diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh old mode 100644 new mode 100755 From a8e2ae1b1b422a7686465fe1d124e5720a7f02db Mon Sep 17 00:00:00 2001 From: KirillKurdyukov Date: Wed, 14 Oct 2026 22:23:54 +0300 Subject: [PATCH 3/3] Remove the extra SLO subprocess launcher --- tests/slo/docker-entrypoint.sh | 31 ++----------------------------- 1 file changed, 2 insertions(+), 29 deletions(-) diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh index 5dcfeb83d..55d933ec3 100755 --- a/tests/slo/docker-entrypoint.sh +++ b/tests/slo/docker-entrypoint.sh @@ -44,35 +44,8 @@ if [ "$PREFIX" = "topic" ]; then EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}" fi -if [ "$PREFIX" = "table" ]; then - TABLE_NAME="key_value_$(python -c 'import uuid; print(uuid.uuid4().hex)')" - python ./tests/slo/src table-create "$ENDPOINT" "$DATABASE" --table-name "$TABLE_NAME" - exec python -c ' -import signal -import subprocess -import sys - -table, endpoint, database, duration = sys.argv[1:5] -common = [sys.executable, "./tests/slo/src"] -workload = subprocess.Popen([ - *common, "table-run", endpoint, database, "--time", duration, - *sys.argv[5:], "--table-name", table, -]) -for signum in (signal.SIGTERM, signal.SIGINT, signal.SIGQUIT): - signal.signal(signum, lambda received, _frame: workload.send_signal(received)) -try: - status = workload.wait() -finally: - cleanup = subprocess.run([ - *common, "table-cleanup", endpoint, database, "--table-name", table, - ], check=False) - if cleanup.returncode: - print("WARN: table cleanup failed", file=sys.stderr) -sys.exit(status if status >= 0 else 128 - status) -' "$TABLE_NAME" "$ENDPOINT" "$DATABASE" "$DURATION" "$@" -fi - -# Topic schema preparation retains its existing behavior. +# 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" $EXTRA_ARGS \ || echo "WARN: ${PREFIX}-create exited non-zero (treated as already-prepared)" >&2