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