# SPRINT-GAP-001: Streaming Race Conditions

| Attribute | Value |
|-----------|-------|
| **Sprint** | GAP-001 |
| **Size** | L (Large) |
| **Owner** | TBD |
| **Phase** | Hardening |
| **Prerequisites** | None |

## Open Issues

- [ ] Define backpressure strategy (drop frames vs block vs buffer)
- [ ] Determine maximum acceptable latency for backpressure signals
- [x] Decide on timeout values for stream initialization lock (5 seconds)

## Validation Status

| Component | Exists | Needs Work |
|-----------|--------|------------|
| `_stream_init_lock` | Yes | ~~Needs timeout~~ ✅ Done (5s timeout) |
| `_active_streams` set | Yes | ~~Needs cleanup guarantees~~ ✅ Done |
| Client error propagation | ~~No~~ Yes | ~~Required~~ ✅ Done (onError callback) |
| Audio chunk acknowledgment | ~~No~~ Yes | ~~Required~~ ✅ Done (Phase 2) |
| Backpressure signaling | No | Required (Phase 3) |

## Objective

Eliminate race conditions in the bidirectional audio streaming pipeline that can cause data loss, resource leaks, or undefined behavior during concurrent operations.

## Key Decisions

| Decision | Choice | Rationale |
|----------|--------|-----------|
| Chunk send strategy | Fire-and-forget → Acknowledged | Current `.catch(() => {})` silently drops failures |
| Backpressure mechanism | gRPC flow control | Native support in HTTP/2 |
| Lock timeout | 5 seconds | Prevent deadlock on init failures |
| Stream cleanup | Guaranteed via finally | Current cleanup can be bypassed |

## What Already Exists

### Backend (`src/noteflow/grpc/_mixins/streaming/`)
- `_stream_init_lock` (asyncio.Lock) protects concurrent initialization
- `_active_streams` (set) tracks active meeting streams
- `_stop_requested` (set) for graceful shutdown signaling
- `cleanup_stream_resources()` for resource cleanup

### Client (`client/src/api/tauri-adapter.ts`)
- `TauriTranscriptionStream` class manages stream lifecycle
- `send()` method dispatches audio chunks
- `close()` method terminates stream

## Identified Issues

### 1. Fire-and-Forget Audio Chunk Dispatch (Critical)

**Location**: `client/src/api/tauri-adapter.ts:107`

```typescript
send(chunk: AudioChunk): void {
  // ...
  this.invoke(TauriCommands.SEND_AUDIO_CHUNK, args).catch((_err) => {});
}
```

**Problem**: Audio chunks are sent without acknowledgment. If the gRPC connection fails mid-stream:
- Client continues sending chunks that are silently dropped
- No error propagation to UI
- Audio data is lost permanently

**Impact**: Data loss during network instability or server restart.

### 2. Stop Recording Also Fire-and-Forget (High)

**Location**: `client/src/api/tauri-adapter.ts:124`

```typescript
close(): void {
  if (this.unlistenFn) { ... }
  this.invoke(TauriCommands.STOP_RECORDING, { meeting_id: this.meetingId }).catch((_err) => {});
}
```

**Problem**: Stream close doesn't await confirmation. If stop fails:
- Meeting may remain in RECORDING state
- Resources may not be cleaned up on server
- Client shows "stopped" but server is still streaming

### 3. Stream Initialization Lock Without Timeout (Medium)

**Location**: `src/noteflow/grpc/_mixins/streaming/_session.py:47-52`

```python
async with host._stream_init_lock:
    if meeting_id in host._active_streams:
        await abort_failed_precondition(...)
    host._active_streams.add(meeting_id)
```

**Problem**: Lock acquisition has no timeout. If a previous initialization hangs:
- All subsequent stream requests block indefinitely
- No way to detect or recover from deadlock

### 4. Cleanup on Exception Path (Medium)

**Location**: `src/noteflow/grpc/_mixins/streaming/_mixin.py:93-95`

```python
finally:
    if current_meeting_id:
        cleanup_stream_resources(self, current_meeting_id)
```

**Problem**: If `current_meeting_id` is never set (early exception), cleanup doesn't run. Also, if exception occurs before `_active_streams.add()` completes but after some resources are allocated.

### 5. No Backpressure Mechanism (Medium)

**Problem**: Client sends audio at capture rate regardless of server processing capacity:
- Server may queue unbounded audio data
- Memory pressure during high load
- Latency accumulation in transcription

## Scope

### Task Breakdown

| Task | Effort | Description |
|------|--------|-------------|
| Add chunk acknowledgment protocol | M | Return ack with sequence number from server |
| Implement client-side retry logic | M | Retry failed chunks with exponential backoff |
| Add lock timeout to stream init | S | Use `asyncio.timeout()` wrapper |
| Fix cleanup guarantees | S | Track allocated resources separately from meeting_id |
| Implement backpressure signaling | L | Use gRPC flow control or custom protocol |
| Add stream health monitoring | M | Periodic heartbeat with timeout detection |
| Surface errors to UI | S | Emit error events from TauriTranscriptionStream |

### Files to Modify

**Backend:**
- `src/noteflow/grpc/_mixins/streaming/_mixin.py`
- `src/noteflow/grpc/_mixins/streaming/_session.py`
- `src/noteflow/grpc/_mixins/streaming/_cleanup.py`
- `src/noteflow/grpc/proto/noteflow.proto` (if adding ack messages)

**Client:**
- `client/src-tauri/src/commands/recording/capture.rs`
- `client/src/api/tauri-adapter.ts`
- `client/src/api/transcription-stream.ts`
- `client/src/hooks/use-recording.ts` (error handling)

## API Schema Changes

Consider adding acknowledgment to `TranscriptUpdate`:

```protobuf
message TranscriptUpdate {
  // Existing fields...

  // New: Acknowledge received chunks up to this sequence number
  optional int64 ack_sequence = 10;
}
```

Or a separate acknowledgment message:

```protobuf
message AudioChunk {
  // Existing fields...
  int64 sequence_number = 6;  // New: For tracking
}

message ChunkAcknowledgment {
  int64 last_ack_sequence = 1;
  bool backpressure = 2;  // Signal client to slow down
}
```

## Migration Strategy

### Phase 1: Add Observability (Low Risk)
- Add metrics for chunk send failures
- Log errors instead of silencing them
- Add health check endpoint for stream status

### Phase 2: Implement Acknowledgment (Medium Risk)
- Add sequence numbers to chunks (backward compatible)
- Server sends periodic acks
- Client tracks unacknowledged chunks

### Phase 3: Add Backpressure (Higher Risk)
- Implement flow control signals
- Client buffers or drops frames when signaled
- Requires UI feedback for "buffering" state

## Deliverables

### Backend
- [x] Stream initialization with timeout (5s default)
- [x] Improved cleanup that handles partial initialization
- [x] Chunk acknowledgment emission (ack every 5 chunks)
- [x] Sequence tracking with gap detection logging
- [ ] Backpressure signal emission
- [ ] Health check RPC for stream status

### Client
- [x] Error propagation from chunk send (via `onError` callback)
- [x] Consecutive failure tracking with threshold-based error emission
- [x] Sequence numbering for outgoing chunks (Rust)
- [x] Acknowledgment tracking (Rust + TypeScript)
- [ ] Retry logic for failed chunks (requires Rust-level buffering)
- [ ] Backpressure response (buffer/drop)
- [ ] UI feedback for stream health

### Tests
- [ ] Integration test: network interruption mid-stream
- [ ] Integration test: server restart during stream
- [x] Unit test: lock timeout behavior
- [x] Unit test: cleanup on partial initialization
- [x] Unit test: chunk sequence tracking and ack emission (12 tests)
- [ ] Load test: backpressure under high throughput

## Test Strategy

### Fixtures
- Mock gRPC server that can simulate failures
- Network partition simulation
- Slow server simulation for backpressure

### Test Cases

| Case | Input | Expected |
|------|-------|----------|
| Network drop mid-stream | 100 chunks, drop at 50 | Error emitted, retry attempted, data preserved |
| Server restart | Active stream, server restart | Reconnection, stream resume or error |
| Lock timeout | Hung initialization | Timeout error, no deadlock |
| Cleanup on early error | Exception before meeting_id set | Resources cleaned up |
| Backpressure | 10x normal audio rate | Client receives backpressure signal, adapts |

## Quality Gates

- [x] Zero `.catch(() => {})` patterns in streaming code
- [x] All stream resources have cleanup guarantees
- [x] Lock operations have timeouts
- [x] Error events emitted for all failure modes (via `onError` callback)
- [x] Chunk sequences assigned (monotonically increasing from 1)
- [x] Server emits acks every 5 chunks
- [x] Gap detection logs warnings for non-contiguous sequences
- [x] Ack sequence flows through to TypeScript client
- [ ] Integration tests pass with simulated failures
- [ ] No regression in normal streaming latency

## Phase 2 Implementation Notes

**Completed 2026-01-02**

### Proto Changes
- `AudioChunk.chunk_sequence` (int64) - sequence number per stream
- `TranscriptUpdate.ack_sequence` (optional int64) - highest contiguous received

### Backend (`src/noteflow/grpc/_mixins/streaming/_processing.py`)
- `_track_chunk_sequence()` tracks highest sequence and emits ack every 5 chunks
- `create_ack_update()` converter in `_mixins/converters/_domain.py`
- Cleanup added for `_chunk_sequences` and `_chunk_counts` dicts

### Rust Client (`client/src-tauri/src/grpc/streaming.rs`)
- `StreamState::Active` includes `sequence_counter` and `last_acked_sequence` atomics
- Outbound stream assigns sequences starting at 1
- Inbound task tracks acks from server
- Pure ack updates (no transcript content) filtered from frontend emission

### TypeScript Client (`client/src/api/`)
- `TranscriptUpdate.ack_sequence` added to types
- `TauriTranscriptionStream.lastAckedSequence` tracking
- `getLastAckedSequence()` method for monitoring

### Backwards Compatibility
- Zero sequence treated as legacy (not tracked)
- Optional ack_sequence ignored by old clients
