# NoteFlow Resilience Test Plan

**Sprint**: Sprint 15 - Platform Hardening
**Version**: 1.0
**Date**: 2025-12-29

---

## Executive Summary

This document defines a rigorous testing and hardening strategy for NoteFlow's teardown, resource management, and crash resilience capabilities. Based on comprehensive codebase analysis, we've identified:

- **22 critical gaps** requiring new tests
- **4 high-priority code fixes** needed before tests can pass
- **3 categories of risk**: Teardown, Resource, Crash Recovery

---

## Current State Assessment

### Strengths (Already Well-Tested)

| Area | Coverage | Key Tests |
|------|----------|-----------|
| State cleanup cycles | ✅ Excellent | `test_concurrency_stress.py` (100+ cycles) |
| Recovery service | ✅ Good | `test_recovery_service.py` (meeting + job recovery) |
| Transaction rollback | ✅ Good | `test_transaction_boundaries.py` |
| Audio file integrity | ✅ Good | `test_audio_integrity.py` (truncation, corruption) |
| Segmenter fuzz | ✅ Good | `test_segmenter_fuzz.py` (1000 patterns) |
| Error handling | ✅ Good | `test_error_handling.py` (70+ scenarios) |
| Webhook retry | ✅ Partial | `test_executor.py` (delivery, HMAC) |

### Critical Gaps (Not Tested)

| Category | Gap | Risk Level |
|----------|-----|------------|
| Teardown | gRPC stream cancellation cleanup | 🔴 Critical |
| Teardown | Signal handler (SIGTERM/SIGKILL) resource cleanup | 🔴 Critical |
| Resource | File descriptor leak detection | 🔴 Critical |
| Resource | Memory profiling under load | 🟡 High |
| Resource | Thread/coroutine leak detection | 🟡 High |
| Resource | Database connection pool exhaustion | 🔴 Critical |
| Crash | Partial audio segment loss | 🔴 Critical |
| Crash | Concurrent crash recovery race | 🟡 High |
| Crash | Long-running operation timeout | 🟡 High |
| Crash | Database unavailability handling | 🔴 Critical |

---

## Part 1: Code Hardening (Pre-Test Fixes)

Before implementing tests, these issues must be fixed to prevent false failures:

### Fix 1: Diarization Session GPU Memory Leak

**File**: `src/noteflow/infrastructure/diarization/session.py:145`

```python
# CURRENT
def close(self) -> None:
    if self._closed:
        return
    self._closed = True
    self._turns.clear()

# PROPOSED
def close(self) -> None:
    if self._closed:
        return
    self._closed = True
    self._turns.clear()
    # Explicitly release pipeline reference to allow GC
    self._pipeline = None  # <-- ADD THIS
```

### Fix 2: Webhook Executor Close Timeout

**File**: `src/noteflow/infrastructure/webhooks/executor.py:284`

```python
# CURRENT
async def close(self) -> None:
    if self._client is not None:
        await self._client.aclose()

# PROPOSED
async def close(self) -> None:
    if self._client is not None:
        try:
            await asyncio.wait_for(self._client.aclose(), timeout=5.0)
        except asyncio.TimeoutError:
            logger.warning("Webhook client close timed out, forcing shutdown")
        finally:
            self._client = None
```

### Fix 3: Audio Writer Thread Join Timeout

**File**: `src/noteflow/infrastructure/audio/writer.py:244`

```python
# CURRENT
self._flush_thread.join(timeout=1.0)

# PROPOSED
self._flush_thread.join(timeout=3.0)  # Increase from 1.0s to 3.0s
```

### Fix 4: Diarization Job Status on Task Cancel

**File**: `src/noteflow/grpc/service.py:388-395`

```python
# CURRENT - only marks failed on orderly shutdown
for job_id, task in list(self._diarization_tasks.items()):
    if not task.done():
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task

# PROPOSED - mark failed on cancellation
for job_id, task in list(self._diarization_tasks.items()):
    if not task.done():
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task
        # Mark job as failed if it was cancelled
        if job_id in self._diarization_jobs:
            job = self._diarization_jobs[job_id]
            if job.status in (DiarizationJobStatus.QUEUED, DiarizationJobStatus.RUNNING):
                job.status = DiarizationJobStatus.FAILED
                job.error_message = "ERR_TASK_CANCELLED"
```

---

## Part 2: New Test Suites

### Test Suite 1: gRPC Stream Lifecycle (`tests/grpc/test_stream_lifecycle.py`)

**Purpose**: Validate resource cleanup when streams terminate abnormally.

```python
"""Tests for gRPC streaming lifecycle edge cases."""
import asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.fixture
def mock_context():
    """Mock gRPC context with cancellation support."""
    context = MagicMock()
    context.cancelled = MagicMock(return_value=False)
    context.add_done_callback = MagicMock()
    return context

class TestStreamCancellation:
    """Test cleanup when client disconnects mid-stream."""

    async def test_client_disconnect_during_audio_processing(
        self, memory_servicer, mock_context
    ):
        """Verify state cleanup when client cancels stream."""
        meeting_id = "test-cancel-001"

        # Start stream
        await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)

        # Verify state exists
        assert meeting_id in memory_servicer._active_streams
        assert meeting_id in memory_servicer._vad_instances

        # Simulate cancellation
        memory_servicer._cleanup_streaming_state(meeting_id)
        memory_servicer._close_audio_writer(meeting_id)
        memory_servicer._active_streams.discard(meeting_id)

        # Verify complete cleanup
        assert meeting_id not in memory_servicer._active_streams
        assert meeting_id not in memory_servicer._vad_instances
        assert meeting_id not in memory_servicer._segmenters
        assert meeting_id not in memory_servicer._partial_buffers

    async def test_grpc_deadline_exceeded_cleanup(self, memory_servicer):
        """Verify cleanup when gRPC deadline is exceeded."""
        meeting_id = "test-deadline-001"

        # Initialize stream
        await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)

        # Simulate deadline exceeded (context.time_remaining() <= 0)
        # Stream should still clean up
        memory_servicer._cleanup_streaming_state(meeting_id)

        assert meeting_id not in memory_servicer._partial_buffers

    async def test_concurrent_cancel_and_stop(self, memory_servicer):
        """Verify cleanup when cancel and stop race."""
        meeting_id = "test-race-001"

        await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)

        # Add to stop_requested AND cancel simultaneously
        memory_servicer._stop_requested.add(meeting_id)

        # Cleanup should be idempotent
        memory_servicer._cleanup_streaming_state(meeting_id)
        memory_servicer._cleanup_streaming_state(meeting_id)  # Second call

        assert meeting_id not in memory_servicer._active_streams

class TestStreamInitializationFailure:
    """Test cleanup when stream initialization fails partway."""

    async def test_cleanup_on_vad_init_failure(self, memory_servicer):
        """Verify no state leaked if VAD initialization fails."""
        meeting_id = "test-vad-fail-001"

        # Patch to make VAD init fail
        with pytest.raises(RuntimeError, match="VAD init failed"):
            # Force failure during init
            memory_servicer._vad_instances[meeting_id] = None
            raise RuntimeError("VAD init failed")

        # Should not leave partial state
        # (In reality, _init_stream_for_meeting should clean up on failure)

    async def test_cleanup_on_audio_writer_failure(self, memory_servicer):
        """Verify cleanup if audio writer fails to open."""
        meeting_id = "test-writer-fail-001"

        # This tests the case where meeting state is added but writer fails
        memory_servicer._active_streams.add(meeting_id)

        # Simulate writer failure - cleanup should still work
        memory_servicer._cleanup_streaming_state(meeting_id)
        memory_servicer._active_streams.discard(meeting_id)

        assert meeting_id not in memory_servicer._active_streams
```

### Test Suite 2: Resource Leak Detection (`tests/stress/test_resource_leaks.py`)

**Purpose**: Detect file descriptor, memory, and coroutine leaks under load.

```python
"""Resource leak detection tests."""
import asyncio
import gc
import os
import pytest
import resource
import sys
import tracemalloc

from noteflow.grpc.service import NoteFlowServicer

@pytest.fixture
def fd_count_before():
    """Capture file descriptor count before test."""
    return len(os.listdir(f"/proc/{os.getpid()}/fd"))

class TestFileDescriptorLeaks:
    """Detect file descriptor leaks."""

    @pytest.mark.slow
    async def test_streaming_fd_cleanup(self, memory_servicer, fd_count_before):
        """Verify file descriptors released after streaming cycles."""
        initial_fds = fd_count_before

        # Run 50 streaming init/cleanup cycles
        for i in range(50):
            meeting_id = f"fd-test-{i:03d}"
            await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)
            memory_servicer._cleanup_streaming_state(meeting_id)
            memory_servicer._close_audio_writer(meeting_id)
            memory_servicer._active_streams.discard(meeting_id)

        # Force garbage collection
        gc.collect()
        await asyncio.sleep(0.1)  # Allow async cleanup

        final_fds = len(os.listdir(f"/proc/{os.getpid()}/fd"))

        # Allow small variance (logging, etc.) but detect major leaks
        assert final_fds <= initial_fds + 10, (
            f"File descriptor leak: started with {initial_fds}, ended with {final_fds}"
        )

    @pytest.mark.slow
    async def test_audio_writer_fd_cleanup(self, memory_servicer, fd_count_before, tmp_path):
        """Verify audio writer closes file handles."""
        from noteflow.infrastructure.audio.writer import MeetingAudioWriter
        from noteflow.infrastructure.security.crypto import Crypto

        crypto = Crypto.generate_new()
        initial_fds = fd_count_before

        # Create and close 20 writers
        for i in range(20):
            writer = MeetingAudioWriter(crypto, tmp_path, buffer_size=1024)
            writer.open(f"meeting-{i}", b"0" * 32, sample_rate=16000, channels=1)
            writer.close()

        gc.collect()
        await asyncio.sleep(0.1)

        final_fds = len(os.listdir(f"/proc/{os.getpid()}/fd"))
        assert final_fds <= initial_fds + 5

class TestMemoryLeaks:
    """Detect memory leaks under load."""

    @pytest.mark.slow
    async def test_streaming_memory_stability(self, memory_servicer):
        """Verify memory doesn't grow unbounded during streaming cycles."""
        tracemalloc.start()

        # Baseline
        gc.collect()
        snapshot1 = tracemalloc.take_snapshot()

        # Run 100 cycles
        for i in range(100):
            meeting_id = f"mem-test-{i:03d}"
            await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)
            memory_servicer._cleanup_streaming_state(meeting_id)

        gc.collect()
        snapshot2 = tracemalloc.take_snapshot()

        # Compare
        top_stats = snapshot2.compare_to(snapshot1, "lineno")

        # Log top memory consumers
        for stat in top_stats[:5]:
            if stat.size_diff > 1024 * 1024:  # > 1MB growth
                pytest.fail(f"Memory leak detected: {stat}")

        tracemalloc.stop()

class TestCoroutineLeaks:
    """Detect leaked coroutines/tasks."""

    async def test_no_orphaned_tasks_after_shutdown(self, memory_servicer):
        """Verify no tasks remain after servicer shutdown."""
        # Create some diarization tasks
        for i in range(5):
            task = asyncio.create_task(asyncio.sleep(10))
            memory_servicer._diarization_tasks[f"job-{i}"] = task

        # Shutdown
        await memory_servicer.shutdown()

        # Verify all tasks cancelled and cleared
        assert len(memory_servicer._diarization_tasks) == 0

        # Check no orphaned tasks in event loop
        pending = [t for t in asyncio.all_tasks() if not t.done() and t != asyncio.current_task()]
        # Filter out test framework tasks
        orphaned = [t for t in pending if "diarization" in str(t)]

        assert len(orphaned) == 0, f"Orphaned tasks: {orphaned}"
```

### Test Suite 3: Database Resilience (`tests/integration/test_database_resilience.py`)

**Purpose**: Test database failure scenarios and recovery.

```python
"""Database resilience and connection pool tests."""
import asyncio
import pytest
from unittest.mock import AsyncMock, patch

from sqlalchemy.exc import OperationalError

class TestConnectionPoolExhaustion:
    """Test behavior when connection pool is exhausted."""

    @pytest.mark.integration
    async def test_concurrent_operations_within_pool_limit(
        self, db_session_factory
    ):
        """Verify system handles concurrent ops within pool size."""
        from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork

        pool_size = 5  # Match settings.db_pool_size

        async def concurrent_operation(idx: int):
            async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
                # Simulate work
                await asyncio.sleep(0.1)
                return idx

        # Run exactly pool_size concurrent operations
        tasks = [concurrent_operation(i) for i in range(pool_size)]
        results = await asyncio.gather(*tasks)

        assert len(results) == pool_size

    @pytest.mark.integration
    async def test_graceful_handling_beyond_pool_limit(
        self, db_session_factory
    ):
        """Verify operations queue when pool is exhausted (not crash)."""
        from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork

        pool_size = 5
        overflow = 3

        async def long_operation(idx: int):
            async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
                await asyncio.sleep(0.5)  # Hold connection
                return idx

        # Run more operations than pool size
        tasks = [long_operation(i) for i in range(pool_size + overflow)]

        # Should complete (with queueing), not crash
        results = await asyncio.wait_for(
            asyncio.gather(*tasks),
            timeout=10.0  # Allow time for queueing
        )

        assert len(results) == pool_size + overflow

class TestDatabaseUnavailability:
    """Test handling of database unavailability."""

    @pytest.mark.integration
    async def test_recovery_on_reconnect(self, db_session_factory):
        """Verify recovery after temporary database disconnect."""
        from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork

        # First operation succeeds
        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            pass  # Connection established

        # Simulate brief unavailability - pool_pre_ping should handle
        # In production, this tests automatic reconnection

        # Second operation should succeed after pool_pre_ping detects stale connection
        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            pass  # Should reconnect

    @pytest.mark.integration
    async def test_rollback_on_commit_failure(self, db_session_factory):
        """Verify rollback when commit fails."""
        from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork
        from noteflow.domain.entities import Meeting, MeetingId

        meeting = Meeting(id=MeetingId.generate(), title="Rollback Test")

        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            await uow.meetings.add(meeting)
            # Don't commit - should rollback on exit

        # Verify not persisted
        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            retrieved = await uow.meetings.get(meeting.id.value)
            assert retrieved is None

class TestTransactionIsolation:
    """Test transaction isolation under concurrent modifications."""

    @pytest.mark.integration
    async def test_concurrent_meeting_updates(self, db_session_factory):
        """Verify concurrent updates don't corrupt data."""
        from noteflow.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork
        from noteflow.domain.entities import Meeting, MeetingId

        meeting_id = MeetingId.generate()
        meeting = Meeting(id=meeting_id, title="Concurrent Test")

        # Create meeting
        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            await uow.meetings.add(meeting)
            await uow.commit()

        # Concurrent updates
        async def update_title(suffix: str):
            async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
                m = await uow.meetings.get(meeting_id.value)
                m.title = f"Updated-{suffix}"
                await uow.commit()

        # Run concurrent updates
        await asyncio.gather(
            update_title("A"),
            update_title("B"),
            update_title("C"),
        )

        # Verify final state is consistent (one of A, B, or C)
        async with SqlAlchemyUnitOfWork(db_session_factory, ".") as uow:
            final = await uow.meetings.get(meeting_id.value)
            assert final.title.startswith("Updated-")
```

### Test Suite 4: Signal Handler Tests (`tests/integration/test_signal_handling.py`)

**Purpose**: Verify graceful shutdown on SIGTERM/SIGINT.

```python
"""Signal handling and graceful shutdown tests."""
import asyncio
import os
import signal
import subprocess
import sys
import time
import pytest

class TestGracefulShutdown:
    """Test server graceful shutdown on signals."""

    @pytest.mark.integration
    @pytest.mark.slow
    async def test_sigterm_graceful_shutdown(self, tmp_path):
        """Verify SIGTERM triggers graceful shutdown."""
        # Start server in subprocess
        env = os.environ.copy()
        env["NOTEFLOW_DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
        env["NOTEFLOW_MEETINGS_DIR"] = str(tmp_path)

        proc = subprocess.Popen(
            [sys.executable, "-m", "noteflow.grpc.server", "--port", "50052"],
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        try:
            # Wait for server to start
            await asyncio.sleep(2.0)
            assert proc.poll() is None, "Server failed to start"

            # Send SIGTERM
            proc.send_signal(signal.SIGTERM)

            # Wait for graceful shutdown (should be < 10 seconds)
            try:
                proc.wait(timeout=10.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                pytest.fail("Server did not shutdown gracefully within timeout")

            # Check exit code (0 = graceful, non-zero = error)
            assert proc.returncode == 0, f"Server exited with code {proc.returncode}"

        finally:
            if proc.poll() is None:
                proc.kill()

    @pytest.mark.integration
    @pytest.mark.slow
    async def test_sigint_graceful_shutdown(self, tmp_path):
        """Verify SIGINT (Ctrl+C) triggers graceful shutdown."""
        env = os.environ.copy()
        env["NOTEFLOW_DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
        env["NOTEFLOW_MEETINGS_DIR"] = str(tmp_path)

        proc = subprocess.Popen(
            [sys.executable, "-m", "noteflow.grpc.server", "--port", "50053"],
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        try:
            await asyncio.sleep(2.0)
            assert proc.poll() is None

            # Send SIGINT (Ctrl+C)
            proc.send_signal(signal.SIGINT)

            try:
                proc.wait(timeout=10.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                pytest.fail("Server did not shutdown on SIGINT")

            assert proc.returncode == 0

        finally:
            if proc.poll() is None:
                proc.kill()

    @pytest.mark.integration
    async def test_shutdown_with_active_streams(self, memory_servicer):
        """Verify shutdown cleans up active streams."""
        # Create active streams
        for i in range(3):
            meeting_id = f"shutdown-stream-{i}"
            await memory_servicer._init_stream_for_meeting(meeting_id, 16000, 1)

        assert len(memory_servicer._active_streams) == 3

        # Shutdown
        await memory_servicer.shutdown()

        # Verify all streams cleaned up
        assert len(memory_servicer._active_streams) == 0
        assert len(memory_servicer._audio_writers) == 0
        assert len(memory_servicer._vad_instances) == 0
```

### Test Suite 5: Crash Recovery (`tests/integration/test_crash_scenarios.py`)

**Purpose**: Verify recovery from various crash scenarios.

```python
"""Crash scenario recovery tests."""
import asyncio
import pytest
from datetime import datetime, timezone

from noteflow.domain.entities import Meeting, MeetingId, MeetingState
from noteflow.application.services import RecoveryService

class TestMeetingCrashRecovery:
    """Test meeting recovery after crashes."""

    @pytest.mark.integration
    async def test_recording_meeting_recovery(self, db_uow_factory, tmp_path):
        """Verify RECORDING meeting marked ERROR after crash."""
        # Create meeting in RECORDING state
        meeting = Meeting(id=MeetingId.generate(), title="Crash Test")
        meeting.start_recording()  # State = RECORDING

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.commit()

        # Simulate crash recovery
        recovery_service = RecoveryService(db_uow_factory, tmp_path)
        result = await recovery_service.recover_all()

        # Verify recovered
        assert result.meetings_recovered == 1

        async with db_uow_factory() as uow:
            recovered = await uow.meetings.get(meeting.id.value)
            assert recovered.state == MeetingState.ERROR
            assert recovered.metadata.get("crash_recovered") is True
            assert recovered.metadata.get("crash_previous_state") == "RECORDING"

    @pytest.mark.integration
    async def test_stopping_meeting_recovery(self, db_uow_factory, tmp_path):
        """Verify STOPPING meeting marked ERROR after crash."""
        meeting = Meeting(id=MeetingId.generate(), title="Stopping Crash")
        meeting.start_recording()
        meeting._state = MeetingState.STOPPING  # Simulate mid-stop crash

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)
        result = await recovery_service.recover_all()

        assert result.meetings_recovered == 1

    @pytest.mark.integration
    async def test_audio_validation_during_recovery(self, db_uow_factory, tmp_path):
        """Verify audio file validation during recovery."""
        meeting = Meeting(id=MeetingId.generate(), title="Audio Validation")
        meeting.start_recording()
        # Don't create audio files - should mark audio_valid=False

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)
        await recovery_service.recover_all()

        async with db_uow_factory() as uow:
            recovered = await uow.meetings.get(meeting.id.value)
            # Should be marked with audio validation status
            assert "audio_valid" in recovered.metadata or "audio_error" in recovered.metadata

class TestDiarizationJobRecovery:
    """Test diarization job recovery after crashes."""

    @pytest.mark.integration
    async def test_running_job_marked_failed(self, db_uow_factory, tmp_path):
        """Verify RUNNING diarization jobs marked FAILED."""
        from noteflow.domain.entities import DiarizationJob, DiarizationJobStatus

        meeting = Meeting(id=MeetingId.generate(), title="Job Recovery")
        job = DiarizationJob(
            job_id="job-crash-test",
            meeting_id=meeting.id.value,
            status=DiarizationJobStatus.RUNNING,
        )

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.diarization_jobs.add(job)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)
        result = await recovery_service.recover_all()

        assert result.jobs_recovered == 1

        async with db_uow_factory() as uow:
            recovered_job = await uow.diarization_jobs.get(job.job_id)
            assert recovered_job.status == DiarizationJobStatus.FAILED
            assert "ERR_SERVER_RESTARTED" in (recovered_job.error_message or "")

    @pytest.mark.integration
    async def test_queued_job_marked_failed(self, db_uow_factory, tmp_path):
        """Verify QUEUED diarization jobs marked FAILED."""
        from noteflow.domain.entities import DiarizationJob, DiarizationJobStatus

        meeting = Meeting(id=MeetingId.generate(), title="Queued Job")
        job = DiarizationJob(
            job_id="job-queued-test",
            meeting_id=meeting.id.value,
            status=DiarizationJobStatus.QUEUED,
        )

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.diarization_jobs.add(job)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)
        result = await recovery_service.recover_all()

        assert result.jobs_recovered >= 1

class TestConcurrentRecovery:
    """Test recovery race conditions."""

    @pytest.mark.integration
    async def test_recovery_is_idempotent(self, db_uow_factory, tmp_path):
        """Verify running recovery twice doesn't corrupt state."""
        meeting = Meeting(id=MeetingId.generate(), title="Idempotent Test")
        meeting.start_recording()

        async with db_uow_factory() as uow:
            await uow.meetings.add(meeting)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)

        # Run recovery twice
        result1 = await recovery_service.recover_all()
        result2 = await recovery_service.recover_all()

        # Second run should find nothing to recover
        assert result1.meetings_recovered == 1
        assert result2.meetings_recovered == 0

    @pytest.mark.integration
    async def test_concurrent_recovery_calls(self, db_uow_factory, tmp_path):
        """Verify concurrent recovery calls don't corrupt data."""
        # Create multiple crashed meetings
        meetings = []
        async with db_uow_factory() as uow:
            for i in range(5):
                meeting = Meeting(id=MeetingId.generate(), title=f"Concurrent-{i}")
                meeting.start_recording()
                await uow.meetings.add(meeting)
                meetings.append(meeting)
            await uow.commit()

        recovery_service = RecoveryService(db_uow_factory, tmp_path)

        # Run recovery concurrently
        results = await asyncio.gather(
            recovery_service.recover_all(),
            recovery_service.recover_all(),
            recovery_service.recover_all(),
        )

        # Combined should be exactly 5 (one wins, others find nothing)
        total_recovered = sum(r.meetings_recovered for r in results)
        assert total_recovered == 5

        # Verify all in ERROR state
        async with db_uow_factory() as uow:
            for meeting in meetings:
                recovered = await uow.meetings.get(meeting.id.value)
                assert recovered.state == MeetingState.ERROR
```

---

## Part 3: Test Execution Plan

### Phase 1: Pre-Requisites (Week 1)

1. **Implement Code Fixes** (Fixes 1-4 above)
2. **Update conftest.py** with new fixtures:
   - `fd_count_before` - File descriptor baseline
   - `db_uow_factory` - Database UoW factory for integration tests
   - `memory_servicer` - Already exists, ensure fully configured

### Phase 2: Unit/Stress Tests (Week 2)

| Priority | Test Suite | Est. Time | Marker |
|----------|------------|-----------|--------|
| P0 | `test_stream_lifecycle.py` | 2 days | `@pytest.mark.asyncio` |
| P0 | `test_resource_leaks.py` | 3 days | `@pytest.mark.slow` |
| P1 | `test_database_resilience.py` | 2 days | `@pytest.mark.integration` |

### Phase 3: Integration Tests (Week 3)

| Priority | Test Suite | Est. Time | Marker |
|----------|------------|-----------|--------|
| P0 | `test_signal_handling.py` | 2 days | `@pytest.mark.integration` |
| P0 | `test_crash_scenarios.py` | 3 days | `@pytest.mark.integration` |
| P1 | Extend `test_e2e_streaming.py` | 2 days | `@pytest.mark.integration` |

### Phase 4: Load/Chaos Testing (Week 4)

| Priority | Test Suite | Est. Time | Marker |
|----------|------------|-----------|--------|
| P1 | Long-running soak test | 2 days | `@pytest.mark.slow` |
| P2 | Chaos engineering (random failures) | 3 days | `@pytest.mark.chaos` |

---

## Part 4: Success Criteria

### Minimum Viable Coverage

| Category | Metric | Target |
|----------|--------|--------|
| Stream cancellation | Tests pass | 100% |
| Resource leaks | No FD growth > 10 over 50 cycles | Pass |
| Memory stability | No > 1MB growth over 100 cycles | Pass |
| DB pool exhaustion | Graceful queueing, no crash | Pass |
| Signal shutdown | Exit code 0, < 10 seconds | Pass |
| Crash recovery | All RECORDING → ERROR | 100% |

### CI Integration

Add to `.github/workflows/tests.yml`:

```yaml
- name: Run resilience tests
  run: |
    pytest tests/grpc/test_stream_lifecycle.py -v
    pytest tests/stress/test_resource_leaks.py -v --slow
    pytest tests/integration/test_database_resilience.py -v --integration
    pytest tests/integration/test_signal_handling.py -v --integration
    pytest tests/integration/test_crash_scenarios.py -v --integration
```

---

## Part 5: Risk Register

| Risk | Likelihood | Impact | Mitigation |
|------|------------|--------|------------|
| FD leak tests flaky on CI | Medium | Low | Use tolerance threshold |
| Memory tests environment-dependent | High | Medium | Run in Docker with fixed resources |
| Signal tests need root on some CI | Low | Medium | Skip in restricted environments |
| DB pool tests need real Postgres | Medium | Low | Use testcontainers |
| Concurrent recovery races | Low | High | Use database-level locking |

---

## Appendix A: File Locations

| New File | Purpose |
|----------|---------|
| `tests/grpc/test_stream_lifecycle.py` | gRPC stream cancellation tests |
| `tests/stress/test_resource_leaks.py` | FD, memory, coroutine leak tests |
| `tests/integration/test_database_resilience.py` | DB pool, unavailability tests |
| `tests/integration/test_signal_handling.py` | SIGTERM/SIGINT shutdown tests |
| `tests/integration/test_crash_scenarios.py` | Meeting/job recovery tests |

## Appendix B: Related Documentation

- [CLAUDE.md](../../../CLAUDE.md) - Project overview and architecture
- [docs/triage.md](../../../triage.md) - Known issues and technical debt
- [tests/quality/](../../../../tests/quality/) - Test quality checks

---

*Document generated from codebase analysis on 2025-12-29*
