peat_mesh/storage/

sync_channel.rs

//! Persistent sync channels for efficient stream reuse (Issue #438 Phase 2)
//!
//! This module provides persistent bidirectional channels for sync operations,
//! eliminating the need to open a new QUIC stream for each sync batch.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────┐
//! │   SyncChannelManager│
//! │                     │
//! │  ┌───────────────┐  │
//! │  │ channels:     │  │
//! │  │  peer_1 → Ch1 │  │
//! │  │  peer_2 → Ch2 │  │
//! │  │  ...          │  │
//! │  └───────────────┘  │
//! └─────────────────────┘
//!          │
//!          ▼
//! ┌─────────────────────┐
//! │    SyncChannel      │
//! │                     │
//! │  ┌──────────────┐   │
//! │  │ send stream  │   │  ← Persistent, reused for all batches
//! │  ├──────────────┤   │
//! │  │ recv task    │   │  ← Background receiver
//! │  ├──────────────┤   │
//! │  │ state        │   │  ← Connected/Reconnecting/Closed
//! │  └──────────────┘   │
//! └─────────────────────┘
//! ```
//!
//! # Benefits
//!
//! - Eliminates stream-per-batch overhead
//! - Reduces QUIC handshake latency
//! - Enables message multiplexing
//! - Automatic reconnection on failure
//!
//! # Lock ordering
//!
//! ## `SyncChannel`
//!
//! Each channel contains two async `Mutex`es and two sync `RwLock`s:
//!
//! | Lock | Type | Protects |
//! |------|------|----------|
//! | `send` | `tokio::sync::Mutex<Option<SendStream>>` | Outbound QUIC stream |
//! | `recv_task` | `tokio::sync::Mutex<Option<JoinHandle>>` | Background receiver handle |
//! | `state` | `std::sync::RwLock<ChannelState>` | Channel lifecycle state |
//! | `last_send` | `std::sync::RwLock<Instant>` | Timestamp bookkeeping |
//!
//! **Required acquisition order (when multiple locks are needed):**
//!
//! 1. `state` (sync RwLock) -- read or write, then **release** before step 2
//! 2. `send` (async Mutex)
//! 3. `state` / `last_send` (sync RwLock) -- may re-acquire after `send`
//!
//! All current methods (`send`, `reconnect`, `close`) follow this discipline:
//! they snapshot `state` into a local variable, drop the `RwLock` guard, and
//! only then await `send.lock()`. This avoids holding a sync lock across an
//! `.await` point and prevents deadlocks between `state` and `send`.
//!
//! **Invariant:** never hold `state` (or `last_send`) while awaiting `send`
//! or `recv_task`. Sync `RwLock` guards are not `Send` and would block the
//! async runtime if held across an await.
//!
//! ## `SyncChannelManager`
//!
//! | Lock | Type | Protects |
//! |------|------|----------|
//! | `channels` | `std::sync::RwLock<HashMap<EndpointId, Arc<SyncChannel>>>` | Per-peer channel map |
//!
//! `channels` is always acquired briefly (read to look up, write to insert or
//! remove) and released before calling any async method on a `SyncChannel`.
//! This ensures the manager lock is never held while a channel lock is
//! contended.

#[cfg(feature = "automerge-backend")]
use anyhow::{Context, Result};
#[cfg(feature = "automerge-backend")]
use iroh::endpoint::{RecvStream, SendStream};
#[cfg(feature = "automerge-backend")]
use iroh::EndpointId;
#[cfg(feature = "automerge-backend")]
use std::collections::HashMap;
#[cfg(feature = "automerge-backend")]
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
#[cfg(feature = "automerge-backend")]
use std::sync::{Arc, RwLock};
#[cfg(feature = "automerge-backend")]
use std::time::{Duration, Instant};
#[cfg(feature = "automerge-backend")]
use tokio::sync::Mutex;
#[cfg(feature = "automerge-backend")]
use tokio::task::JoinHandle;

#[cfg(feature = "automerge-backend")]
use super::automerge_sync::{AutomergeSyncCoordinator, SyncBatch, SyncMessageType};
#[cfg(feature = "automerge-backend")]
use super::sync_transport::SyncTransport;

/// Channel state
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChannelState {
    /// Channel is connected and ready for use
    Connected,
    /// Channel is attempting to reconnect
    Reconnecting,
    /// Channel is closed and cannot be used
    Closed,
}

/// Persistent sync channel to a peer
///
/// Maintains a single bidirectional QUIC stream for all sync operations
/// with automatic reconnection on failure.
#[cfg(feature = "automerge-backend")]
pub struct SyncChannel {
    /// Peer ID this channel connects to
    peer_id: EndpointId,
    /// Transport reference for reconnection
    transport: Arc<dyn SyncTransport>,
    /// Outbound send stream (protected by mutex for single writer)
    send: Arc<Mutex<Option<SendStream>>>,
    /// Inbound receiver task handle
    recv_task: Arc<Mutex<Option<JoinHandle<()>>>>,
    /// Channel state
    state: Arc<RwLock<ChannelState>>,
    /// Reconnection attempts counter
    reconnect_attempts: AtomicU32,
    /// Last successful send time
    last_send: Arc<RwLock<Instant>>,
    /// Total bytes sent through this channel
    bytes_sent: AtomicU64,
    /// Total batches sent through this channel
    batches_sent: AtomicU64,
}

#[cfg(feature = "automerge-backend")]
impl SyncChannel {
    /// Maximum reconnection attempts before giving up
    const MAX_RECONNECT_ATTEMPTS: u32 = 3;
    /// Delay between reconnection attempts
    const RECONNECT_DELAY: Duration = Duration::from_millis(500);
    /// Timeout for individual read operations in the receive loop.
    /// If a peer stops sending mid-message, the loop breaks after this duration.
    const RECV_TIMEOUT: Duration = Duration::from_secs(30);

    /// Create a new sync channel to a peer
    ///
    /// Opens a bidirectional stream and spawns a receiver task.
    /// An optional [`CancellationToken`](tokio_util::sync::CancellationToken)
    /// can be provided; when cancelled, the receive loop exits promptly.
    pub async fn connect(
        transport: Arc<dyn SyncTransport>,
        peer_id: EndpointId,
        coordinator: Arc<AutomergeSyncCoordinator>,
    ) -> Result<Self> {
        Self::connect_with_token(transport, peer_id, coordinator, None).await
    }

    /// Create a new sync channel to a peer with an explicit cancellation token.
    ///
    /// See [`connect`](Self::connect) for details.
    pub async fn connect_with_token(
        transport: Arc<dyn SyncTransport>,
        peer_id: EndpointId,
        coordinator: Arc<AutomergeSyncCoordinator>,
        cancel: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Self> {
        // Get connection to peer (or establish authenticated one)
        let conn = transport.get_or_connect(&peer_id).await?;

        // Open bidirectional stream
        let (send, recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream")?;

        let channel = Self {
            peer_id,
            transport,
            send: Arc::new(Mutex::new(Some(send))),
            recv_task: Arc::new(Mutex::new(None)),
            state: Arc::new(RwLock::new(ChannelState::Connected)),
            reconnect_attempts: AtomicU32::new(0),
            last_send: Arc::new(RwLock::new(Instant::now())),
            bytes_sent: AtomicU64::new(0),
            batches_sent: AtomicU64::new(0),
        };

        // Spawn receiver task
        channel.spawn_receiver(recv, coordinator, cancel);

        tracing::debug!("Sync channel connected to peer {:?}", peer_id);
        Ok(channel)
    }

    /// Spawn the inbound receiver task
    fn spawn_receiver(
        &self,
        recv: RecvStream,
        coordinator: Arc<AutomergeSyncCoordinator>,
        cancel: Option<tokio_util::sync::CancellationToken>,
    ) {
        let peer_id = self.peer_id;
        let state = Arc::clone(&self.state);
        let recv_task = Arc::clone(&self.recv_task);

        let task = tokio::spawn(async move {
            tracing::debug!("Sync channel receiver started for peer {:?}", peer_id);

            if let Err(e) = Self::receive_loop(recv, peer_id, coordinator, cancel).await {
                tracing::warn!(
                    "Sync channel receiver for peer {:?} ended with error: {}",
                    peer_id,
                    e
                );
            }

            // Mark channel as needing reconnection
            *state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Reconnecting;
            tracing::debug!("Sync channel receiver ended for peer {:?}", peer_id);
        });

        // Store task handle (spawn a task to avoid blocking)
        tokio::spawn(async move {
            *recv_task.lock().await = Some(task);
        });
    }

    /// Main receive loop - processes incoming batches
    ///
    /// When a cancellation token is provided, the loop checks it before each
    /// read and races the token against the receive timeout so that shutdown
    /// is observed promptly.
    async fn receive_loop(
        mut recv: RecvStream,
        peer_id: EndpointId,
        coordinator: Arc<AutomergeSyncCoordinator>,
        cancel: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<()> {
        loop {
            // Exit immediately if cancellation has been requested.
            if let Some(ref token) = cancel {
                if token.is_cancelled() {
                    tracing::debug!("Sync channel receive loop for peer {:?} cancelled", peer_id);
                    return Ok(());
                }
            }

            // Read message type marker (race with cancellation token if present)
            let mut marker = [0u8; 1];
            let read_result = if let Some(ref token) = cancel {
                tokio::select! {
                    res = tokio::time::timeout(Self::RECV_TIMEOUT, recv.read_exact(&mut marker)) => res,
                    () = token.cancelled() => {
                        tracing::debug!(
                            "Sync channel receive loop for peer {:?} cancelled during read",
                            peer_id
                        );
                        return Ok(());
                    }
                }
            } else {
                tokio::time::timeout(Self::RECV_TIMEOUT, recv.read_exact(&mut marker)).await
            };

            match read_result {
                Ok(Ok(_)) => {}
                Ok(Err(e)) => {
                    // Stream closed or error
                    return Err(anyhow::anyhow!("Stream read error: {}", e));
                }
                Err(_) => {
                    tracing::warn!(
                        "Sync channel receive timeout for peer {:?} (no data for {:?})",
                        peer_id,
                        Self::RECV_TIMEOUT,
                    );
                    return Err(anyhow::anyhow!(
                        "Receive timeout waiting for message marker"
                    ));
                }
            }

            // Check if it's a batch message (0x07)
            if marker[0] != SyncMessageType::SyncBatch as u8 {
                tracing::warn!(
                    "Unexpected message type on sync channel: 0x{:02x}",
                    marker[0]
                );
                continue;
            }

            // Read batch length (4 bytes, big-endian)
            let mut len_bytes = [0u8; 4];
            tokio::time::timeout(Self::RECV_TIMEOUT, recv.read_exact(&mut len_bytes))
                .await
                .map_err(|_| {
                    tracing::warn!(
                        "Sync channel receive timeout for peer {:?} reading batch length",
                        peer_id,
                    );
                    anyhow::anyhow!("Receive timeout reading batch length")
                })?
                .context("Failed to read batch length")?;
            let batch_len = u32::from_be_bytes(len_bytes) as usize;

            // Read batch data
            let mut batch_data = vec![0u8; batch_len];
            tokio::time::timeout(Self::RECV_TIMEOUT, recv.read_exact(&mut batch_data))
                .await
                .map_err(|_| {
                    tracing::warn!(
                        "Sync channel receive timeout for peer {:?} reading batch data ({} bytes)",
                        peer_id,
                        batch_len,
                    );
                    anyhow::anyhow!("Receive timeout reading batch data")
                })?
                .context("Failed to read batch data")?;

            // Decode and process batch
            match SyncBatch::decode(&batch_data) {
                Ok(batch) => {
                    let total_bytes = 1 + 4 + batch_len; // marker + len + data
                    if let Err(e) = coordinator
                        .receive_batch_message(peer_id, batch, total_bytes)
                        .await
                    {
                        tracing::warn!("Failed to process batch from peer {:?}: {}", peer_id, e);
                    }
                }
                Err(e) => {
                    tracing::warn!("Failed to decode batch from peer {:?}: {}", peer_id, e);
                }
            }
        }
    }

    /// Send a batch through this channel
    ///
    /// Wire format (compatible with receive_sync_payload_from_stream):
    /// ```text
    /// [2 bytes: doc_key_len][N bytes: "batch"][1 byte: 0x07][4 bytes: batch_len][batch_bytes...]
    /// ```
    pub async fn send(&self, batch: &SyncBatch) -> Result<()> {
        // Check channel state (read and release lock before any await)
        let needs_reconnect = {
            let state = *self.state.read().unwrap_or_else(|e| e.into_inner());
            match state {
                ChannelState::Closed => return Err(anyhow::anyhow!("Channel is closed")),
                ChannelState::Reconnecting => true,
                ChannelState::Connected => false,
            }
        };

        // Reconnect if needed (outside the lock)
        if needs_reconnect {
            self.reconnect().await?;
        }

        let batch_bytes = batch.encode();
        let mut send_guard = self.send.lock().await;

        let send = send_guard
            .as_mut()
            .ok_or_else(|| anyhow::anyhow!("No send stream available"))?;

        // Write doc_key length prefix (2 bytes) - "batch" = 5 chars
        let doc_key = b"batch";
        send.write_all(&(doc_key.len() as u16).to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;

        // Write doc_key
        send.write_all(doc_key)
            .await
            .context("Failed to write doc_key")?;

        // Write message type marker
        send.write_all(&[SyncMessageType::SyncBatch as u8])
            .await
            .context("Failed to write batch marker")?;

        // Write batch length
        let batch_len = batch_bytes.len() as u32;
        send.write_all(&batch_len.to_be_bytes())
            .await
            .context("Failed to write batch length")?;

        // Write batch data
        send.write_all(&batch_bytes)
            .await
            .context("Failed to write batch data")?;

        // Update statistics (2 + doc_key_len + 1 + 4 + batch_bytes)
        let total_bytes = 2 + doc_key.len() + 1 + 4 + batch_bytes.len();
        self.bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);
        self.batches_sent.fetch_add(1, Ordering::Relaxed);
        *self.last_send.write().unwrap_or_else(|e| e.into_inner()) = Instant::now();

        tracing::trace!(
            "Sent batch {} ({} entries, {} bytes) to peer {:?}",
            batch.batch_id,
            batch.len(),
            total_bytes,
            self.peer_id
        );

        Ok(())
    }

    /// Attempt to reconnect the channel
    pub async fn reconnect(&self) -> Result<()> {
        // Update state
        *self.state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Reconnecting;

        let attempts = self.reconnect_attempts.fetch_add(1, Ordering::Relaxed);
        if attempts >= Self::MAX_RECONNECT_ATTEMPTS {
            *self.state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Closed;
            return Err(anyhow::anyhow!(
                "Max reconnection attempts ({}) exceeded",
                Self::MAX_RECONNECT_ATTEMPTS
            ));
        }

        tracing::info!(
            "Attempting reconnection to peer {:?} (attempt {})",
            self.peer_id,
            attempts + 1
        );

        // Wait before reconnecting
        tokio::time::sleep(Self::RECONNECT_DELAY).await;

        // Get connection (or establish authenticated one)
        let conn = self.transport.get_or_connect(&self.peer_id).await?;

        // Open new stream
        let (send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream for reconnection")?;

        // Issue #435: Explicitly stop recv stream to prevent resource accumulation
        // (reconnect only uses send half, recv is not needed)
        let _ = recv.stop(0u32.into());

        // Update send stream
        *self.send.lock().await = Some(send);
        *self.state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Connected;
        self.reconnect_attempts.store(0, Ordering::Relaxed);

        tracing::info!("Reconnected sync channel to peer {:?}", self.peer_id);
        Ok(())
    }

    /// Check if channel is connected
    pub fn is_connected(&self) -> bool {
        *self.state.read().unwrap_or_else(|e| e.into_inner()) == ChannelState::Connected
    }

    /// Get channel state
    pub fn state(&self) -> ChannelState {
        *self.state.read().unwrap_or_else(|e| e.into_inner())
    }

    /// Get peer ID
    pub fn peer_id(&self) -> EndpointId {
        self.peer_id
    }

    /// Get total bytes sent
    pub fn bytes_sent(&self) -> u64 {
        self.bytes_sent.load(Ordering::Relaxed)
    }

    /// Get total batches sent
    pub fn batches_sent(&self) -> u64 {
        self.batches_sent.load(Ordering::Relaxed)
    }

    /// Close the channel
    pub async fn close(&self) {
        *self.state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Closed;

        // Cancel receiver task
        if let Some(task) = self.recv_task.lock().await.take() {
            task.abort();
        }

        // Close send stream
        if let Some(mut send) = self.send.lock().await.take() {
            let _ = send.finish();
        }

        tracing::debug!("Sync channel to peer {:?} closed", self.peer_id);
    }
}

/// Manager for sync channels to all peers
///
/// Automatically creates and manages persistent channels for each connected peer.
#[cfg(feature = "automerge-backend")]
pub struct SyncChannelManager {
    /// Active channels per peer
    channels: Arc<RwLock<HashMap<EndpointId, Arc<SyncChannel>>>>,
    /// Transport reference
    transport: Arc<dyn SyncTransport>,
    /// Coordinator reference for processing received batches
    coordinator: Arc<AutomergeSyncCoordinator>,
    /// Whether the manager is active
    active: Arc<std::sync::atomic::AtomicBool>,
}

#[cfg(feature = "automerge-backend")]
impl SyncChannelManager {
    /// Create a new channel manager
    pub fn new(
        transport: Arc<dyn SyncTransport>,
        coordinator: Arc<AutomergeSyncCoordinator>,
    ) -> Self {
        Self {
            channels: Arc::new(RwLock::new(HashMap::new())),
            transport,
            coordinator,
            active: Arc::new(std::sync::atomic::AtomicBool::new(true)),
        }
    }

    /// Get or create a channel to a peer
    pub async fn get_channel(&self, peer_id: EndpointId) -> Result<Arc<SyncChannel>> {
        // Check if we have an existing connected channel
        {
            let channels = self.channels.read().unwrap_or_else(|e| e.into_inner());
            if let Some(channel) = channels.get(&peer_id) {
                if channel.is_connected() {
                    return Ok(Arc::clone(channel));
                }
            }
        }

        // Need to create or reconnect
        let channel = SyncChannel::connect(
            Arc::clone(&self.transport),
            peer_id,
            Arc::clone(&self.coordinator),
        )
        .await?;

        let channel = Arc::new(channel);
        self.channels
            .write()
            .unwrap()
            .insert(peer_id, Arc::clone(&channel));

        Ok(channel)
    }

    /// Send batch to a peer through persistent channel
    pub async fn send_to_peer(&self, peer_id: EndpointId, batch: &SyncBatch) -> Result<()> {
        let channel = self.get_channel(peer_id).await?;
        channel.send(batch).await
    }

    /// Broadcast batch to all connected peers
    pub async fn broadcast(&self, batch: &SyncBatch) -> Result<()> {
        let peer_ids = self.transport.connected_peers();

        for peer_id in peer_ids {
            if let Err(e) = self.send_to_peer(peer_id, batch).await {
                tracing::warn!("Failed to send batch to peer {:?}: {}", peer_id, e);
            }
        }

        Ok(())
    }

    /// Send a delta sync message to a specific peer through persistent channel
    ///
    /// Packages the message into a SyncBatch and sends it.
    pub async fn send_delta_sync(
        &self,
        peer_id: EndpointId,
        doc_key: &str,
        message: &automerge::sync::Message,
    ) -> Result<usize> {
        let encoded = message.clone().encode();
        let payload_len = encoded.len();

        let mut batch = SyncBatch::new();
        batch.entries.push(super::automerge_sync::SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::DeltaSync,
            encoded,
        ));

        self.send_to_peer(peer_id, &batch).await?;

        // Return bytes sent: marker + len + batch_bytes
        Ok(1 + 4 + batch.encode().len() + payload_len)
    }

    /// Send a state snapshot to a specific peer through persistent channel
    pub async fn send_state_snapshot(
        &self,
        peer_id: EndpointId,
        doc_key: &str,
        state_bytes: Vec<u8>,
    ) -> Result<usize> {
        let payload_len = state_bytes.len();

        let mut batch = SyncBatch::new();
        batch.entries.push(super::automerge_sync::SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::StateSnapshot,
            state_bytes,
        ));

        self.send_to_peer(peer_id, &batch).await?;

        Ok(1 + 4 + batch.encode().len() + payload_len)
    }

    /// Send a tombstone to a specific peer through persistent channel
    pub async fn send_tombstone(
        &self,
        peer_id: EndpointId,
        tombstone_msg: &crate::qos::TombstoneSyncMessage,
    ) -> Result<usize> {
        let mut batch = SyncBatch::new();
        batch.add_tombstone(tombstone_msg);

        let batch_bytes = batch.encode();
        self.send_to_peer(peer_id, &batch).await?;

        Ok(1 + 4 + batch_bytes.len())
    }

    /// Send multiple tombstones to a specific peer through persistent channel
    pub async fn send_tombstone_batch(
        &self,
        peer_id: EndpointId,
        tombstones: &[crate::qos::TombstoneSyncMessage],
    ) -> Result<usize> {
        let mut batch = SyncBatch::new();
        for tombstone in tombstones {
            batch.add_tombstone(tombstone);
        }

        let batch_bytes = batch.encode();
        self.send_to_peer(peer_id, &batch).await?;

        Ok(1 + 4 + batch_bytes.len())
    }

    /// Broadcast a delta sync message to all connected peers
    pub async fn broadcast_delta_sync(
        &self,
        doc_key: &str,
        message: &automerge::sync::Message,
    ) -> Result<()> {
        let encoded = message.clone().encode();

        let mut batch = SyncBatch::new();
        batch.entries.push(super::automerge_sync::SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::DeltaSync,
            encoded,
        ));

        self.broadcast(&batch).await
    }

    /// Broadcast a state snapshot to all connected peers
    pub async fn broadcast_state_snapshot(
        &self,
        doc_key: &str,
        state_bytes: Vec<u8>,
    ) -> Result<()> {
        let mut batch = SyncBatch::new();
        batch.entries.push(super::automerge_sync::SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::StateSnapshot,
            state_bytes,
        ));

        self.broadcast(&batch).await
    }

    /// Broadcast a tombstone to all connected peers
    pub async fn broadcast_tombstone(
        &self,
        tombstone_msg: &crate::qos::TombstoneSyncMessage,
    ) -> Result<()> {
        let mut batch = SyncBatch::new();
        batch.add_tombstone(tombstone_msg);

        self.broadcast(&batch).await
    }

    /// Remove channel for a peer (e.g., on disconnect)
    pub async fn remove_channel(&self, peer_id: &EndpointId) {
        // Extract channel from map first, then close outside the lock
        let channel = self
            .channels
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .remove(peer_id);
        if let Some(channel) = channel {
            channel.close().await;
        }
    }

    /// Get number of active channels
    pub fn channel_count(&self) -> usize {
        self.channels
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .len()
    }

    /// Get statistics for all channels
    pub fn stats(&self) -> ChannelManagerStats {
        let channels = self.channels.read().unwrap_or_else(|e| e.into_inner());
        let mut total_bytes = 0u64;
        let mut total_batches = 0u64;
        let mut connected = 0usize;

        for channel in channels.values() {
            total_bytes += channel.bytes_sent();
            total_batches += channel.batches_sent();
            if channel.is_connected() {
                connected += 1;
            }
        }

        ChannelManagerStats {
            total_channels: channels.len(),
            connected_channels: connected,
            total_bytes_sent: total_bytes,
            total_batches_sent: total_batches,
        }
    }

    /// Shutdown the manager and close all channels
    pub async fn shutdown(&self) {
        self.active.store(false, Ordering::Relaxed);

        let channels: Vec<Arc<SyncChannel>> = {
            let mut channels = self.channels.write().unwrap_or_else(|e| e.into_inner());
            channels.drain().map(|(_, c)| c).collect()
        };

        for channel in channels {
            channel.close().await;
        }

        tracing::debug!("SyncChannelManager shutdown complete");
    }
}

/// Statistics for channel manager
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone, Default)]
pub struct ChannelManagerStats {
    /// Total number of channels (active + reconnecting)
    pub total_channels: usize,
    /// Number of connected channels
    pub connected_channels: usize,
    /// Total bytes sent across all channels
    pub total_bytes_sent: u64,
    /// Total batches sent across all channels
    pub total_batches_sent: u64,
}

#[cfg(all(test, feature = "automerge-backend"))]
mod tests {
    use super::*;

    #[test]
    fn test_channel_state_enum() {
        assert_ne!(ChannelState::Connected, ChannelState::Reconnecting);
        assert_ne!(ChannelState::Reconnecting, ChannelState::Closed);
        assert_eq!(ChannelState::Connected, ChannelState::Connected);
    }

    #[test]
    fn test_channel_manager_stats_default() {
        let stats = ChannelManagerStats::default();
        assert_eq!(stats.total_channels, 0);
        assert_eq!(stats.connected_channels, 0);
        assert_eq!(stats.total_bytes_sent, 0);
        assert_eq!(stats.total_batches_sent, 0);
    }
}