Intermittent Elasticsearch Test errors

[strawgate]

FAILED tests/stores/elasticsearch/test_elasticsearch.py::TestElasticsearchStoreNativeMode::test_empty_put[9.0.0-no_ctx_manager] - key_value.shared.errors.store.StoreSetupError: Failed to setup collection: BadRequestError(400, 'resource_already_exists_exception', 'index [kv-store-e2e-test-test/bWbuVhNnQxS2_2bwXj3Oow] already exists'): (collection: test)
FAILED tests/stores/elasticsearch/test_elasticsearch.py::TestElasticsearchStoreNativeMode::test_empty_ttl[9.0.0-with_ctx_manager] - Failed: Timeout (>10.0s) from pytest-timeout.

maybe we're, for some reason, not deleting the indices?

[claude]

Claude finished @strawgate's task —— View job

---

Investigation Complete ✓

I've identified the root cause of the intermittent Elasticsearch test failures.

---

Recommendation

Fix the setup_elasticsearch fixture scope mismatch that causes race conditions between parameterized test runs.

The issue stems from a fixture scoping problem in test_elasticsearch.py:102-128:

@pytest.fixture(autouse=True, scope="session", params=ELASTICSEARCH_VERSIONS_TO_TEST)
async def setup_elasticsearch(self, request: pytest.FixtureRequest) -> AsyncGenerator[None, None]:
    # Creates ONE container per version for the ENTIRE test session
    # But multiple test classes (TestElasticsearchStoreNativeMode, TestElasticsearchStoreNonNativeMode)
    # share the SAME container

The problem: Session-scoped fixture + parameterization creates one container per version across ALL test classes, but function-scoped cleanup runs per test.

What's happening:

1. setup_elasticsearch is scope="session" with params=["9.0.0", "9.2.0"]
2. pytest creates one Elasticsearch container per version for the entire test session
3. Both TestElasticsearchStoreNativeMode and TestElasticsearchStoreNonNativeMode use the same container
4. When tests from different classes run in parallel or interleaved, they interfere with each other's indices
5. The cleanup_elasticsearch_indices fixture (lines 124-128) runs per test function, but multiple tests can be setting up collections simultaneously on the shared container

Race condition:

• Test A (native mode): calls store.put() → triggers _setup_collection() → checks if index exists
• Test B (non-native mode): calls store.put() → triggers _setup_collection() → checks if index exists
• Test A: tries to create index → succeeds
• Test B: tries to create index → fails with "resource_already_exists_exception" ❌

---

Findings

Root Cause Analysis

The Fixture Scoping Issue

The setup_elasticsearch fixture at test_elasticsearch.py:102-117 has conflicting scopes:

@pytest.fixture(autouse=True, scope="session", params=ELASTICSEARCH_VERSIONS_TO_TEST)
async def setup_elasticsearch(self, request: pytest.FixtureRequest) -> AsyncGenerator[None, None]:
    version = request.param
    es_image = f"docker.elastic.co/elasticsearch/elasticsearch:{version}"
    
    with docker_container(
        f"elasticsearch-test-{version}",
        es_image,
        {str(ES_CONTAINER_PORT): ES_PORT},
        {"discovery.type": "single-node", "xpack.security.enabled": "false"},
    ):
        # ...
        yield

Problems:

1. scope="session" means one container is created per version for ALL tests in the session
2. Both test classes (TestElasticsearchStoreNativeMode and TestElasticsearchStoreNonNativeMode) inherit this fixture
3. Tests from both classes share the same Elasticsearch container
4. The cleanup_elasticsearch_indices fixture (lines 124-128) is function-scoped, running before/after each test
5. BUT the cleanup can't prevent race conditions when multiple tests try to create the same index simultaneously

The _setup_collection Race Condition

In store.py:224-230:

async def _setup_collection(self, *, collection: str) -> None:
    index_name = self._sanitize_index_name(collection=collection)
    
    if await self._client.options(ignore_status=404).indices.exists(index=index_name):
        return  # ← Returns if index already exists
    
    _ = await self._client.options(ignore_status=404).indices.create(index=index_name, mappings=DEFAULT_MAPPING, settings={})

The race:

1. Test A checks if index exists → False
2. Test B checks if index exists → False (happens before Test A creates it)
3. Test A creates index → Success
4. Test B tries to create index → Fails with resource_already_exists_exception

Even though line 230 has ignore_status=404, Elasticsearch returns a 400 error (BadRequest) for "index already exists", not a 404.

Why It's Intermittent

The error is intermittent because:

• It only occurs when tests from different test classes run close enough in time to hit the race condition
• With --dist=loadfile, tests in the same file run on the same worker, but different test classes can still interleave
• The parametrization creates multiple test instances ([9.0.0-no_ctx_manager], [9.0.0-with_ctx_manager], etc.)
• Timing varies based on system load, causing the race to trigger unpredictably

The Timeout Error

The timeout in test_empty_ttl[9.0.0-with_ctx_manager] is likely a side effect of the first error:

• When one test fails during index creation, it may leave the Elasticsearch container in an inconsistent state
• Subsequent tests may hang waiting for operations to complete
• The 10-second pytest timeout kicks in

Related Code Sections

File	Lines	Description
test_elasticsearch.py	102-117	setup_elasticsearch fixture with scope="session"
test_elasticsearch.py	124-128	cleanup_elasticsearch_indices fixture (function-scoped)
test_elasticsearch.py	152-158	TestElasticsearchStoreNativeMode store fixture
test_elasticsearch.py	212-218	TestElasticsearchStoreNonNativeMode store fixture
store.py	224-230	_setup_collection with race condition

---

Detailed Action Plan

Option 1: Change Fixture Scope to Class (Recommended)

Change setup_elasticsearch from scope="session" to scope="class":

@pytest.fixture(autouse=True, scope="class", params=ELASTICSEARCH_VERSIONS_TO_TEST)  # ← Changed to "class"
async def setup_elasticsearch(self, request: pytest.FixtureRequest) -> AsyncGenerator[None, None]:
    version = request.param
    es_image = f"docker.elastic.co/elasticsearch/elasticsearch:{version}"
    
    with docker_container(
        f"elasticsearch-test-{version}",
        es_image,
        {str(ES_CONTAINER_PORT): ES_PORT},
        {"discovery.type": "single-node", "xpack.security.enabled": "false"},
    ):
        if not await async_wait_for_true(bool_fn=ping_elasticsearch, tries=WAIT_FOR_ELASTICSEARCH_TIMEOUT, wait_time=2):
            msg = f"Elasticsearch {version} failed to start"
            raise ElasticsearchFailedToStartError(msg)
        
        yield

Benefits:

• Each test class gets its own Elasticsearch container
• No cross-contamination between native and non-native mode tests
• Preserves parameterization (tests still run against multiple ES versions)
• Minimal code change

Tradeoffs:

• Slightly longer test execution (2 containers instead of 1 per version)
• More Docker resource usage during tests

---

Option 2: Add Unique Collection Prefixes Per Test Class

Modify each test class to use a different collection prefix:

class TestElasticsearchStoreNativeMode(BaseTestElasticsearchStore):
    @override
    @pytest.fixture
    async def store(self) -> ElasticsearchStore:
        return ElasticsearchStore(url=ES_URL, index_prefix="kv-store-e2e-test-native", native_storage=True)
    
class TestElasticsearchStoreNonNativeMode(BaseTestElasticsearchStore):
    @override
    @pytest.fixture
    async def store(self) -> ElasticsearchStore:
        return ElasticsearchStore(url=ES_URL, index_prefix="kv-store-e2e-test-nonnative", native_storage=False)

Benefits:

• Keeps session-scoped fixture (faster tests)
• Different test classes create different indices

Tradeoffs:

• Doesn't prevent races within the same test class if tests run in parallel
• Less clean separation between test runs

---

Option 3: Make Index Creation Idempotent

Update _setup_collection to handle the race condition gracefully:

async def _setup_collection(self, *, collection: str) -> None:
    index_name = self._sanitize_index_name(collection=collection)
    
    if await self._client.options(ignore_status=404).indices.exists(index=index_name):
        return
    
    try:
        _ = await self._client.options(ignore_status=[404, 400]).indices.create(
            index=index_name, 
            mappings=DEFAULT_MAPPING, 
            settings={}
        )
    except Exception as e:
        # Check if the error is "resource_already_exists_exception"
        if "resource_already_exists_exception" not in str(e):
            raise
        # Index was created by another concurrent request, that's fine

Benefits:

• Handles the race condition at the source
• More robust for production use cases with concurrent requests

Tradeoffs:

• Doesn't address the fundamental fixture scoping issue
• May hide other legitimate errors

---

Recommended Solution: Option 1 (Change to scope="class")

This is the cleanest fix that addresses the root cause without complexity. The test execution time increase is negligible compared to the benefit of eliminating flaky tests.

---

Related Items

Related Issues and Pull Requests

Repository	Issue or PR	Relevance
strawgate/py-key-value	Issue #166 - MultiDiskStore on Windows test flaky	Similar flaky test issue resolved by isolating test instances
strawgate/py-key-value	Issue #129 - Ubuntu 22.04 tests hanging	Another test timeout issue, resolved by fixing pytest-xdist race condition

Related Files

Repository	File	Relevance	Sections
strawgate/py-key-value	test_elasticsearch.py	Contains the failing tests and problematic fixture	102-128
strawgate/py-key-value	store.py	ElasticsearchStore implementation with race condition in _setup_collection	224-230
strawgate/py-key-value	pyproject.toml	pytest configuration with --dist=loadfile	53-64
strawgate/py-key-value	conftest.py	Docker container management utilities	148-174

Related Webpages

Name	URL	Relevance
pytest fixtures scope documentation	https://docs.pytest.org/en/stable/how-to/fixtures.html#scope-sharing-fixtures-across-classes-modules-packages-or-session	Official documentation explaining fixture scopes and their implications for test isolation
pytest-xdist documentation	https://pytest-xdist.readthedocs.io/en/stable/distribution.html	Explains --dist=loadfile and how tests are distributed across workers
Elasticsearch Index Creation API	https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html	Documents the index creation API and error responses including resource_already_exists_exception

---

Problems Encountered During Investigation

None - all investigation tools were accessible and provided comprehensive information about the codebase, test infrastructure, and Elasticsearch integration.

---

[strawgate]

@claude xdist is per-file so those tests can't run at the same time

[claude]

Claude finished @strawgate's task —— View job

---

Root Cause Identified!

Thanks for the correction about xdist! After re-examining the code with that understanding, I found the actual root cause:

---

The Bug

File: test_elasticsearch.py:43-46

async def cleanup_elasticsearch_indices(elasticsearch_client: AsyncElasticsearch):
    indices = await elasticsearch_client.options(ignore_status=404).indices.get(index="kv-store-e2e-test-*")
    for index in indices:
        _ = await elasticsearch_client.options(ignore_status=404).indices.delete(index=index)

The Problem:

This cleanup function has a subtle but critical bug. When indices.get() is called with a wildcard pattern and no indices exist, it returns an empty response (or raises a 404 that's ignored). The code then tries to iterate over this response.

However, the more serious issue is that when indices DO exist but the cleanup fails (e.g., due to timing issues, Elasticsearch being busy, or index locks), the ignore_status=404 silently swallows the errors, and the indices aren't actually deleted.

Additionally, there's no explicit check for whether the get operation succeeded before attempting to iterate and delete.

---

Why This Causes Intermittent Failures

1. Cleanup silently fails: When indices can't be deleted (Elasticsearch timing issues, locks, etc.), the ignore_status=404 masks the failure
2. Indices persist: Failed cleanup means indices from previous tests remain
3. Index already exists error: When _setup_collection() runs in the next test, it tries to create an index that already exists → resource_already_exists_exception
4. Timeout: After the first error, subsequent tests may hang waiting for Elasticsearch operations
5. Intermittent: The cleanup fails unpredictably based on Elasticsearch's internal state and timing

---

The Fix

Option 1: Improve the cleanup function (Recommended)

async def cleanup_elasticsearch_indices(elasticsearch_client: AsyncElasticsearch):
    try:
        indices = await elasticsearch_client.indices.get(index="kv-store-e2e-test-*")
        if indices:
            for index_name in indices:
                try:
                    await elasticsearch_client.indices.delete(index=index_name)
                except Exception as e:
                    logger.warning(f"Failed to delete index {index_name}: {e}")
    except Exception as e:
        # No indices to clean up, or error getting indices
        logger.debug(f"No indices to clean up or error: {e}")

Option 2: Make index creation truly idempotent in the store

Modify store.py:224-230 to handle the race condition gracefully:

async def _setup_collection(self, *, collection: str) -> None:
    index_name = self._sanitize_index_name(collection=collection)
    
    if await self._client.options(ignore_status=404).indices.exists(index=index_name):
        return
    
    try:
        _ = await self._client.indices.create(
            index=index_name, 
            mappings=DEFAULT_MAPPING, 
            settings={}
        )
    except Exception as e:
        # Check if error is "resource_already_exists_exception"
        if "resource_already_exists_exception" in str(e):
            # Index was created between our check and create call, that's fine
            return
        raise

Option 3: Both (Belt and suspenders)

Implement both fixes for maximum reliability.

---

Tasks

• Acknowledge the correction about xdist configuration
• Re-examine test fixtures and cleanup mechanisms
• Identify actual root cause
• Implement the fix

Should I implement the fix? Which option do you prefer?

---