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