Spaces:

techprotrade
/

annator-command-center

Paused

annator-command-center / tests /soak /test_cache_stability.py

Deploy ATOM FastAPI command center runtime (part 9)

81e3673 verified 3 months ago

12 kB

"""
Soak Tests for Cache Stability Under Concurrent Load
Extended duration tests (30 minutes) to validate cache behavior:
- Concurrent operations (10 workers, 10000 ops each)
- Cache consistency (no race conditions, no data loss)
- TTL expiration (entries expire correctly)
- Cache unbounded growth detection
Cache issues detected:
- Race conditions in concurrent get/set operations
- Cache corruption (data loss, incorrect values)
- TTL not working (cache grows unbounded)
- Deadlocks under concurrent load
Tests use ThreadPoolExecutor for realistic concurrent access patterns.
"""

import gc
import pytest
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from core.governance_cache import GovernanceCache


@pytest.mark.soak
@pytest.mark.timeout(1800)  # 30 minutes
def test_cache_concurrent_operations_30min(
    memory_monitor,
    enable_gc_control,
    soak_test_config
):
    """
    Soak test for cache stability under concurrent operations (30 minutes).
    Validates:
    - Cache remains consistent under concurrent load (10 workers)
    - No race conditions or deadlocks
    - No cache corruption (data loss, incorrect values)
    - Zero cache errors (get returns None after set)
    Test pattern:
    - Run for 30 minutes (1800 seconds)
    - Use ThreadPoolExecutor with 10 workers
    - Each worker performs 10000 cache operations (set + get)
    - Validate cache consistency: if get returns None after set, fail
    - Log worker completions
    - Force GC every 10 iterations
    Duration: 30 minutes
    Workers: 10 concurrent threads
    Operations per worker: 10000 (set + get)
    Failure: Any cache inconsistency detected
    Race conditions detected:
    - Get returns None after set (data loss)
    - Incorrect value returned (corruption)
    - Deadlock (test hangs)
    """
    cache = GovernanceCache(max_size=10000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    errors = []
    iterations = 0

    def worker_operations(worker_id: int):
        """
        Worker function for concurrent cache operations.
        Each worker performs 10000 cache operations:
        - Set key with unique worker_id to avoid conflicts
        - Get key immediately after set
        - Validate value is not None (cache consistency check)
        Raises AssertionError if cache inconsistency detected.
        """
        for i in range(10000):
            # Concurrent write with unique key (worker_id avoids conflicts)
            cache.set(
                agent_id=f"agent_{worker_id}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

            # Concurrent read
            result = cache.get(
                agent_id=f"agent_{worker_id}_{i}",
                action_type="test_action"
            )

            # Validate cache consistency (get should not return None after set)
            if result is None:
                raise AssertionError(
                    f"Cache inconsistency: key agent_{worker_id}_{i} not found after write"
                )

        return f"Worker {worker_id} completed"

    start_time = time.time()

    # Run concurrent workers for 30 minutes
    while time.time() - start_time < 1800:
        # Submit 10 concurrent workers
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(worker_operations, worker_id)
                for worker_id in range(10)
            ]

            # Wait for all workers to complete, collect errors
            for future in as_completed(futures):
                try:
                    result = future.result()
                    print(result)
                except AssertionError as e:
                    errors.append(str(e))
                except Exception as e:
                    errors.append(f"Unexpected error: {e}")

        iterations += 1

        # Force garbage collection every 10 iterations
        if iterations % 10 == 0:
            enable_gc_control["collect"]()

        # Log memory growth every 60 iterations
        if iterations % 60 == 0:
            current_memory = process.memory_info().rss / 1024 / 1024
            memory_growth = current_memory - initial_memory

            print(
                f"Iteration {iterations}: "
                f"Memory growth = {memory_growth:.2f}MB, "
                f"Errors = {len(errors)}"
            )

            # Fail fast if memory growth > 500MB
            if memory_growth > soak_test_config["fail_fast_threshold_mb"]:
                pytest.fail(
                    f"FAIL-FAST: Memory leak detected - {memory_growth:.2f}MB growth "
                    f"(threshold: {soak_test_config['fail_fast_threshold_mb']}MB)"
                )

    # Final validation: zero cache errors
    assert len(errors) == 0, (
        f"Cache inconsistencies detected: {len(errors)} errors\n"
        f"Errors: {errors[:5]}"  # Show first 5 errors
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, zero cache errors")


@pytest.mark.soak
@pytest.mark.timeout(900)  # 15 minutes
def test_cache_ttl_expiration_15min(memory_monitor, enable_gc_control):
    """
    Soak test for cache TTL expiration (15 minutes).
    Validates:
    - Cache entries expire after TTL (60 seconds)
    - Cache doesn't grow unbounded (TTL prevents memory leaks)
    - Expired entries are not returned (get returns None)
    Test pattern:
    - Create cache with TTL=60 seconds
    - Add 1000 entries per iteration
    - Validate entries exist immediately after adding
    - Wait 70 seconds, validate entries are gone (expired)
    - Repeat for 15 minutes
    - Monitor memory usage
    Duration: 15 minutes
    TTL: 60 seconds
    Validation: Entries expire correctly, cache doesn't grow unbounded
    Purpose: Detect TTL bugs that cause unbounded cache growth.
    """
    cache = GovernanceCache(max_size=10000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    iterations = 0

    start_time = time.time()

    # Run TTL validation for 15 minutes
    while time.time() - start_time < 900:
        # Add 1000 entries
        for i in range(1000):
            cache.set(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

        # Validate entries exist immediately
        for i in range(100):  # Check 100 random entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            assert result is not None, f"Entry not found immediately after set: agent_{iterations}_{i}"

        print(f"Iteration {iterations}: Added 1000 entries, validated existence")

        # Wait 70 seconds (TTL=60, +10 buffer)
        time.sleep(70)

        # Validate entries are expired (get returns None)
        expired_count = 0
        for i in range(100):  # Check 100 random entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            if result is None:
                expired_count += 1

        # All entries should be expired
        assert expired_count == 100, (
            f"TTL expiration failed: {expired_count}/100 entries expired "
            f"(expected: 100)"
        )

        print(f"Iteration {iterations}: All entries expired correctly (100/100)")

        iterations += 1

        # Force garbage collection
        enable_gc_control["collect"]()

        # Log memory growth
        current_memory = process.memory_info().rss / 1024 / 1024
        memory_growth = current_memory - initial_memory

        print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

        # Fail fast if memory growth > 500MB (indicates TTL not working)
        if memory_growth > 500:
            pytest.fail(
                f"TTL expiration not working: cache grew {memory_growth:.2f}MB "
                f"(entries should be expiring)"
            )

    # Final validation: memory growth should be reasonable (< 200MB for 15min)
    final_memory = process.memory_info().rss / 1024 / 1024
    memory_growth = final_memory - initial_memory

    assert memory_growth < 200, (
        f"Cache grew unbounded: {memory_growth:.2f}MB growth over 15 minutes "
        f"(TTL expiration may not be working)"
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, TTL expiration working correctly")


@pytest.mark.soak
@pytest.mark.timeout(1800)  # 30 minutes
def test_cache_lru_eviction_stability_30min(memory_monitor, enable_gc_control):
    """
    Soak test for cache LRU eviction stability (30 minutes).
    Validates:
    - LRU eviction works correctly (cache doesn't exceed max_size)
    - Evicted entries are not returned (get returns None)
    - Cache performance remains stable under load
    Test pattern:
    - Create cache with max_size=1000
    - Add 2000 entries (exceeds max_size, triggers LRU eviction)
    - Validate cache size <= max_size
    - Verify oldest entries are evicted (get returns None)
    - Repeat for 30 minutes
    Duration: 30 minutes
    Max size: 1000 entries
    Validation: LRU eviction works, cache size stable
    Purpose: Detect LRU eviction bugs that cause unbounded cache growth.
    """
    cache = GovernanceCache(max_size=1000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    iterations = 0

    start_time = time.time()

    # Run LRU eviction validation for 30 minutes
    while time.time() - start_time < 1800:
        # Add 2000 entries (exceeds max_size=1000)
        for i in range(2000):
            cache.set(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

        # Validate oldest entries are evicted (first 1000 should be gone)
        evicted_count = 0
        for i in range(1000):  # Check oldest 1000 entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            if result is None:
                evicted_count += 1

        # At least some entries should be evicted (LRU working)
        # Not all 1000 will be evicted (TTL may not have expired yet)
        # But at least 500 should be evicted due to LRU
        assert evicted_count >= 500, (
            f"LRU eviction not working: only {evicted_count}/1000 entries evicted "
            f"(expected: >= 500)"
        )

        print(f"Iteration {iterations}: LRU eviction working ({evicted_count}/1000 evicted)")

        iterations += 1

        # Force garbage collection
        enable_gc_control["collect"]()

        # Log memory growth
        current_memory = process.memory_info().rss / 1024 / 1024
        memory_growth = current_memory - initial_memory

        print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

        # Fail fast if memory growth > 500MB (indicates LRU not working)
        if memory_growth > 500:
            pytest.fail(
                f"LRU eviction not working: cache grew {memory_growth:.2f}MB "
                f"(entries should be evicted)"
            )

    # Final validation: memory growth should be reasonable
    final_memory = process.memory_info().rss / 1024 / 1024
    memory_growth = final_memory - initial_memory

    assert memory_growth < 200, (
        f"Cache grew unbounded: {memory_growth:.2f}MB growth over 30 minutes "
        f"(LRU eviction may not be working)"
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, LRU eviction working correctly")

	"""
	Soak Tests for Cache Stability Under Concurrent Load

	Extended duration tests (30 minutes) to validate cache behavior:
	- Concurrent operations (10 workers, 10000 ops each)
	- Cache consistency (no race conditions, no data loss)
	- TTL expiration (entries expire correctly)
	- Cache unbounded growth detection

	Cache issues detected:
	- Race conditions in concurrent get/set operations
	- Cache corruption (data loss, incorrect values)
	- TTL not working (cache grows unbounded)
	- Deadlocks under concurrent load

	Tests use ThreadPoolExecutor for realistic concurrent access patterns.
	"""

	import gc
	import pytest
	import time
	from concurrent.futures import ThreadPoolExecutor, as_completed

	from core.governance_cache import GovernanceCache


	@pytest.mark.soak
	@pytest.mark.timeout(1800) # 30 minutes
	def test_cache_concurrent_operations_30min(
	memory_monitor,
	enable_gc_control,
	soak_test_config
	):
	"""
	Soak test for cache stability under concurrent operations (30 minutes).

	Validates:
	- Cache remains consistent under concurrent load (10 workers)
	- No race conditions or deadlocks
	- No cache corruption (data loss, incorrect values)
	- Zero cache errors (get returns None after set)

	Test pattern:
	- Run for 30 minutes (1800 seconds)
	- Use ThreadPoolExecutor with 10 workers
	- Each worker performs 10000 cache operations (set + get)
	- Validate cache consistency: if get returns None after set, fail
	- Log worker completions
	- Force GC every 10 iterations

	Duration: 30 minutes
	Workers: 10 concurrent threads
	Operations per worker: 10000 (set + get)
	Failure: Any cache inconsistency detected

	Race conditions detected:
	- Get returns None after set (data loss)
	- Incorrect value returned (corruption)
	- Deadlock (test hangs)
	"""
	cache = GovernanceCache(max_size=10000, ttl_seconds=60)
	process = memory_monitor["process"]
	initial_memory = memory_monitor["initial_memory_mb"]

	errors = []
	iterations = 0

	def worker_operations(worker_id: int):
	"""
	Worker function for concurrent cache operations.

	Each worker performs 10000 cache operations:
	- Set key with unique worker_id to avoid conflicts
	- Get key immediately after set
	- Validate value is not None (cache consistency check)

	Raises AssertionError if cache inconsistency detected.
	"""
	for i in range(10000):
	# Concurrent write with unique key (worker_id avoids conflicts)
	cache.set(
	agent_id=f"agent_{worker_id}_{i}",
	action_type="test_action",
	data={"allowed": True, "maturity_level": "AUTONOMOUS"}
	)

	# Concurrent read
	result = cache.get(
	agent_id=f"agent_{worker_id}_{i}",
	action_type="test_action"
	)

	# Validate cache consistency (get should not return None after set)
	if result is None:
	raise AssertionError(
	f"Cache inconsistency: key agent_{worker_id}_{i} not found after write"
	)

	return f"Worker {worker_id} completed"

	start_time = time.time()

	# Run concurrent workers for 30 minutes
	while time.time() - start_time < 1800:
	# Submit 10 concurrent workers
	with ThreadPoolExecutor(max_workers=10) as executor:
	futures = [
	executor.submit(worker_operations, worker_id)
	for worker_id in range(10)
	]

	# Wait for all workers to complete, collect errors
	for future in as_completed(futures):
	try:
	result = future.result()
	print(result)
	except AssertionError as e:
	errors.append(str(e))
	except Exception as e:
	errors.append(f"Unexpected error: {e}")

	iterations += 1

	# Force garbage collection every 10 iterations
	if iterations % 10 == 0:
	enable_gc_control["collect"]()

	# Log memory growth every 60 iterations
	if iterations % 60 == 0:
	current_memory = process.memory_info().rss / 1024 / 1024
	memory_growth = current_memory - initial_memory

	print(
	f"Iteration {iterations}: "
	f"Memory growth = {memory_growth:.2f}MB, "
	f"Errors = {len(errors)}"
	)

	# Fail fast if memory growth > 500MB
	if memory_growth > soak_test_config["fail_fast_threshold_mb"]:
	pytest.fail(
	f"FAIL-FAST: Memory leak detected - {memory_growth:.2f}MB growth "
	f"(threshold: {soak_test_config['fail_fast_threshold_mb']}MB)"
	)

	# Final validation: zero cache errors
	assert len(errors) == 0, (
	f"Cache inconsistencies detected: {len(errors)} errors\n"
	f"Errors: {errors[:5]}" # Show first 5 errors
	)

	print(f"\n✅ Soak test complete: {iterations} iterations, zero cache errors")


	@pytest.mark.soak
	@pytest.mark.timeout(900) # 15 minutes
	def test_cache_ttl_expiration_15min(memory_monitor, enable_gc_control):
	"""
	Soak test for cache TTL expiration (15 minutes).

	Validates:
	- Cache entries expire after TTL (60 seconds)
	- Cache doesn't grow unbounded (TTL prevents memory leaks)
	- Expired entries are not returned (get returns None)

	Test pattern:
	- Create cache with TTL=60 seconds
	- Add 1000 entries per iteration
	- Validate entries exist immediately after adding
	- Wait 70 seconds, validate entries are gone (expired)
	- Repeat for 15 minutes
	- Monitor memory usage

	Duration: 15 minutes
	TTL: 60 seconds
	Validation: Entries expire correctly, cache doesn't grow unbounded

	Purpose: Detect TTL bugs that cause unbounded cache growth.
	"""
	cache = GovernanceCache(max_size=10000, ttl_seconds=60)
	process = memory_monitor["process"]
	initial_memory = memory_monitor["initial_memory_mb"]

	iterations = 0

	start_time = time.time()

	# Run TTL validation for 15 minutes
	while time.time() - start_time < 900:
	# Add 1000 entries
	for i in range(1000):
	cache.set(
	agent_id=f"agent_{iterations}_{i}",
	action_type="test_action",
	data={"allowed": True, "maturity_level": "AUTONOMOUS"}
	)

	# Validate entries exist immediately
	for i in range(100): # Check 100 random entries
	result = cache.get(
	agent_id=f"agent_{iterations}_{i}",
	action_type="test_action"
	)
	assert result is not None, f"Entry not found immediately after set: agent_{iterations}_{i}"

	print(f"Iteration {iterations}: Added 1000 entries, validated existence")

	# Wait 70 seconds (TTL=60, +10 buffer)
	time.sleep(70)

	# Validate entries are expired (get returns None)
	expired_count = 0
	for i in range(100): # Check 100 random entries
	result = cache.get(
	agent_id=f"agent_{iterations}_{i}",
	action_type="test_action"
	)
	if result is None:
	expired_count += 1

	# All entries should be expired
	assert expired_count == 100, (
	f"TTL expiration failed: {expired_count}/100 entries expired "
	f"(expected: 100)"
	)

	print(f"Iteration {iterations}: All entries expired correctly (100/100)")

	iterations += 1

	# Force garbage collection
	enable_gc_control["collect"]()

	# Log memory growth
	current_memory = process.memory_info().rss / 1024 / 1024
	memory_growth = current_memory - initial_memory

	print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

	# Fail fast if memory growth > 500MB (indicates TTL not working)
	if memory_growth > 500:
	pytest.fail(
	f"TTL expiration not working: cache grew {memory_growth:.2f}MB "
	f"(entries should be expiring)"
	)

	# Final validation: memory growth should be reasonable (< 200MB for 15min)
	final_memory = process.memory_info().rss / 1024 / 1024
	memory_growth = final_memory - initial_memory

	assert memory_growth < 200, (
	f"Cache grew unbounded: {memory_growth:.2f}MB growth over 15 minutes "
	f"(TTL expiration may not be working)"
	)

	print(f"\n✅ Soak test complete: {iterations} iterations, TTL expiration working correctly")


	@pytest.mark.soak
	@pytest.mark.timeout(1800) # 30 minutes
	def test_cache_lru_eviction_stability_30min(memory_monitor, enable_gc_control):
	"""
	Soak test for cache LRU eviction stability (30 minutes).

	Validates:
	- LRU eviction works correctly (cache doesn't exceed max_size)
	- Evicted entries are not returned (get returns None)
	- Cache performance remains stable under load

	Test pattern:
	- Create cache with max_size=1000
	- Add 2000 entries (exceeds max_size, triggers LRU eviction)
	- Validate cache size <= max_size
	- Verify oldest entries are evicted (get returns None)
	- Repeat for 30 minutes

	Duration: 30 minutes
	Max size: 1000 entries
	Validation: LRU eviction works, cache size stable

	Purpose: Detect LRU eviction bugs that cause unbounded cache growth.
	"""
	cache = GovernanceCache(max_size=1000, ttl_seconds=60)
	process = memory_monitor["process"]
	initial_memory = memory_monitor["initial_memory_mb"]

	iterations = 0

	start_time = time.time()

	# Run LRU eviction validation for 30 minutes
	while time.time() - start_time < 1800:
	# Add 2000 entries (exceeds max_size=1000)
	for i in range(2000):
	cache.set(
	agent_id=f"agent_{iterations}_{i}",
	action_type="test_action",
	data={"allowed": True, "maturity_level": "AUTONOMOUS"}
	)

	# Validate oldest entries are evicted (first 1000 should be gone)
	evicted_count = 0
	for i in range(1000): # Check oldest 1000 entries
	result = cache.get(
	agent_id=f"agent_{iterations}_{i}",
	action_type="test_action"
	)
	if result is None:
	evicted_count += 1

	# At least some entries should be evicted (LRU working)
	# Not all 1000 will be evicted (TTL may not have expired yet)
	# But at least 500 should be evicted due to LRU
	assert evicted_count >= 500, (
	f"LRU eviction not working: only {evicted_count}/1000 entries evicted "
	f"(expected: >= 500)"
	)

	print(f"Iteration {iterations}: LRU eviction working ({evicted_count}/1000 evicted)")

	iterations += 1

	# Force garbage collection
	enable_gc_control["collect"]()

	# Log memory growth
	current_memory = process.memory_info().rss / 1024 / 1024
	memory_growth = current_memory - initial_memory

	print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

	# Fail fast if memory growth > 500MB (indicates LRU not working)
	if memory_growth > 500:
	pytest.fail(
	f"LRU eviction not working: cache grew {memory_growth:.2f}MB "
	f"(entries should be evicted)"
	)

	# Final validation: memory growth should be reasonable
	final_memory = process.memory_info().rss / 1024 / 1024
	memory_growth = final_memory - initial_memory

	assert memory_growth < 200, (
	f"Cache grew unbounded: {memory_growth:.2f}MB growth over 30 minutes "
	f"(LRU eviction may not be working)"
	)

	print(f"\n✅ Soak test complete: {iterations} iterations, LRU eviction working correctly")