Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 8 additions & 15 deletions tests/python/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand Down
28 changes: 28 additions & 0 deletions tests/python/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
43 changes: 34 additions & 9 deletions tests/python/test_rfc0003_concurrency.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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},
Expand Down Expand Up @@ -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},
Expand Down Expand Up @@ -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,
Expand All @@ -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",
Expand Down Expand Up @@ -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,
Expand All @@ -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}},
)
Expand Down
Loading