epics_base_rs/server/database/mod.rs
1use std::future::Future;
2
3pub mod db_access;
4mod field_io;
5pub mod filters;
6mod link_set;
7mod links;
8mod processing;
9mod record_lock;
10mod scan_index;
11
12pub use link_set::{
13 DynLinkSet, LinkDbfType, LinkMetadata, LinkPutOp, LinkSet, LinkSetRegistry, RemoteAlarm,
14};
15pub use processing::{AsyncDbHandle, AsyncToken};
16pub use record_lock::{ManyRecordWriteGuard, RecordWriteGuard};
17
18use crate::error::{CaError, CaResult};
19use crate::runtime::sync::RwLock;
20use std::collections::{BTreeSet, HashMap};
21use std::sync::Arc;
22
23use crate::server::pv::ProcessVariable;
24use crate::server::record::{Record, RecordInstance, ScanList};
25use crate::types::EpicsValue;
26
27/// What a `.db` definition carries into the creation sink alongside the record
28/// itself: the `dbCommon` fields `db_loader::apply_fields` could not route to
29/// the record's own `field_list`, and the record's `info(...)` tags.
30///
31/// It exists so that [`PvDatabase::add_loaded_record`] receives a record's
32/// COMPLETE loaded state in one call. A caller cannot add the record and then
33/// apply its `.db` fields, because the sink runs C's `iocInit` passes — whose
34/// result depends on those fields — before the record is reachable at all.
35#[derive(Default, Debug, Clone)]
36pub struct RecordLoad {
37 /// `dbCommon` fields, in `.db` file order (a later `field(UDF,…)` wins).
38 pub common_fields: Vec<(String, EpicsValue)>,
39 /// `info(key, "value")` tags.
40 pub info_tags: Vec<(String, String)>,
41}
42
43impl RecordLoad {
44 /// The common fields alone — the shape every `.db` loader path produces
45 /// from [`crate::server::db_loader::apply_fields`].
46 pub fn from_common_fields(common_fields: Vec<(String, EpicsValue)>) -> Self {
47 Self {
48 common_fields,
49 info_tags: Vec::new(),
50 }
51 }
52}
53
54/// Parse a PV name into (base_name, field_name).
55/// "TEMP.EGU" → ("TEMP", "EGU")
56/// "TEMP" → ("TEMP", "VAL")
57pub fn parse_pv_name(name: &str) -> (&str, &str) {
58 match name.rsplit_once('.') {
59 Some((base, field)) => (base, field),
60 None => (name, "VAL"),
61 }
62}
63
64/// C `dbIsValueField` (`dbAccess.c:463-469`): is this field the record
65/// type's *value* field?
66///
67/// A record type's value field is the one the DBD names `VAL` — the DBD
68/// parser records exactly that field's index as `indvalFlddes`
69/// (`dbLexRoutines.c:777-780`), which is what `dbIsValueField` compares
70/// against. Metadata that C/pvxs apply "to VAL only" (e.g. QSRV's
71/// `Q:form` → `display.form.index`, `iocsource.cpp:53`) key on this
72/// predicate, so it lives beside [`parse_pv_name`], whose `"REC"` → `VAL`
73/// default is the other half of the same rule.
74pub fn is_value_field(field: &str) -> bool {
75 field.eq_ignore_ascii_case("VAL")
76}
77
78/// Apply timestamp to a record based on its TSE field.
79/// `is_soft` indicates a Soft Channel device type.
80///
81/// Mirrors C `recGblGetTimeStampSimm` (recGbl.c:310-343). The TSE
82/// constants are defined in `epicsTime.h:102-104`:
83///
84/// - `epicsTimeEventCurrentTime = 0` → wall-clock now
85/// - `epicsTimeEventBestTime = -1` → generalTime BestTime providers
86/// - `epicsTimeEventDeviceTime = -2` → device support already set time
87/// - `1..` → event-number providers
88///
89/// The C path is symmetric: every non-`-2` case unconditionally
90/// overwrites `precord->time` via `epicsTimeGetEvent(tse)`, which
91/// delegates to `epicsTimeGetCurrent` for `tse==0` and to
92/// `generalTimeGetEventPriority` otherwise. Only `-2` (device time)
93/// is left untouched because the device support has already written
94/// the timestamp before `recGblGetTimeStamp` is called.
95fn apply_timestamp(common: &mut super::record::CommonFields, _is_soft: bool) {
96 // Single owner of TSE -> TIME resolution; device support that must
97 // format the record's resolved time during `read()` routes through the
98 // same helper so the two never drift (see `recgbl::get_time_stamp`).
99 // For TSE=-2 the helper returns `common.time` unchanged, preserving the
100 // device-time "leave it alone" semantics.
101 common.time = crate::server::recgbl::get_time_stamp(common.tse, common.time);
102}
103
104/// Unified entry in the PV database.
105pub enum PvEntry {
106 Simple(Arc<ProcessVariable>),
107 Record(Arc<RwLock<RecordInstance>>),
108}
109
110/// Callback for resolving external PV names (CA/PVA links).
111/// Returns the current value of the external PV, or None if unavailable.
112///
113/// **Async**, for the same reason [`LinkSet`] is: an external link's value
114/// lives on a tokio runtime, and a sync closure could only reach it through a
115/// blocking bridge that panics on a current-thread runtime. The one caller
116/// ([`PvDatabase::resolve_external_pv`]) is already async.
117pub type ExternalPvResolver = Arc<
118 dyn for<'a> Fn(
119 &'a str,
120 )
121 -> std::pin::Pin<Box<dyn Future<Output = Option<EpicsValue>> + Send + 'a>>
122 + Send
123 + Sync,
124>;
125
126/// Async hook invoked by [`PvDatabase::has_name`] when a name is not yet
127/// in the database. Used by the CA gateway and similar proxy components
128/// to lazily populate PVs on first search.
129///
130/// The resolver should:
131/// 1. Determine whether the name should be served (e.g., check ACL)
132/// 2. Take whatever action is needed to make `has_name` return true on
133/// a subsequent call (e.g., subscribe to an upstream IOC and call
134/// `add_pv` with a placeholder value)
135/// 3. Return `true` if the name is now resolvable, `false` otherwise
136///
137/// Returning `true` causes `has_name` to re-check the database. The
138/// resolver may take some time (TCP search, upstream connect handshake);
139/// the caller (UDP search responder, TCP CREATE_CHANNEL handler) will
140/// `.await` it.
141/// The second argument is the downstream client's socket address when
142/// the lookup originates from a CA/PVA search or channel-create on
143/// behalf of an identified peer (`None` for host-less internal lookups:
144/// preload, iocsh, link processing). the CA gateway needs
145/// this to evaluate `.pvlist` `DENY FROM host` rules at search time, the
146/// way C ca-gateway's `pvExistTest` passes the client host to
147/// `gateAs::findEntry`.
148pub type SearchResolver = Arc<
149 dyn Fn(
150 String,
151 Option<std::net::SocketAddr>,
152 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>
153 + Send
154 + Sync,
155>;
156
157/// Per-request admission gate for an **already-registered** simple PV.
158///
159/// A plain IOC's simple PVs are authoritative: once registered they
160/// exist unconditionally, so no gate is installed and the cached-PV
161/// short-circuit in [`PvDatabase::find_entry_from`] /
162/// [`PvDatabase::has_name_from`] is unchanged. A CA gateway is
163/// different — its shadow PVs are projections of an upstream that can be
164/// host-denied for a given requester or disconnected — so it installs a
165/// gate that the lookup path consults *before* returning a cached simple
166/// PV. Returning `false` makes the database answer "does not exist" for
167/// that requester, exactly as C ca-gateway's `pvExistTest` returns
168/// `pverDoesNotExistHere` for a host-denied or disconnected PV
169/// (`gateServer.cc:1516-1637`) — without removing the PV object, so its
170/// cached value stays available for diagnostics and re-admission.
171///
172/// The first argument is the filter-suffix-stripped record path (the
173/// same key the simple-PV map and the gateway cache use); the second is
174/// the requesting peer (`None` for host-less internal lookups). The gate
175/// governs **only** simple PVs — records and aliases are never
176/// gateway-managed and bypass it.
177pub type ExistenceGate = Arc<
178 dyn Fn(
179 String,
180 Option<std::net::SocketAddr>,
181 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>>
182 + Send
183 + Sync,
184>;
185
186/// Internal state of [`PvDatabase`].
187///
188/// # Invariant — alias-aware lookup (epics-base PR #336)
189///
190/// **MUST**: every record-name lookup that originates from an
191/// external API (CA/PVA server, link processing, iocsh, bridge
192/// providers) MUST go through [`PvDatabase::get_record`] /
193/// [`PvDatabase::find_entry`] / [`PvDatabase::has_name`], never
194/// `inner.records.read().await.get(...)` directly.
195///
196/// **MUST NOT**: a function that takes an arbitrary record-name
197/// `&str` and reads `inner.records` directly, unless one of:
198/// - the function is itself an alias-management primitive
199/// (`add_record`, `remove_record`, `add_alias`,
200/// `find_entry_no_resolve`, `has_name_no_resolve`,
201/// `get_record_no_resolve`, `all_record_names`), OR
202/// - the name has been normalised to canonical earlier in the
203/// same scope (the `let canonical_owned; let name: &str = ...`
204/// pattern in `process_record_with_links_inner` /
205/// `complete_async_record_inner` / `put_record_field_from_ca` /
206/// `put_pv`).
207///
208/// **Owner/Gate:** `PvDatabase::get_record` (alias-aware path).
209///
210/// New code that adds a record-name entry point should call
211/// `get_record` first OR run the canonical-normalisation snippet
212/// at function entry. Direct `inner.records` access is reserved
213/// for the alias-management primitives listed above.
214/// One CP/CPP edge in the [`PvDatabaseInner::cp_links`] index: the record
215/// to (re)process when the source record changes.
216///
217/// `passive_only` distinguishes CPP from CP. C adds the `CA_DBPROCESS`
218/// action for a CP link unconditionally, but for a CPP link only when the
219/// link-holding record's `SCAN` is Passive (`dbCa.c:854,994,1072`). CP
220/// edges clear the flag; CPP edges set it, and `dispatch_cp_targets`
221/// honours it.
222#[derive(Clone, Debug)]
223pub struct CpTarget {
224 pub record: String,
225 pub passive_only: bool,
226}
227
228struct PvDatabaseInner {
229 simple_pvs: RwLock<HashMap<String, Arc<ProcessVariable>>>,
230 records: RwLock<HashMap<String, Arc<RwLock<RecordInstance>>>>,
231 /// Scan index: maps scan type → sorted set of
232 /// `(PHAS, load_order, record_name)`.
233 ///
234 /// C parity (`dbScan.c:1052-1095`): `buildScanLists` walks records
235 /// in database / record-type **load order** and `addToList`
236 /// inserts each after the last element with `phas <= precord->phas`
237 /// — so within one PHAS value the scan list is a stable FIFO in
238 /// load order. The secondary sort key is the per-record
239 /// `load_order` sequence (NOT the record name), so two records
240 /// sharing a PHAS scan in the order they were loaded, matching a
241 /// C IOC built from the same `.db` file.
242 /// Keyed by [`ScanList`], not `ScanType`: a `Passive` or illegal SCAN names
243 /// no list (C `scanAdd`, dbScan.c:241-251) and so cannot be a key at all.
244 scan_index: RwLock<HashMap<ScanList, BTreeSet<(i16, u64, String)>>>,
245 /// Per-record load-order sequence number, assigned monotonically
246 /// at `add_record`. Used as the secondary scan-index sort key so
247 /// same-PHAS records preserve database load order. Survives a
248 /// `remove_record` + re-`add_record` (the re-add gets a fresh,
249 /// higher sequence — matching a fresh `.db` reload).
250 load_order: RwLock<HashMap<String, u64>>,
251 /// Monotonic counter feeding `load_order`.
252 load_order_counter: std::sync::atomic::AtomicU64,
253 /// CP/CPP link index: maps source_record → target edges to process when
254 /// the source changes. Each edge carries the CP-vs-CPP distinction (see
255 /// [`CpTarget`]).
256 cp_links: RwLock<HashMap<String, Vec<CpTarget>>>,
257 /// External (CA/PVA) CP/CPP link index: maps the *external PV name*
258 /// (the cross-IOC source, e.g. `OTHER:PV` from `INP="OTHER:PV CP"`)
259 /// → holder edges to process when that remote PV changes. The local
260 /// [`Self::cp_links`] index is keyed by a local source RECORD that
261 /// processes here; a cross-IOC source never processes locally, so its
262 /// only trigger is the calink/pvalink CA monitor callback, which calls
263 /// [`PvDatabase::dispatch_external_cp_targets`]. Parity with C
264 /// `dbCa.c:993-994` `eventCallback` adding `CA_DBPROCESS`.
265 external_cp_links: RwLock<HashMap<String, Vec<CpTarget>>>,
266 /// Alias map: alternate-name → real-record-name. Mirrors epics-base
267 /// PR #336 (alias name validation + parsing). `find_entry` and
268 /// related lookups consult this map after the canonical record
269 /// table so an alias resolves transparently to its target.
270 aliases: RwLock<HashMap<String, String>>,
271 /// Single gate that serializes
272 /// every `add_pv` / `add_pv_with_hook` / `add_record` /
273 /// `add_alias` / `remove_record` / `remove_simple_pv` /
274 /// `remove_alias`. Without this, the per-method write-lock
275 /// orders (`simple_pvs` first vs. `records` first vs.
276 /// `aliases` first) could deadlock under concurrent registrations,
277 /// and `add_record`'s post-insert `scan_index.write()` had a
278 /// TOCTOU window where `remove_record` could land between the
279 /// records map insert and the scan-index insert and leave a
280 /// phantom scan entry.
281 ///
282 /// Holding this mutex makes the cross-namespace `check_name_free`
283 /// peek atomic with the target-map insert, eliminates the
284 /// scan-index race, and lets `remove_*` purge dangling aliases
285 /// without a second pass.
286 registration_mutex: tokio::sync::Mutex<()>,
287 /// The IOC lifecycle phase — the port's `iocInit` boundary. See
288 /// [`DbInitPhase`], [`PvDatabase::begin_load`],
289 /// [`PvDatabase::schedule_record_init`] and [`PvDatabase::ioc_init`].
290 init_phase: std::sync::Mutex<DbInitPhase>,
291 /// Lines queued by the iocsh `afterIocRunning <command>` directive
292 /// (epics-base PR #558). Drained by the IOC application after PINI
293 /// completes, then re-executed through a fresh IocShell so the
294 /// commands run with the database in its post-init state.
295 after_ioc_running: std::sync::Mutex<Vec<String>>,
296 /// Optional resolver for external PVs (ca://, pva:// links).
297 external_resolver: RwLock<Option<ExternalPvResolver>>,
298 /// Optional async resolver invoked on `has_name` misses (e.g. CA gateway).
299 search_resolver: RwLock<Option<SearchResolver>>,
300 /// Optional per-request gate consulted before a *cached* simple PV is
301 /// advertised as existing (e.g. CA gateway host/state admission). See
302 /// [`ExistenceGate`]. `None` for a plain IOC (short-circuit unchanged).
303 existence_gate: RwLock<Option<ExistenceGate>>,
304 /// Per-scheme link sets — pluggable backends for `pva://` /
305 /// `ca://` link resolution. Consulted before the legacy
306 /// [`ExternalPvResolver`] in [`Self::resolve_external_pv`].
307 /// Mirrors the C-EPICS lset abstraction.
308 link_sets: RwLock<link_set::LinkSetRegistry>,
309 /// True once the ScanScheduler has been started for this DB.
310 /// Prevents duplicate scan tasks when multiple protocol servers (CA + PVA)
311 /// both try to start scanning on the same DB.
312 scan_started: std::sync::atomic::AtomicBool,
313 /// True once PINI processing has completed. Non-owner schedulers await
314 /// this before running their hooks, preserving the "PINI before hooks"
315 /// ordering contract.
316 pini_done: std::sync::atomic::AtomicBool,
317 /// Fired by the scan owner after PINI completes. Non-owners register
318 /// interest on this before re-checking `pini_done` to avoid missing the
319 /// signal (`notify_waiters` does not store a permit).
320 pini_notify: tokio::sync::Notify,
321 /// Per-record advisory write gates — the Rust
322 /// counterpart of the C-EPICS `dbScanLock` / `dbLocker`
323 /// machinery. Every plain CA/PVA write, the QSRV atomic group
324 /// PUT/GET, and the pvalink atomic scan-on-update epoch all
325 /// acquire these gates, so no two of them can interleave on a
326 /// shared record. See [`record_lock`].
327 record_locks: record_lock::RecordLockRegistry,
328 /// Subroutine functions by name, retained at runtime so the processing
329 /// path can re-resolve an aSub's subroutine when its name changes
330 /// (C `aSubRecord.c::fetch_values` `registryFunctionFind`, LFLG=READ /
331 /// SUBL). Populated once at iocInit from the IocApp/IocBuilder registry;
332 /// read-only thereafter.
333 subroutine_registry: RwLock<HashMap<String, Arc<crate::server::record::SubroutineFn>>>,
334 /// Breakpoint tables by name (C `bptList`), shared by every db-load path so
335 /// `ai`/`ao` records with `LINR >= 3` resolve their linearisation table. An
336 /// `Arc` snapshot is installed on each record at creation; the master grows
337 /// (copy-on-write via [`PvDatabase::add_breaktables`]) as `dbLoadRecords`
338 /// loads more `breaktable(...)` definitions, so build-time and runtime
339 /// loads share one registry.
340 breaktable_registry: RwLock<Arc<crate::server::cvt_bpt::BreakTableRegistry>>,
341}
342
343/// Database of all process variables hosted by this server.
344#[derive(Clone)]
345pub struct PvDatabase {
346 inner: Arc<PvDatabaseInner>,
347}
348
349/// A record initialisation owed to `iocInit` — the port's `init_record`
350/// tail. Built by a record's `refresh_link_status` and handed to
351/// [`PvDatabase::schedule_record_init`].
352type RecordInit = std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>;
353
354/// The IOC lifecycle phase, and with it the answer to "may a record's links be
355/// classified against the database as it stands right now?".
356///
357/// C runs `init_record` — where a record classifies its links (`checkLinks`,
358/// `dbNameToAddr`) — from `iocInit`, i.e. after EVERY `dbLoadRecords` block
359/// has been read. A forward reference across two `dbLoadRecords` calls in one
360/// `st.cmd` is therefore a LOCAL link, deterministically, and the classified
361/// value is final the moment `iocInit` returns (`dbgf` is refused before it).
362///
363/// The boundary is `iocInit`, NOT a load group: gating on the load group left
364/// the multi-`dbLoadRecords` case every real `st.cmd` uses racing 9-in-15
365/// (R18-92). So the phase here is an ioc-lifecycle state.
366///
367/// # The lifecycle is ONE-WAY: `Unloaded → Loading → Running`
368///
369/// R18-92 modelled it with two states, `Loading` and `Complete`, where
370/// `Complete` meant BOTH "never loaded" and "iocInit has run" — so `begin_load`
371/// needed a `Complete → Loading` arm to open the phase at all, and that arm ran
372/// on a post-iocInit load too. One `dbLoadRecords` typed after `iocInit` then
373/// re-armed the queue that only `ioc_init` drains, and every later
374/// classification — including every runtime `special()` link re-point — was
375/// pushed into a `Vec` nothing polls (R19-62, measured: `iocInit;
376/// dbLoadRecords(b.db); dbpf CO.INPA "9.5"` froze `CO.INAV` at 0).
377///
378/// Splitting the two meanings is what closes it: `Loading` is now produced ONLY
379/// from `Unloaded`, so no function in the crate can transition backwards out of
380/// `Running`. The one-way-ness is a property of the transitions that exist, not
381/// of a runtime check.
382enum DbInitPhase {
383 /// No load has begun. `iocInit` is owed nothing, so a classification runs
384 /// immediately — a programmatically built or unit-test database.
385 Unloaded,
386 /// Between the first `dbLoadRecords`/builder load and `iocInit`; holds the
387 /// classifications owed, in issue order. A half-built database is never
388 /// observed, because no classification code runs against one.
389 Loading(Vec<RecordInit>),
390 /// `iocInit` has run: the database is final and every link status is
391 /// classified. A classification issued now runs immediately, which is what a
392 /// runtime re-point (`special()` on a link field) needs. TERMINAL — nothing
393 /// re-opens the load phase.
394 Running,
395}
396
397/// [`PvDatabase::begin_load`] was called on a database whose `iocInit` has
398/// already run — C's `getIocState() != iocVoid` (R19-63).
399///
400/// The `Display` text is C's `errSymMsg(S_dbLib_postInitRecRegister)` verbatim
401/// (`dbStaticLib.h:269`), which is what `dbCreateRecord` prints:
402///
403/// ```text
404/// epics> dbCreateRecord(pdbbase,"ai","NEWREC")
405/// ERROR: 33554463 IOC already initialized - No new records can be added
406/// ```
407#[derive(Debug, Clone, Copy, PartialEq, Eq)]
408pub struct IocAlreadyInitialized;
409
410impl std::fmt::Display for IocAlreadyInitialized {
411 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
412 f.write_str("IOC already initialized - No new records can be added")
413 }
414}
415
416impl std::error::Error for IocAlreadyInitialized {}
417
418/// Which record kind a SELM link selection is being computed for.
419/// The Specified/Mask base differs between record types in C, so the
420/// shared selector must know the caller.
421#[derive(Clone, Copy, PartialEq, Eq, Debug)]
422pub(crate) enum SelmKind {
423 /// `fanout` / `seq`: Specified index is `SELN + OFFS` (0-based over
424 /// LNK0..LNKF / group 0..15); Mask is shifted by `SHFT`.
425 /// Mirrors `fanoutRecord.c:106-141` and `seqRecord.c:147-178`.
426 FanoutSeq,
427 /// `dfanout`: Specified index is `SELN - 1` (1-based, `SELN==0`
428 /// means "drive nothing", `SELN > OUT_ARG_MAX` is invalid); Mask
429 /// has NO `SHFT` and `SELN==0` means "no output".
430 /// Mirrors `dfanoutRecord.c:307-339`.
431 Dfanout,
432}
433
434/// Result of resolving a SELM/SELN selection.
435#[derive(Clone, Debug, Default)]
436pub(crate) struct SelmResult {
437 /// 0-based link indices to drive (into the LNK0../OUTA.. array).
438 pub indices: Vec<usize>,
439 /// `Some` when C would raise an alarm for an out-of-range
440 /// `SELN`/`OFFS`/`SHFT`. C uses `recGblSetSevr(prec, SOFT_ALARM,
441 /// INVALID_ALARM)` in every such path.
442 pub alarm: Option<(u16, crate::server::record::AlarmSeverity)>,
443}
444
445/// Convert a link value to `epicsUInt16` with C `dbGetLink(.., DBR_USHORT,
446/// ..)` semantics, for the fanout/dfanout/seq `SELL`→`SELN` read — so a
447/// constant, DB, CA, or PVA link source all convert by the one rule C applies
448/// through `dbFastGetConvertRoutine`.
449///
450/// # The source type decides the rule, because in C it decides the routine
451///
452/// `dbFastGetConvertRoutine` is a 2-D table indexed by *both* the source DBF
453/// and the destination DBR (`dbConvert.c:1571-1638`): a `DBF_LONG` source
454/// reaches `getLongUshort`, a `DBF_DOUBLE` source reaches `getDoubleUshort`.
455/// They are different functions, and C gives them different semantics:
456///
457/// * **Integer source** — `(epicsUInt16)(epicsInt32)v`. Conversion of an
458/// out-of-range *integer* to an unsigned type is **defined** in C
459/// (C17 6.3.1.3p2: reduce modulo `USHRT_MAX + 1`). Every compiler and
460/// every target agrees, so this is a real contract and the port keeps it:
461/// `SELL` pointing at a `DBF_LONG` field holding `-1` gives `SELN = 65535`.
462/// * **Float source** — `(epicsUInt16)d`. Conversion of an out-of-range
463/// *float* is **undefined** (C17 6.3.1.4p1), so compiled C is not
464/// single-valued: x86-64 wraps, aarch64 saturates. What the port does about
465/// that is [`crate::types::c_cast`]'s call — the single owner of the policy —
466/// and deliberately not restated here.
467///
468/// Both rules already live in [`EpicsValue::convert_to`], the single
469/// value-coercion owner: it takes the integer view (`as_int_i64`) when the
470/// source has one and falls back to `c_cast` only for a genuine float. So this
471/// is a thin projection onto that owner, NOT a second conversion table.
472///
473/// The previous revision called `c_cast::f64_to_u16(value.to_f64())` directly,
474/// bypassing the owner — which silently applied the float rule to integer
475/// sources too, losing the one wrap C actually defines.
476pub(crate) fn dbr_ushort_cast(value: &EpicsValue) -> u16 {
477 match value.convert_to(crate::types::DbFieldType::UShort) {
478 EpicsValue::UShort(v) => v,
479 // A link that delivers an array converts element-wise; C's
480 // `dbGetLink(.., &prec->seln, 0, 0)` requests ONE element, so SELN
481 // takes the first (an empty array leaves it 0).
482 EpicsValue::UShortArray(v) => v.first().copied().unwrap_or(0),
483 // `convert_to(UShort)` returns no other variant.
484 _ => 0,
485 }
486}
487
488/// Select which link indices are active based on SELM/SELN, applying
489/// the record-type-specific `OFFS`/`SHFT` bias.
490///
491/// SELM: 0 = All, 1 = Specified, 2 = Mask. `count` is the number of
492/// link slots (16 for fanout/dfanout/seq).
493///
494/// `seln` is the native `DBF_USHORT` value: C declares `SELN` as
495/// `epicsUInt16`, so every comparison below is unsigned, matching C's
496/// selection arithmetic — never `-1`. What an out-of-range `SELL` converts
497/// *to* is [`dbr_ushort_cast`]'s decision, not this function's.
498///
499/// C references:
500/// * fanout — `fanoutRecord.c:106-141`
501/// * dfanout — `dfanoutRecord.c:307-339`
502/// * seq — `seqRecord.c:147-178`
503pub(crate) fn select_link_indices_ex(
504 kind: SelmKind,
505 selm: i16,
506 seln: u16,
507 offs: i16,
508 shft: i16,
509 count: usize,
510) -> SelmResult {
511 use crate::server::recgbl::alarm_status::SOFT_ALARM;
512 use crate::server::record::AlarmSeverity;
513
514 let invalid = || SelmResult {
515 indices: Vec::new(),
516 alarm: Some((SOFT_ALARM, AlarmSeverity::Invalid)),
517 };
518 let ok = |indices: Vec<usize>| SelmResult {
519 indices,
520 alarm: None,
521 };
522
523 match selm {
524 // All — every slot.
525 0 => ok((0..count).collect()),
526 // Specified.
527 1 => match kind {
528 SelmKind::FanoutSeq => {
529 // C: `i = seln + offs;` with `seln` unsigned (epicsUInt16),
530 // 0-based; `i<0 || i>=NLINKS` → INVALID. So `SELN=65535`
531 // (from `SELL=-1`) yields `i>=NLINKS` → INVALID, never
532 // drives link 0.
533 let i = seln as i32 + offs as i32;
534 if i < 0 || i >= count as i32 {
535 invalid()
536 } else {
537 ok(vec![i as usize])
538 }
539 }
540 SelmKind::Dfanout => {
541 // C `dfanoutRecord.c:315-320`: `if (prec->seln > OUT_ARG_MAX)`
542 // with `seln` unsigned → INVALID; `seln == 0` → no output;
543 // otherwise drive `seln - 1`. OFFS is not a dfanout field.
544 // `SELL=-1` → `SELN=65535` > count → INVALID (the signed
545 // read used to see `-1`, take the `<= 0` branch, and drive
546 // nothing with no alarm).
547 let seln_i = seln as i32;
548 if seln_i > count as i32 {
549 invalid()
550 } else if seln_i == 0 {
551 ok(Vec::new())
552 } else {
553 ok(vec![(seln_i - 1) as usize])
554 }
555 }
556 },
557 // Mask.
558 2 => {
559 let mask: u32 = match kind {
560 SelmKind::FanoutSeq => {
561 // C: SHFT shift first, with `shft` range-checked to [-15,15].
562 if !(-15..=15).contains(&shft) {
563 return invalid();
564 }
565 let raw = seln as u32;
566 if shft >= 0 {
567 raw >> shft
568 } else {
569 raw << (-shft)
570 }
571 }
572 // dfanout Mask has no SHFT.
573 SelmKind::Dfanout => seln as u32,
574 };
575 ok((0..count).filter(|i| mask & (1 << i) != 0).collect())
576 }
577 // Any other SELM value → C `default:` raises INVALID.
578 _ => invalid(),
579 }
580}
581
582impl PvDatabase {
583 pub fn new() -> Self {
584 Self {
585 inner: Arc::new(PvDatabaseInner {
586 simple_pvs: RwLock::new(HashMap::new()),
587 external_resolver: RwLock::new(None),
588 search_resolver: RwLock::new(None),
589 existence_gate: RwLock::new(None),
590 link_sets: RwLock::new(link_set::LinkSetRegistry::new()),
591 records: RwLock::new(HashMap::new()),
592 scan_index: RwLock::new(HashMap::new()),
593 load_order: RwLock::new(HashMap::new()),
594 load_order_counter: std::sync::atomic::AtomicU64::new(0),
595 cp_links: RwLock::new(HashMap::new()),
596 external_cp_links: RwLock::new(HashMap::new()),
597 aliases: RwLock::new(HashMap::new()),
598 registration_mutex: tokio::sync::Mutex::new(()),
599 init_phase: std::sync::Mutex::new(DbInitPhase::Unloaded),
600 after_ioc_running: std::sync::Mutex::new(Vec::new()),
601 scan_started: std::sync::atomic::AtomicBool::new(false),
602 pini_done: std::sync::atomic::AtomicBool::new(false),
603 pini_notify: tokio::sync::Notify::new(),
604 record_locks: record_lock::RecordLockRegistry::default(),
605 subroutine_registry: RwLock::new(HashMap::new()),
606 breaktable_registry: RwLock::new(Arc::new(
607 crate::server::cvt_bpt::BreakTableRegistry::new(),
608 )),
609 }),
610 }
611 }
612
613 /// Merge `tables` into the shared breakpoint-table registry (C `bptList`
614 /// accumulation across `dbLoadDatabase`/`dbLoadRecords`) and return the new
615 /// snapshot. Copy-on-write: a new merged registry replaces the old one.
616 ///
617 /// `add_breaktables` is the single registry-mutation owner, so it also
618 /// restores the invariant *every record can resolve against the current
619 /// registry* on mutation: the new snapshot is re-installed into every
620 /// existing record. That covers a record created before its table was
621 /// loaded (an inline record added before `dbLoadRecords`, or a merge-reload
622 /// that repoints `LINR` to a table loaded in the same command) — neither
623 /// of which goes back through `add_record`'s install. `install_*` is a
624 /// no-op for non-ai/ao records and resets the cached table so the new
625 /// registry wins. Returns the current snapshot unchanged when `tables` is
626 /// empty (no mutation, so no re-install).
627 pub async fn add_breaktables(
628 &self,
629 tables: Vec<crate::server::cvt_bpt::BrkTable>,
630 ) -> Arc<crate::server::cvt_bpt::BreakTableRegistry> {
631 // Hold the registration gate across the registry write AND the record
632 // snapshot below so this mutation cannot interleave with `add_record`'s
633 // [registry read -> records-map insert] — both are gated by the same
634 // mutex. Without it a record created concurrently could read the
635 // pre-mutation registry (miss the just-loaded table) while not yet
636 // being in the records map for the re-install below, leaving a
637 // table-not-found alarm until the next load / LINR put. `add_record`
638 // holds this gate across its whole body (registry read + map insert),
639 // so taking it here closes that TOCTOU window. No `add_breaktables`
640 // caller already holds the gate, so this is reentrancy-safe.
641 let _gate = self.inner.registration_mutex.lock().await;
642 let snapshot = {
643 let mut guard = self.inner.breaktable_registry.write().await;
644 if tables.is_empty() {
645 return guard.clone();
646 }
647 let mut next = (**guard).clone();
648 for table in tables {
649 next.insert(table);
650 }
651 let snapshot = Arc::new(next);
652 *guard = snapshot.clone();
653 snapshot
654 };
655 // Re-install into existing records. Snapshot the instance handles
656 // under a brief read, then release the map lock BEFORE taking any
657 // per-record write lock — collect-then-act, keeping the invariant
658 // "never hold the records-map lock across a per-record lock" uniform
659 // across the codebase (a7f5a74f). This is defensive: no current path
660 // takes the per-record lock then the records-map lock, so there is no
661 // confirmed cycle; uniform order forecloses one. Same idiom as
662 // `all_record_names`. (The registry write lock was released above.)
663 let instances: Vec<_> = self.inner.records.read().await.values().cloned().collect();
664 for inst in instances {
665 inst.write()
666 .await
667 .record
668 .install_breaktable_registry(snapshot.clone());
669 }
670 snapshot
671 }
672
673 /// Install the by-name subroutine registry, retained for runtime
674 /// re-resolution (aSub LFLG=READ / SUBL). Called once at iocInit with the
675 /// IocApp/IocBuilder registry. See [`Self::find_subroutine_named`].
676 pub async fn install_subroutine_registry(
677 &self,
678 registry: HashMap<String, Arc<crate::server::record::SubroutineFn>>,
679 ) {
680 *self.inner.subroutine_registry.write().await = registry;
681 }
682
683 /// Look up a registered subroutine by name. The processing path uses this
684 /// to re-resolve an aSub's subroutine when SNAM changes (C `fetch_values`
685 /// `registryFunctionFind`). `None` when the name is not registered, which
686 /// the caller treats as C's `S_db_BadSub` (skip running the subroutine).
687 pub(crate) async fn find_subroutine_named(
688 &self,
689 name: &str,
690 ) -> Option<Arc<crate::server::record::SubroutineFn>> {
691 self.inner
692 .subroutine_registry
693 .read()
694 .await
695 .get(name)
696 .cloned()
697 }
698
699 /// Atomically claim the right to start the scan scheduler for this DB.
700 /// Returns `true` on the first call, `false` on subsequent calls.
701 /// Used by `ScanScheduler::run_with_hooks` to prevent duplicate scan tasks
702 /// when multiple protocol servers (CA + PVA) both try to start scanning.
703 pub fn try_claim_scan_start(&self) -> bool {
704 self.inner
705 .scan_started
706 .compare_exchange(
707 false,
708 true,
709 std::sync::atomic::Ordering::AcqRel,
710 std::sync::atomic::Ordering::Acquire,
711 )
712 .is_ok()
713 }
714
715 /// Mark PINI processing complete. Wakes any non-owner scan schedulers
716 /// that were waiting before running their hooks.
717 pub fn mark_pini_done(&self) {
718 self.inner
719 .pini_done
720 .store(true, std::sync::atomic::Ordering::Release);
721 self.inner.pini_notify.notify_waiters();
722 }
723
724 /// Wait until the scan owner has completed PINI processing.
725 /// Returns immediately if PINI has already completed.
726 pub async fn wait_for_pini(&self) {
727 if self
728 .inner
729 .pini_done
730 .load(std::sync::atomic::Ordering::Acquire)
731 {
732 return;
733 }
734 // Register interest BEFORE re-checking the flag to avoid missing a
735 // signal that arrives between the load and the await — `notify_waiters`
736 // does not store a permit for late subscribers.
737 let notified = self.inner.pini_notify.notified();
738 if self
739 .inner
740 .pini_done
741 .load(std::sync::atomic::Ordering::Acquire)
742 {
743 return;
744 }
745 notified.await;
746 }
747
748 /// Install an async resolver invoked when [`PvDatabase::has_name`]
749 /// fails to find a name. Used by proxy/gateway implementations to
750 /// lazily populate PVs on first search.
751 pub async fn set_search_resolver(&self, resolver: SearchResolver) {
752 *self.inner.search_resolver.write().await = Some(resolver);
753 }
754
755 /// Remove the previously installed search resolver, if any.
756 pub async fn clear_search_resolver(&self) {
757 *self.inner.search_resolver.write().await = None;
758 }
759
760 /// Install the per-request existence gate (see [`ExistenceGate`]).
761 /// Replaces any previously installed gate. Used by the CA gateway so
762 /// a cached shadow PV re-runs host/state admission per request.
763 pub async fn set_existence_gate(&self, gate: ExistenceGate) {
764 *self.inner.existence_gate.write().await = Some(gate);
765 }
766
767 /// Remove the previously installed existence gate, if any.
768 pub async fn clear_existence_gate(&self) {
769 *self.inner.existence_gate.write().await = None;
770 }
771
772 /// True when a cached simple PV named `name` must be treated as
773 /// non-existent for `peer` because the installed [`ExistenceGate`]
774 /// denied it. Always `false` when no gate is installed (a plain IOC)
775 /// or when `name` does not resolve to a simple PV — records and
776 /// aliases are never gateway-managed and bypass the gate.
777 ///
778 /// The single consultation point for the gate, shared by
779 /// [`Self::find_entry_from`] and [`Self::has_name_from`] so the
780 /// "cached simple PV ⇒ exists" short-circuit is closed uniformly on
781 /// both the create and search paths.
782 async fn simple_pv_gate_denies(&self, name: &str, peer: Option<std::net::SocketAddr>) -> bool {
783 let gate = match self.inner.existence_gate.read().await.clone() {
784 Some(g) => g,
785 None => return false,
786 };
787 // Strip the channel-filter suffix exactly as the lookups do
788 // (CA-FR-8) so the gate sees the same record-path key the
789 // simple-PV map and the gateway cache are keyed on.
790 let record_path = filters::split_channel_name(name).record_path;
791 if !self
792 .inner
793 .simple_pvs
794 .read()
795 .await
796 .contains_key(record_path.as_str())
797 {
798 return false;
799 }
800 !gate(record_path, peer).await
801 }
802
803 /// Set an external PV resolver for CA/PVA link resolution.
804 /// The resolver is called synchronously from link reads.
805 pub async fn set_external_resolver(&self, resolver: ExternalPvResolver) {
806 *self.inner.external_resolver.write().await = Some(resolver);
807 }
808
809 /// Register a [`LinkSet`] under `scheme` (e.g. `"pva"` /
810 /// `"ca"`). The lset is consulted for `ParsedLink::Pva` /
811 /// `ParsedLink::Ca` link reads/writes before falling back to
812 /// the legacy [`ExternalPvResolver`]. Subsequent calls for the
813 /// same scheme replace the previous binding.
814 pub async fn register_link_set(&self, scheme: &str, lset: link_set::DynLinkSet) {
815 self.inner.link_sets.write().await.register(scheme, lset);
816 }
817
818 /// Look up the lset for `scheme`, if any.
819 pub async fn link_set(&self, scheme: &str) -> Option<link_set::DynLinkSet> {
820 self.inner.link_sets.read().await.get(scheme)
821 }
822
823 /// Snapshot of every registered scheme name. Stable order for
824 /// `dbpvxr` dumps.
825 pub async fn registered_link_schemes(&self) -> Vec<String> {
826 let mut s = self.inner.link_sets.read().await.schemes();
827 s.sort();
828 s
829 }
830
831 /// Wait for the CA links to local records to report
832 /// `is_connected() == true`. Mirrors `dbCa: iocInit wait for local CA
833 /// links to connect` (epics-base PR #768/#856). The working set is
834 /// exactly [`Self::external_link_targets`]: only the CA facility's
835 /// local-target links — `pva://` links and non-local CA links connect
836 /// in the background and are never waited on (pvxs parity).
837 ///
838 /// Polls every 100 ms. Returns:
839 /// * `Ok(connected_count)` — the number of links that ended up
840 /// connected. May be smaller than the total when the timeout
841 /// expired before everyone was ready.
842 /// * The total link count — i.e. the size of the working set
843 /// that was checked. `(connected, total)` lets the caller log
844 /// "M/N CA links connected".
845 ///
846 /// Pure no-op when no CA link set is registered, or when its
847 /// `link_names()` has no local-target link yet (e.g. lazy-open lsets
848 /// that haven't observed any record link — record processing creates
849 /// the entries on first read, after iocInit returns).
850 pub async fn wait_for_external_links(&self, timeout: std::time::Duration) -> (usize, usize) {
851 // Collect (lset, name) pairs once. `link_names()` may grow
852 // as record processing opens new links, but iocInit's wait
853 // is bounded by the records loaded *before* Phase 3 — every
854 // such link is already opened by the time wire_device_support
855 // and setup_cp_links return.
856 let targets = self.external_link_targets().await;
857 let total = targets.len();
858 if total == 0 {
859 return (0, 0);
860 }
861 let deadline = tokio::time::Instant::now() + timeout;
862 loop {
863 let mut connected = 0usize;
864 for (lset, name) in &targets {
865 if lset.is_connected(name).await {
866 connected += 1;
867 }
868 }
869 if connected == total {
870 return (connected, total);
871 }
872 if tokio::time::Instant::now() >= deadline {
873 return (connected, total);
874 }
875 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
876 }
877 }
878
879 /// Snapshot the `(lset, link_name)` pairs the iocInit external-link
880 /// wait reasons over. Shared by [`Self::wait_for_external_links`] and
881 /// [`Self::unconnected_external_links`] so both see the identical
882 /// working set.
883 ///
884 /// C parity: the iocInit connection-wait is a property of the CA link
885 /// facility (dbCa) alone. `dbCaRun` (dbCa.c:370-380) blocks on
886 /// `initOutstanding`, the count of CA links flagged
887 /// `DBCA_CALLBACK_INIT_WAIT` — set only for a CA link whose target is
888 /// a LOCAL record (dbLink.c:128-130):
889 /// int isLocal = dbChannelTest(pvname) == 0;
890 /// dbCaAddLinkCallbackOpt(..., isLocal ? DBCA_CALLBACK_INIT_WAIT : 0)
891 /// No other external facility waits: pvxs pvalink's `linkGlobal_t::init`
892 /// (ioc/pvalink.cpp) only calls `chan->open()` per channel — it opens
893 /// in the background and never blocks iocInit. So the wait targets
894 /// exactly the CA link set's local-target links; a non-local CA link
895 /// (e.g. areaDetector's `ShutterStatusEPICS_RBV.INP = "test CP MS"`
896 /// placeholder) and every `pva://` link connect asynchronously and are
897 /// never held by iocInit, like C.
898 async fn external_link_targets(&self) -> Vec<(link_set::DynLinkSet, String)> {
899 // Only the CA facility participates — look it up directly rather
900 // than iterating every registered scheme. `has_name_no_resolve`
901 // is the `dbChannelTest` twin (target is a local record).
902 let Some(ca_lset) = ({
903 let registry = self.inner.link_sets.read().await;
904 registry.get("ca")
905 }) else {
906 return Vec::new();
907 };
908 let mut targets: Vec<(link_set::DynLinkSet, String)> = Vec::new();
909 for n in ca_lset.link_names().await {
910 if self.has_name_no_resolve(&n).await {
911 targets.push((ca_lset.clone(), n));
912 }
913 }
914 targets
915 }
916
917 /// Names of the waited-on CA links (local-target, per
918 /// [`Self::external_link_targets`]) that are opened but not yet
919 /// connected. iocInit calls this after
920 /// [`Self::wait_for_external_links`] times out so the
921 /// "M/N connected" diagnostic can name the `N-M` it proceeded
922 /// without, instead of leaving the operator to run `dbcar`.
923 /// `pva://` links are not in this set — they never block iocInit.
924 pub async fn unconnected_external_links(&self) -> Vec<String> {
925 let mut names = Vec::new();
926 for (lset, name) in self.external_link_targets().await {
927 if !lset.is_connected(&name).await {
928 names.push(name);
929 }
930 }
931 names
932 }
933
934 /// Enumerate every link-shaped field on `record_name`. Returns
935 /// `(field_name, link_string, parsed)` tuples for fields whose
936 /// raw value parses as a non-trivial link via
937 /// [`crate::server::record::parse_link_v2`]. Used by `dbpvxr` to
938 /// dump per-record link state without hardcoding the field-name
939 /// list — works across record types as long as they expose link
940 /// strings via [`Record::get_field`].
941 ///
942 /// Returns an empty Vec when the record doesn't exist.
943 pub async fn record_link_fields(
944 &self,
945 record_name: &str,
946 ) -> Vec<(String, String, crate::server::record::ParsedLink)> {
947 let rec = match self.get_record(record_name).await {
948 Some(r) => r,
949 None => return Vec::new(),
950 };
951 let inst = rec.read().await;
952 let mut out = Vec::new();
953 // Each field is parsed for ITS OWN link-field type: C `dbPutFieldLink`
954 // passes `pfldDes->field_type` to `dbParseLink` (`dbAccess.c:1094`),
955 // which then masks the modifiers by that type (`dbStaticLib.c:2380-2391`).
956 // `OUT` is `DBF_OUTLINK`, so its CP/CPP is discarded here rather than
957 // reaching `setup_cp_links` — an `OUT` link must never be registered as
958 // a CP holder.
959 let push = |field: &str,
960 raw: &str,
961 ftype: crate::server::record::LinkFieldType,
962 out: &mut Vec<_>| {
963 if raw.is_empty() {
964 return;
965 }
966 let parsed = crate::server::record::parse_link_field(raw, ftype);
967 if !matches!(parsed, crate::server::record::ParsedLink::None) {
968 out.push((field.to_string(), raw.to_string(), parsed));
969 }
970 };
971 use crate::server::record::LinkFieldType;
972 // Canonical link-bearing fields stored on `CommonFields` as raw
973 // String. These do NOT appear as `DbFieldType::String` entries in
974 // `field_list()`: an `ai`'s `INP` / an `ao`'s `OUT` carry
975 // `DBF_INLINK` / `DBF_OUTLINK` descriptors (and `INP`/`OUT` are
976 // not in the record's static field table at all), so the previous
977 // `field_list()` scan filtered by `String` silently dropped every
978 // device-support link — the holder's pvalink monitor was never
979 // opened. Enumerate the canonical storage directly so this method
980 // is the single owner of "which fields on a record are links",
981 // shared by `setup_cp_links` (CA CP/CPP) and the pvalink install
982 // scan (PVA CP/CPP).
983 push("INP", &inst.common.inp, LinkFieldType::In, &mut out);
984 push("OUT", &inst.common.out, LinkFieldType::Out, &mut out);
985 push("TSEL", &inst.common.tsel, LinkFieldType::In, &mut out);
986 push("SDIS", &inst.common.sdis, LinkFieldType::In, &mut out);
987 // Record-specific multi-input links (INPA..INPL for
988 // calc/calcout/sel/sub) and the CP-capable input link fields
989 // (DOL family, NVL, SELL, SGNL).
990 let mut field_names: Vec<&str> = inst
991 .record
992 .multi_input_links()
993 .iter()
994 .map(|(lf, _vf)| *lf)
995 .collect();
996 field_names.extend_from_slice(crate::server::database::links::CP_INPUT_LINK_FIELDS);
997 for field in field_names {
998 if let Some(EpicsValue::String(s)) = inst.record.get_field(field) {
999 push(field, &s.as_str_lossy(), LinkFieldType::In, &mut out);
1000 }
1001 }
1002 out
1003 }
1004
1005 /// Resolve an external PV name. Dispatches through the
1006 /// `(scheme, name)` lset if one is registered; otherwise falls
1007 /// back to the legacy [`ExternalPvResolver`] closure. `name`
1008 /// may be the bare PV name (in which case `pva://` is assumed
1009 /// when an lset is registered for that scheme) or a fully
1010 /// scheme-prefixed string.
1011 pub(crate) async fn resolve_external_pv(&self, name: &str) -> Option<EpicsValue> {
1012 // Try lsets first. We accept both "scheme://body" and the
1013 // bare body (stored in ParsedLink::Pva/Ca after the
1014 // dispatch in record/link.rs).
1015 let (scheme, body) = if let Some(rest) = name.strip_prefix("pva://") {
1016 ("pva", rest)
1017 } else if let Some(rest) = name.strip_prefix("ca://") {
1018 ("ca", rest)
1019 } else {
1020 // No prefix — try every registered lset in turn. The
1021 // first one with a value for `name` wins. Schemes are
1022 // single-digit so this is cheap.
1023 let registry = self.inner.link_sets.read().await;
1024 let lsets: Vec<_> = registry
1025 .schemes()
1026 .iter()
1027 .filter_map(|s| registry.get(s))
1028 .collect();
1029 drop(registry);
1030 for lset in lsets {
1031 if let Some(v) = lset.get_value(name).await {
1032 return Some(v);
1033 }
1034 }
1035 // Fall through to legacy resolver.
1036 let resolver = self.inner.external_resolver.read().await.clone();
1037 return match resolver {
1038 Some(r) => r(name).await,
1039 None => None,
1040 };
1041 };
1042 let lset = self.inner.link_sets.read().await.get(scheme);
1043 if let Some(lset) = lset {
1044 if let Some(v) = lset.get_value(body).await {
1045 return Some(v);
1046 }
1047 }
1048 let resolver = self.inner.external_resolver.read().await.clone();
1049 match resolver {
1050 Some(r) => r(name).await,
1051 None => None,
1052 }
1053 }
1054
1055 /// Add a simple PV with an initial value.
1056 ///
1057 /// Returns `Err` when `name` is already registered as a simple PV,
1058 /// a record, or an alias — mirroring epics-base C IOC which treats
1059 /// duplicate `dbLoadRecords` names as a fatal error. Callers that
1060 /// want replace-on-overwrite semantics must first call
1061 /// `remove_simple_pv` / `remove_record` / `remove_alias`.
1062 ///
1063 /// Serialized through `registration_mutex` so the
1064 /// cross-namespace check is atomic with the insert and the lock
1065 /// order across all add_*/remove_* methods is identical (no
1066 /// cross-namespace deadlock).
1067 pub async fn add_pv(&self, name: &str, initial: EpicsValue) -> CaResult<()> {
1068 let _gate = self.inner.registration_mutex.lock().await;
1069 self.check_name_free(name).await?;
1070 let pv = Arc::new(ProcessVariable::new(name.to_string(), initial));
1071 self.inner
1072 .simple_pvs
1073 .write()
1074 .await
1075 .insert(name.to_string(), pv);
1076 Ok(())
1077 }
1078
1079 /// Add a simple PV that already has a [`WriteHook`] installed.
1080 ///
1081 /// Equivalent to `add_pv` followed by `find_pv` + `set_write_hook`,
1082 /// but the PV is constructed with the hook in place so it is
1083 /// inserted into the `simple_pvs` map ATOMICALLY with the hook
1084 /// already attached. Closes a small race in proxy/gateway code
1085 /// where a downstream client could (in principle) `CREATE_CHAN` +
1086 /// `WRITE_NOTIFY` between the two awaits and hit the local
1087 /// `pv.set()` fallback path before the hook landed.
1088 ///
1089 /// Returns `Err` on duplicate name (see [`add_pv`]).
1090 pub async fn add_pv_with_hook(
1091 &self,
1092 name: &str,
1093 initial: EpicsValue,
1094 hook: crate::server::pv::WriteHook,
1095 ) -> CaResult<()> {
1096 self.add_pv_with_hooks(name, initial, hook, None).await
1097 }
1098
1099 /// like [`Self::add_pv_with_hook`] but also installs an
1100 /// optional [`AccessHook`](crate::server::pv::AccessHook) so the CA
1101 /// gateway can route this shadow PV's read/write access-rights
1102 /// decision through its own ACF. Both hooks are attached before the
1103 /// PV is inserted into `simple_pvs`, so a downstream `CREATE_CHAN`
1104 /// cannot observe the PV without its access hook bound.
1105 pub async fn add_pv_with_hooks(
1106 &self,
1107 name: &str,
1108 initial: EpicsValue,
1109 write_hook: crate::server::pv::WriteHook,
1110 access_hook: Option<crate::server::pv::AccessHook>,
1111 ) -> CaResult<()> {
1112 self.add_pv_with_hooks_full(name, initial, write_hook, access_hook, None)
1113 .await
1114 }
1115
1116 /// like [`Self::add_pv_with_hooks`] but also installs an optional
1117 /// [`ReadHook`](crate::server::pv::ReadHook) so a proxy (the CA
1118 /// gateway in no-cache mode) can serve each downstream GET from a
1119 /// fresh upstream fetch instead of the stored value. All three hooks
1120 /// are attached before the PV is inserted into `simple_pvs`, so a
1121 /// downstream `CREATE_CHAN` cannot observe the PV without its hooks
1122 /// bound — the read hook lands atomically with registration, closing
1123 /// the same race the write/access hooks already close. `read_hook:
1124 /// None` is identical to [`Self::add_pv_with_hooks`].
1125 pub async fn add_pv_with_hooks_full(
1126 &self,
1127 name: &str,
1128 initial: EpicsValue,
1129 write_hook: crate::server::pv::WriteHook,
1130 access_hook: Option<crate::server::pv::AccessHook>,
1131 read_hook: Option<crate::server::pv::ReadHook>,
1132 ) -> CaResult<()> {
1133 let _gate = self.inner.registration_mutex.lock().await;
1134 self.check_name_free(name).await?;
1135 let pv = Arc::new(ProcessVariable::new(name.to_string(), initial));
1136 pv.set_write_hook(write_hook);
1137 if let Some(access) = access_hook {
1138 pv.set_access_hook(access);
1139 }
1140 if let Some(read) = read_hook {
1141 pv.set_read_hook(read);
1142 }
1143 self.inner
1144 .simple_pvs
1145 .write()
1146 .await
1147 .insert(name.to_string(), pv);
1148 Ok(())
1149 }
1150
1151 /// Remove a simple PV by name. Returns `Some(pv)` if a PV was
1152 /// removed. Used by the gateway sweep so an evicted upstream
1153 /// subscription doesn't leave a stale shadow PV (with a now-dead
1154 /// `WriteHook` capturing an aborted upstream channel).
1155 ///
1156 /// Also purges any aliases that pointed AT this name
1157 /// (otherwise a re-add of the same alias name would fail with
1158 /// "already registered as an alias" even though its target is
1159 /// gone).
1160 pub async fn remove_simple_pv(&self, name: &str) -> Option<Arc<ProcessVariable>> {
1161 let _gate = self.inner.registration_mutex.lock().await;
1162 // Simple PVs cannot be alias targets (aliases point at
1163 // records), but a stale alias whose name MATCHES this PV
1164 // would have been rejected at add_alias time. No alias
1165 // cleanup needed for simple-PV removal.
1166 self.inner.simple_pvs.write().await.remove(name)
1167 }
1168
1169 /// Enter the LOAD phase: records are being created and the database is not
1170 /// yet the one C would classify links against. Called by every path that
1171 /// begins creating records for an IOC — an `IocBuilder` build, an iocsh
1172 /// `dbLoadRecords` / `dbCreateRecord`, `IocApp::run` — and idempotent within
1173 /// the phase, because an `st.cmd` issues several loads and they are all one
1174 /// `iocInit` (R18-92).
1175 ///
1176 /// # Refused once the IOC is running (R19-63)
1177 ///
1178 /// C admits no record creation after `iocInit`: `dbReadCOM`
1179 /// (`dbLexRoutines.c:236`) fails every `.db`/`.dbd` read with `-2` once
1180 /// `getIocState() != iocVoid`, and `dbCreateRecordCallFunc`
1181 /// (`dbStaticIocRegister.c:288`) fails with `S_dbLib_postInitRecRegister`.
1182 /// Asking to create records IS asking to enter the load phase, so the answer
1183 /// lives here and is a `Result` the caller cannot ignore — a creator that
1184 /// never asked cannot be written by accident, and one that asked cannot
1185 /// proceed on a refusal.
1186 ///
1187 /// The phase is left ONLY by [`Self::ioc_init`], and once left it is
1188 /// TERMINAL (R19-62): the queue is drained by exactly one `ioc_init`, so
1189 /// nothing can be pushed into it afterwards and stranded. A load that fails
1190 /// halfway leaves the phase open, which strands nothing: a queued
1191 /// classification blocks no caller, and it is dropped with the database.
1192 #[must_use = "C refuses a load after iocInit (dbReadCOM, dbLexRoutines.c:236); \
1193 the refusal must be reported and no record created"]
1194 pub fn begin_load(&self) -> Result<(), IocAlreadyInitialized> {
1195 let mut phase = self.inner.init_phase.lock().unwrap();
1196 match *phase {
1197 // The only producer of `Loading`.
1198 DbInitPhase::Unloaded => {
1199 *phase = DbInitPhase::Loading(Vec::new());
1200 Ok(())
1201 }
1202 // An `st.cmd` issues several loads; they are all one `iocInit`.
1203 DbInitPhase::Loading(_) => Ok(()),
1204 // Post-`iocInit`: terminal. Refused, as C refuses it.
1205 DbInitPhase::Running => Err(IocAlreadyInitialized),
1206 }
1207 }
1208
1209 /// Schedule a record's link-status classification — the port's
1210 /// `init_record` tail (C `checkLinks`).
1211 ///
1212 /// During the LOAD phase the future is QUEUED for [`Self::ioc_init`]; a
1213 /// half-built database cannot be classified against because the code that
1214 /// would do it has not been polled. Before any load, and once `iocInit` has
1215 /// run, it is spawned at once — which is what a runtime `special()` link
1216 /// re-point needs.
1217 pub(crate) fn schedule_record_init(
1218 &self,
1219 init: impl std::future::Future<Output = ()> + Send + 'static,
1220 ) {
1221 let mut phase = self.inner.init_phase.lock().unwrap();
1222 match &mut *phase {
1223 DbInitPhase::Loading(queued) => queued.push(Box::pin(init)),
1224 DbInitPhase::Unloaded | DbInitPhase::Running => {
1225 drop(phase);
1226 tokio::spawn(init);
1227 }
1228 }
1229 }
1230
1231 /// The `iocInit` barrier: end the LOAD phase and run every classification
1232 /// owed, to completion.
1233 ///
1234 /// After this returns the database is complete and every link status is
1235 /// FINAL — C's guarantee, where `init_record` runs inside `iocInit` and a
1236 /// `dbgf REC.INAV` right after it reads the classified value (before it, C
1237 /// refuses `dbgf` outright). Idempotent: an `st.cmd` that spells `iocInit`
1238 /// out and the `IocApp` that runs one anyway are the same single boundary.
1239 pub async fn ioc_init(&self) {
1240 let owed = {
1241 let mut phase = self.inner.init_phase.lock().unwrap();
1242 match std::mem::replace(&mut *phase, DbInitPhase::Running) {
1243 DbInitPhase::Loading(queued) => queued,
1244 // An IOC that loaded nothing (programmatic / unit-test database)
1245 // still crosses the barrier: the phase becomes terminal.
1246 DbInitPhase::Unloaded => return,
1247 DbInitPhase::Running => return,
1248 }
1249 };
1250 // Sequential, in issue order: each classification is a short read of a
1251 // now-immutable record set, and C's `init_record` pass is a loop too.
1252 for init in owed {
1253 init.await;
1254 }
1255 }
1256
1257 /// Add a record (accepts a boxed Record to avoid double-boxing).
1258 ///
1259 /// Returns `Err` when `name` collides with an existing record,
1260 /// simple PV, or alias. The C IOC's `dbLoadRecords` treats this as
1261 /// fatal; do not silently replace.
1262 ///
1263 /// The records-map insert AND scan-index insert run
1264 /// under the same `registration_mutex` hold, eliminating the
1265 /// TOCTOU window where `remove_record` could land between them
1266 /// and leave a phantom scan entry.
1267 pub async fn add_record(&self, name: &str, record: Box<dyn Record>) -> CaResult<()> {
1268 self.add_loaded_record(name, record, RecordLoad::default())
1269 .await
1270 }
1271
1272 /// Add a record together with the field set its `.db` definition loaded
1273 /// into it — the creation sink for every `dbLoadRecords` path.
1274 ///
1275 /// C's `dbLoadRecords` writes a record's ENTIRE field set through
1276 /// `dbStaticLib` (including the `UDF = 0` that `dbPutString`
1277 /// (`dbStaticLib.c:2653-2661`) implies for any put to a field named
1278 /// `VAL`), and only afterwards does `iocInit::doInitRecord0`
1279 /// (`iocInit.c:508-536`) evaluate `if (udf && stat == UDF_ALARM) sevr =
1280 /// udfs`. The port used to add the record first and apply its loaded common
1281 /// fields afterwards, so the init passes ran against a PRE-LOAD field set:
1282 /// every record with a `field(VAL,…)` latched `SEVR = INVALID` at creation
1283 /// and the `UDF = 0` that arrived a moment later could not lower it again.
1284 /// A whole `.db` of setpoint defaults and sim constants came up red.
1285 ///
1286 /// Taking the loaded fields here is what makes C's ordering hold by
1287 /// construction: there is no window in which the init passes can observe a
1288 /// record whose `.db` fields have not landed, because the record is not
1289 /// reachable until they have. [`RecordInstance::run_init_passes`] is
1290 /// crate-private for the same reason — the sink is the only caller.
1291 pub async fn add_loaded_record(
1292 &self,
1293 name: &str,
1294 record: Box<dyn Record>,
1295 load: RecordLoad,
1296 ) -> CaResult<()> {
1297 let _gate = self.inner.registration_mutex.lock().await;
1298 self.check_name_free(name).await?;
1299 let mut instance = RecordInstance::new_boxed(name.to_string(), record);
1300 // Hand the record a cycle-free handle to its own database so it can
1301 // post out-of-band field updates / wire completion-driven re-entry
1302 // (asyn TRACE callback, sseq WAITn) without owning the database.
1303 // C records reach `dbCommon::pdba`/the IOC the same way at
1304 // `dbDefineRecord` init; the framework supplies the back-reference,
1305 // the record never constructs it. Defaulted no-op for records that
1306 // do not need it.
1307 instance
1308 .record
1309 .set_async_context(name.to_string(), self.async_handle());
1310
1311 // Hand the record the current breakpoint-table registry snapshot so a
1312 // LINR>=3 ai/ao record can resolve its table lazily at convert time.
1313 // add_record is the single creation sink (IocBuilder, dbLoadRecords,
1314 // dbCreateRecord, inline records all funnel through here), so this one
1315 // install covers every creation path uniformly. The trait default is a
1316 // no-op for records that don't use it; skipped when no tables are
1317 // loaded so the common case pays no Arc clone. A record created before
1318 // its table is loaded is re-installed by `add_breaktables`.
1319 {
1320 let snapshot = self.inner.breaktable_registry.read().await.clone();
1321 if !snapshot.is_empty() {
1322 instance.record.install_breaktable_registry(snapshot);
1323 }
1324 }
1325
1326 // The `.db` load, applied to the instance BEFORE the init passes below
1327 // — C's `dbLoadRecords` → `iocInit` ordering. The `.db` value coercion
1328 // (`put_common_field_db_load`) differs from a runtime `dbPut`'s: C's
1329 // loader converter has a wider menu bound (`dbStaticRun.c`).
1330 //
1331 // The scan-index entry is built from `instance.common.scan` further
1332 // down, i.e. from the POST-load field set, so a `field(SCAN,…)` needs
1333 // no index fix-up here — the record has not been published yet.
1334 for (field, value) in load.common_fields {
1335 if let Err(e) = instance.put_common_field_db_load(&field, value) {
1336 eprintln!("put_common_field({field}) failed for {name}: {e}");
1337 }
1338 }
1339 // `info(...)` tags land before `init_record`, so device support that
1340 // reads them at init sees the values.
1341 for (key, value) in &load.info_tags {
1342 instance.set_info(key, value);
1343 }
1344
1345 // C's `iocInit` init passes, through their owner (the `doInitRecord0`
1346 // prologue — `pact = FALSE` plus the initial UDF severity — then
1347 // `init_record(0)`, `init_record(1)`, and the UDF tail). This sink is
1348 // the single site that runs them: a record built programmatically, by
1349 // iocsh `dbCreateRecord`, or from a `.db` is initialised the same way,
1350 // and — since the load above has already landed — always against its
1351 // FINAL field set.
1352 instance.run_init_passes(name);
1353
1354 // The init-seed owner: every CONSTANT link the record declares
1355 // (`Record::constant_init_links`) is loaded into its value field ONCE,
1356 // here — a constant delivers NOTHING at process time
1357 // (`dbConstLink.c:219-225`). `add_record` is the creation sink every
1358 // path funnels through, so this covers a record built programmatically
1359 // as well as one loaded from a .db; `IocBuilder`/`dbLoadRecords` call
1360 // the owner again after `init_record(1)`, once the record's final
1361 // NELM/FTVL buffer exists for an array constant to land in. Seeding
1362 // twice is a no-op — both run before any client put.
1363 super::database::processing::seed_constant_links(&mut instance);
1364
1365 let scan = instance.common.scan;
1366 let phas = instance.common.phas;
1367 self.inner
1368 .records
1369 .write()
1370 .await
1371 .insert(name.to_string(), Arc::new(RwLock::new(instance)));
1372
1373 // Assign a monotonic load-order sequence — the scan-index
1374 // secondary sort key, so same-PHAS records keep load order.
1375 let seq = self
1376 .inner
1377 .load_order_counter
1378 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1379 self.inner
1380 .load_order
1381 .write()
1382 .await
1383 .insert(name.to_string(), seq);
1384
1385 if let Some(list) = scan.scan_list() {
1386 self.inner
1387 .scan_index
1388 .write()
1389 .await
1390 .entry(list)
1391 .or_default()
1392 .insert((phas, seq, name.to_string()));
1393 }
1394 Ok(())
1395 }
1396
1397 /// Verify that `name` is not currently registered in any of the
1398 /// three namespaces. Caller MUST hold `registration_mutex` so the
1399 /// peek-then-insert sequence is atomic — without that, two tasks
1400 /// can both see the name as free and race the insert.
1401 async fn check_name_free(&self, name: &str) -> CaResult<()> {
1402 let kind = if self.inner.simple_pvs.read().await.contains_key(name) {
1403 Some("simple PV")
1404 } else if self.inner.records.read().await.contains_key(name) {
1405 Some("record")
1406 } else if self.inner.aliases.read().await.contains_key(name) {
1407 Some("alias")
1408 } else {
1409 None
1410 };
1411 if let Some(kind) = kind {
1412 return Err(CaError::DbParseError {
1413 line: 0,
1414 column: 0,
1415 message: format!("name '{name}' is already registered as a {kind}"),
1416 });
1417 }
1418 Ok(())
1419 }
1420
1421 /// Remove a record by name. Returns `true` if a record was removed,
1422 /// `false` if no such name was registered. Mirrors epics-base PR
1423 /// #505 — deletion at database creation, exposed here as a public
1424 /// API so iocsh `dbDeleteRecord` and tests can drive it.
1425 ///
1426 /// The cleanup covers the three indices that `add_record` populates:
1427 /// the records map, the scan index, and CP-link source/target lists.
1428 /// Live subscribers on the removed record drop their `Sender` clone
1429 /// when the `RecordInstance` is dropped — they observe `Closed` on
1430 /// next recv, matching the existing dbEvent cancel flow.
1431 pub async fn remove_record(&self, name: &str) -> bool {
1432 let _gate = self.inner.registration_mutex.lock().await;
1433 // 1) Remove from main map; keep scan + phas for scan-index cleanup.
1434 let removed = self.inner.records.write().await.remove(name);
1435 let Some(rec_arc) = removed else {
1436 return false;
1437 };
1438 let scan = {
1439 let inst = rec_arc.read().await;
1440 inst.common.scan
1441 };
1442
1443 // 2) Drop from scan index if it was scheduled. Match by record
1444 // name only — PHAS and load_order are not needed and may be
1445 // stale relative to the entry actually present.
1446 if let Some(list) = scan.scan_list() {
1447 let mut idx = self.inner.scan_index.write().await;
1448 if let Some(set) = idx.get_mut(&list) {
1449 set.retain(|(_, _, n)| n != name);
1450 if set.is_empty() {
1451 idx.remove(&list);
1452 }
1453 }
1454 }
1455
1456 // 2b) Drop the load-order entry.
1457 self.inner.load_order.write().await.remove(name);
1458
1459 // 3) Drop from CP-link tables. Removed both as source (channel
1460 // change → trigger targets) and as target (other channels'
1461 // CP lists may still reference this name).
1462 let mut cp = self.inner.cp_links.write().await;
1463 cp.remove(name);
1464 for targets in cp.values_mut() {
1465 targets.retain(|t| t.record != name);
1466 }
1467 drop(cp);
1468
1469 // 4) Purge aliases that pointed AT the
1470 // removed record. Otherwise `find_pv("ALT")` returns None
1471 // (target gone) but `add_pv("ALT", ...)` still fails with
1472 // "already registered as an alias" — orphan blocks reuse.
1473 let mut aliases = self.inner.aliases.write().await;
1474 aliases.retain(|_alias, target| target != name);
1475
1476 true
1477 }
1478
1479 /// Internal: synchronous lookup without invoking the search resolver.
1480 async fn find_entry_no_resolve(&self, name: &str) -> Option<PvEntry> {
1481 // a channel name may carry a `.{"arr":...}` filter
1482 // suffix. Strip it before lookup — the suffix is a per-channel
1483 // filter spec, not part of the PV identity. `split_channel_name`
1484 // is the single owner of "channel name → record_path" and is
1485 // idempotent on an already-stripped name. Without this a
1486 // filtered SimplePv (`SP.{"arr":...}`) never matches
1487 // `simple_pvs` (keyed by the bare PV name) and even a filtered
1488 // record fails when the JSON contains a `.` (e.g.
1489 // `{"dbnd":{"d":0.5}}`), because the bare `parse_pv_name` last-dot
1490 // split would tear the suffix apart instead of removing it.
1491 let record_path = filters::split_channel_name(name).record_path;
1492 let (base, _field) = parse_pv_name(&record_path);
1493
1494 if let Some(pv) = self.inner.simple_pvs.read().await.get(record_path.as_str()) {
1495 return Some(PvEntry::Simple(pv.clone()));
1496 }
1497 if let Some(rec) = self.inner.records.read().await.get(base) {
1498 return Some(PvEntry::Record(rec.clone()));
1499 }
1500 // Alias resolve (epics-base PR #336): the alternate name maps
1501 // to a canonical record name. Look up the real record after
1502 // translating the base.
1503 if let Some(target) = self.inner.aliases.read().await.get(base).cloned() {
1504 if let Some(rec) = self.inner.records.read().await.get(&target) {
1505 return Some(PvEntry::Record(rec.clone()));
1506 }
1507 }
1508 None
1509 }
1510
1511 /// Register an alias `alias` for an existing record `target`.
1512 /// Mirrors epics-base PR #336. Returns `Err(...)` when the target
1513 /// does not exist or the alias name is already in use anywhere
1514 /// in the database (records, simple PVs, or other aliases).
1515 ///
1516 /// Pre-fix the alias path checked only
1517 /// `records` and `aliases` — a simple-PV with the same name as
1518 /// the proposed alias was missed, leaving the database in a
1519 /// state where `find_pv(alias)` could resolve to either the
1520 /// simple PV or the alias-mapped record depending on lookup
1521 /// order. Now we run the same cross-namespace `check_name_free`
1522 /// guard the other add_* paths use.
1523 pub async fn add_alias(&self, alias: &str, target: &str) -> CaResult<()> {
1524 let _gate = self.inner.registration_mutex.lock().await;
1525 if !self.inner.records.read().await.contains_key(target) {
1526 return Err(CaError::ChannelNotFound(format!(
1527 "alias target '{target}' is not a registered record"
1528 )));
1529 }
1530 self.check_name_free(alias).await?;
1531 self.inner
1532 .aliases
1533 .write()
1534 .await
1535 .insert(alias.to_string(), target.to_string());
1536 Ok(())
1537 }
1538
1539 /// Resolve an alias to its target record name, or `None` when the
1540 /// name is not an alias.
1541 pub async fn resolve_alias(&self, name: &str) -> Option<String> {
1542 self.inner.aliases.read().await.get(name).cloned()
1543 }
1544
1545 /// Queue an iocsh command line for post-PINI execution.
1546 /// Mirrors epics-base PR #558 — `afterIocRunning <command>` lets
1547 /// the startup script schedule actions that run after iocInit
1548 /// completes (when the record set is fully wired up).
1549 pub fn queue_after_ioc_running(&self, line: impl Into<String>) {
1550 self.inner
1551 .after_ioc_running
1552 .lock()
1553 .unwrap()
1554 .push(line.into());
1555 }
1556
1557 /// Drain the post-PINI iocsh command queue. Called by
1558 /// `IocApplication::run` after PINI processing.
1559 pub fn take_after_ioc_running(&self) -> Vec<String> {
1560 std::mem::take(&mut *self.inner.after_ioc_running.lock().unwrap())
1561 }
1562
1563 /// Internal: synchronous existence check without resolver.
1564 async fn has_name_no_resolve(&self, name: &str) -> bool {
1565 // strip the channel-filter suffix before lookup so a
1566 // filtered channel (`SP.{"arr":...}` / `REC.{"dbnd":{"d":0.5}}`)
1567 // resolves to its underlying PV at UDP-search time. This is the
1568 // search-side twin of `find_entry_no_resolve`; without it a
1569 // filtered SimplePv never answers a SEARCH and the client never
1570 // reaches CREATE_CHAN. See that function for the full rationale.
1571 let record_path = filters::split_channel_name(name).record_path;
1572 let (base, _) = parse_pv_name(&record_path);
1573 if self
1574 .inner
1575 .simple_pvs
1576 .read()
1577 .await
1578 .contains_key(record_path.as_str())
1579 {
1580 return true;
1581 }
1582 if self.inner.records.read().await.contains_key(base) {
1583 return true;
1584 }
1585 // Alias entry exists and points to a live record
1586 // (epics-base PR #336).
1587 if let Some(target) = self.inner.aliases.read().await.get(base) {
1588 return self.inner.records.read().await.contains_key(target);
1589 }
1590 false
1591 }
1592
1593 /// Look up an entry by name. Supports "record.FIELD" syntax.
1594 ///
1595 /// If the name is not found and a search resolver is installed,
1596 /// the resolver is invoked once. If the resolver returns true, the
1597 /// database is re-checked.
1598 pub async fn find_entry(&self, name: &str) -> Option<PvEntry> {
1599 self.find_entry_from(name, None).await
1600 }
1601
1602 /// Like [`Self::find_entry`], but threads the downstream client's
1603 /// socket address into the search resolver. The CA TCP CREATE_CHANNEL
1604 /// handler passes the connection peer so the gateway can apply
1605 /// host-scoped `.pvlist` admission.
1606 pub async fn find_entry_from(
1607 &self,
1608 name: &str,
1609 peer: Option<std::net::SocketAddr>,
1610 ) -> Option<PvEntry> {
1611 if let Some(entry) = self.find_entry_no_resolve(name).await {
1612 // A cached simple PV must still pass the per-request
1613 // existence gate (CA gateway host/state admission). When the
1614 // gate denies it, answer does-not-exist for this requester
1615 // instead of returning the stale shadow entry — C ca-gateway
1616 // re-runs `gateAs::findEntry`/cache-state on every
1617 // `pvExistTest` (gateServer.cc:1516-1637). Records/aliases
1618 // bypass the gate (see `simple_pv_gate_denies`).
1619 if matches!(entry, PvEntry::Simple(_)) && self.simple_pv_gate_denies(name, peer).await {
1620 return None;
1621 }
1622 return Some(entry);
1623 }
1624 // Try the search resolver
1625 let resolver = self.inner.search_resolver.read().await.clone();
1626 if let Some(r) = resolver {
1627 if r(name.to_string(), peer).await {
1628 return self.find_entry_no_resolve(name).await;
1629 }
1630 }
1631 None
1632 }
1633
1634 /// Check if a base name exists (for UDP search).
1635 ///
1636 /// If the name is not in the database and a search resolver is installed,
1637 /// the resolver is invoked. The resolver may populate the database
1638 /// (e.g., subscribe to an upstream IOC and add a placeholder PV) and
1639 /// return true; this method then re-checks.
1640 pub async fn has_name(&self, name: &str) -> bool {
1641 self.has_name_from(name, None).await
1642 }
1643
1644 /// Like [`Self::has_name`], but threads the downstream client's
1645 /// socket address into the search resolver. The CA UDP search
1646 /// responder passes the datagram source address so the gateway can
1647 /// apply host-scoped `.pvlist` admission.
1648 pub async fn has_name_from(&self, name: &str, peer: Option<std::net::SocketAddr>) -> bool {
1649 if self.has_name_no_resolve(name).await {
1650 // Same per-request gate as `find_entry_from`: a cached simple
1651 // PV the gateway's host/state admission denies must answer
1652 // does-not-exist at search time. Records/aliases bypass.
1653 if self.simple_pv_gate_denies(name, peer).await {
1654 return false;
1655 }
1656 return true;
1657 }
1658 let resolver = self.inner.search_resolver.read().await.clone();
1659 if let Some(r) = resolver {
1660 if r(name.to_string(), peer).await {
1661 return self.has_name_no_resolve(name).await;
1662 }
1663 }
1664 false
1665 }
1666
1667 /// Look up a simple PV by name (backward-compatible).
1668 pub async fn find_pv(&self, name: &str) -> Option<Arc<ProcessVariable>> {
1669 if let Some(pv) = self.inner.simple_pvs.read().await.get(name) {
1670 return Some(pv.clone());
1671 }
1672 None
1673 }
1674
1675 /// Get a record Arc by name. Alias-aware (epics-base PR #336):
1676 /// when `name` is not a canonical record but matches a registered
1677 /// alias, the alias' target record is returned. Mirrors base
1678 /// `dbNameToAddr` behaviour, so dbpf/dbpr/dbgf, CA channel lookup,
1679 /// and DB-link target resolution all work transparently for
1680 /// aliases.
1681 ///
1682 /// Use [`Self::get_record_no_resolve`] when the caller already
1683 /// holds a canonical name and wants to suppress the alias path
1684 /// (e.g. to detect alias collisions during builder wiring).
1685 pub async fn get_record(&self, name: &str) -> Option<Arc<RwLock<RecordInstance>>> {
1686 if let Some(rec) = self.inner.records.read().await.get(name).cloned() {
1687 return Some(rec);
1688 }
1689 let target = self.inner.aliases.read().await.get(name).cloned()?;
1690 self.inner.records.read().await.get(&target).cloned()
1691 }
1692
1693 /// Strict variant of [`Self::get_record`] — does NOT consult the
1694 /// alias table. Returns `Some` only when a canonical record with
1695 /// that exact name exists.
1696 pub async fn get_record_no_resolve(&self, name: &str) -> Option<Arc<RwLock<RecordInstance>>> {
1697 self.inner.records.read().await.get(name).cloned()
1698 }
1699
1700 /// Every record name, in **database load order** — the single owner of
1701 /// whole-database iteration order.
1702 ///
1703 /// C parity: `dbFirstRecord`/`dbNextRecord` walk the record list of each
1704 /// record type in the order `dbReadDatabase` appended them, so every
1705 /// whole-database pass a C IOC makes — `initDevSup`, `initDatabase`,
1706 /// `initialProcess` (PINI), `dbl`/`dbgrep` dumps — visits records in load
1707 /// order. Device support is written against that contract: a dynamic device
1708 /// support whose record references another record (epics-modules/opcua's
1709 /// element records require their `opcuaItem` record to have bound first,
1710 /// linkParser.cpp:226-234) only boots if the referenced record was wired
1711 /// first, which the `.db` guarantees by declaring it first.
1712 ///
1713 /// The names live in a `HashMap`, so returning `keys()` made that order the
1714 /// hash order: neither load order nor even stable across runs of the same
1715 /// binary (`RandomState` reseeds per process). Booting the same database
1716 /// twice could wire records in two different orders — one boot succeeding
1717 /// and the next failing. Ordering here, at the one accessor every
1718 /// whole-database walk already goes through, makes every such pass
1719 /// deterministic and load-ordered at once.
1720 ///
1721 /// The order key is the existing per-record `load_order` sequence (the
1722 /// scan-index's secondary sort key), so this ordering and the scan lists'
1723 /// ordering are the same fact, not two. A record with no sequence — none
1724 /// exists; `add_record` is the only insertion path — would sort last by
1725 /// name rather than nondeterministically.
1726 pub async fn all_record_names(&self) -> Vec<String> {
1727 // Lock order records → load_order, matching `add_record`/`remove_record`.
1728 let records = self.inner.records.read().await;
1729 let load_order = self.inner.load_order.read().await;
1730 let mut names: Vec<String> = records.keys().cloned().collect();
1731 names.sort_by(|a, b| {
1732 let seq = |n: &String| load_order.get(n).copied().unwrap_or(u64::MAX);
1733 seq(a).cmp(&seq(b)).then_with(|| a.cmp(b))
1734 });
1735 names
1736 }
1737
1738 /// Get all alias names registered against existing records.
1739 /// Mirrors the alias-half of base's `dbFirstRecord` iteration —
1740 /// `dbgrep` / `dbglob` / `dbsr` walk both record names and
1741 /// aliases when matching a glob.
1742 pub async fn all_alias_names(&self) -> Vec<String> {
1743 self.inner.aliases.read().await.keys().cloned().collect()
1744 }
1745
1746 /// Return every alias that points at `canonical`. Sorted for
1747 /// stable output; empty when the record has no aliases. Used by
1748 /// `dbpr` to surface alias-form names so admins can see how
1749 /// clients may reach the record.
1750 pub async fn aliases_for_record(&self, canonical: &str) -> Vec<String> {
1751 let aliases = self.inner.aliases.read().await;
1752 let mut hits: Vec<String> = aliases
1753 .iter()
1754 .filter_map(|(alias, target)| {
1755 if target == canonical {
1756 Some(alias.clone())
1757 } else {
1758 None
1759 }
1760 })
1761 .collect();
1762 hits.sort();
1763 hits
1764 }
1765
1766 /// Get all simple PV names.
1767 pub async fn all_simple_pv_names(&self) -> Vec<String> {
1768 self.inner.simple_pvs.read().await.keys().cloned().collect()
1769 }
1770}
1771
1772#[cfg(test)]
1773mod tests {
1774 use super::*;
1775
1776 /// C `recGblGetTimeStampSimm` (recGbl.c:310-343) maps TSE values
1777 /// to epicsTime sources via the constants in `epicsTime.h:102-104`.
1778 /// The Rust port previously misread TSE=-1 as "device-provided
1779 /// with BestTime fallback" and gated the BestTime call on a
1780 /// UNIX_EPOCH check. C calls `epicsTimeGetEvent(-1)`
1781 /// unconditionally; only TSE=-2 (epicsTimeEventDeviceTime) leaves
1782 /// `precord->time` untouched.
1783 ///
1784 /// Regression: a stale device write (any non-epoch SystemTime)
1785 /// suppressed every BestTime refresh thereafter.
1786 #[test]
1787 fn apply_timestamp_tse_minus_one_always_overwrites_with_best_time() {
1788 use crate::server::record::CommonFields;
1789 use std::time::{Duration, SystemTime};
1790
1791 // Pre-populate `time` with a stale but non-epoch sentinel.
1792 let stale = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
1793 let mut common = CommonFields::default();
1794 common.tse = -1;
1795 common.time = stale;
1796
1797 apply_timestamp(&mut common, false);
1798
1799 // BestTime must have run unconditionally — `common.time` is
1800 // no longer the stale sentinel.
1801 assert_ne!(
1802 common.time, stale,
1803 "TSE=-1 must always overwrite via generalTime BestTime, \
1804 matching C epicsTimeGetEvent(-1) called unconditionally"
1805 );
1806 }
1807
1808 /// C `epicsTimeEventDeviceTime = -2` (epicsTime.h:104). The C
1809 /// path does NOT call `epicsTimeGetEvent` for this TSE value;
1810 /// device support has already set `precord->time` before the
1811 /// recGbl call. The Rust port must leave `common.time` untouched.
1812 #[test]
1813 fn apply_timestamp_tse_minus_two_preserves_device_provided_time() {
1814 use crate::server::record::CommonFields;
1815 use std::time::{Duration, SystemTime};
1816
1817 let device_time = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000_000);
1818 let mut common = CommonFields::default();
1819 common.tse = -2;
1820 common.time = device_time;
1821
1822 apply_timestamp(&mut common, false);
1823
1824 assert_eq!(
1825 common.time, device_time,
1826 "TSE=-2 (epicsTimeEventDeviceTime) must preserve device-provided time"
1827 );
1828 }
1829
1830 #[test]
1831 fn select_link_indices_fanout_all_specified_mask() {
1832 use crate::server::record::AlarmSeverity;
1833 // All — every slot.
1834 let r = select_link_indices_ex(SelmKind::FanoutSeq, 0, 0, 0, 0, 16);
1835 assert_eq!(r.indices, (0..16).collect::<Vec<_>>());
1836 assert!(r.alarm.is_none());
1837
1838 // Specified, 0-based: SELN=0 selects LNK0 (C parity, fanout).
1839 let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 0, 0, 0, 16);
1840 assert_eq!(r.indices, vec![0]);
1841 // Specified with OFFS bias: SELN=2 + OFFS=3 → index 5.
1842 let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 2, 3, 0, 16);
1843 assert_eq!(r.indices, vec![5]);
1844 // Out-of-range Specified → INVALID alarm, no links.
1845 let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 20, 0, 0, 16);
1846 assert!(r.indices.is_empty());
1847 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1848 // Negative resolved index (SELN + negative OFFS) → INVALID.
1849 let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, 0, -1, 0, 16);
1850 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1851
1852 // Mask: SELN=0b101 → bits 0 and 2.
1853 let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 0, 16);
1854 assert_eq!(r.indices, vec![0, 2]);
1855 // Mask with SHFT: SELN=0b101 >> 1 = 0b10 → bit 1.
1856 let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 1, 16);
1857 assert_eq!(r.indices, vec![1]);
1858 // Mask with negative SHFT: SELN=0b101 << 1 = 0b1010 → bits 1,3.
1859 let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, -1, 16);
1860 assert_eq!(r.indices, vec![1, 3]);
1861 // SHFT out of [-15,15] → INVALID.
1862 let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, 5, 0, 16, 16);
1863 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1864
1865 // Unknown SELM → INVALID.
1866 let r = select_link_indices_ex(SelmKind::FanoutSeq, 9, 0, 0, 0, 16);
1867 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1868 }
1869
1870 #[test]
1871 fn select_link_indices_dfanout_specified_is_one_based() {
1872 use crate::server::record::AlarmSeverity;
1873 // dfanout Specified is 1-based: SELN=1 → OUTA (index 0).
1874 let r = select_link_indices_ex(SelmKind::Dfanout, 1, 1, 0, 0, 16);
1875 assert_eq!(r.indices, vec![0]);
1876 // SELN=2 → OUTB (index 1).
1877 let r = select_link_indices_ex(SelmKind::Dfanout, 1, 2, 0, 0, 16);
1878 assert_eq!(r.indices, vec![1]);
1879 // SELN=0 → drive nothing, NO alarm.
1880 let r = select_link_indices_ex(SelmKind::Dfanout, 1, 0, 0, 0, 16);
1881 assert!(r.indices.is_empty());
1882 assert!(r.alarm.is_none());
1883 // SELN > 16 → INVALID.
1884 let r = select_link_indices_ex(SelmKind::Dfanout, 1, 17, 0, 0, 16);
1885 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1886 // dfanout Mask has no SHFT — SHFT arg ignored.
1887 let r = select_link_indices_ex(SelmKind::Dfanout, 2, 5, 0, 7, 16);
1888 assert_eq!(r.indices, vec![0, 2]);
1889 }
1890
1891 /// `SELN` is unsigned, and the rule that makes it unsigned depends on the
1892 /// SOURCE type — because in C the source type picks the conversion routine.
1893 #[test]
1894 fn seln_cast_follows_the_source_type() {
1895 // Integer source -> `getLongUshort`, `(epicsUInt16)(epicsInt32)v`.
1896 // C DEFINES this (C17 6.3.1.3p2, modulo 2^16), so we reproduce it.
1897 assert_eq!(dbr_ushort_cast(&EpicsValue::Long(-1)), 65535);
1898 assert_eq!(dbr_ushort_cast(&EpicsValue::Long(65536)), 0);
1899 assert_eq!(dbr_ushort_cast(&EpicsValue::Short(-1)), 65535);
1900 assert_eq!(dbr_ushort_cast(&EpicsValue::Int64(-1)), 65535);
1901 assert_eq!(dbr_ushort_cast(&EpicsValue::Long(3)), 3);
1902
1903 // Float source -> `getDoubleUshort`, `(epicsUInt16)d`. C leaves this
1904 // UNDEFINED (C17 6.3.1.4p1), and whatever `types::c_cast` decides to do
1905 // about that is a SEPARATE question from this one — the point here is
1906 // only that the float source takes the float rule and the integer
1907 // source does not.
1908 assert_eq!(
1909 dbr_ushort_cast(&EpicsValue::Double(-1.0)),
1910 crate::types::c_cast::f64_to_u16(-1.0)
1911 );
1912 assert_eq!(
1913 dbr_ushort_cast(&EpicsValue::Double(65536.0)),
1914 crate::types::c_cast::f64_to_u16(65536.0)
1915 );
1916 // In range: no policy in play, both rules truncate toward zero.
1917 assert_eq!(dbr_ushort_cast(&EpicsValue::Double(3.7)), 3);
1918 }
1919
1920 /// Whatever produced it, a `SELN` of 65535 selects nothing under Specified
1921 /// (out of range -> INVALID) and everything under Mask.
1922 #[test]
1923 fn seln_at_the_unsigned_maximum_selects_by_selm() {
1924 use crate::server::record::AlarmSeverity;
1925 let seln_max = 65535u16;
1926 // fanout/seq Specified: C `i = (epicsUInt16)seln + offs` = 65535 →
1927 // out of range → INVALID. A signed read would clamp to 0 and wrongly
1928 // drive link 0.
1929 let r = select_link_indices_ex(SelmKind::FanoutSeq, 1, seln_max, 0, 0, 16);
1930 assert!(r.indices.is_empty());
1931 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1932 // fanout/seq Mask: 65535 → all 16 low bits set → every link. A signed
1933 // read would produce an empty mask.
1934 let r = select_link_indices_ex(SelmKind::FanoutSeq, 2, seln_max, 0, 0, 16);
1935 assert_eq!(r.indices, (0..16).collect::<Vec<_>>());
1936 // dfanout Specified: 65535 > count → INVALID. A signed read would see
1937 // -1 ≤ 0 → drive nothing, with no alarm.
1938 let r = select_link_indices_ex(SelmKind::Dfanout, 1, seln_max, 0, 0, 16);
1939 assert!(r.indices.is_empty());
1940 assert_eq!(r.alarm, Some((15, AlarmSeverity::Invalid)));
1941 }
1942
1943 /// Lset that flips to "connected" after a configurable delay.
1944 /// Drives the wait_for_external_links time-budget tests below.
1945 struct DelayedConnectLset {
1946 names: Vec<String>,
1947 connect_at: tokio::time::Instant,
1948 }
1949
1950 #[async_trait::async_trait]
1951 impl link_set::LinkSet for DelayedConnectLset {
1952 async fn is_connected(&self, _: &str) -> bool {
1953 tokio::time::Instant::now() >= self.connect_at
1954 }
1955 async fn get_value(&self, _: &str) -> Option<EpicsValue> {
1956 None
1957 }
1958 async fn link_names(&self) -> Vec<String> {
1959 self.names.clone()
1960 }
1961 }
1962
1963 #[tokio::test]
1964 async fn wait_for_external_links_returns_zero_zero_when_no_lsets() {
1965 let db = PvDatabase::new();
1966 let (c, t) = db
1967 .wait_for_external_links(std::time::Duration::from_millis(50))
1968 .await;
1969 assert_eq!((c, t), (0, 0));
1970 }
1971
1972 #[tokio::test]
1973 async fn wait_for_external_links_connected_quickly() {
1974 let db = PvDatabase::new();
1975 // Local-target forced-CA links (dbChannelTest==0 → isLocal): these
1976 // get DBCA_CALLBACK_INIT_WAIT, so iocInit waits for them.
1977 db.add_pv("pv:A", EpicsValue::Long(0)).await.unwrap();
1978 db.add_pv("pv:B", EpicsValue::Long(0)).await.unwrap();
1979 let lset = Arc::new(DelayedConnectLset {
1980 names: vec!["pv:A".to_string(), "pv:B".to_string()],
1981 connect_at: tokio::time::Instant::now(),
1982 });
1983 // Registered under "ca": the iocInit wait is CA-facility only, so
1984 // the working set comes from the "ca" link set (these forced-CA
1985 // local-target links), never from a "pva" set.
1986 db.register_link_set("ca", lset).await;
1987 let (c, t) = db
1988 .wait_for_external_links(std::time::Duration::from_secs(1))
1989 .await;
1990 assert_eq!((c, t), (2, 2));
1991 }
1992
1993 #[tokio::test]
1994 async fn wait_for_external_links_returns_partial_on_timeout() {
1995 let db = PvDatabase::new();
1996 // Local target so the link is in the init-wait set (dbLink.c:130);
1997 // connect-time well past the budget below, so the wait must return
1998 // (0, 1) instead of blocking.
1999 db.add_pv("slow:pv", EpicsValue::Long(0)).await.unwrap();
2000 let lset = Arc::new(DelayedConnectLset {
2001 names: vec!["slow:pv".to_string()],
2002 connect_at: tokio::time::Instant::now() + std::time::Duration::from_secs(60),
2003 });
2004 db.register_link_set("ca", lset).await;
2005 let started = tokio::time::Instant::now();
2006 let (c, t) = db
2007 .wait_for_external_links(std::time::Duration::from_millis(250))
2008 .await;
2009 let elapsed = started.elapsed();
2010 assert_eq!((c, t), (0, 1));
2011 assert!(
2012 elapsed >= std::time::Duration::from_millis(200),
2013 "wait must consume at least the configured budget, got {:?}",
2014 elapsed
2015 );
2016 assert!(
2017 elapsed < std::time::Duration::from_secs(2),
2018 "wait must not exceed the budget by much, got {:?}",
2019 elapsed
2020 );
2021 }
2022
2023 /// C parity (dbLink.c:130): a link whose target is NOT a local record
2024 /// (`dbChannelTest != 0`) gets no DBCA_CALLBACK_INIT_WAIT, so iocInit
2025 /// must not block on it. An areaDetector `test CP MS` placeholder — a CP
2026 /// link to a PV that exists nowhere — must drop straight through, leaving
2027 /// the link to connect (or dangle) asynchronously and silently, like C.
2028 #[tokio::test]
2029 async fn wait_for_external_links_skips_nonlocal_targets() {
2030 let db = PvDatabase::new();
2031 // "test" has no local record and would never connect.
2032 let lset = Arc::new(DelayedConnectLset {
2033 names: vec!["test".to_string()],
2034 connect_at: tokio::time::Instant::now() + std::time::Duration::from_secs(60),
2035 });
2036 db.register_link_set("ca", lset).await;
2037 let started = tokio::time::Instant::now();
2038 let (c, t) = db
2039 .wait_for_external_links(std::time::Duration::from_secs(10))
2040 .await;
2041 // Non-local target is excluded from the wait set entirely, so the
2042 // call returns (0, 0) immediately rather than blocking the budget.
2043 assert_eq!((c, t), (0, 0));
2044 assert!(
2045 started.elapsed() < std::time::Duration::from_secs(1),
2046 "non-local link must not be waited on, got {:?}",
2047 started.elapsed()
2048 );
2049 // And it is reported as unconnected by neither path (silent, like C).
2050 assert!(db.unconnected_external_links().await.is_empty());
2051 }
2052
2053 // epics-base PR #336 — alias parsing + lookup integration tests.
2054
2055 #[tokio::test]
2056 async fn alias_resolves_through_find_entry() {
2057 let db = PvDatabase::new();
2058 db.add_record(
2059 "TARGET",
2060 Box::new(crate::server::records::ai::AiRecord::new(42.0)),
2061 )
2062 .await
2063 .unwrap();
2064 db.add_alias("ALIAS_NAME", "TARGET").await.unwrap();
2065
2066 // find_entry on the alias must return the same record as
2067 // find_entry on the target.
2068 let via_alias = db.find_entry("ALIAS_NAME").await;
2069 let via_target = db.find_entry("TARGET").await;
2070 assert!(via_alias.is_some());
2071 assert!(via_target.is_some());
2072 // has_name flips true for the alias too.
2073 assert!(db.has_name("ALIAS_NAME").await);
2074 assert!(db.has_name("TARGET").await);
2075 assert!(!db.has_name("NOT:THERE").await);
2076 }
2077
2078 #[tokio::test]
2079 async fn alias_target_must_exist() {
2080 let db = PvDatabase::new();
2081 let err = db.add_alias("DANGLING", "MISSING_TARGET").await;
2082 assert!(err.is_err(), "alias to missing target must be rejected");
2083 }
2084
2085 #[tokio::test]
2086 async fn alias_collision_with_existing_record_rejected() {
2087 let db = PvDatabase::new();
2088 db.add_record(
2089 "EXISTING",
2090 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2091 )
2092 .await
2093 .unwrap();
2094 db.add_record(
2095 "OTHER",
2096 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2097 )
2098 .await
2099 .unwrap();
2100 let err = db.add_alias("EXISTING", "OTHER").await;
2101 assert!(
2102 err.is_err(),
2103 "alias name colliding with record must be rejected"
2104 );
2105 }
2106
2107 #[tokio::test]
2108 async fn get_record_resolves_alias() {
2109 // Regression: get_record must transparently resolve
2110 // aliases so dbpf / dbgf / dbpr / CA put paths see the same
2111 // record whether the caller uses the canonical name or the
2112 // alias.
2113 let db = PvDatabase::new();
2114 db.add_record(
2115 "TARGET",
2116 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2117 )
2118 .await
2119 .unwrap();
2120 db.add_alias("ALIAS", "TARGET").await.unwrap();
2121
2122 let via_canonical = db.get_record("TARGET").await;
2123 let via_alias = db.get_record("ALIAS").await;
2124 assert!(via_canonical.is_some());
2125 assert!(via_alias.is_some(), "get_record must resolve alias");
2126 // Both calls return the same Arc (pointer equality).
2127 assert!(Arc::ptr_eq(&via_canonical.unwrap(), &via_alias.unwrap()));
2128 }
2129
2130 /// `add_record` is the single creation sink: a record added AFTER its
2131 /// breakpoint table is loaded must receive the registry snapshot so a
2132 /// `LINR >= 3` conversion resolves — without any explicit per-call-site
2133 /// `install_breaktable_registry`. This covers the dbCreateRecord and
2134 /// inline-record creation paths that previously skipped the install.
2135 #[tokio::test]
2136 async fn add_record_installs_breaktable_registry_from_snapshot() {
2137 let db = PvDatabase::new();
2138 let ramp = crate::server::cvt_bpt::BrkTable::build(
2139 "ramp",
2140 &[(0.0, 0.0), (100.0, 10.0), (300.0, 30.0)],
2141 )
2142 .unwrap();
2143 db.add_breaktables(vec![ramp]).await;
2144
2145 let mut rec = crate::server::records::ai::AiRecord::new(0.0);
2146 rec.put_field("LINR", EpicsValue::Short(15)).unwrap(); // ramp = first user-table index
2147 db.add_record("AI:BPT", Box::new(rec)).await.unwrap();
2148
2149 let arc = db.get_record("AI:BPT").await.unwrap();
2150 let mut inst = arc.write().await;
2151 inst.record.put_field("RVAL", EpicsValue::Long(50)).unwrap();
2152 inst.record.process().unwrap();
2153 // raw 50 in [0,100] -> eng 5.0, proving the registry was installed by
2154 // add_record alone.
2155 assert_eq!(inst.record.get_field("VAL"), Some(EpicsValue::Double(5.0)));
2156 }
2157
2158 /// `add_breaktables` re-installs the new snapshot into records that
2159 /// already exist, so a record created BEFORE its table was loaded (inline
2160 /// records added before dbLoadRecords; merge-reloads repointing LINR) can
2161 /// still resolve `LINR >= 3`. Without the re-install the record keeps an
2162 /// empty registry and never linearises.
2163 #[tokio::test]
2164 async fn add_breaktables_reinstalls_registry_into_existing_records() {
2165 let db = PvDatabase::new();
2166 // Record added while the registry is still empty: add_record installs
2167 // nothing (the inline-record / pre-load ordering case).
2168 let mut rec = crate::server::records::ai::AiRecord::new(0.0);
2169 rec.put_field("LINR", EpicsValue::Short(15)).unwrap(); // ramp = first user-table index
2170 db.add_record("AI:BPT", Box::new(rec)).await.unwrap();
2171
2172 // Load the table afterwards — re-install must reach the existing record.
2173 let ramp = crate::server::cvt_bpt::BrkTable::build(
2174 "ramp",
2175 &[(0.0, 0.0), (100.0, 10.0), (300.0, 30.0)],
2176 )
2177 .unwrap();
2178 db.add_breaktables(vec![ramp]).await;
2179
2180 let arc = db.get_record("AI:BPT").await.unwrap();
2181 let mut inst = arc.write().await;
2182 inst.record.put_field("RVAL", EpicsValue::Long(50)).unwrap();
2183 inst.record.process().unwrap();
2184 assert_eq!(inst.record.get_field("VAL"), Some(EpicsValue::Double(5.0)));
2185 }
2186
2187 #[tokio::test]
2188 async fn get_record_no_resolve_skips_alias_table() {
2189 // Strict variant must NOT see aliases — keeps the canonical
2190 // distinction available for builder code paths.
2191 let db = PvDatabase::new();
2192 db.add_record(
2193 "TARGET",
2194 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2195 )
2196 .await
2197 .unwrap();
2198 db.add_alias("ALIAS", "TARGET").await.unwrap();
2199
2200 assert!(db.get_record_no_resolve("TARGET").await.is_some());
2201 assert!(
2202 db.get_record_no_resolve("ALIAS").await.is_none(),
2203 "get_record_no_resolve must not follow alias table"
2204 );
2205 }
2206
2207 #[tokio::test]
2208 async fn register_cp_link_normalises_alias_to_canonical() {
2209 // Regression: CP link registration must store the
2210 // canonical record names. dispatch_cp_targets looks up by
2211 // canonical, so an alias-keyed entry is functionally dead.
2212 let db = PvDatabase::new();
2213 db.add_record(
2214 "SRC_REAL",
2215 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2216 )
2217 .await
2218 .unwrap();
2219 db.add_record(
2220 "DST_REAL",
2221 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2222 )
2223 .await
2224 .unwrap();
2225 db.add_alias("SRC_ALIAS", "SRC_REAL").await.unwrap();
2226 db.add_alias("DST_ALIAS", "DST_REAL").await.unwrap();
2227
2228 // Register using the alias forms (CP edge: passive_only = false).
2229 db.register_cp_link("SRC_ALIAS", "DST_ALIAS", false).await;
2230
2231 // Lookup must succeed via the canonical source name.
2232 let targets = db.get_cp_targets("SRC_REAL").await;
2233 assert_eq!(targets.len(), 1);
2234 assert_eq!(targets[0].record, "DST_REAL");
2235 assert!(!targets[0].passive_only);
2236 // Alias-keyed lookup must NOT have been registered.
2237 let alias_lookup = db.get_cp_targets("SRC_ALIAS").await;
2238 assert!(alias_lookup.is_empty());
2239 }
2240
2241 #[tokio::test]
2242 async fn aliases_for_record_returns_sorted_targets_only() {
2243 let db = PvDatabase::new();
2244 db.add_record(
2245 "TARGET",
2246 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2247 )
2248 .await
2249 .unwrap();
2250 db.add_record(
2251 "OTHER",
2252 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2253 )
2254 .await
2255 .unwrap();
2256 db.add_alias("ZZ", "TARGET").await.unwrap();
2257 db.add_alias("AA", "TARGET").await.unwrap();
2258 db.add_alias("MM", "OTHER").await.unwrap();
2259
2260 // Sorted, only TARGET's aliases.
2261 assert_eq!(
2262 db.aliases_for_record("TARGET").await,
2263 vec!["AA".to_string(), "ZZ".to_string()]
2264 );
2265 // OTHER's alone.
2266 assert_eq!(db.aliases_for_record("OTHER").await, vec!["MM".to_string()]);
2267 // Unknown record → empty, not None.
2268 assert!(db.aliases_for_record("MISSING").await.is_empty());
2269 }
2270
2271 #[tokio::test]
2272 async fn all_alias_names_returns_registered_aliases() {
2273 let db = PvDatabase::new();
2274 db.add_record(
2275 "TARGET",
2276 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2277 )
2278 .await
2279 .unwrap();
2280 db.add_alias("ALIAS_A", "TARGET").await.unwrap();
2281 db.add_alias("ALIAS_B", "TARGET").await.unwrap();
2282
2283 let mut aliases = db.all_alias_names().await;
2284 aliases.sort();
2285 assert_eq!(aliases, vec!["ALIAS_A".to_string(), "ALIAS_B".to_string()]);
2286 // Canonical names are NOT returned here.
2287 assert!(!aliases.contains(&"TARGET".to_string()));
2288 }
2289
2290 #[tokio::test]
2291 async fn complete_async_record_accepts_alias() {
2292 // Invariant audit: complete_async_record (the
2293 // entry point used by async device-support callbacks to
2294 // finish processing) must accept an alias name. Pre-fix it
2295 // walked `inner.records` directly and would
2296 // `ChannelNotFound` if the original name was an alias.
2297 let db = PvDatabase::new();
2298 db.add_record(
2299 "TARGET",
2300 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2301 )
2302 .await
2303 .unwrap();
2304 db.add_alias("ALIAS", "TARGET").await.unwrap();
2305
2306 // Use complete_async_record by alias — must not error.
2307 db.complete_async_record("ALIAS").await.unwrap();
2308 // And by canonical too — keeps existing behaviour.
2309 db.complete_async_record("TARGET").await.unwrap();
2310 }
2311
2312 #[tokio::test]
2313 async fn process_record_accepts_alias() {
2314 // Regression: process_record must accept an alias
2315 // name. Pre-fix it walked `inner.records` directly.
2316 let db = PvDatabase::new();
2317 db.add_record(
2318 "TARGET",
2319 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2320 )
2321 .await
2322 .unwrap();
2323 db.add_alias("ALIAS", "TARGET").await.unwrap();
2324
2325 // Both should succeed and reach the same record.
2326 db.process_record("TARGET").await.unwrap();
2327 db.process_record("ALIAS").await.unwrap();
2328
2329 // A bogus name still errors.
2330 assert!(db.process_record("MISSING").await.is_err());
2331 }
2332
2333 #[tokio::test]
2334 async fn process_record_with_links_accepts_alias_and_avoids_cycle() {
2335 // Regression: process_record_with_links normalises
2336 // the alias so that (a) the records-map lookup hits and
2337 // (b) the cycle-detection set doesn't treat alias and
2338 // canonical as two distinct entries (which would let a
2339 // self-loop slip past the visited check).
2340 let db = PvDatabase::new();
2341 db.add_record(
2342 "TARGET",
2343 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2344 )
2345 .await
2346 .unwrap();
2347 db.add_alias("ALIAS", "TARGET").await.unwrap();
2348
2349 let mut visited = std::collections::HashSet::new();
2350 db.process_record_with_links("ALIAS", &mut visited, 0)
2351 .await
2352 .unwrap();
2353
2354 // visited should contain the *canonical* name only.
2355 assert!(
2356 visited.contains("TARGET"),
2357 "visited must record the canonical name: {visited:?}",
2358 );
2359 assert!(
2360 !visited.contains("ALIAS"),
2361 "visited must NOT record the alias form: {visited:?}",
2362 );
2363 }
2364
2365 #[tokio::test]
2366 async fn alias_duplicate_rejected() {
2367 let db = PvDatabase::new();
2368 db.add_record(
2369 "TARGET",
2370 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2371 )
2372 .await
2373 .unwrap();
2374 db.add_alias("ALIAS", "TARGET").await.unwrap();
2375 // Re-registering the same alias name (even to the same target)
2376 // must fail — base behaviour: aliases are inserted once.
2377 let err = db.add_alias("ALIAS", "TARGET").await;
2378 assert!(err.is_err(), "duplicate alias name must be rejected");
2379 }
2380
2381 /// `add_pv`, `add_pv_with_hook`, and `add_record` must
2382 /// refuse to silently replace an existing registration. Mirrors
2383 /// epics-base C IOC which treats a duplicate `dbLoadRecords` name
2384 /// as a fatal load error.
2385 #[tokio::test]
2386 async fn add_pv_and_add_record_reject_duplicates_across_namespaces() {
2387 use crate::server::records::ai::AiRecord;
2388
2389 let db = PvDatabase::new();
2390 db.add_pv("A", EpicsValue::Double(1.0)).await.unwrap();
2391 // Same name as simple_pv — every namespace must see it.
2392 assert!(db.add_pv("A", EpicsValue::Double(2.0)).await.is_err());
2393 let noop_hook: crate::server::pv::WriteHook =
2394 std::sync::Arc::new(|_v, _ctx| Box::pin(async { Ok(()) }));
2395 assert!(
2396 db.add_pv_with_hook("A", EpicsValue::Double(2.0), noop_hook)
2397 .await
2398 .is_err()
2399 );
2400 assert!(
2401 db.add_record("A", Box::new(AiRecord::new(0.0)))
2402 .await
2403 .is_err()
2404 );
2405 assert!(db.add_alias("A", "A").await.is_err());
2406
2407 db.add_record("R", Box::new(AiRecord::new(0.0)))
2408 .await
2409 .unwrap();
2410 assert!(
2411 db.add_record("R", Box::new(AiRecord::new(1.0)))
2412 .await
2413 .is_err()
2414 );
2415 assert!(db.add_pv("R", EpicsValue::Double(0.0)).await.is_err());
2416 assert!(db.add_alias("R", "R").await.is_err());
2417
2418 db.add_alias("AL", "R").await.unwrap();
2419 assert!(db.add_pv("AL", EpicsValue::Double(0.0)).await.is_err());
2420 assert!(
2421 db.add_record("AL", Box::new(AiRecord::new(0.0)))
2422 .await
2423 .is_err()
2424 );
2425 }
2426
2427 /// Removing a record must purge aliases
2428 /// that pointed AT it. Otherwise the alias name stays
2429 /// "registered" forever and `add_pv` / `add_record` rejecting
2430 /// reuse causes a permanent name leak.
2431 #[tokio::test]
2432 async fn remove_record_purges_dangling_aliases() {
2433 use crate::server::records::ai::AiRecord;
2434
2435 let db = PvDatabase::new();
2436 db.add_record("R", Box::new(AiRecord::new(0.0)))
2437 .await
2438 .unwrap();
2439 db.add_alias("ALT1", "R").await.unwrap();
2440 db.add_alias("ALT2", "R").await.unwrap();
2441 // An alias that points elsewhere must NOT be touched.
2442 db.add_record("OTHER", Box::new(AiRecord::new(0.0)))
2443 .await
2444 .unwrap();
2445 db.add_alias("KEEPER", "OTHER").await.unwrap();
2446
2447 assert!(db.remove_record("R").await);
2448
2449 // Both aliases pointing at R should be gone — `add_pv` of
2450 // those names succeeds again.
2451 db.add_pv("ALT1", EpicsValue::Double(0.0)).await.unwrap();
2452 db.add_pv("ALT2", EpicsValue::Double(0.0)).await.unwrap();
2453 // The unrelated alias must survive.
2454 assert_eq!(db.resolve_alias("KEEPER").await, Some("OTHER".to_string()));
2455 }
2456
2457 /// `add_alias` must reject collisions with
2458 /// every namespace, including simple PVs (which the pre-fix
2459 /// code missed).
2460 #[tokio::test]
2461 async fn add_alias_rejects_simple_pv_collision() {
2462 use crate::server::records::ai::AiRecord;
2463
2464 let db = PvDatabase::new();
2465 db.add_pv("PVX", EpicsValue::Double(0.0)).await.unwrap();
2466 db.add_record("TARGET", Box::new(AiRecord::new(0.0)))
2467 .await
2468 .unwrap();
2469 // alias name "PVX" collides with the simple PV — must fail.
2470 assert!(db.add_alias("PVX", "TARGET").await.is_err());
2471 }
2472
2473 /// Concurrent `add_pv` and `add_record` with
2474 /// the same name must not deadlock and must serialize so that
2475 /// exactly one succeeds. Pre-fix the two methods grabbed
2476 /// different write locks first, opening a cross-lock-order
2477 /// deadlock window.
2478 #[tokio::test]
2479 async fn concurrent_add_pv_and_add_record_do_not_deadlock() {
2480 use crate::server::records::ai::AiRecord;
2481
2482 let db = std::sync::Arc::new(PvDatabase::new());
2483 let db1 = db.clone();
2484 let db2 = db.clone();
2485 let h1 = tokio::spawn(async move { db1.add_pv("RACE", EpicsValue::Double(1.0)).await });
2486 let h2 =
2487 tokio::spawn(async move { db2.add_record("RACE", Box::new(AiRecord::new(0.0))).await });
2488 // Both complete within a reasonable bound — pre-fix this
2489 // could hang because T1 holds simple_pvs.write and waits
2490 // for records.read while T2 holds records.write and waits
2491 // for simple_pvs.read.
2492 let r1 = tokio::time::timeout(std::time::Duration::from_secs(2), h1)
2493 .await
2494 .expect("add_pv must not block on add_record");
2495 let r2 = tokio::time::timeout(std::time::Duration::from_secs(2), h2)
2496 .await
2497 .expect("add_record must not block on add_pv");
2498 let r1 = r1.unwrap();
2499 let r2 = r2.unwrap();
2500 // Exactly one of the two wins; the other reports
2501 // "already registered".
2502 assert!(
2503 (r1.is_ok() && r2.is_err()) || (r1.is_err() && r2.is_ok()),
2504 "exactly one of the racing inserts must succeed: r1={r1:?} r2={r2:?}",
2505 );
2506 }
2507
2508 #[tokio::test]
2509 async fn existence_gate_blocks_cached_simple_pv_per_request() {
2510 // A cached simple PV must re-pass the installed existence gate on
2511 // both the search (`has_name_from`) and create (`find_entry_from`)
2512 // paths. Records bypass the gate. With no gate the short-circuit
2513 // is unchanged (plain-IOC behaviour).
2514 use std::net::SocketAddr;
2515
2516 let db = PvDatabase::new();
2517 db.add_pv("SHADOW:x", EpicsValue::Double(1.0))
2518 .await
2519 .unwrap();
2520 db.add_record(
2521 "REC",
2522 Box::new(crate::server::records::ai::AiRecord::new(0.0)),
2523 )
2524 .await
2525 .unwrap();
2526
2527 let denied: SocketAddr = "127.0.0.1:5064".parse().unwrap();
2528 let allowed: SocketAddr = "192.0.2.5:5064".parse().unwrap();
2529
2530 // No gate installed: the cached simple PV resolves unconditionally.
2531 assert!(db.has_name_from("SHADOW:x", Some(denied)).await);
2532 assert!(db.find_entry_from("SHADOW:x", Some(denied)).await.is_some());
2533
2534 // Gate denies the simple PV only for `denied` (the gateway's
2535 // host-scoped `.pvlist` admission has exactly this shape).
2536 let gate: ExistenceGate = Arc::new(move |name, peer| {
2537 Box::pin(async move { !(name == "SHADOW:x" && peer == Some(denied)) })
2538 });
2539 db.set_existence_gate(gate).await;
2540
2541 // Denied peer: does-not-exist on both paths despite the PV being
2542 // cached in `simple_pvs`.
2543 assert!(!db.has_name_from("SHADOW:x", Some(denied)).await);
2544 assert!(db.find_entry_from("SHADOW:x", Some(denied)).await.is_none());
2545
2546 // Allowed peer: still resolves.
2547 assert!(db.has_name_from("SHADOW:x", Some(allowed)).await);
2548 assert!(
2549 db.find_entry_from("SHADOW:x", Some(allowed))
2550 .await
2551 .is_some()
2552 );
2553
2554 // Records are never gateway-managed — the gate must not gate them
2555 // even for the denied peer.
2556 assert!(db.has_name_from("REC", Some(denied)).await);
2557 assert!(db.find_entry_from("REC", Some(denied)).await.is_some());
2558 }
2559
2560 /// `record_link_fields` must surface a record's device-support `INP`
2561 /// link. An `ai`'s `INP` is a `DBF_INLINK` field stored in
2562 /// `common.inp` — it is not a `DbFieldType::String` entry in
2563 /// `field_list()` — so the earlier `field_list()` scan filtered by
2564 /// `String` silently dropped it. The pvalink install scan walks this
2565 /// method, so a Passive `ai` carrying a CP/CPP pvalink `INP` never had
2566 /// its monitor opened at iocInit. Enumerating the canonical
2567 /// `common.inp` storage fixes it; C `dbpvar`/`dbcar` likewise dump
2568 /// every link field including device-support INP/OUT.
2569 #[tokio::test]
2570 async fn record_link_fields_surfaces_device_support_inp() {
2571 use crate::server::record::ParsedLink;
2572 use crate::server::records::ai::AiRecord;
2573
2574 let db = PvDatabase::new();
2575 db.add_record("AI", Box::new(AiRecord::new(0.0)))
2576 .await
2577 .unwrap();
2578 // Device-support INP lives in `common.inp` (DBF_INLINK), the
2579 // exact storage a `field_list()` String scan cannot reach.
2580 {
2581 let rec = db.get_record("AI").await.unwrap();
2582 rec.write().await.common.inp = "pva://mini:current?proc=CP".to_string();
2583 }
2584
2585 let links = db.record_link_fields("AI").await;
2586 let inp = links
2587 .iter()
2588 .find(|(f, _, _)| f == "INP")
2589 .unwrap_or_else(|| panic!("INP link must be surfaced, got {links:?}"));
2590 assert_eq!(inp.1, "pva://mini:current?proc=CP");
2591 assert!(
2592 matches!(inp.2, ParsedLink::Pva(_)),
2593 "a pva:// INP must parse to ParsedLink::Pva, got {:?}",
2594 inp.2
2595 );
2596 }
2597}