epics-base-rs 0.28.2

//! External-link OUT-write queue — the Rust counterpart of `dbCa`'s
//! `workList` (`modules/database/src/ioc/db/dbCa.c`).
//!
//! # The C model this mirrors
//!
//! C never performs a Channel Access network write inside `dbScanLock`.
//! `dbCaPutLink` (`dbCa.c:598-602`) forwards to `dbCaPutLinkCallback`
//! (`dbCa.c:515-596`), which — under the per-link `pca->lock`, *not* the
//! record lock —
//!
//! 1. refuses the put outright when the link is not connected or has no
//!    write access, returning `-1` with nothing staged (`dbCa.c:529-532`);
//! 2. converts the value into the link's single pending-value buffer
//!    `pca->pputNative` / `pca->pputString` (`dbCa.c:533-584`);
//! 3. records the delivery flavour — `CA_PUT` for a plain put,
//!    `CA_PUT_CALLBACK` plus the completion callback for
//!    `dbCaPutLinkCallback` (`dbCa.c:585-592`);
//! 4. calls `addAction(pca, CA_WRITE_NATIVE)` (`dbCa.c:130-159`), which ORs
//!    the action bit, appends the link to `workList` **once**
//!    (`callAdd = (pca->link_action == 0)`, `dbCa.c:135`, `:153-155`) and
//!    signals `workListEvent`;
//! 5. **returns**.
//!
//! The network write happens later on the `dbCaTask` thread
//! (`dbCa.c:1093-1260`): it pops one link, clears `link_action` *before*
//! doing the work (`dbCa.c:1130`) so a put arriving mid-flight re-queues the
//! link, and issues `ca_array_put` (plain) or `ca_array_put_callback`
//! (completion flavour) at `dbCa.c:1161-1183`.
//!
//! Three consequences this module reproduces verbatim:
//!
//! * **Latest-wins per link, not a per-put queue.** `pputNative` is one
//!   buffer. A second put staged before the task services the link
//!   overwrites the first and only bumps a counter
//!   (`if (pca->newOutNative) pca->nNoWrite++;`, `dbCa.c:582-583`; the
//!   string twin at `:548-549`). There is therefore no "queue full" state
//!   in C to match — the bound is one pending value per link, times the
//!   number of links, and that is the bound here too.
//! * **The link is enqueued at most once.** `addAction`'s `callAdd` gate
//!   (`dbCa.c:135`) appends the `caLink` to `workList` only when it carries
//!   no pending action; a put on an already-queued link just ORs the bit.
//! * **The returned status is the *staging* status, never the network
//!   status.** `dbCaPutLinkCallback` returns the conversion status (or the
//!   `-1` refusal); the outcome of the wire write is reported only through
//!   `errlogPrintf` on the task (`dbCa.c:1175-1179`) or, for the completion
//!   flavour, through `putComplete` (`dbCa.c:994-1012`). C's
//!   `dbPutLink` → `setLinkAlarm` gate (`dbLink.c:434-448`) therefore alarms
//!   on *refusal to stage*, not on a failed transmission.
//!
//! # What this module owns
//!
//! One [`LinkPutQueue`] per [`super::PvDatabase`]. A record-processing
//! thread stages a write and returns; a single owner task drains the queue
//! and performs the actual `LinkSet::put_value` / `LinkSet::flush_puts`
//! away from the record's advisory write gate.
//!
//! **Invariant.** *Every staged completion is resolved exactly once* —
//! by the owner (with the lset's result), by a superseding put, or by the
//! queue's teardown. [`PutCompletion`] enforces it by construction: its
//! `Drop` resolves an unresolved completion, so no path — early return,
//! `?`, panic, or the whole database dropping — can drop a pending put
//! silently.

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::Weak;

use super::link_set::{DynLinkSet, LinkPutOp};
use crate::types::EpicsValue;

/// Which lset a staged write is bound to.
///
/// `Scheme(s)` is a `pva://`/`ca://` prefixed link — the write goes to that
/// one lset. `Any` is the bare-name form (the body stored in
/// `ParsedLink::Ca`/`Pva` once `record/link.rs` has stripped the scheme):
/// every registered lset is tried in turn, first to accept wins, which is
/// what the pre-queue `write_external_pv` did.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub(crate) enum LinkTarget {
    Scheme(String),
    Any,
}

/// Identity of one external link: the lset it belongs to plus the name that
/// lset is addressed with. The unit of latest-wins coalescing, exactly as
/// `caLink` is in C.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub(crate) struct LinkKey {
    pub(crate) target: LinkTarget,
    pub(crate) name: String,
}

/// The completion half of a staged put — the `pca->putCallback` slot
/// (`dbCaPvt.h:83`, set at `dbCa.c:585-588`, consumed by `putComplete` at
/// `:994-1011`).
///
/// Unlike C, which simply overwrites `putCallback` when a second
/// `dbCaPutLinkCallback` lands before the first completes — dropping the
/// earlier callback with no notification (`dbCa.c:585-592`) — a
/// `PutCompletion` cannot be lost: `Drop` resolves it. That is strictly
/// stronger than C and is what makes "no pending put loses its
/// classification silently" hold by construction rather than by review.
struct PutCompletion(Option<tokio::sync::oneshot::Sender<Result<(), String>>>);

impl PutCompletion {
    /// Resolve with the outcome the owner (or a superseding put) determined.
    fn resolve(mut self, result: Result<(), String>) {
        if let Some(tx) = self.0.take() {
            let _ = tx.send(result);
        }
    }
}

impl Drop for PutCompletion {
    fn drop(&mut self) {
        if let Some(tx) = self.0.take() {
            let _ = tx.send(Err(
                "external link put dropped before completion (link queue torn down)".to_string(),
            ));
        }
    }
}

/// One link's pending value — `pca->pputNative` plus the `putType` /
/// `putCallback` pair that names its flavour (`dbCaPvt.h:81-84`).
struct StagedPut {
    value: EpicsValue,
    op: LinkPutOp,
    /// `Some` only for [`LinkPutOp::Async`] — C's `CA_PUT_CALLBACK`.
    completion: Option<PutCompletion>,
}

/// Per-link state. Membership of the ready deque is exactly `Queued`, so the
/// deque cannot disagree with the map: the illegal combinations
/// ("in the deque but in flight", "staged but nobody will pick it up") are
/// not representable.
enum LinkState {
    /// Staged and waiting for the owner. In the ready deque exactly once.
    Queued(StagedPut),
    /// The owner took the value; a network op is running. Not in the deque.
    InFlight,
    /// A network op is running AND a newer value was staged behind it.
    /// Not in the deque; `finish` moves it back to `Queued` and enqueues.
    InFlightRestaged(StagedPut),
}

/// Per-link OPEN state — C's `CA_CONNECT` action, staged once per link by
/// `dbCaAddLink` (`dbCa.c:397-401`) and serviced on the `dbCaTask`
/// (`ca_create_channel` + `ca_add_array_event`).
///
/// `Done` is terminal on purpose: C stages exactly one `CA_CONNECT` per link
/// and libca owns every reconnection attempt from then on. Re-staging on each
/// cache miss would turn a down remote into one connect attempt per record
/// scan.
#[derive(PartialEq, Eq, Debug)]
enum OpenState {
    Queued,
    InFlight,
    Done,
}

#[derive(Default)]
struct QueueInner {
    links: HashMap<LinkKey, LinkState>,
    ready: VecDeque<LinkKey>,
    opens: HashMap<LinkKey, OpenState>,
    ready_opens: VecDeque<LinkKey>,
    /// Per-link count of staged values a later put on the same link
    /// overwrote — C's `pca->nNoWrite` (`dbCa.c:548`, `:582`), which lives on
    /// the `caLink` and is therefore per link, not per IOC. Keyed rather than
    /// summed because `dbcar` prints it beside each link's name
    /// (`dbCaTest.c:111-116`) and a whole-queue total cannot be attributed
    /// back to one of them.
    ///
    /// Never pruned: C's counter lives as long as the `caLink`, whereas an
    /// entry in `links` exists only while a write is outstanding, so pruning
    /// with the state would reset the count every time the link went idle.
    coalesced: HashMap<LinkKey, u64>,
    /// Number of writes the owner has completed. Diagnostic.
    completed: u64,
    /// Number of link opens the owner has completed. Diagnostic.
    opened: u64,
}

impl QueueInner {
    fn is_idle(&self) -> bool {
        self.links.is_empty()
            && !self
                .opens
                .values()
                .any(|s| matches!(s, OpenState::Queued | OpenState::InFlight))
    }
}

/// The queue itself. Lives in `PvDatabaseInner`; one owner task drains it.
pub(crate) struct LinkPutQueue {
    inner: parking_lot::Mutex<QueueInner>,
    /// Wakes the owner when a link becomes `Queued`. Held by the owner as an
    /// `Arc` so it can park on the signal without holding the queue itself
    /// alive — the queue must be free to drop when the database does.
    work: Arc<tokio::sync::Notify>,
    /// Wakes [`LinkPutQueue::sync`] waiters when the queue goes idle — the
    /// `dbCaSync` barrier (`dbCa.c:1126-1129`, `CA_SYNC`).
    idle: tokio::sync::Notify,
    /// Set once the owner task has been spawned.
    owner_started: std::sync::atomic::AtomicBool,
    /// The reactor the network half of this queue needs, captured where one is
    /// known to exist — construction time, which for an IOC is inside
    /// `IocApplication::run` and for a test is inside its `#[tokio::test]`.
    ///
    /// The queue is the one place on the record-processing chain that cannot
    /// live on the background executor: `LinkSet::connect_link` for a `ca://`
    /// target builds a `CaClient`, whose transport manager and search engine
    /// own `tokio::net` sockets. Staging, by contrast, happens on whatever
    /// thread processed the record — often a blocking connection thread with
    /// no runtime — so the capability has to be carried here rather than read
    /// off the staging thread. `None` means no runtime existed at all when the
    /// database was built, and then an external link cannot work under any
    /// arrangement — so it is refused rather than dispatched somewhere that
    /// would only move the panic; see [`LinkPutQueue::network`]. A purely
    /// local `LinkSet` is unaffected: it never reaches this queue.
    reactor: Option<crate::runtime::task::BlockingBridge>,
}

impl Default for LinkPutQueue {
    fn default() -> Self {
        Self {
            inner: parking_lot::Mutex::new(QueueInner::default()),
            work: Arc::new(tokio::sync::Notify::new()),
            idle: tokio::sync::Notify::new(),
            owner_started: std::sync::atomic::AtomicBool::new(false),
            reactor: crate::runtime::task::BlockingBridge::try_capture(),
        }
    }
}

impl Drop for LinkPutQueue {
    fn drop(&mut self) {
        // Teardown resolves every staged completion. `PutCompletion::drop`
        // does the resolving, so clearing the map is enough — and stays
        // correct if a future state variant carries a completion somewhere
        // this function would forget to visit.
        self.inner.get_mut().links.clear();
        // Wake the owner so it observes the dead `Weak` and exits instead of
        // parking on a signal nobody will ever send again.
        self.work.notify_waiters();
    }
}

impl LinkPutQueue {
    /// The reactor every network future dispatched by this queue runs on —
    /// the single owner of "this future may touch a socket", so no other site
    /// in the queue has to know which executor it is on.
    ///
    /// `None` is the no-runtime-anywhere state described on
    /// [`Self::reactor`], and it is a *refusal*, never a fallback: there is no
    /// second executor that could do the work. The process-global background
    /// executor cannot — `runtime::task::spawn_background` documents that it
    /// deliberately has no `tokio::net` reactor — so sending a `ca://` open or
    /// put there only moves the panic onto a callback-pool worker. Callers
    /// refuse instead; [`super::PvDatabase::external_put_gate`] turns that
    /// into the record's LINK alarm before anything is staged.
    pub(crate) fn network(&self) -> Option<&crate::runtime::task::BlockingBridge> {
        self.reactor.as_ref()
    }

    /// Stage `value` on `key`, coalescing latest-wins onto any pending value
    /// (C `dbCa.c:582-583`). Returns the completion receiver for an
    /// [`LinkPutOp::Async`] put — the caller awaits it, which is the
    /// `dbCaPutLinkCallback` route; [`LinkPutOp::Plain`] gets `None` and the
    /// caller returns immediately, which is `dbCaPutLink`.
    pub(crate) fn stage_put(
        &self,
        key: LinkKey,
        value: EpicsValue,
        op: LinkPutOp,
    ) -> Option<tokio::sync::oneshot::Receiver<Result<(), String>>> {
        let (completion, rx) = match op {
            LinkPutOp::Plain => (None, None),
            LinkPutOp::Async => {
                let (tx, rx) = tokio::sync::oneshot::channel();
                (Some(PutCompletion(Some(tx))), Some(rx))
            }
        };
        let staged = StagedPut {
            value,
            op,
            completion,
        };
        let wake = {
            let mut inner = self.inner.lock();
            match inner.links.remove(&key) {
                None => {
                    inner.links.insert(key.clone(), LinkState::Queued(staged));
                    inner.ready.push_back(key);
                    true
                }
                Some(LinkState::Queued(prev)) => {
                    Self::supersede(&mut inner, &key, prev);
                    // Already in the ready deque — C's `callAdd` is false
                    // here too (`dbCa.c:135`), so no second enqueue.
                    inner.links.insert(key, LinkState::Queued(staged));
                    false
                }
                Some(LinkState::InFlight) => {
                    inner.links.insert(key, LinkState::InFlightRestaged(staged));
                    false
                }
                Some(LinkState::InFlightRestaged(prev)) => {
                    Self::supersede(&mut inner, &key, prev);
                    inner.links.insert(key, LinkState::InFlightRestaged(staged));
                    false
                }
            }
        };
        if wake {
            self.work.notify_one();
        }
        rx
    }

    /// Resolve a displaced value's completion — the only place a staged put
    /// is superseded, so the `nNoWrite` accounting has one owner.
    fn supersede(inner: &mut QueueInner, key: &LinkKey, prev: StagedPut) {
        *inner.coalesced.entry(key.clone()).or_default() += 1;
        if let Some(c) = prev.completion {
            c.resolve(Err(
                "external link put superseded by a newer put on the same link".to_string(),
            ));
        }
    }

    /// Owner-only: take the next ready link, moving it `Queued` → `InFlight`.
    /// Mirrors `dbCaTask` popping `workList` and clearing `link_action`
    /// before doing the work (`dbCa.c:1115-1132`).
    fn take_ready(&self) -> Option<(LinkKey, StagedPut)> {
        let mut inner = self.inner.lock();
        loop {
            let key = inner.ready.pop_front()?;
            match inner.links.remove(&key) {
                Some(LinkState::Queued(staged)) => {
                    inner.links.insert(key.clone(), LinkState::InFlight);
                    return Some((key, staged));
                }
                // Only `Queued` is ever in the deque; restore anything else
                // rather than silently dropping it.
                Some(other) => {
                    inner.links.insert(key, other);
                }
                None => {}
            }
        }
    }

    /// Owner-only: the write for `key` finished. `InFlight` → idle;
    /// `InFlightRestaged` → `Queued` and back in the deque, which is how a
    /// put that landed mid-flight is not lost (C clears `link_action` before
    /// the wire write for the same reason, `dbCa.c:1130`).
    fn finish(&self, key: LinkKey) {
        let (wake_work, wake_idle) = {
            let mut inner = self.inner.lock();
            inner.completed += 1;
            let wake_work = match inner.links.remove(&key) {
                Some(LinkState::InFlight) | None => false,
                Some(LinkState::InFlightRestaged(staged)) => {
                    inner.links.insert(key.clone(), LinkState::Queued(staged));
                    inner.ready.push_back(key);
                    true
                }
                // Cannot happen: only `finish` and `take_ready` move a key
                // out of `InFlight*`, and both run on the owner path.
                Some(other) => {
                    inner.links.insert(key, other);
                    false
                }
            };
            (wake_work, inner.is_idle())
        };
        if wake_work {
            self.work.notify_one();
        }
        if wake_idle {
            self.idle.notify_waiters();
        }
    }

    /// Stage the OPEN of `key` — C `dbCaAddLink` (`dbCa.c:397-401`)
    /// delegates to `dbCaAddLinkCallback` (`:373-395`), whose one
    /// `addAction(pca, CA_CONNECT)` is `dbCa.c:393`. The checkout
    /// `R7.0.10-146-g8f5015b66` routes this through
    /// `dbCaAddLinkCallbackOpt` instead, which arrived with the iocInit
    /// connection-wait (`717d69e1f`, post-`R7.0.10` and in no tag) and does
    /// not exist at the pin. See `PvDatabase::external_link_targets`.
    /// Idempotent and
    /// terminal: a link that has ever been staged is never staged again, so a
    /// record scanning at 10 Hz against a down remote produces one connect
    /// attempt, not ten per second. Reconnection is the lset's job, as it is
    /// libca's in C.
    ///
    /// Returns true when this call is the one that staged it.
    pub(crate) fn stage_open(&self, key: LinkKey) -> bool {
        let staged = {
            let mut inner = self.inner.lock();
            if inner.opens.contains_key(&key) {
                false
            } else {
                inner.opens.insert(key.clone(), OpenState::Queued);
                inner.ready_opens.push_back(key);
                true
            }
        };
        if staged {
            self.work.notify_one();
        }
        staged
    }

    /// Owner-only: take the next link to open, `Queued` → `InFlight`.
    fn take_ready_open(&self) -> Option<LinkKey> {
        let mut inner = self.inner.lock();
        loop {
            let key = inner.ready_opens.pop_front()?;
            match inner.opens.get_mut(&key) {
                Some(state @ OpenState::Queued) => {
                    *state = OpenState::InFlight;
                    return Some(key);
                }
                // Only `Queued` is ever in the deque.
                _ => continue,
            }
        }
    }

    /// Owner-only: the open for `key` returned. `InFlight` → `Done`.
    fn finish_open(&self, key: LinkKey) {
        let wake_idle = {
            let mut inner = self.inner.lock();
            inner.opened += 1;
            inner.opens.insert(key, OpenState::Done);
            inner.is_idle()
        };
        if wake_idle {
            self.idle.notify_waiters();
        }
    }

    /// True when nothing is staged and nothing is in flight.
    pub(crate) fn is_idle(&self) -> bool {
        self.inner.lock().is_idle()
    }

    /// Number of link opens the owner has completed. Diagnostic.
    pub(crate) fn opened_count(&self) -> u64 {
        self.inner.lock().opened
    }

    /// Number of staged values a later put on the same link overwrote,
    /// summed over every link — C's `pca->nNoWrite` (`dbCa.c:548`, `:582`)
    /// aggregated the way `dbcar`'s trailing `(%lu disconnects, %lu writes
    /// prohibited)` total does (`dbCaTest.c:153-154`).
    pub(crate) fn coalesced_count(&self) -> u64 {
        self.inner.lock().coalesced.values().sum()
    }

    /// The same count for one link — the per-`caLink` `pca->nNoWrite`
    /// `dbcar` prints beside the link's name.
    pub(crate) fn coalesced_for(&self, key: &LinkKey) -> u64 {
        self.inner.lock().coalesced.get(key).copied().unwrap_or(0)
    }

    /// Number of writes the owner has completed.
    pub(crate) fn completed_count(&self) -> u64 {
        self.inner.lock().completed
    }

    /// Block until the queue is idle — the `dbCaSync` barrier
    /// (`dbCa.c:1126-1129`: a `CA_SYNC` action the task answers only once it
    /// has drained everything ahead of it). Tests use it to make an enqueued
    /// put observable; `iocPause`-class callers use it to quiesce.
    pub(crate) async fn sync(&self) {
        loop {
            let notified = self.idle.notified();
            if self.is_idle() {
                return;
            }
            notified.await;
        }
    }

    /// Start the single owner task, once. `db` is weak so the task cannot
    /// keep the database alive; it exits when the database is dropped.
    ///
    /// Reached from the staging path, which record processing enters on
    /// whatever thread processed the record, so the loop itself goes to the
    /// background executor. The network work it dispatches goes to
    /// [`Self::reactor`] instead — see that field.
    pub(crate) fn ensure_owner(self: &Arc<Self>, db: Weak<super::PvDatabaseInner>) {
        if self
            .owner_started
            .swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            return;
        }
        let queue = Arc::downgrade(self);
        let work = self.work.clone();
        // Middle band: one owner loop drains the queued link puts of every
        // record, so there is no single record whose PRIO it could take.
        crate::runtime::task::spawn_background(
            crate::runtime::task::CallbackPriority::Medium,
            owner_loop(queue, work, db),
        );
    }
}

/// The single consumer — `dbCaTask`'s loop (`dbCa.c:1105-1260`).
///
/// Per dequeued link it resolves the lset and spawns the network write,
/// keeping at most one write in flight per link (so per-link write ordering
/// is preserved) while distinct links proceed independently. C gets that for
/// free because `ca_array_put` only queues into libca's send buffer; ours has
/// to spawn because `LinkSet::put_value` for `pva://` is a full PUT round
/// trip, and one slow target must not stall every other link's writes.
async fn owner_loop(
    queue: Weak<LinkPutQueue>,
    work: Arc<tokio::sync::Notify>,
    db: Weak<super::PvDatabaseInner>,
) {
    // C `dbCaTask` registers with the watchdog before its loop
    // (`dbCa.c:1096`) and removes when told to exit (`:1262`). Unbounded: the
    // loop parks on `work` whenever no link has anything staged, which is an
    // IOC with quiet output links rather than a fault.
    let _watched = crate::runtime::taskwd::taskwd_insert(
        "dbCaLink",
        crate::runtime::taskwd::CheckIn::Unbounded,
        None,
    );
    loop {
        // Register on the wake signal BEFORE draining, so a put staged
        // during the drain cannot be missed.
        let notified = work.notified();
        tokio::pin!(notified);
        notified.as_mut().enable();
        {
            let Some(q) = queue.upgrade() else { return };
            match q.network() {
                Some(reactor) => {
                    // Opens first: a pending open is what a blocked cached read
                    // is waiting on, and C services `CA_CONNECT` on the same
                    // task ahead of that link's other actions
                    // (`dbCa.c:1133-1160`).
                    while let Some(key) = q.take_ready_open() {
                        let Some(inner) = db.upgrade() else { return };
                        let lsets = resolve_lsets(&inner, &key.target);
                        drop(inner);
                        let q2 = q.clone();
                        reactor.spawn(async move {
                            for lset in &lsets {
                                lset.connect_link(&key.name).await;
                            }
                            q2.finish_open(key);
                        });
                    }
                    while let Some((key, staged)) = q.take_ready() {
                        let Some(inner) = db.upgrade() else { return };
                        let lsets = resolve_lsets(&inner, &key.target);
                        drop(inner);
                        let q2 = q.clone();
                        reactor.spawn(async move {
                            run_put(&lsets, &key, staged).await;
                            q2.finish(key);
                        });
                    }
                }
                // No reactor anywhere. Refuse what is here rather than
                // dispatch it — see [`LinkPutQueue::network`]. Reachable only
                // through a direct `stage_put`/`stage_open`, because
                // `external_put_gate` refuses this database's external puts
                // before they are staged; draining here so a caller that
                // bypassed the gate still gets its completion resolved rather
                // than left pending until teardown.
                None => {
                    while let Some(key) = q.take_ready_open() {
                        q.finish_open(key);
                    }
                    while let Some((key, staged)) = q.take_ready() {
                        refuse_for_want_of_a_reactor(&key, staged);
                        q.finish(key);
                    }
                }
            }
            // The strong reference must not survive the await below, or the
            // queue could never drop and its teardown would never run.
        }
        notified.await;
    }
}

/// The lsets a write (or an open) may be offered to, in order. Single owner
/// of the target → lset mapping, shared with `PvDatabase::resolve_external_pv`
/// so a link's cached read and its staged open address the same lsets.
pub(super) fn resolve_lsets(
    inner: &super::PvDatabaseInner,
    target: &LinkTarget,
) -> Vec<DynLinkSet> {
    let registry = inner.link_sets.load();
    match target {
        LinkTarget::Scheme(s) => registry.get(s).into_iter().collect(),
        LinkTarget::Any => registry
            .schemes()
            .iter()
            .filter_map(|s| registry.get(s))
            .collect(),
    }
}

/// Resolve a staged write that no owner can perform, because the database
/// captured no reactor ([`LinkPutQueue::network`]).
///
/// The failure is reported the way C reports a put it could not deliver — on
/// the task, through `errlogPrintf` (`db/dbCa.c:1111-1114`
/// (`dbCaTask`); epics-base R7.0.10) — and the completion is resolved so the
/// put-notify chain unblocks with an error instead of waiting for a wire write
/// that will never be attempted.
fn refuse_for_want_of_a_reactor(key: &LinkKey, staged: StagedPut) {
    let msg = format!(
        "external link put to '{}' refused: this database was built with no \
         tokio runtime, so no link set can reach the network",
        key.name
    );
    eprintln!("dbCa: {msg}");
    if let Some(c) = staged.completion {
        c.resolve(Err(msg));
    }
}

/// Perform one dequeued write. The network suspension lives here — on the
/// owner's spawned task, never on a record-processing thread.
async fn run_put(lsets: &[DynLinkSet], key: &LinkKey, staged: StagedPut) {
    let StagedPut {
        value,
        op,
        completion,
    } = staged;
    let mut result = Err(format!(
        "no link set accepted the write to external link '{}'",
        key.name
    ));
    for lset in lsets {
        match lset.put_value(&key.name, value.clone(), op).await {
            Ok(()) => {
                // Production drain of any retry-queued OUT writes
                // on this lset now that a write has reached it (the
                // channel may have just reconnected). pvxs replays
                // the queued put from record processing, not
                // test code (`ioc/pvalink_channel.cpp:220-280`).
                lset.flush_puts().await;
                result = Ok(());
                break;
            }
            Err(e) => result = Err(e),
        }
    }
    // C reports a failed wire write on the task, via `errlogPrintf`
    // (`dbCa.c:1175-1179`) — for BOTH flavours: the `status != ECA_NORMAL`
    // arm sits below the `CA_PUT` / `CA_PUT_CALLBACK` fork and covers each
    // (`dbCa.c:1161-1183`). The record has already returned; its alarm came
    // from the staging gate, exactly as `dbCaPutLink`'s `-1` does.
    if let Err(e) = &result {
        eprintln!("dbCa: external link put to '{}' failed: {e}", key.name);
    }
    if let Some(c) = completion {
        c.resolve(result);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::future::Future;
    use std::task::{Context, Poll, Waker};

    fn key(name: &str) -> LinkKey {
        LinkKey {
            target: LinkTarget::Scheme("ca".to_string()),
            name: name.to_string(),
        }
    }

    /// Read a completion out of its receiver with no runtime at all.
    ///
    /// The resolutions these tests assert — supersession and teardown — are
    /// performed *synchronously*, inside `stage_put` and inside `Drop`, so
    /// the receiver is `Ready` on its first poll and no reactor is involved.
    /// Polling by hand rather than `.await`ing says exactly that: `Pending`
    /// means the completion was left unresolved, which is the invariant
    /// under test, and it fails here instead of hanging. It also keeps both
    /// tests running on the exec backend and adds no site to this file's
    /// RTEMS-EXEC-MODEL-ALLOW census.
    fn resolved(rx: tokio::sync::oneshot::Receiver<Result<(), String>>) -> Result<(), String> {
        let mut rx = std::pin::pin!(rx);
        match rx.as_mut().poll(&mut Context::from_waker(Waker::noop())) {
            Poll::Ready(out) => out.expect("completion resolved, not dropped"),
            Poll::Pending => panic!("completion was left unresolved"),
        }
    }

    /// Boundary: two plain puts staged on ONE link before the owner picks it
    /// up. C keeps a single `pputNative` buffer, so the second overwrites the
    /// first and only bumps `nNoWrite` (`dbCa.c:582-583`) — latest-wins, and
    /// the link is enqueued once (`callAdd = (pca->link_action == 0)`,
    /// `dbCa.c:135`). This is C's answer to "what happens when the queue is
    /// full": there is no queue to fill.
    #[test]
    fn second_put_on_same_link_coalesces_latest_wins() {
        let q = LinkPutQueue::default();
        assert!(
            q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain)
                .is_none()
        );
        assert!(
            q.stage_put(key("A"), EpicsValue::Long(2), LinkPutOp::Plain)
                .is_none()
        );

        assert_eq!(q.coalesced_count(), 1, "one put must be coalesced away");
        let (k, staged) = q.take_ready().expect("one ready link");
        assert_eq!(k, key("A"));
        assert_eq!(staged.value, EpicsValue::Long(2), "latest value wins");
        assert!(
            q.take_ready().is_none(),
            "the link is enqueued exactly once"
        );
    }

    /// Boundary: the count is attributed to the LINK that lost the value,
    /// not to the queue. C keeps `nNoWrite` on the `caLink` (`dbCa.c:548`,
    /// `:582`) and `dbcar` prints it beside that link's name, so a second
    /// link's supersessions must not show up against the first.
    #[test]
    fn the_coalesce_count_is_attributed_per_link() {
        let q = LinkPutQueue::default();
        for v in [1, 2, 3] {
            q.stage_put(key("A"), EpicsValue::Long(v), LinkPutOp::Plain);
        }
        q.stage_put(key("B"), EpicsValue::Long(9), LinkPutOp::Plain);

        assert_eq!(q.coalesced_for(&key("A")), 2);
        assert_eq!(q.coalesced_for(&key("B")), 0);
        // Never staged at all — C's `pca == NULL`, zero rather than absent.
        assert_eq!(q.coalesced_for(&key("C")), 0);
        assert_eq!(q.coalesced_count(), 2, "the total is still the sum");
    }

    /// Boundary: the per-link count SURVIVES the link going idle. C's counter
    /// lives on the `caLink`, which outlives any one staged value, so a link
    /// that drained and was written to again must not report a fresh zero.
    #[test]
    fn the_coalesce_count_outlives_the_links_pending_state() {
        let q = LinkPutQueue::default();
        q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain);
        q.stage_put(key("A"), EpicsValue::Long(2), LinkPutOp::Plain);
        assert_eq!(q.coalesced_for(&key("A")), 1);
        let (k, _staged) = q.take_ready().expect("ready");
        q.finish(k);
        assert!(q.is_idle(), "nothing pending on any link");
        assert_eq!(q.coalesced_for(&key("A")), 1);
    }

    /// Boundary: distinct links do NOT coalesce — one ready entry each, in
    /// staging order (`workList` is a FIFO: `ellAdd` / `ellGet`).
    #[test]
    fn distinct_links_keep_their_own_pending_value() {
        let q = LinkPutQueue::default();
        q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain);
        q.stage_put(key("B"), EpicsValue::Long(2), LinkPutOp::Plain);
        assert_eq!(q.coalesced_count(), 0);
        assert_eq!(q.take_ready().expect("A first").0, key("A"));
        assert_eq!(q.take_ready().expect("B second").0, key("B"));
        assert!(q.take_ready().is_none());
    }

    /// Boundary: a put staged while the link is IN FLIGHT is not lost — it
    /// becomes ready again when the in-flight write finishes, and never
    /// overtakes it. C gets this by clearing `link_action` before the wire
    /// write (`dbCa.c:1130`), so a concurrent `addAction` re-appends the link.
    #[test]
    fn put_staged_during_flight_is_requeued_on_finish() {
        let q = LinkPutQueue::default();
        q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain);
        let (k, first) = q.take_ready().expect("ready");
        assert_eq!(first.value, EpicsValue::Long(1));
        assert!(q.take_ready().is_none(), "in flight, nothing ready");

        q.stage_put(key("A"), EpicsValue::Long(9), LinkPutOp::Plain);
        assert!(
            q.take_ready().is_none(),
            "still in flight — at most one write per link keeps write order"
        );

        q.finish(k);
        let (_, second) = q.take_ready().expect("re-queued after finish");
        assert_eq!(second.value, EpicsValue::Long(9));
    }

    /// Boundary: the completion flavour is resolved — never dropped — when a
    /// newer put supersedes it. C overwrites `pca->putCallback`
    /// (`dbCa.c:585-592`) and the earlier callback never fires; we resolve it
    /// with an error instead, so no pending put loses its classification.
    #[test]
    fn superseded_completion_is_resolved_not_dropped() {
        let q = LinkPutQueue::default();
        let rx = q
            .stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Async)
            .expect("Async put yields a completion");
        q.stage_put(key("A"), EpicsValue::Long(2), LinkPutOp::Async);

        assert!(
            resolved(rx).unwrap_err().contains("superseded"),
            "a superseded completion must say so"
        );
    }

    /// Boundary: tearing the queue down resolves every pending completion.
    /// The `Drop` on `PutCompletion` is what makes this hold on *every* exit
    /// path, not just this one.
    #[test]
    fn teardown_resolves_pending_completions() {
        let q = LinkPutQueue::default();
        let rx = q
            .stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Async)
            .expect("Async put yields a completion");
        drop(q);
        assert!(
            resolved(rx).is_err(),
            "a torn-down pending put must not report success"
        );
    }

    /// Boundary: a plain put yields no completion channel at all — it is
    /// fire-and-forget by type, so there is no receiver a caller could
    /// accidentally await. C's `dbCaPutLink` passes a null callback
    /// (`dbCa.c:601`).
    #[test]
    fn plain_put_has_no_completion() {
        let q = LinkPutQueue::default();
        assert!(
            q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain)
                .is_none()
        );
    }

    /// Boundary: idle accounting. The queue is idle before anything is
    /// staged, busy while a write is in flight, idle again after `finish`.
    #[test]
    fn idle_tracks_staged_and_in_flight() {
        let q = LinkPutQueue::default();
        assert!(q.is_idle());
        q.stage_put(key("A"), EpicsValue::Long(1), LinkPutOp::Plain);
        assert!(!q.is_idle(), "staged but not run");
        let (k, _) = q.take_ready().expect("ready");
        assert!(!q.is_idle(), "in flight");
        q.finish(k);
        assert!(q.is_idle());
        assert_eq!(q.completed_count(), 1);
    }
}