Reliability¶

async def enqueue(self, event: UDMEvent) -> bool:
    """
    Add an event to the buffer.

    Args:
        event: The UDM event to buffer

    Returns:
        True if successfully buffered, False if buffer is full
    """
    async with self._lock:
        # Check size limit
        stats = self._get_stats_sync()
        if stats.size_bytes >= self.max_size_bytes:
            # Try to prune old events first
            self._prune_old_events()
            stats = self._get_stats_sync()
            if stats.size_bytes >= self.max_size_bytes:
                return False

        # Generate unique filename with timestamp for ordering
        self._counter += 1
        timestamp = int(time.time() * 1000000)
        filename = f"{timestamp}_{self._counter}_{event.event_id}.json"
        filepath = self.storage_path / filename

        # Serialize and write atomically
        event_data = {
            "buffered_at": datetime.utcnow().isoformat(),
            "event": event.to_dict(),
        }

        # Write to temp file then rename (atomic on POSIX)
        temp_path = filepath.with_suffix(".tmp")
        try:
            with open(temp_path, "w") as f:
                json.dump(event_data, f)
            temp_path.rename(filepath)
            return True
        except OSError:
            if temp_path.exists():
                temp_path.unlink()
            return False

async def dequeue(self, max_events: int = 100) -> list[UDMEvent]:
    """
    Get events from the buffer (oldest first).

    Args:
        max_events: Maximum number of events to retrieve

    Returns:
        List of UDM events (does not remove them from buffer)
    """
    async with self._lock:
        events: list[UDMEvent] = []

        # Get sorted list of buffer files
        files = sorted(self.storage_path.glob("*.json"))

        for filepath in files[:max_events]:
            try:
                with open(filepath, "r") as f:
                    data = json.load(f)
                event = UDMEvent.from_dict(data["event"])
                events.append(event)
            except (json.JSONDecodeError, KeyError, OSError):
                # Skip corrupted files
                continue

        return events

async def remove(self, event_ids: list[str]) -> int:
    """
    Remove specific events from the buffer.

    Args:
        event_ids: List of event IDs to remove

    Returns:
        Number of events removed
    """
    async with self._lock:
        removed = 0
        event_id_set = set(event_ids)

        for filepath in self.storage_path.glob("*.json"):
            # Extract event_id from filename
            parts = filepath.stem.split("_")
            if len(parts) >= 3:
                file_event_id = "_".join(parts[2:])
                if file_event_id in event_id_set:
                    try:
                        filepath.unlink()
                        removed += 1
                    except OSError:
                        pass

        return removed

async def flush(
    self, transport: Transport, batch_size: int = 100
) -> BatchSendResult:
    """
    Send all buffered events when connection is available.

    Args:
        transport: The transport to use for sending
        batch_size: Number of events to send per batch

    Returns:
        Combined BatchSendResult for all events
    """
    total_events = 0
    successful_events = 0
    failed_events = 0
    all_results = []

    while True:
        # Get next batch
        events = await self.dequeue(batch_size)
        if not events:
            break

        # Send batch
        result = await transport.send_batch(events)

        total_events += result.total_events
        successful_events += result.successful_events
        failed_events += result.failed_events
        all_results.extend(result.results)

        # Remove successfully sent events
        successful_ids = [r.event_id for r in result.results if r.success]
        await self.remove(successful_ids)

        # If batch failed completely, stop flushing
        if result.failed_events == result.total_events:
            break

    return BatchSendResult(
        success=failed_events == 0,
        total_events=total_events,
        successful_events=successful_events,
        failed_events=failed_events,
        results=all_results,
    )

async def get_stats(self) -> BufferStats:
    """Get buffer statistics."""
    async with self._lock:
        return self._get_stats_sync()

async def clear(self) -> int:
    """Clear all buffered events."""
    async with self._lock:
        removed = 0
        for filepath in self.storage_path.glob("*.json"):
            try:
                filepath.unlink()
                removed += 1
            except OSError:
                continue
        return removed

async def __aiter__(self) -> AsyncIterator[UDMEvent]:
    """Iterate over buffered events."""
    events = await self.dequeue(max_events=10000)
    for event in events:
        yield event

def get_delay(self, attempt: int) -> float:
    """
    Calculate delay for given attempt number.

    Args:
        attempt: Attempt number (0-indexed)

    Returns:
        Delay in seconds
    """
    # Calculate exponential delay
    delay_ms = self.initial_delay_ms * (self.exponential_base**attempt)

    # Cap at max delay
    delay_ms = min(delay_ms, self.max_delay_ms)

    # Add jitter (±25%)
    if self.jitter:
        jitter_factor = 1.0 + (random.random() - 0.5) * 0.5
        delay_ms *= jitter_factor

    return delay_ms / 1000.0  # Convert to seconds

def should_retry(self, error_type: ErrorType, attempt: int) -> bool:
    """
    Determine if a retry should be attempted.

    Args:
        error_type: Type of error that occurred
        attempt: Current attempt number (0-indexed)

    Returns:
        True if should retry, False otherwise
    """
    if attempt >= self.max_retries:
        return False

    return error_type in self.retryable_errors

async def execute_with_retry(
    self,
    func: Callable[[], T],
    on_retry: Callable[[int, Exception], None] | None = None,
) -> T:
    """
    Execute a function with retry logic.

    Args:
        func: Async function to execute
        on_retry: Optional callback on each retry

    Returns:
        Result of the function

    Raises:
        Exception: If all retries exhausted
    """
    last_exception: Exception | None = None

    for attempt in range(self.max_retries + 1):
        try:
            return await func()
        except Exception as e:
            last_exception = e

            # Check if error type is retryable
            error_type = getattr(e, "error_type", ErrorType.UNKNOWN)
            if not self.should_retry(error_type, attempt):
                raise

            # Call retry callback
            if on_retry:
                on_retry(attempt, e)

            # Wait before retrying
            delay = self.get_delay(attempt)
            await asyncio.sleep(delay)

    # All retries exhausted
    if last_exception:
        raise last_exception
    raise RuntimeError("Retry logic error: no exception captured")

async def allow_request(self) -> bool:
    """
    Check if a request should be allowed.

    Returns:
        True if request should proceed, False if rejected
    """
    async with self._lock:
        if self._state == CircuitState.CLOSED:
            return True

        if self._state == CircuitState.OPEN:
            # Check if recovery timeout has passed
            if self._last_failure_time:
                elapsed = time.time() - self._last_failure_time
                if elapsed >= self.recovery_timeout_sec:
                    # Transition to half-open
                    self._state = CircuitState.HALF_OPEN
                    self._half_open_successes = 0
                    self._half_open_failures = 0
                    return True
            return False

        if self._state == CircuitState.HALF_OPEN:
            # Allow limited requests
            total_half_open = self._half_open_successes + self._half_open_failures
            return total_half_open < self.half_open_requests

        return False

async def record_success(self) -> None:
    """Record a successful request."""
    async with self._lock:
        if self._state == CircuitState.HALF_OPEN:
            self._half_open_successes += 1

            # If enough successes, close the circuit
            if self._half_open_successes >= self.half_open_requests:
                self._state = CircuitState.CLOSED
                self._failure_count = 0
                self._half_open_successes = 0
                self._half_open_failures = 0

        elif self._state == CircuitState.CLOSED:
            # Reset failure count on success
            self._failure_count = 0

async def record_failure(self) -> None:
    """Record a failed request."""
    async with self._lock:
        self._last_failure_time = time.time()

        if self._state == CircuitState.HALF_OPEN:
            self._half_open_failures += 1

            # If any failure in half-open, open the circuit
            self._state = CircuitState.OPEN

        elif self._state == CircuitState.CLOSED:
            self._failure_count += 1

            # Check if threshold reached
            if self._failure_count >= self.failure_threshold:
                self._state = CircuitState.OPEN

async def reset(self) -> None:
    """Reset the circuit breaker to closed state."""
    async with self._lock:
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._last_failure_time = None
        self._half_open_successes = 0
        self._half_open_failures = 0

async def execute(
    self,
    func: Callable[[], T],
) -> T:
    """
    Execute a function with circuit breaker protection.

    Args:
        func: Async function to execute

    Returns:
        Result of the function

    Raises:
        CircuitOpenError: If circuit is open
        Exception: If function raises an error
    """
    if not await self.allow_request():
        raise CircuitOpenError("Circuit breaker is open")

    try:
        result = await func()
        await self.record_success()
        return result
    except Exception as e:
        await self.record_failure()
        raise

async def start(self) -> None:
    """Start the batch processor."""
    self._running = True
    self._flush_task = asyncio.create_task(self._flush_loop())

async def stop(self) -> None:
    """Stop the batch processor and flush remaining events."""
    self._running = False
    if self._flush_task:
        self._flush_task.cancel()
        try:
            await self._flush_task
        except asyncio.CancelledError:
            pass

    # Flush any remaining events
    await self.flush()

async def add(self, event: UDMEvent) -> Batch | None:
    """
    Add an event to the current batch.

    Args:
        event: The event to add

    Returns:
        A Batch if one was flushed, None otherwise
    """
    async with self._lock:
        # Initialize batch timestamp if needed
        if self._batch_created_at is None:
            self._batch_created_at = time.time()

        self._current_batch.append(event)

        # Check if batch is full
        if len(self._current_batch) >= self.config.max_batch_size:
            return await self._flush_batch()

        return None

async def flush(self) -> Batch | None:
    """
    Flush the current batch.

    Returns:
        The flushed batch, or None if empty
    """
    async with self._lock:
        return await self._flush_batch()