monarch_rdma/backend/ibverbs/

manager_actor.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 Manager
//!
//! Contains ibverbs-specific RDMA logic.
//!
//! Manages ibverbs resources including:
//! - Memory registration (CPU and CUDA via dmabuf or segment scanning)
//! - Queue pair creation and connection establishment
//! - RDMA domain and protection domain management
//! - Device selection and PCI-to-RDMA device mapping

use std::collections::HashMap;
use std::fmt::Write as _;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration;

use anyhow::Result;
use async_trait::async_trait;
use dashmap::DashMap;
use hyperactor::Actor;
use hyperactor::ActorHandle;
use hyperactor::ActorId;
use hyperactor::ActorRef;
use hyperactor::Context;
use hyperactor::Endpoint as _;
use hyperactor::HandleClient;
use hyperactor::Handler;
use hyperactor::Instance;
use hyperactor::OncePortHandle;
use hyperactor::OncePortRef;
use hyperactor::actor::Referable;
use rand::seq::SliceRandom;
use serde::Deserialize;
use serde::Serialize;
use tokio::sync::mpsc;
use typeuri::Named;

use super::IbvOp;
use super::cq_actor::CompletionQueueActor;
use super::device::IbvDevice;
use super::device::IbvDeviceImpl;
use super::device_selection::PeerDeviceAffinityPolicy;
use super::device_selection::configured_peer_device_affinity;
use super::device_selection::resolve_target;
use super::device_selection::select_optimal_ibv_devices;
use super::domain::IbvDomain;
use super::domain::IbvDomainImpl;
use super::efa_device::EfaDevice;
use super::memory_region::IbvMemoryRegionView;
use super::memory_region::IbvRemoteMemoryRegionView;
use super::mlx_device::MlxDevice;
use super::primitives::IbvConfig;
use super::primitives::IbvCq;
use super::primitives::IbvQpInfo;
use super::primitives::ibverbs_supported;
use super::queue_pair::IbvQueuePair;
use super::queue_pair::ProcessOps;
use super::queue_pair::QpKey;
use super::queue_pair::QueuePairActor;
use super::queue_pair::QueuePairHandle;
use super::queue_pair::QueuePairOp;
use super::queue_pair::StripeId;
use super::queue_pair::StripeResult;
use super::queue_pair::legacy;
use crate::RdmaOp;
use crate::RdmaOpType;
use crate::RdmaTransportLevel;
use crate::backend::RdmaBackend;
use crate::backend::RdmaConfig;
use crate::backend::ResolveRemoteBackendContext;
use crate::local_memory::KeepaliveLocalMemory;
use crate::rdma_components::RdmaRemoteBuffer;
use crate::rdma_manager_actor::RdmaManagerActor;
use crate::validate_execution_context;

const KIB: usize = 1024;
// GB200 host-memory RDMA reaches full bandwidth only when each stripe starts
// on a cache-line boundary.
const STRIPE_ALIGNMENT: usize = 64;

fn configured_min_stripe_size() -> NonZeroUsize {
    let kilobytes = hyperactor_config::global::get(crate::config::RDMA_MIN_STRIPE_SIZE_KB).get();
    NonZeroUsize::new(
        kilobytes
            .checked_mul(KIB)
            .expect("RDMA_MIN_STRIPE_SIZE_KB fits in bytes"),
    )
    .expect("a non-zero KiB count produces a non-zero byte count")
}

/// Cross-proc message: the active side asks the peer's manager to
/// create and connect a mirror QP for an in-flight [`QueuePairActor`].
/// Generic over the manager actor type so test code can swap in a
/// mock.
#[derive(Debug, Serialize, Deserialize, Named)]
#[serde(bound(serialize = "", deserialize = ""))]
pub(super) struct CreatePeerQueuePair<M: Referable> {
    /// The active side's manager.
    pub(super) sender: ActorRef<M>,
    /// Device the active side picked for its QP.
    pub(super) sender_device: String,
    /// Device the peer should create its mirror QP on.
    pub(super) receiver_device: String,
    /// Distinguishes parallel QPs for the same device route.
    pub(super) qp_index: usize,
    /// Active side's endpoint, captured right after QP creation.
    pub(super) sender_info: IbvQpInfo,
    /// One-shot reply carrying the peer's endpoint, or an error.
    pub(super) reply: OncePortRef<Result<IbvQpInfo, String>>,
}
wirevalue::register_type!(CreatePeerQueuePair<IbvManagerActor<MlxDevice>>);
wirevalue::register_type!(CreatePeerQueuePair<IbvManagerActor<EfaDevice>>);

/// Local-only message: submit a batch of RDMA ops for end-to-end
/// execution. The manager iterates the batch, resolves each op's
/// local MRs via [`IbvManagerActor::resolve_local_mrs`], splits the operation
/// across compatible registration pairs with [`QueuePairRouter::plan_ops`],
/// and dispatches each stripe to the active-side [`QueuePairActor`] for its
/// [`QpKey`].
///
/// Per-stripe completion notifications stream back on `reply` as
/// [`StripeResult`] values.
pub(super) struct SubmitOps<I: IbvDeviceImpl> {
    ops: Vec<ManagerOp<I>>,
    reply: mpsc::UnboundedSender<StripeResult>,
}

enum ManagerOp<I: IbvDeviceImpl> {
    NeedsMemReg {
        op_idx: usize,
        op: IbvOp<IbvManagerActor<I>>,
    },
    NeedsQueuePair {
        peer_manager: ActorRef<IbvManagerActor<I>>,
        ops: Vec<QueuePairOp>,
    },
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct QpRoute {
    self_device: String,
    other_id: ActorId,
    other_device: String,
}

/// Shared state for selecting a compatible registration pair and dispatching
/// operations directly to initialized queue-pair workers.
///
/// The manager installs a route after creating a queue pair and removes all
/// routes before tearing its queue pairs down. Each [`IbvBackend`] shares this
/// router with the manager so warm submissions can bypass the manager mailbox;
/// operations without local registrations or a matching route fall back to
/// [`SubmitOps`].
#[derive(Debug)]
struct QueuePairRouter {
    peer_device_affinity: PeerDeviceAffinityPolicy,
    qps_per_peer: NonZeroUsize,
    next_qp_indices: DashMap<QpRoute, usize>,
    queue_pairs: DashMap<QpKey, mpsc::UnboundedSender<ProcessOps>>,
}

impl QueuePairRouter {
    fn new(peer_device_affinity: PeerDeviceAffinityPolicy, qps_per_peer: NonZeroUsize) -> Self {
        Self {
            peer_device_affinity,
            qps_per_peer,
            next_qp_indices: DashMap::new(),
            queue_pairs: DashMap::new(),
        }
    }

    fn next_qp_index(&self, route: QpRoute) -> usize {
        let mut next = self.next_qp_indices.entry(route).or_insert(0);
        let current = *next;
        *next = (current + 1) % self.qps_per_peer.get();
        current
    }

    /// Returns a disjoint list of registration pairings allowed by the affinity policy.
    fn pair_registrations<'a>(
        &self,
        local: &'a [IbvMemoryRegionView],
        remote: &'a [IbvRemoteMemoryRegionView],
    ) -> Result<Vec<(&'a IbvMemoryRegionView, &'a IbvRemoteMemoryRegionView)>, anyhow::Error> {
        // `PeerDeviceAffinityPolicy::pairs` is sensitive to input order. Sort
        // both sides so selection depends only on the unordered registration
        // sets and is consistent across processes.
        let mut local: Vec<&IbvMemoryRegionView> = local.iter().collect();
        local.sort_by(|a, b| a.device_name.cmp(&b.device_name));
        let mut remote: Vec<&IbvRemoteMemoryRegionView> = remote.iter().collect();
        remote.sort_by(|a, b| a.device_name.cmp(&b.device_name));

        let local_names: Vec<String> = local.iter().map(|mr| mr.device_name.clone()).collect();
        let remote_names: Vec<String> = remote.iter().map(|mr| mr.device_name.clone()).collect();
        let pairs = self
            .peer_device_affinity
            .pairs(&local_names, &remote_names)
            .into_iter()
            .enumerate()
            .filter_map(|(i, peer)| peer.map(|j| (local[i], remote[j])))
            .collect::<Vec<_>>();
        anyhow::ensure!(
            !pairs.is_empty(),
            "no NIC of {local_names:?} pairs with a peer NIC of {remote_names:?} under {:?}",
            self.peer_device_affinity,
        );
        Ok(pairs)
    }

    /// Splits an operation into stripes over its compatible NIC pairs.
    ///
    /// An operation gets at least `qps_per_peer` stripes, rounded up to a
    /// multiple of the pair count, as far as
    /// [`RDMA_MIN_STRIPE_SIZE_KB`](crate::config::RDMA_MIN_STRIPE_SIZE_KB)
    /// allows. One that is too small for a stripe on every pair uses a random
    /// subset of the pairs, one stripe each. Otherwise the pairs take stripes
    /// in turn, in device-name order. Each stripe uses the next queue pair on
    /// its route.
    fn plan_ops<I: IbvDeviceImpl>(
        &self,
        op_idx: usize,
        op: &IbvOp<IbvManagerActor<I>>,
    ) -> Result<Vec<QueuePairOp>, anyhow::Error> {
        let local_mrs = op.local_memory.registered_mrs::<I>();
        let mut pairs = self.pair_registrations(&local_mrs, &op.remote_buffers)?;
        let other_id = op.remote_manager.actor_addr().id().clone();

        let size = match op.op_type {
            RdmaOpType::WriteFromLocal => op.local_memory.size(),
            RdmaOpType::ReadIntoLocal => op
                .remote_buffers
                .first()
                .map(|remote| remote.size)
                .unwrap_or(0),
        };
        let max_stripes = (size / configured_min_stripe_size().get()).max(1);
        let stripe_count = if max_stripes < pairs.len() {
            pairs.shuffle(&mut rand::rng());
            max_stripes
        } else {
            // Every pair takes the same number of stripes, so no NIC carries
            // more than its share.
            let wanted_per_pair = self.qps_per_peer.get().div_ceil(pairs.len());
            let allowed_per_pair = max_stripes / pairs.len();
            pairs.len() * wanted_per_pair.min(allowed_per_pair)
        };

        let stripe_size = size / stripe_count / STRIPE_ALIGNMENT * STRIPE_ALIGNMENT;
        let mut offset = 0;
        (0..stripe_count)
            .map(|stripe_idx| {
                let (local, remote) = pairs[stripe_idx % pairs.len()];
                let size = if stripe_idx + 1 == stripe_count {
                    size - offset
                } else {
                    stripe_size
                };
                let route = QpRoute {
                    self_device: local.device_name.clone(),
                    other_id: other_id.clone(),
                    other_device: remote.device_name.clone(),
                };
                let op = QueuePairOp {
                    stripe_id: StripeId {
                        op_idx,
                        stripe_idx,
                        stripe_count,
                    },
                    qp_index: self.next_qp_index(route),
                    op_type: op.op_type,
                    local_memory: op.local_memory.clone(),
                    local: local.try_slice(offset, size)?,
                    remote: remote.try_slice(offset, size)?,
                };
                offset += size;
                Ok(op)
            })
            .collect()
    }

    fn try_dispatch<I: IbvDeviceImpl>(
        &self,
        op_idx: usize,
        op: IbvOp<IbvManagerActor<I>>,
        reply: &mpsc::UnboundedSender<StripeResult>,
    ) -> Option<ManagerOp<I>> {
        if op.local_memory.registered_mrs::<I>().is_empty() {
            return Some(ManagerOp::NeedsMemReg { op_idx, op });
        }
        let planned = match self.plan_ops(op_idx, &op) {
            Ok(planned) => planned,
            Err(error) => {
                let _ = reply.send(StripeResult {
                    stripe_id: StripeId {
                        op_idx,
                        stripe_idx: 0,
                        stripe_count: 1,
                    },
                    result: Err(error.to_string()),
                });
                return None;
            }
        };
        let senders = planned
            .iter()
            .map(|planned| {
                let key = QpKey {
                    self_device: planned.local.device_name.clone(),
                    other_id: op.remote_manager.actor_addr().id().clone(),
                    other_device: planned.remote.device_name.clone(),
                    qp_index: planned.qp_index,
                };
                self.queue_pairs
                    .get(&key)
                    .map(|sender| sender.value().clone())
            })
            .collect::<Option<Vec<_>>>();
        let Some(senders) = senders else {
            return Some(ManagerOp::NeedsQueuePair {
                peer_manager: op.remote_manager,
                ops: planned,
            });
        };

        for (sender, planned) in senders.into_iter().zip(planned) {
            let stripe_id = planned.stripe_id;
            if sender
                .send(ProcessOps {
                    items: vec![planned],
                    reply: reply.clone(),
                })
                .is_err()
            {
                let _ = reply.send(StripeResult {
                    stripe_id,
                    result: Err("queue pair worker stopped".to_owned()),
                });
            }
        }
        None
    }
}

/// Local-only message: create a fresh, unconnected legacy
/// [`legacy::IbvQueuePair`] on `self_device` and return it. The caller drives
/// the connection itself — exchange [`IbvQpInfo`] with the other endpoint's QP
/// and call `connect` on each side. Lets doorbell tests and the
/// `cuda_ping_pong` example poke a real QP without going through
/// [`QueuePairActor`]; both want the legacy queue pair for its direct
/// device-doorbell data path, independent of the backend's production
/// [`IbvDomainImpl::QueuePair`].
pub struct RawQueuePair {
    pub self_device: String,
    pub reply: OncePortHandle<Result<legacy::IbvQueuePair, String>>,
}

/// Local-only messages for [`IbvManagerActor`].
#[derive(Handler, HandleClient, Debug)]
pub enum IbvManagerLocalMessage {
    /// Register `local`'s MRs and reply with one
    /// [`IbvRemoteMemoryRegionView`] per registration. Called by
    /// [`crate::rdma_manager_actor::RdmaManagerActor::request_buffer`]
    /// at buffer-creation time.
    ///
    /// The registrations live on `local` itself, so they stay in force for as
    /// long as any holder of that handle does.
    RegisterRemoteBuffer {
        local: KeepaliveLocalMemory,
        #[reply]
        reply: OncePortHandle<Result<Vec<IbvRemoteMemoryRegionView>, String>>,
    },
}

/// Default key used for the per-device protection domain inside
/// each [`IbvDevice<I>`] entry of [`IbvManagerActor::devices`].
const DEFAULT_DOMAIN: &str = "default";

/// Manages all ibverbs-specific RDMA resources and operations.
///
/// This struct handles memory registration, queue pair management,
/// and connection establishment using the ibverbs API.
///
/// Generic over `I: IbvDeviceImpl` so the same actor implementation
/// drives every concrete backend (`IbvManagerActor<MlxDevice>`,
/// `IbvManagerActor<EfaDevice>`, ...).
#[derive(Debug)]
#[hyperactor::export(
    handlers = [
        CreatePeerQueuePair<IbvManagerActor<I>>,
    ],
)]
pub struct IbvManagerActor<I: IbvDeviceImpl> {
    owner: OnceLock<ActorHandle<RdmaManagerActor>>,

    /// Active-side queue-pair workers and their supervising actor, keyed
    /// from this manager's perspective. Lazily populated on the first
    /// [`SubmitOps`] that targets a new `(self_device, peer,
    /// other_device, qp_index)` tuple.
    qp_handles: HashMap<
        QpKey,
        QueuePairHandle<IbvManagerActor<I>, <I::Domain as IbvDomainImpl>::QueuePair>,
    >,

    /// Passive-side mirror QPs, created in response to a peer's
    /// [`CreatePeerQueuePair`]. The peer's [`QueuePairActor`] owns
    /// the active side; we hold the connected mirror here so the
    /// peer can read/write our memory. The map's `Drop` destroys
    /// each QP via its own `Drop`.
    peer_created_qps: HashMap<QpKey, <I::Domain as IbvDomainImpl>::QueuePair>,

    /// Map of RDMA device names to their opened [`IbvDevice<I>`], each of
    /// which owns the per-device `Arc<IbvContext>` and the `DEFAULT_DOMAIN`
    /// `Arc<IbvDomain>`.
    devices: HashMap<String, IbvDevice<I>>,

    /// Which of a peer's NICs each of this manager's NICs may pair with, from
    /// [`RDMA_PEER_DEVICE_AFFINITY`](crate::config::RDMA_PEER_DEVICE_AFFINITY).
    /// Read once, when the manager starts.
    peer_device_affinity: PeerDeviceAffinityPolicy,

    /// Warm QP routes shared with [`IbvBackend`] for direct steady-state
    /// submission without a manager-mailbox round trip.
    queue_pair_router: Arc<QueuePairRouter>,

    /// Completion pollers, keyed by device name when each device gets its own
    /// poller and by the empty string when all devices share one.
    cq_pollers: HashMap<String, ActorHandle<CompletionQueueActor<IbvCq>>>,

    config: IbvConfig,
}

#[async_trait]
impl<I: IbvDeviceImpl> Actor for IbvManagerActor<I> {
    async fn init(&mut self, this: &Instance<Self>) -> Result<(), anyhow::Error> {
        this.set_system();
        let owner = if let Some(owner) = this.parent_handle() {
            owner
        } else {
            anyhow::bail!("RdmaManagerActor not found as parent of IbvManagerActor");
        };
        self.owner
            .set(owner)
            .expect("owner should only be set once during init");
        Ok(())
    }

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

impl<I: IbvDeviceImpl> Drop for IbvManagerActor<I> {
    fn drop(&mut self) {
        self.queue_pair_router.queue_pairs.clear();

        // Stop each supervising QP actor; its cleanup cancels and joins the
        // data-plane worker that owns the QP.
        for (_key, handle) in self.qp_handles.drain() {
            let _ = handle.drain_and_stop("IbvManagerActor dropped");
        }

        for (_key, poller) in self.cq_pollers.drain() {
            let _ = poller.drain_and_stop("IbvManagerActor dropped");
        }

        // The remaining fields (`peer_created_qps`, `devices`) free their FFI
        // resources through their elements' `Drop`s when this struct is dropped.
    }
}

impl<I: IbvDeviceImpl> IbvManagerActor<I> {
    /// Create a new IbvManagerActor with the given configuration.
    pub async fn new(params: Option<IbvConfig>) -> Result<Self, anyhow::Error> {
        if !ibverbs_supported() {
            return Err(anyhow::anyhow!(
                "Cannot create IbvManagerActor because RDMA is not supported on this machine"
            ));
        }

        // Use the caller's config; when none is given, start from the
        // defaults and let the backend seed its own.
        let mut config = match params {
            Some(config) => config,
            None => {
                let mut config = IbvConfig::default();
                I::apply_config_defaults(&mut config);
                config
            }
        };

        // Preserve an explicit per-manager target. Otherwise parse the process
        // default from `RDMA_IBVERBS_TARGET`; empty preserves automatic selection.
        if config.target.is_none() {
            config.target = super::device_selection::configured_ibverbs_target()?;
        }
        tracing::debug!("rdma is enabled, config target: {:?}", config.target);

        // check config and hardware support align
        if config.use_gpu_direct {
            match validate_execution_context().await {
                Ok(_) => {
                    tracing::info!("GPU Direct RDMA execution context validated successfully");
                }
                Err(e) => {
                    tracing::warn!(
                        "GPU Direct RDMA execution context validation failed: {}. Downgrading to standard ibverbs mode.",
                        e
                    );
                    config.use_gpu_direct = false;
                }
            }
        }

        let peer_device_affinity = configured_peer_device_affinity()?;
        let actor = Self {
            owner: OnceLock::new(),
            qp_handles: HashMap::new(),
            peer_created_qps: HashMap::new(),
            devices: HashMap::new(),
            peer_device_affinity: peer_device_affinity.clone(),
            queue_pair_router: Arc::new(QueuePairRouter::new(
                peer_device_affinity,
                hyperactor_config::global::get(crate::config::RDMA_QPS_PER_PEER).into_std(),
            )),
            cq_pollers: HashMap::new(),
            config,
        };

        Ok(actor)
    }

    fn queue_pair_router(&self) -> Arc<QueuePairRouter> {
        Arc::clone(&self.queue_pair_router)
    }

    /// Get or create the `DEFAULT_DOMAIN` for the named RDMA device, opening
    /// the device on first use.
    fn get_or_create_device_domain(
        &mut self,
        device_name: &str,
    ) -> Result<&IbvDomain<I::Domain>, anyhow::Error> {
        self.get_or_create_device(device_name)?
            .get_or_create_domain(DEFAULT_DOMAIN)
    }

    /// Get the named RDMA device, opening it on first use.
    fn get_or_create_device(
        &mut self,
        device_name: &str,
    ) -> Result<&mut IbvDevice<I>, anyhow::Error> {
        if !self.devices.contains_key(device_name) {
            let device = IbvDevice::<I>::try_open(device_name, self.config.clone())?;
            // Print device info if MONARCH_DEBUG_RDMA=1 is set.
            crate::print_device_info_if_debug_enabled(device.context().as_ptr());
            self.devices.insert(device_name.to_string(), device);
        }
        Ok(self
            .devices
            .get_mut(device_name)
            .expect("device just inserted or already present"))
    }

    /// Chooses a set of NICs on which to register `mem`, then registers
    /// `mem` on those NICs (caching the registrations inside the handle)
    /// and returns the relevant [`IbvMemoryRegionView`]s. The call fails
    /// only when there are no successful registrations.
    ///
    /// If an explicit `config.target` is set, then that is the NIC that is
    /// chosen. Otherwise, NICs are chosen based on `pick_optimal_devices`.
    ///
    /// A region already registered on this backend keeps the NICs it has: the
    /// set is chosen once, on the first call.
    fn resolve_local_mrs(
        &mut self,
        mem: &KeepaliveLocalMemory,
    ) -> Result<Vec<IbvMemoryRegionView>, anyhow::Error> {
        let already_serving = mem.registered_mrs::<I>();
        if !already_serving.is_empty() {
            return Ok(already_serving);
        }

        let device_names = match &self.config.target {
            Some(target) => vec![
                resolve_target::<I>(target)?
                    .ok_or_else(|| {
                        anyhow::anyhow!("configured device target {:?} not found", target)
                    })?
                    .name()
                    .clone(),
            ],
            None => self.pick_optimal_devices(mem)?,
        };

        let mut views = Vec::with_capacity(device_names.len());
        let mut failure: Option<anyhow::Error> = None;
        for device_name in &device_names {
            match self.resolve_local_mr_on(mem, device_name) {
                Ok(view) => views.push(view),
                Err(error) => {
                    tracing::warn!(
                        "not serving [{:#x}, {:#x}) from {device_name}: {error:#}",
                        mem.addr(),
                        mem.addr() + mem.size(),
                    );
                    failure.get_or_insert(error);
                }
            }
        }
        if views.is_empty() {
            return Err(failure.expect("`device_names` is never empty, so one of them failed"));
        }
        Ok(views)
    }

    /// Resolve `mem` to its [`IbvMemoryRegionView`] on `device_name`,
    /// registering the region there on first use.
    ///
    /// The registration lives on `mem` itself, shared by every clone of that
    /// handle, so one handle registers a given device at most once. Nothing
    /// deduplicates separate handles over the same region, though: each
    /// carries its own registrations, so a region covered by two handles is
    /// registered once per handle. A failure is recorded on `mem` too and
    /// returned to later callers rather than retried.
    fn resolve_local_mr_on(
        &mut self,
        mem: &KeepaliveLocalMemory,
        device_name: &str,
    ) -> Result<IbvMemoryRegionView, anyhow::Error> {
        if let Some(mrv) = mem.registered_mr::<I>(device_name)? {
            return Ok(mrv);
        }
        tracing::debug!(
            "Using RDMA device: {} for memory at 0x{:x}",
            device_name,
            mem.addr()
        );

        // The backend strategy handles host vs. device memory (standard MR,
        // dmabuf MR, or a device-specific segment binding).
        let registered = self
            .get_or_create_device_domain(device_name)
            .and_then(|domain| domain.register_mr(mem));
        match registered {
            Ok(mrv) => mem.install_mr::<I>(mrv),
            Err(error) => {
                mem.record_mr_failure::<I>(device_name, &error);
                Err(error)
            }
        }
    }

    /// The NICs of backend `I` to serve `mem` from: up to
    /// [`RDMA_MAX_NICS_PER_BUFFER`](crate::config::RDMA_MAX_NICS_PER_BUFFER)
    /// of the optimal NICs returned by [`select_optimal_ibv_devices`], chosen
    /// using [`PeerDeviceAffinityPolicy::choose`].
    fn pick_optimal_devices(
        &self,
        mem: &KeepaliveLocalMemory,
    ) -> Result<Vec<String>, anyhow::Error> {
        let location = mem.location();
        let devices = select_optimal_ibv_devices::<I>(location)?;
        anyhow::ensure!(
            !devices.is_empty(),
            "no {} RDMA device has a path to {location:?}",
            I::backend_name(),
        );
        let names: Vec<String> = devices.iter().map(|device| device.name().clone()).collect();
        let max = hyperactor_config::global::get(crate::config::RDMA_MAX_NICS_PER_BUFFER);
        // `names` is passed in the order returned by `select_optimal_ibv_devices`, which is
        // lexicographic. `PeerDeviceAffinityPolicy::choose` is sensitive to input order,
        // so ensuring `names` has the same order across procs is important for consistency.
        Ok(self
            .peer_device_affinity
            .choose(&names, max.map(hyperactor_config::NonZeroUsize::into_std)))
    }

    /// Build a passive-side mirror QP for `qp_key`, connect it to
    /// `sender_info`, and store it in [`Self::peer_created_qps`].
    /// Returns the local endpoint the active side needs to finish
    /// its own `connect`. Called from
    /// [`Handler<CreatePeerQueuePair>`].
    fn create_peer_qp(
        &mut self,
        qp_key: &QpKey,
        sender_info: &IbvQpInfo,
    ) -> Result<IbvQpInfo, anyhow::Error> {
        if self.peer_created_qps.contains_key(qp_key) {
            anyhow::bail!("peer queue pair already exists for {qp_key:?}");
        }
        let self_device = qp_key.self_device.clone();
        let config = self.config.clone();
        let mut qp = self
            .get_or_create_device(&self_device)?
            .create_non_posting_queue_pair(DEFAULT_DOMAIN, &config)
            .map_err(|e| anyhow::anyhow!("could not create peer IbvQueuePair: {}", e))?;
        let local_info = qp
            .get_qp_info()
            .map_err(|e| anyhow::anyhow!("could not extract peer QP info: {}", e))?;
        qp.connect(sender_info)
            .map_err(|e| anyhow::anyhow!("could not connect peer QP: {}", e))?;
        self.peer_created_qps.insert(qp_key.clone(), qp);
        Ok(local_info)
    }

    /// Returns the poller for `device`, spawning it on first use.
    fn ensure_cq_poller(
        &mut self,
        cx: &Context<'_, Self>,
        device: &str,
    ) -> ActorHandle<CompletionQueueActor<IbvCq>> {
        let key = if hyperactor_config::global::get(crate::config::RDMA_CQ_POLLER_PER_DEVICE) {
            device.to_owned()
        } else {
            String::new()
        };
        self.cq_pollers
            .entry(key)
            .or_insert_with(|| cx.spawn(CompletionQueueActor::new()))
            .clone()
    }

    /// Lazy active-side QP worker: if `qp_key` is absent from
    /// [`Self::qp_handles`], create an [`IbvQueuePair`] on the
    /// requested device and spawn a [`QueuePairActor`] to drive its handshake
    /// and supervise its data-plane task. Returns a handle to the QP actor.
    fn ensure_qp_actor(
        &mut self,
        cx: &Context<'_, Self>,
        qp_key: &QpKey,
        peer_manager: ActorRef<Self>,
    ) -> Result<QueuePairHandle<Self, <I::Domain as IbvDomainImpl>::QueuePair>, anyhow::Error> {
        if let Some(h) = self.qp_handles.get(qp_key) {
            return Ok(h.clone());
        }
        let self_device = qp_key.self_device.clone();
        let config = self.config.clone();
        let cq_poller = self.ensure_cq_poller(cx, &self_device);
        let (qp, cq_lease) = self
            .get_or_create_device(&self_device)?
            .create_queue_pair(DEFAULT_DOMAIN, &config)
            .map_err(|e| anyhow::anyhow!("could not create IbvQueuePair for {qp_key:?}: {}", e))?;
        let local_manager: ActorRef<Self> = cx.bind();
        let is_loopback = local_manager.actor_addr() == peer_manager.actor_addr()
            && qp_key.self_device == qp_key.other_device;
        let (actor, sender) = QueuePairActor::new(
            qp_key.clone(),
            local_manager,
            peer_manager,
            qp,
            cq_poller,
            cq_lease,
            is_loopback,
            config.max_send_wr,
        );
        let handle = QueuePairHandle::new(cx.spawn(actor), sender);
        self.queue_pair_router
            .queue_pairs
            .insert(qp_key.clone(), handle.sender());
        self.qp_handles.insert(qp_key.clone(), handle.clone());
        Ok(handle)
    }

    fn dispatch_ops(
        &mut self,
        cx: &Context<'_, Self>,
        peer_manager: ActorRef<Self>,
        ops: Vec<QueuePairOp>,
        reply: &mpsc::UnboundedSender<StripeResult>,
    ) {
        for op in ops {
            let stripe_id = op.stripe_id;
            let qp_key = QpKey {
                self_device: op.local.device_name.clone(),
                other_id: peer_manager.actor_addr().id().clone(),
                other_device: op.remote.device_name.clone(),
                qp_index: op.qp_index,
            };
            let handle = match self.ensure_qp_actor(cx, &qp_key, peer_manager.clone()) {
                Ok(handle) => handle,
                Err(error) => {
                    let _ = reply.send(StripeResult {
                        stripe_id,
                        result: Err(error.to_string()),
                    });
                    continue;
                }
            };
            if let Err(error) = handle.send(ProcessOps {
                items: vec![op],
                reply: reply.clone(),
            }) {
                let _ = reply.send(StripeResult {
                    stripe_id,
                    result: Err(error.to_string()),
                });
            }
        }
    }
}

#[async_trait]
impl<I: IbvDeviceImpl> Handler<SubmitOps<I>> for IbvManagerActor<I> {
    async fn handle(&mut self, cx: &Context<Self>, msg: SubmitOps<I>) -> Result<(), anyhow::Error> {
        let SubmitOps { ops, reply } = msg;

        // Interleave MR resolution with QP dispatch: as soon as an op is
        // planned, send its stripes to their queue-pair workers. They can post
        // the stripes while the manager resolves the next op's registrations.
        for op in ops {
            let (peer_manager, planned) = match op {
                ManagerOp::NeedsQueuePair { peer_manager, ops } => (peer_manager, ops),
                ManagerOp::NeedsMemReg { op_idx, op } => {
                    let whole_op = StripeId {
                        op_idx,
                        stripe_idx: 0,
                        stripe_count: 1,
                    };
                    if let Err(error) = self.resolve_local_mrs(&op.local_memory) {
                        let _ = reply.send(StripeResult {
                            stripe_id: whole_op,
                            result: Err(error.to_string()),
                        });
                        continue;
                    }
                    let planned = match self.queue_pair_router.plan_ops(op_idx, &op) {
                        Ok(planned) => planned,
                        Err(error) => {
                            let _ = reply.send(StripeResult {
                                stripe_id: whole_op,
                                result: Err(error.to_string()),
                            });
                            continue;
                        }
                    };
                    (op.remote_manager, planned)
                }
            };
            self.dispatch_ops(cx, peer_manager, planned, &reply);
        }
        Ok(())
    }
}

#[async_trait]
impl<I: IbvDeviceImpl> Handler<RawQueuePair> for IbvManagerActor<I> {
    async fn handle(&mut self, cx: &Context<Self>, msg: RawQueuePair) -> Result<(), anyhow::Error> {
        let RawQueuePair { self_device, reply } = msg;
        // Build a fresh, unconnected legacy QP on `self_device` and hand it
        // back; the caller exchanges endpoint info and connects it.
        let config = self.config.clone();
        let result = self
            .get_or_create_device_domain(&self_device)
            .and_then(|domain| legacy::IbvQueuePair::new(domain, config))
            .map_err(|e| e.to_string());
        let _ = reply.try_post(cx, result);
        Ok(())
    }
}

#[async_trait]
impl<I: IbvDeviceImpl> Handler<CreatePeerQueuePair<IbvManagerActor<I>>> for IbvManagerActor<I> {
    async fn handle(
        &mut self,
        cx: &Context<Self>,
        msg: CreatePeerQueuePair<IbvManagerActor<I>>,
    ) -> Result<(), anyhow::Error> {
        let CreatePeerQueuePair {
            sender,
            sender_device,
            receiver_device,
            qp_index,
            sender_info,
            reply,
        } = msg;
        let qp_key = QpKey {
            self_device: receiver_device,
            other_id: sender.actor_addr().id().clone(),
            other_device: sender_device,
            qp_index,
        };
        match self.create_peer_qp(&qp_key, &sender_info) {
            Ok(local_info) => reply.post(cx, Ok(local_info)),
            Err(e) => reply.post(cx, Err(e.to_string())),
        }
        Ok(())
    }
}

#[async_trait]
impl<I: IbvDeviceImpl> IbvManagerLocalMessageHandler for IbvManagerActor<I> {
    async fn register_remote_buffer(
        &mut self,
        _cx: &Context<Self>,
        local: KeepaliveLocalMemory,
    ) -> Result<Result<Vec<IbvRemoteMemoryRegionView>, String>, anyhow::Error> {
        // The registration is installed on `local`'s shared map, so every clone
        // of this handle — including the one the caller holds — reuses it rather
        // than registering the same region on that device again.
        Ok(self
            .resolve_local_mrs(&local)
            .map(|mrs| mrs.iter().map(IbvRemoteMemoryRegionView::from).collect())
            .map_err(|error| format!("{error:#}")))
    }
}

// `#[hyperactor::handle(IbvManagerLocalMessage)]` would generate a non-generic
// `impl Handler<...> for IbvManagerActor<I>` that can't see `I`; we write the
// generic delegation by hand.
#[async_trait]
impl<I: IbvDeviceImpl> Handler<IbvManagerLocalMessage> for IbvManagerActor<I> {
    async fn handle(
        &mut self,
        cx: &Context<Self>,
        message: IbvManagerLocalMessage,
    ) -> Result<(), anyhow::Error> {
        <Self as IbvManagerLocalMessageHandler>::handle(self, cx, message).await
    }
}

/// Wrapper around [`ActorHandle<IbvManagerActor<I>>`] that moves the RDMA
/// data-plane (post send/recv, poll CQ) off the actor loop while keeping
/// state-mutating operations (MR registration/deregistration, QP management)
/// serialized through actor messages.
#[derive(Debug)]
pub struct IbvBackend<I: IbvDeviceImpl> {
    manager: ActorHandle<IbvManagerActor<I>>,
    queue_pair_router: Arc<QueuePairRouter>,
}

impl<I: IbvDeviceImpl> Clone for IbvBackend<I> {
    fn clone(&self) -> Self {
        Self {
            manager: self.manager.clone(),
            queue_pair_router: Arc::clone(&self.queue_pair_router),
        }
    }
}

impl<I: IbvDeviceImpl> std::ops::Deref for IbvBackend<I> {
    type Target = ActorHandle<IbvManagerActor<I>>;
    fn deref(&self) -> &Self::Target {
        &self.manager
    }
}

/// Serializable per-buffer context for an ibverbs backend: the manager to route
/// ops through and the wire description of every MR the buffer is registered
/// under, one per NIC serving it.
#[derive(Serialize, Deserialize, Named)]
#[serde(bound = "")]
pub struct IbvRemoteBackendContext<I: IbvDeviceImpl> {
    pub manager: ActorRef<IbvManagerActor<I>>,
    pub buffers: Vec<IbvRemoteMemoryRegionView>,
}

// `Clone` and `Debug` are hand-rolled to avoid the spurious `I: Clone`
// and `I: Debug` bounds the derives would impose; neither field depends
// on `I` implementing them.
impl<I: IbvDeviceImpl> Clone for IbvRemoteBackendContext<I> {
    fn clone(&self) -> Self {
        Self {
            manager: self.manager.clone(),
            buffers: self.buffers.clone(),
        }
    }
}

impl<I: IbvDeviceImpl> std::fmt::Debug for IbvRemoteBackendContext<I> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("IbvRemoteBackendContext")
            .field("manager", &self.manager)
            .field("buffers", &self.buffers)
            .finish()
    }
}

#[async_trait]
impl<I: IbvDeviceImpl> RdmaBackend for IbvBackend<I>
where
    RdmaRemoteBuffer: ResolveRemoteBackendContext<IbvBackend<I>>,
{
    type RemoteBackendContext = IbvRemoteBackendContext<I>;
    type TransportInfo = ();

    fn available() -> bool {
        if IbvDevice::<I>::available() {
            if hyperactor_config::global::get(crate::config::RDMA_DISABLE_IBVERBS) {
                tracing::warn!(
                    "ibverbs ({}) is available, but it was disabled by configuration (RDMA_DISABLE_IBVERBS=true)",
                    I::backend_name()
                );
                return false;
            }
            return true;
        }
        false
    }

    fn transport_level(&self) -> RdmaTransportLevel {
        RdmaTransportLevel::Nic
    }

    fn transport_info(&self) -> Option<Self::TransportInfo> {
        None
    }

    async fn spawn(
        cx: &(impl hyperactor::context::Actor + Send + Sync),
        config: &RdmaConfig,
    ) -> Result<Self> {
        let actor = IbvManagerActor::<I>::new(config.ibv.clone()).await?;
        let queue_pair_router = actor.queue_pair_router();
        Ok(Self {
            manager: cx.spawn(actor),
            queue_pair_router,
        })
    }

    async fn register_remote_buffer(
        &self,
        cx: &(impl hyperactor::context::Actor + Send + Sync),
        _remote_buf_id: usize,
        local: KeepaliveLocalMemory,
    ) -> Result<IbvRemoteBackendContext<I>> {
        let buffers = self
            .manager
            .register_remote_buffer(cx, local)
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        Ok(IbvRemoteBackendContext {
            manager: self.manager.bind(),
            buffers,
        })
    }

    /// No-op: this backend holds no per-buffer state. A region's registrations
    /// live on the [`KeepaliveLocalMemory`] they were made for, so a buffer is
    /// released by its owner dropping that handle.
    async fn release_buffer(
        &self,
        _cx: &(impl hyperactor::context::Actor + Send + Sync),
        _remote_buf_id: usize,
    ) -> Result<()> {
        Ok(())
    }

    /// Submit a batch of RDMA operations.
    ///
    /// Translates each op to an `IbvOp`, then forwards them directly to the
    /// relevant queue-pairs when possible; any ops that cannot be forwarded
    /// directly (memory registration needed, QPs don't exist yet) are shipped to
    /// [`IbvManagerActor`] via [`SubmitOps`]. The manager interleaves local-MR
    /// resolution with dispatch.
    ///
    /// Waits for every stripe of every operation. Every stripe failure is
    /// collected into a single error, grouped by operation.
    async fn submit(
        &self,
        cx: &(impl hyperactor::context::Actor + Send + Sync),
        ops: Vec<RdmaOp>,
        timeout: Duration,
    ) -> Result<(), anyhow::Error> {
        let mut ibv_ops = Vec::with_capacity(ops.len());
        for op in ops {
            let ctx = <RdmaRemoteBuffer as ResolveRemoteBackendContext<IbvBackend<I>>>::resolve(
                &op.remote,
            )
            .expect("op routed to incompatible backend");
            ibv_ops.push(IbvOp {
                op_type: op.op_type,
                local_memory: op.local.clone(),
                remote_buffers: ctx.buffers,
                remote_manager: ctx.manager,
            });
        }
        let n = ibv_ops.len();
        let (reply, mut reply_rx) = mpsc::unbounded_channel();

        let mut manager_ops = Vec::new();
        for (op_idx, op) in ibv_ops.into_iter().enumerate() {
            if let Some(op) = self.queue_pair_router.try_dispatch::<I>(op_idx, op, &reply) {
                manager_ops.push(op);
            }
        }
        if !manager_ops.is_empty() {
            self.manager.try_post(
                cx,
                SubmitOps {
                    ops: manager_ops,
                    reply,
                },
            )?;
        }

        let mut outstanding: HashMap<usize, usize> = HashMap::with_capacity(n);
        // Per operation, the index and failure message of each of its stripes
        // that failed.
        let mut failures: HashMap<usize, Vec<(usize, String)>> = HashMap::new();
        let mut completed = 0usize;
        let mut received_stripes = 0usize;
        let mut terminal: Option<String> = None;
        let deadline = tokio::time::Instant::now() + timeout;
        while completed < n {
            tokio::select! {
                () = tokio::time::sleep_until(deadline) => {
                    terminal = Some(format!(
                        "submit timed out after {completed}/{n} ops ({received_stripes} stripe replies) with {} failed ops",
                        failures.len()
                    ));
                    break;
                }
                recv = reply_rx.recv() => {
                    match recv {
                        Some(StripeResult { stripe_id, result }) => {
                            received_stripes += 1;
                            if let Err(error) = result {
                                failures
                                    .entry(stripe_id.op_idx)
                                    .or_default()
                                    .push((stripe_id.stripe_idx, error));
                            }
                            let remaining = outstanding
                                .entry(stripe_id.op_idx)
                                .or_insert(stripe_id.stripe_count);
                            *remaining = remaining
                                .checked_sub(1)
                                .expect("an operation reports no more than its stripe count");
                            if *remaining == 0 {
                                outstanding.remove(&stripe_id.op_idx);
                                completed += 1;
                            }
                        }
                        None => {
                            terminal = Some(format!(
                                "operation result channel closed after {completed}/{n} ops ({received_stripes} stripe replies) with {} failed ops: \
                                some built-in RDMA actor likely stopped or failed, see logs for more details",
                                failures.len()
                            ));
                            break;
                        }
                    }
                }
            }
        }

        if terminal.is_none() && failures.is_empty() {
            return Ok(());
        }

        let mut failures: Vec<_> = failures.into_iter().collect();
        failures.sort_by_key(|(idx, _)| *idx);
        let mut msg = terminal.unwrap_or_else(|| format!("{}/{n} ops failed", failures.len()));
        if !failures.is_empty() {
            msg.push(':');
            for (idx, op_failures) in &mut failures {
                op_failures.sort_by_key(|(stripe_idx, _)| *stripe_idx);
                write!(msg, "\n  op {idx}:").expect("infallible String write");
                for (stripe_idx, err) in op_failures {
                    write!(msg, "\n    stripe {stripe_idx}: {err}")
                        .expect("infallible String write");
                }
            }
        }
        Err(anyhow::anyhow!(msg))
    }
}

#[cfg(test)]
mod tests {
    //! End-to-end coverage of the [`SubmitOps`] → [`ProcessOps`] →
    //! [`QueuePairActor`] data path.
    //!
    //! Each test stands up two RDMA participants in two
    //! [`Proc::direct`] procs in the test process. Each proc hosts an
    //! [`RdmaManagerActor`] and a [`BufferHelperActor`]. Tests
    //! allocate buffers on either side via the helpers, drive RDMA
    //! through [`IbvBackend::submit`] (called inside the helper
    //! actor), and verify by reading back local contents through
    //! [`BufferHelperMessage::ReadContents`]. The
    //! [`BufferHelperActor::cleanup`] impl releases any CUDA
    //! allocations when the actor stops; [`TestEnv::shutdown`]
    //! explicitly drains both procs.

    use std::collections::BTreeMap;
    use std::collections::BTreeSet;
    use std::num::NonZeroUsize;
    use std::sync::Arc;
    use std::sync::atomic::AtomicUsize;
    use std::sync::atomic::Ordering;
    use std::time::Duration;

    use async_trait::async_trait;
    use hyperactor::Actor;
    use hyperactor::ActorAddr;
    use hyperactor::ActorEnvironment;
    use hyperactor::ActorRef;
    use hyperactor::Context;
    use hyperactor::Handler;
    use hyperactor::Instance;
    use hyperactor::Label;
    use hyperactor::OncePortRef;
    use hyperactor::Proc;
    use hyperactor::RefClient;
    use hyperactor::RemoteSpawn;
    use hyperactor::Uid;
    use hyperactor::actor::ActorError;
    use hyperactor::channel::ChannelAddr;
    use hyperactor::channel::ChannelTransport;
    use serde::Deserialize;
    use serde::Serialize;
    use typeuri::Named;

    use super::IbvManagerActor;
    use super::IbvOp;
    use super::KIB;
    use super::QpRoute;
    use super::QueuePairRouter;
    use crate::IbvConfig;
    use crate::RdmaManagerActor;
    use crate::RdmaManagerMessageClient;
    use crate::RdmaOp;
    use crate::RdmaOpType;
    use crate::RdmaRemoteBuffer;
    use crate::backend::RdmaBackendHandle;
    use crate::backend::cuda_test_utils::CudaAllocation;
    use crate::backend::cuda_test_utils::CudaAllocator;
    use crate::backend::ibverbs::device::list_all_devices;
    use crate::backend::ibverbs::device_selection::IbvDeviceTarget;
    use crate::backend::ibverbs::device_selection::PeerDeviceAffinityPolicy;
    use crate::backend::ibverbs::device_selection::resolve_target;
    use crate::backend::ibverbs::device_selection::select_optimal_ibv_devices;
    use crate::backend::ibverbs::memory_region::IbvMemoryRegionView;
    use crate::backend::ibverbs::memory_region::IbvRemoteMemoryRegionView;
    use crate::backend::ibverbs::mlx_device::MlxDevice;
    use crate::backend::ibverbs::primitives::IbvQpType;
    use crate::device_selection::MemoryLocation;
    use crate::local_memory::KeepaliveLocalMemory;

    // ====================================================================
    // BufferHelperActor
    // ====================================================================

    /// Device a test buffer is allocated on.
    #[derive(Debug, Clone, Copy, Serialize, Deserialize, Named)]
    pub enum BufferDevice {
        Cpu,
        Cuda(i32),
    }

    /// One op for [`BufferHelperMessage::Submit`]. The helper looks up
    /// the local memory behind `local_buf` (registered earlier via
    /// `Allocate`) and pairs it with `remote_buf` to form an
    /// [`RdmaOp`].
    #[derive(Debug, Clone, Serialize, Deserialize, Named)]
    pub struct BufferHelperOp {
        op_type: RdmaOpType,
        local_buf: RdmaRemoteBuffer,
        remote_buf: RdmaRemoteBuffer,
    }

    /// Test helper that owns local buffers (CPU or CUDA) and drives
    /// [`IbvBackend::submit`] against its own [`RdmaManagerActor`].
    #[hyperactor::export(handlers = [BufferHelperMessage])]
    #[hyperactor::spawnable]
    #[derive(Debug)]
    pub struct BufferHelperActor {
        rdma_manager: ActorRef<RdmaManagerActor>,
        /// CUDA allocations tracked for cleanup. Each is also held as
        /// `Keepalive` inside the registered `KeepaliveLocalMemory`;
        /// both clones must drop before the FFI memory is released.
        cuda_allocs: Vec<CudaAllocation>,
    }

    #[async_trait]
    impl Actor for BufferHelperActor {
        async fn cleanup(
            &mut self,
            _this: &Instance<Self>,
            _err: Option<&ActorError>,
        ) -> Result<(), anyhow::Error> {
            for alloc in self.cuda_allocs.drain(..) {
                alloc.try_free();
            }
            Ok(())
        }
    }

    #[async_trait]
    impl RemoteSpawn for BufferHelperActor {
        type Params = ActorRef<RdmaManagerActor>;

        async fn new(
            rdma_manager: ActorRef<RdmaManagerActor>,
            _env: &ActorEnvironment,
        ) -> Result<Self, anyhow::Error> {
            Ok(Self {
                rdma_manager,
                cuda_allocs: Vec::new(),
            })
        }
    }

    #[derive(Handler, RefClient, Named, Serialize, Deserialize, Debug)]
    pub enum BufferHelperMessage {
        /// Allocate `size` bytes on `device`, pre-fill with `pattern`,
        /// register with the local `RdmaManagerActor`, and reply with
        /// the resulting `RdmaRemoteBuffer`.
        Allocate {
            size: usize,
            device: BufferDevice,
            pattern: u8,
            #[reply]
            reply: OncePortRef<RdmaRemoteBuffer>,
        },
        /// Like `Allocate`, but fills byte `i` with [`positional_byte`]`(i)`.
        AllocatePositional {
            size: usize,
            device: BufferDevice,
            #[reply]
            reply: OncePortRef<RdmaRemoteBuffer>,
        },
        /// Look up the local memory behind `remote.id` and reply with
        /// the byte range `[offset, offset + len)`. Tests use this to
        /// sample buffers too large to ship over a single actor
        /// message in one piece.
        ReadContents {
            remote: Box<RdmaRemoteBuffer>,
            offset: usize,
            len: usize,
            #[reply]
            reply: OncePortRef<Vec<u8>>,
        },
        /// Drive a batch of RDMA ops through `IbvBackend::submit`.
        /// Each op's `local_buf` is resolved against this helper's
        /// `RdmaManagerActor`; `remote_buf` is shipped as-is to the
        /// peer.
        Submit {
            ops: Vec<BufferHelperOp>,
            timeout_secs: u64,
            #[reply]
            reply: OncePortRef<Result<(), String>>,
        },
    }

    impl BufferHelperActor {
        async fn allocate_impl(
            &mut self,
            cx: &Context<'_, Self>,
            device: BufferDevice,
            contents: Vec<u8>,
        ) -> Result<RdmaRemoteBuffer, anyhow::Error> {
            let size = contents.len();
            let local = match device {
                BufferDevice::Cpu => {
                    KeepaliveLocalMemory::try_new(Arc::new(contents.into_boxed_slice()))?
                }
                BufferDevice::Cuda(device_id) => {
                    let alloc = CudaAllocator::get().allocate(device_id, size, size);
                    let local = KeepaliveLocalMemory::try_new(Arc::new(alloc.clone()))?;
                    self.cuda_allocs.push(alloc);
                    // SAFETY: `local` is freshly constructed; no other
                    // holder touches this CUDA range yet.
                    unsafe { local.write_at(0, &contents) }?;
                    local
                }
            };
            let handle = self
                .rdma_manager
                .downcast_handle(cx)
                .ok_or_else(|| anyhow::anyhow!("rdma_manager not local to BufferHelperActor"))?;
            handle.request_buffer(cx, local).await
        }

        async fn read_contents_impl(
            &mut self,
            cx: &Context<'_, Self>,
            remote: RdmaRemoteBuffer,
            offset: usize,
            len: usize,
        ) -> Result<Vec<u8>, anyhow::Error> {
            let handle = self
                .rdma_manager
                .downcast_handle(cx)
                .ok_or_else(|| anyhow::anyhow!("rdma_manager not local"))?;
            let local = handle
                .request_local_memory(cx, remote.id)
                .await?
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "no local memory registered on this side for remote_buf_id={}",
                        remote.id,
                    )
                })?;
            let mut out = vec![0u8; len];
            // SAFETY: by convention the caller has ensured all RDMA
            // ops against this buffer have completed before invoking
            // ReadContents.
            unsafe { local.read_at(offset, &mut out)? };
            Ok(out)
        }

        async fn submit_impl(
            &mut self,
            cx: &Context<'_, Self>,
            ops: Vec<BufferHelperOp>,
            timeout_secs: u64,
        ) -> Result<Result<(), String>, anyhow::Error> {
            let handle = self
                .rdma_manager
                .downcast_handle(cx)
                .ok_or_else(|| anyhow::anyhow!("rdma_manager not local"))?;
            let mut rdma_ops = Vec::with_capacity(ops.len());
            for (i, op) in ops.into_iter().enumerate() {
                let local = handle
                    .request_local_memory(cx, op.local_buf.id)
                    .await?
                    .ok_or_else(|| {
                        anyhow::anyhow!(
                            "op {i}: no local memory registered for remote_buf_id={}",
                            op.local_buf.id,
                        )
                    })?;
                rdma_ops.push(RdmaOp {
                    op_type: op.op_type,
                    local,
                    remote: op.remote_buf,
                });
            }
            let nic = RdmaManagerActor::local_handle(cx)
                .get_backend_handles(cx)
                .await?
                .into_iter()
                .find(|h| !matches!(h, RdmaBackendHandle::Tcp(_)))
                .ok_or_else(|| anyhow::anyhow!("no NIC backend on this proc"))?;
            let result = nic
                .submit(cx, rdma_ops, Duration::from_secs(timeout_secs))
                .await;
            Ok(result.map_err(|e| format!("{e}")))
        }
    }

    #[async_trait]
    #[hyperactor::handle(BufferHelperMessage)]
    impl BufferHelperMessageHandler for BufferHelperActor {
        async fn allocate(
            &mut self,
            cx: &Context<Self>,
            size: usize,
            device: BufferDevice,
            pattern: u8,
        ) -> Result<RdmaRemoteBuffer, anyhow::Error> {
            self.allocate_impl(cx, device, vec![pattern; size]).await
        }

        async fn allocate_positional(
            &mut self,
            cx: &Context<Self>,
            size: usize,
            device: BufferDevice,
        ) -> Result<RdmaRemoteBuffer, anyhow::Error> {
            self.allocate_impl(cx, device, positional_contents(size))
                .await
        }

        async fn read_contents(
            &mut self,
            cx: &Context<Self>,
            remote: Box<RdmaRemoteBuffer>,
            offset: usize,
            len: usize,
        ) -> Result<Vec<u8>, anyhow::Error> {
            self.read_contents_impl(cx, *remote, offset, len).await
        }

        async fn submit(
            &mut self,
            cx: &Context<Self>,
            ops: Vec<BufferHelperOp>,
            timeout_secs: u64,
        ) -> Result<Result<(), String>, anyhow::Error> {
            self.submit_impl(cx, ops, timeout_secs).await
        }
    }

    // ====================================================================
    // TestEnv
    // ====================================================================

    static COUNTER: AtomicUsize = AtomicUsize::new(0);

    /// Two-sided test environment.
    ///
    /// Each side is a `Proc::direct` in the test process hosting its
    /// own `RdmaManagerActor` and `BufferHelperActor`. A client minted
    /// from `proc_b` drives both helpers through their `ActorRef`s.
    struct TestEnv {
        client: hyperactor::Client,
        proc_a: Proc,
        helper_a: ActorRef<BufferHelperActor>,
        proc_b: Proc,
        helper_b: ActorRef<BufferHelperActor>,
    }

    impl TestEnv {
        /// Asymmetric setup: side A uses `config_a`, side B uses `config_b`.
        async fn new(config_a: IbvConfig, config_b: IbvConfig) -> Result<Self, anyhow::Error> {
            let id = COUNTER.fetch_add(1, Ordering::Relaxed);
            let proc_a = Proc::direct(
                ChannelAddr::any(ChannelTransport::Unix),
                format!("rdma_side_a_{id}"),
            )?;
            let helper_a = Self::spawn_side(&proc_a, config_a).await?;
            let proc_b = Proc::direct(
                ChannelAddr::any(ChannelTransport::Unix),
                format!("rdma_side_b_{id}"),
            )?;
            let helper_b = Self::spawn_side(&proc_b, config_b).await?;
            let client = proc_b.client("test_client");
            Ok(Self {
                client,
                proc_a,
                helper_a,
                proc_b,
                helper_b,
            })
        }

        /// Symmetric setup: both sides use `config`.
        async fn same_config(config: IbvConfig) -> Result<Self, anyhow::Error> {
            Self::new(config.clone(), config).await
        }

        /// Spawn an `RdmaManagerActor` + `BufferHelperActor` on `proc`
        /// and return the helper's `ActorRef`.
        async fn spawn_side(
            proc: &Proc,
            config: IbvConfig,
        ) -> Result<ActorRef<BufferHelperActor>, anyhow::Error> {
            let rdma_actor =
                RdmaManagerActor::new(Some(config), &ActorEnvironment::default()).await?;
            // Must match `RdmaManagerActor::local_handle`'s singleton lookup of "rdma_manager".
            let rdma_handle =
                proc.spawn_with_uid(Uid::singleton(Label::strip("rdma_manager")), rdma_actor)?;
            let rdma: ActorRef<RdmaManagerActor> = rdma_handle.bind();
            let helper_actor = BufferHelperActor::new(rdma, &ActorEnvironment::default()).await?;
            let helper_handle = proc.spawn_with_label("helper", helper_actor);
            Ok(helper_handle.bind())
        }

        async fn shutdown(mut self) -> Result<(), anyhow::Error> {
            let _ = self
                .proc_a
                .destroy_and_wait(Duration::from_secs(10), "TestEnv shutdown proc_a")
                .await?;
            let _ = self
                .proc_b
                .destroy_and_wait(Duration::from_secs(10), "TestEnv shutdown proc_b")
                .await?;
            Ok(())
        }
    }

    // ====================================================================
    // Shared test bodies
    // ====================================================================

    async fn assert_remote_pattern(
        helper: &ActorRef<BufferHelperActor>,
        cx: &hyperactor::Client,
        remote: RdmaRemoteBuffer,
        size: usize,
        pattern: u8,
    ) -> Result<(), anyhow::Error> {
        let got = helper.read_contents(cx, Box::new(remote), 0, size).await?;
        assert_eq!(got, vec![pattern; size]);
        Ok(())
    }

    /// The byte at `offset` in a buffer filled by position. 251 is prime, so
    /// the sequence never lines up with a stripe or alignment boundary, and a
    /// range copied to the wrong offset reads back wrong.
    fn positional_byte(offset: usize) -> u8 {
        (offset % 251) as u8
    }

    fn positional_contents(size: usize) -> Vec<u8> {
        (0..size).map(positional_byte).collect()
    }

    /// Panics at the first byte of `got` that does not hold its own
    /// [`positional_byte`].
    fn assert_positional(got: &[u8], size: usize, what: &str) {
        assert_eq!(got.len(), size, "{what}: read back the wrong length");
        if let Some(offset) = (0..size).find(|&i| got[i] != positional_byte(i)) {
            panic!(
                "{what}: byte {offset} of {size} is {:#04x}, expected {:#04x}",
                got[offset],
                positional_byte(offset),
            );
        }
    }

    /// Writes a buffer filled by position from side A into side B, reads one
    /// from side B into side A, and asserts that every byte lands at its own
    /// offset.
    async fn run_positional_write_and_read(
        env: &TestEnv,
        size: usize,
        timeout_secs: u64,
    ) -> Result<(), anyhow::Error> {
        let write_src = env
            .helper_a
            .allocate_positional(&env.client, size, BufferDevice::Cpu)
            .await?;
        let write_dst = env
            .helper_b
            .allocate(&env.client, size, BufferDevice::Cpu, 0)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: write_src,
                    remote_buf: write_dst.clone(),
                }],
                timeout_secs,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        let got = env
            .helper_b
            .read_contents(&env.client, Box::new(write_dst), 0, size)
            .await?;
        assert_positional(&got, size, "write");

        let read_dst = env
            .helper_a
            .allocate(&env.client, size, BufferDevice::Cpu, 0)
            .await?;
        let read_src = env
            .helper_b
            .allocate_positional(&env.client, size, BufferDevice::Cpu)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::ReadIntoLocal,
                    local_buf: read_dst.clone(),
                    remote_buf: read_src,
                }],
                timeout_secs,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        let got = env
            .helper_a
            .read_contents(&env.client, Box::new(read_dst), 0, size)
            .await?;
        assert_positional(&got, size, "read");
        Ok(())
    }

    /// Drive a single write from side A's buffer into side B's buffer
    /// and assert the destination now matches the pattern.
    async fn run_cross_actor_write(
        env: &TestEnv,
        src_dev: BufferDevice,
        dst_dev: BufferDevice,
        size: usize,
        pattern: u8,
        timeout_secs: u64,
    ) -> Result<(), anyhow::Error> {
        let src = env
            .helper_a
            .allocate(&env.client, size, src_dev, pattern)
            .await?;
        let dst = env.helper_b.allocate(&env.client, size, dst_dev, 0).await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: src,
                    remote_buf: dst.clone(),
                }],
                timeout_secs,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_b, &env.client, dst, size, pattern).await
    }

    /// Drive a single read from side B's buffer into side A's buffer
    /// and assert the destination now matches the pattern.
    async fn run_cross_actor_read(
        env: &TestEnv,
        dst_dev: BufferDevice,
        src_dev: BufferDevice,
        size: usize,
        pattern: u8,
        timeout_secs: u64,
    ) -> Result<(), anyhow::Error> {
        let dst = env.helper_a.allocate(&env.client, size, dst_dev, 0).await?;
        let src = env
            .helper_b
            .allocate(&env.client, size, src_dev, pattern)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::ReadIntoLocal,
                    local_buf: dst.clone(),
                    remote_buf: src,
                }],
                timeout_secs,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_a, &env.client, dst, size, pattern).await
    }

    /// Drive both a write and a read in a single
    /// `IbvBackend::submit` batch — both ops target the same peer
    /// QP and so resolve to a single `ProcessOps` group. After the
    /// batch completes, side B's `write_dst` and side A's `read_dst`
    /// both contain their respective patterns.
    async fn run_multi_op_same_qp(
        env: &TestEnv,
        dev_a: BufferDevice,
        dev_b: BufferDevice,
        size: usize,
        timeout_secs: u64,
    ) -> Result<(), anyhow::Error> {
        const WRITE_PATTERN: u8 = 0xa1;
        const READ_PATTERN: u8 = 0xb2;
        let write_src = env
            .helper_a
            .allocate(&env.client, size, dev_a, WRITE_PATTERN)
            .await?;
        let write_dst = env.helper_b.allocate(&env.client, size, dev_b, 0).await?;
        let read_dst = env.helper_a.allocate(&env.client, size, dev_a, 0).await?;
        let read_src = env
            .helper_b
            .allocate(&env.client, size, dev_b, READ_PATTERN)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: write_src,
                        remote_buf: write_dst.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::ReadIntoLocal,
                        local_buf: read_dst.clone(),
                        remote_buf: read_src,
                    },
                ],
                timeout_secs,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_b, &env.client, write_dst, size, WRITE_PATTERN).await?;
        assert_remote_pattern(&env.helper_a, &env.client, read_dst, size, READ_PATTERN).await
    }

    /// Drive a write + read between two buffers registered with the
    /// *same* `RdmaManagerActor` on the *same* device. Exercises the
    /// loopback path (`is_loopback = true`) where the active actor
    /// connects its QP to its own endpoint and skips the
    /// `CreatePeerQueuePair` round trip.
    async fn run_true_loopback(
        env: &TestEnv,
        dev: BufferDevice,
        size: usize,
    ) -> Result<(), anyhow::Error> {
        const PATTERN: u8 = 0x5d;
        let src = env
            .helper_a
            .allocate(&env.client, size, dev, PATTERN)
            .await?;
        let dst = env.helper_a.allocate(&env.client, size, dev, 0).await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: src,
                    remote_buf: dst.clone(),
                }],
                5,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_a, &env.client, dst, size, PATTERN).await
    }

    // ====================================================================
    // Helpers
    // ====================================================================

    fn require_rdma() {
        if list_all_devices().is_empty() {
            panic!("SKIPPED: no RDMA devices available");
        }
    }

    fn require_cuda() {
        if !crate::is_cuda_available() {
            panic!("SKIPPED: CUDA not available");
        }
    }

    fn make_plan_op(size: usize, pair_count: usize) -> IbvOp<IbvManagerActor<MlxDevice>> {
        let allocation: Box<[u8]> = vec![0; size.max(1)].into_boxed_slice();
        let local_memory =
            KeepaliveLocalMemory::try_new(Arc::new(allocation)).expect("valid CPU allocation");
        let mut remote_buffers = Vec::with_capacity(pair_count);
        for pair in 0..pair_count {
            let device = format!("mlx5_{pair}");
            let mut local = IbvMemoryRegionView::for_test(&device, pair as u32);
            local.virtual_addr = local_memory.addr();
            local.rdma_addr = local_memory.addr();
            local.size = local_memory.size();
            local_memory
                .install_mr::<MlxDevice>(local)
                .expect("install test registration");
            remote_buffers.push(IbvRemoteMemoryRegionView {
                rkey: pair as u32,
                addr: 0x1000,
                size,
                device_name: device,
            });
        }
        let remote_manager = ActorRef::attest(
            "manager.proc@inproc://0"
                .parse::<ActorAddr>()
                .expect("valid test actor address"),
        );
        IbvOp {
            op_type: RdmaOpType::ReadIntoLocal,
            local_memory,
            remote_buffers,
            remote_manager,
        }
    }

    fn plan_ranges(
        size: usize,
        pair_count: usize,
        min_stripe_kb: usize,
        qps_per_peer: usize,
    ) -> Vec<(usize, usize)> {
        let op = make_plan_op(size, pair_count);
        let local_addr = op.local_memory.addr();
        let lock = hyperactor_config::global::lock();
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(min_stripe_kb)
                .expect("minimum stripe size is non-zero"),
        );
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(qps_per_peer).expect("qps_per_peer is non-zero"),
        );
        let mut ranges: Vec<_> = router
            .plan_ops::<MlxDevice>(0, &op)
            .expect("plan test operation")
            .into_iter()
            .map(|op| (op.local.virtual_addr - local_addr, op.local.size))
            .collect();
        ranges.sort_unstable();
        ranges
    }

    #[test]
    fn plan_ops_balances_stripes_without_an_undersized_tail() {
        assert_eq!(plan_ranges(511 * KIB, 4, 512, 1), vec![(0, 511 * KIB)]);
        assert_eq!(plan_ranges(513 * KIB, 4, 512, 1), vec![(0, 513 * KIB)]);
        assert_eq!(
            plan_ranges(12 * KIB, 3, 1, 1),
            vec![(0, 4 * KIB), (4 * KIB, 4 * KIB), (8 * KIB, 4 * KIB)]
        );
        assert_eq!(
            plan_ranges(15 * KIB, 8, 4, 1),
            vec![(0, 5 * KIB), (5 * KIB, 5 * KIB), (10 * KIB, 5 * KIB)]
        );
        assert_eq!(
            plan_ranges(10 * KIB, 3, 1, 1),
            vec![(0, 3392), (3392, 3392), (6784, 3456)]
        );
        assert_eq!(
            plan_ranges(32_000_000, 4, 8192, 1),
            vec![
                (0, 10_666_624),
                (10_666_624, 10_666_624),
                (21_333_248, 10_666_752),
            ]
        );
        assert_eq!(plan_ranges(0, 4, 1, 1), vec![(0, 0)]);
    }

    #[test]
    fn plan_ops_stripes_across_queue_pairs() {
        let even =
            |count: usize, size: usize| (0..count).map(|i| (i * size, size)).collect::<Vec<_>>();
        // One pair with four queue pairs takes one stripe per queue pair.
        assert_eq!(plan_ranges(4 * KIB, 1, 1, 4), even(4, KIB));
        // The minimum stripe size still caps the count: 2 KiB at a 1 KiB
        // minimum makes two stripes, not four.
        assert_eq!(plan_ranges(2 * KIB, 1, 1, 4), even(2, KIB));
        // A larger transfer on one pair still makes four stripes, not sixteen.
        assert_eq!(plan_ranges(16 * KIB, 1, 1, 4), even(4, 4 * KIB));
        // Two pairs make four stripes, two per pair.
        assert_eq!(plan_ranges(16 * KIB, 2, 1, 4), even(4, 4 * KIB));
        // Three pairs round four stripes up to six, two per pair.
        assert_eq!(plan_ranges(12 * KIB, 3, 1, 4), even(6, 2 * KIB));
        // With at least as many pairs as queue pairs, each pair takes one stripe.
        assert_eq!(plan_ranges(16 * KIB, 8, 1, 4), even(8, 2 * KIB));
        // Room for three stripes on two pairs rounds down to two, one per pair.
        assert_eq!(plan_ranges(3 * KIB, 2, 1, 4), even(2, 1536));
        // A transfer below the minimum stays one stripe.
        assert_eq!(plan_ranges(KIB - 1, 2, 1, 4), vec![(0, KIB - 1)]);
    }

    /// Plans `op` and returns each stripe's `(device, qp_index)` in stripe order.
    fn plan_queue_pairs(
        router: &QueuePairRouter,
        op_idx: usize,
        op: &IbvOp<IbvManagerActor<MlxDevice>>,
    ) -> Vec<(String, usize)> {
        let mut planned = router
            .plan_ops::<MlxDevice>(op_idx, op)
            .expect("plan test operation");
        planned.sort_unstable_by_key(|planned| planned.stripe_id.stripe_idx);
        planned
            .into_iter()
            .map(|planned| (planned.local.device_name, planned.qp_index))
            .collect()
    }

    #[test]
    fn plan_ops_spreads_one_op_over_distinct_queue_pairs() {
        let lock = hyperactor_config::global::lock();
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(1).expect("1 is non-zero"),
        );

        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(3).expect("3 is non-zero"),
        );
        let qp_indices: Vec<usize> = plan_queue_pairs(&router, 0, &make_plan_op(3 * KIB, 1))
            .into_iter()
            .map(|(_, qp_index)| qp_index)
            .collect();
        assert_eq!(qp_indices, vec![0, 1, 2], "one stripe per queue pair");

        // Three pairs with four queue pairs: two stripes per pair, on
        // different queue pairs.
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(4).expect("4 is non-zero"),
        );
        assert_eq!(
            queue_pairs_by_device(&router, &make_plan_op(12 * KIB, 3)),
            vec![vec![0, 1]; 3]
        );

        // Room for three stripes on two pairs: one stripe per pair, so neither
        // pair carries two thirds of the transfer.
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(4).expect("4 is non-zero"),
        );
        assert_eq!(
            queue_pairs_by_device(&router, &make_plan_op(3 * KIB, 2)),
            vec![vec![0]; 2]
        );
    }

    #[test]
    fn plan_ops_assigns_stripes_to_pairs_in_name_order() {
        let lock = hyperactor_config::global::lock();
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(1).expect("1 is non-zero"),
        );
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(4).expect("4 is non-zero"),
        );

        // Every op of this size takes the same pair for the same byte range.
        for op_idx in 0..3 {
            let devices: Vec<String> = plan_queue_pairs(&router, op_idx, &make_plan_op(4 * KIB, 2))
                .into_iter()
                .map(|(device, _)| device)
                .collect();
            assert_eq!(devices, vec!["mlx5_0", "mlx5_1", "mlx5_0", "mlx5_1"]);
        }
    }

    /// Plans `op` and returns each device's queue-pair indices in stripe
    /// order, with devices sorted by name.
    fn queue_pairs_by_device(
        router: &QueuePairRouter,
        op: &IbvOp<IbvManagerActor<MlxDevice>>,
    ) -> Vec<Vec<usize>> {
        let mut by_device: BTreeMap<String, Vec<usize>> = BTreeMap::new();
        for (device, qp_index) in plan_queue_pairs(router, 0, op) {
            by_device.entry(device).or_default().push(qp_index);
        }
        by_device.into_values().collect()
    }

    #[test]
    fn plan_ops_continues_the_rotation_across_ops() {
        let lock = hyperactor_config::global::lock();
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(1).expect("1 is non-zero"),
        );
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(3).expect("3 is non-zero"),
        );
        let op = make_plan_op(2 * KIB, 1);

        // Two stripes per op on a route with three queue pairs.
        let qp_indices: Vec<Vec<usize>> = (0..3)
            .map(|op_idx| {
                plan_queue_pairs(&router, op_idx, &op)
                    .into_iter()
                    .map(|(_, qp_index)| qp_index)
                    .collect()
            })
            .collect();
        assert_eq!(qp_indices, vec![vec![0, 1], vec![2, 0], vec![1, 2]]);
    }

    #[test]
    fn queue_pair_indices_advance_independently_per_route() {
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(3).expect("3 is non-zero"),
        );
        let other_id = "manager.proc@inproc://0"
            .parse::<ActorAddr>()
            .expect("valid test actor address")
            .id()
            .clone();
        let route_a = QpRoute {
            self_device: "mlx5_0".into(),
            other_id: other_id.clone(),
            other_device: "mlx5_1".into(),
        };
        let route_b = QpRoute {
            self_device: "mlx5_2".into(),
            other_id,
            other_device: "mlx5_3".into(),
        };

        assert_eq!(
            (0..4)
                .map(|_| router.next_qp_index(route_a.clone()))
                .collect::<Vec<_>>(),
            vec![0, 1, 2, 0],
        );
        assert_eq!(router.next_qp_index(route_b), 0);
        assert_eq!(router.next_qp_index(route_a), 1);
    }

    #[test]
    fn plan_ops_round_robins_queue_pair_indices() {
        let router = QueuePairRouter::new(
            PeerDeviceAffinityPolicy::Any,
            NonZeroUsize::new(3).expect("3 is non-zero"),
        );
        let op = make_plan_op(1, 1);

        let qp_indices = (0..4)
            .map(|op_idx| {
                let planned = router
                    .plan_ops::<MlxDevice>(op_idx, &op)
                    .expect("plan test operation");
                let [planned] = planned.as_slice() else {
                    panic!("one NIC pair should produce one stripe");
                };
                planned.qp_index
            })
            .collect::<Vec<_>>();

        assert_eq!(qp_indices, vec![0, 1, 2, 0]);
    }

    // ====================================================================
    // Tests
    // ====================================================================

    /// `register_remote_buffer` must record its registration on the
    /// `KeepaliveLocalMemory` it is handed — under the name of the device it
    /// registered on — so that later `resolve_local_mrs` calls for that device
    /// reuse it instead of registering the same region again.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_host_buffers_use_per_allocation_mrs_by_default() -> Result<(), anyhow::Error> {
        require_rdma();
        let lock = hyperactor_config::global::lock();
        let _guard = lock.override_key(crate::config::RDMA_HOST_ODP, false);
        let target = IbvDeviceTarget::cpu(0);
        let device = resolve_target::<MlxDevice>(&target)?
            .expect("cpu:0 should resolve to a NIC")
            .name()
            .clone();
        let env = TestEnv::same_config(IbvConfig::targeting(target)).await?;
        let buf: Box<[u8]> = vec![0u8; 1024].into_boxed_slice();
        let local = KeepaliveLocalMemory::try_new(Arc::new(buf))?;
        assert!(
            local.registered_mr::<MlxDevice>(&device)?.is_none(),
            "the region should have no registration before it is registered",
        );
        RdmaManagerActor::local_handle(&env.client)
            .request_buffer(&env.client, local.clone())
            .await?;
        let first = local
            .registered_mr::<MlxDevice>(&device)?
            .expect("registration should be recorded under the pinned device");

        let second_buf: Box<[u8]> = vec![0u8; 1024].into_boxed_slice();
        let second_local = KeepaliveLocalMemory::try_new(Arc::new(second_buf))?;
        RdmaManagerActor::local_handle(&env.client)
            .request_buffer(&env.client, second_local.clone())
            .await?;
        let second = second_local
            .registered_mr::<MlxDevice>(&device)?
            .expect("second registration should be recorded under the pinned device");
        assert_ne!(
            (first.lkey, first.rkey),
            (second.lkey, second.rkey),
            "the default path should retain per-allocation registrations"
        );
        env.shutdown().await
    }

    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_host_odp_shares_a_key_and_advises_destinations() -> Result<(), anyhow::Error> {
        require_rdma();
        let lock = hyperactor_config::global::lock();
        let _guard = lock.override_key(crate::config::RDMA_HOST_ODP, true);
        let target = IbvDeviceTarget::cpu(0);
        let device = resolve_target::<MlxDevice>(&target)?
            .expect("cpu:0 should resolve to a NIC")
            .name()
            .clone();
        let env = TestEnv::same_config(IbvConfig::targeting(target)).await?;

        let first_buf: Box<[u8]> = vec![0u8; 1024].into_boxed_slice();
        let first_local = KeepaliveLocalMemory::try_new(Arc::new(first_buf))?;
        RdmaManagerActor::local_handle(&env.client)
            .request_buffer(&env.client, first_local.clone())
            .await?;
        let first = first_local
            .registered_mr::<MlxDevice>(&device)?
            .expect("first ODP registration should be recorded");

        let second_buf: Box<[u8]> = vec![0u8; 1024].into_boxed_slice();
        let second_local = KeepaliveLocalMemory::try_new(Arc::new(second_buf))?;
        RdmaManagerActor::local_handle(&env.client)
            .request_buffer(&env.client, second_local.clone())
            .await?;
        let second = second_local
            .registered_mr::<MlxDevice>(&device)?
            .expect("second ODP registration should be recorded");

        assert_eq!(
            (first.lkey, first.rkey),
            (second.lkey, second.rkey),
            "host buffers on one device should share its implicit ODP MR"
        );
        run_cross_actor_write(
            &env,
            BufferDevice::Cpu,
            BufferDevice::Cpu,
            1024 * 1024,
            0xa5,
            10,
        )
        .await?;
        run_cross_actor_read(
            &env,
            BufferDevice::Cpu,
            BufferDevice::Cpu,
            1024 * 1024,
            0x5a,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// With `rdma_max_nics_per_buffer` above 1, a buffer is registered on that
    /// many of its tied-for-best NICs. Under the `match_name` policy, a remote
    /// buffer must always be registered on the first N devices returned by
    /// `select_optimal_ibv_devices` -- otherwise, a mismatch could make a transfer
    /// impossible.
    #[timed_test::async_timed_test(timeout_secs = 120)]
    async fn test_match_name_nic_selection() -> Result<(), anyhow::Error> {
        require_rdma();
        const MAX_NICS: usize = 4;
        let lock = hyperactor_config::global::lock();
        let _max_guard = lock.override_key(
            crate::config::RDMA_MAX_NICS_PER_BUFFER,
            Some(hyperactor_config::NonZeroUsize::new(MAX_NICS).expect("MAX_NICS is 4")),
        );
        let _policy_guard = lock.override_key(
            crate::config::RDMA_PEER_DEVICE_AFFINITY,
            "match_name".to_string(),
        );
        let expected: BTreeSet<String> =
            select_optimal_ibv_devices::<MlxDevice>(MemoryLocation::Cpu(None))?
                .iter()
                .take(MAX_NICS)
                .map(|nic| nic.name().clone())
                .collect();

        let env = TestEnv::same_config(IbvConfig::default()).await?;
        let buffer = env
            .helper_a
            .allocate(&env.client, 32, BufferDevice::Cpu, 0)
            .await?;
        let served_by: BTreeSet<String> = buffer
            .resolve_mlx()
            .expect("the buffer is registered on a Mellanox NIC")
            .buffers
            .iter()
            .map(|mr| mr.device_name.clone())
            .collect();
        assert_eq!(served_by, expected);

        for pattern in 0..2 * MAX_NICS as u8 {
            run_cross_actor_write(&env, BufferDevice::Cpu, BufferDevice::Cpu, 32, pattern, 5)
                .await?;
        }
        env.shutdown().await
    }

    #[timed_test::async_timed_test(timeout_secs = 120)]
    async fn test_striped_transfers_land_every_byte() -> Result<(), anyhow::Error> {
        require_rdma();
        const MAX_NICS: usize = 4;
        const MIN_STRIPE_KB: usize = 512;
        // The extra 17 bytes make the size prime, so the stripes split unevenly.
        const SIZE: usize = MAX_NICS * MIN_STRIPE_KB * KIB + 17;
        let tied = select_optimal_ibv_devices::<MlxDevice>(MemoryLocation::Cpu(None))?.len();
        if tied < 2 {
            panic!("SKIPPED: this host has fewer than two NICs tied for host memory");
        }

        let lock = hyperactor_config::global::lock();
        let _max_guard = lock.override_key(
            crate::config::RDMA_MAX_NICS_PER_BUFFER,
            Some(hyperactor_config::NonZeroUsize::new(MAX_NICS).expect("MAX_NICS is non-zero")),
        );
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(MIN_STRIPE_KB).expect("MIN_STRIPE_KB is non-zero"),
        );
        let _policy_guard =
            lock.override_key(crate::config::RDMA_PEER_DEVICE_AFFINITY, "any".to_string());

        let env = TestEnv::same_config(IbvConfig::default()).await?;
        run_positional_write_and_read(&env, SIZE, 20).await?;
        env.shutdown().await
    }

    /// With a single NIC pair, a transfer is split across that pair's queue pairs.
    #[timed_test::async_timed_test(timeout_secs = 120)]
    async fn test_queue_pair_stripes_land_every_byte() -> Result<(), anyhow::Error> {
        require_rdma();
        const QPS_PER_PEER: usize = 4;
        const MIN_STRIPE_KB: usize = 512;
        // The extra 17 bytes make the size prime, so the stripes split unevenly.
        const SIZE: usize = QPS_PER_PEER * MIN_STRIPE_KB * KIB + 17;

        let lock = hyperactor_config::global::lock();
        let _qps_guard = lock.override_key(
            crate::config::RDMA_QPS_PER_PEER,
            hyperactor_config::NonZeroUsize::new(QPS_PER_PEER).expect("QPS_PER_PEER is non-zero"),
        );
        let _stripe_guard = lock.override_key(
            crate::config::RDMA_MIN_STRIPE_SIZE_KB,
            hyperactor_config::NonZeroUsize::new(MIN_STRIPE_KB).expect("MIN_STRIPE_KB is non-zero"),
        );

        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        run_positional_write_and_read(&env, SIZE, 20).await?;
        env.shutdown().await
    }

    /// Under `any` a buffer starts at a NIC drawn at random, so at one NIC per
    /// buffer the buffers spread over the tied NICs instead of all landing on
    /// the first.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_any_nic_selection() -> Result<(), anyhow::Error> {
        require_rdma();
        let lock = hyperactor_config::global::lock();
        let _max_guard = lock.override_key(
            crate::config::RDMA_MAX_NICS_PER_BUFFER,
            Some(hyperactor_config::NonZeroUsize::MIN),
        );
        let _policy_guard =
            lock.override_key(crate::config::RDMA_PEER_DEVICE_AFFINITY, "any".to_string());
        let tied = select_optimal_ibv_devices::<MlxDevice>(MemoryLocation::Cpu(None))?.len();

        let env = TestEnv::same_config(IbvConfig::default()).await?;
        let mut served_by: BTreeSet<String> = BTreeSet::new();
        for _ in 0..16 {
            let buffer = env
                .helper_a
                .allocate(&env.client, 32, BufferDevice::Cpu, 0)
                .await?;
            let registrations = buffer
                .resolve_mlx()
                .expect("the buffer is registered on a Mellanox NIC")
                .buffers;
            let [mr] = registrations.as_slice() else {
                panic!("one NIC serves a buffer here, got {registrations:?}");
            };
            served_by.insert(mr.device_name.clone());
        }
        // Sixteen buffers all drawing the same one of several NICs would be a
        // one-in-billions coincidence.
        assert_eq!(
            served_by.len() > 1,
            tied > 1,
            "{tied} NICs tie for host memory, but the buffers landed on {served_by:?}",
        );
        env.shutdown().await
    }

    /// Cross-actor RDMA write over a single device (both sides target
    /// `cpu:0`). The two `RdmaManagerActor`s differ — this exercises
    /// the asymmetric `CreatePeerQueuePair` handshake even though the
    /// underlying device is shared.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cross_actor_same_device_write() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        run_cross_actor_write(&env, BufferDevice::Cpu, BufferDevice::Cpu, 32, 0xa5, 5).await?;
        env.shutdown().await
    }

    /// Cross-actor RDMA read over a single device.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cross_actor_same_device_read() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        run_cross_actor_read(&env, BufferDevice::Cpu, BufferDevice::Cpu, 32, 0x3c, 5).await?;
        env.shutdown().await
    }

    /// True loopback write: both buffers registered with the same
    /// `RdmaManagerActor` on the same device. The `QueuePairActor`
    /// sees `is_loopback = true` and connects its QP to its own
    /// endpoint without going through `CreatePeerQueuePair`.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_loopback_write() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        run_true_loopback(&env, BufferDevice::Cpu, 32).await?;
        env.shutdown().await
    }

    /// Cross-device write (cpu:0 → cpu:1).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cross_device_write() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::cpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::cpu(1)),
        )
        .await?;
        run_cross_actor_write(&env, BufferDevice::Cpu, BufferDevice::Cpu, 32, 0x77, 5).await?;
        env.shutdown().await
    }

    /// Cross-device read (cpu:0 ← cpu:1).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cross_device_read() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::cpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::cpu(1)),
        )
        .await?;
        run_cross_actor_read(&env, BufferDevice::Cpu, BufferDevice::Cpu, 32, 0x88, 5).await?;
        env.shutdown().await
    }

    /// One write + one read in a single `IbvBackend::submit` batch.
    /// Both ops share the same `QpKey` so the manager groups them
    /// into a single `ProcessOps` dispatched to one `QueuePairActor`.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_multi_op_same_qp_cpu() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        run_multi_op_same_qp(&env, BufferDevice::Cpu, BufferDevice::Cpu, 64, 5).await?;
        env.shutdown().await
    }

    /// Same as `test_multi_op_same_qp_cpu` but with 2 MiB CUDA buffers,
    /// pulled apart into a separate test because the buffer-size +
    /// device split is the only thing that differs.
    #[timed_test::async_timed_test(timeout_secs = 120)]
    async fn test_multi_op_same_qp_cuda() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::gpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::gpu(1)),
        )
        .await?;
        run_multi_op_same_qp(&env, BufferDevice::Cuda(0), BufferDevice::Cuda(1), SIZE, 10).await?;
        env.shutdown().await
    }

    /// CUDA → CPU write.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cuda_to_cpu_write() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::gpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::cpu(1)),
        )
        .await?;
        run_cross_actor_write(
            &env,
            BufferDevice::Cuda(0),
            BufferDevice::Cpu,
            SIZE,
            0x9b,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CUDA → CPU read (source is the CUDA side).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cuda_to_cpu_read() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::cpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::gpu(1)),
        )
        .await?;
        run_cross_actor_read(
            &env,
            BufferDevice::Cpu,
            BufferDevice::Cuda(1),
            SIZE,
            0x37,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CPU → CUDA write.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cpu_to_cuda_write() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::cpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::gpu(1)),
        )
        .await?;
        run_cross_actor_write(
            &env,
            BufferDevice::Cpu,
            BufferDevice::Cuda(1),
            SIZE,
            0x5a,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CPU → CUDA read (source is the CPU side).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cpu_to_cuda_read() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::gpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::cpu(1)),
        )
        .await?;
        run_cross_actor_read(
            &env,
            BufferDevice::Cuda(0),
            BufferDevice::Cpu,
            SIZE,
            0x4e,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CUDA → CUDA write.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cuda_to_cuda_write() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::gpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::gpu(1)),
        )
        .await?;
        run_cross_actor_write(
            &env,
            BufferDevice::Cuda(0),
            BufferDevice::Cuda(1),
            SIZE,
            0xee,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CUDA → CUDA read.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_cuda_to_cuda_read() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;
        let env = TestEnv::new(
            IbvConfig::targeting(IbvDeviceTarget::gpu(0)),
            IbvConfig::targeting(IbvDeviceTarget::gpu(1)),
        )
        .await?;
        run_cross_actor_read(
            &env,
            BufferDevice::Cuda(0),
            BufferDevice::Cuda(1),
            SIZE,
            0x42,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// CUDA buffers with `IbvQpType::Standard` (no mlx5dv).
    /// Exercises the per-buffer dmabuf MR-registration path: without
    /// mlx5dv the manager cannot use indirect mkeys via segment
    /// scanning and instead registers each buffer as a standalone
    /// dmabuf MR (`ibv_reg_dmabuf_mr`).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_standard_qp_cuda_dmabuf_fallback() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 16 * 1024 * 1024;
        let mut config_a = IbvConfig::targeting(IbvDeviceTarget::gpu(0));
        config_a.qp_type = IbvQpType::Standard;
        let mut config_b = IbvConfig::targeting(IbvDeviceTarget::gpu(1));
        config_b.qp_type = IbvQpType::Standard;
        let env = TestEnv::new(config_a, config_b).await?;
        run_cross_actor_write(
            &env,
            BufferDevice::Cuda(0),
            BufferDevice::Cuda(1),
            SIZE,
            0x33,
            10,
        )
        .await?;
        env.shutdown().await
    }

    /// Two `IbvBackend::submit` calls back-to-back through the same
    /// helper. The second batch reuses the cached `QueuePairActor`
    /// from the first (the manager's `qp_handles` entry persists
    /// across submits).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_multi_batch_same_qp() -> Result<(), anyhow::Error> {
        require_rdma();
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        let src1 = env
            .helper_a
            .allocate(&env.client, 32, BufferDevice::Cpu, 0xa1)
            .await?;
        let dst1 = env
            .helper_b
            .allocate(&env.client, 32, BufferDevice::Cpu, 0)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: src1,
                    remote_buf: dst1.clone(),
                }],
                5,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_b, &env.client, dst1, 32, 0xa1).await?;

        let src2 = env
            .helper_a
            .allocate(&env.client, 32, BufferDevice::Cpu, 0xb2)
            .await?;
        let dst2 = env
            .helper_b
            .allocate(&env.client, 32, BufferDevice::Cpu, 0)
            .await?;
        env.helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: src2,
                    remote_buf: dst2.clone(),
                }],
                5,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;
        assert_remote_pattern(&env.helper_b, &env.client, dst2, 32, 0xb2).await?;
        env.shutdown().await
    }

    /// Single submit batch with ops landing on multiple `QpKey`
    /// groups: loopback (helper_a → helper_a), cross-actor cpu↔cpu,
    /// cpu↔cuda, cuda↔cpu, cuda↔cuda. Exercises the manager's
    /// per-QP slicing and concurrent multi-QP dispatch.
    #[timed_test::async_timed_test(timeout_secs = 120)]
    async fn test_multi_op_multi_qp() -> Result<(), anyhow::Error> {
        require_rdma();
        require_cuda();
        const SIZE: usize = 2 * 1024 * 1024;

        let env = TestEnv::same_config(IbvConfig::default()).await?;

        const LOOPBACK_PAT: u8 = 0x11;
        let lb_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, LOOPBACK_PAT)
            .await?;
        let lb_dst = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;

        const CC_PAT: u8 = 0x22;
        let cc_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, CC_PAT)
            .await?;
        let cc_dst = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;

        const CG_PAT: u8 = 0x33;
        let cg_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, CG_PAT)
            .await?;
        let cg_dst = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cuda(1), 0)
            .await?;

        const GC_PAT: u8 = 0x44;
        let gc_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cuda(0), GC_PAT)
            .await?;
        let gc_dst = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;

        const GG_PAT: u8 = 0x55;
        let gg_src = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cuda(1), GG_PAT)
            .await?;
        let gg_dst = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cuda(0), 0)
            .await?;

        env.helper_a
            .submit(
                &env.client,
                vec![
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: lb_src,
                        remote_buf: lb_dst.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: cc_src,
                        remote_buf: cc_dst.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: cg_src,
                        remote_buf: cg_dst.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: gc_src,
                        remote_buf: gc_dst.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::ReadIntoLocal,
                        local_buf: gg_dst.clone(),
                        remote_buf: gg_src,
                    },
                ],
                30,
            )
            .await?
            .map_err(|e| anyhow::anyhow!(e))?;

        assert_remote_pattern(&env.helper_a, &env.client, lb_dst, SIZE, LOOPBACK_PAT).await?;
        assert_remote_pattern(&env.helper_b, &env.client, cc_dst, SIZE, CC_PAT).await?;
        assert_remote_pattern(&env.helper_b, &env.client, cg_dst, SIZE, CG_PAT).await?;
        assert_remote_pattern(&env.helper_b, &env.client, gc_dst, SIZE, GC_PAT).await?;
        assert_remote_pattern(&env.helper_a, &env.client, gg_dst, SIZE, GG_PAT).await?;

        env.shutdown().await
    }

    /// Force the timeout branch in `IbvBackend::submit`. A near-zero
    /// timeout fires before any per-op replies arrive, so the
    /// aggregated error reports the timeout terminal cause.
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_submit_timeout() -> Result<(), anyhow::Error> {
        require_rdma();
        const SIZE: usize = 1024 * 1024;
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;
        let src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0x77)
            .await?;
        let dst = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;
        let result = env
            .helper_a
            .submit(
                &env.client,
                vec![BufferHelperOp {
                    op_type: RdmaOpType::WriteFromLocal,
                    local_buf: src,
                    remote_buf: dst,
                }],
                0,
            )
            .await?;
        let err = result.expect_err("expected submit to time out");
        assert!(
            err.contains("submit timed out"),
            "unexpected error message: {err}",
        );
        env.shutdown().await
    }

    /// Submit a batch with a bogus op in the middle. RC
    /// completions fire in posting order, so the good op before it
    /// completes normally, the bogus op fails with `REM_ACCESS_ERR`
    /// (it puts the QP into error state), and the good op after it
    /// gets flushed with `WC_WR_FLUSH_ERR`. Verifies (a) op 0 is
    /// absent from the aggregated error and its bytes transferred,
    /// (b) ops 1 and 2 both appear in the error, and (c) op 2's
    /// destination was *not* written (the flush meant nothing was
    /// transferred).
    #[timed_test::async_timed_test(timeout_secs = 60)]
    async fn test_partial_failure_batch() -> Result<(), anyhow::Error> {
        require_rdma();
        let lock = hyperactor_config::global::lock();
        let _guard = lock.override_key(
            crate::config::RDMA_QPS_PER_PEER,
            hyperactor_config::NonZeroUsize::new(1).expect("1 is non-zero"),
        );
        const SIZE: usize = 32;
        let env = TestEnv::same_config(IbvConfig::targeting(IbvDeviceTarget::cpu(0))).await?;

        const GOOD_PAT: u8 = 0xc3;
        const POST_FLUSH_PAT: u8 = 0xde;

        let good_src_0 = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, GOOD_PAT)
            .await?;
        let good_dst_0 = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;

        let bogus_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0xee)
            .await?;
        let real_remote = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;
        let mut bogus_remote = real_remote.clone();
        let bufs = [
            bogus_remote
                .backends
                .mlx
                .as_mut()
                .map(|ctx| &mut ctx.buffers),
            bogus_remote
                .backends
                .efa
                .as_mut()
                .map(|ctx| &mut ctx.buffers),
        ];
        for buf in bufs.into_iter().flatten().flatten() {
            buf.rkey = 0xdead_beef;
            buf.addr = 0xdead_0000;
        }

        let post_flush_src = env
            .helper_a
            .allocate(&env.client, SIZE, BufferDevice::Cpu, POST_FLUSH_PAT)
            .await?;
        let post_flush_dst = env
            .helper_b
            .allocate(&env.client, SIZE, BufferDevice::Cpu, 0)
            .await?;

        let result = env
            .helper_a
            .submit(
                &env.client,
                vec![
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: good_src_0,
                        remote_buf: good_dst_0.clone(),
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: bogus_src,
                        remote_buf: bogus_remote,
                    },
                    BufferHelperOp {
                        op_type: RdmaOpType::WriteFromLocal,
                        local_buf: post_flush_src,
                        remote_buf: post_flush_dst.clone(),
                    },
                ],
                10,
            )
            .await?;
        let err = result.expect_err("expected at least one op to fail");
        let rem_access = format!(
            "status={:?}",
            rdmaxcel_sys::ibv_wc_status::IBV_WC_REM_ACCESS_ERR,
        );
        let wr_flush = format!(
            "status={:?}",
            rdmaxcel_sys::ibv_wc_status::IBV_WC_WR_FLUSH_ERR,
        );
        /// Returns the stripe failure lines an error reports under `op
        /// {op_idx}:`.
        fn op_block(err: &str, op_idx: usize) -> &str {
            let header = format!("\n  op {op_idx}:\n");
            let start = err
                .find(&header)
                .unwrap_or_else(|| panic!("expected op {op_idx} in error: {err}"));
            let block = &err[start + header.len()..];
            &block[..block.find("\n  op ").unwrap_or(block.len())]
        }

        assert!(
            !err.contains("op 0:"),
            "op 0 should not appear in error: {err}",
        );
        let op1 = op_block(&err, 1);
        assert!(
            op1.lines().all(|line| line.starts_with("    stripe ")),
            "expected op 1 to list one line per failed stripe: {op1}",
        );
        assert!(
            op1.contains("completion failed") && op1.contains(&rem_access),
            "expected op 1 to fail with REM_ACCESS_ERR: {op1}",
        );
        let op2 = op_block(&err, 2);
        assert!(
            op2.contains("completion failed") && op2.contains(&wr_flush),
            "expected op 2 to be flushed with WR_FLUSH_ERR: {op2}",
        );

        assert_remote_pattern(&env.helper_b, &env.client, good_dst_0, SIZE, GOOD_PAT).await?;
        // The flushed op was never transferred; destination stays zero.
        assert_remote_pattern(&env.helper_b, &env.client, post_flush_dst, SIZE, 0).await?;

        env.shutdown().await
    }
}