Skip to main content

epics_base_rs/server/database/
mod.rs

1pub mod db_access;
2mod field_io;
3pub mod filters;
4mod link_set;
5mod links;
6mod processing;
7mod record_lock;
8mod scan_index;
9
10pub use link_set::{
11    DynLinkSet, LinkDbfType, LinkMetadata, LinkPutOp, LinkSet, LinkSetRegistry, RemoteAlarm,
12};
13pub use processing::{AsyncDbHandle, AsyncToken};
14pub use record_lock::{ManyRecordWriteGuard, RecordWriteGuard};
15
16use crate::error::{CaError, CaResult};
17use crate::runtime::sync::RwLock;
18use std::collections::{BTreeSet, HashMap};
19use std::sync::Arc;
20
21use crate::server::pv::ProcessVariable;
22use crate::server::record::{Record, RecordInstance, ScanType};
23use crate::types::EpicsValue;
24
25/// Parse a PV name into (base_name, field_name).
26/// "TEMP.EGU" → ("TEMP", "EGU")
27/// "TEMP"     → ("TEMP", "VAL")
28pub fn parse_pv_name(name: &str) -> (&str, &str) {
29    match name.rsplit_once('.') {
30        Some((base, field)) => (base, field),
31        None => (name, "VAL"),
32    }
33}
34
35/// Apply timestamp to a record based on its TSE field.
36/// `is_soft` indicates a Soft Channel device type.
37///
38/// Mirrors C `recGblGetTimeStampSimm` (recGbl.c:310-343). The TSE
39/// constants are defined in `epicsTime.h:102-104`:
40///
41///   - `epicsTimeEventCurrentTime = 0` → wall-clock now
42///   - `epicsTimeEventBestTime    = -1` → generalTime BestTime providers
43///   - `epicsTimeEventDeviceTime  = -2` → device support already set time
44///   - `1..` → event-number providers
45///
46/// The C path is symmetric: every non-`-2` case unconditionally
47/// overwrites `precord->time` via `epicsTimeGetEvent(tse)`, which
48/// delegates to `epicsTimeGetCurrent` for `tse==0` and to
49/// `generalTimeGetEventPriority` otherwise. Only `-2` (device time)
50/// is left untouched because the device support has already written
51/// the timestamp before `recGblGetTimeStamp` is called.
52fn apply_timestamp(common: &mut super::record::CommonFields, _is_soft: bool) {
53    // Single owner of TSE -> TIME resolution; device support that must
54    // format the record's resolved time during `read()` routes through the
55    // same helper so the two never drift (see `recgbl::get_time_stamp`).
56    // For TSE=-2 the helper returns `common.time` unchanged, preserving the
57    // device-time "leave it alone" semantics.
58    common.time = crate::server::recgbl::get_time_stamp(common.tse, common.time);
59}
60
61/// Unified entry in the PV database.
62pub enum PvEntry {
63    Simple(Arc<ProcessVariable>),
64    Record(Arc<RwLock<RecordInstance>>),
65}
66
67/// Callback for resolving external PV names (CA/PVA links).
68/// Returns the current value of the external PV, or None if unavailable.
69pub type ExternalPvResolver = Arc<dyn Fn(&str) -> Option<EpicsValue> + Send + Sync>;
70
71/// Async hook invoked by [`PvDatabase::has_name`] when a name is not yet
72/// in the database. Used by the CA gateway and similar proxy components
73/// to lazily populate PVs on first search.
74///
75/// The resolver should:
76/// 1. Determine whether the name should be served (e.g., check ACL)
77/// 2. Take whatever action is needed to make `has_name` return true on
78///    a subsequent call (e.g., subscribe to an upstream IOC and call
79///    `add_pv` with a placeholder value)
80/// 3. Return `true` if the name is now resolvable, `false` otherwise
81///
82/// Returning `true` causes `has_name` to re-check the database. The
83/// resolver may take some time (TCP search, upstream connect handshake);
84/// the caller (UDP search responder, TCP CREATE_CHANNEL handler) will
85/// `.await` it.
86/// The second argument is the downstream client's socket address when
87/// the lookup originates from a CA/PVA search or channel-create on
88/// behalf of an identified peer (`None` for host-less internal lookups:
89/// preload, iocsh, link processing). the CA gateway needs
90/// this to evaluate `.pvlist` `DENY FROM host` rules at search time, the
91/// way C ca-gateway's `pvExistTest` passes the client host to
92/// `gateAs::findEntry`.
93pub type SearchResolver = Arc<
94    dyn Fn(
95            String,
96            Option<std::net::SocketAddr>,
97        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>
98        + Send
99        + Sync,
100>;
101
102/// Per-request admission gate for an **already-registered** simple PV.
103///
104/// A plain IOC's simple PVs are authoritative: once registered they
105/// exist unconditionally, so no gate is installed and the cached-PV
106/// short-circuit in [`PvDatabase::find_entry_from`] /
107/// [`PvDatabase::has_name_from`] is unchanged. A CA gateway is
108/// different — its shadow PVs are projections of an upstream that can be
109/// host-denied for a given requester or disconnected — so it installs a
110/// gate that the lookup path consults *before* returning a cached simple
111/// PV. Returning `false` makes the database answer "does not exist" for
112/// that requester, exactly as C ca-gateway's `pvExistTest` returns
113/// `pverDoesNotExistHere` for a host-denied or disconnected PV
114/// (`gateServer.cc:1516-1637`) — without removing the PV object, so its
115/// cached value stays available for diagnostics and re-admission.
116///
117/// The first argument is the filter-suffix-stripped record path (the
118/// same key the simple-PV map and the gateway cache use); the second is
119/// the requesting peer (`None` for host-less internal lookups). The gate
120/// governs **only** simple PVs — records and aliases are never
121/// gateway-managed and bypass it.
122pub type ExistenceGate = Arc<
123    dyn Fn(
124            String,
125            Option<std::net::SocketAddr>,
126        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>
127        + Send
128        + Sync,
129>;
130
131/// Internal state of [`PvDatabase`].
132///
133/// # Invariant — alias-aware lookup (epics-base PR #336)
134///
135/// **MUST**: every record-name lookup that originates from an
136/// external API (CA/PVA server, link processing, iocsh, bridge
137/// providers) MUST go through [`PvDatabase::get_record`] /
138/// [`PvDatabase::find_entry`] / [`PvDatabase::has_name`], never
139/// `inner.records.read().await.get(...)` directly.
140///
141/// **MUST NOT**: a function that takes an arbitrary record-name
142/// `&str` and reads `inner.records` directly, unless one of:
143/// - the function is itself an alias-management primitive
144///   (`add_record`, `remove_record`, `add_alias`,
145///   `find_entry_no_resolve`, `has_name_no_resolve`,
146///   `get_record_no_resolve`, `all_record_names`), OR
147/// - the name has been normalised to canonical earlier in the
148///   same scope (the `let canonical_owned; let name: &str = ...`
149///   pattern in `process_record_with_links_inner` /
150///   `complete_async_record_inner` / `put_record_field_from_ca` /
151///   `put_pv`).
152///
153/// **Owner/Gate:** `PvDatabase::get_record` (alias-aware path).
154///
155/// New code that adds a record-name entry point should call
156/// `get_record` first OR run the canonical-normalisation snippet
157/// at function entry. Direct `inner.records` access is reserved
158/// for the alias-management primitives listed above.
159/// One CP/CPP edge in the [`PvDatabaseInner::cp_links`] index: the record
160/// to (re)process when the source record changes.
161///
162/// `passive_only` distinguishes CPP from CP. C adds the `CA_DBPROCESS`
163/// action for a CP link unconditionally, but for a CPP link only when the
164/// link-holding record's `SCAN` is Passive (`dbCa.c:854,994,1072`). CP
165/// edges clear the flag; CPP edges set it, and `dispatch_cp_targets`
166/// honours it.
167#[derive(Clone, Debug)]
168pub struct CpTarget {
169    pub record: String,
170    pub passive_only: bool,
171}
172
173struct PvDatabaseInner {
174    simple_pvs: RwLock<HashMap<String, Arc<ProcessVariable>>>,
175    records: RwLock<HashMap<String, Arc<RwLock<RecordInstance>>>>,
176    /// Scan index: maps scan type → sorted set of
177    /// `(PHAS, load_order, record_name)`.
178    ///
179    /// C parity (`dbScan.c:1052-1095`): `buildScanLists` walks records
180    /// in database / record-type **load order** and `addToList`
181    /// inserts each after the last element with `phas <= precord->phas`
182    /// — so within one PHAS value the scan list is a stable FIFO in
183    /// load order. The secondary sort key is the per-record
184    /// `load_order` sequence (NOT the record name), so two records
185    /// sharing a PHAS scan in the order they were loaded, matching a
186    /// C IOC built from the same `.db` file.
187    scan_index: RwLock<HashMap<ScanType, BTreeSet<(i16, u64, String)>>>,
188    /// Per-record load-order sequence number, assigned monotonically
189    /// at `add_record`. Used as the secondary scan-index sort key so
190    /// same-PHAS records preserve database load order. Survives a
191    /// `remove_record` + re-`add_record` (the re-add gets a fresh,
192    /// higher sequence — matching a fresh `.db` reload).
193    load_order: RwLock<HashMap<String, u64>>,
194    /// Monotonic counter feeding `load_order`.
195    load_order_counter: std::sync::atomic::AtomicU64,
196    /// CP/CPP link index: maps source_record → target edges to process when
197    /// the source changes. Each edge carries the CP-vs-CPP distinction (see
198    /// [`CpTarget`]).
199    cp_links: RwLock<HashMap<String, Vec<CpTarget>>>,
200    /// External (CA/PVA) CP/CPP link index: maps the *external PV name*
201    /// (the cross-IOC source, e.g. `OTHER:PV` from `INP="OTHER:PV CP CA"`)
202    /// → holder edges to process when that remote PV changes. The local
203    /// [`Self::cp_links`] index is keyed by a local source RECORD that
204    /// processes here; a cross-IOC source never processes locally, so its
205    /// only trigger is the calink/pvalink CA monitor callback, which calls
206    /// [`PvDatabase::dispatch_external_cp_targets`]. Parity with C
207    /// `dbCa.c:993-994` `eventCallback` adding `CA_DBPROCESS`.
208    external_cp_links: RwLock<HashMap<String, Vec<CpTarget>>>,
209    /// Alias map: alternate-name → real-record-name. Mirrors epics-base
210    /// PR #336 (alias name validation + parsing). `find_entry` and
211    /// related lookups consult this map after the canonical record
212    /// table so an alias resolves transparently to its target.
213    aliases: RwLock<HashMap<String, String>>,
214    /// Single gate that serializes
215    /// every `add_pv` / `add_pv_with_hook` / `add_record` /
216    /// `add_alias` / `remove_record` / `remove_simple_pv` /
217    /// `remove_alias`. Without this, the per-method write-lock
218    /// orders (`simple_pvs` first vs. `records` first vs.
219    /// `aliases` first) could deadlock under concurrent registrations,
220    /// and `add_record`'s post-insert `scan_index.write()` had a
221    /// TOCTOU window where `remove_record` could land between the
222    /// records map insert and the scan-index insert and leave a
223    /// phantom scan entry.
224    ///
225    /// Holding this mutex makes the cross-namespace `check_name_free`
226    /// peek atomic with the target-map insert, eliminates the
227    /// scan-index race, and lets `remove_*` purge dangling aliases
228    /// without a second pass.
229    registration_mutex: tokio::sync::Mutex<()>,
230    /// Lines queued by the iocsh `afterIocRunning <command>` directive
231    /// (epics-base PR #558). Drained by the IOC application after PINI
232    /// completes, then re-executed through a fresh IocShell so the
233    /// commands run with the database in its post-init state.
234    after_ioc_running: std::sync::Mutex<Vec<String>>,
235    /// Optional resolver for external PVs (ca://, pva:// links).
236    external_resolver: RwLock<Option<ExternalPvResolver>>,
237    /// Optional async resolver invoked on `has_name` misses (e.g. CA gateway).
238    search_resolver: RwLock<Option<SearchResolver>>,
239    /// Optional per-request gate consulted before a *cached* simple PV is
240    /// advertised as existing (e.g. CA gateway host/state admission). See
241    /// [`ExistenceGate`]. `None` for a plain IOC (short-circuit unchanged).
242    existence_gate: RwLock<Option<ExistenceGate>>,
243    /// Per-scheme link sets — pluggable backends for `pva://` /
244    /// `ca://` link resolution. Consulted before the legacy
245    /// [`ExternalPvResolver`] in [`Self::resolve_external_pv`].
246    /// Mirrors the C-EPICS lset abstraction.
247    link_sets: RwLock<link_set::LinkSetRegistry>,
248    /// True once the ScanScheduler has been started for this DB.
249    /// Prevents duplicate scan tasks when multiple protocol servers (CA + PVA)
250    /// both try to start scanning on the same DB.
251    scan_started: std::sync::atomic::AtomicBool,
252    /// True once PINI processing has completed. Non-owner schedulers await
253    /// this before running their hooks, preserving the "PINI before hooks"
254    /// ordering contract.
255    pini_done: std::sync::atomic::AtomicBool,
256    /// Fired by the scan owner after PINI completes. Non-owners register
257    /// interest on this before re-checking `pini_done` to avoid missing the
258    /// signal (`notify_waiters` does not store a permit).
259    pini_notify: tokio::sync::Notify,
260    /// Per-record advisory write gates — the Rust
261    /// counterpart of the C-EPICS `dbScanLock` / `dbLocker`
262    /// machinery. Every plain CA/PVA write, the QSRV atomic group
263    /// PUT/GET, and the pvalink atomic scan-on-update epoch all
264    /// acquire these gates, so no two of them can interleave on a
265    /// shared record. See [`record_lock`].
266    record_locks: record_lock::RecordLockRegistry,
267}
268
269/// Database of all process variables hosted by this server.
270#[derive(Clone)]
271pub struct PvDatabase {
272    inner: Arc<PvDatabaseInner>,
273}
274
275/// Which record kind a SELM link selection is being computed for.
276/// The Specified/Mask base differs between record types in C, so the
277/// shared selector must know the caller.
278#[derive(Clone, Copy, PartialEq, Eq, Debug)]
279pub(crate) enum SelmKind {
280    /// `fanout` / `seq`: Specified index is `SELN + OFFS` (0-based over
281    /// LNK0..LNKF / group 0..15); Mask is shifted by `SHFT`.
282    /// Mirrors `fanoutRecord.c:106-141` and `seqRecord.c:147-178`.
283    FanoutSeq,
284    /// `dfanout`: Specified index is `SELN - 1` (1-based, `SELN==0`
285    /// means "drive nothing", `SELN > OUT_ARG_MAX` is invalid); Mask
286    /// has NO `SHFT` and `SELN==0` means "no output".
287    /// Mirrors `dfanoutRecord.c:307-339`.
288    Dfanout,
289}
290
291/// Result of resolving a SELM/SELN selection.
292#[derive(Clone, Debug, Default)]
293pub(crate) struct SelmResult {
294    /// 0-based link indices to drive (into the LNK0../OUTA.. array).
295    pub indices: Vec<usize>,
296    /// `Some` when C would raise an alarm for an out-of-range
297    /// `SELN`/`OFFS`/`SHFT`. C uses `recGblSetSevr(prec, SOFT_ALARM,
298    /// INVALID_ALARM)` in every such path.
299    pub alarm: Option<(u16, crate::server::record::AlarmSeverity)>,
300}
301
302/// Convert a link value to `epicsUInt16` with C `dbGetLink(.., DBR_USHORT,
303/// ..)` cast semantics. The dbConvert GET macro stores `*pdst =
304/// (epicsUInt16) *psrc` (`dbConvert.c:63-70`) — a C cast that truncates
305/// toward zero then wraps modulo 2^16, NOT a clamp. So `-1` becomes
306/// `65535` and `65536` becomes `0`. Used for fanout/dfanout/seq
307/// `SELL`→`SELN` so a constant, DB, CA, or PVA link source all convert
308/// by the one rule C applies through `dbFastGetConvertRoutine`.
309pub(crate) fn dbr_ushort_cast(value: &EpicsValue) -> u16 {
310    (value.to_f64().unwrap_or(0.0) as i64) as u16
311}
312
313/// Select which link indices are active based on SELM/SELN, applying
314/// the record-type-specific `OFFS`/`SHFT` bias.
315///
316/// SELM: 0 = All, 1 = Specified, 2 = Mask. `count` is the number of
317/// link slots (16 for fanout/dfanout/seq).
318///
319/// `seln` is the native `DBF_USHORT` value: C declares `SELN` as
320/// `epicsUInt16`, so every comparison below is unsigned, matching C's
321/// selection arithmetic. A `SELL=-1` link read therefore selects
322/// `65535` (out of range → INVALID for Specified, all-bits for Mask),
323/// not `-1`.
324///
325/// C references:
326/// * fanout — `fanoutRecord.c:106-141`
327/// * dfanout — `dfanoutRecord.c:307-339`
328/// * seq — `seqRecord.c:147-178`
329pub(crate) fn select_link_indices_ex(
330    kind: SelmKind,
331    selm: i16,
332    seln: u16,
333    offs: i16,
334    shft: i16,
335    count: usize,
336) -> SelmResult {
337    use crate::server::recgbl::alarm_status::SOFT_ALARM;
338    use crate::server::record::AlarmSeverity;
339
340    let invalid = || SelmResult {
341        indices: Vec::new(),
342        alarm: Some((SOFT_ALARM, AlarmSeverity::Invalid)),
343    };
344    let ok = |indices: Vec<usize>| SelmResult {
345        indices,
346        alarm: None,
347    };
348
349    match selm {
350        // All — every slot.
351        0 => ok((0..count).collect()),
352        // Specified.
353        1 => match kind {
354            SelmKind::FanoutSeq => {
355                // C: `i = seln + offs;` with `seln` unsigned (epicsUInt16),
356                // 0-based; `i<0 || i>=NLINKS` → INVALID. So `SELN=65535`
357                // (from `SELL=-1`) yields `i>=NLINKS` → INVALID, never
358                // drives link 0.
359                let i = seln as i32 + offs as i32;
360                if i < 0 || i >= count as i32 {
361                    invalid()
362                } else {
363                    ok(vec![i as usize])
364                }
365            }
366            SelmKind::Dfanout => {
367                // C `dfanoutRecord.c:315-320`: `if (prec->seln > OUT_ARG_MAX)`
368                // with `seln` unsigned → INVALID; `seln == 0` → no output;
369                // otherwise drive `seln - 1`. OFFS is not a dfanout field.
370                // `SELL=-1` → `SELN=65535` > count → INVALID (the signed
371                // read used to see `-1`, take the `<= 0` branch, and drive
372                // nothing with no alarm).
373                let seln_i = seln as i32;
374                if seln_i > count as i32 {
375                    invalid()
376                } else if seln_i == 0 {
377                    ok(Vec::new())
378                } else {
379                    ok(vec![(seln_i - 1) as usize])
380                }
381            }
382        },
383        // Mask.
384        2 => {
385            let mask: u32 = match kind {
386                SelmKind::FanoutSeq => {
387                    // C: SHFT shift first, with `shft` range-checked to [-15,15].
388                    if !(-15..=15).contains(&shft) {
389                        return invalid();
390                    }
391                    let raw = seln as u32;
392                    if shft >= 0 {
393                        raw >> shft
394                    } else {
395                        raw << (-shft)
396                    }
397                }
398                // dfanout Mask has no SHFT.
399                SelmKind::Dfanout => seln as u32,
400            };
401            ok((0..count).filter(|i| mask & (1 << i) != 0).collect())
402        }
403        // Any other SELM value → C `default:` raises INVALID.
404        _ => invalid(),
405    }
406}
407
408impl PvDatabase {
409    pub fn new() -> Self {
410        Self {
411            inner: Arc::new(PvDatabaseInner {
412                simple_pvs: RwLock::new(HashMap::new()),
413                external_resolver: RwLock::new(None),
414                search_resolver: RwLock::new(None),
415                existence_gate: RwLock::new(None),
416                link_sets: RwLock::new(link_set::LinkSetRegistry::new()),
417                records: RwLock::new(HashMap::new()),
418                scan_index: RwLock::new(HashMap::new()),
419                load_order: RwLock::new(HashMap::new()),
420                load_order_counter: std::sync::atomic::AtomicU64::new(0),
421                cp_links: RwLock::new(HashMap::new()),
422                external_cp_links: RwLock::new(HashMap::new()),
423                aliases: RwLock::new(HashMap::new()),
424                registration_mutex: tokio::sync::Mutex::new(()),
425                after_ioc_running: std::sync::Mutex::new(Vec::new()),
426                scan_started: std::sync::atomic::AtomicBool::new(false),
427                pini_done: std::sync::atomic::AtomicBool::new(false),
428                pini_notify: tokio::sync::Notify::new(),
429                record_locks: record_lock::RecordLockRegistry::default(),
430            }),
431        }
432    }
433
434    /// Atomically claim the right to start the scan scheduler for this DB.
435    /// Returns `true` on the first call, `false` on subsequent calls.
436    /// Used by `ScanScheduler::run_with_hooks` to prevent duplicate scan tasks
437    /// when multiple protocol servers (CA + PVA) both try to start scanning.
438    pub fn try_claim_scan_start(&self) -> bool {
439        self.inner
440            .scan_started
441            .compare_exchange(
442                false,
443                true,
444                std::sync::atomic::Ordering::AcqRel,
445                std::sync::atomic::Ordering::Acquire,
446            )
447            .is_ok()
448    }
449
450    /// Mark PINI processing complete. Wakes any non-owner scan schedulers
451    /// that were waiting before running their hooks.
452    pub fn mark_pini_done(&self) {
453        self.inner
454            .pini_done
455            .store(true, std::sync::atomic::Ordering::Release);
456        self.inner.pini_notify.notify_waiters();
457    }
458
459    /// Wait until the scan owner has completed PINI processing.
460    /// Returns immediately if PINI has already completed.
461    pub async fn wait_for_pini(&self) {
462        if self
463            .inner
464            .pini_done
465            .load(std::sync::atomic::Ordering::Acquire)
466        {
467            return;
468        }
469        // Register interest BEFORE re-checking the flag to avoid missing a
470        // signal that arrives between the load and the await — `notify_waiters`
471        // does not store a permit for late subscribers.
472        let notified = self.inner.pini_notify.notified();
473        if self
474            .inner
475            .pini_done
476            .load(std::sync::atomic::Ordering::Acquire)
477        {
478            return;
479        }
480        notified.await;
481    }
482
483    /// Install an async resolver invoked when [`PvDatabase::has_name`]
484    /// fails to find a name. Used by proxy/gateway implementations to
485    /// lazily populate PVs on first search.
486    pub async fn set_search_resolver(&self, resolver: SearchResolver) {
487        *self.inner.search_resolver.write().await = Some(resolver);
488    }
489
490    /// Remove the previously installed search resolver, if any.
491    pub async fn clear_search_resolver(&self) {
492        *self.inner.search_resolver.write().await = None;
493    }
494
495    /// Install the per-request existence gate (see [`ExistenceGate`]).
496    /// Replaces any previously installed gate. Used by the CA gateway so
497    /// a cached shadow PV re-runs host/state admission per request.
498    pub async fn set_existence_gate(&self, gate: ExistenceGate) {
499        *self.inner.existence_gate.write().await = Some(gate);
500    }
501
502    /// Remove the previously installed existence gate, if any.
503    pub async fn clear_existence_gate(&self) {
504        *self.inner.existence_gate.write().await = None;
505    }
506
507    /// True when a cached simple PV named `name` must be treated as
508    /// non-existent for `peer` because the installed [`ExistenceGate`]
509    /// denied it. Always `false` when no gate is installed (a plain IOC)
510    /// or when `name` does not resolve to a simple PV — records and
511    /// aliases are never gateway-managed and bypass the gate.
512    ///
513    /// The single consultation point for the gate, shared by
514    /// [`Self::find_entry_from`] and [`Self::has_name_from`] so the
515    /// "cached simple PV ⇒ exists" short-circuit is closed uniformly on
516    /// both the create and search paths.
517    async fn simple_pv_gate_denies(&self, name: &str, peer: Option<std::net::SocketAddr>) -> bool {
518        let gate = match self.inner.existence_gate.read().await.clone() {
519            Some(g) => g,
520            None => return false,
521        };
522        // Strip the channel-filter suffix exactly as the lookups do
523        // (CA-FR-8) so the gate sees the same record-path key the
524        // simple-PV map and the gateway cache are keyed on.
525        let record_path = filters::split_channel_name(name).record_path;
526        if !self
527            .inner
528            .simple_pvs
529            .read()
530            .await
531            .contains_key(record_path.as_str())
532        {
533            return false;
534        }
535        !gate(record_path, peer).await
536    }
537
538    /// Set an external PV resolver for CA/PVA link resolution.
539    /// The resolver is called synchronously from link reads.
540    pub async fn set_external_resolver(&self, resolver: ExternalPvResolver) {
541        *self.inner.external_resolver.write().await = Some(resolver);
542    }
543
544    /// Register a [`LinkSet`] under `scheme` (e.g. `"pva"` /
545    /// `"ca"`). The lset is consulted for `ParsedLink::Pva` /
546    /// `ParsedLink::Ca` link reads/writes before falling back to
547    /// the legacy [`ExternalPvResolver`]. Subsequent calls for the
548    /// same scheme replace the previous binding.
549    pub async fn register_link_set(&self, scheme: &str, lset: link_set::DynLinkSet) {
550        self.inner.link_sets.write().await.register(scheme, lset);
551    }
552
553    /// Look up the lset for `scheme`, if any.
554    pub async fn link_set(&self, scheme: &str) -> Option<link_set::DynLinkSet> {
555        self.inner.link_sets.read().await.get(scheme)
556    }
557
558    /// Snapshot of every registered scheme name. Stable order for
559    /// `dbpvxr` dumps.
560    pub async fn registered_link_schemes(&self) -> Vec<String> {
561        let mut s = self.inner.link_sets.read().await.schemes();
562        s.sort();
563        s
564    }
565
566    /// Wait for the CA links to local records to report
567    /// `is_connected() == true`. Mirrors `dbCa: iocInit wait for local CA
568    /// links to connect` (epics-base PR #768/#856). The working set is
569    /// exactly [`Self::external_link_targets`]: only the CA facility's
570    /// local-target links — `pva://` links and non-local CA links connect
571    /// in the background and are never waited on (pvxs parity).
572    ///
573    /// Polls every 100 ms. Returns:
574    /// * `Ok(connected_count)` — the number of links that ended up
575    ///   connected. May be smaller than the total when the timeout
576    ///   expired before everyone was ready.
577    /// * The total link count — i.e. the size of the working set
578    ///   that was checked. `(connected, total)` lets the caller log
579    ///   "M/N CA links connected".
580    ///
581    /// Pure no-op when no CA link set is registered, or when its
582    /// `link_names()` has no local-target link yet (e.g. lazy-open lsets
583    /// that haven't observed any record link — record processing creates
584    /// the entries on first read, after iocInit returns).
585    pub async fn wait_for_external_links(&self, timeout: std::time::Duration) -> (usize, usize) {
586        // Collect (lset, name) pairs once. `link_names()` may grow
587        // as record processing opens new links, but iocInit's wait
588        // is bounded by the records loaded *before* Phase 3 — every
589        // such link is already opened by the time wire_device_support
590        // and setup_cp_links return.
591        let targets = self.external_link_targets().await;
592        let total = targets.len();
593        if total == 0 {
594            return (0, 0);
595        }
596        let deadline = tokio::time::Instant::now() + timeout;
597        loop {
598            let connected = targets
599                .iter()
600                .filter(|(lset, name)| lset.is_connected(name))
601                .count();
602            if connected == total {
603                return (connected, total);
604            }
605            if tokio::time::Instant::now() >= deadline {
606                return (connected, total);
607            }
608            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
609        }
610    }
611
612    /// Snapshot the `(lset, link_name)` pairs the iocInit external-link
613    /// wait reasons over. Shared by [`Self::wait_for_external_links`] and
614    /// [`Self::unconnected_external_links`] so both see the identical
615    /// working set.
616    ///
617    /// C parity: the iocInit connection-wait is a property of the CA link
618    /// facility (dbCa) alone. `dbCaRun` (dbCa.c:370-380) blocks on
619    /// `initOutstanding`, the count of CA links flagged
620    /// `DBCA_CALLBACK_INIT_WAIT` — set only for a CA link whose target is
621    /// a LOCAL record (dbLink.c:128-130):
622    ///   int isLocal = dbChannelTest(pvname) == 0;
623    ///   dbCaAddLinkCallbackOpt(..., isLocal ? DBCA_CALLBACK_INIT_WAIT : 0)
624    /// No other external facility waits: pvxs pvalink's `linkGlobal_t::init`
625    /// (ioc/pvalink.cpp) only calls `chan->open()` per channel — it opens
626    /// in the background and never blocks iocInit. So the wait targets
627    /// exactly the CA link set's local-target links; a non-local CA link
628    /// (e.g. areaDetector's `ShutterStatusEPICS_RBV.INP = "test CP MS"`
629    /// placeholder) and every `pva://` link connect asynchronously and are
630    /// never held by iocInit, like C.
631    async fn external_link_targets(&self) -> Vec<(link_set::DynLinkSet, String)> {
632        // Only the CA facility participates — look it up directly rather
633        // than iterating every registered scheme. `has_name_no_resolve`
634        // is the `dbChannelTest` twin (target is a local record).
635        let Some(ca_lset) = ({
636            let registry = self.inner.link_sets.read().await;
637            registry.get("ca")
638        }) else {
639            return Vec::new();
640        };
641        let mut targets: Vec<(link_set::DynLinkSet, String)> = Vec::new();
642        for n in ca_lset.link_names() {
643            if self.has_name_no_resolve(&n).await {
644                targets.push((ca_lset.clone(), n));
645            }
646        }
647        targets
648    }
649
650    /// Names of the waited-on CA links (local-target, per
651    /// [`Self::external_link_targets`]) that are opened but not yet
652    /// connected. iocInit calls this after
653    /// [`Self::wait_for_external_links`] times out so the
654    /// "M/N connected" diagnostic can name the `N-M` it proceeded
655    /// without, instead of leaving the operator to run `dbcar`.
656    /// `pva://` links are not in this set — they never block iocInit.
657    pub async fn unconnected_external_links(&self) -> Vec<String> {
658        self.external_link_targets()
659            .await
660            .into_iter()
661            .filter(|(lset, name)| !lset.is_connected(name))
662            .map(|(_, name)| name)
663            .collect()
664    }
665
666    /// Enumerate every link-shaped field on `record_name`. Returns
667    /// `(field_name, link_string, parsed)` tuples for fields whose
668    /// raw value parses as a non-trivial link via
669    /// [`crate::server::record::parse_link_v2`]. Used by `dbpvxr` to
670    /// dump per-record link state without hardcoding the field-name
671    /// list — works across record types as long as they expose link
672    /// strings via [`Record::get_field`].
673    ///
674    /// Returns an empty Vec when the record doesn't exist.
675    pub async fn record_link_fields(
676        &self,
677        record_name: &str,
678    ) -> Vec<(String, String, crate::server::record::ParsedLink)> {
679        let rec = match self.get_record(record_name).await {
680            Some(r) => r,
681            None => return Vec::new(),
682        };
683        let inst = rec.read().await;
684        let mut out = Vec::new();
685        let push = |field: &str, raw: &str, out: &mut Vec<_>| {
686            if raw.is_empty() {
687                return;
688            }
689            let parsed = crate::server::record::parse_link_v2(raw);
690            if !matches!(parsed, crate::server::record::ParsedLink::None) {
691                out.push((field.to_string(), raw.to_string(), parsed));
692            }
693        };
694        // Canonical link-bearing fields stored on `CommonFields` as raw
695        // String. These do NOT appear as `DbFieldType::String` entries in
696        // `field_list()`: an `ai`'s `INP` / an `ao`'s `OUT` carry
697        // `DBF_INLINK` / `DBF_OUTLINK` descriptors (and `INP`/`OUT` are
698        // not in the record's static field table at all), so the previous
699        // `field_list()` scan filtered by `String` silently dropped every
700        // device-support link — the holder's pvalink monitor was never
701        // opened. Enumerate the canonical storage directly so this method
702        // is the single owner of "which fields on a record are links",
703        // shared by `setup_cp_links` (CA CP/CPP) and the pvalink install
704        // scan (PVA CP/CPP).
705        push("INP", &inst.common.inp, &mut out);
706        push("OUT", &inst.common.out, &mut out);
707        push("TSEL", &inst.common.tsel, &mut out);
708        push("SDIS", &inst.common.sdis, &mut out);
709        // Record-specific multi-input links (INPA..INPL for
710        // calc/calcout/sel/sub) and the CP-capable input link fields
711        // (DOL family, NVL, SELL, SGNL).
712        let mut field_names: Vec<&str> = inst
713            .record
714            .multi_input_links()
715            .iter()
716            .map(|(lf, _vf)| *lf)
717            .collect();
718        field_names.extend_from_slice(crate::server::database::links::CP_INPUT_LINK_FIELDS);
719        for field in field_names {
720            if let Some(EpicsValue::String(s)) = inst.record.get_field(field) {
721                push(field, &s.as_str_lossy(), &mut out);
722            }
723        }
724        out
725    }
726
727    /// Resolve an external PV name. Dispatches through the
728    /// `(scheme, name)` lset if one is registered; otherwise falls
729    /// back to the legacy [`ExternalPvResolver`] closure. `name`
730    /// may be the bare PV name (in which case `pva://` is assumed
731    /// when an lset is registered for that scheme) or a fully
732    /// scheme-prefixed string.
733    pub(crate) async fn resolve_external_pv(&self, name: &str) -> Option<EpicsValue> {
734        // Try lsets first. We accept both "scheme://body" and the
735        // bare body (stored in ParsedLink::Pva/Ca after the
736        // dispatch in record/link.rs).
737        let (scheme, body) = if let Some(rest) = name.strip_prefix("pva://") {
738            ("pva", rest)
739        } else if let Some(rest) = name.strip_prefix("ca://") {
740            ("ca", rest)
741        } else {
742            // No prefix — try every registered lset in turn. The
743            // first one with a value for `name` wins. Schemes are
744            // single-digit so this is cheap.
745            let registry = self.inner.link_sets.read().await;
746            for s in registry.schemes() {
747                if let Some(lset) = registry.get(&s) {
748                    if let Some(v) = lset.get_value(name) {
749                        return Some(v);
750                    }
751                }
752            }
753            drop(registry);
754            // Fall through to legacy resolver.
755            let resolver = self.inner.external_resolver.read().await;
756            return resolver.as_ref().and_then(|r| r(name));
757        };
758        if let Some(lset) = self.inner.link_sets.read().await.get(scheme) {
759            if let Some(v) = lset.get_value(body) {
760                return Some(v);
761            }
762        }
763        let resolver = self.inner.external_resolver.read().await;
764        resolver.as_ref().and_then(|r| r(name))
765    }
766
767    /// Add a simple PV with an initial value.
768    ///
769    /// Returns `Err` when `name` is already registered as a simple PV,
770    /// a record, or an alias — mirroring epics-base C IOC which treats
771    /// duplicate `dbLoadRecords` names as a fatal error. Callers that
772    /// want replace-on-overwrite semantics must first call
773    /// `remove_simple_pv` / `remove_record` / `remove_alias`.
774    ///
775    /// Serialized through `registration_mutex` so the
776    /// cross-namespace check is atomic with the insert and the lock
777    /// order across all add_*/remove_* methods is identical (no
778    /// cross-namespace deadlock).
779    pub async fn add_pv(&self, name: &str, initial: EpicsValue) -> CaResult<()> {
780        let _gate = self.inner.registration_mutex.lock().await;
781        self.check_name_free(name).await?;
782        let pv = Arc::new(ProcessVariable::new(name.to_string(), initial));
783        self.inner
784            .simple_pvs
785            .write()
786            .await
787            .insert(name.to_string(), pv);
788        Ok(())
789    }
790
791    /// Add a simple PV that already has a [`WriteHook`] installed.
792    ///
793    /// Equivalent to `add_pv` followed by `find_pv` + `set_write_hook`,
794    /// but the PV is constructed with the hook in place so it is
795    /// inserted into the `simple_pvs` map ATOMICALLY with the hook
796    /// already attached. Closes a small race in proxy/gateway code
797    /// where a downstream client could (in principle) `CREATE_CHAN` +
798    /// `WRITE_NOTIFY` between the two awaits and hit the local
799    /// `pv.set()` fallback path before the hook landed.
800    ///
801    /// Returns `Err` on duplicate name (see [`add_pv`]).
802    pub async fn add_pv_with_hook(
803        &self,
804        name: &str,
805        initial: EpicsValue,
806        hook: crate::server::pv::WriteHook,
807    ) -> CaResult<()> {
808        self.add_pv_with_hooks(name, initial, hook, None).await
809    }
810
811    /// like [`Self::add_pv_with_hook`] but also installs an
812    /// optional [`AccessHook`](crate::server::pv::AccessHook) so the CA
813    /// gateway can route this shadow PV's read/write access-rights
814    /// decision through its own ACF. Both hooks are attached before the
815    /// PV is inserted into `simple_pvs`, so a downstream `CREATE_CHAN`
816    /// cannot observe the PV without its access hook bound.
817    pub async fn add_pv_with_hooks(
818        &self,
819        name: &str,
820        initial: EpicsValue,
821        write_hook: crate::server::pv::WriteHook,
822        access_hook: Option<crate::server::pv::AccessHook>,
823    ) -> CaResult<()> {
824        self.add_pv_with_hooks_full(name, initial, write_hook, access_hook, None)
825            .await
826    }
827
828    /// like [`Self::add_pv_with_hooks`] but also installs an optional
829    /// [`ReadHook`](crate::server::pv::ReadHook) so a proxy (the CA
830    /// gateway in no-cache mode) can serve each downstream GET from a
831    /// fresh upstream fetch instead of the stored value. All three hooks
832    /// are attached before the PV is inserted into `simple_pvs`, so a
833    /// downstream `CREATE_CHAN` cannot observe the PV without its hooks
834    /// bound — the read hook lands atomically with registration, closing
835    /// the same race the write/access hooks already close. `read_hook:
836    /// None` is identical to [`Self::add_pv_with_hooks`].
837    pub async fn add_pv_with_hooks_full(
838        &self,
839        name: &str,
840        initial: EpicsValue,
841        write_hook: crate::server::pv::WriteHook,
842        access_hook: Option<crate::server::pv::AccessHook>,
843        read_hook: Option<crate::server::pv::ReadHook>,
844    ) -> CaResult<()> {
845        let _gate = self.inner.registration_mutex.lock().await;
846        self.check_name_free(name).await?;
847        let pv = Arc::new(ProcessVariable::new(name.to_string(), initial));
848        pv.set_write_hook(write_hook);
849        if let Some(access) = access_hook {
850            pv.set_access_hook(access);
851        }
852        if let Some(read) = read_hook {
853            pv.set_read_hook(read);
854        }
855        self.inner
856            .simple_pvs
857            .write()
858            .await
859            .insert(name.to_string(), pv);
860        Ok(())
861    }
862
863    /// Remove a simple PV by name. Returns `Some(pv)` if a PV was
864    /// removed. Used by the gateway sweep so an evicted upstream
865    /// subscription doesn't leave a stale shadow PV (with a now-dead
866    /// `WriteHook` capturing an aborted upstream channel).
867    ///
868    /// Also purges any aliases that pointed AT this name
869    /// (otherwise a re-add of the same alias name would fail with
870    /// "already registered as an alias" even though its target is
871    /// gone).
872    pub async fn remove_simple_pv(&self, name: &str) -> Option<Arc<ProcessVariable>> {
873        let _gate = self.inner.registration_mutex.lock().await;
874        // Simple PVs cannot be alias targets (aliases point at
875        // records), but a stale alias whose name MATCHES this PV
876        // would have been rejected at add_alias time. No alias
877        // cleanup needed for simple-PV removal.
878        self.inner.simple_pvs.write().await.remove(name)
879    }
880
881    /// Add a record (accepts a boxed Record to avoid double-boxing).
882    ///
883    /// Returns `Err` when `name` collides with an existing record,
884    /// simple PV, or alias. The C IOC's `dbLoadRecords` treats this as
885    /// fatal; do not silently replace.
886    ///
887    /// The records-map insert AND scan-index insert run
888    /// under the same `registration_mutex` hold, eliminating the
889    /// TOCTOU window where `remove_record` could land between them
890    /// and leave a phantom scan entry.
891    pub async fn add_record(&self, name: &str, record: Box<dyn Record>) -> CaResult<()> {
892        let _gate = self.inner.registration_mutex.lock().await;
893        self.check_name_free(name).await?;
894        let mut instance = RecordInstance::new_boxed(name.to_string(), record);
895        // Hand the record a cycle-free handle to its own database so it can
896        // post out-of-band field updates / wire completion-driven re-entry
897        // (asyn TRACE callback, sseq WAITn) without owning the database.
898        // C records reach `dbCommon::pdba`/the IOC the same way at
899        // `dbDefineRecord` init; the framework supplies the back-reference,
900        // the record never constructs it. Defaulted no-op for records that
901        // do not need it.
902        instance
903            .record
904            .set_async_context(name.to_string(), self.async_handle());
905        let scan = instance.common.scan;
906        let phas = instance.common.phas;
907        self.inner
908            .records
909            .write()
910            .await
911            .insert(name.to_string(), Arc::new(RwLock::new(instance)));
912
913        // Assign a monotonic load-order sequence — the scan-index
914        // secondary sort key, so same-PHAS records keep load order.
915        let seq = self
916            .inner
917            .load_order_counter
918            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
919        self.inner
920            .load_order
921            .write()
922            .await
923            .insert(name.to_string(), seq);
924
925        if scan != ScanType::Passive {
926            self.inner
927                .scan_index
928                .write()
929                .await
930                .entry(scan)
931                .or_default()
932                .insert((phas, seq, name.to_string()));
933        }
934        Ok(())
935    }
936
937    /// Verify that `name` is not currently registered in any of the
938    /// three namespaces. Caller MUST hold `registration_mutex` so the
939    /// peek-then-insert sequence is atomic — without that, two tasks
940    /// can both see the name as free and race the insert.
941    async fn check_name_free(&self, name: &str) -> CaResult<()> {
942        let kind = if self.inner.simple_pvs.read().await.contains_key(name) {
943            Some("simple PV")
944        } else if self.inner.records.read().await.contains_key(name) {
945            Some("record")
946        } else if self.inner.aliases.read().await.contains_key(name) {
947            Some("alias")
948        } else {
949            None
950        };
951        if let Some(kind) = kind {
952            return Err(CaError::DbParseError {
953                line: 0,
954                column: 0,
955                message: format!("name '{name}' is already registered as a {kind}"),
956            });
957        }
958        Ok(())
959    }
960
961    /// Remove a record by name. Returns `true` if a record was removed,
962    /// `false` if no such name was registered. Mirrors epics-base PR
963    /// #505 — deletion at database creation, exposed here as a public
964    /// API so iocsh `dbDeleteRecord` and tests can drive it.
965    ///
966    /// The cleanup covers the three indices that `add_record` populates:
967    /// the records map, the scan index, and CP-link source/target lists.
968    /// Live subscribers on the removed record drop their `Sender` clone
969    /// when the `RecordInstance` is dropped — they observe `Closed` on
970    /// next recv, matching the existing dbEvent cancel flow.
971    pub async fn remove_record(&self, name: &str) -> bool {
972        let _gate = self.inner.registration_mutex.lock().await;
973        // 1) Remove from main map; keep scan + phas for scan-index cleanup.
974        let removed = self.inner.records.write().await.remove(name);
975        let Some(rec_arc) = removed else {
976            return false;
977        };
978        let scan = {
979            let inst = rec_arc.read().await;
980            inst.common.scan
981        };
982
983        // 2) Drop from scan index if it was scheduled. Match by record
984        // name only — PHAS and load_order are not needed and may be
985        // stale relative to the entry actually present.
986        if scan != ScanType::Passive {
987            let mut idx = self.inner.scan_index.write().await;
988            if let Some(set) = idx.get_mut(&scan) {
989                set.retain(|(_, _, n)| n != name);
990                if set.is_empty() {
991                    idx.remove(&scan);
992                }
993            }
994        }
995
996        // 2b) Drop the load-order entry.
997        self.inner.load_order.write().await.remove(name);
998
999        // 3) Drop from CP-link tables. Removed both as source (channel
1000        // change → trigger targets) and as target (other channels'
1001        // CP lists may still reference this name).
1002        let mut cp = self.inner.cp_links.write().await;
1003        cp.remove(name);
1004        for targets in cp.values_mut() {
1005            targets.retain(|t| t.record != name);
1006        }
1007        drop(cp);
1008
1009        // 4) Purge aliases that pointed AT the
1010        // removed record. Otherwise `find_pv("ALT")` returns None
1011        // (target gone) but `add_pv("ALT", ...)` still fails with
1012        // "already registered as an alias" — orphan blocks reuse.
1013        let mut aliases = self.inner.aliases.write().await;
1014        aliases.retain(|_alias, target| target != name);
1015
1016        true
1017    }
1018
1019    /// Internal: synchronous lookup without invoking the search resolver.
1020    async fn find_entry_no_resolve(&self, name: &str) -> Option<PvEntry> {
1021        // a channel name may carry a `.{"arr":...}` filter
1022        // suffix. Strip it before lookup — the suffix is a per-channel
1023        // filter spec, not part of the PV identity. `split_channel_name`
1024        // is the single owner of "channel name → record_path" and is
1025        // idempotent on an already-stripped name. Without this a
1026        // filtered SimplePv (`SP.{"arr":...}`) never matches
1027        // `simple_pvs` (keyed by the bare PV name) and even a filtered
1028        // record fails when the JSON contains a `.` (e.g.
1029        // `{"dbnd":{"d":0.5}}`), because the bare `parse_pv_name` last-dot
1030        // split would tear the suffix apart instead of removing it.
1031        let record_path = filters::split_channel_name(name).record_path;
1032        let (base, _field) = parse_pv_name(&record_path);
1033
1034        if let Some(pv) = self.inner.simple_pvs.read().await.get(record_path.as_str()) {
1035            return Some(PvEntry::Simple(pv.clone()));
1036        }
1037        if let Some(rec) = self.inner.records.read().await.get(base) {
1038            return Some(PvEntry::Record(rec.clone()));
1039        }
1040        // Alias resolve (epics-base PR #336): the alternate name maps
1041        // to a canonical record name. Look up the real record after
1042        // translating the base.
1043        if let Some(target) = self.inner.aliases.read().await.get(base).cloned() {
1044            if let Some(rec) = self.inner.records.read().await.get(&target) {
1045                return Some(PvEntry::Record(rec.clone()));
1046            }
1047        }
1048        None
1049    }
1050
1051    /// Register an alias `alias` for an existing record `target`.
1052    /// Mirrors epics-base PR #336. Returns `Err(...)` when the target
1053    /// does not exist or the alias name is already in use anywhere
1054    /// in the database (records, simple PVs, or other aliases).
1055    ///
1056    /// Pre-fix the alias path checked only
1057    /// `records` and `aliases` — a simple-PV with the same name as
1058    /// the proposed alias was missed, leaving the database in a
1059    /// state where `find_pv(alias)` could resolve to either the
1060    /// simple PV or the alias-mapped record depending on lookup
1061    /// order. Now we run the same cross-namespace `check_name_free`
1062    /// guard the other add_* paths use.
1063    pub async fn add_alias(&self, alias: &str, target: &str) -> CaResult<()> {
1064        let _gate = self.inner.registration_mutex.lock().await;
1065        if !self.inner.records.read().await.contains_key(target) {
1066            return Err(CaError::ChannelNotFound(format!(
1067                "alias target '{target}' is not a registered record"
1068            )));
1069        }
1070        self.check_name_free(alias).await?;
1071        self.inner
1072            .aliases
1073            .write()
1074            .await
1075            .insert(alias.to_string(), target.to_string());
1076        Ok(())
1077    }
1078
1079    /// Resolve an alias to its target record name, or `None` when the
1080    /// name is not an alias.
1081    pub async fn resolve_alias(&self, name: &str) -> Option<String> {
1082        self.inner.aliases.read().await.get(name).cloned()
1083    }
1084
1085    /// Queue an iocsh command line for post-PINI execution.
1086    /// Mirrors epics-base PR #558 — `afterIocRunning <command>` lets
1087    /// the startup script schedule actions that run after iocInit
1088    /// completes (when the record set is fully wired up).
1089    pub fn queue_after_ioc_running(&self, line: impl Into<String>) {
1090        self.inner
1091            .after_ioc_running
1092            .lock()
1093            .unwrap()
1094            .push(line.into());
1095    }
1096
1097    /// Drain the post-PINI iocsh command queue. Called by
1098    /// `IocApplication::run` after PINI processing.
1099    pub fn take_after_ioc_running(&self) -> Vec<String> {
1100        std::mem::take(&mut *self.inner.after_ioc_running.lock().unwrap())
1101    }
1102
1103    /// Internal: synchronous existence check without resolver.
1104    async fn has_name_no_resolve(&self, name: &str) -> bool {
1105        // strip the channel-filter suffix before lookup so a
1106        // filtered channel (`SP.{"arr":...}` / `REC.{"dbnd":{"d":0.5}}`)
1107        // resolves to its underlying PV at UDP-search time. This is the
1108        // search-side twin of `find_entry_no_resolve`; without it a
1109        // filtered SimplePv never answers a SEARCH and the client never
1110        // reaches CREATE_CHAN. See that function for the full rationale.
1111        let record_path = filters::split_channel_name(name).record_path;
1112        let (base, _) = parse_pv_name(&record_path);
1113        if self
1114            .inner
1115            .simple_pvs
1116            .read()
1117            .await
1118            .contains_key(record_path.as_str())
1119        {
1120            return true;
1121        }
1122        if self.inner.records.read().await.contains_key(base) {
1123            return true;
1124        }
1125        // Alias entry exists and points to a live record
1126        // (epics-base PR #336).
1127        if let Some(target) = self.inner.aliases.read().await.get(base) {
1128            return self.inner.records.read().await.contains_key(target);
1129        }
1130        false
1131    }
1132
1133    /// Look up an entry by name. Supports "record.FIELD" syntax.
1134    ///
1135    /// If the name is not found and a search resolver is installed,
1136    /// the resolver is invoked once. If the resolver returns true, the
1137    /// database is re-checked.
1138    pub async fn find_entry(&self, name: &str) -> Option<PvEntry> {
1139        self.find_entry_from(name, None).await
1140    }
1141
1142    /// Like [`Self::find_entry`], but threads the downstream client's
1143    /// socket address into the search resolver. The CA TCP CREATE_CHANNEL
1144    /// handler passes the connection peer so the gateway can apply
1145    /// host-scoped `.pvlist` admission.
1146    pub async fn find_entry_from(
1147        &self,
1148        name: &str,
1149        peer: Option<std::net::SocketAddr>,
1150    ) -> Option<PvEntry> {
1151        if let Some(entry) = self.find_entry_no_resolve(name).await {
1152            // A cached simple PV must still pass the per-request
1153            // existence gate (CA gateway host/state admission). When the
1154            // gate denies it, answer does-not-exist for this requester
1155            // instead of returning the stale shadow entry — C ca-gateway
1156            // re-runs `gateAs::findEntry`/cache-state on every
1157            // `pvExistTest` (gateServer.cc:1516-1637). Records/aliases
1158            // bypass the gate (see `simple_pv_gate_denies`).
1159            if matches!(entry, PvEntry::Simple(_)) && self.simple_pv_gate_denies(name, peer).await {
1160                return None;
1161            }
1162            return Some(entry);
1163        }
1164        // Try the search resolver
1165        let resolver = self.inner.search_resolver.read().await.clone();
1166        if let Some(r) = resolver {
1167            if r(name.to_string(), peer).await {
1168                return self.find_entry_no_resolve(name).await;
1169            }
1170        }
1171        None
1172    }
1173
1174    /// Check if a base name exists (for UDP search).
1175    ///
1176    /// If the name is not in the database and a search resolver is installed,
1177    /// the resolver is invoked. The resolver may populate the database
1178    /// (e.g., subscribe to an upstream IOC and add a placeholder PV) and
1179    /// return true; this method then re-checks.
1180    pub async fn has_name(&self, name: &str) -> bool {
1181        self.has_name_from(name, None).await
1182    }
1183
1184    /// Like [`Self::has_name`], but threads the downstream client's
1185    /// socket address into the search resolver. The CA UDP search
1186    /// responder passes the datagram source address so the gateway can
1187    /// apply host-scoped `.pvlist` admission.
1188    pub async fn has_name_from(&self, name: &str, peer: Option<std::net::SocketAddr>) -> bool {
1189        if self.has_name_no_resolve(name).await {
1190            // Same per-request gate as `find_entry_from`: a cached simple
1191            // PV the gateway's host/state admission denies must answer
1192            // does-not-exist at search time. Records/aliases bypass.
1193            if self.simple_pv_gate_denies(name, peer).await {
1194                return false;
1195            }
1196            return true;
1197        }
1198        let resolver = self.inner.search_resolver.read().await.clone();
1199        if let Some(r) = resolver {
1200            if r(name.to_string(), peer).await {
1201                return self.has_name_no_resolve(name).await;
1202            }
1203        }
1204        false
1205    }
1206
1207    /// Look up a simple PV by name (backward-compatible).
1208    pub async fn find_pv(&self, name: &str) -> Option<Arc<ProcessVariable>> {
1209        if let Some(pv) = self.inner.simple_pvs.read().await.get(name) {
1210            return Some(pv.clone());
1211        }
1212        None
1213    }
1214
1215    /// Get a record Arc by name. Alias-aware (epics-base PR #336):
1216    /// when `name` is not a canonical record but matches a registered
1217    /// alias, the alias' target record is returned. Mirrors base
1218    /// `dbNameToAddr` behaviour, so dbpf/dbpr/dbgf, CA channel lookup,
1219    /// and DB-link target resolution all work transparently for
1220    /// aliases.
1221    ///
1222    /// Use [`Self::get_record_no_resolve`] when the caller already
1223    /// holds a canonical name and wants to suppress the alias path
1224    /// (e.g. to detect alias collisions during builder wiring).
1225    pub async fn get_record(&self, name: &str) -> Option<Arc<RwLock<RecordInstance>>> {
1226        if let Some(rec) = self.inner.records.read().await.get(name).cloned() {
1227            return Some(rec);
1228        }
1229        let target = self.inner.aliases.read().await.get(name).cloned()?;
1230        self.inner.records.read().await.get(&target).cloned()
1231    }
1232
1233    /// Strict variant of [`Self::get_record`] — does NOT consult the
1234    /// alias table. Returns `Some` only when a canonical record with
1235    /// that exact name exists.
1236    pub async fn get_record_no_resolve(&self, name: &str) -> Option<Arc<RwLock<RecordInstance>>> {
1237        self.inner.records.read().await.get(name).cloned()
1238    }
1239
1240    /// Get all record names.
1241    pub async fn all_record_names(&self) -> Vec<String> {
1242        self.inner.records.read().await.keys().cloned().collect()
1243    }
1244
1245    /// Get all alias names registered against existing records.
1246    /// Mirrors the alias-half of base's `dbFirstRecord` iteration —
1247    /// `dbgrep` / `dbglob` / `dbsr` walk both record names and
1248    /// aliases when matching a glob.
1249    pub async fn all_alias_names(&self) -> Vec<String> {
1250        self.inner.aliases.read().await.keys().cloned().collect()
1251    }
1252
1253    /// Return every alias that points at `canonical`. Sorted for
1254    /// stable output; empty when the record has no aliases. Used by
1255    /// `dbpr` to surface alias-form names so admins can see how
1256    /// clients may reach the record.
1257    pub async fn aliases_for_record(&self, canonical: &str) -> Vec<String> {
1258        let aliases = self.inner.aliases.read().await;
1259        let mut hits: Vec<String> = aliases
1260            .iter()
1261            .filter_map(|(alias, target)| {
1262                if target == canonical {
1263                    Some(alias.clone())
1264                } else {
1265                    None
1266                }
1267            })
1268            .collect();
1269        hits.sort();
1270        hits
1271    }
1272
1273    /// Get all simple PV names.
1274    pub async fn all_simple_pv_names(&self) -> Vec<String> {
1275        self.inner.simple_pvs.read().await.keys().cloned().collect()
1276    }
1277}
1278
1279#[cfg(test)]
1280mod tests {
1281    use super::*;
1282
1283    /// C `recGblGetTimeStampSimm` (recGbl.c:310-343) maps TSE values
1284    /// to epicsTime sources via the constants in `epicsTime.h:102-104`.
1285    /// The Rust port previously misread TSE=-1 as "device-provided
1286    /// with BestTime fallback" and gated the BestTime call on a
1287    /// UNIX_EPOCH check. C calls `epicsTimeGetEvent(-1)`
1288    /// unconditionally; only TSE=-2 (epicsTimeEventDeviceTime) leaves
1289    /// `precord->time` untouched.
1290    ///
1291    /// Regression: a stale device write (any non-epoch SystemTime)
1292    /// suppressed every BestTime refresh thereafter.
1293    #[test]
1294    fn apply_timestamp_tse_minus_one_always_overwrites_with_best_time() {
1295        use crate::server::record::CommonFields;
1296        use std::time::{Duration, SystemTime};
1297
1298        // Pre-populate `time` with a stale but non-epoch sentinel.
1299        let stale = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
1300        let mut common = CommonFields::default();
1301        common.tse = -1;
1302        common.time = stale;
1303
1304        apply_timestamp(&mut common, false);
1305
1306        // BestTime must have run unconditionally — `common.time` is
1307        // no longer the stale sentinel.
1308        assert_ne!(
1309            common.time, stale,
1310            "TSE=-1 must always overwrite via generalTime BestTime, \
1311             matching C epicsTimeGetEvent(-1) called unconditionally"
1312        );
1313    }
1314
1315    /// C `epicsTimeEventDeviceTime = -2` (epicsTime.h:104). The C
1316    /// path does NOT call `epicsTimeGetEvent` for this TSE value;
1317    /// device support has already set `precord->time` before the
1318    /// recGbl call. The Rust port must leave `common.time` untouched.
1319    #[test]
1320    fn apply_timestamp_tse_minus_two_preserves_device_provided_time() {
1321        use crate::server::record::CommonFields;
1322        use std::time::{Duration, SystemTime};
1323
1324        let device_time = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000_000);
1325        let mut common = CommonFields::default();
1326        common.tse = -2;
1327        common.time = device_time;
1328
1329        apply_timestamp(&mut common, false);
1330
1331        assert_eq!(
1332            common.time, device_time,
1333            "TSE=-2 (epicsTimeEventDeviceTime) must preserve device-provided time"
1334        );
1335    }
1336
1337    #[test]
1338    fn select_link_indices_fanout_all_specified_mask() {
1339        use crate::server::record::AlarmSeverity;
1340        // All — every slot.
1341        let r = select_link_indices_ex(SelmKind::FanoutSeq, 0, 0, 0, 0, 16);
1342        assert_eq!(r.indices, (0..16).collect::<Vec<_>>());
1343        assert!(r.alarm.is_none());
1344
1345        // Specified, 0-based: SELN=0 selects LNK0 (C parity, fanout).
1346        let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 0, 0, 0, 16);
1347        assert_eq!(r.indices, vec![0]);
1348        // Specified with OFFS bias: SELN=2 + OFFS=3 → index 5.
1349        let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 2, 3, 0, 16);
1350        assert_eq!(r.indices, vec![5]);
1351        // Out-of-range Specified → INVALID alarm, no links.
1352        let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 20, 0, 0, 16);
1353        assert!(r.indices.is_empty());
1354        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1355        // Negative resolved index (SELN + negative OFFS) → INVALID.
1356        let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 0, -1, 0, 16);
1357        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1358
1359        // Mask: SELN=0b101 → bits 0 and 2.
1360        let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 0, 16);
1361        assert_eq!(r.indices, vec![0, 2]);
1362        // Mask with SHFT: SELN=0b101 >> 1 = 0b10 → bit 1.
1363        let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 1, 16);
1364        assert_eq!(r.indices, vec![1]);
1365        // Mask with negative SHFT: SELN=0b101 << 1 = 0b1010 → bits 1,3.
1366        let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, -1, 16);
1367        assert_eq!(r.indices, vec![1, 3]);
1368        // SHFT out of [-15,15] → INVALID.
1369        let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 16, 16);
1370        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1371
1372        // Unknown SELM → INVALID.
1373        let r = select_link_indices_ex(SelmKind::FanoutSeq, 9, 0, 0, 0, 16);
1374        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1375    }
1376
1377    #[test]
1378    fn select_link_indices_dfanout_specified_is_one_based() {
1379        use crate::server::record::AlarmSeverity;
1380        // dfanout Specified is 1-based: SELN=1 → OUTA (index 0).
1381        let r = select_link_indices_ex(SelmKind::Dfanout, 1, 1, 0, 0, 16);
1382        assert_eq!(r.indices, vec![0]);
1383        // SELN=2 → OUTB (index 1).
1384        let r = select_link_indices_ex(SelmKind::Dfanout, 1, 2, 0, 0, 16);
1385        assert_eq!(r.indices, vec![1]);
1386        // SELN=0 → drive nothing, NO alarm.
1387        let r = select_link_indices_ex(SelmKind::Dfanout, 1, 0, 0, 0, 16);
1388        assert!(r.indices.is_empty());
1389        assert!(r.alarm.is_none());
1390        // SELN > 16 → INVALID.
1391        let r = select_link_indices_ex(SelmKind::Dfanout, 1, 17, 0, 0, 16);
1392        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1393        // dfanout Mask has no SHFT — SHFT arg ignored.
1394        let r = select_link_indices_ex(SelmKind::Dfanout, 2, 5, 0, 7, 16);
1395        assert_eq!(r.indices, vec![0, 2]);
1396    }
1397
1398    #[test]
1399    fn seln_is_unsigned_dbr_ushort_cast() {
1400        use crate::server::record::AlarmSeverity;
1401        // `dbr_ushort_cast` reproduces C's `(epicsUInt16)` cast
1402        // (dbConvert.c:63-70): truncate toward zero, wrap mod 2^16, NOT a
1403        // clamp. SELL=-1 → 65535, SELL=65536 → 0.
1404        assert_eq!(dbr_ushort_cast(&EpicsValue::Double(-1.0)), 65535);
1405        assert_eq!(dbr_ushort_cast(&EpicsValue::Double(65536.0)), 0);
1406        assert_eq!(dbr_ushort_cast(&EpicsValue::Double(3.7)), 3);
1407        assert_eq!(dbr_ushort_cast(&EpicsValue::Long(-1)), 65535);
1408
1409        // SELL=-1 casts to the unsigned `DBF_USHORT` value 65535.
1410        let seln_max = 65535u16;
1411        // fanout/seq Specified: C `i = (epicsUInt16)seln + offs` =
1412        // 65535 → out of range → INVALID. The old signed read clamped to
1413        // 0 and drove link 0.
1414        let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, seln_max, 0, 0, 16);
1415        assert!(r.indices.is_empty());
1416        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1417        // fanout/seq Mask: 65535 → all 16 low bits set → every link. The
1418        // old clamp produced an empty mask.
1419        let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, seln_max, 0, 0, 16);
1420        assert_eq!(r.indices, (0..16).collect::<Vec<_>>());
1421        // dfanout Specified: 65535 > count → INVALID. The old signed read
1422        // saw -1 ≤ 0 → drove nothing with no alarm.
1423        let r = select_link_indices_ex(SelmKind::Dfanout, 1, seln_max, 0, 0, 16);
1424        assert!(r.indices.is_empty());
1425        assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1426    }
1427
1428    /// Lset that flips to "connected" after a configurable delay.
1429    /// Drives the wait_for_external_links time-budget tests below.
1430    struct DelayedConnectLset {
1431        names: Vec<String>,
1432        connect_at: tokio::time::Instant,
1433    }
1434
1435    impl link_set::LinkSet for DelayedConnectLset {
1436        fn is_connected(&self, _: &str) -> bool {
1437            tokio::time::Instant::now() >= self.connect_at
1438        }
1439        fn get_value(&self, _: &str) -> Option<EpicsValue> {
1440            None
1441        }
1442        fn link_names(&self) -> Vec<String> {
1443            self.names.clone()
1444        }
1445    }
1446
1447    #[tokio::test]
1448    async fn wait_for_external_links_returns_zero_zero_when_no_lsets() {
1449        let db = PvDatabase::new();
1450        let (c, t) = db
1451            .wait_for_external_links(std::time::Duration::from_millis(50))
1452            .await;
1453        assert_eq!((c, t), (0, 0));
1454    }
1455
1456    #[tokio::test]
1457    async fn wait_for_external_links_connected_quickly() {
1458        let db = PvDatabase::new();
1459        // Local-target forced-CA links (dbChannelTest==0 → isLocal): these
1460        // get DBCA_CALLBACK_INIT_WAIT, so iocInit waits for them.
1461        db.add_pv("pv:A", EpicsValue::Long(0)).await.unwrap();
1462        db.add_pv("pv:B", EpicsValue::Long(0)).await.unwrap();
1463        let lset = Arc::new(DelayedConnectLset {
1464            names: vec!["pv:A".to_string(), "pv:B".to_string()],
1465            connect_at: tokio::time::Instant::now(),
1466        });
1467        // Registered under "ca": the iocInit wait is CA-facility only, so
1468        // the working set comes from the "ca" link set (these forced-CA
1469        // local-target links), never from a "pva" set.
1470        db.register_link_set("ca", lset).await;
1471        let (c, t) = db
1472            .wait_for_external_links(std::time::Duration::from_secs(1))
1473            .await;
1474        assert_eq!((c, t), (2, 2));
1475    }
1476
1477    #[tokio::test]
1478    async fn wait_for_external_links_returns_partial_on_timeout() {
1479        let db = PvDatabase::new();
1480        // Local target so the link is in the init-wait set (dbLink.c:130);
1481        // connect-time well past the budget below, so the wait must return
1482        // (0, 1) instead of blocking.
1483        db.add_pv("slow:pv", EpicsValue::Long(0)).await.unwrap();
1484        let lset = Arc::new(DelayedConnectLset {
1485            names: vec!["slow:pv".to_string()],
1486            connect_at: tokio::time::Instant::now() + std::time::Duration::from_secs(60),
1487        });
1488        db.register_link_set("ca", lset).await;
1489        let started = tokio::time::Instant::now();
1490        let (c, t) = db
1491            .wait_for_external_links(std::time::Duration::from_millis(250))
1492            .await;
1493        let elapsed = started.elapsed();
1494        assert_eq!((c, t), (0, 1));
1495        assert!(
1496            elapsed >= std::time::Duration::from_millis(200),
1497            "wait must consume at least the configured budget, got {:?}",
1498            elapsed
1499        );
1500        assert!(
1501            elapsed < std::time::Duration::from_secs(2),
1502            "wait must not exceed the budget by much, got {:?}",
1503            elapsed
1504        );
1505    }
1506
1507    /// C parity (dbLink.c:130): a link whose target is NOT a local record
1508    /// (`dbChannelTest != 0`) gets no DBCA_CALLBACK_INIT_WAIT, so iocInit
1509    /// must not block on it. An areaDetector `test CP MS` placeholder — a CP
1510    /// link to a PV that exists nowhere — must drop straight through, leaving
1511    /// the link to connect (or dangle) asynchronously and silently, like C.
1512    #[tokio::test]
1513    async fn wait_for_external_links_skips_nonlocal_targets() {
1514        let db = PvDatabase::new();
1515        // "test" has no local record and would never connect.
1516        let lset = Arc::new(DelayedConnectLset {
1517            names: vec!["test".to_string()],
1518            connect_at: tokio::time::Instant::now() + std::time::Duration::from_secs(60),
1519        });
1520        db.register_link_set("ca", lset).await;
1521        let started = tokio::time::Instant::now();
1522        let (c, t) = db
1523            .wait_for_external_links(std::time::Duration::from_secs(10))
1524            .await;
1525        // Non-local target is excluded from the wait set entirely, so the
1526        // call returns (0, 0) immediately rather than blocking the budget.
1527        assert_eq!((c, t), (0, 0));
1528        assert!(
1529            started.elapsed() < std::time::Duration::from_secs(1),
1530            "non-local link must not be waited on, got {:?}",
1531            started.elapsed()
1532        );
1533        // And it is reported as unconnected by neither path (silent, like C).
1534        assert!(db.unconnected_external_links().await.is_empty());
1535    }
1536
1537    // epics-base PR #336 — alias parsing + lookup integration tests.
1538
1539    #[tokio::test]
1540    async fn alias_resolves_through_find_entry() {
1541        let db = PvDatabase::new();
1542        db.add_record(
1543            "TARGET",
1544            Box::new(crate::server::records::ai::AiRecord::new(42.0)),
1545        )
1546        .await
1547        .unwrap();
1548        db.add_alias("ALIAS_NAME", "TARGET").await.unwrap();
1549
1550        // find_entry on the alias must return the same record as
1551        // find_entry on the target.
1552        let via_alias = db.find_entry("ALIAS_NAME").await;
1553        let via_target = db.find_entry("TARGET").await;
1554        assert!(via_alias.is_some());
1555        assert!(via_target.is_some());
1556        // has_name flips true for the alias too.
1557        assert!(db.has_name("ALIAS_NAME").await);
1558        assert!(db.has_name("TARGET").await);
1559        assert!(!db.has_name("NOT:THERE").await);
1560    }
1561
1562    #[tokio::test]
1563    async fn alias_target_must_exist() {
1564        let db = PvDatabase::new();
1565        let err = db.add_alias("DANGLING", "MISSING_TARGET").await;
1566        assert!(err.is_err(), "alias to missing target must be rejected");
1567    }
1568
1569    #[tokio::test]
1570    async fn alias_collision_with_existing_record_rejected() {
1571        let db = PvDatabase::new();
1572        db.add_record(
1573            "EXISTING",
1574            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1575        )
1576        .await
1577        .unwrap();
1578        db.add_record(
1579            "OTHER",
1580            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1581        )
1582        .await
1583        .unwrap();
1584        let err = db.add_alias("EXISTING", "OTHER").await;
1585        assert!(
1586            err.is_err(),
1587            "alias name colliding with record must be rejected"
1588        );
1589    }
1590
1591    #[tokio::test]
1592    async fn get_record_resolves_alias() {
1593        // Regression: get_record must transparently resolve
1594        // aliases so dbpf / dbgf / dbpr / CA put paths see the same
1595        // record whether the caller uses the canonical name or the
1596        // alias.
1597        let db = PvDatabase::new();
1598        db.add_record(
1599            "TARGET",
1600            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1601        )
1602        .await
1603        .unwrap();
1604        db.add_alias("ALIAS", "TARGET").await.unwrap();
1605
1606        let via_canonical = db.get_record("TARGET").await;
1607        let via_alias = db.get_record("ALIAS").await;
1608        assert!(via_canonical.is_some());
1609        assert!(via_alias.is_some(), "get_record must resolve alias");
1610        // Both calls return the same Arc (pointer equality).
1611        assert!(Arc::ptr_eq(&via_canonical.unwrap(), &via_alias.unwrap()));
1612    }
1613
1614    #[tokio::test]
1615    async fn get_record_no_resolve_skips_alias_table() {
1616        // Strict variant must NOT see aliases — keeps the canonical
1617        // distinction available for builder code paths.
1618        let db = PvDatabase::new();
1619        db.add_record(
1620            "TARGET",
1621            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1622        )
1623        .await
1624        .unwrap();
1625        db.add_alias("ALIAS", "TARGET").await.unwrap();
1626
1627        assert!(db.get_record_no_resolve("TARGET").await.is_some());
1628        assert!(
1629            db.get_record_no_resolve("ALIAS").await.is_none(),
1630            "get_record_no_resolve must not follow alias table"
1631        );
1632    }
1633
1634    #[tokio::test]
1635    async fn register_cp_link_normalises_alias_to_canonical() {
1636        // Regression: CP link registration must store the
1637        // canonical record names. dispatch_cp_targets looks up by
1638        // canonical, so an alias-keyed entry is functionally dead.
1639        let db = PvDatabase::new();
1640        db.add_record(
1641            "SRC_REAL",
1642            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1643        )
1644        .await
1645        .unwrap();
1646        db.add_record(
1647            "DST_REAL",
1648            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1649        )
1650        .await
1651        .unwrap();
1652        db.add_alias("SRC_ALIAS", "SRC_REAL").await.unwrap();
1653        db.add_alias("DST_ALIAS", "DST_REAL").await.unwrap();
1654
1655        // Register using the alias forms (CP edge: passive_only = false).
1656        db.register_cp_link("SRC_ALIAS", "DST_ALIAS", false).await;
1657
1658        // Lookup must succeed via the canonical source name.
1659        let targets = db.get_cp_targets("SRC_REAL").await;
1660        assert_eq!(targets.len(), 1);
1661        assert_eq!(targets[0].record, "DST_REAL");
1662        assert!(!targets[0].passive_only);
1663        // Alias-keyed lookup must NOT have been registered.
1664        let alias_lookup = db.get_cp_targets("SRC_ALIAS").await;
1665        assert!(alias_lookup.is_empty());
1666    }
1667
1668    #[tokio::test]
1669    async fn aliases_for_record_returns_sorted_targets_only() {
1670        let db = PvDatabase::new();
1671        db.add_record(
1672            "TARGET",
1673            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1674        )
1675        .await
1676        .unwrap();
1677        db.add_record(
1678            "OTHER",
1679            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1680        )
1681        .await
1682        .unwrap();
1683        db.add_alias("ZZ", "TARGET").await.unwrap();
1684        db.add_alias("AA", "TARGET").await.unwrap();
1685        db.add_alias("MM", "OTHER").await.unwrap();
1686
1687        // Sorted, only TARGET's aliases.
1688        assert_eq!(
1689            db.aliases_for_record("TARGET").await,
1690            vec!["AA".to_string(), "ZZ".to_string()]
1691        );
1692        // OTHER's alone.
1693        assert_eq!(db.aliases_for_record("OTHER").await, vec!["MM".to_string()]);
1694        // Unknown record → empty, not None.
1695        assert!(db.aliases_for_record("MISSING").await.is_empty());
1696    }
1697
1698    #[tokio::test]
1699    async fn all_alias_names_returns_registered_aliases() {
1700        let db = PvDatabase::new();
1701        db.add_record(
1702            "TARGET",
1703            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1704        )
1705        .await
1706        .unwrap();
1707        db.add_alias("ALIAS_A", "TARGET").await.unwrap();
1708        db.add_alias("ALIAS_B", "TARGET").await.unwrap();
1709
1710        let mut aliases = db.all_alias_names().await;
1711        aliases.sort();
1712        assert_eq!(aliases, vec!["ALIAS_A".to_string(), "ALIAS_B".to_string()]);
1713        // Canonical names are NOT returned here.
1714        assert!(!aliases.contains(&"TARGET".to_string()));
1715    }
1716
1717    #[tokio::test]
1718    async fn complete_async_record_accepts_alias() {
1719        // Invariant audit: complete_async_record (the
1720        // entry point used by async device-support callbacks to
1721        // finish processing) must accept an alias name. Pre-fix it
1722        // walked `inner.records` directly and would
1723        // `ChannelNotFound` if the original name was an alias.
1724        let db = PvDatabase::new();
1725        db.add_record(
1726            "TARGET",
1727            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1728        )
1729        .await
1730        .unwrap();
1731        db.add_alias("ALIAS", "TARGET").await.unwrap();
1732
1733        // Use complete_async_record by alias — must not error.
1734        db.complete_async_record("ALIAS").await.unwrap();
1735        // And by canonical too — keeps existing behaviour.
1736        db.complete_async_record("TARGET").await.unwrap();
1737    }
1738
1739    #[tokio::test]
1740    async fn process_record_accepts_alias() {
1741        // Regression: process_record must accept an alias
1742        // name. Pre-fix it walked `inner.records` directly.
1743        let db = PvDatabase::new();
1744        db.add_record(
1745            "TARGET",
1746            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1747        )
1748        .await
1749        .unwrap();
1750        db.add_alias("ALIAS", "TARGET").await.unwrap();
1751
1752        // Both should succeed and reach the same record.
1753        db.process_record("TARGET").await.unwrap();
1754        db.process_record("ALIAS").await.unwrap();
1755
1756        // A bogus name still errors.
1757        assert!(db.process_record("MISSING").await.is_err());
1758    }
1759
1760    #[tokio::test]
1761    async fn process_record_with_links_accepts_alias_and_avoids_cycle() {
1762        // Regression: process_record_with_links normalises
1763        // the alias so that (a) the records-map lookup hits and
1764        // (b) the cycle-detection set doesn't treat alias and
1765        // canonical as two distinct entries (which would let a
1766        // self-loop slip past the visited check).
1767        let db = PvDatabase::new();
1768        db.add_record(
1769            "TARGET",
1770            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1771        )
1772        .await
1773        .unwrap();
1774        db.add_alias("ALIAS", "TARGET").await.unwrap();
1775
1776        let mut visited = std::collections::HashSet::new();
1777        db.process_record_with_links("ALIAS", &mut visited, 0)
1778            .await
1779            .unwrap();
1780
1781        // visited should contain the *canonical* name only.
1782        assert!(
1783            visited.contains("TARGET"),
1784            "visited must record the canonical name: {visited:?}",
1785        );
1786        assert!(
1787            !visited.contains("ALIAS"),
1788            "visited must NOT record the alias form: {visited:?}",
1789        );
1790    }
1791
1792    #[tokio::test]
1793    async fn alias_duplicate_rejected() {
1794        let db = PvDatabase::new();
1795        db.add_record(
1796            "TARGET",
1797            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1798        )
1799        .await
1800        .unwrap();
1801        db.add_alias("ALIAS", "TARGET").await.unwrap();
1802        // Re-registering the same alias name (even to the same target)
1803        // must fail — base behaviour: aliases are inserted once.
1804        let err = db.add_alias("ALIAS", "TARGET").await;
1805        assert!(err.is_err(), "duplicate alias name must be rejected");
1806    }
1807
1808    /// `add_pv`, `add_pv_with_hook`, and `add_record` must
1809    /// refuse to silently replace an existing registration. Mirrors
1810    /// epics-base C IOC which treats a duplicate `dbLoadRecords` name
1811    /// as a fatal load error.
1812    #[tokio::test]
1813    async fn add_pv_and_add_record_reject_duplicates_across_namespaces() {
1814        use crate::server::records::ai::AiRecord;
1815
1816        let db = PvDatabase::new();
1817        db.add_pv("A", EpicsValue::Double(1.0)).await.unwrap();
1818        // Same name as simple_pv — every namespace must see it.
1819        assert!(db.add_pv("A", EpicsValue::Double(2.0)).await.is_err());
1820        let noop_hook: crate::server::pv::WriteHook =
1821            std::sync::Arc::new(|_v, _ctx| Box::pin(async { Ok(()) }));
1822        assert!(
1823            db.add_pv_with_hook("A", EpicsValue::Double(2.0), noop_hook)
1824                .await
1825                .is_err()
1826        );
1827        assert!(
1828            db.add_record("A", Box::new(AiRecord::new(0.0)))
1829                .await
1830                .is_err()
1831        );
1832        assert!(db.add_alias("A", "A").await.is_err());
1833
1834        db.add_record("R", Box::new(AiRecord::new(0.0)))
1835            .await
1836            .unwrap();
1837        assert!(
1838            db.add_record("R", Box::new(AiRecord::new(1.0)))
1839                .await
1840                .is_err()
1841        );
1842        assert!(db.add_pv("R", EpicsValue::Double(0.0)).await.is_err());
1843        assert!(db.add_alias("R", "R").await.is_err());
1844
1845        db.add_alias("AL", "R").await.unwrap();
1846        assert!(db.add_pv("AL", EpicsValue::Double(0.0)).await.is_err());
1847        assert!(
1848            db.add_record("AL", Box::new(AiRecord::new(0.0)))
1849                .await
1850                .is_err()
1851        );
1852    }
1853
1854    /// Removing a record must purge aliases
1855    /// that pointed AT it. Otherwise the alias name stays
1856    /// "registered" forever and `add_pv` / `add_record` rejecting
1857    /// reuse causes a permanent name leak.
1858    #[tokio::test]
1859    async fn remove_record_purges_dangling_aliases() {
1860        use crate::server::records::ai::AiRecord;
1861
1862        let db = PvDatabase::new();
1863        db.add_record("R", Box::new(AiRecord::new(0.0)))
1864            .await
1865            .unwrap();
1866        db.add_alias("ALT1", "R").await.unwrap();
1867        db.add_alias("ALT2", "R").await.unwrap();
1868        // An alias that points elsewhere must NOT be touched.
1869        db.add_record("OTHER", Box::new(AiRecord::new(0.0)))
1870            .await
1871            .unwrap();
1872        db.add_alias("KEEPER", "OTHER").await.unwrap();
1873
1874        assert!(db.remove_record("R").await);
1875
1876        // Both aliases pointing at R should be gone — `add_pv` of
1877        // those names succeeds again.
1878        db.add_pv("ALT1", EpicsValue::Double(0.0)).await.unwrap();
1879        db.add_pv("ALT2", EpicsValue::Double(0.0)).await.unwrap();
1880        // The unrelated alias must survive.
1881        assert_eq!(db.resolve_alias("KEEPER").await, Some("OTHER".to_string()));
1882    }
1883
1884    /// `add_alias` must reject collisions with
1885    /// every namespace, including simple PVs (which the pre-fix
1886    /// code missed).
1887    #[tokio::test]
1888    async fn add_alias_rejects_simple_pv_collision() {
1889        use crate::server::records::ai::AiRecord;
1890
1891        let db = PvDatabase::new();
1892        db.add_pv("PVX", EpicsValue::Double(0.0)).await.unwrap();
1893        db.add_record("TARGET", Box::new(AiRecord::new(0.0)))
1894            .await
1895            .unwrap();
1896        // alias name "PVX" collides with the simple PV — must fail.
1897        assert!(db.add_alias("PVX", "TARGET").await.is_err());
1898    }
1899
1900    /// Concurrent `add_pv` and `add_record` with
1901    /// the same name must not deadlock and must serialize so that
1902    /// exactly one succeeds. Pre-fix the two methods grabbed
1903    /// different write locks first, opening a cross-lock-order
1904    /// deadlock window.
1905    #[tokio::test]
1906    async fn concurrent_add_pv_and_add_record_do_not_deadlock() {
1907        use crate::server::records::ai::AiRecord;
1908
1909        let db = std::sync::Arc::new(PvDatabase::new());
1910        let db1 = db.clone();
1911        let db2 = db.clone();
1912        let h1 = tokio::spawn(async move { db1.add_pv("RACE", EpicsValue::Double(1.0)).await });
1913        let h2 =
1914            tokio::spawn(async move { db2.add_record("RACE", Box::new(AiRecord::new(0.0))).await });
1915        // Both complete within a reasonable bound — pre-fix this
1916        // could hang because T1 holds simple_pvs.write and waits
1917        // for records.read while T2 holds records.write and waits
1918        // for simple_pvs.read.
1919        let r1 = tokio::time::timeout(std::time::Duration::from_secs(2), h1)
1920            .await
1921            .expect("add_pv must not block on add_record");
1922        let r2 = tokio::time::timeout(std::time::Duration::from_secs(2), h2)
1923            .await
1924            .expect("add_record must not block on add_pv");
1925        let r1 = r1.unwrap();
1926        let r2 = r2.unwrap();
1927        // Exactly one of the two wins; the other reports
1928        // "already registered".
1929        assert!(
1930            (r1.is_ok() && r2.is_err()) || (r1.is_err() && r2.is_ok()),
1931            "exactly one of the racing inserts must succeed: r1={r1:?} r2={r2:?}",
1932        );
1933    }
1934
1935    #[tokio::test]
1936    async fn existence_gate_blocks_cached_simple_pv_per_request() {
1937        // A cached simple PV must re-pass the installed existence gate on
1938        // both the search (`has_name_from`) and create (`find_entry_from`)
1939        // paths. Records bypass the gate. With no gate the short-circuit
1940        // is unchanged (plain-IOC behaviour).
1941        use std::net::SocketAddr;
1942
1943        let db = PvDatabase::new();
1944        db.add_pv("SHADOW:x", EpicsValue::Double(1.0))
1945            .await
1946            .unwrap();
1947        db.add_record(
1948            "REC",
1949            Box::new(crate::server::records::ai::AiRecord::new(0.0)),
1950        )
1951        .await
1952        .unwrap();
1953
1954        let denied: SocketAddr = "127.0.0.1:5064".parse().unwrap();
1955        let allowed: SocketAddr = "192.0.2.5:5064".parse().unwrap();
1956
1957        // No gate installed: the cached simple PV resolves unconditionally.
1958        assert!(db.has_name_from("SHADOW:x", Some(denied)).await);
1959        assert!(db.find_entry_from("SHADOW:x", Some(denied)).await.is_some());
1960
1961        // Gate denies the simple PV only for `denied` (the gateway's
1962        // host-scoped `.pvlist` admission has exactly this shape).
1963        let gate: ExistenceGate = Arc::new(move |name, peer| {
1964            Box::pin(async move { !(name == "SHADOW:x" && peer == Some(denied)) })
1965        });
1966        db.set_existence_gate(gate).await;
1967
1968        // Denied peer: does-not-exist on both paths despite the PV being
1969        // cached in `simple_pvs`.
1970        assert!(!db.has_name_from("SHADOW:x", Some(denied)).await);
1971        assert!(db.find_entry_from("SHADOW:x", Some(denied)).await.is_none());
1972
1973        // Allowed peer: still resolves.
1974        assert!(db.has_name_from("SHADOW:x", Some(allowed)).await);
1975        assert!(
1976            db.find_entry_from("SHADOW:x", Some(allowed))
1977                .await
1978                .is_some()
1979        );
1980
1981        // Records are never gateway-managed — the gate must not gate them
1982        // even for the denied peer.
1983        assert!(db.has_name_from("REC", Some(denied)).await);
1984        assert!(db.find_entry_from("REC", Some(denied)).await.is_some());
1985    }
1986
1987    /// `record_link_fields` must surface a record's device-support `INP`
1988    /// link. An `ai`'s `INP` is a `DBF_INLINK` field stored in
1989    /// `common.inp` — it is not a `DbFieldType::String` entry in
1990    /// `field_list()` — so the earlier `field_list()` scan filtered by
1991    /// `String` silently dropped it. The pvalink install scan walks this
1992    /// method, so a Passive `ai` carrying a CP/CPP pvalink `INP` never had
1993    /// its monitor opened at iocInit. Enumerating the canonical
1994    /// `common.inp` storage fixes it; C `dbpvar`/`dbcar` likewise dump
1995    /// every link field including device-support INP/OUT.
1996    #[tokio::test]
1997    async fn record_link_fields_surfaces_device_support_inp() {
1998        use crate::server::record::ParsedLink;
1999        use crate::server::records::ai::AiRecord;
2000
2001        let db = PvDatabase::new();
2002        db.add_record("AI", Box::new(AiRecord::new(0.0)))
2003            .await
2004            .unwrap();
2005        // Device-support INP lives in `common.inp` (DBF_INLINK), the
2006        // exact storage a `field_list()` String scan cannot reach.
2007        {
2008            let rec = db.get_record("AI").await.unwrap();
2009            rec.write().await.common.inp = "pva://mini:current?proc=CP".to_string();
2010        }
2011
2012        let links = db.record_link_fields("AI").await;
2013        let inp = links
2014            .iter()
2015            .find(|(f, _, _)| f == "INP")
2016            .unwrap_or_else(|| panic!("INP link must be surfaced, got {links:?}"));
2017        assert_eq!(inp.1, "pva://mini:current?proc=CP");
2018        assert!(
2019            matches!(inp.2, ParsedLink::Pva(_)),
2020            "a pva:// INP must parse to ParsedLink::Pva, got {:?}",
2021            inp.2
2022        );
2023    }
2024}