From 979d8ac24b9425fa9d69a4d4801f967b067deb56 Mon Sep 17 00:00:00 2001 From: Scott Robinson Date: Wed, 7 Oct 2026 03:01:25 +0000 Subject: [PATCH] test(python): give each RFC-0003 concurrency worker its own client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four RFC-0003 §4 stress tests drive the server from 50 threads through the session-scoped boto3 client. A boto3 client is not safe for concurrent use: two threads interleaving on one of its TLS connections corrupt the record stream, and the request fails with botocore.exceptions.SSLError: SSL validation failed ... [SSL: TLSV1_ALERT_DECODE_ERROR] tlsv1 alert decode error which looks like a server fault and is not. It fired on the MongoDB leg of #382, a pull request that touches only the PostgreSQL transaction path; the rerun passed. The client construction moves from the conftest fixture into `helpers.make_dynamodb_client`, the fixture calls it, and the four tests take a `thread_client` fixture that builds one client per worker thread, lazily, in a `threading.local`. Setup and verification on the main thread keep using the session client. Run three times against a PostgreSQL server built from main: 4 passed each time. Signed-off-by: Scott Robinson --- tests/python/conftest.py | 23 +++++-------- tests/python/helpers.py | 28 +++++++++++++++ tests/python/test_rfc0003_concurrency.py | 43 +++++++++++++++++++----- 3 files changed, 70 insertions(+), 24 deletions(-) diff --git a/tests/python/conftest.py b/tests/python/conftest.py index 83b7384af..956ea3f53 100755 --- a/tests/python/conftest.py +++ b/tests/python/conftest.py @@ -27,7 +27,7 @@ import pytest import urllib3 -from helpers import unique_name, wait_for_active, wait_for_deleted # noqa: F401 +from helpers import make_dynamodb_client, unique_name, wait_for_active, wait_for_deleted # noqa: F401 # Suppress InsecureRequestWarning for self-signed TLS certs. urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) @@ -65,20 +65,13 @@ def is_extenddb(endpoint_url) -> bool: @pytest.fixture(scope="session") def dynamodb_client(endpoint_url: str | None): - """Session-scoped boto3 DynamoDB client.""" - kwargs: dict = { - "service_name": "dynamodb", - "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), - "config": botocore.config.Config( - retries={"max_attempts": 0}, # No SDK retries — we want raw errors - ), - } - if endpoint_url: - kwargs["endpoint_url"] = endpoint_url - if endpoint_url.startswith("https://"): - ca_cert = os.environ.get("EXTENDDB_CA_CERT", "") - kwargs["verify"] = ca_cert if ca_cert else False - return boto3.client(**kwargs) + """Session-scoped boto3 DynamoDB client. + + Not safe to share across threads; a test that drives the server from a + thread pool builds one client per thread with + `helpers.make_dynamodb_client` instead. + """ + return make_dynamodb_client(endpoint_url) @pytest.fixture(scope="session") diff --git a/tests/python/helpers.py b/tests/python/helpers.py index 448a1607e..f4ed85479 100755 --- a/tests/python/helpers.py +++ b/tests/python/helpers.py @@ -9,9 +9,37 @@ from __future__ import annotations +import os import time import uuid +import boto3 +import botocore.config + + +def make_dynamodb_client(endpoint_url: str | None): + """A boto3 DynamoDB client with SDK retries off, for the configured target. + + The session fixture in conftest is one of these. A test that drives the + server from several threads must build one per thread with this function: + a boto3 client is not safe for concurrent use, and sharing one across + threads interleaves requests on its TLS connections, which surfaces as + spurious `SSLError: TLSV1_ALERT_DECODE_ERROR` under load. + """ + kwargs: dict = { + "service_name": "dynamodb", + "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + "config": botocore.config.Config( + retries={"max_attempts": 0}, # No SDK retries — we want raw errors + ), + } + if endpoint_url: + kwargs["endpoint_url"] = endpoint_url + if endpoint_url.startswith("https://"): + ca_cert = os.environ.get("EXTENDDB_CA_CERT", "") + kwargs["verify"] = ca_cert if ca_cert else False + return boto3.client(**kwargs) + def wait_for_active(client, table_name: str, timeout: float = 60.0) -> None: """Poll DescribeTable until status is ACTIVE with exponential backoff.""" diff --git a/tests/python/test_rfc0003_concurrency.py b/tests/python/test_rfc0003_concurrency.py index d8e3c4555..1d5d72342 100644 --- a/tests/python/test_rfc0003_concurrency.py +++ b/tests/python/test_rfc0003_concurrency.py @@ -26,23 +26,48 @@ class (or no error at all) even under sustained contention. If the These tests deliberately use a boto3 client with `retries={"max_attempts": 0}` so any InternalServerError surfaces immediately instead of being masked by the SDK's retry policy. + +Each worker thread uses its own client. A boto3 client is not safe for +concurrent use; the shared session client, driven from 50 threads, would +interleave requests on one TLS connection and fail at random with +`SSLError: TLSV1_ALERT_DECODE_ERROR`, which is the SDK, not the backend. """ from __future__ import annotations +import threading import uuid from concurrent.futures import ThreadPoolExecutor, as_completed import pytest from botocore.exceptions import ClientError -from helpers import unique_name, wait_for_active, wait_for_deleted +from helpers import make_dynamodb_client, unique_name, wait_for_active, wait_for_deleted NUM_THREADS = 50 INCREMENTS_PER_THREAD = 20 # 50 * 20 = 1_000 increments +@pytest.fixture() +def thread_client(endpoint_url): + """A callable returning the current thread's own DynamoDB client. + + Built lazily, once per thread, so each of the pool's workers talks to the + server over connections nothing else touches. + """ + local = threading.local() + + def get(): + client = getattr(local, "client", None) + if client is None: + client = make_dynamodb_client(endpoint_url) + local.client = client + return client + + return get + + @pytest.fixture() def counter_table(dynamodb_client): """A plain HASH-keyed table for the RFC-0003 §4.x scenarios.""" @@ -73,13 +98,13 @@ class TestRfc0003UnconditionalPutOnHotKey: attempts of a snapshot-txn WriteConflict loop. """ - def test_all_writers_succeed(self, dynamodb_client, counter_table): + def test_all_writers_succeed(self, dynamodb_client, thread_client, counter_table): key = f"hot-{uuid.uuid4().hex[:8]}" errors: list[str] = [] def _write(thread_id: int) -> None: try: - dynamodb_client.put_item( + thread_client().put_item( TableName=counter_table, Item={ "pk": {"S": key}, @@ -114,13 +139,13 @@ class TestRfc0003ConditionalPutOnHotKey: `ConditionalCheckFailedException`, not `InternalServerError`. """ - def test_one_winner_rest_ccf(self, dynamodb_client, counter_table): + def test_one_winner_rest_ccf(self, dynamodb_client, thread_client, counter_table): key = f"race-{uuid.uuid4().hex[:8]}" outcomes: list[tuple[str, str]] = [] # (result, error_code) def _conditional_put(thread_id: int) -> None: try: - dynamodb_client.put_item( + thread_client().put_item( TableName=counter_table, Item={ "pk": {"S": key}, @@ -163,7 +188,7 @@ class TestRfc0003AtomicCounterAdd: calls converge at MongoDB's doc-lock level without OCC retries. """ - def test_all_increments_apply(self, dynamodb_client, counter_table): + def test_all_increments_apply(self, dynamodb_client, thread_client, counter_table): key = f"counter-{uuid.uuid4().hex[:8]}" dynamodb_client.put_item( TableName=counter_table, @@ -175,7 +200,7 @@ def _increment(thread_id: int) -> int: done = 0 for _ in range(INCREMENTS_PER_THREAD): try: - dynamodb_client.update_item( + thread_client().update_item( TableName=counter_table, Key={"pk": {"S": key}}, UpdateExpression="ADD #c :one", @@ -211,7 +236,7 @@ class TestRfc0003UnconditionalDeleteOnHotKey: are no-ops; if it doesn't, all are no-ops. Never an error. """ - def test_all_deletes_succeed(self, dynamodb_client, counter_table): + def test_all_deletes_succeed(self, dynamodb_client, thread_client, counter_table): key = f"delkey-{uuid.uuid4().hex[:8]}" dynamodb_client.put_item( TableName=counter_table, @@ -221,7 +246,7 @@ def test_all_deletes_succeed(self, dynamodb_client, counter_table): def _delete(_thread_id: int) -> None: try: - dynamodb_client.delete_item( + thread_client().delete_item( TableName=counter_table, Key={"pk": {"S": key}}, )