//! Persistent sync channels for efficient stream reuse (Issue #438 Phase 2 /
//! peat-mesh#175 / ADR-063).
//!
//! 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()  ──┐                                        │
//! │   send()  ──┤                                        │
//! │   send()  ──┼─► mpsc::Sender<Vec<u8>> ─► writer ─► QUIC SendStream
//! │   send()  ──┤                            task         (persistent)
//! │   send()  ──┘                                        │
//! │                                                      │
//! │   ┌──────────────┐                                   │
//! │   │ recv task    │  ◄─ QUIC RecvStream (persistent)  │
//! │   └──────────────┘                                   │
//! │                                                      │
//! │   ┌──────────────┐                                   │
//! │   │ state        │  Connected / Reconnecting / Closed│
//! │   └──────────────┘                                   │
//! └──────────────────────────────────────────────────────┘
//! ```
//!
//! # Send-path design (peat-mesh#175 UAT-driven refactor)
//!
//! An earlier shape of this module held a `tokio::sync::Mutex<SendStream>`
//! per channel; every caller of `SyncChannel::send` had to lock the mutex
//! across the four `write_all` calls that make up one v2 frame. Under
//! fan-out workloads — `sync_document_with_all_peers` issues per-peer
//! sync concurrently for the same write event — that mutex serialised
//! the per-peer send path. Lab measurement (peat-mesh#175 UAT,
//! single-platform-emitter sweep) showed the persistent-stream design
//! was ~2pp _worse_ than the legacy per-message-stream path it replaces,
//! because the legacy path lets the per-message bi-streams pipeline
//! through QUIC's stream multiplexer while the persistent design
//! single-stream-serialised everything.
//!
//! The refactor: each `SyncChannel` owns a `tokio::sync::mpsc::Sender<Vec<u8>>`
//! that pre-framed bytes are pushed onto. A single writer task drains the
//! mpsc and writes to the QUIC `SendStream`. User-visible `send()` is
//! non-blocking (returns when the frame is queued, not when written),
//! eliminating contention on the user-facing path. Wire byte ordering is
//! still serial — the mpsc is single-producer-from-the-stream's-perspective —
//! but the API-level concurrency is unconstrained.
//!
//! Bounded mpsc capacity = 1024 frames so backpressure flows: a writer that
//! can't keep up makes `send()` callers await for buffer space, rather than
//! the queue growing unboundedly under sustained overload.
//!
//! # Benefits
//!
//! - Eliminates stream-per-batch overhead on the wire (one stream per peer)
//! - Decouples API latency from QUIC write latency (mpsc queueing)
//! - Lets parallel callers of `send()` enqueue without contention
//! - Automatic reconnection on writer-task failure
//!
//! # Lock ordering
//!
//! ## `SyncChannel`
//!
//! | Field | Type | Protects |
//! |-------|------|----------|
//! | `send_tx` | `std::sync::RwLock<Option<mpsc::Sender<Vec<u8>>>>` | Outbound frame queue (writer task drain side) |
//! | `recv_task` | `tokio::sync::Mutex<Option<JoinHandle>>` | Background receiver handle |
//! | `writer_task` | `tokio::sync::Mutex<Option<JoinHandle>>` | Background writer handle |
//! | `state` | `std::sync::RwLock<ChannelState>` | Channel lifecycle state |
//! | `last_send` | `std::sync::RwLock<Instant>` | Timestamp bookkeeping |
//!
//! `send_tx` is held only briefly (clone the sender, drop the guard) on
//! the hot path; reconnect takes the write lock to swap senders.
//!
//! **Invariant:** never hold `state` (or `last_send`) while awaiting any
//! async lock. Sync `RwLock` guards are not `Send`.
//!
//! ## `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`.

#[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::{mpsc, 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;
#[cfg(feature = "automerge-backend")]
use crate::qos::QoSClass;

/// 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,
}

/// Outbound frame senders for a channel's two-lane writer (peat-mesh#247).
///
/// Both senders feed the same QUIC SendStream via the writer task, which
/// drains the priority lane ahead of the bulk lane. Holding both in one
/// struct keeps the reconnect/close lifecycle simple: dropping the
/// `Option<OutboundSenders>` drops both lanes together, so the writer sees
/// both receivers close and drains-then-finishes exactly as the
/// single-lane design did.
#[cfg(feature = "automerge-backend")]
struct OutboundSenders {
    /// Small / mission-priority frames. Drained first by the writer so a
    /// tiny interactive write (e.g. a status update) is not stuck behind a
    /// backlog of large state snapshots to the same peer.
    priority: mpsc::Sender<Vec<u8>>,
    /// Large, lower-priority frames (bulk state snapshots, archival docs).
    bulk: mpsc::Sender<Vec<u8>>,
}

/// 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 frame queues (priority + bulk lanes) → writer task → QUIC
    /// SendStream. Held as `Option` so reconnect can swap senders without
    /// dropping the channel; the writer task exits when both receiver
    /// halves drop. See [`OutboundSenders`] and [`Self::is_priority_frame`].
    send_tx: Arc<RwLock<Option<OutboundSenders>>>,
    /// Inbound receiver task handle
    recv_task: Arc<Mutex<Option<JoinHandle<()>>>>,
    /// Outbound writer task handle
    writer_task: Arc<Mutex<Option<JoinHandle<()>>>>,
    /// Current writer generation — incremented on every reconnect, used by
    /// `spawn_writer` to gate state writes against stale writers. See
    /// `spawn_writer`'s doc-comment for the race this closes.
    current_writer_generation: Arc<AtomicU64>,
    /// 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);
    /// Bounded mpsc capacity (per-peer frame count) for the outbound
    /// frame queue. Backpressure flows back to `send()` callers when the
    /// queue is full.
    ///
    /// This is a per-FRAME bound, not a per-BYTE bound. For the
    /// dominant `DeltaSync` / small `Tombstone` workload (avg ~500 B
    /// per frame), 1024 frames ≈ 512 KB per-peer ceiling. Under
    /// `StateSnapshot` pressure (a freshly-joined peer catching up,
    /// where each frame carries a full Automerge `doc.save()` blob —
    /// commonly many MB), 1024 frames can be gigabytes. Sized-Semaphore
    /// byte-accounting is the right next step if snapshot pressure
    /// becomes a deployment concern; until then, the frame bound is
    /// what we ship. peat-mesh#178 QA [SUGGESTION].
    const SEND_QUEUE_CAPACITY: usize = 1024;

    /// Frame-size threshold (bytes) for the priority lane (peat-mesh#247).
    ///
    /// Frames at or below this size take the priority lane regardless of
    /// QoS class — a small, interactive write (delta sync, status update,
    /// tombstone; the dominant traffic averages ~500 B/frame) must not wait
    /// behind a backlog of large `StateSnapshot` blobs serialized to the
    /// same peer's single QUIC stream. Larger frames are routed by QoS
    /// class instead (see [`Self::is_priority_frame`]). 16 KiB sits well
    /// above the typical delta/tombstone frame and below the full-document
    /// snapshots that constitute the bulk backlog.
    const PRIORITY_FRAME_BYTES: usize = 16 * 1024;

    /// 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")?;

        // Set up the outbound mpsc lanes + spawn the writer task. The
        // writer task owns the SendStream and drains both receivers
        // (priority ahead of bulk); user `send()` calls push pre-framed
        // bytes onto the lane selected by `is_priority_frame`.
        let (priority_tx, priority_rx) = mpsc::channel::<Vec<u8>>(Self::SEND_QUEUE_CAPACITY);
        let (bulk_tx, bulk_rx) = mpsc::channel::<Vec<u8>>(Self::SEND_QUEUE_CAPACITY);
        let state = Arc::new(RwLock::new(ChannelState::Connected));
        let current_writer_generation = Arc::new(AtomicU64::new(0));
        let writer = Self::spawn_writer(
            send,
            priority_rx,
            bulk_rx,
            peer_id,
            Arc::clone(&state),
            0,
            Arc::clone(&current_writer_generation),
        );

        let channel = Self {
            peer_id,
            transport,
            send_tx: Arc::new(RwLock::new(Some(OutboundSenders {
                priority: priority_tx,
                bulk: bulk_tx,
            }))),
            recv_task: Arc::new(Mutex::new(None)),
            writer_task: Arc::new(Mutex::new(Some(writer))),
            current_writer_generation,
            state,
            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 outbound writer task that drains the two lane queues and
    /// writes pre-framed bytes to the QUIC SendStream.
    ///
    /// The drain is a biased `select!`: the priority lane is always polled
    /// first, so a small / mission-priority frame is written ahead of any
    /// bulk frames already queued. A frame already mid-`write_all` is not
    /// preempted (frames are bounded), so a priority frame waits at most one
    /// in-flight bulk frame — collapsing the head-of-line stall a backlog of
    /// large snapshots otherwise imposes on the single per-peer stream
    /// (peat-mesh#247).
    ///
    /// On any write error the task marks the channel as Reconnecting
    /// (matching the receiver-task convention) and exits. Reconnect will
    /// swap in a new SendStream + new lanes + new writer task.
    ///
    /// When all `send_tx` senders are dropped (both lanes close together —
    /// they share one `OutboundSenders`), both receivers return None, the
    /// `else` arm fires and the task exits cleanly after draining whatever
    /// remained (priority first). Used by `close()` to terminate the writer
    /// without abort.
    ///
    /// `generation` defends against a reconnect race (peat-mesh#178 QA
    /// [WARNING]): a `send()` caller can clone the old mpsc::Sender
    /// before reconnect drops the channel's send_tx, keeping the old
    /// mpsc alive past `reconnect()`'s sender-drop step. The old writer
    /// then eventually fails its write_all, observes the shared `state`,
    /// and would clobber the freshly-installed `Connected` with
    /// `Reconnecting` — triggering an unnecessary reconnect cycle on
    /// what is, from the channel's perspective, a healthy stream. The
    /// writer only updates state if its generation still matches the
    /// channel's current generation; stale writers from prior incarnations
    /// log and exit silently.
    fn spawn_writer(
        mut send: SendStream,
        mut priority_rx: mpsc::Receiver<Vec<u8>>,
        mut bulk_rx: mpsc::Receiver<Vec<u8>>,
        peer_id: EndpointId,
        state: Arc<RwLock<ChannelState>>,
        generation: u64,
        current_generation: Arc<AtomicU64>,
    ) -> JoinHandle<()> {
        tokio::spawn(async move {
            tracing::debug!(peer = ?peer_id, gen = generation, "Sync channel writer started");
            loop {
                // Biased: drain the priority lane ahead of bulk. A closed-but-
                // empty lane's `recv()` returns None, disabling only that arm
                // for this iteration (so a closed priority lane falls through
                // to bulk); when BOTH lanes are closed and drained, `else`
                // fires and we break to the clean-shutdown finish() below.
                let frame = tokio::select! {
                    biased;
                    Some(frame) = priority_rx.recv() => frame,
                    Some(frame) = bulk_rx.recv() => frame,
                    else => break,
                };
                if let Err(e) = send.write_all(&frame).await {
                    if current_generation.load(Ordering::Acquire) == generation {
                        tracing::warn!(
                            peer = ?peer_id,
                            gen = generation,
                            error = %e,
                            "Sync channel writer error; marking channel Reconnecting"
                        );
                        *state.write().unwrap_or_else(|e| e.into_inner()) =
                            ChannelState::Reconnecting;
                    } else {
                        // Stale writer from a prior incarnation — a newer
                        // writer task is the current source of truth, do
                        // not clobber its state.
                        tracing::debug!(
                            peer = ?peer_id,
                            gen = generation,
                            current_gen = current_generation.load(Ordering::Acquire),
                            error = %e,
                            "Stale sync channel writer exiting silently"
                        );
                    }
                    return;
                }
            }
            // All senders dropped — clean shutdown (e.g. close() was called).
            // `send.finish()` half-closes our side; the peer's recv_loop
            // sees FinishedEarly(0) and terminates cleanly.
            let _ = send.finish();
            tracing::debug!(peer = ?peer_id, gen = generation, "Sync channel writer ended (queue closed)");
        })
    }

    /// 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 sync frames on the persistent
    /// stream (peat-mesh#175 / ADR-063).
    ///
    /// Frames use the v2 wire format
    /// `[2 doc_key_len][N doc_key][1 msg_type][4 payload_len][payload]`,
    /// identical to the legacy per-message stream's framing parsed by
    /// `AutomergeSyncCoordinator::receive_sync_payload_from_stream`. The loop
    /// reads one frame, dispatches it to `receive_batch_message`, and
    /// repeats until the peer half-closes the stream.
    ///
    /// Today `SyncChannelManager::send_*` always wraps its payload in a
    /// `SyncBatch` (`msg_type = 0x07`), so the persistent stream's traffic
    /// pattern is "stream of SyncBatch frames." Frames of other msg_types
    /// are decoded by `receive_sync_payload_from_stream` and re-wrapped
    /// into single-entry batches before dispatch, so the connect-side
    /// loop stays one entry point wide regardless of what variant arrives.
    /// Negentropy frames — which require a response on the same stream —
    /// are not expected on the connect side (the responder side handles
    /// queries), and are logged-and-skipped if one appears.
    ///
    /// The receive loop applies NO application-level idle timeout to the
    /// wait for the next frame (peat-mesh#239): a quiet channel is healthy,
    /// not failed, and an idle timeout here tore down live channels whose
    /// receiver was then never re-armed, wedging delivery after any quiet
    /// period. Connection liveness is delegated to QUIC (1s keepalive / 5s
    /// max_idle_timeout from the tactical transport config, Issue #315 /
    /// #226): a genuinely dead connection surfaces as a stream read error
    /// and drives reconnection; a live idle one simply blocks for the next
    /// frame. `read_exact`s within a frame inherit QUIC's stream timeouts.
    ///
    /// The pre-#175 implementation read `[1 marker][4 len][N data]`
    /// directly and rejected any frame where the first byte was not
    /// `SyncBatch (0x07)`. Because the sender writes the v2 framing
    /// (`[2 doc_key_len][5 "batch"][1 0x07]...`), the receiver
    /// desynchronized on the very first byte — `0x00` (high byte of
    /// `doc_key_len=5`) was warned-and-skipped as `"Unexpected message
    /// type"`, and every subsequent frame's parse was off by the same
    /// shift.
    async fn receive_loop(
        mut recv: RecvStream,
        peer_id: EndpointId,
        coordinator: Arc<AutomergeSyncCoordinator>,
        cancel: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<()> {
        use super::automerge_sync::ReceivedSyncPayload;

        loop {
            if let Some(ref token) = cancel {
                if token.is_cancelled() {
                    tracing::debug!(
                        peer = ?peer_id,
                        "Sync channel receive loop cancelled"
                    );
                    return Ok(());
                }
            }

            // Block for the next frame with NO application-level idle timeout.
            // A quiet channel (no sync traffic for a while) is healthy, not
            // failed — wrapping this wait in a 30s idle timeout (peat-mesh#239)
            // tore down idle-but-live channels, and the recovery path only
            // re-armed the writer, never the receiver, wedging delivery after
            // any quiet period until a full re-bootstrap. Connection liveness is
            // the QUIC layer's job: the tactical transport config applies a 1s
            // keepalive + 5s max_idle_timeout (Issue #315 / #226), so a genuinely
            // dead connection surfaces here as a stream read error (handled
            // below → reconnect), while a live idle one simply waits.
            let frame_result = if let Some(ref token) = cancel {
                tokio::select! {
                    res = coordinator.receive_sync_payload_from_stream(&mut recv) => res,
                    () = token.cancelled() => {
                        tracing::debug!(
                            peer = ?peer_id,
                            "Sync channel receive loop cancelled during read"
                        );
                        return Ok(());
                    }
                }
            } else {
                coordinator
                    .receive_sync_payload_from_stream(&mut recv)
                    .await
            };

            let (doc_key, payload, payload_size) = match frame_result {
                Ok(frame) => frame,
                Err(e) if super::automerge_sync::is_graceful_stream_end(&e) => {
                    tracing::debug!(
                        peer = ?peer_id,
                        "Sync channel stream ended cleanly at frame boundary"
                    );
                    return Ok(());
                }
                Err(e) => {
                    return Err(e.context("Failed to read sync frame on persistent channel"));
                }
            };

            // Repackage non-Batch variants into single-entry SyncBatches so
            // the connect-side loop has one dispatch entry point. Today
            // SyncChannelManager only sends Batch frames, but normalising
            // here keeps the loop forward-compatible.
            let batch = match payload {
                ReceivedSyncPayload::Batch(b) => b,
                ReceivedSyncPayload::Delta(message) => {
                    let mut b = SyncBatch::new();
                    b.entries.push(super::automerge_sync::SyncEntry::new(
                        doc_key.clone(),
                        SyncMessageType::DeltaSync,
                        message.encode(),
                    ));
                    b
                }
                ReceivedSyncPayload::StateSnapshot(bytes) => {
                    let mut b = SyncBatch::new();
                    b.entries.push(super::automerge_sync::SyncEntry::new(
                        doc_key.clone(),
                        SyncMessageType::StateSnapshot,
                        bytes,
                    ));
                    b
                }
                ReceivedSyncPayload::Tombstone(t) => {
                    let mut b = SyncBatch::new();
                    b.add_tombstone(&t);
                    b
                }
                ReceivedSyncPayload::TombstoneBatch(tb) => {
                    let mut b = SyncBatch::new();
                    for t in &tb.tombstones {
                        b.add_tombstone(t);
                    }
                    b
                }
                ReceivedSyncPayload::NegentropyInit(_)
                | ReceivedSyncPayload::NegentropyResponse(_) => {
                    tracing::warn!(
                        peer = ?peer_id,
                        doc = %doc_key,
                        "Ignoring Negentropy frame on connect-side persistent stream; responder side handles these"
                    );
                    continue;
                }
            };

            if let Err(e) = coordinator
                .receive_batch_message(peer_id, batch, payload_size)
                .await
            {
                tracing::warn!(peer = ?peer_id, error = %e, "Frame dispatch failed");
            }
        }
    }

    /// Send a batch through this channel.
    ///
    /// Pre-frames the batch using the v2 wire format
    /// `[2 doc_key_len][5 "batch"][1 0x07][4 batch_len][batch_bytes]` then
    /// pushes the framed bytes onto the per-peer mpsc queue. The writer
    /// task drains the queue and writes to QUIC. Multiple concurrent
    /// `send()` calls do not contend — they enqueue independently and the
    /// writer task serialises bytes on the wire.
    ///
    /// Awaits if the queue is at capacity (peat-mesh#175 UAT-driven
    /// backpressure): the bounded mpsc surfaces sustained-overload
    /// pressure back to callers rather than letting the queue grow
    /// without bound. Returns Err if the channel is Closed or if the
    /// writer task has exited (e.g. after a write error).
    ///
    /// **Delivery semantics:** `Ok(())` means "frame enqueued, will be
    /// best-effort delivered" — not "written to the wire." If the
    /// writer task subsequently fails its `write_all` (peer disconnect,
    /// QUIC reset), the in-flight frame and any frames queued behind it
    /// are dropped when the mpsc receiver is dropped on writer exit;
    /// the caller has no error signal beyond the channel's state going
    /// to `Reconnecting` on the next `send()`. For `Batch` / `DeltaSync`
    /// Automerge recovers via the next round of vector-clock
    /// negotiation. For `Tombstone` / `StateSnapshot` (unidirectional
    /// advisories), this is at-most-once-eventually with silent drops
    /// across reconnects — caller-side retry on top is still future
    /// work. peat-mesh#178 QA [SUGGESTION].
    ///
    /// **`bytes_sent` semantics:** the counter is incremented when the
    /// frame is enqueued, not when written to the wire. After a writer
    /// failure the counter overcounts by the failed frame's size; the
    /// drift is bounded by the mpsc queue depth (≤ `SEND_QUEUE_CAPACITY`
    /// frames per reconnect cycle). Acceptable for the `tracing` use
    /// cases the counter currently feeds; consumers that need
    /// on-wire-byte accuracy should source from QUIC stats instead.
    pub async fn send(&self, batch: &SyncBatch) -> Result<()> {
        // Check channel state (read and release the std::RwLock guard
        // before any .await — sync locks are not Send across awaits).
        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,
            }
        };

        if needs_reconnect {
            self.reconnect().await?;
        }

        // Pre-frame: build the v2 wire bytes in a single Vec so the
        // writer's single `write_all` is one syscall.
        let batch_bytes = batch.encode();
        let doc_key = b"batch";
        let total_bytes = 2 + doc_key.len() + 1 + 4 + batch_bytes.len();
        let mut frame = Vec::with_capacity(total_bytes);
        frame.extend_from_slice(&(doc_key.len() as u16).to_be_bytes());
        frame.extend_from_slice(doc_key);
        frame.push(SyncMessageType::SyncBatch as u8);
        frame.extend_from_slice(&(batch_bytes.len() as u32).to_be_bytes());
        frame.extend_from_slice(&batch_bytes);

        // Route to the priority or bulk lane (peat-mesh#247). Decide before
        // taking the lock so the lock is held only to clone the chosen
        // sender. Clone the sender (cheap — mpsc::Sender is internally
        // Arc'd), then drop the std::RwLock guard before awaiting on send.
        let priority = Self::is_priority_frame(batch, frame.len());
        let sender = {
            let guard = self.send_tx.read().unwrap_or_else(|e| e.into_inner());
            let senders = guard
                .as_ref()
                .ok_or_else(|| anyhow::anyhow!("Channel has no send queue"))?;
            if priority {
                senders.priority.clone()
            } else {
                senders.bulk.clone()
            }
        };

        sender
            .send(frame)
            .await
            .map_err(|_| anyhow::anyhow!("Sync channel writer task has exited"))?;

        // Update statistics. Bytes_sent is "bytes enqueued for write" —
        // not "bytes on the wire" — but the difference is the mpsc
        // queue depth, which is bounded and drains continuously.
        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!(
            "Enqueued batch {} ({} entries, {} bytes, lane={}) to peer {:?}",
            batch.batch_id,
            batch.len(),
            total_bytes,
            if priority { "priority" } else { "bulk" },
            self.peer_id
        );

        Ok(())
    }

    /// Decide which lane a framed batch takes (peat-mesh#247).
    ///
    /// Fast lane (`true`) when **either**:
    /// - the frame is small (`<= PRIORITY_FRAME_BYTES`) — the size tiebreak:
    ///   a tiny interactive write (delta sync, status update, tombstone)
    ///   must not wait behind a backlog of large state snapshots on the
    ///   single per-peer stream; **or**
    /// - any entry's collection is a mission-priority QoS class
    ///   (`Critical`/`High`) — large but latency-sensitive docs still
    ///   preempt bulk transfers.
    ///
    /// Everything else (large `Normal`/`Low`/`Bulk` frames — full-document
    /// snapshots, archival data) takes the bulk lane. `QoSClass` does not
    /// derive `Ord`, so priority is compared via the `repr(u8)` discriminant
    /// (lower = higher priority; `Critical = 1`, `High = 2`).
    fn is_priority_frame(batch: &SyncBatch, frame_len: usize) -> bool {
        if frame_len <= Self::PRIORITY_FRAME_BYTES {
            return true;
        }
        batch.entries.iter().any(|entry| {
            let collection = entry.doc_key.split(':').next().unwrap_or("");
            (QoSClass::for_collection(collection) as u8) <= (QoSClass::High as u8)
        })
    }

    /// Attempt to reconnect the channel.
    ///
    /// Tears down the old writer task by dropping its sender (the task
    /// observes `rx.recv() = None` and exits, half-closing its
    /// SendStream). Opens a fresh bi-stream, builds a new mpsc + writer
    /// task, and stores the new sender under `send_tx`.
    pub async fn reconnect(&self) -> Result<()> {
        *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
        );

        tokio::time::sleep(Self::RECONNECT_DELAY).await;

        let conn = self.transport.get_or_connect(&self.peer_id).await?;

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

        // Issue #435: explicitly stop recv on the reconnect stream — the
        // existing receiver task is bound to the original stream's recv
        // half (still owned by `recv_task`); the recv half of the new
        // bi-stream is unused by reconnect.
        let _ = recv.stop(0u32.into());

        // Drop the old sender first so the previous writer task drains
        // its queue and exits cleanly.
        {
            let mut guard = self.send_tx.write().unwrap_or_else(|e| e.into_inner());
            *guard = None;
        }
        if let Some(handle) = self.writer_task.lock().await.take() {
            // Best-effort: give the old writer a chance to drain. If it
            // hangs (e.g. peer not reading), abort after 500ms — we'd
            // rather take a fresh stream than wait indefinitely.
            let _ = tokio::time::timeout(Duration::from_millis(500), handle).await;
        }

        // Bump the writer generation BEFORE spawning the new writer.
        // Any old writer still alive (e.g. a `send()` caller held a
        // cloned sender past reconnect's sender-drop above) will load
        // a stale generation when it next tries to write and exit
        // silently instead of clobbering the freshly-Connected state.
        let new_generation = self
            .current_writer_generation
            .fetch_add(1, Ordering::AcqRel)
            + 1;

        let (priority_tx, priority_rx) = mpsc::channel::<Vec<u8>>(Self::SEND_QUEUE_CAPACITY);
        let (bulk_tx, bulk_rx) = mpsc::channel::<Vec<u8>>(Self::SEND_QUEUE_CAPACITY);
        let writer = Self::spawn_writer(
            send,
            priority_rx,
            bulk_rx,
            self.peer_id,
            Arc::clone(&self.state),
            new_generation,
            Arc::clone(&self.current_writer_generation),
        );

        {
            let mut guard = self.send_tx.write().unwrap_or_else(|e| e.into_inner());
            *guard = Some(OutboundSenders {
                priority: priority_tx,
                bulk: bulk_tx,
            });
        }
        *self.writer_task.lock().await = Some(writer);
        *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.
    ///
    /// Drops the mpsc sender so the writer task drains pending frames
    /// and exits cleanly (half-closing the SendStream via `finish()`).
    /// Aborts the receiver task — we don't need its remaining frames.
    pub async fn close(&self) {
        *self.state.write().unwrap_or_else(|e| e.into_inner()) = ChannelState::Closed;

        // Drop the sender → writer task observes rx.recv()=None and
        // exits after draining.
        {
            let mut guard = self.send_tx.write().unwrap_or_else(|e| e.into_inner());
            *guard = None;
        }

        // Receive task: abort outright. We don't process trailing
        // inbound frames after close.
        if let Some(task) = self.recv_task.lock().await.take() {
            task.abort();
        }

        // Writer task: await its clean exit briefly. If draining hangs
        // (peer not reading), abort.
        if let Some(handle) = self.writer_task.lock().await.take() {
            if tokio::time::timeout(Duration::from_millis(500), handle)
                .await
                .is_err()
            {
                tracing::debug!(
                    peer = ?self.peer_id,
                    "Writer task did not drain in 500ms; aborting"
                );
            }
        }

        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);
    }

    // ---- peat-mesh#247 priority-lane pins ----------------------------------

    use crate::storage::automerge_sync::SyncEntry;

    /// Build a single-entry `StateSnapshot` batch for `collection:id` with the
    /// given payload — the shape `SyncChannelManager::send_state_snapshot`
    /// produces, so it routes through `SyncChannel::send` like real traffic.
    fn snapshot_batch(collection: &str, id: &str, payload: Vec<u8>) -> SyncBatch {
        let mut batch = SyncBatch::new();
        batch.entries.push(SyncEntry::new(
            format!("{collection}:{id}"),
            SyncMessageType::StateSnapshot,
            payload,
        ));
        batch
    }

    /// Pins the lane routing policy (`is_priority_frame`): the size tiebreak,
    /// the QoS discriminant, and the size boundary. Catches the silent
    /// inversions QA flagged — flipping the `<=` discriminant (which would
    /// route Bulk to the priority lane), moving `PRIORITY_FRAME_BYTES`, or
    /// changing the boundary. peat-mesh#247.
    #[test]
    fn priority_frame_routing_policy() {
        // `sitreps` → Bulk QoS; `alerts` → Critical; `nodes` → High;
        // `beacons` → Normal (see qos::QoSClass::for_collection).

        // Size tiebreak: a small frame takes the priority lane regardless of
        // class — even a Bulk-QoS collection.
        let small = snapshot_batch("sitreps", "s", vec![0u8; 8]);
        assert!(
            SyncChannel::is_priority_frame(&small, 512),
            "small frame must take the priority lane regardless of QoS class"
        );

        // A large frame in a Bulk/Normal-QoS collection takes the bulk lane.
        let big_bulk = snapshot_batch("sitreps", "b", vec![0u8; 64 * 1024]);
        assert!(
            !SyncChannel::is_priority_frame(&big_bulk, 64 * 1024),
            "large Bulk-QoS frame must take the bulk lane"
        );
        let big_normal = snapshot_batch("beacons", "n", vec![0u8; 64 * 1024]);
        assert!(
            !SyncChannel::is_priority_frame(&big_normal, 64 * 1024),
            "large Normal-QoS frame must take the bulk lane"
        );

        // A large frame in a Critical/High-QoS collection still preempts: QoS
        // overrides size. (If the discriminant were inverted to `>=`, these
        // would route to bulk — the regression this pins.)
        let big_critical = snapshot_batch("alerts", "c", vec![0u8; 64 * 1024]);
        assert!(
            SyncChannel::is_priority_frame(&big_critical, 64 * 1024),
            "large Critical-QoS frame must take the priority lane"
        );
        let big_high = snapshot_batch("nodes", "h", vec![0u8; 64 * 1024]);
        assert!(
            SyncChannel::is_priority_frame(&big_high, 64 * 1024),
            "large High-QoS frame must take the priority lane"
        );

        // Size boundary is inclusive (`<=`).
        let at_boundary = snapshot_batch("sitreps", "edge", vec![]);
        assert!(
            SyncChannel::is_priority_frame(&at_boundary, SyncChannel::PRIORITY_FRAME_BYTES),
            "frame exactly at PRIORITY_FRAME_BYTES must take the priority lane"
        );
        assert!(
            !SyncChannel::is_priority_frame(&at_boundary, SyncChannel::PRIORITY_FRAME_BYTES + 1),
            "frame one byte over PRIORITY_FRAME_BYTES (Bulk QoS) must take the bulk lane"
        );
    }

    /// Bind two loopback iroh endpoints and return a connected `Connection`
    /// pair `(conn_a, conn_b)` (a accepts, b connects). Endpoints are returned
    /// so the caller keeps them alive for the duration of the test.
    async fn loopback_conn_pair() -> (
        iroh::Endpoint,
        iroh::Endpoint,
        iroh::endpoint::Connection,
        iroh::endpoint::Connection,
    ) {
        use iroh::address_lookup::memory::MemoryLookup;
        use iroh::Endpoint;

        const ALPN: &[u8] = b"test/priority-lane/1";
        let timeout = Duration::from_secs(15);

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::Empty)
            .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
            .address_lookup(MemoryLookup::new())
            .secret_key(iroh::SecretKey::from_bytes(&[7u8; 32]))
            .alpns(vec![ALPN.to_vec()])
            .bind()
            .await
            .expect("bind endpoint_a");

        let lookup_b = MemoryLookup::new();
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::Empty)
            .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
            .address_lookup(lookup_b.clone())
            .secret_key(iroh::SecretKey::from_bytes(&[8u8; 32]))
            .bind()
            .await
            .expect("bind endpoint_b");
        lookup_b.add_endpoint_info(endpoint_a.addr());

        let endpoint_a_clone = endpoint_a.clone();
        let accept_task = tokio::spawn(async move {
            let incoming = endpoint_a_clone.accept().await.expect("no incoming");
            let accepting = incoming.accept().expect("accept incoming");
            accepting.await.expect("accept handshake")
        });
        let conn_b = tokio::time::timeout(timeout, endpoint_b.connect(endpoint_a.id(), ALPN))
            .await
            .expect("b → a connect timed out")
            .expect("b → a connect");
        let conn_a = tokio::time::timeout(timeout, accept_task)
            .await
            .expect("a accept timed out")
            .expect("accept task panicked");
        (endpoint_a, endpoint_b, conn_a, conn_b)
    }

    /// End-to-end pin through the public `SyncChannel::send`: with bulk frames
    /// enqueued *first* and small/critical frames last, the writer must drain
    /// the priority lane ahead of the bulk lane on the wire. Catches a flipped
    /// `biased;` in `spawn_writer` and a writer that drains only one lane.
    /// peat-mesh#247.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_writes_priority_frames_ahead_of_queued_bulk() {
        let (_ep_a, _ep_b, conn_a, conn_b) = loopback_conn_pair().await;

        // A opens the bi-stream the writer owns; B reads the wire bytes.
        let (send_a, _recv_a) = conn_a.open_bi().await.expect("A open_bi");

        // Build the two lanes; park the senders in a `SyncChannel` so `send`
        // routes through the real public path, and hold the receivers to hand
        // to the writer after the frames are enqueued (so all frames are
        // buffered before the writer starts — makes the drain order
        // deterministic rather than timing-dependent).
        let (priority_tx, priority_rx) = mpsc::channel::<Vec<u8>>(SyncChannel::SEND_QUEUE_CAPACITY);
        let (bulk_tx, bulk_rx) = mpsc::channel::<Vec<u8>>(SyncChannel::SEND_QUEUE_CAPACITY);

        struct DummyTransport;
        impl SyncTransport for DummyTransport {
            fn get_connection(&self, _peer_id: &EndpointId) -> Option<iroh::endpoint::Connection> {
                None
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                Vec::new()
            }
        }

        let peer_id = iroh::SecretKey::from_bytes(&[9u8; 32]).public();
        let state = Arc::new(RwLock::new(ChannelState::Connected));
        let channel = SyncChannel {
            peer_id,
            transport: Arc::new(DummyTransport),
            send_tx: Arc::new(RwLock::new(Some(OutboundSenders {
                priority: priority_tx,
                bulk: bulk_tx,
            }))),
            recv_task: Arc::new(Mutex::new(None)),
            writer_task: Arc::new(Mutex::new(None)),
            current_writer_generation: Arc::new(AtomicU64::new(0)),
            state: Arc::clone(&state),
            reconnect_attempts: AtomicU32::new(0),
            last_send: Arc::new(RwLock::new(Instant::now())),
            bytes_sent: AtomicU64::new(0),
            batches_sent: AtomicU64::new(0),
        };

        // Enqueue 3 large Bulk-QoS frames FIRST, then 2 small Critical-QoS
        // frames. Each payload carries a distinguishable marker.
        let bulk_payload = |i: usize| {
            let mut p = b"BULKMARK".to_vec();
            p.resize(32 * 1024, b'0' + i as u8);
            p
        };
        for i in 0..3 {
            channel
                .send(&snapshot_batch(
                    "sitreps",
                    &format!("b{i}"),
                    bulk_payload(i),
                ))
                .await
                .expect("send bulk");
        }
        for i in 0..2 {
            channel
                .send(&snapshot_batch(
                    "alerts",
                    &format!("c{i}"),
                    b"CRITMARK".to_vec(),
                ))
                .await
                .expect("send critical");
        }

        // Now start the writer over the held receivers, then drop the channel
        // so both lane senders close and the writer drains-then-finishes.
        let writer = SyncChannel::spawn_writer(
            send_a,
            priority_rx,
            bulk_rx,
            peer_id,
            state,
            0,
            Arc::new(AtomicU64::new(0)),
        );
        drop(channel);

        // Read the whole wire stream on B and check marker order.
        let (_send_b, mut recv_b) = conn_b.accept_bi().await.expect("B accept_bi");
        let bytes =
            tokio::time::timeout(Duration::from_secs(15), recv_b.read_to_end(8 * 1024 * 1024))
                .await
                .expect("read timed out")
                .expect("read_to_end");
        let _ = writer.await;

        let find = |needle: &[u8]| -> Vec<usize> {
            bytes
                .windows(needle.len())
                .enumerate()
                .filter(|(_, w)| *w == needle)
                .map(|(i, _)| i)
                .collect::<Vec<_>>()
        };
        let crit = find(b"CRITMARK");
        let bulk = find(b"BULKMARK");
        assert_eq!(crit.len(), 2, "both critical frames must reach the wire");
        assert_eq!(bulk.len(), 3, "all bulk frames must reach the wire");
        assert!(
            crit.iter().max() < bulk.iter().min(),
            "priority frames must be written ahead of bulk frames on the wire \
             (crit offsets {crit:?}, bulk offsets {bulk:?}) — biased writer drain \
             (peat-mesh#247) regressed"
        );
    }
}
