monarch_rdma/backend/ibverbs/

queue_pair.rs

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * All rights reserved.
 *
 * This source code is licensed under the BSD-style license found in the
 * LICENSE file in the root directory of this source tree.
 */

//! ibverbs queue pair, doorbell, and completion polling.
//!
//! An [`IbvQueuePair`] encapsulates the send and receive queues, completion
//! queues, and other resources needed for RDMA communication. It provides
//! methods for establishing connections and performing RDMA operations.

/// Maximum size for a single RDMA operation in bytes (1 GiB).
const MAX_RDMA_MSG_SIZE: usize = 1024 * 1024 * 1024;

use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::io::Error;
use std::panic::AssertUnwindSafe;
use std::result::Result;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;

use async_trait::async_trait;
use futures::FutureExt as _;
use hyperactor::Actor;
use hyperactor::ActorHandle;
use hyperactor::ActorId;
use hyperactor::ActorRef;
use hyperactor::Context;
use hyperactor::Endpoint as _;
use hyperactor::Handler;
use hyperactor::Instance;
use hyperactor::PortHandle;
use hyperactor::actor::ActorError;
use hyperactor::actor::Referable;
use hyperactor::actor::RemoteHandles;
use hyperactor::context::Mailbox;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use typeuri::Named;

use super::cq_actor::Attach;
use super::cq_actor::CompletionInbox;
use super::cq_actor::CompletionQueueActor;
use super::cq_actor::CompletionResult;
use super::cq_actor::CompletionRoute;
use super::cq_actor::CqId;
use super::cq_actor::Detach;
use super::cq_actor::DetachedQueuePair;
use super::cq_actor::IbvCompletionQueue as _;
use super::cq_actor::PollerWake;
use super::cq_pool::CqLease;
use super::domain::IbvDomain;
use super::domain::IbvDomainImpl;
use super::manager_actor::CreatePeerQueuePair;
use super::memory_region::IbvMemoryRegionView;
use super::memory_region::IbvRemoteMemoryRegionView;
use super::primitives::Gid;
use super::primitives::GidScope;
use super::primitives::GidType;
use super::primitives::IbvConfig;
use super::primitives::IbvCq;
use super::primitives::IbvOperation;
use super::primitives::IbvQp;
use super::primitives::IbvQpInfo;
#[cfg(test)]
use super::primitives::IbvWc;
use super::primitives::resolve_qp_type;
use crate::RdmaOpType;
use crate::local_memory::KeepaliveLocalMemory;

/// A per-work-request completion failure: a work request completed with
/// a non-success `ibv_wc_status`. Carries the `wr_id`, status, and vendor
/// error code so callers can correlate the failure to the request and
/// match on the status without string parsing. The QP may or may not be
/// in error state.
#[derive(Debug)]
pub struct WorkRequestError {
    pub wr_id: u64,
    pub status: rdmaxcel_sys::ibv_wc_status::Type,
    pub vendor_err: u32,
    message: String,
}

impl std::fmt::Display for WorkRequestError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.message)
    }
}

impl std::error::Error for WorkRequestError {}

impl WorkRequestError {
    /// Returns `true` when the completion status is `IBV_WC_WR_FLUSH_ERR`,
    /// which typically indicates a secondary failure after the QP entered
    /// error state due to a different work request's failure.
    pub fn is_wr_flush_err(&self) -> bool {
        self.status == rdmaxcel_sys::ibv_wc_status::IBV_WC_WR_FLUSH_ERR
    }

    /// Builds the failure a work completion reported, formatting its message from
    /// the status fields.
    pub(super) fn from_status(
        wr_id: u64,
        status: rdmaxcel_sys::ibv_wc_status::Type,
        vendor_err: u32,
    ) -> Self {
        Self {
            wr_id,
            status,
            vendor_err,
            message: format!(
                "completion failed for wr_id={wr_id}: status={status:?}, vendor_err={vendor_err}"
            ),
        }
    }

    #[cfg(test)]
    pub(super) fn for_test(wr_id: u64, message: &str) -> Self {
        Self {
            wr_id,
            status: rdmaxcel_sys::ibv_wc_status::IBV_WC_GENERAL_ERR,
            vendor_err: 0,
            message: message.to_string(),
        }
    }
}

/// A CQ-level poll failure: `ibv_poll_cq` itself failed, naming no work request.
/// The entry that caused it, if any, has been consumed.
#[derive(Debug)]
pub struct PollCompletionError {
    message: String,
}

impl std::fmt::Display for PollCompletionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.message)
    }
}

impl std::error::Error for PollCompletionError {}

impl PollCompletionError {
    /// A poll that failed, described by `message`.
    pub(super) fn new(message: String) -> Self {
        Self { message }
    }
}

/// A doorbell trigger for batched RDMA operations.
///
/// Rings the hardware doorbell to execute previously enqueued work requests.
#[derive(Debug, Named, Clone, Serialize, Deserialize)]
pub struct DoorBell {
    pub src_ptr: usize,
    pub dst_ptr: usize,
    pub size: usize,
}
wirevalue::register_type!(DoorBell);

/// Specifies which completion queue to poll.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum PollTarget {
    Send,
    Recv,
}

/// The legacy single-type queue pair, retained while the backends migrate onto
/// the [`IbvQueuePair`] trait.
pub mod legacy;

/// Identifies a per-peer queue pair held by one
/// [`super::manager_actor::IbvManagerActor`]. The same conceptual
/// QP is referenced by two distinct keys, one from each side: each
/// manager stores the local view (its own device, the peer's actor
/// id, the peer's device).
#[derive(Clone, Hash, Eq, PartialEq, Debug, Serialize, Deserialize, Named)]
pub(super) struct QpKey {
    pub(super) self_device: String,
    pub(super) other_id: ActorId,
    pub(super) other_device: String,
    pub(super) qp_index: usize,
}

// =====================================================================
// QueuePairActor
// =====================================================================
//
// Owns one simplex queue pair to a peer. The active side spawns a
// `QueuePairActor` that owns and operates the local QP; the peer's
// manager eagerly creates its mirror QP during the handshake, stores
// it as a passive endpoint, and never wraps it in an actor. If the
// peer later wants to RDMA us, it spawns its own `QueuePairActor`
// locally, which creates a fresh pair the same way.

/// A NIC-backend queue pair: the unit of RDMA communication between two
/// endpoints, and the operations a [`QueuePairActor`] performs on it. Each
/// [`IbvDomainImpl`] names its concrete queue-pair type via
/// [`IbvDomainImpl::QueuePair`](super::domain::IbvDomainImpl::QueuePair) and builds
/// it through [`Self::new`].
pub trait IbvQueuePair: std::fmt::Debug + Send + Sync + 'static + Sized {
    /// Creates a queue pair against `domain` in the RESET state, reporting its
    /// completions on `send_cq` and `recv_cq`; [`Self::connect`] transitions it
    /// to RTS before use.
    ///
    /// The completion queues are created by the caller rather than here, so that
    /// one queue can back many queue pairs. Implementers hold a clone of each,
    /// which is what keeps them alive for the QP's lifetime.
    ///
    /// # Safety
    ///
    /// `domain`'s PD (`domain.as_ptr()`) must be null or a valid protection
    /// domain. Callers must ensure the PD outlives the QP; the easiest way to
    /// do this is for implementers to store a clone of `domain.pd()` (an
    /// `Arc<IbvPd>`) inside the QP. `send_cq` and `recv_cq` must have been
    /// created on `domain`'s device context.
    unsafe fn new<I: IbvDomainImpl<QueuePair = Self>>(
        domain: &IbvDomain<I>,
        config: IbvConfig,
        send_cq: Arc<IbvCq>,
        recv_cq: Arc<IbvCq>,
    ) -> Result<Self, anyhow::Error>;

    /// Transitions the QP through `INIT -> RTR -> RTS`, connected to `info`.
    fn connect(&mut self, info: &IbvQpInfo) -> Result<(), anyhow::Error>;

    /// Returns the local endpoint other peers need in order to connect to this
    /// QP.
    fn get_qp_info(&mut self) -> Result<IbvQpInfo, anyhow::Error>;

    /// Returns the current `ibv_qp_state` of the QP.
    fn state(&mut self) -> Result<u32, anyhow::Error>;

    /// Largest transfer this QP issues as a single work request; [`Self::put`]
    /// and [`Self::get`] split anything larger into that many WRs. The owning
    /// [`QueuePairActor`] budgets send-queue credits against this value, so an
    /// implementation that chunks more finely must report it here or it will
    /// over-commit the send queue.
    fn max_msg_size(&self) -> usize {
        MAX_RDMA_MSG_SIZE
    }

    /// Post an RDMA WRITE of `local_src` into `remote_dst`. The request may
    /// be chunked into multiple WRs. This method returns the list of WR ids
    /// that were posted.
    fn put(
        &mut self,
        remote_dst: IbvRemoteMemoryRegionView,
        local_src: IbvMemoryRegionView,
    ) -> Result<Vec<u64>, anyhow::Error>;

    /// Post an RDMA READ of `remote_src` into `local_dst`. The request may
    /// be chunked into multiple WRs. This method returns the list of WR ids
    /// that were posted.
    fn get(
        &mut self,
        local_dst: IbvMemoryRegionView,
        remote_src: IbvRemoteMemoryRegionView,
    ) -> Result<Vec<u64>, anyhow::Error>;
}

/// Queries the local endpoint info for `qp`, whose device `context` and the QP
/// `config` (port, PSN) describe the connection. `gid` is the port's source GID.
///
/// # Safety
///
/// `qp` must be a live `ibv_qp` (non-null) and `context` must be its live device
/// context.
pub(super) unsafe fn get_qp_info(
    qp: *mut rdmaxcel_sys::ibv_qp,
    context: *mut rdmaxcel_sys::ibv_context,
    config: &IbvConfig,
    gid: Gid,
) -> Result<IbvQpInfo, anyhow::Error> {
    let mut port_attr = rdmaxcel_sys::ibv_port_attr::default();
    // SAFETY: `context` is a live device context (caller contract); the
    // out-param is a writable, properly aligned `ibv_port_attr`. `ibv_query_port`
    // returns the errno on failure.
    let errno = unsafe {
        rdmaxcel_sys::ibv_query_port(
            context,
            config.port_num,
            &mut port_attr as *mut rdmaxcel_sys::ibv_port_attr as *mut _,
        )
    };
    if errno != 0 {
        return Err(anyhow::anyhow!(
            "failed to query port attributes: {}",
            Error::from_raw_os_error(errno)
        ));
    }

    // SAFETY: `qp` is a live `ibv_qp` (caller contract).
    let qp_num = unsafe { (*qp).qp_num };
    Ok(IbvQpInfo {
        qp_num,
        lid: port_attr.lid,
        gid: Some(gid),
        psn: config.psn,
    })
}

/// Returns the current `ibv_qp_state` of `qp`.
///
/// # Safety
///
/// `qp` must be a live `ibv_qp` (non-null).
pub(super) unsafe fn state(qp: *mut rdmaxcel_sys::ibv_qp) -> Result<u32, anyhow::Error> {
    let mut qp_attr = rdmaxcel_sys::ibv_qp_attr::default();
    let mut qp_init_attr = rdmaxcel_sys::ibv_qp_init_attr::default();
    let mask = rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_STATE;
    // SAFETY: `qp` wraps a live `ibv_qp` (caller contract); the out-params are
    // writable, properly aligned attr structs. `ibv_query_qp` returns the errno.
    let errno =
        unsafe { rdmaxcel_sys::ibv_query_qp(qp, &mut qp_attr, mask.0 as i32, &mut qp_init_attr) };
    if errno != 0 {
        return Err(anyhow::anyhow!(
            "failed to query QP state: {}",
            Error::from_raw_os_error(errno)
        ));
    }
    Ok(qp_attr.qp_state)
}

/// Transitions `qp` through `INIT -> RTR -> RTS`, connected to `info`, granting
/// remote peers `access_flags` and using the connection parameters in `config`.
/// `sgid_index` is the local source-GID table index, used as the address
/// handle's `sgid_index` when `info` carries a (remote) GID.
///
/// # Safety
///
/// `qp` must be a live `ibv_qp` (non-null).
pub(super) unsafe fn connect(
    qp: *mut rdmaxcel_sys::ibv_qp,
    config: &IbvConfig,
    access_flags: i32,
    info: &IbvQpInfo,
    sgid_index: u8,
) -> Result<(), anyhow::Error> {
    // Transition to INIT.
    let mut qp_attr = rdmaxcel_sys::ibv_qp_attr {
        qp_state: rdmaxcel_sys::ibv_qp_state::IBV_QPS_INIT,
        qp_access_flags: access_flags as u32,
        pkey_index: config.pkey_index,
        port_num: config.port_num,
        ..Default::default()
    };
    let mask = rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_STATE
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_PKEY_INDEX
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_PORT
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_ACCESS_FLAGS;
    // SAFETY: `qp` is a live `ibv_qp` (caller contract); `qp_attr` is a valid
    // `ibv_qp_attr` whose populated fields match `mask`. `ibv_modify_qp` returns
    // the errno.
    let errno = unsafe { rdmaxcel_sys::ibv_modify_qp(qp, &mut qp_attr, mask.0 as i32) };
    if errno != 0 {
        return Err(anyhow::anyhow!(
            "failed to transition QP to INIT: {}",
            Error::from_raw_os_error(errno)
        ));
    }

    // Transition to RTR (Ready to Receive).
    let mut qp_attr = rdmaxcel_sys::ibv_qp_attr {
        qp_state: rdmaxcel_sys::ibv_qp_state::IBV_QPS_RTR,
        path_mtu: config.path_mtu,
        dest_qp_num: info.qp_num,
        rq_psn: info.psn,
        max_dest_rd_atomic: config.max_dest_rd_atomic,
        min_rnr_timer: config.min_rnr_timer,
        ah_attr: rdmaxcel_sys::ibv_ah_attr {
            dlid: info.lid,
            sl: 0,
            src_path_bits: 0,
            port_num: config.port_num,
            grh: Default::default(),
            ..Default::default()
        },
        ..Default::default()
    };
    if let Some(gid) = info.gid {
        qp_attr.ah_attr.is_global = 1;
        qp_attr.ah_attr.grh.dgid = rdmaxcel_sys::ibv_gid::from(gid);
        qp_attr.ah_attr.grh.hop_limit = 0xff;
        qp_attr.ah_attr.grh.sgid_index = sgid_index;
    } else {
        qp_attr.ah_attr.is_global = 0;
    }
    let mask = rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_STATE
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_AV
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_PATH_MTU
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_DEST_QPN
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_RQ_PSN
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_MAX_DEST_RD_ATOMIC
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_MIN_RNR_TIMER;
    // SAFETY: as for the INIT transition above.
    let errno = unsafe { rdmaxcel_sys::ibv_modify_qp(qp, &mut qp_attr, mask.0 as i32) };
    if errno != 0 {
        return Err(anyhow::anyhow!(
            "failed to transition QP to RTR: {}",
            Error::from_raw_os_error(errno)
        ));
    }

    // Transition to RTS (Ready to Send).
    let mut qp_attr = rdmaxcel_sys::ibv_qp_attr {
        qp_state: rdmaxcel_sys::ibv_qp_state::IBV_QPS_RTS,
        sq_psn: config.psn,
        max_rd_atomic: config.max_rd_atomic,
        retry_cnt: config.retry_cnt,
        rnr_retry: config.rnr_retry,
        timeout: config.qp_timeout,
        ..Default::default()
    };
    let mask = rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_STATE
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_TIMEOUT
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_RETRY_CNT
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_SQ_PSN
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_RNR_RETRY
        | rdmaxcel_sys::ibv_qp_attr_mask::IBV_QP_MAX_QP_RD_ATOMIC;
    // SAFETY: as for the INIT transition above.
    let errno = unsafe { rdmaxcel_sys::ibv_modify_qp(qp, &mut qp_attr, mask.0 as i32) };
    if errno != 0 {
        return Err(anyhow::anyhow!(
            "failed to transition QP to RTS: {}",
            Error::from_raw_os_error(errno)
        ));
    }
    Ok(())
}

/// An RDMA reliable-connected (RC) queue pair built on plain ibverbs
/// (`ibv_post_send`), independent of any device-specific verbs.
///
/// Single-owner: it owns the [`IbvQp`] — which in turn owns its two completion
/// queues and the protection domain — and destroys them on drop, so the type is
/// intentionally `!Clone`. The device context and completion queues are reached
/// through the [`IbvQp`].
#[derive(Debug)]
pub struct RCQueuePair {
    qp: IbvQp,
    config: IbvConfig,
    /// The source GID (carrying its table index), resolved from the owning
    /// device's `IbvDeviceInfo` at construction: the first global RoCE v2 GID on
    /// `config.port_num`.
    gid: Gid,
    /// Remote-access flags granted to peers at connect time, taken from the
    /// owning domain at construction.
    access_flags: i32,
    /// Monotonic work-request id, handed out one per posted WR. Standard
    /// ibverbs carries no internal counter, so the QP tracks its own.
    next_wr_id: u64,
}

impl RCQueuePair {
    /// Assembles an `RCQueuePair` that owns `qp` (and, through it, its
    /// completion queues, protection domain, and device context). `access_flags`
    /// is granted to peers at [`Self::connect`]; it is taken from the owning
    /// domain by the caller.
    ///
    /// # Safety
    ///
    /// `qp` must wrap a live, non-null `ibv_qp`, and everything reached through
    /// it — its send/recv completion queues, protection domain, and device
    /// context — must likewise be non-null and valid: the data-path verbs invoke
    /// them without re-checking.
    pub(super) unsafe fn from_qp(
        qp: IbvQp,
        config: IbvConfig,
        gid: Gid,
        access_flags: i32,
    ) -> Self {
        RCQueuePair {
            qp,
            config,
            gid,
            access_flags,
            next_wr_id: 0,
        }
    }

    /// Posts `op` over `[laddr, laddr + total_size)` to `[raddr, ...)`,
    /// splitting into `MAX_RDMA_MSG_SIZE`-bound chunks and returning one wr_id
    /// per chunk.
    fn post_chunked(
        &mut self,
        op: IbvOperation,
        laddr: usize,
        lkey: u32,
        raddr: usize,
        rkey: u32,
        total_size: usize,
    ) -> Result<Vec<u64>, anyhow::Error> {
        let mut remaining = total_size;
        let mut offset = 0;
        let mut wr_ids = Vec::new();
        while remaining > 0 {
            let chunk = std::cmp::min(remaining, MAX_RDMA_MSG_SIZE);
            let wr_id = self.next_wr_id;
            self.next_wr_id += 1;
            self.post_one(op, laddr + offset, lkey, chunk, raddr + offset, rkey, wr_id)?;
            wr_ids.push(wr_id);
            remaining -= chunk;
            offset += chunk;
        }
        Ok(wr_ids)
    }

    /// Posts a single signaled RDMA `op` work request via `ibv_post_send`.
    fn post_one(
        &self,
        op: IbvOperation,
        laddr: usize,
        lkey: u32,
        length: usize,
        raddr: usize,
        rkey: u32,
        wr_id: u64,
    ) -> Result<(), anyhow::Error> {
        let qp = self.qp.as_ptr();
        let context = self.qp.context().as_ptr();
        let mut sge = rdmaxcel_sys::ibv_sge {
            addr: laddr as u64,
            length: length as u32,
            lkey,
        };
        let mut wr = rdmaxcel_sys::ibv_send_wr {
            wr_id,
            next: std::ptr::null_mut(),
            sg_list: &mut sge as *mut _,
            num_sge: 1,
            opcode: op.into(),
            send_flags: rdmaxcel_sys::ibv_send_flags::IBV_SEND_SIGNALED.0,
            wr: Default::default(),
            qp_type: Default::default(),
            __bindgen_anon_1: Default::default(),
            __bindgen_anon_2: Default::default(),
        };
        // Set the RDMA target. Writing through a union field is safe; the device
        // reads the `rdma` member for the RDMA_WRITE/RDMA_READ opcodes set above.
        wr.wr.rdma.remote_addr = raddr as u64;
        wr.wr.rdma.rkey = rkey;
        let mut bad_wr: *mut rdmaxcel_sys::ibv_send_wr = std::ptr::null_mut();
        // SAFETY: `context` is the QP's live device context (read from the live
        // `qp`); we invoke its `post_send` verb through the ops table. `qp` is
        // live and `wr`/`sge`/`bad_wr` are valid for the duration of the call.
        let errno = unsafe {
            let post_send = (*context)
                .ops
                .post_send
                .expect("post_send verb missing from ibv_context ops");
            post_send(qp, &mut wr as *mut _, &mut bad_wr)
        };
        if errno != 0 {
            return Err(anyhow::anyhow!(
                "failed to post {:?} request: {}",
                op,
                Error::from_raw_os_error(errno)
            ));
        }
        Ok(())
    }
}

impl IbvQueuePair for RCQueuePair {
    unsafe fn new<I: IbvDomainImpl<QueuePair = Self>>(
        domain: &IbvDomain<I>,
        config: IbvConfig,
        send_cq: Arc<IbvCq>,
        recv_cq: Arc<IbvCq>,
    ) -> Result<Self, anyhow::Error> {
        tracing::debug!("creating an RCQueuePair from config {}", config);
        // `IbvDomain`'s `pd` accessor permits null (e.g. a test domain); a real
        // QP needs one, so reject null up front.
        let pd = domain.as_ptr();
        if pd.is_null() {
            anyhow::bail!("cannot create an RCQueuePair on a null protection domain");
        }

        // Resolve the source GID up front (before allocating any FFI resources),
        // so a port without a global RoCE v2 GID fails cleanly here.
        let gid = domain.device_info().select_gid(
            config.port_num,
            Some(GidScope::Global),
            Some(GidType::RoCEv2),
        )?;

        // A standard RC QP with the caps from `config`.
        let mut init_attr = rdmaxcel_sys::ibv_qp_init_attr {
            send_cq: send_cq.as_ptr(),
            recv_cq: recv_cq.as_ptr(),
            cap: rdmaxcel_sys::ibv_qp_cap {
                max_send_wr: config.max_send_wr,
                max_recv_wr: config.max_recv_wr,
                max_send_sge: config.max_send_sge,
                max_recv_sge: config.max_recv_sge,
                max_inline_data: 0,
            },
            qp_type: rdmaxcel_sys::ibv_qp_type::IBV_QPT_RC,
            sq_sig_all: 0,
            ..Default::default()
        };
        // SAFETY: `pd` is non-null (checked above) and live per `IbvDomain`'s
        // construction contract; `init_attr` is a fully initialized
        // `ibv_qp_init_attr`. `ibv_create_qp` returns null on failure.
        let qp = unsafe { rdmaxcel_sys::ibv_create_qp(pd, &mut init_attr) };
        if qp.is_null() {
            anyhow::bail!(
                "failed to create queue pair (QP): {}",
                Error::last_os_error()
            );
        }
        // SAFETY: `qp` is a live RC QP just created against `pd` with
        // `send_cq`/`recv_cq`; `IbvQp` holds a clone of each, keeping them alive
        // for at least as long as the QP it destroys on drop.
        let qp = unsafe { IbvQp::from_raw(qp, send_cq, recv_cq, domain.pd().clone()) };
        let access_flags = domain.access_flags();
        // SAFETY: `qp` and its `send_cq`/`recv_cq`/PD/context were all created
        // non-null.
        Ok(unsafe { Self::from_qp(qp, config, gid, access_flags) })
    }

    fn connect(&mut self, info: &IbvQpInfo) -> Result<(), anyhow::Error> {
        // SAFETY: `self.qp` is the live QP, kept alive for `self`'s lifetime.
        unsafe {
            connect(
                self.qp.as_ptr(),
                &self.config,
                self.access_flags,
                info,
                self.gid.index(),
            )
        }
    }

    fn get_qp_info(&mut self) -> Result<IbvQpInfo, anyhow::Error> {
        let context = self.qp.context().as_ptr();
        // SAFETY: `self.qp` is the live QP and `context` its non-null device
        // context (validated inside `new`), both valid for `self`'s lifetime.
        unsafe { get_qp_info(self.qp.as_ptr(), context, &self.config, self.gid) }
    }

    fn state(&mut self) -> Result<u32, anyhow::Error> {
        // SAFETY: `self.qp` is the live QP, kept alive for `self`'s lifetime.
        unsafe { state(self.qp.as_ptr()) }
    }

    fn put(
        &mut self,
        remote_dst: IbvRemoteMemoryRegionView,
        local_src: IbvMemoryRegionView,
    ) -> Result<Vec<u64>, anyhow::Error> {
        if remote_dst.size < local_src.size {
            return Err(anyhow::anyhow!(
                "remote buffer size ({}) is smaller than local buffer size ({})",
                remote_dst.size,
                local_src.size
            ));
        }
        self.post_chunked(
            IbvOperation::Write,
            local_src.rdma_addr,
            local_src.lkey,
            remote_dst.addr,
            remote_dst.rkey,
            local_src.size,
        )
    }

    fn get(
        &mut self,
        local_dst: IbvMemoryRegionView,
        remote_src: IbvRemoteMemoryRegionView,
    ) -> Result<Vec<u64>, anyhow::Error> {
        if local_dst.size < remote_src.size {
            return Err(anyhow::anyhow!(
                "local buffer size ({}) is smaller than remote buffer size ({})",
                local_dst.size,
                remote_src.size
            ));
        }
        self.post_chunked(
            IbvOperation::Read,
            local_dst.rdma_addr,
            local_dst.lkey,
            remote_src.addr,
            remote_src.rkey,
            remote_src.size,
        )
    }
}

/// Bundle of trait bounds for an actor type that can serve as the
/// peer manager — i.e. the recipient of [`CreatePeerQueuePair`].
pub(super) trait Manager:
    Actor + Referable + RemoteHandles<CreatePeerQueuePair<Self>>
{
}

impl<T> Manager for T where T: Actor + Referable + RemoteHandles<CreatePeerQueuePair<T>> {}

/// Identifies one stripe of an operation in a submission batch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct StripeId {
    pub(super) op_idx: usize,
    pub(super) stripe_idx: usize,
    pub(super) stripe_count: usize,
}

impl std::fmt::Display for StripeId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "op_idx={}, stripe {}/{}",
            self.op_idx, self.stripe_idx, self.stripe_count
        )
    }
}

/// Per-stripe completion result sent from a queue-pair worker to the submitter.
#[derive(Debug)]
pub(super) struct StripeResult {
    pub(super) stripe_id: StripeId,
    pub(super) result: Result<(), String>,
}

/// One transfer stripe assigned to a queue pair.
#[derive(Debug)]
pub(super) struct QueuePairOp {
    pub(super) stripe_id: StripeId,
    /// Selects one of the queue pairs for this NIC route.
    pub(super) qp_index: usize,
    pub(super) op_type: RdmaOpType,
    pub(super) local_memory: KeepaliveLocalMemory,
    /// The local and remote views are already sliced to this stripe's range.
    pub(super) local: IbvMemoryRegionView,
    pub(super) remote: IbvRemoteMemoryRegionView,
}

/// Local-only request: enqueue a batch of ops on this QP. As each op resolves
/// the worker sends one [`StripeResult`] on `reply`. Its inner `Result` is `Ok(())`
/// if every WR for the op completed successfully, otherwise `Err` carrying the
/// first per-WR error observed (held back until the op's other WRs also report,
/// so the memory handle and MR registration outlive the data path).
#[derive(Debug)]
pub(super) struct ProcessOps {
    pub(super) items: Vec<QueuePairOp>,
    pub(super) reply: mpsc::UnboundedSender<StripeResult>,
}

/// An op accepted by the worker but not yet posted to the QP.
#[derive(Debug)]
struct PendingOp {
    op: QueuePairOp,
    reply: mpsc::UnboundedSender<StripeResult>,
    /// WR count this op will issue when posted, computed once at
    /// construction so retries from a credit head-block don't redo
    /// the work.
    wrs: u32,
}

/// State of an op whose WRs are in flight on the QP.
#[derive(Debug)]
struct PostedOpEntry {
    stripe_id: StripeId,
    pending_wrs: HashSet<u64>,
    /// Kept alive so the memory outlives every in-flight WR touching it.
    _local_memory: KeepaliveLocalMemory,
    /// Kept alive so the memory *registration* outlives every in-flight
    /// WR touching it.
    _local_mrv: IbvMemoryRegionView,
    reply: mpsc::UnboundedSender<StripeResult>,
    /// First per-WR error observed for this op. The op's final reply is
    /// held back until `pending_wrs.is_empty()`, so the two fields above
    /// outlive every WR still in flight.
    first_error: Option<String>,
}

/// Owns one queue pair and the state used to schedule and complete its work.
#[derive(Debug)]
struct QueuePairState<Qp: IbvQueuePair> {
    qp_key: QpKey,
    qp: Qp,
    /// QP-wide cap on outstanding send-queue WRs (reads + writes).
    max_send_wr: u32,
    /// Single FIFO of ops awaiting their first post attempt. If the head
    /// op does not fit in the free send-queue slots, ops queued behind it
    /// stall even where they would have fit; smarter interleaving is
    /// future work.
    queue: VecDeque<PendingOp>,
    /// op-id → entry, for tracking WR completion. op-ids are local
    /// to this queue pair (monotonic counter); they exist so we can
    /// route per-WR completions to the right `PostedOpEntry`
    /// without making any assumptions about uniqueness of `op_idx`
    /// across batches.
    posted: HashMap<u64, PostedOpEntry>,
    /// wr_id → local op-id.
    wr_to_op: HashMap<u64, u64>,
    next_op_id: u64,
    /// WRs posted to the send queue and not yet reaped:
    /// what `max_send_wr` gates against.
    in_flight: u32,
    /// Work requests posted by this queue pair. The completion poller compares
    /// this with its consumed count to decide when polling is useful.
    posted_wrs: Arc<AtomicU64>,
    poller_wake: Option<PollerWake>,
}

#[derive(Debug)]
struct DetachedQueuePairResources<Qp> {
    _qp: Qp,
    _posted: HashMap<u64, PostedOpEntry>,
}

#[derive(Debug)]
struct QueuePairWorkerFailed {
    error: String,
}

/// Local data-plane handle for submitting work to a queue pair while retaining
/// its actor handle for lifecycle management.
#[derive(Debug)]
pub(super) struct QueuePairHandle<M: Manager, Qp: IbvQueuePair> {
    actor: ActorHandle<QueuePairActor<M, Qp>>,
    sender: mpsc::UnboundedSender<ProcessOps>,
}

impl<M: Manager, Qp: IbvQueuePair> Clone for QueuePairHandle<M, Qp> {
    fn clone(&self) -> Self {
        Self {
            actor: self.actor.clone(),
            sender: self.sender.clone(),
        }
    }
}

impl<M: Manager, Qp: IbvQueuePair> QueuePairHandle<M, Qp> {
    pub(super) fn new(
        actor: ActorHandle<QueuePairActor<M, Qp>>,
        sender: mpsc::UnboundedSender<ProcessOps>,
    ) -> Self {
        Self { actor, sender }
    }

    pub(super) fn send(&self, message: ProcessOps) -> Result<(), anyhow::Error> {
        self.sender
            .send(message)
            .map_err(|_| anyhow::anyhow!("queue pair worker stopped"))
    }

    pub(super) fn sender(&self) -> mpsc::UnboundedSender<ProcessOps> {
        self.sender.clone()
    }

    pub(super) fn drain_and_stop(&self, reason: &str) -> Result<(), ActorError> {
        self.actor.drain_and_stop(reason)
    }

    #[cfg(test)]
    fn actor(&self) -> &ActorHandle<QueuePairActor<M, Qp>> {
        &self.actor
    }
}

#[derive(Debug)]
struct QueuePairWorker<Qp: IbvQueuePair> {
    state: Option<QueuePairState<Qp>>,
    completions: CompletionInbox,
    completion_route: Option<CompletionRoute>,
    process_ops: mpsc::UnboundedReceiver<ProcessOps>,
    _submission_sender: mpsc::UnboundedSender<ProcessOps>,
    cq_poller: ActorHandle<CompletionQueueActor<IbvCq>>,
    detach_target: Option<(CqId, u32)>,
    cq_lease: Option<CqLease>,
}

/// Supervises one per-peer queue pair's data-plane task.
///
/// The actor performs the connection handshake and integrates the worker with
/// the supervision tree. After initialization, [`QueuePairWorker`] exclusively
/// owns the QP and its mutable bookkeeping and receives submissions and
/// completions through separate local channels.
#[derive(Debug)]
pub(super) struct QueuePairActor<M: Manager, Qp: IbvQueuePair> {
    /// Filled into [`CreatePeerQueuePair::sender`] so the peer can
    /// build its own [`QpKey`] from our identity.
    local_manager: ActorRef<M>,
    /// Recipient of [`CreatePeerQueuePair`].
    peer_manager: ActorRef<M>,
    worker: Option<QueuePairWorker<Qp>>,
    /// `true` when the peer QP is colocated with this actor's QP —
    /// i.e. both endpoints live in the same `IbvManagerActor` *and*
    /// target the same RDMA device. In that case `init` connects
    /// the QP to its own endpoint and skips the cross-actor
    /// handshake.
    is_loopback: bool,
    init_timeout: Duration,
    cancellation: CancellationToken,
    task: Option<tokio::task::JoinHandle<()>>,
    #[cfg(test)]
    completion_sender: mpsc::UnboundedSender<CompletionResult>,
}

impl<M: Manager, Qp: IbvQueuePair> QueuePairActor<M, Qp> {
    fn worker_mut(&mut self) -> &mut QueuePairWorker<Qp> {
        self.worker
            .as_mut()
            .expect("the queue pair worker has not started")
    }

    pub(super) fn new(
        qp_key: QpKey,
        local_manager: ActorRef<M>,
        peer_manager: ActorRef<M>,
        qp: Qp,
        cq_poller: ActorHandle<CompletionQueueActor<IbvCq>>,
        cq_lease: CqLease,
        is_loopback: bool,
        max_send_wr: u32,
    ) -> (Self, mpsc::UnboundedSender<ProcessOps>) {
        let init_timeout = hyperactor_config::global::get(crate::config::RDMA_QP_INIT_TIMEOUT);
        let posted_wrs = Arc::new(AtomicU64::new(0));
        let (completions, completion_route) = CompletionInbox::new();
        let (sender, process_ops) = mpsc::unbounded_channel();
        #[cfg(test)]
        let completion_sender = completion_route.sender_for_test();
        let state = QueuePairState::new(qp_key, qp, max_send_wr, posted_wrs);
        (
            Self {
                worker: Some(QueuePairWorker {
                    state: Some(state),
                    completions,
                    completion_route: Some(completion_route),
                    process_ops,
                    _submission_sender: sender.clone(),
                    cq_poller,
                    detach_target: None,
                    cq_lease: Some(cq_lease),
                }),
                local_manager,
                peer_manager,
                is_loopback,
                init_timeout,
                cancellation: CancellationToken::new(),
                task: None,
                #[cfg(test)]
                completion_sender,
            },
            sender,
        )
    }
}

impl<M: Manager, Qp: IbvQueuePair> Drop for QueuePairActor<M, Qp> {
    fn drop(&mut self) {
        self.cancellation.cancel();
    }
}

impl<Qp: IbvQueuePair> QueuePairState<Qp> {
    fn new(qp_key: QpKey, qp: Qp, max_send_wr: u32, posted_wrs: Arc<AtomicU64>) -> Self {
        Self {
            qp_key,
            qp,
            max_send_wr,
            queue: VecDeque::new(),
            posted: HashMap::new(),
            wr_to_op: HashMap::new(),
            next_op_id: 0,
            in_flight: 0,
            posted_wrs,
            poller_wake: None,
        }
    }

    fn enqueue(&mut self, message: ProcessOps) {
        let max_msg_size = self.qp.max_msg_size();
        self.queue
            .extend(message.items.into_iter().map(|op| PendingOp {
                wrs: op.local.size.div_ceil(max_msg_size).max(1) as u32,
                op,
                reply: message.reply.clone(),
            }));
    }

    /// Try to post the head of `queue`.
    ///
    /// * `Ok(true)` — head was either posted or rejected with a
    ///   per-op error (e.g. op too large for this QP); caller
    ///   should attempt the next head.
    /// * `Ok(false)` — head can't be posted because the QP is at
    ///   `max_send_wr`. The op stays at the head; caller should stop
    ///   walking.
    /// * `Err(_)` — `qp.put`/`qp.get` failed (e.g. the QP is in
    ///   error state). Fatal: the worker reports this to its supervising actor.
    fn try_post_head(&mut self) -> Result<bool, anyhow::Error> {
        let pending = self.queue.pop_front().expect("non-empty queue");
        let PendingOp { op, reply, wrs } = pending;
        let stripe_id = op.stripe_id;

        // 1. Per-op fatal: op alone exceeds the QP's capacity.
        if wrs > self.max_send_wr {
            let err = format!(
                "op too large for this QP [{stripe_id}, qp_key={:?}, op_type={:?}, wrs={}, max_send_wr={}, local: {:?}, remote: {:?}]",
                self.qp_key, op.op_type, wrs, self.max_send_wr, op.local_memory, op.remote,
            );
            let _ = reply.send(StripeResult {
                stripe_id,
                result: Err(err),
            });
            return Ok(true);
        }

        // 2. Credit gating. Every op consumes send-queue slots.
        //    A head op that does not fit stalls the queue behind
        //    it until enough WRs are reaped.
        if self.in_flight + wrs > self.max_send_wr {
            self.queue.push_front(PendingOp { op, reply, wrs });
            return Ok(false);
        }

        // 3. Post.
        let local = op.local;
        let post_result = match op.op_type {
            RdmaOpType::WriteFromLocal => self.qp.put(op.remote.clone(), local.clone()),
            RdmaOpType::ReadIntoLocal => self.qp.get(local.clone(), op.remote.clone()),
        };
        let wr_ids = post_result.map_err(|e| {
            anyhow::anyhow!(
                "qp.{} failed [{stripe_id}, qp_key={:?}, local: {:?}, remote: {:?}]: {e}",
                if matches!(op.op_type, RdmaOpType::ReadIntoLocal) {
                    "get"
                } else {
                    "put"
                },
                self.qp_key,
                local,
                op.remote,
            )
        })?;

        // 4. Track in-flight state.
        let op_id = self.next_op_id;
        self.next_op_id += 1;
        let mut pending_wrs = HashSet::with_capacity(wr_ids.len());
        for id in &wr_ids {
            self.wr_to_op.insert(*id, op_id);
            pending_wrs.insert(*id);
        }
        self.in_flight += wr_ids.len() as u32;
        self.posted.insert(
            op_id,
            PostedOpEntry {
                stripe_id,
                pending_wrs,
                _local_memory: op.local_memory,
                _local_mrv: local,
                reply,
                first_error: None,
            },
        );
        // This state is the counter's only writer. The following `unpark`
        // publishes the increment before a parked poller checks the count.
        self.posted_wrs
            .fetch_add(wr_ids.len() as u64, Ordering::Relaxed);
        self.poller_wake
            .as_ref()
            .expect("the queue pair attached before posting work")
            .notify()
            .map_err(|error| {
                anyhow::anyhow!(
                    "could not wake the completion queue poller [qp_key={:?}]: {error}",
                    self.qp_key,
                )
            })?;
        Ok(true)
    }

    fn post_ready(&mut self) -> Result<(), anyhow::Error> {
        while !self.queue.is_empty() {
            if !self.try_post_head()? {
                break;
            }
        }
        Ok(())
    }

    fn complete_wr(&mut self, completion: CompletionResult) -> Result<(), anyhow::Error> {
        let (wr_id, wr_error) = match completion {
            Ok(wc) => (wc.wr_id(), None),
            Err(per_wr) => (per_wr.wr_id, Some(per_wr.to_string())),
        };
        let op_id = self
            .wr_to_op
            .remove(&wr_id)
            .expect("completed wr_id missing from wr_to_op");
        let entry = self
            .posted
            .get_mut(&op_id)
            .expect("op_id missing from posted");
        entry.pending_wrs.remove(&wr_id);
        self.in_flight -= 1;
        if let Some(err) = wr_error
            && entry.first_error.is_none()
        {
            entry.first_error = Some(err);
        }
        if entry.pending_wrs.is_empty() {
            let entry = self.posted.remove(&op_id).expect("just verified");
            let result = match entry.first_error {
                Some(err) => Err(err),
                None => Ok(()),
            };
            let _ = entry.reply.send(StripeResult {
                stripe_id: entry.stripe_id,
                result,
            });
        }
        Ok(())
    }

    fn into_detached(self) -> DetachedQueuePair {
        DetachedQueuePair::new(DetachedQueuePairResources {
            _qp: self.qp,
            // If we dropped these now, the user would get an error immediately
            // even though the operations are still being processed and it isn't
            // safe for them to access the relevant memory yet.
            _posted: self.posted,
        })
    }
}

impl<Qp: IbvQueuePair> QueuePairWorker<Qp> {
    fn state(&self) -> &QueuePairState<Qp> {
        self.state
            .as_ref()
            .expect("the queue pair has not been detached")
    }

    fn state_mut(&mut self) -> &mut QueuePairState<Qp> {
        self.state
            .as_mut()
            .expect("the queue pair has not been detached")
    }

    async fn run(&mut self, cancellation: CancellationToken) -> Result<(), anyhow::Error> {
        if cancellation.is_cancelled() {
            return Ok(());
        }
        let qp_num = self.state_mut().qp.get_qp_info()?.qp_num;
        let cq = Arc::clone(
            self.cq_lease
                .as_ref()
                .expect("the queue pair worker starts with a CQ lease")
                .cq(),
        );
        let cq_id = cq.cq_id();
        let route = self
            .completion_route
            .take()
            .expect("the queue pair attaches its completion route once");
        let posted = Arc::clone(&self.state().posted_wrs);
        let (reply, installed) = tokio::sync::oneshot::channel();
        self.cq_poller.try_post(
            Instance::<()>::self_client(),
            Attach {
                cq,
                qp_num,
                route,
                posted,
                reply,
            },
        )?;
        self.detach_target = Some((cq_id, qp_num));
        let poller_wake = tokio::select! {
            biased;

            _ = cancellation.cancelled() => return Ok(()),
            result = installed => result.map_err(|error| {
                anyhow::anyhow!("CQ poller did not attach queue pair: {error}")
            })?,
        };
        self.state_mut().poller_wake = Some(poller_wake);

        loop {
            tokio::select! {
                biased;

                _ = cancellation.cancelled() => return Ok(()),
                // A closed completion channel is only a QP failure while work
                // is outstanding. An idle QP need not observe CQ teardown.
                completion = self.completions.recv(), if self.state().in_flight > 0 => {
                    let completion = completion.ok_or_else(|| {
                        anyhow::anyhow!(
                            "completion queue poller stopped [qp_key={:?}]",
                            self.state().qp_key,
                        )
                    })?;
                    self.state_mut().complete_wr(completion)?;
                    self.state_mut().post_ready()?;
                }
                message = self.process_ops.recv() => {
                    let message = message.expect(
                        "the queue pair worker retains a submission sender while it runs",
                    );
                    self.state_mut().enqueue(message);
                    self.state_mut().post_ready()?;
                }
            }
        }
    }
}

impl<Qp: IbvQueuePair> Drop for QueuePairWorker<Qp> {
    fn drop(&mut self) {
        let Some((cq_id, qp_num)) = self.detach_target else {
            return;
        };
        let state = self
            .state
            .take()
            .expect("a queue pair with a detach target has state");
        let lease = self
            .cq_lease
            .take()
            .expect("a queue pair with a detach target has a CQ lease");
        let qp_key = state.qp_key.clone();
        if let Err(error) = self.cq_poller.try_post(
            Instance::<()>::self_client(),
            Detach {
                cq_id,
                qp_num,
                qp: state.into_detached(),
                lease,
            },
        ) {
            tracing::warn!(
                ?qp_key,
                qp_num,
                ?cq_id,
                %error,
                "handing the queue pair to its poller failed",
            );
        }
    }
}

async fn run_queue_pair_worker<Qp: IbvQueuePair>(
    mut worker: QueuePairWorker<Qp>,
    cancellation: CancellationToken,
    failed: PortHandle<QueuePairWorkerFailed>,
) {
    let result = AssertUnwindSafe(worker.run(cancellation))
        .catch_unwind()
        .await;
    let error = match result {
        Ok(Ok(())) => return,
        Ok(Err(error)) => error.to_string(),
        Err(_) => "queue pair worker panicked".to_owned(),
    };
    let error_for_log = error.clone();
    if let Err(post_error) = failed.try_post(
        Instance::<()>::self_client(),
        QueuePairWorkerFailed { error },
    ) {
        tracing::error!(
            error = %error_for_log,
            %post_error,
            "reporting queue pair worker failure failed",
        );
    }
}

#[async_trait]
impl<M: Manager, Qp: IbvQueuePair> Actor for QueuePairActor<M, Qp> {
    async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
        this.set_system();
        let qp_key = self.worker_mut().state().qp_key.clone();
        let local_info = self
            .worker_mut()
            .state_mut()
            .qp
            .get_qp_info()
            .map_err(|e| {
                tracing::error!(?qp_key, error = %e, "QueuePairActor init: get_qp_info failed");
                anyhow::anyhow!("could not extract local QP info: {e}")
            })?;

        let peer_info = if self.is_loopback {
            // The "peer" is ourselves; skip the round-trip and
            // connect to our own endpoint.
            local_info.clone()
        } else {
            let (reply, rx) = this.mailbox().open_once_port::<Result<IbvQpInfo, String>>();
            self.peer_manager.post(
                this,
                CreatePeerQueuePair {
                    sender: self.local_manager.clone(),
                    sender_device: qp_key.self_device.clone(),
                    receiver_device: qp_key.other_device.clone(),
                    qp_index: qp_key.qp_index,
                    sender_info: local_info,
                    reply: reply.bind(),
                },
            );
            match tokio::time::timeout(self.init_timeout, rx.recv()).await {
                Ok(Ok(Ok(info))) => info,
                Ok(Ok(Err(e))) => {
                    tracing::error!(
                        ?qp_key,
                        peer_manager = ?self.peer_manager,
                        error = %e,
                        "QueuePairActor init: peer manager rejected CreatePeerQueuePair",
                    );
                    return Err(anyhow::anyhow!("peer manager rejected QP request: {e}"));
                }
                Ok(Err(e)) => {
                    tracing::error!(
                        ?qp_key,
                        peer_manager = ?self.peer_manager,
                        error = %e,
                        "QueuePairActor init: peer reply port closed",
                    );
                    return Err(anyhow::anyhow!("peer reply port closed: {e}"));
                }
                Err(_) => {
                    tracing::error!(
                        ?qp_key,
                        peer_manager = ?self.peer_manager,
                        timeout = ?self.init_timeout,
                        "QueuePairActor init: timed out waiting for peer reply",
                    );
                    return Err(anyhow::anyhow!(
                        "QP initialization timed out after {:?}",
                        self.init_timeout
                    ));
                }
            }
        };

        self.worker_mut()
            .state_mut()
            .qp
            .connect(&peer_info)
            .map_err(|e| {
                tracing::error!(
                    ?qp_key,
                    peer_info = ?peer_info,
                    error = %e,
                    "QueuePairActor init: connect failed",
                );
                anyhow::anyhow!("could not connect QP to peer: {e}")
            })?;

        let worker = self
            .worker
            .take()
            .expect("the queue pair worker starts only once");
        let failed = this.handle().port::<QueuePairWorkerFailed>();
        self.task = Some(crate::rdma_runtime::spawn_on_rdma_runtime(
            run_queue_pair_worker(worker, self.cancellation.clone(), failed),
        ));
        Ok(())
    }

    async fn cleanup(
        &mut self,
        _this: &Instance<Self>,
        _err: Option<&ActorError>,
    ) -> Result<(), anyhow::Error> {
        self.cancellation.cancel();
        let Some(task) = self.task.take() else {
            return Ok(());
        };
        task.await
            .map_err(|error| anyhow::anyhow!("joining queue pair worker failed: {error}"))
    }

    // This actor is implemented in Rust, but the RDMA registration path may enter
    // Python and take the GIL. Run its loop on the dedicated rdma runtime rather
    // than the shared control-plane runtime; see `crate::rdma_runtime`.
    fn spawn_server_task<F>(future: F) -> tokio::task::JoinHandle<F::Output>
    where
        F: std::future::Future + Send + 'static,
        F::Output: Send + 'static,
    {
        crate::rdma_runtime::spawn_on_rdma_runtime(future)
    }
}

#[async_trait]
impl<M: Manager, Qp: IbvQueuePair> Handler<QueuePairWorkerFailed> for QueuePairActor<M, Qp> {
    async fn handle(
        &mut self,
        _cx: &Context<Self>,
        message: QueuePairWorkerFailed,
    ) -> Result<(), anyhow::Error> {
        Err(anyhow::anyhow!(
            "queue pair worker failed: {}",
            message.error
        ))
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::Mutex;
    use std::sync::atomic::AtomicBool;
    use std::sync::atomic::Ordering;
    use std::time::Duration;

    use anyhow::Result;
    use async_trait::async_trait;
    use hyperactor::ActorHandle;
    use hyperactor::Context;
    use hyperactor::Handler;
    use hyperactor::proc::Proc;

    use super::super::memory_region::IbvMemoryRegionKeepalive;
    use super::*;
    use crate::backend::ibverbs::device::IbvDevice;
    use crate::backend::ibverbs::device::list_all_devices;
    use crate::backend::ibverbs::device_selection::IbvDeviceTarget;
    use crate::backend::ibverbs::device_selection::resolve_target;
    use crate::backend::ibverbs::mlx_device::MlxDevice;
    use crate::backend::ibverbs::primitives::IbvConfig;

    #[test]
    fn test_create_connection() {
        if list_all_devices().is_empty() {
            println!("Skipping test: RDMA devices not available");
            return;
        }

        let config = IbvConfig {
            use_gpu_direct: false,
            ..Default::default()
        };
        let device_info = resolve_target::<MlxDevice>(&IbvDeviceTarget::cpu(0))
            .expect("resolving cpu:0 should succeed")
            .expect("cpu:0 should resolve to a NIC");
        let mut device = IbvDevice::<MlxDevice>::try_open(device_info.name(), config.clone())
            .expect("resolved device should open");
        let domain = device
            .get_or_create_domain("test")
            .expect("domain creation should succeed");
        let queue_pair = legacy::IbvQueuePair::new(domain, config.clone());
        assert!(queue_pair.is_ok());
    }

    #[test]
    fn test_loopback_connection() {
        if list_all_devices().is_empty() {
            println!("Skipping test: RDMA devices not available");
            return;
        }

        let server_config = IbvConfig {
            use_gpu_direct: false,
            ..Default::default()
        };
        let client_config = IbvConfig {
            use_gpu_direct: false,
            ..Default::default()
        };

        let server_info = resolve_target::<MlxDevice>(&IbvDeviceTarget::cpu(0))
            .expect("resolving cpu:0 should succeed")
            .expect("cpu:0 should resolve to a NIC");
        let mut server_device =
            IbvDevice::<MlxDevice>::try_open(server_info.name(), server_config.clone())
                .expect("server device should open");
        let server_domain = server_device
            .get_or_create_domain("test")
            .expect("server domain creation should succeed");
        let client_info = resolve_target::<MlxDevice>(&IbvDeviceTarget::cpu(0))
            .expect("resolving cpu:0 should succeed")
            .expect("cpu:0 should resolve to a NIC");
        let mut client_device =
            IbvDevice::<MlxDevice>::try_open(client_info.name(), client_config.clone())
                .expect("client device should open");
        let client_domain = client_device
            .get_or_create_domain("test")
            .expect("client domain creation should succeed");

        let mut server_qp =
            legacy::IbvQueuePair::new(server_domain, server_config.clone()).unwrap();
        let mut client_qp =
            legacy::IbvQueuePair::new(client_domain, client_config.clone()).unwrap();

        let server_connection_info = server_qp.get_qp_info().unwrap();
        let client_connection_info = client_qp.get_qp_info().unwrap();

        assert!(server_qp.connect(&client_connection_info).is_ok());
        assert!(client_qp.connect(&server_connection_info).is_ok());
    }

    // =================================================================
    // QueuePairActor init handshake
    // =================================================================

    /// Captured fields from a `CreatePeerQueuePair` message; we
    /// can't keep the original because the `reply` port is consumed
    /// to send the response.
    #[derive(Debug, Clone)]
    struct CreateCapture {
        sender_id: hyperactor::ActorId,
        sender_device: String,
        receiver_device: String,
        qp_index: usize,
        sender_qp_num: u32,
    }

    #[derive(Debug)]
    struct QpaMockState {
        creates: Vec<CreateCapture>,
        response: Option<Result<IbvQpInfo, String>>,
        /// Forwards every supervision event the parent receives to
        /// the test.
        supervision_tx: Option<
            tokio::sync::mpsc::UnboundedSender<hyperactor::supervision::ActorSupervisionEvent>,
        >,
    }

    /// Mock manager used by `QueuePairActor` tests.
    ///
    /// Plays two roles:
    /// 1. As the *parent*, it spawns `QueuePairActor` children via
    ///    [`SpawnQpaChild`].
    /// 2. As the *peer*, it handles `CreatePeerQueuePair`.
    #[derive(Debug)]
    #[hyperactor::export(handlers = [CreatePeerQueuePair<QpaMockManager>])]
    struct QpaMockManager {
        state: Arc<Mutex<QpaMockState>>,
    }

    #[async_trait]
    impl Actor for QpaMockManager {
        async fn handle_supervision_event(
            &mut self,
            _this: &Instance<Self>,
            event: &hyperactor::supervision::ActorSupervisionEvent,
        ) -> Result<bool> {
            let tx = self.state.lock().unwrap().supervision_tx.clone();
            let Some(tx) = tx else {
                return Ok(!event.is_error());
            };
            tx.send(event.clone())
                .map_err(|e| anyhow::anyhow!("supervision_tx send failed: {e}"))?;
            Ok(true)
        }
    }

    #[async_trait]
    impl Handler<CreatePeerQueuePair<QpaMockManager>> for QpaMockManager {
        async fn handle(
            &mut self,
            cx: &Context<Self>,
            msg: CreatePeerQueuePair<QpaMockManager>,
        ) -> Result<()> {
            let response = {
                let mut state = self.state.lock().unwrap();
                state.creates.push(CreateCapture {
                    sender_id: msg.sender.actor_addr().id().clone(),
                    sender_device: msg.sender_device.clone(),
                    receiver_device: msg.receiver_device.clone(),
                    qp_index: msg.qp_index,
                    sender_qp_num: msg.sender_info.qp_num,
                });
                state.response.take()
            };
            if let Some(response) = response {
                msg.reply.post(cx, response);
            }
            // None → drop reply intentionally to test the timeout path.
            Ok(())
        }
    }

    /// Local message that spawns a `QueuePairActor` as a child of
    /// this manager so supervision events route here. The reply
    /// carries the resulting handle so the test can observe
    /// lifecycle transitions.
    #[derive(Debug)]
    struct SpawnQpaChild {
        qp_key: QpKey,
        peer_manager: ActorRef<QpaMockManager>,
        qp: MockQp,
        is_loopback: bool,
        max_send_wr: u32,
        reply: hyperactor::OncePortHandle<QueuePairHandle<QpaMockManager, MockQp>>,
    }

    #[async_trait]
    impl Handler<SpawnQpaChild> for QpaMockManager {
        async fn handle(&mut self, cx: &Context<Self>, msg: SpawnQpaChild) -> Result<()> {
            let local_manager = cx.bind::<QpaMockManager>();
            let cq_poller = cx.spawn(CompletionQueueActor::<IbvCq>::new());
            let test_qp = msg.qp.clone();
            let (actor, sender) = QueuePairActor::new(
                msg.qp_key,
                local_manager,
                msg.peer_manager,
                msg.qp,
                cq_poller,
                CqLease::for_test(),
                msg.is_loopback,
                msg.max_send_wr,
            );
            test_qp.set_completion_sender(actor.completion_sender.clone());
            let handle = QueuePairHandle::new(cx.spawn(actor), sender);
            msg.reply.try_post(cx, handle)?;
            Ok(())
        }
    }

    /// One `put` or `get` call recorded by the mock, surfaced to the
    /// test via the `posted_rx` channel returned alongside the QP.
    #[derive(Debug)]
    enum PostedOp {
        Put {
            remote_dst: IbvRemoteMemoryRegionView,
            local_src: IbvMemoryRegionView,
            wr_ids: Vec<u64>,
        },
        Get {
            local_dst: IbvMemoryRegionView,
            remote_src: IbvRemoteMemoryRegionView,
            wr_ids: Vec<u64>,
        },
    }

    #[derive(Debug)]
    struct MockQpInner {
        connect_calls: Vec<IbvQpInfo>,
        next_wr_id: u64,
        /// Every `put`/`get` call is forwarded here in order, so the
        /// test body can `await` rather than poll.
        posted_tx: tokio::sync::mpsc::UnboundedSender<PostedOp>,
        completion_sender: Option<mpsc::UnboundedSender<CompletionResult>>,
        /// One-shot error for the next `put` or `get` call.
        post_error: Option<String>,
    }

    /// `IbvQueuePair` mock used by `QueuePairActor` tests. Cloning is
    /// cheap (shared `Arc<Mutex<...>>`); the test typically holds
    /// one clone while handing another to the actor.
    #[derive(Debug, Clone)]
    struct MockQp {
        info: IbvQpInfo,
        inner: Arc<Mutex<MockQpInner>>,
    }

    impl MockQp {
        /// Returns the mock QP and the receiver that observes its
        /// posts. Tests that don't care about post events can drop
        /// the receiver.
        fn new(qp_num: u32, psn: u32) -> (Self, tokio::sync::mpsc::UnboundedReceiver<PostedOp>) {
            let (posted_tx, posted_rx) = tokio::sync::mpsc::unbounded_channel();
            let qp = Self {
                info: IbvQpInfo {
                    qp_num,
                    lid: 0,
                    gid: None,
                    psn,
                },
                inner: Arc::new(Mutex::new(MockQpInner {
                    connect_calls: Vec::new(),
                    next_wr_id: 0,
                    posted_tx,
                    completion_sender: None,
                    post_error: None,
                })),
            };
            (qp, posted_rx)
        }

        fn connect_calls(&self) -> Vec<IbvQpInfo> {
            self.inner.lock().unwrap().connect_calls.clone()
        }

        fn set_completion_sender(&self, sender: mpsc::UnboundedSender<CompletionResult>) {
            self.inner.lock().unwrap().completion_sender = Some(sender);
        }

        fn queue_completion(&self, wr_id: u64) {
            self.queue_completion_result(Ok(IbvWc::for_test(wr_id, true)));
        }

        fn queue_per_wr_error(&self, wr_id: u64, message: &str) {
            self.queue_completion_result(Err(WorkRequestError::for_test(wr_id, message)));
        }

        fn queue_completion_result(&self, result: CompletionResult) {
            self.inner
                .lock()
                .unwrap()
                .completion_sender
                .as_ref()
                .expect("the mock queue pair is attached")
                .send(result)
                .expect("the completion inbox should remain open");
        }

        /// Make the next `put`/`get` return this error (one-shot).
        fn queue_post_error(&self, message: &str) {
            self.inner.lock().unwrap().post_error = Some(message.to_string());
        }
    }

    impl IbvQueuePair for MockQp {
        unsafe fn new<I: IbvDomainImpl<QueuePair = Self>>(
            _domain: &IbvDomain<I>,
            _config: IbvConfig,
            _send_cq: Arc<IbvCq>,
            _recv_cq: Arc<IbvCq>,
        ) -> Result<Self> {
            // No `IbvDomainImpl` sets `Q = MockQp`, so this is never reached;
            // the mock is built directly via `MockQp::new`.
            unreachable!("MockQp is constructed directly, not from a domain")
        }

        fn connect(&mut self, info: &IbvQpInfo) -> Result<()> {
            self.inner.lock().unwrap().connect_calls.push(info.clone());
            Ok(())
        }

        fn get_qp_info(&mut self) -> Result<IbvQpInfo> {
            Ok(self.info.clone())
        }

        fn state(&mut self) -> Result<u32> {
            Ok(rdmaxcel_sys::ibv_qp_state::IBV_QPS_RTS)
        }

        fn put(
            &mut self,
            remote_dst: IbvRemoteMemoryRegionView,
            local_src: IbvMemoryRegionView,
        ) -> Result<Vec<u64>> {
            let mut inner = self.inner.lock().unwrap();
            if let Some(msg) = inner.post_error.take() {
                return Err(anyhow::anyhow!(msg));
            }
            let wrs = local_src.size.div_ceil(MAX_RDMA_MSG_SIZE).max(1);
            let mut wr_ids = Vec::with_capacity(wrs);
            for _ in 0..wrs {
                wr_ids.push(inner.next_wr_id);
                inner.next_wr_id += 1;
            }
            let _ = inner.posted_tx.send(PostedOp::Put {
                remote_dst,
                local_src,
                wr_ids: wr_ids.clone(),
            });
            Ok(wr_ids)
        }

        fn get(
            &mut self,
            local_dst: IbvMemoryRegionView,
            remote_src: IbvRemoteMemoryRegionView,
        ) -> Result<Vec<u64>> {
            let mut inner = self.inner.lock().unwrap();
            if let Some(msg) = inner.post_error.take() {
                return Err(anyhow::anyhow!(msg));
            }
            let wrs = local_dst.size.div_ceil(MAX_RDMA_MSG_SIZE).max(1);
            let mut wr_ids = Vec::with_capacity(wrs);
            for _ in 0..wrs {
                wr_ids.push(inner.next_wr_id);
                inner.next_wr_id += 1;
            }
            let _ = inner.posted_tx.send(PostedOp::Get {
                local_dst,
                remote_src,
                wr_ids: wr_ids.clone(),
            });
            Ok(wr_ids)
        }
    }

    struct QpaHarness {
        proc: Proc,
        parent: ActorHandle<QpaMockManager>,
        peer: ActorHandle<QpaMockManager>,
        peer_state: Arc<Mutex<QpaMockState>>,
        client: hyperactor::Client,
        supervision_rx:
            tokio::sync::mpsc::UnboundedReceiver<hyperactor::supervision::ActorSupervisionEvent>,
    }

    impl QpaHarness {
        fn build() -> Result<Self> {
            let proc = Proc::anonymous();
            let (supervision_tx, supervision_rx) = tokio::sync::mpsc::unbounded_channel();
            let parent = proc.spawn_with_label(
                "parent",
                QpaMockManager {
                    state: Arc::new(Mutex::new(QpaMockState {
                        creates: Vec::new(),
                        response: None,
                        supervision_tx: Some(supervision_tx),
                    })),
                },
            );
            let peer_state = Arc::new(Mutex::new(QpaMockState {
                creates: Vec::new(),
                response: None,
                supervision_tx: None,
            }));
            let peer = proc.spawn_with_label(
                "peer",
                QpaMockManager {
                    state: Arc::clone(&peer_state),
                },
            );
            let client = proc.client("client");
            Ok(Self {
                proc,
                parent,
                peer,
                peer_state,
                client,
                supervision_rx,
            })
        }

        fn peer_id(&self) -> hyperactor::ActorId {
            self.peer.actor_addr().id().clone()
        }

        fn parent_id(&self) -> hyperactor::ActorId {
            self.parent.actor_addr().id().clone()
        }

        /// Await the next forwarded child-error supervision event.
        async fn next_supervision_event(
            &mut self,
        ) -> hyperactor::supervision::ActorSupervisionEvent {
            tokio::time::timeout(Duration::from_secs(5), self.supervision_rx.recv())
                .await
                .expect("timed out waiting for child supervision event")
                .expect("supervision channel closed")
        }

        /// Destroys the proc, closes the supervision channel, drains
        /// every remaining event, and asserts none are unexpected.
        async fn teardown(mut self) {
            self.supervision_rx.close();
            self.proc
                .destroy_and_wait(Duration::from_secs(30), "test teardown")
                .await
                .expect("destroy_and_wait failed");
            let mut leftover = Vec::new();
            while let Some(event) = self.supervision_rx.recv().await {
                leftover.push(event);
            }
            assert!(
                leftover.is_empty(),
                "unexpected supervision events at teardown: {leftover:?}",
            );
        }

        async fn spawn_actor(
            &self,
            qp_key: QpKey,
            peer_manager: ActorRef<QpaMockManager>,
            qp: MockQp,
            is_loopback: bool,
        ) -> Result<QueuePairHandle<QpaMockManager, MockQp>> {
            self.spawn_actor_with_caps(qp_key, peer_manager, qp, is_loopback, 4)
                .await
        }

        async fn spawn_actor_with_caps(
            &self,
            qp_key: QpKey,
            peer_manager: ActorRef<QpaMockManager>,
            qp: MockQp,
            is_loopback: bool,
            max_send_wr: u32,
        ) -> Result<QueuePairHandle<QpaMockManager, MockQp>> {
            let (reply, rx) = self.client.mailbox().open_once_port();
            self.parent.try_post(
                &self.client,
                SpawnQpaChild {
                    qp_key,
                    peer_manager,
                    qp,
                    is_loopback,
                    max_send_wr,
                    reply,
                },
            )?;
            Ok(rx.recv().await?)
        }
    }

    async fn await_status(
        handle: &QueuePairHandle<QpaMockManager, MockQp>,
        expected: impl Fn(&hyperactor::actor::ActorStatus) -> bool,
    ) -> hyperactor::actor::ActorStatus {
        let mut status = handle.actor().status();
        status.wait_for(|s| expected(s)).await.unwrap();
        status.borrow().clone()
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_init_succeeds_via_peer() -> Result<()> {
        let harness = QpaHarness::build()?;
        let peer_info = IbvQpInfo {
            qp_num: 0xbeef,
            lid: 0,
            gid: None,
            psn: 0xc0ffee,
        };
        harness.peer_state.lock().unwrap().response = Some(Ok(peer_info.clone()));

        let (qp, _posted_rx) = MockQp::new(0x1234, 0xdead);
        let qp_key = QpKey {
            self_device: "mlx5_0".into(),
            other_id: harness.peer_id(),
            other_device: "mlx5_1".into(),
            qp_index: 3,
        };
        let handle = harness
            .spawn_actor(
                qp_key.clone(),
                harness.peer.bind::<QpaMockManager>(),
                qp.clone(),
                false,
            )
            .await?;

        await_status(&handle, |s| {
            matches!(s, hyperactor::actor::ActorStatus::Idle)
        })
        .await;

        let creates = harness.peer_state.lock().unwrap().creates.clone();
        assert_eq!(creates.len(), 1);
        assert_eq!(creates[0].sender_device, "mlx5_0");
        assert_eq!(creates[0].receiver_device, "mlx5_1");
        assert_eq!(creates[0].qp_index, 3);
        assert_eq!(creates[0].sender_qp_num, 0x1234);
        // The sender ref carries the local manager's identity so the
        // receiver can build its own `QpKey`.
        assert_eq!(creates[0].sender_id, harness.parent_id());

        let connects = qp.connect_calls();
        assert_eq!(connects.len(), 1);
        assert_eq!(connects[0].qp_num, peer_info.qp_num);
        assert_eq!(connects[0].psn, peer_info.psn);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_init_loopback_skips_peer_round_trip() -> Result<()> {
        let harness = QpaHarness::build()?;

        let (qp, _posted_rx) = MockQp::new(0xaaa, 0xbbb);
        let qp_key = QpKey {
            self_device: "mlx5_0".into(),
            other_id: harness.parent_id(),
            other_device: "mlx5_0".into(),
            qp_index: 0,
        };
        let handle = harness
            .spawn_actor(
                qp_key,
                harness.peer.bind::<QpaMockManager>(),
                qp.clone(),
                true,
            )
            .await?;

        await_status(&handle, |s| {
            matches!(s, hyperactor::actor::ActorStatus::Idle)
        })
        .await;
        assert!(harness.peer_state.lock().unwrap().creates.is_empty());

        let connects = qp.connect_calls();
        assert_eq!(connects.len(), 1);
        assert_eq!(connects[0].qp_num, 0xaaa);
        assert_eq!(connects[0].psn, 0xbbb);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_init_peer_error_fails_actor() -> Result<()> {
        let mut harness = QpaHarness::build()?;
        harness.peer_state.lock().unwrap().response =
            Some(Err("peer rejected, no domain on receiver_device".into()));

        let qp_key = QpKey {
            self_device: "mlx5_0".into(),
            other_id: harness.peer_id(),
            other_device: "mlx5_99".into(),
            qp_index: 0,
        };
        let (qp, _posted_rx) = MockQp::new(1, 2);
        let handle = harness
            .spawn_actor(qp_key, harness.peer.bind::<QpaMockManager>(), qp, false)
            .await?;

        let event = harness.next_supervision_event().await;
        assert_eq!(&event.actor_id, handle.actor().actor_addr());
        let report = event.failure_report().expect("event should be a failure");
        assert!(
            report.contains("peer rejected"),
            "failure report should surface the peer's error string; got: {report}"
        );
        await_status(&handle, |s| {
            matches!(s, hyperactor::actor::ActorStatus::Failed(_))
        })
        .await;
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_init_timeout_fails_actor() -> Result<()> {
        let lock = hyperactor_config::global::lock();
        let _guard = lock.override_key(
            crate::config::RDMA_QP_INIT_TIMEOUT,
            Duration::from_millis(100),
        );

        let mut harness = QpaHarness::build()?;

        let qp_key = QpKey {
            self_device: "mlx5_0".into(),
            other_id: harness.peer_id(),
            other_device: "mlx5_1".into(),
            qp_index: 0,
        };
        let (qp, _posted_rx) = MockQp::new(1, 2);
        let handle = harness
            .spawn_actor(qp_key, harness.peer.bind::<QpaMockManager>(), qp, false)
            .await?;

        let event = harness.next_supervision_event().await;
        assert_eq!(&event.actor_id, handle.actor().actor_addr());
        let report = event.failure_report().expect("event should be a failure");
        assert!(
            report.contains("timed out"),
            "failure report should mention timeout; got: {report}"
        );
        await_status(&handle, |s| {
            matches!(s, hyperactor::actor::ActorStatus::Failed(_))
        })
        .await;
        harness.teardown().await;
        Ok(())
    }

    // =================================================================
    // QueuePairActor op processing
    // =================================================================

    use crate::backend::ibverbs::primitives::IbvMr;
    use crate::local_memory::Keepalive;
    use crate::local_memory::KeepaliveLocalMemory;

    /// No-op [`Keepalive`] for tests that mint a [`KeepaliveLocalMemory`]
    /// from a fake address never actually read or written through.
    struct FakeKeepalive {
        addr: usize,
        size: usize,
    }
    impl Keepalive for FakeKeepalive {
        fn addr(&self) -> usize {
            self.addr
        }
        fn size(&self) -> usize {
            self.size
        }
    }

    /// The allocation an op's [`QueuePairOp::local`] view describes.
    fn fake_local_memory(addr: usize, size: usize) -> KeepaliveLocalMemory {
        KeepaliveLocalMemory::try_new(Arc::new(FakeKeepalive { addr, size }))
            .expect("a fake host address has a location")
    }

    /// [`Keepalive`] that sets `dropped` when it goes away, so a test can
    /// tell whether anything still pins the allocation.
    struct DropFlagKeepalive {
        addr: usize,
        size: usize,
        dropped: Arc<AtomicBool>,
    }
    impl Keepalive for DropFlagKeepalive {
        fn addr(&self) -> usize {
            self.addr
        }
        fn size(&self) -> usize {
            self.size
        }
    }
    impl Drop for DropFlagKeepalive {
        fn drop(&mut self) {
            self.dropped.store(true, Ordering::SeqCst);
        }
    }

    fn fake_mrv(addr: usize, size: usize) -> IbvMemoryRegionView {
        IbvMemoryRegionView::new(
            addr,
            addr,
            size,
            0x1234,
            0x5678,
            SELF_DEVICE.to_string(),
            // A null MR keepalive: its `Drop` is a no-op.
            Arc::new(IbvMr::null()),
        )
    }

    /// Devices of the QP `spawn_ready_actor` builds.
    const SELF_DEVICE: &str = "mlx5_0";
    const PEER_DEVICE: &str = "mlx5_0";

    fn make_op(op_idx: usize, op_type: RdmaOpType, addr: usize, size: usize) -> QueuePairOp {
        make_stripe(op_idx, 0, 1, op_type, addr, size, 0, size)
    }

    #[expect(clippy::too_many_arguments, reason = "a test fixture for one stripe")]
    fn make_stripe(
        op_idx: usize,
        stripe_idx: usize,
        stripe_count: usize,
        op_type: RdmaOpType,
        addr: usize,
        region_size: usize,
        offset: usize,
        size: usize,
    ) -> QueuePairOp {
        let local = fake_mrv(addr, region_size)
            .try_slice(offset, size)
            .expect("a valid local slice");
        let remote = IbvRemoteMemoryRegionView {
            rkey: 0,
            addr: 0x4000_0000,
            size: region_size,
            device_name: PEER_DEVICE.to_string(),
        }
        .try_slice(offset, size)
        .expect("a valid remote slice");
        QueuePairOp {
            stripe_id: StripeId {
                op_idx,
                stripe_idx,
                stripe_count,
            },
            qp_index: 0,
            op_type,
            local_memory: fake_local_memory(addr, region_size),
            local,
            remote,
        }
    }

    /// MR keepalive that sets `dropped` when it goes away, so a test can tell
    /// whether anything still holds the registration open.
    #[derive(Debug)]
    struct DropFlagMr {
        dropped: Arc<AtomicBool>,
    }
    impl IbvMemoryRegionKeepalive for DropFlagMr {}
    impl Drop for DropFlagMr {
        fn drop(&mut self) {
            self.dropped.store(true, Ordering::SeqCst);
        }
    }

    /// Like [`make_op`], but the registration reports when it drops.
    fn make_op_watching_mr_drop(
        op_idx: usize,
        op_type: RdmaOpType,
        addr: usize,
        size: usize,
        dropped: Arc<AtomicBool>,
    ) -> QueuePairOp {
        QueuePairOp {
            local: IbvMemoryRegionView::new(
                addr,
                addr,
                size,
                0x1234,
                0x5678,
                SELF_DEVICE.to_string(),
                Arc::new(DropFlagMr { dropped }),
            ),
            ..make_op(op_idx, op_type, addr, size)
        }
    }

    /// Like [`make_op`], but the local memory reports when it drops.
    fn make_op_watching_drop(
        op_idx: usize,
        op_type: RdmaOpType,
        addr: usize,
        size: usize,
        dropped: Arc<AtomicBool>,
    ) -> QueuePairOp {
        QueuePairOp {
            local_memory: KeepaliveLocalMemory::try_new(Arc::new(DropFlagKeepalive {
                addr,
                size,
                dropped,
            }))
            .expect("a fake host address has a location"),
            ..make_op(op_idx, op_type, addr, size)
        }
    }

    /// Await `flag`, which the actor sets from a `Drop` it runs just after
    /// posting the op's reply — so the reply can be observed a moment
    /// before the drop lands.
    async fn await_dropped(flag: &AtomicBool) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while !flag.load(Ordering::SeqCst) {
            assert!(
                tokio::time::Instant::now() < deadline,
                "the pin on the allocation was never released",
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }

    impl QpaHarness {
        /// Spawn a `QueuePairActor` in loopback mode (no peer round
        /// trip) so init completes immediately; await it reaching
        /// `Idle`. Returns the actor handle along with the mock QP
        /// the test can drive.
        async fn spawn_ready_actor(
            &self,
            max_send_wr: u32,
        ) -> Result<(
            QueuePairHandle<QpaMockManager, MockQp>,
            MockQp,
            tokio::sync::mpsc::UnboundedReceiver<PostedOp>,
        )> {
            let (qp, posted_rx) = MockQp::new(0x1, 0x2);
            let qp_key = QpKey {
                self_device: SELF_DEVICE.into(),
                other_id: self.parent_id(),
                other_device: PEER_DEVICE.into(),
                qp_index: 0,
            };
            let handle = self
                .spawn_actor_with_caps(
                    qp_key,
                    self.peer.bind::<QpaMockManager>(),
                    qp.clone(),
                    true,
                    max_send_wr,
                )
                .await?;
            await_status(&handle, |s| {
                matches!(s, hyperactor::actor::ActorStatus::Idle)
            })
            .await;
            Ok((handle, qp, posted_rx))
        }
    }

    /// Await the next `PostedOp` from the mock; panics on timeout.
    async fn recv_posted(rx: &mut tokio::sync::mpsc::UnboundedReceiver<PostedOp>) -> PostedOp {
        tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .expect("timed out waiting for post on MockQp")
            .expect("MockQp post channel closed")
    }

    /// Assert that no further `PostedOp` arrives in `wait`. Used to
    /// pin down a credit-blocked state.
    async fn assert_no_post(
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<PostedOp>,
        wait: Duration,
    ) {
        if let Ok(Some(p)) = tokio::time::timeout(wait, rx.recv()).await {
            panic!("unexpected post on MockQp: {p:?}");
        }
    }

    fn expect_put(p: PostedOp) -> (IbvMemoryRegionView, IbvRemoteMemoryRegionView, Vec<u64>) {
        match p {
            PostedOp::Put {
                remote_dst,
                local_src,
                wr_ids,
            } => (local_src, remote_dst, wr_ids),
            other => panic!("expected Put, got {other:?}"),
        }
    }

    fn expect_get(p: PostedOp) -> (IbvMemoryRegionView, IbvRemoteMemoryRegionView, Vec<u64>) {
        match p {
            PostedOp::Get {
                local_dst,
                remote_src,
                wr_ids,
            } => (local_dst, remote_src, wr_ids),
            other => panic!("expected Get, got {other:?}"),
        }
    }

    /// Open a result channel and send a `ProcessOps` batch. Returns
    /// the receiver so the caller can await per-op results.
    fn submit_ops(
        actor: &QueuePairHandle<QpaMockManager, MockQp>,
        items: Vec<QueuePairOp>,
    ) -> Result<mpsc::UnboundedReceiver<StripeResult>> {
        let (reply, rx) = mpsc::unbounded_channel();
        actor.send(ProcessOps { items, reply })?;
        Ok(rx)
    }

    /// Collect exactly `n` replies with a per-recv timeout, sorted
    /// by op_idx for deterministic comparison.
    async fn collect_replies(
        rx: &mut mpsc::UnboundedReceiver<StripeResult>,
        n: usize,
    ) -> Vec<(usize, Result<(), String>)> {
        let mut out = Vec::with_capacity(n);
        for _ in 0..n {
            let m = tokio::time::timeout(Duration::from_secs(5), rx.recv())
                .await
                .expect("timed out waiting for ProcessOps reply")
                .expect("ProcessOps result channel closed");
            out.push((m.stripe_id.op_idx, m.result));
        }
        out.sort_by_key(|(i, _)| *i);
        out
    }

    /// Try to recv with a short timeout, returning `None` on timeout
    /// so callers can assert "no reply yet".
    async fn try_recv(
        rx: &mut mpsc::UnboundedReceiver<StripeResult>,
        wait: Duration,
    ) -> Option<(usize, Result<(), String>)> {
        match tokio::time::timeout(wait, rx.recv()).await {
            Ok(Some(m)) => Some((m.stripe_id.op_idx, m.result)),
            Ok(None) => panic!("ProcessOps result channel closed"),
            Err(_) => None,
        }
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_processes_single_write() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        let items = vec![make_op(7, RdmaOpType::WriteFromLocal, 0x1000, 4096)];
        let mut rx = submit_ops(&actor, items)?;

        // wr_ids start at 0 (fresh MockQp), so the single WR is wr 0.
        let (lhandle, rhandle, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(wr_ids, vec![0]);
        assert_eq!(lhandle.rdma_addr, 0x1000);
        assert_eq!(lhandle.size, 4096);
        assert_eq!(lhandle.lkey, 0x1234);
        assert_eq!(lhandle.rkey, 0x5678);
        assert_eq!(lhandle.device_name, SELF_DEVICE);
        assert_eq!(rhandle.addr, 0x4000_0000);
        assert_eq!(rhandle.size, 4096);
        assert_eq!(rhandle.device_name, PEER_DEVICE);

        qp.queue_completion(0);
        let replies = collect_replies(&mut rx, 1).await;
        assert_eq!(replies, vec![(7, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_posts_only_its_stripe() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        let items = vec![make_stripe(
            7,
            2,
            4,
            RdmaOpType::WriteFromLocal,
            0x1000,
            4096,
            2048,
            1024,
        )];
        let mut rx = submit_ops(&actor, items)?;

        let (local, remote, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!((local.rdma_addr, local.size), (0x1800, 1024));
        assert_eq!((remote.addr, remote.size), (0x4000_0800, 1024));

        qp.queue_completion(wr_ids[0]);
        let reply = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .expect("timed out waiting for the stripe reply")
            .expect("reply channel closed");
        assert_eq!(
            reply.stripe_id,
            StripeId {
                op_idx: 7,
                stripe_idx: 2,
                stripe_count: 4,
            }
        );
        assert_eq!(reply.result, Ok(()));
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_cleanup_joins_worker_and_closes_submission_channel() -> Result<()> {
        let mut harness = QpaHarness::build()?;
        let (actor, _qp, _posted_rx) = harness.spawn_ready_actor(4).await?;

        actor.drain_and_stop("test complete")?;
        await_status(&actor, |status| {
            matches!(status, hyperactor::actor::ActorStatus::Stopped(_))
        })
        .await;
        let event = harness.next_supervision_event().await;
        assert_eq!(&event.actor_id, actor.actor().actor_addr());
        assert!(matches!(
            event.actor_status,
            hyperactor::actor::ActorStatus::Stopped(_)
        ));

        let (reply, _rx) = mpsc::unbounded_channel();
        assert!(
            actor
                .send(ProcessOps {
                    items: Vec::new(),
                    reply,
                })
                .is_err(),
            "the submission receiver must be closed after actor cleanup",
        );
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_stays_alive_after_external_submission_handles_are_dropped() -> Result<()> {
        let mut harness = QpaHarness::build()?;
        let (actor, _qp, _posted_rx) = harness.spawn_ready_actor(4).await?;
        let actor_handle = actor.actor().clone();
        let mut status = actor_handle.status();

        drop(actor);
        assert!(
            tokio::time::timeout(
                Duration::from_secs(1),
                status.wait_for(hyperactor::actor::ActorStatus::is_terminal),
            )
            .await
            .is_err(),
            "dropping external submission handles must not stop the worker",
        );

        actor_handle.drain_and_stop("test complete")?;
        let event = harness.next_supervision_event().await;
        assert_eq!(&event.actor_id, actor_handle.actor_addr());
        assert!(matches!(
            event.actor_status,
            hyperactor::actor::ActorStatus::Stopped(_)
        ));
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_survives_a_dropped_operation_result_receiver() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(1).await?;
        let abandoned = submit_ops(
            &actor,
            vec![make_op(0, RdmaOpType::WriteFromLocal, 0x1000, 4096)],
        )?;
        drop(abandoned);
        let (_, _, first_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        qp.queue_completion(first_wrs[0]);

        let mut reply = submit_ops(
            &actor,
            vec![make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096)],
        )?;
        let (_, _, second_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        qp.queue_completion(second_wrs[0]);

        assert_eq!(collect_replies(&mut reply, 1).await, vec![(1, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_posted_op_pins_local_memory() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        // Two ops: the second exists only to prove the first has left
        // `try_post_head`, which is where the `QueuePairOp` — the batch's own
        // hold on the allocation — drops. Until then the assertion below
        // would pass whether or not the posted entry pins anything.
        let dropped = Arc::new(AtomicBool::new(false));
        let items = vec![
            make_op_watching_drop(
                0,
                RdmaOpType::WriteFromLocal,
                0x1000,
                4096,
                Arc::clone(&dropped),
            ),
            make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096),
        ];
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, other_wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert!(
            !dropped.load(Ordering::SeqCst),
            "the allocation must stay mapped while its WR is in flight: `submit` can \
             return before every WR retires, and the caller is then free to drop it",
        );

        // The pin lasts exactly as long as the op: retiring op 0 alone
        // releases it, which would not hold if the entry that pinned the
        // allocation were some other op's.
        qp.queue_completion(wr_ids[0]);
        await_dropped(&dropped).await;

        // Retire op 1 too, so no op is left in flight at teardown.
        qp.queue_completion(other_wr_ids[0]);
        assert_eq!(
            collect_replies(&mut rx, 2).await,
            vec![(0, Ok(())), (1, Ok(()))]
        );
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_posted_op_pins_local_registration() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        // As above, op 1 only proves op 0 has left `try_post_head`.
        let dropped = Arc::new(AtomicBool::new(false));
        let items = vec![
            make_op_watching_mr_drop(
                0,
                RdmaOpType::WriteFromLocal,
                0x1000,
                4096,
                Arc::clone(&dropped),
            ),
            make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096),
        ];
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, other_wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert!(
            !dropped.load(Ordering::SeqCst),
            "the registration must stay alive while a WR that touches it is in flight"
        );

        qp.queue_completion(wr_ids[0]);
        await_dropped(&dropped).await;

        qp.queue_completion(other_wr_ids[0]);
        assert_eq!(
            collect_replies(&mut rx, 2).await,
            vec![(0, Ok(())), (1, Ok(()))]
        );
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_chunked_op_waits_for_all_wrs() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(8).await?;

        // A 3-chunk write (3 * MAX_RDMA_MSG_SIZE) splits into 3 WRs;
        // the op's reply must be held back until all 3 complete.
        let items = vec![make_op(
            11,
            RdmaOpType::WriteFromLocal,
            0x1000,
            3 * MAX_RDMA_MSG_SIZE,
        )];
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(wr_ids, vec![0, 1, 2]);

        // Complete the first two — no reply yet.
        qp.queue_completion(wr_ids[0]);
        qp.queue_completion(wr_ids[1]);
        assert!(
            try_recv(&mut rx, Duration::from_millis(100))
                .await
                .is_none(),
            "op should not be reported until every WR has completed",
        );

        // Complete the third — now the op resolves.
        qp.queue_completion(wr_ids[2]);
        let replies = collect_replies(&mut rx, 1).await;
        assert_eq!(replies, vec![(11, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_per_wr_error_held_until_all_wrs_complete() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(8).await?;

        // 3-WR write: simulate the second WR failing first; the op's
        // Err must not fire until the other 2 WRs have also reported
        // so the MR registration and local memory outlive every in-flight WR.
        let items = vec![make_op(
            42,
            RdmaOpType::WriteFromLocal,
            0x1000,
            3 * MAX_RDMA_MSG_SIZE,
        )];
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, wr_ids) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(wr_ids.len(), 3);

        // Deliver the per-WR error first.
        qp.queue_per_wr_error(wr_ids[1], "simulated WR fail");
        assert!(
            try_recv(&mut rx, Duration::from_millis(100))
                .await
                .is_none(),
            "op error must wait for the other WRs to drain",
        );

        // The remaining WRs complete (flush-style or otherwise) and
        // finally the op's Err is reported.
        qp.queue_completion(wr_ids[0]);
        qp.queue_completion(wr_ids[2]);

        let replies = collect_replies(&mut rx, 1).await;
        assert_eq!(replies.len(), 1);
        assert_eq!(replies[0].0, 42);
        let err = replies[0]
            .1
            .as_ref()
            .expect_err("op should fail because one WR errored");
        assert!(
            err.contains("simulated WR fail"),
            "op error should surface the per-WR error: {err}",
        );
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_per_wr_error_isolates_to_failing_op() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(8).await?;

        // Two writes share a batch: op_idx 0 is multi-WR (3 chunks),
        // op_idx 1 is single-WR. One of op_idx 0's WRs fails;
        // op_idx 1's WR succeeds independently.
        let items = vec![
            make_op(0, RdmaOpType::WriteFromLocal, 0x1000, 3 * MAX_RDMA_MSG_SIZE),
            make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096),
        ];
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, big_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, small_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(big_wrs, vec![0, 1, 2]);
        assert_eq!(small_wrs, vec![3]);

        // op_idx 1 completes successfully on its own.
        qp.queue_completion(small_wrs[0]);
        // op_idx 0: middle WR fails, others succeed.
        qp.queue_per_wr_error(big_wrs[1], "isolated per-WR fail");
        qp.queue_completion(big_wrs[0]);
        qp.queue_completion(big_wrs[2]);

        let replies = collect_replies(&mut rx, 2).await;
        assert_eq!(replies.len(), 2);
        assert_eq!(replies[0].0, 0);
        let err = replies[0]
            .1
            .as_ref()
            .expect_err("op_idx 0 should fail because one WR errored");

        assert!(
            err.contains("isolated per-WR fail"),
            "op_idx 0 error should name the per-WR error: {err}",
        );
        assert_eq!(replies[1], (1usize, Ok(())));
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_reads_gate_on_send_slots_like_writes() -> Result<()> {
        let harness = QpaHarness::build()?;
        // Reads hold no credit of their own: three of them against
        // max_send_wr=2 gate on send-queue slots, just as writes would.
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(2).await?;

        let items = (0..3usize)
            .map(|i| make_op(i, RdmaOpType::ReadIntoLocal, 0x1000 + i * 0x1000, 4096))
            .collect();
        let mut rx = submit_ops(&actor, items)?;

        // Two reads fit the send-queue cap; the third parks.
        let (_, _, r0_wrs) = expect_get(recv_posted(&mut posted_rx).await);
        let (_, _, r1_wrs) = expect_get(recv_posted(&mut posted_rx).await);
        assert_eq!(r0_wrs, vec![0]);
        assert_eq!(r1_wrs, vec![1]);
        assert_no_post(&mut posted_rx, Duration::from_millis(50)).await;

        // Freeing one slot lets the third read post.
        qp.queue_completion(r0_wrs[0]);
        let (_, _, r2_wrs) = expect_get(recv_posted(&mut posted_rx).await);
        assert_eq!(r2_wrs, vec![2]);

        qp.queue_completion(r1_wrs[0]);
        qp.queue_completion(r2_wrs[0]);
        let replies = collect_replies(&mut rx, 3).await;
        assert_eq!(replies, vec![(0, Ok(())), (1, Ok(())), (2, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_write_slot_gating() -> Result<()> {
        let harness = QpaHarness::build()?;
        // max_send_wr=2 caps total in-flight WRs; submit 4 writes.
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(2).await?;

        let items = (0..4usize)
            .map(|i| make_op(i, RdmaOpType::WriteFromLocal, 0x1000 + i * 0x1000, 4096))
            .collect();
        let mut rx = submit_ops(&actor, items)?;

        let (_, _, w0_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, w1_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(w0_wrs, vec![0]);
        assert_eq!(w1_wrs, vec![1]);
        assert_no_post(&mut posted_rx, Duration::from_millis(50)).await;

        // Complete one — the third write should post; one slot still busy.
        qp.queue_completion(w0_wrs[0]);
        let (_, _, w2_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(w2_wrs, vec![2]);
        assert_no_post(&mut posted_rx, Duration::from_millis(50)).await;

        // Complete another — the fourth write posts.
        qp.queue_completion(w1_wrs[0]);
        let (_, _, w3_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(w3_wrs, vec![3]);

        qp.queue_completion(w2_wrs[0]);
        qp.queue_completion(w3_wrs[0]);
        let replies = collect_replies(&mut rx, 4).await;
        assert_eq!(
            replies,
            vec![(0, Ok(())), (1, Ok(())), (2, Ok(())), (3, Ok(()))],
        );
        harness.teardown().await;
        Ok(())
    }

    /// Reads and writes draw on one pool of send-queue slots, so a mixed
    /// batch parks the head op until enough WRs of either kind are reaped.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_reads_and_writes_share_send_slots() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        let items = vec![
            make_op(0, RdmaOpType::ReadIntoLocal, 0x1000, 4096),
            make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096),
            make_op(2, RdmaOpType::WriteFromLocal, 0x3000, 4096),
            make_op(3, RdmaOpType::WriteFromLocal, 0x4000, 4096),
            make_op(4, RdmaOpType::ReadIntoLocal, 0x5000, 2 * MAX_RDMA_MSG_SIZE),
        ];
        let mut rx = submit_ops(&actor, items)?;

        // The first four ops fill all four slots: 1 read WR + 3 write WRs.
        let (_, _, r0_wrs) = expect_get(recv_posted(&mut posted_rx).await);
        let (_, _, w1_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, w2_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        let (_, _, w3_wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(r0_wrs, vec![0]);
        assert_eq!(w1_wrs, vec![1]);
        assert_eq!(w2_wrs, vec![2]);
        assert_eq!(w3_wrs, vec![3]);
        // The 2-WR read at op_idx 4 needs two free slots, so it parks.
        assert_no_post(&mut posted_rx, Duration::from_millis(50)).await;

        // One completion frees one slot: still one short.
        qp.queue_completion(r0_wrs[0]);
        assert_eq!(collect_replies(&mut rx, 1).await, vec![(0, Ok(()))]);
        assert_no_post(&mut posted_rx, Duration::from_millis(50)).await;

        // The second completion -- a write this time -- frees the slot the
        // read was waiting for, so it posts.
        qp.queue_completion(w1_wrs[0]);
        assert_eq!(collect_replies(&mut rx, 1).await, vec![(1, Ok(()))]);
        let (_, _, r4_wrs) = expect_get(recv_posted(&mut posted_rx).await);
        assert_eq!(r4_wrs, vec![4, 5]);

        // Drain everything else.
        qp.queue_completion(w2_wrs[0]);
        qp.queue_completion(w3_wrs[0]);
        for &id in &r4_wrs {
            qp.queue_completion(id);
        }
        let rest = collect_replies(&mut rx, 3).await;
        assert_eq!(rest, vec![(2, Ok(())), (3, Ok(())), (4, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_multiple_batches_share_credit() -> Result<()> {
        let harness = QpaHarness::build()?;
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(4).await?;

        // Batch A: 2 writes with op_idx 10, 11.
        let batch_a = vec![
            make_op(10, RdmaOpType::WriteFromLocal, 0x1000, 4096),
            make_op(11, RdmaOpType::WriteFromLocal, 0x2000, 4096),
        ];
        let mut rx_a = submit_ops(&actor, batch_a)?;

        // Batch B: 2 writes with op_idx 20, 21. Shares the QP with
        // Batch A — together they sit at 4/4 max_send_wr.
        let batch_b = vec![
            make_op(20, RdmaOpType::WriteFromLocal, 0x3000, 4096),
            make_op(21, RdmaOpType::WriteFromLocal, 0x4000, 4096),
        ];
        let mut rx_b = submit_ops(&actor, batch_b)?;

        // Collect 4 post events (one per write).
        let mut all_wr_ids = Vec::new();
        for _ in 0..4 {
            let (_, _, wrs) = expect_put(recv_posted(&mut posted_rx).await);
            all_wr_ids.extend(wrs);
        }
        for &id in &all_wr_ids {
            qp.queue_completion(id);
        }

        let replies_a = collect_replies(&mut rx_a, 2).await;
        assert_eq!(replies_a, vec![(10, Ok(())), (11, Ok(()))]);
        let replies_b = collect_replies(&mut rx_b, 2).await;
        assert_eq!(replies_b, vec![(20, Ok(())), (21, Ok(()))]);
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_op_too_large_for_qp() -> Result<()> {
        let harness = QpaHarness::build()?;
        // max_send_wr=1 with a 2-chunk write → can never fit.
        let (actor, qp, mut posted_rx) = harness.spawn_ready_actor(1).await?;

        let items = vec![
            make_op(0, RdmaOpType::WriteFromLocal, 0x1000, 2 * MAX_RDMA_MSG_SIZE),
            make_op(1, RdmaOpType::WriteFromLocal, 0x2000, 4096),
        ];
        let mut rx = submit_ops(&actor, items)?;

        // Only op_idx 1 reaches the wire (op_idx 0 was rejected).
        let (_, _, wrs) = expect_put(recv_posted(&mut posted_rx).await);
        assert_eq!(wrs, vec![0]);
        qp.queue_completion(0);

        let replies = collect_replies(&mut rx, 2).await;
        // op_idx 0 must report a "too large" error; op_idx 1 succeeds.
        assert_eq!(replies.len(), 2);
        assert_eq!(replies[0].0, 0);
        let err = replies[0]
            .1
            .as_ref()
            .expect_err("op_idx 0 should fail as too large");
        assert!(err.contains("too large"), "expected too-large error: {err}");
        assert_eq!(replies[1], (1usize, Ok(())));
        harness.teardown().await;
        Ok(())
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn qpa_post_error_kills_actor_via_supervision() -> Result<()> {
        let mut harness = QpaHarness::build()?;
        let (actor, qp, _posted_rx) = harness.spawn_ready_actor(4).await?;

        qp.queue_post_error("simulated post failure");
        let items = vec![make_op(0, RdmaOpType::WriteFromLocal, 0x1000, 4096)];
        let _rx = submit_ops(&actor, items)?;

        let event = harness.next_supervision_event().await;
        assert_eq!(&event.actor_id, actor.actor().actor_addr());
        let report = event.failure_report().expect("event should be a failure");
        assert!(
            report.contains("qp.put failed") && report.contains("simulated post failure"),
            "supervision report should name the post failure: {report}",
        );
        await_status(&actor, |s| {
            matches!(s, hyperactor::actor::ActorStatus::Failed(_))
        })
        .await;
        harness.teardown().await;
        Ok(())
    }
}