epics_base_rs/server/database/processing.rs
1use std::collections::HashSet;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use crate::error::{CaError, CaResult};
6use crate::runtime::sync::RwLock;
7use crate::server::record::{NotifyWaitSet, RecordInstance};
8use crate::types::EpicsValue;
9
10use super::{PvDatabase, apply_timestamp};
11
12/// A cancellable, generation-gated handle that re-enters an async record's
13/// `process()` exactly once.
14///
15/// C parity: epics-base `callbackRequest` / `callbackRequestDelayed`
16/// (`callback.c`) post a one-shot callback that later runs the record's
17/// `(*prset->process)(precord)` directly, bypassing `dbProcess`'s PACT
18/// entry guard. Here, firing the token re-enters via
19/// [`PvDatabase::process_record_continuation`] (the owner-driven
20/// continuation that also bypasses the PACT guard).
21///
22/// # Cancellation is structural, not a runtime check
23///
24/// The record owns a monotonic generation counter (`reprocess_generation`).
25/// Minting a token snapshots that counter as the token's `epoch` *after*
26/// bumping it, so:
27///
28/// - minting a newer token for the same record (C `callbackRequestDelayed`
29/// replacing an outstanding delayed callback), or
30/// - [`PvDatabase::cancel_async_reentry`] (C `callbackCancelDelayed`),
31///
32/// each advance the counter past every outstanding token's `epoch`. A
33/// stale token therefore re-enters *nothing*: [`AsyncToken::fire`] is the
34/// sole re-entry path, the epoch comparison is owned in one place, and the
35/// token is consumed (`self` by value) so it cannot fire twice. A consumer
36/// never writes an `if generation == ...` guard — it holds the token and
37/// calls `fire`; the no-op-when-stale is guaranteed by construction.
38pub struct AsyncToken {
39 /// Canonical record name to re-enter.
40 name: String,
41 /// Shared generation counter owned by the record
42 /// (`RecordInstance::reprocess_generation`).
43 generation: Arc<AtomicU64>,
44 /// Generation value captured at mint time. The token is current iff
45 /// `generation == epoch`.
46 epoch: u64,
47}
48
49impl AsyncToken {
50 /// The record this token re-enters.
51 pub fn record_name(&self) -> &str {
52 &self.name
53 }
54
55 /// True iff this token is still the current generation — no newer
56 /// token was minted and no [`PvDatabase::cancel_async_reentry`] has
57 /// run for the record since this token was minted. Read-only.
58 pub fn is_current(&self) -> bool {
59 self.generation.load(Ordering::Acquire) == self.epoch
60 }
61
62 /// Cancel this token (C `callbackCancelDelayed` for the holder's own
63 /// pending re-entry): advance the generation so this and any other
64 /// outstanding token for the record become stale, then consume the
65 /// token. Use when the holder itself decides not to re-enter; use
66 /// [`PvDatabase::cancel_async_reentry`] to cancel a token already
67 /// handed to a timer / notify task.
68 pub fn cancel(self) {
69 self.generation.fetch_add(1, Ordering::AcqRel);
70 }
71
72 /// Fire the continuation: if still current, re-enter the record's
73 /// `process()` via [`PvDatabase::process_record_continuation`]. A
74 /// stale (superseded / cancelled) token is a no-op. Consumes the
75 /// token so it cannot fire twice.
76 pub async fn fire(self, db: &PvDatabase) -> CaResult<()> {
77 if self.generation.load(Ordering::Acquire) != self.epoch {
78 return Ok(());
79 }
80 let mut visited = HashSet::new();
81 db.process_record_continuation(&self.name, &mut visited, 0)
82 .await
83 }
84}
85
86/// A cycle-free handle for driving async-side database updates from
87/// OUTSIDE a record's `process()` cycle.
88///
89/// Wraps a [`std::sync::Weak`] reference to the database: a record stashes
90/// it (via [`crate::server::record::Record::set_async_context`]) without
91/// creating an ownership cycle — the database owns the record, so a strong
92/// `Arc<PvDatabaseInner>` stored on the record would leak the whole
93/// database. Every call upgrades the `Weak` to a temporary [`PvDatabase`];
94/// once the last strong owner drops, the upgrade fails and the call is a
95/// no-op (nothing is stranded).
96///
97/// This is the out-of-band counterpart to the in-band re-entry
98/// [`crate::server::record::ProcessAction`]s: a driver / callback thread
99/// (asyn TRACE post, AQR cancel, motor intermediate readback) holds the
100/// handle and pushes field updates or wires a completion-driven re-entry
101/// without going through `process()`. It exposes exactly the c401e2f0
102/// PACT primitive surface, each call guarded by the live-database check.
103#[derive(Clone)]
104pub struct AsyncDbHandle {
105 inner: std::sync::Weak<super::PvDatabaseInner>,
106}
107
108impl AsyncDbHandle {
109 /// Upgrade to a temporary owning [`PvDatabase`], or `None` if the
110 /// database has been dropped.
111 fn db(&self) -> Option<PvDatabase> {
112 self.inner.upgrade().map(|inner| PvDatabase { inner })
113 }
114
115 /// True while the backing database is still alive.
116 pub fn is_alive(&self) -> bool {
117 self.inner.strong_count() > 0
118 }
119
120 /// Out-of-band field post — see [`PvDatabase::post_fields`]. Returns an
121 /// empty `Vec` (no-op) if the database has been dropped.
122 pub async fn post_fields(
123 &self,
124 name: &str,
125 fields: Vec<(String, EpicsValue)>,
126 ) -> CaResult<Vec<String>> {
127 match self.db() {
128 Some(db) => db.post_fields(name, fields).await,
129 None => Ok(Vec::new()),
130 }
131 }
132
133 /// Resolve a link's target field type for the sseq link-status
134 /// diagnostics — see [`PvDatabase::link_target_field_type`]. `None` if
135 /// the link is constant / external / unresolvable, or the database is
136 /// gone. (Distinct from the free `server::record::link_field_type`,
137 /// which returns the link *class* `LinkType`, not the target's type.)
138 pub async fn link_target_field_type(&self, link: &str) -> Option<crate::types::DbFieldType> {
139 match self.db() {
140 Some(db) => db.link_target_field_type(link).await,
141 None => None,
142 }
143 }
144
145 /// Mint an async re-entry token — see [`PvDatabase::mint_async_token`].
146 /// `None` if the record is absent or the database has been dropped.
147 pub async fn mint_async_token(&self, name: &str) -> Option<AsyncToken> {
148 match self.db() {
149 Some(db) => db.mint_async_token(name).await,
150 None => None,
151 }
152 }
153
154 /// Cancel an outstanding async re-entry — see
155 /// [`PvDatabase::cancel_async_reentry`]. No-op if the database is gone.
156 pub async fn cancel_async_reentry(&self, name: &str) {
157 if let Some(db) = self.db() {
158 db.cancel_async_reentry(name).await;
159 }
160 }
161
162 /// Arm a put-notify wait-set — see [`PvDatabase::new_put_notify`].
163 /// Database-independent (re-exported associated fn).
164 pub fn new_put_notify() -> (
165 Arc<NotifyWaitSet>,
166 crate::runtime::sync::oneshot::Receiver<()>,
167 ) {
168 PvDatabase::new_put_notify()
169 }
170
171 /// Wire a completion oneshot to an async re-entry — see
172 /// [`PvDatabase::reprocess_on_notify`]. `None` if the database is gone
173 /// (the `completion` receiver is dropped, stranding nothing).
174 pub fn reprocess_on_notify(
175 &self,
176 token: AsyncToken,
177 completion: crate::runtime::sync::oneshot::Receiver<()>,
178 ) -> Option<tokio::task::JoinHandle<()>> {
179 self.db()
180 .map(|db| db.reprocess_on_notify(token, completion))
181 }
182
183 /// Issue a non-blocking put-with-completion to an OUT link — see
184 /// [`PvDatabase::put_link_notify`]. `None` if the database is gone or
185 /// the source record is missing.
186 pub async fn put_link_notify(
187 &self,
188 record_name: &str,
189 link_str: &str,
190 value: EpicsValue,
191 ) -> Option<crate::runtime::sync::oneshot::Receiver<()>> {
192 match self.db() {
193 Some(db) => db.put_link_notify(record_name, link_str, value).await,
194 None => None,
195 }
196 }
197}
198
199/// C `dbNotifyAdd`: a will-process PP target (FLNK / OUT) joins the active
200/// put-notify wait-set exactly once, so the completion waits for it. Called
201/// only on the `!pact` (will-process) branch — a busy target sets RPRO and
202/// does not join (matching the pre-fix drop behaviour), and the
203/// `notify.is_none()` guard prevents a double-join when a record is reached
204/// again within the same chain.
205pub(super) fn join_put_notify(
206 target: &mut RecordInstance,
207 src_notify: Option<&Arc<NotifyWaitSet>>,
208) {
209 if target.notify.is_none() {
210 if let Some(ws) = src_notify {
211 target.notify = Some(ws.clone());
212 ws.enter();
213 }
214 }
215}
216
217/// C `dbNotifyCompletion`: this record finished its contribution to the
218/// put-notify (sync completion, async completion, or SDIS-disable bail).
219/// Take its wait-set membership and leave — the completion oneshot fires on
220/// the `leave` that empties the set. Idempotent: a record not in any
221/// put-notify is a no-op.
222fn complete_put_notify(inst: &mut RecordInstance) {
223 if let Some(ws) = inst.notify.take() {
224 ws.leave();
225 }
226}
227
228/// If a CA TSEL link's pvname targets a record's `.TIME` field, return
229/// the record name with the `.TIME` suffix stripped; otherwise `None`.
230///
231/// Mirrors C `TSEL_modified` (dbLink.c:80-86): a `PV_LINK` tsel whose
232/// pvname contains `.TIME` is flagged `DBLINK_FLAG_TSELisTIME` and the
233/// name is truncated at `.TIME` to address the record. Matched on the
234/// `.TIME` suffix (the realistic spelling) case-insensitively, to stay
235/// consistent with the DB branch's `field.eq_ignore_ascii_case("TIME")`.
236fn ca_tsel_time_record(pv: &str) -> Option<&str> {
237 let idx = pv.len().checked_sub(".TIME".len())?;
238 pv[idx..]
239 .eq_ignore_ascii_case(".TIME")
240 .then_some(&pv[..idx])
241}
242
243/// Convert an lset `(seconds_past_epoch, nanos, userTag)` timestamp
244/// triple into the record-side `(SystemTime, userTag)` pair, clamping
245/// seconds/nanos to the valid `Duration` range. Shared by the TSEL
246/// `.TIME` Ca arm and the non-local Db arm — both read a `ca://` `.TIME`
247/// source through `external_link_time` and adopt the result identically.
248fn ext_time_pair((secs, ns, utag): (i64, i32, u64)) -> (std::time::SystemTime, u64) {
249 let secs = secs.max(0) as u64;
250 let ns = (ns.max(0) as u32).min(999_999_999);
251 (
252 std::time::UNIX_EPOCH + std::time::Duration::new(secs, ns),
253 utag,
254 )
255}
256
257/// The source record's put-propagation context for the forward-link tail.
258/// C `processTarget` (dbDbLink.c:460-474) carries `psrc->putf` and
259/// `psrc->ppn` to each target as a unit — the PUTF bit and the put-notify
260/// wait-set always travel together. Bundled so the tail threads one
261/// snapshot instead of a `(putf, notify)` pair.
262#[derive(Clone, Copy)]
263struct PutNotifyCtx<'a> {
264 putf: bool,
265 notify: Option<&'a Arc<NotifyWaitSet>>,
266}
267
268/// Result of the simulation-mode check.
269///
270/// C `aiRecord.c:151-168` handles simulation entirely inside
271/// `readValue()`; `process()` then ALWAYS runs `convert`/`checkAlarms`/
272/// `monitor`/`recGblFwdLink(prec)`. A simulated record therefore must
273/// NOT skip the forward-link / CP / RPRO tail — only the device read
274/// and record-support body are replaced by the SIOL round-trip.
275enum SimOutcome {
276 /// SIMM disabled / no simulation link configured: run the record
277 /// body normally.
278 NotSimulated,
279 /// Simulation handled the record value (SIOL read/write done).
280 /// The caller must still run the forward-link / CP / RPRO tail
281 /// exactly as `recGblFwdLink` does for a real process cycle.
282 Simulated,
283}
284
285impl PvDatabase {
286 /// Process a record by name (process_local + notify).
287 /// Alias-aware (epics-base PR #336).
288 pub async fn process_record(&self, name: &str) -> CaResult<()> {
289 self.process_record_inner(name, true).await
290 }
291
292 /// `process_record` variant for a caller that already
293 /// owns the record's advisory write gate — the QSRV atomic group
294 /// PUT applying a `+proc` member. The gate `Mutex` is not
295 /// reentrant; the atomic group path MUST use this entry. See
296 /// [`crate::server::database::PvDatabase::lock_records`].
297 pub async fn process_record_already_locked(&self, name: &str) -> CaResult<()> {
298 self.process_record_inner(name, false).await
299 }
300
301 async fn process_record_inner(&self, name: &str, acquire_gate: bool) -> CaResult<()> {
302 let rec = self.get_record(name).await;
303
304 if let Some(rec) = rec {
305 // advisory write gate (`dbScanLock` analogue). A
306 // QSRV atomic group with a `+proc` member holds this
307 // record's gate via `lock_records`; a direct
308 // `process_record` on the same backing record must block
309 // until the atomic group transaction completes. Skipped
310 // when the caller already owns the gate.
311 let _record_gate = if acquire_gate {
312 let canonical = self
313 .resolve_alias(name)
314 .await
315 .unwrap_or_else(|| name.to_string());
316 Some(self.lock_record(&canonical).await)
317 } else {
318 None
319 };
320 let (snapshot, alarm_posts) = {
321 let mut instance = rec.write().await;
322 instance.process_local()?
323 };
324 // Notify outside lock
325 let instance = rec.read().await;
326 instance.notify_from_snapshot(&snapshot);
327 // Post the alarm fields (SEVR/STAT/ACKS) with their
328 // individual C masks — see `process_local` / recGblResetAlarms.
329 for &(field, mask) in &alarm_posts {
330 instance.notify_field(field, mask);
331 }
332 Ok(())
333 } else {
334 Err(CaError::ChannelNotFound(name.to_string()))
335 }
336 }
337
338 /// Process a record with full link handling (INP -> process -> alarms -> OUT -> FLNK).
339 /// Uses visited set for cycle detection and depth limit.
340 ///
341 /// Foreign-caller entry: FLNK dispatch, scan loop, scan_event, CA put,
342 /// process(PROC=1) etc. Hits the PACT entry guard (mirrors C `dbProcess`
343 /// at `dbAccess.c:537-559`) when the record is mid-async.
344 ///
345 /// this is a *foreign* full-processing entry, so it acquires
346 /// the record's advisory write gate (`dbScanLock` analogue) for the
347 /// entry record before processing. A QSRV atomic group or pvalink
348 /// atomic scan-on-update epoch that holds `lock_records` over the
349 /// same record blocks a foreign scan/event/FLNK-dispatch caller
350 /// here, and vice versa — restoring the `DBManyLock` exclusion. The
351 /// recursive FLNK / OUT / CP fan-out within one chain does NOT
352 /// re-acquire the gate (`process_record_with_links_recursive`),
353 /// mirroring C `processTarget` (`dbDbLink.c:436`) which asserts the
354 /// target's lock set is already owned by the calling thread; the
355 /// `visited` cycle guard prevents re-processing the entry record.
356 pub fn process_record_with_links<'a>(
357 &'a self,
358 name: &'a str,
359 visited: &'a mut HashSet<String>,
360 depth: usize,
361 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
362 Box::pin(async move {
363 self.process_record_with_links_inner(name, visited, depth, false, true)
364 .await
365 })
366 }
367
368 /// full-processing entry for a caller that already owns the
369 /// record's advisory write gate via [`PvDatabase::lock_records`] —
370 /// the QSRV atomic group GET/PUT and the pvalink atomic
371 /// scan-on-update epoch. The advisory gate `Mutex` is not
372 /// reentrant; a transaction owner holding `lock_records` over the
373 /// member set MUST use this entry to scan a member record, or it
374 /// would deadlock against its own epoch guard. Foreign (non-owner)
375 /// callers must use [`Self::process_record_with_links`] so the gate
376 /// is taken.
377 pub fn process_record_with_links_already_locked<'a>(
378 &'a self,
379 name: &'a str,
380 visited: &'a mut HashSet<String>,
381 depth: usize,
382 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
383 Box::pin(async move {
384 self.process_record_with_links_inner(name, visited, depth, false, false)
385 .await
386 })
387 }
388
389 /// recursive FLNK / OUT / CP fan-out entry within a single
390 /// processing chain. Does NOT re-acquire the advisory write gate:
391 /// the chain is one transaction whose entry record's gate is
392 /// already held by the foreign entry, and C `processTarget`
393 /// (`dbDbLink.c:436`) processes a link target under the lock set
394 /// already owned by the calling thread. Re-acquiring per chain
395 /// member would also create a lock-ordering deadlock between
396 /// reverse FLNK chains.
397 pub(crate) fn process_record_with_links_recursive<'a>(
398 &'a self,
399 name: &'a str,
400 visited: &'a mut HashSet<String>,
401 depth: usize,
402 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
403 Box::pin(async move {
404 self.process_record_with_links_inner(name, visited, depth, false, false)
405 .await
406 })
407 }
408
409 /// Owner-driven continuation re-entry — bypasses the PACT entry guard.
410 ///
411 /// Used by `ProcessAction::ReprocessAfter` timer fires: the spawned
412 /// re-entry task IS the owner of the async cycle, equivalent to C
413 /// `callbackRequestDelayed`'s direct call to the record's `process()`
414 /// (which bypasses `dbProcess`). Foreign callers must still go through
415 /// `process_record_with_links` so FLNK / scan / CA put cannot race
416 /// during the wait window.
417 ///
418 /// the timer fire is a fresh task — the original cycle's
419 /// advisory gate was released when `process_record_with_links`
420 /// returned async-pending. In C, `callbackRequestDelayed` dispatches
421 /// through a callback that re-takes `dbScanLock(precord)` for the
422 /// completion `process()`. This entry therefore re-acquires the
423 /// advisory write gate, so the continuation cannot interleave with a
424 /// QSRV atomic group or another foreign scan of the same record.
425 pub fn process_record_continuation<'a>(
426 &'a self,
427 name: &'a str,
428 visited: &'a mut HashSet<String>,
429 depth: usize,
430 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
431 Box::pin(async move {
432 self.process_record_with_links_inner(name, visited, depth, true, true)
433 .await
434 })
435 }
436
437 /// A cycle-free [`AsyncDbHandle`] for this database, handed to each
438 /// record via [`crate::server::record::Record::set_async_context`] at
439 /// registration. Holds only a `Weak` reference, so a record stashing
440 /// it never keeps the database alive.
441 pub fn async_handle(&self) -> AsyncDbHandle {
442 AsyncDbHandle {
443 inner: Arc::downgrade(&self.inner),
444 }
445 }
446
447 /// Mint a fresh async re-entry [`AsyncToken`] for `name`.
448 ///
449 /// Minting advances the record's generation counter, so any
450 /// previously-minted token for the same record is superseded — its
451 /// [`AsyncToken::fire`] becomes a structural no-op. This mirrors C
452 /// `callbackRequestDelayed` replacing an outstanding delayed callback
453 /// for a record. `name` must be the canonical record name (the value
454 /// of `RecordInstance::name`). Returns `None` if the record is absent.
455 pub async fn mint_async_token(&self, name: &str) -> Option<AsyncToken> {
456 let records = self.inner.records.read().await;
457 let rec = records.get(name)?;
458 let generation = rec.read().await.reprocess_generation.clone();
459 let epoch = generation.fetch_add(1, Ordering::AcqRel) + 1;
460 Some(AsyncToken {
461 name: name.to_string(),
462 generation,
463 epoch,
464 })
465 }
466
467 /// Cancel any outstanding async re-entry token for `name` (C
468 /// `callbackCancelDelayed`): advance the record's generation counter so
469 /// every previously-minted [`AsyncToken`] for it becomes stale and its
470 /// `fire` is a no-op. A subsequent [`Self::mint_async_token`] produces a
471 /// fresh, current token. No-op if the record is absent.
472 pub async fn cancel_async_reentry(&self, name: &str) {
473 let records = self.inner.records.read().await;
474 if let Some(rec) = records.get(name) {
475 rec.read()
476 .await
477 .reprocess_generation
478 .fetch_add(1, Ordering::AcqRel);
479 }
480 }
481
482 /// Post an async-side field update for `name` — the C `db_post_events`
483 /// analogue called from device-support / async-callback context.
484 ///
485 /// Each `(field, value)` is written through the internal put (bypassing
486 /// the read-only field gate, like a record's own `process()` writes)
487 /// and a monitor event is posted with `DBE_VALUE | DBE_LOG` — the mask C
488 /// device support uses for an out-of-process value post
489 /// (`db_post_events(precord, &prec->field, DBE_VALUE | DBE_LOG)`).
490 /// Metadata-class writes invalidate the metadata cache via
491 /// `notify_field_written`, honouring the snapshot-cache contract.
492 ///
493 /// Unlike [`Self::complete_async_record`], this runs *no* alarm /
494 /// timestamp / FLNK tail: it is the immediate "push these fields to
495 /// monitors now" primitive (e.g. asyn TRACE info, motor intermediate
496 /// readback) that is independent of any process cycle. Returns the
497 /// field names actually posted, or [`CaError::ChannelNotFound`] if the
498 /// record is absent.
499 pub async fn post_fields(
500 &self,
501 name: &str,
502 fields: Vec<(String, EpicsValue)>,
503 ) -> CaResult<Vec<String>> {
504 let rec = {
505 let records = self.inner.records.read().await;
506 records.get(name).cloned()
507 };
508 let rec = rec.ok_or_else(|| CaError::ChannelNotFound(name.to_string()))?;
509 let mut inst = rec.write().await;
510 let mut posted = Vec::with_capacity(fields.len());
511 for (field, value) in fields {
512 inst.record.put_field_internal(&field, value)?;
513 // Snapshot-cache contract: a metadata-class write must
514 // invalidate the cache before the monitor snapshot is built.
515 inst.notify_field_written(&field);
516 inst.notify_field(
517 &field,
518 crate::server::recgbl::EventMask::VALUE | crate::server::recgbl::EventMask::LOG,
519 );
520 posted.push(field);
521 }
522 Ok(posted)
523 }
524
525 /// Resolve a link's target field [`DbFieldType`] for a LOCAL `DB_LINK`,
526 /// or `None` for a constant / external / unresolvable link.
527 ///
528 /// Parity of C `dbGetLinkDBFtype` as `sseqRecord.c:checkLinks`
529 /// (sseqRecord.c:884-941) uses it to fill the `DTn`/`LTn` diagnostics:
530 /// a `DB_LINK` whose target record is on this IOC reports its addressed
531 /// field's type (C `dbNameToAddr` → `pAddr->field_type`). A constant or
532 /// `CA`/`PVA` (external) link returns `None` — epics-base-rs has no
533 /// client-side introspection of a remote field's type, so the caller
534 /// renders those as the `DBF_unknown` sentinel.
535 pub(crate) async fn link_target_field_type(
536 &self,
537 link: &str,
538 ) -> Option<crate::types::DbFieldType> {
539 let db = match crate::server::record::parse_link_v2(link) {
540 crate::server::record::ParsedLink::Db(db) => db,
541 _ => return None,
542 };
543 let rec = self.get_record(&db.record).await?;
544 let inst = rec.read().await;
545 let field = if db.field.is_empty() {
546 "VAL"
547 } else {
548 db.field.as_str()
549 };
550 inst.record
551 .field_list()
552 .iter()
553 .find(|f| f.name.eq_ignore_ascii_case(field))
554 .map(|f| f.dbf_type)
555 }
556
557 /// Create a put-notify wait-set for a downstream operation a record is
558 /// about to drive, returning the wait-set (to attach to the downstream
559 /// target instance's `notify`) and the completion receiver.
560 ///
561 /// C `dbNotify.c` `processNotify`: the set arms `pending = 1` for the
562 /// downstream operation and fires the oneshot when that slot (plus any
563 /// FLNK/OUT chain members that `enter` it) drains to zero — i.e. on
564 /// `dbNotifyCompletion`. Pair with [`Self::reprocess_on_notify`] to
565 /// re-enter a waiting record when the downstream completes (SSEQ
566 /// `WAITn`).
567 pub fn new_put_notify() -> (
568 Arc<NotifyWaitSet>,
569 crate::runtime::sync::oneshot::Receiver<()>,
570 ) {
571 let (tx, rx) = crate::runtime::sync::oneshot::channel();
572 (NotifyWaitSet::new(tx), rx)
573 }
574
575 /// Wire a downstream put-notify completion to an async re-entry: spawn a
576 /// task that awaits `completion` (the oneshot from
577 /// [`Self::new_put_notify`], fired on `dbNotifyCompletion`) and then
578 /// `token.fire`s, re-entering the waiting record's `process()`. A
579 /// superseded / cancelled token re-enters nothing. Returns the spawned
580 /// task handle; fire-and-forget callers may drop it.
581 pub fn reprocess_on_notify(
582 &self,
583 token: AsyncToken,
584 completion: crate::runtime::sync::oneshot::Receiver<()>,
585 ) -> tokio::task::JoinHandle<()> {
586 let db = self.clone();
587 tokio::spawn(async move {
588 // `Err` means the sender was dropped without firing (the
589 // downstream op vanished); treat it the same as completion so a
590 // waiting record is never stranded — `fire` is a no-op if the
591 // token was meanwhile superseded.
592 let _ = completion.await;
593 let _ = token.fire(&db).await;
594 })
595 }
596
597 /// Issue a put-WITH-completion to an OUT link and hand the caller only
598 /// the completion receiver — the non-blocking sibling of
599 /// [`Self::reprocess_on_notify`].
600 ///
601 /// Each call mints its own put-notify wait-set (C `dbProcessNotify`),
602 /// writes the link through it with the source record's committed PUTF /
603 /// alarm propagated (C `recGblInheritSevrMsg`), releases the initiator
604 /// count, and returns the oneshot that fires on `dbNotifyCompletion`.
605 /// The caller owns when (and whether) to await each receiver, so several
606 /// puts can be outstanding at once — unlike
607 /// [`ProcessAction::WriteDbLinkNotify`], which wires the completion
608 /// straight to a single superseding async re-entry token and so allows
609 /// only one outstanding put per record. This is the seam C
610 /// `calcApp/src/sseqRecord.c` needs to run multiple `WAITn` put-callbacks
611 /// concurrently in flight (`processNextLink`).
612 ///
613 /// `record_name` is the source whose PUTF/alarm propagate into the
614 /// target, `link_str` the already-resolved OUT link spelling, `value`
615 /// the value to write. `None` if the source record is gone; an empty
616 /// `link_str` returns a receiver that fires immediately (nothing joined
617 /// the set).
618 pub async fn put_link_notify(
619 &self,
620 record_name: &str,
621 link_str: &str,
622 value: EpicsValue,
623 ) -> Option<crate::runtime::sync::oneshot::Receiver<()>> {
624 let (src_putf, src_alarm) = {
625 let rec = {
626 let records = self.inner.records.read().await;
627 records.get(record_name)?.clone()
628 };
629 let instance = rec.read().await;
630 (
631 instance.common.putf,
632 super::links::LinkAlarm {
633 stat: instance.common.stat,
634 sevr: instance.common.sevr,
635 amsg: instance.common.amsg.clone(),
636 },
637 )
638 };
639 let (waitset, completion) = Self::new_put_notify();
640 if !link_str.is_empty() {
641 let parsed = crate::server::record::parse_link_v2(link_str);
642 // Seed the cycle-guard with the source so a target linking back
643 // does not re-process it, exactly as a top-level OUT-link write
644 // does (`process_record_with_links_inner` inserts its own name).
645 let mut visited = HashSet::new();
646 visited.insert(record_name.to_string());
647 self.write_out_link_value(
648 &parsed,
649 value,
650 super::links::OutLinkSrc {
651 putf: src_putf,
652 notify: Some(&waitset),
653 alarm: &src_alarm,
654 },
655 &mut visited,
656 0,
657 )
658 .await;
659 }
660 // Release the initiator's own count (C `dbProcessNotify` holds one
661 // count for the requester and drops it after issuing the put). The
662 // set then drains — firing `completion` — when the downstream
663 // target(s) that joined via `join_put_notify` finish, or immediately
664 // when the link was empty / the target completed synchronously.
665 waitset.leave();
666 Some(completion)
667 }
668
669 async fn process_record_with_links_inner(
670 &self,
671 name: &str,
672 visited: &mut HashSet<String>,
673 depth: usize,
674 is_continuation: bool,
675 acquire_gate: bool,
676 ) -> CaResult<()> {
677 const MAX_LINK_DEPTH: usize = 16;
678 const MAX_LINK_OPS: usize = 256;
679
680 // Normalise to the canonical record name once at entry — both
681 // for cycle-detection (`visited` would otherwise treat alias
682 // and canonical as distinct entries) and for the records-map
683 // lookup below. Mirrors epics-base PR #336.
684 let canonical_owned;
685 let name: &str = if let Some(target) = self.resolve_alias(name).await {
686 canonical_owned = target;
687 &canonical_owned
688 } else {
689 name
690 };
691
692 if depth >= MAX_LINK_DEPTH {
693 eprintln!("link chain depth limit reached at record {name}");
694 return Ok(());
695 }
696 if visited.len() >= MAX_LINK_OPS {
697 eprintln!("link chain ops budget exhausted at record {name}");
698 return Ok(());
699 }
700 if !visited.insert(name.to_string()) {
701 return Ok(()); // Cycle detected, skip
702 }
703
704 let rec = {
705 let records = self.inner.records.read().await;
706 records.get(name).cloned()
707 };
708
709 let rec = match rec {
710 Some(r) => r,
711 None => return Err(CaError::ChannelNotFound(name.to_string())),
712 };
713
714 // advisory write gate (`dbScanLock(precord)` analogue).
715 // A foreign full-processing entry (scan loop, scan_event, FLNK
716 // dispatch from another chain, CA put, PINI/startup) acquires
717 // the entry record's gate so it cannot interleave with a QSRV
718 // atomic group or a pvalink atomic scan epoch holding
719 // `lock_records` over the same record. `name` is already the
720 // alias-resolved canonical name, the same key `lock_records`
721 // uses. Not acquired when `acquire_gate` is false: either a
722 // transaction owner already holds the gate via `lock_records`
723 // (`process_record_with_links_already_locked`), or this is a
724 // recursive FLNK/OUT/CP call within one chain
725 // (`process_record_with_links_recursive`) — C `processTarget`
726 // processes a link target under the lock set the caller already
727 // owns, and re-acquiring would deadlock the non-reentrant gate.
728 let _record_gate = if acquire_gate {
729 Some(self.lock_record(name).await)
730 } else {
731 None
732 };
733
734 // 0a. PACT entry guard — mirrors C `dbProcess` (dbAccess.c:537-559).
735 // If the record is currently mid-async (PACT=true), do NOT re-enter
736 // the body. Instead increment LCNT; after MAX_LOCK=10 consecutive
737 // attempts raise SCAN_ALARM/INVALID with "Async in progress" and
738 // post a monitor on VAL (DBE_VALUE|DBE_LOG). Up to MAX_LOCK we just
739 // bail out silently so transient back-to-back scans don't immediately
740 // alarm the record.
741 //
742 // Without this guard, FLNK / scan-loop / event scans dispatched onto
743 // a record whose first cycle is still pending (async device support,
744 // CA put_notify on PUTF) would re-enter `record.process()` while the
745 // device's first response is still in flight — corrupting the
746 // record's internal state machine and bypassing the C-parity
747 // contract that callers see for `dbProcess`. The pre-existing
748 // `dispatch_cp_targets` path already did this check (sets RPRO=true
749 // and skips); the main entry was missing it.
750 if !is_continuation {
751 const MAX_LOCK: i16 = 10;
752 let mut instance = rec.write().await;
753 if instance.is_processing() {
754 // C `dbAccess.c:539-541` — when TPRO is set on a record
755 // whose PACT is true, print the diagnostic line before
756 // the bail decision. The C path emits:
757 // "%s: dbProcess of Active '%s' with RPRO=%d"
758 // mirroring the same context format the regular trace
759 // path below uses (thread/client name + record name +
760 // current RPRO bit). Without this, an operator
761 // debugging a stuck async record sees NO sign that the
762 // entry guard is firing — they only notice the
763 // eventual SCAN_ALARM after MAX_LOCK=10 attempts.
764 if instance.common.tpro {
765 eprintln!(
766 "[TPRO] {}: dbProcess of Active '{}' with RPRO={}",
767 instance.name,
768 instance.name,
769 if instance.common.rpro { 1 } else { 0 },
770 );
771 }
772 let stat = instance.common.stat;
773 let already_invalid =
774 instance.common.sevr >= crate::server::record::AlarmSeverity::Invalid;
775 let already_scan_alarm = stat == crate::server::recgbl::alarm_status::SCAN_ALARM;
776 let lcnt_before = instance.common.lcnt;
777 instance.common.lcnt = lcnt_before.saturating_add(1);
778 if already_scan_alarm || lcnt_before < MAX_LOCK || already_invalid {
779 // Bail out without raising alarm yet.
780 return Ok(());
781 }
782 // Raise SCAN_ALARM/INVALID, reset alarm transition,
783 // and post VAL monitor (DBE_VALUE | DBE_LOG).
784 crate::server::recgbl::rec_gbl_set_sevr_msg(
785 &mut instance.common,
786 crate::server::recgbl::alarm_status::SCAN_ALARM,
787 crate::server::record::AlarmSeverity::Invalid,
788 "Async in progress",
789 );
790 let _ = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
791 // Post VAL event with VALUE | LOG mask (mirrors C
792 // `db_post_events(prec, &VAL, DBE_VALUE|DBE_LOG)`).
793 let mut changed_fields = Vec::new();
794 if let Some(val) = instance.record.val() {
795 changed_fields.push(("VAL".to_string(), val));
796 }
797 changed_fields.push((
798 "SEVR".to_string(),
799 EpicsValue::Short(instance.common.sevr as i16),
800 ));
801 changed_fields.push((
802 "STAT".to_string(),
803 EpicsValue::Short(instance.common.stat as i16),
804 ));
805 // Include AMSG so subscribers reading the alarm text
806 // observe "Async in progress" alongside the SCAN_ALARM
807 // transition (C `recGbl.c:210-211` posts STAT and AMSG
808 // together when `stat_mask` is non-zero).
809 changed_fields.push((
810 "AMSG".to_string(),
811 EpicsValue::String(instance.common.amsg.clone().into()),
812 ));
813 let snapshot = crate::server::record::ProcessSnapshot {
814 changed_fields,
815 event_mask: crate::server::recgbl::EventMask::VALUE
816 | crate::server::recgbl::EventMask::LOG
817 | crate::server::recgbl::EventMask::ALARM,
818 };
819 drop(instance);
820 let inst = rec.read().await;
821 inst.notify_from_snapshot(&snapshot);
822 return Ok(());
823 }
824 // Not pact: reset lcnt (mirrors C `else { precord->lcnt = 0; }`
825 // at dbAccess.c:559) so the next async cycle starts clean.
826 instance.common.lcnt = 0;
827 }
828
829 // 0. SDIS disable check — C parity dbAccess.c:562-592.
830 //
831 // When the SDIS link evaluates to a value equal to DISV, the
832 // record is disabled and bails before record support runs. C
833 // ALWAYS clears rpro/putf and triggers dbNotifyCompletion at
834 // this point — regardless of whether the alarm transition
835 // fires — because a disabled record must not leave behind
836 // pending reprocess requests or stranded put_notify completion
837 // callbacks. Pre-fix the Rust port only reset
838 // nsta/nsev and updated the alarm state, leaking rpro/putf
839 // into the next cycle and stalling CA WRITE_NOTIFY callers
840 // (the put_notify_tx never fired so the CA dispatcher waited
841 // until socket disconnect to release the operation).
842 {
843 let (sdis_link, disv, diss) = {
844 let instance = rec.read().await;
845 (
846 instance.parsed_sdis.clone(),
847 instance.common.disv,
848 instance.common.diss,
849 )
850 };
851
852 // C `dbGetLink(&precord->sdis, DBR_SHORT, &precord->disa, 0, 0)`
853 // reads the SDIS link regardless of its type (DB / CA / PVA /
854 // constant) via the lset. The pre-fix port only refreshed
855 // `disa` from a `ParsedLink::Db` SDIS, so a remote-sourced
856 // (CA/PVA) or constant enable/disable was silently ignored.
857 if let Some(val) = self.read_link_value_no_process(&sdis_link).await {
858 let disa_val = val.to_f64().unwrap_or(0.0) as i16;
859 let mut instance = rec.write().await;
860 instance.common.disa = disa_val;
861 }
862
863 let disa = rec.read().await.common.disa;
864 if disa == disv {
865 let notify = {
866 let mut instance = rec.write().await;
867 // C `dbAccess.c:575-577` — clear rpro/putf and arm
868 // notifyCompletion BEFORE the alarm check. Disabled
869 // records skip processing entirely, so any pending
870 // reprocess request is dropped (the next non-
871 // disabled cycle will pick up fresh state) and the
872 // CA put-notify caller must be released. A disabled
873 // record drives no FLNK/OUT chain, so leaving the
874 // wait-set here is its whole contribution.
875 instance.common.rpro = false;
876 instance.common.putf = false;
877 let notify = instance.notify.take();
878
879 // Reset nsta/nsev so stale alarm state doesn't bleed
880 // into a subsequent (re-enabled) cycle. C resets
881 // them after the sevr/stat transition; doing it
882 // first here is observationally identical because
883 // the SDIS bail short-circuits any record-support
884 // path that could read them.
885 instance.common.nsta = 0;
886 instance.common.nsev = crate::server::record::AlarmSeverity::NoAlarm;
887
888 // C `dbAccess.c:580-581` — if already in
889 // DISABLE_ALARM, the alarm post is skipped entirely
890 // (the alarm cycle is debounced). The rpro/putf
891 // clear above still ran, matching C's pre-`goto
892 // all_done` ordering.
893 if instance.common.stat != crate::server::recgbl::alarm_status::DISABLE_ALARM {
894 use crate::server::recgbl::EventMask;
895 instance.common.sevr = diss;
896 instance.common.stat = crate::server::recgbl::alarm_status::DISABLE_ALARM;
897 // C `dbAccess.c:586-593` posts each field with
898 // its own mask:
899 // db_post_events(&stat, DBE_VALUE);
900 // db_post_events(&sevr, DBE_VALUE);
901 // db_post_events(&val, DBE_VALUE|DBE_ALARM);
902 // STAT/SEVR get DBE_VALUE only — a DBE_ALARM-only
903 // subscriber on `.STAT`/`.SEVR` must NOT receive
904 // this disable event. Only the value field
905 // carries DBE_ALARM.
906 instance.notify_field("STAT", EventMask::VALUE);
907 instance.notify_field("SEVR", EventMask::VALUE);
908 instance.notify_field("VAL", EventMask::VALUE | EventMask::ALARM);
909 }
910 notify
911 };
912 // Fire dbNotifyCompletion outside the record lock —
913 // C `dbAccess.c:622-623` runs it at `all_done` after
914 // the disable bail. Without this, a CA WRITE_NOTIFY
915 // landing on a disabled record stalls until socket
916 // disconnect. `leave` fires the completion oneshot when
917 // this empties the wait-set.
918 if let Some(ws) = notify {
919 ws.leave();
920 }
921 return Ok(());
922 }
923 }
924
925 // 0.3. TSEL link: C `recGblGetTimeStampSimm` (recGbl.c:310-323).
926 //
927 // When `TSEL` is a non-constant link, C distinguishes two
928 // cases by the link target field:
929 // * the link points at another record's `.TIME` field
930 // (`DBLINK_FLAG_TSELisTIME`) — copy that record's
931 // timestamp directly into `prec->time`;
932 // * otherwise `dbGetLink(&tsel, DBR_SHORT, &prec->tse)` —
933 // load `TSE` from the link before the event lookup.
934 {
935 let tsel_link = {
936 let instance = rec.read().await;
937 instance.parsed_tsel.clone()
938 };
939 // A TSEL link pointing at a `.TIME` field copies that record's
940 // timestamp+utag into `time`/`utag` and marks TSE=-2 so
941 // `apply_timestamp` leaves them alone. C `TSEL_modified`
942 // (dbLink.c:71-87) sets `DBLINK_FLAG_TSELisTIME` for ANY
943 // `PV_LINK` tsel whose pvname contains `.TIME`, set BEFORE the
944 // DB-vs-CA decision (dbLink.c:118) — so a local-DB link AND a
945 // CA link both qualify. `recGblGetTimeStampSimm`
946 // (recGbl.c:316-321) then copies the link's time+utag via
947 // `dbGetTimeStampTag` and RETURNS, never loading TSE from the
948 // value (even when the read fails). A pva link is a
949 // `JSON_LINK` and returns early from `dbInitLink`
950 // (dbLink.c:107) before `TSEL_modified`, so C never flags it;
951 // pva TSEL `.TIME` is intentionally excluded here.
952 let tsel_is_time = match &tsel_link {
953 crate::server::record::ParsedLink::Db(link) => {
954 link.field.eq_ignore_ascii_case("TIME")
955 }
956 crate::server::record::ParsedLink::Ca(ca) => ca_tsel_time_record(&ca.pv).is_some(),
957 _ => false,
958 };
959 if tsel_is_time {
960 // C `dbGetTimeStampTag(plink, &prec->time, &prec->utag)`
961 // (recGbl.c:317) copies BOTH the link's time AND utag.
962 // Read the pair as one consistent snapshot per source.
963 let src_time = match &tsel_link {
964 crate::server::record::ParsedLink::Db(link) => {
965 // C `dbInitLink` locality (`dbLink.c:115-130`):
966 // `TSEL_modified` sets the `TSELisTIME` flag and
967 // strips `.TIME` BEFORE the DB-vs-CA decision
968 // (dbLink.c:115-118), so a TSEL `.TIME` link whose
969 // record is not local still becomes a CA link and
970 // reads its remote `.TIME` via the CA lset
971 // `getTimeStampTag`. Local arm reads the source
972 // record's `(time, utag)`; the non-local arm routes
973 // `ca://REC` through `external_link_time` (CA
974 // carries no userTag, so utag is 0) — uniform with
975 // the `Ca` arm below and the `read_db_link_value`
976 // read-locality fallback.
977 if self.has_name_no_resolve(&link.record).await {
978 match self.get_record(&link.record).await {
979 Some(src) => {
980 let g = src.read().await;
981 Some((g.common.time, g.common.utag))
982 }
983 None => None,
984 }
985 } else {
986 self.external_link_time(&format!("ca://{}", link.record))
987 .await
988 .map(ext_time_pair)
989 }
990 }
991 crate::server::record::ParsedLink::Ca(ca) => {
992 // Strip `.TIME` (C dbLink.c:82-84) and read the CA
993 // link's cached timestamp. `external_link_time`
994 // routes `ca://` to the ungated CA lset
995 // `time_stamp` (CA has no `time=` option; gated
996 // only on `connected`, like C `dbGetTimeStamp`
997 // failing on a disconnected link). CA wire carries
998 // no userTag, so the source contributes utag 0.
999 match ca_tsel_time_record(&ca.pv) {
1000 Some(rec_name) => self
1001 .external_link_time(&format!("ca://{rec_name}"))
1002 .await
1003 .map(ext_time_pair),
1004 None => None,
1005 }
1006 }
1007 _ => None,
1008 };
1009 // C returns after the TSELisTIME branch even when the read
1010 // fails (recGbl.c:317-320): keep the record's current time
1011 // rather than falling through to load TSE from the value.
1012 if let Some((src_time, src_utag)) = src_time {
1013 let mut instance = rec.write().await;
1014 instance.common.time = src_time;
1015 instance.common.utag = src_utag;
1016 instance.common.tse = -2;
1017 }
1018 } else if let Some(val) = self.read_link_value_no_process(&tsel_link).await {
1019 // Non-`.TIME` TSEL: C `dbGetLink(&tsel, DBR_SHORT,
1020 // &prec->tse)` loads TSE from the link regardless of its
1021 // type. The pre-fix port only read a `ParsedLink::Db`
1022 // TSEL, ignoring a CA/PVA/constant TSE source.
1023 let tse_val = val.to_f64().unwrap_or(0.0) as i16;
1024 let mut instance = rec.write().await;
1025 instance.common.tse = tse_val;
1026 }
1027 }
1028
1029 // 0.5. Simulation mode check.
1030 //
1031 // C `aiRecord.c:151-168`: simulation is handled inside
1032 // `readValue()`, then `process()` ALWAYS runs `convert` /
1033 // `checkAlarms` / `monitor` / `recGblFwdLink(prec)`. A
1034 // simulated record therefore must still run the forward-link /
1035 // CP / RPRO tail — only the device read and record-support
1036 // body are replaced by the SIOL round-trip. Returning early
1037 // here would silently break every FLNK / CP chain downstream
1038 // of any record in SIMM mode.
1039 match self.check_simulation_mode(&rec).await {
1040 SimOutcome::NotSimulated => {}
1041 SimOutcome::Simulated => {
1042 self.run_forward_link_tail(name, &rec, visited, depth).await;
1043 return Ok(());
1044 }
1045 }
1046
1047 // 1. Read INP link value and DOL link (outside lock)
1048 let (inp_parsed, is_soft, dol_info) = {
1049 let instance = rec.read().await;
1050 let rtype = instance.record.record_type();
1051
1052 let inp = instance.parsed_inp.clone();
1053 let is_soft = crate::server::device_support::is_soft_dtyp(&instance.common.dtyp);
1054
1055 // DOL link info for output records with OMSL=CLOSED_LOOP.
1056 //
1057 // C parity: every record type whose DBD declares both an
1058 // OMSL `menuOmsl` field AND a DOL link field must honour
1059 // the closed-loop binding. `dfanoutRecord.c:115-122` shows
1060 // dfanout doing this directly via `dbGetLink(&prec->dol,
1061 // DBR_DOUBLE, &prec->val, ...)` when `omsl ==
1062 // menuOmslclosed_loop`. The Rust port previously omitted
1063 // `dfanout`, so a dfanout configured with OMSL=closed_loop
1064 // never sourced VAL from DOL — every cycle silently used
1065 // the previously-cached VAL, breaking any cascaded
1066 // setpoint-distribution chain that relied on dfanout to
1067 // re-read the input.
1068 //
1069 // The `aao` (array analog output) record is the only other
1070 // OMSL-bearing C record; the Rust port does not implement
1071 // aao (confirmed: no `crates/epics-base-rs/src/server/records/aao*.rs`),
1072 // so it is a future gap, not a same-defect-not-fixed site.
1073 let dol = match rtype {
1074 "ao" | "longout" | "int64out" | "bo" | "mbbo" | "mbboDirect" | "stringout"
1075 | "lso" | "dfanout" => {
1076 let omsl = instance
1077 .record
1078 .get_field("OMSL")
1079 .and_then(|v| {
1080 if let EpicsValue::Short(s) = v {
1081 Some(s)
1082 } else {
1083 None
1084 }
1085 })
1086 .unwrap_or(0);
1087 let oif = instance
1088 .record
1089 .get_field("OIF")
1090 .and_then(|v| {
1091 if let EpicsValue::Short(s) = v {
1092 Some(s)
1093 } else {
1094 None
1095 }
1096 })
1097 .unwrap_or(0);
1098 if omsl == 1 {
1099 let dol_parsed = instance
1100 .record
1101 .get_field("DOL")
1102 .and_then(|v| {
1103 if let EpicsValue::String(s) = v {
1104 Some(s)
1105 } else {
1106 None
1107 }
1108 })
1109 .map(|s| {
1110 crate::server::record::parse_link_v2(s.as_str_lossy().as_ref())
1111 })
1112 .unwrap_or(crate::server::record::ParsedLink::None);
1113 Some((dol_parsed, oif))
1114 } else {
1115 None
1116 }
1117 }
1118 _ => None,
1119 };
1120
1121 (inp, is_soft, dol)
1122 };
1123
1124 // 1.1. Pre-input-link actions: actions a record needs the
1125 // framework to execute BEFORE any input-link fetch this cycle.
1126 //
1127 // C `devEpidSoftCallback.c:120-151`: a DB-type readback-trigger
1128 // (TRIG) link is written with `dbPutLink` — which synchronously
1129 // processes the triggered source — and only then does
1130 // `dbGetLink(&pepid->inp, ...)` read CVAL. The trigger write
1131 // must land before the `INP -> CVAL` fetch, in the same pass.
1132 // `pre_process_actions` runs too late (after the input-link
1133 // fetch below), so `pre_input_link_actions` is a strictly
1134 // earlier hook. The record needs `dtyp` to decide whether the
1135 // callback DSET is active, so push the process context first.
1136 {
1137 let pre_input_actions = {
1138 let mut instance = rec.write().await;
1139 let ctx = instance.common.process_context();
1140 instance.record.set_process_context(&ctx);
1141 instance.record.pre_input_link_actions()
1142 };
1143 if !pre_input_actions.is_empty() {
1144 self.execute_process_actions(name, &rec, pre_input_actions, visited, depth)
1145 .await;
1146 }
1147 }
1148
1149 // Read INP value
1150 let inp_value = self
1151 .read_link_value_soft(&inp_parsed, is_soft, visited, depth)
1152 .await;
1153
1154 // epics-base PR #d0cf47c: single-INP MS-class link must also
1155 // propagate the source record's STAT/SEVR/AMSG just like the
1156 // multi-input fetch loop below does. Previously the INPA..L
1157 // path (calc/sub/aSub/sel) propagated alarms but plain single
1158 // INP (ai/bi/longin/mbbi/stringin) silently dropped them —
1159 // downstream MSS readers saw NoAlarm even when the source was
1160 // INVALID. Only fires for soft-channel records: hardware-driver
1161 // alarms travel through device-support's own last_alarm path.
1162 //
1163 // B2: a soft INP that is an external `pva://` / `ca://` link
1164 // also propagates the lset's alarm. The link string carries
1165 // no `MonitorSwitch` (the `?sevr=MS` modifier is stripped by
1166 // the parser before epics-base-rs sees it), so the lset has
1167 // already applied the MS/NMS/MSI gate — a `Some` LinkAlarm
1168 // here is one the lset decided to propagate. We fold it in as
1169 // `MaximizeStatus` so the gated severity AND message both
1170 // reach `LINK_ALARM`, matching pvxs `pvalink_lset.cpp`
1171 // `recGblSetSevrMsg`.
1172 let inp_link_alarm: Option<(
1173 crate::server::record::MonitorSwitch,
1174 super::links::LinkAlarm,
1175 )> = if is_soft {
1176 match inp_parsed {
1177 crate::server::record::ParsedLink::Db(ref db) => {
1178 let (_v, alarm) = self.read_link_with_alarm(&inp_parsed).await;
1179 alarm.map(|a| (db.monitor_switch, a))
1180 }
1181 crate::server::record::ParsedLink::Pva(_)
1182 | crate::server::record::ParsedLink::PvaJson(_) => {
1183 // PVA: the lset already applied the MS/NMS/MSI gate,
1184 // so the returned severity is final — fold it as
1185 // MaximizeStatus to preserve the remote stat+msg
1186 // (pvxs `pvalink_lset.cpp`).
1187 let (_v, alarm) = self.read_link_with_alarm(&inp_parsed).await;
1188 alarm.map(|a| (crate::server::record::MonitorSwitch::MaximizeStatus, a))
1189 }
1190 crate::server::record::ParsedLink::Ca(ref ca) => {
1191 // CA: apply the link's own
1192 // MS/NMS/MSI/MSS gate at the fold boundary, uniform
1193 // with the Db arm above — the resolver returned the
1194 // *raw* remote alarm, not a gated one.
1195 let (_v, alarm) = self.read_link_with_alarm(&inp_parsed).await;
1196 alarm.map(|a| (ca.monitor_switch, a))
1197 }
1198 _ => None,
1199 }
1200 } else {
1201 None
1202 };
1203
1204 // if the single-INP link is an external `pva://` /
1205 // `ca://` link configured with `time=true`, the lset returns
1206 // the latched upstream NT timestamp here and we adopt it
1207 // into the owning record's `common.time` and `common.utag`. The
1208 // lset gates the option internally (returns `None` unless
1209 // `time=true`), so a bare connected link without the flag still
1210 // produces local processing time. Mirrors pvxs
1211 // `pvalink_lset.cpp:427`.
1212 let inp_link_remote_time: Option<(i64, i32, u64)> = match inp_parsed.external_pv_name() {
1213 Some(name) => self.external_link_time(&name).await,
1214 None => None,
1215 };
1216
1217 // Read DOL value
1218 let dol_value = if let Some((ref dol_parsed, _oif)) = dol_info {
1219 self.read_link_value(dol_parsed, visited, depth).await
1220 } else {
1221 None
1222 };
1223
1224 // 1.5. Multi-input link fetch (calc/calcout/sel/sub)
1225 // Also collect alarm info from source records for MS/NMS propagation.
1226 let multi_input_values: Vec<(String, EpicsValue)>;
1227 let mut link_alarms: Vec<(
1228 crate::server::record::MonitorSwitch,
1229 super::links::LinkAlarm,
1230 )> = Vec::new();
1231 // Link fields (the `multi_input_links` first element) whose
1232 // fetch actually produced a value this cycle — pushed to the
1233 // record via `set_resolved_input_links` so its `process()` can
1234 // observe link-fetch success (C `RTN_SUCCESS(dbGetLink(...))`).
1235 let mut resolved_link_fields: Vec<&'static str> = Vec::new();
1236 {
1237 let link_info: Vec<(String, &'static str, String)> = {
1238 let instance = rec.read().await;
1239 instance
1240 .record
1241 .multi_input_links()
1242 .iter()
1243 .map(|(lf, vf)| {
1244 let link_str = instance
1245 .record
1246 .get_field(lf)
1247 .and_then(|v| {
1248 if let EpicsValue::String(s) = v {
1249 Some(s)
1250 } else {
1251 None
1252 }
1253 })
1254 .unwrap_or_default();
1255 (link_str.as_str_lossy().into_owned(), *lf, vf.to_string())
1256 })
1257 .collect()
1258 }; // read lock dropped
1259 let mut results = Vec::new();
1260 for (link_str, link_field, val_field) in &link_info {
1261 if !link_str.is_empty() {
1262 let parsed = crate::server::record::parse_link_v2(link_str);
1263 // C `dbGetLink`: a `ProcessPassive` DB input link
1264 // processes its passive source record before the
1265 // value is read. `read_link_with_alarm` does a bare
1266 // `get_pv`, so process the source here first —
1267 // matching the single-INP `read_link_value_soft`
1268 // path. Without this, calc/sel/sub/aSub INPA..INPL
1269 // PP links read a stale source value.
1270 if let crate::server::record::ParsedLink::Db(ref db) = parsed {
1271 self.process_passive_db_source(db, visited, depth).await;
1272 }
1273 let (value, alarm) = self.read_link_with_alarm(&parsed).await;
1274 if let Some(value) = value {
1275 results.push((val_field.clone(), value));
1276 resolved_link_fields.push(link_field);
1277 }
1278 // B2 / multi-input alarm propagation
1279 // covers external links too. `Db` and `Ca` carry an
1280 // explicit `MonitorSwitch` (CA's was parsed from its
1281 // `MS`/`NMS`/`MSI`/`MSS` modifier); `Pva` is gated by
1282 // its lset, so its already-final severity folds as
1283 // `MaximizeStatus` (preserving remote stat+msg).
1284 if let Some(alarm) = alarm {
1285 match &parsed {
1286 crate::server::record::ParsedLink::Db(db) => {
1287 link_alarms.push((db.monitor_switch, alarm));
1288 }
1289 crate::server::record::ParsedLink::Ca(ca) => {
1290 link_alarms.push((ca.monitor_switch, alarm));
1291 }
1292 crate::server::record::ParsedLink::Pva(_)
1293 | crate::server::record::ParsedLink::PvaJson(_) => {
1294 link_alarms.push((
1295 crate::server::record::MonitorSwitch::MaximizeStatus,
1296 alarm,
1297 ));
1298 }
1299 _ => {}
1300 }
1301 }
1302 }
1303 }
1304 multi_input_values = results;
1305 }
1306 // PR #d0cf47c continued: feed the INP alarm (if any) into the
1307 // same `link_alarms` list the lock-section iterates over. Order
1308 // doesn't matter — `rec_gbl_set_sevr_msg` takes the maximum
1309 // severity across all sources.
1310 if let Some(pair) = inp_link_alarm {
1311 link_alarms.push(pair);
1312 }
1313
1314 // 1.6. Sel NVL link: resolve NVL -> SELN
1315 let sel_nvl_value: Option<EpicsValue> = {
1316 let instance = rec.read().await;
1317 if instance.record.record_type() == "sel" {
1318 let nvl_str = instance
1319 .record
1320 .get_field("NVL")
1321 .and_then(|v| {
1322 if let EpicsValue::String(s) = v {
1323 Some(s)
1324 } else {
1325 None
1326 }
1327 })
1328 .unwrap_or_default();
1329 if !nvl_str.is_empty() {
1330 drop(instance); // release read lock before async read
1331 let parsed =
1332 crate::server::record::parse_link_v2(nvl_str.as_str_lossy().as_ref());
1333 self.read_link_value(&parsed, visited, depth).await
1334 } else {
1335 None
1336 }
1337 } else {
1338 None
1339 }
1340 };
1341
1342 // 2. Lock record, apply INP/DOL, process, evaluate alarms, build snapshot
1343 let (snapshot, out_info, flnk_name, process_actions, alarm_posts) = {
1344 let mut instance = rec.write().await;
1345
1346 // Apply DOL value for output records (OMSL=CLOSED_LOOP)
1347 if let Some(dol_val) = dol_value {
1348 let oif = dol_info.as_ref().map(|(_, oif)| *oif).unwrap_or(0);
1349 if oif == 1 {
1350 // Incremental: VAL += DOL value
1351 if let (Some(cur), Some(dol_f)) = (
1352 instance.record.val().and_then(|v| v.to_f64()),
1353 dol_val.to_f64(),
1354 ) {
1355 let _ = instance.record.set_val(EpicsValue::Double(cur + dol_f));
1356 }
1357 } else {
1358 // Full: VAL = DOL value
1359 let _ = instance.record.set_val(dol_val);
1360 }
1361 }
1362
1363 // Apply INP value. "Soft Channel" sets VAL directly
1364 // (C `read_xxx` return 2, skip RVAL→VAL conversion).
1365 // "Raw Soft Channel" routes the value into RVAL and lets
1366 // the record's RVAL→VAL convert run (epics-base
1367 // f2fe9d12: devBiSoftRaw applies MASK after the read).
1368 // Records opt into the raw path via
1369 // `Record::accepts_raw_soft_input` so DTYPs on records
1370 // that haven't wired raw soft channel stay on the legacy
1371 // VAL-direct path.
1372 let is_raw_soft = instance.common.dtyp == "Raw Soft Channel"
1373 && instance.record.accepts_raw_soft_input();
1374 let soft_inp_applied = inp_value.is_some() && !is_raw_soft;
1375 if let Some(inp_val) = inp_value {
1376 if is_raw_soft {
1377 let _ = instance.record.apply_raw_input(inp_val);
1378 } else {
1379 let _ = instance.record.set_val(inp_val);
1380 }
1381 } else if is_soft
1382 && matches!(
1383 inp_parsed,
1384 crate::server::record::ParsedLink::Db(_)
1385 | crate::server::record::ParsedLink::Ca(_)
1386 | crate::server::record::ParsedLink::Pva(_)
1387 | crate::server::record::ParsedLink::PvaJson(_)
1388 )
1389 {
1390 // epics-base PR #4737901: soft-channel `read_xxx` must
1391 // surface link-read failures via the alarm tree, not
1392 // silently succeed. When the INP link is a real
1393 // Db/Ca/Pva link (i.e. operator expected a value) and
1394 // the read returned None, attach LINK_ALARM/INVALID
1395 // so downstream consumers can react. ParsedLink::None
1396 // and Constant don't fall into this branch — the
1397 // former is "no link configured", the latter has its
1398 // own None-as-no-value semantics.
1399 use crate::server::recgbl::{alarm_status, rec_gbl_set_sevr};
1400 rec_gbl_set_sevr(
1401 &mut instance.common,
1402 alarm_status::LINK_ALARM,
1403 crate::server::record::AlarmSeverity::Invalid,
1404 );
1405 }
1406
1407 // Apply multi-input values (INPA..INPL -> A..L).
1408 //
1409 // Uses `put_field_internal`, not `put_field`: this is the
1410 // framework writing a resolved input-link value into a
1411 // record field, exactly like the `ReadDbLink` apply
1412 // (`execute_read_db_links` / `execute_process_actions`),
1413 // which already routes through `put_field_internal`. Some
1414 // records map an input link to a normally read-only field
1415 // — e.g. the epid record's `INP -> CVAL` — and `put_field`
1416 // rejects those with `ReadOnlyField`, silently dropping the
1417 // value. `put_field_internal` defaults to `put_field`, so
1418 // records with writable targets (calc/sub `A..L`) are
1419 // unaffected.
1420 for (val_field, value) in &multi_input_values {
1421 if let Some(f) = value.to_f64() {
1422 let _ = instance
1423 .record
1424 .put_field_internal(val_field, EpicsValue::Double(f));
1425 }
1426 }
1427
1428 // Tell the record which input link fields actually resolved
1429 // a value this cycle — the framework analogue of C device
1430 // support inspecting `RTN_SUCCESS(dbGetLink(...))`
1431 // (`epidRecord.c:191-193`).
1432 instance
1433 .record
1434 .set_resolved_input_links(&resolved_link_fields);
1435
1436 // Apply sel NVL -> SELN. SELN is DBF_USHORT (selRecord.dbd.pod:295),
1437 // an unsigned 0..65535 index. Carry the native unsigned value so a
1438 // link value in 32768..65535 is not lost to f64->i16 saturation
1439 // before it reaches the field's put.
1440 if let Some(nvl_val) = sel_nvl_value {
1441 if let Some(f) = nvl_val.to_f64() {
1442 let _ = instance
1443 .record
1444 .put_field("SELN", EpicsValue::UShort(f as u16));
1445 }
1446 }
1447
1448 // Device support read (input records only, not output records)
1449 let is_soft = instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
1450 let is_output = instance.record.can_device_write();
1451 let mut device_actions: Vec<crate::server::record::ProcessAction> = Vec::new();
1452 // C `devAiSoft.c:65` `read_ai` (and the other soft-channel
1453 // input `read_xxx`) ALWAYS returns 2 ("don't convert") for a
1454 // Soft-Channel input record — whether the value arrived via
1455 // an INP link or the INP link is constant/unset
1456 // (`dbLinkIsConstant` → `return 2`). Only `aiRecord.c:158`'s
1457 // `if (status==0) convert(prec)` runs RVAL→VAL conversion, so
1458 // for a plain Soft-Channel input record `convert()` must be
1459 // skipped unconditionally. Without this, a soft ai with no
1460 // INP would run `convert()` and clobber a preset VAL — e.g.
1461 // a preset NaN would be rewritten to 0.0, then the framework
1462 // UDF check (`value_is_undefined()`) would see a defined 0.0
1463 // and wrongly clear UDF. `is_raw_soft`
1464 // (Raw Soft Channel, `devAiSoftRaw` returns 0) is excluded —
1465 // it deliberately wants the RVAL→VAL convert.
1466 //
1467 // Gated on `soft_channel_skips_convert()` so this only
1468 // suppresses an `RVAL → VAL` convert step. Records such as
1469 // `epid` also override `set_device_did_compute` but treat it
1470 // as "skip the whole built-in compute" (the PID loop); they
1471 // return `false` here so a Soft-Channel `epid` still runs
1472 // `do_pid()` in `process()`.
1473 let soft_input_skips_convert = is_soft
1474 && !is_output
1475 && !is_raw_soft
1476 && instance.record.soft_channel_skips_convert();
1477 let mut device_did_compute = (soft_inp_applied && is_soft) || soft_input_skips_convert;
1478 if !is_soft && !is_output {
1479 if let Some(mut dev) = instance.device.take() {
1480 // Push framework-owned common state (PHAS/TSE/TSEL/
1481 // UDF) so device support's read() can see it — C
1482 // device support reads `dbCommon` directly
1483 // (`devTimeOfDay.c:122` uses `psi->phas`).
1484 dev.set_process_context(&instance.common.process_context());
1485 match dev.read(&mut *instance.record) {
1486 Ok(read_outcome) => {
1487 device_did_compute = read_outcome.did_compute;
1488 device_actions = read_outcome.actions;
1489 }
1490 Err(e) => {
1491 eprintln!("device read error on {}: {e}", instance.name);
1492 use crate::server::recgbl::{alarm_status, rec_gbl_set_sevr};
1493 rec_gbl_set_sevr(
1494 &mut instance.common,
1495 alarm_status::READ_ALARM,
1496 crate::server::record::AlarmSeverity::Invalid,
1497 );
1498 }
1499 }
1500 instance.device = Some(dev);
1501 }
1502 }
1503
1504 // Pre-process actions: execute ReadDbLink from device support and
1505 // record's pre_process_actions() BEFORE process() so the values
1506 // are immediately available. Matches C dbGetLink() semantics.
1507 let mut pre_actions = instance.record.pre_process_actions();
1508 // Also collect ReadDbLink from device actions
1509 let mut deferred_device_actions = Vec::new();
1510 for action in device_actions {
1511 if matches!(
1512 action,
1513 crate::server::record::ProcessAction::ReadDbLink { .. }
1514 ) {
1515 pre_actions.push(action);
1516 } else {
1517 deferred_device_actions.push(action);
1518 }
1519 }
1520 if !pre_actions.is_empty() {
1521 let rec_name = instance.name.clone();
1522 drop(instance);
1523 self.execute_read_db_links(&rec_name, &rec, &pre_actions, visited, depth)
1524 .await;
1525 instance = rec.write().await;
1526 }
1527
1528 // Note: C EPICS LCNT prevents reentrant processing of the same
1529 // record within a single processing chain. In Rust, this is handled
1530 // by the `visited` HashSet (cycle detection) and the `processing`
1531 // AtomicBool guard. LCNT is not needed as a separate mechanism
1532 // because async processing with visited sets already prevents
1533 // the runaway loops that LCNT guards against in C.
1534
1535 // Tell the record whether device support already computed.
1536 // Records that override set_device_did_compute() use this to
1537 // skip their built-in computation (e.g., ai skips RVAL->VAL).
1538 // Note: field_io.rs may have already called set_device_did_compute(true)
1539 // for CA puts to VAL. We only set true here, never reset to false.
1540 if device_did_compute {
1541 instance.record.set_device_did_compute(true);
1542 }
1543
1544 // TPRO: trace processing (C EPICS dbProcess prints context when TPRO>0)
1545 if instance.common.tpro {
1546 eprintln!(
1547 "[TPRO] {}: process (SCAN={:?}, PACT={})",
1548 instance.name,
1549 instance.common.scan,
1550 instance
1551 .processing
1552 .load(std::sync::atomic::Ordering::Relaxed)
1553 );
1554 }
1555
1556 // Push framework-owned common state (UDF/PHAS/TSE/TSEL) so
1557 // the record's process() can see it — C records read
1558 // `dbCommon` directly (`epidRecord.c:195` checks
1559 // `pepid->udf`, `timestampRecord.c:90` checks `tse`).
1560 {
1561 let ctx = instance.common.process_context();
1562 instance.record.set_process_context(&ctx);
1563 }
1564
1565 // Process
1566 let mut outcome = instance.record.process()?;
1567 // Merge deferred device actions into process outcome actions
1568 outcome.actions.extend(deferred_device_actions);
1569 let process_result = outcome.result;
1570 let process_actions = outcome.actions;
1571
1572 if process_result == crate::server::record::RecordProcessResult::AsyncPending {
1573 // C `dbProcess` contract: when device support / record body
1574 // signals "async pending", `pact` MUST be true so subsequent
1575 // dbProcess attempts on the same record bail at the entry
1576 // guard. Previous Rust port assumed `process_local` had
1577 // already set it via the swap-true at function entry, but
1578 // this main path bypasses `process_local` and calls
1579 // `record.process()` directly — leaving `processing=false`.
1580 // Mirrors `aiRecord.c:122` and similar: `prec->pact = TRUE;
1581 // return 0;` before async work.
1582 instance
1583 .processing
1584 .store(true, std::sync::atomic::Ordering::Release);
1585
1586 // PACT stays set; skip alarm/timestamp/snapshot/OUT/FLNK.
1587 // But still execute any actions (e.g., ReprocessAfter for delayed re-entry).
1588 let rec_name = instance.name.clone();
1589 drop(instance);
1590 self.execute_process_actions(&rec_name, &rec, process_actions, visited, depth)
1591 .await;
1592 return Ok(());
1593 }
1594 if let crate::server::record::RecordProcessResult::AsyncPendingNotify(fields) =
1595 process_result
1596 {
1597 // Intermediate notification (e.g. DMOV=0 at move start).
1598 // Execute device write first so the move command reaches the
1599 // driver, then fire the record's link writes, then flush
1600 // DMOV=0 etc. to monitors. This mirrors the C ordering on an
1601 // async (pact=1) pass: `motorRecord.cc:1491` runs `do_work`
1602 // (the device move), `motorRecord.cc:1495` then fires
1603 // `dbPutLink(&pmr->rlnk, ...)` UNCONDITIONALLY — on every pass
1604 // including the move-start pass where DMOV just went 0 — and
1605 // only `motorRecord.cc:1507` afterwards calls `monitor()`. So
1606 // the requested `WriteDbLink`/`WriteDbLinkNotify` actions must
1607 // run on the pending cycle as well; a put processes a PP target
1608 // even when the value is unchanged, so dropping them changes
1609 // downstream process counts (motor RLNK, asyn async writes).
1610 // The forward link stays deferred: C runs `recGblFwdLink` only
1611 // when `pmr->dmov != 0` (motorRecord.cc:1509), i.e. on async
1612 // completion, not on this pending pass.
1613 if !is_soft {
1614 if let Some(mut dev) = instance.device.take() {
1615 let _ = dev.write(&mut *instance.record);
1616 instance.device = Some(dev);
1617 }
1618 }
1619 apply_timestamp(&mut instance.common, is_soft);
1620 // Filter out fields that haven't changed, update MLST/last_posted.
1621 let mut changed_fields = Vec::new();
1622 for (name, val) in fields {
1623 let changed = match instance.last_posted.get(&name) {
1624 Some(prev) => prev != &val,
1625 None => true,
1626 };
1627 if changed {
1628 if name == "VAL" {
1629 if let Some(f) = val.to_f64() {
1630 instance.put_coerced("MLST", f);
1631 instance.common.mlst = Some(f);
1632 }
1633 }
1634 instance.last_posted.insert(name.clone(), val.clone());
1635 changed_fields.push((name, val));
1636 }
1637 }
1638 let event_mask = if changed_fields.is_empty() {
1639 crate::server::recgbl::EventMask::NONE
1640 } else {
1641 crate::server::recgbl::EventMask::VALUE
1642 | crate::server::recgbl::EventMask::ALARM
1643 };
1644 let snapshot = crate::server::record::ProcessSnapshot {
1645 changed_fields,
1646 event_mask,
1647 };
1648 let rec_name = instance.name.clone();
1649 let rec_clone = rec.clone();
1650 drop(instance);
1651 // Partition exactly as the synchronous Complete path: link
1652 // writes fire here (C `dbPutLink` precedes `monitor()`);
1653 // delayed-reprocess / device-command actions run after the
1654 // notify (the Complete path runs them after the FLNK tail,
1655 // which is deferred to async completion on this pending pass).
1656 let (link_writes, deferred_actions): (Vec<_>, Vec<_>) =
1657 process_actions.into_iter().partition(|a| {
1658 matches!(
1659 a,
1660 crate::server::record::ProcessAction::WriteDbLink { .. }
1661 | crate::server::record::ProcessAction::WriteDbLinkNotify { .. }
1662 )
1663 });
1664 self.execute_process_actions(&rec_name, &rec, link_writes, visited, depth)
1665 .await;
1666 {
1667 let inst = rec_clone.read().await;
1668 inst.notify_from_snapshot(&snapshot);
1669 }
1670 self.execute_process_actions(&rec_name, &rec, deferred_actions, visited, depth)
1671 .await;
1672 return Ok(());
1673 }
1674
1675 // Async-completion PACT clear for the `ReprocessAfter`
1676 // continuation path. C parity `dbAccess.c:583` —
1677 // `prset->process(precord)` for a record whose first cycle
1678 // returned async-pending is the *completion* re-entry; the
1679 // record support clears `pact` itself inside `process()`
1680 // (e.g. `aiRecord.c` second pass sets `prec->pact = FALSE`).
1681 //
1682 // A record that returns `AsyncPending` AND emits a
1683 // `ProcessAction::ReprocessAfter` is re-entered here via
1684 // `process_record_continuation` (`is_continuation == true`,
1685 // PACT entry guard skipped). Reaching this point means the
1686 // continuation's `process()` did NOT return async-pending
1687 // again (both async branches above return early), so the
1688 // async cycle is genuinely complete. The non-continuation
1689 // async-device path clears `processing` in
1690 // `complete_async_record_inner`; the continuation path has
1691 // no such callback, so without this clear `processing`
1692 // stays `true` forever — every later foreign
1693 // `process_record_with_links` then trips the PACT entry
1694 // guard, counts to MAX_LOCK, and raises a spurious
1695 // SCAN_ALARM. Clearing here (record still write-locked,
1696 // before the OUT/FLNK tail) mirrors the C ordering where
1697 // `pact` is already `FALSE` when `recGblFwdLink` runs.
1698 if is_continuation {
1699 instance
1700 .processing
1701 .store(false, std::sync::atomic::Ordering::Release);
1702 }
1703
1704 // MS-class alarm propagation from input links. Mirrors C
1705 // `recGblInheritSevrMsg` (recGbl.c::260):
1706 //
1707 // * NMS — do nothing.
1708 // * MS — DEST gets `LINK_ALARM` (NOT the source stat),
1709 // max-raised sevr, NO amsg propagation.
1710 // * MSI — same as MS, but only when source.sevr == INVALID.
1711 // * MSS — DEST gets source stat, max-raised sevr, source amsg
1712 // (PR d0cf47c is the only branch that propagates msg).
1713 //
1714 // Previous version treated Maximize and MaximizeStatus
1715 // identically, propagating source stat + amsg through both
1716 // — that matches MSS but is wrong for MS (and MSI), which
1717 // C says should always surface as LINK_ALARM with no msg.
1718 // The per-mode switch is shared with the DB OUT-link write
1719 // path via `inherit_sevr_msg` so the two sides cannot drift.
1720 for (ms, alarm) in &link_alarms {
1721 super::links::inherit_sevr_msg(&mut instance.common, *ms, alarm);
1722 }
1723
1724 // UDF update — C parity (aiRecord.c:285, calcRecord.c
1725 // checkAlarms, int64inRecord.c:144): clear UDF only when
1726 // this cycle produced a *defined* value. A NaN computed
1727 // value (calc divide-by-zero) or a failed link read that
1728 // left VAL un-updated must keep UDF true so the following
1729 // `recGblCheckUDF` raises UDF_ALARM at severity UDFS.
1730 //
1731 // This MUST run before `evaluate_alarms()` (which calls
1732 // `rec_gbl_check_udf`): C records set `prec->udf` inside
1733 // `process()` before `checkAlarms()` runs.
1734 if instance.record.clears_udf() {
1735 instance.common.udf = instance.record.value_is_undefined();
1736 }
1737
1738 // Per-record alarm hook — record-type-specific STATE / COS
1739 // / limit / SOFT alarms (C `checkAlarms()`). Records that
1740 // have migrated their alarm logic here raise into
1741 // `nsta`/`nsev`; the rest fall back to the framework's
1742 // centralised `evaluate_alarms` match below.
1743 {
1744 let inst = &mut *instance;
1745 inst.record.check_alarms(&mut inst.common);
1746 }
1747
1748 // Evaluate alarms (accumulates into nsta/nsev)
1749 instance.evaluate_alarms();
1750
1751 // Device support alarm/timestamp override
1752 if !is_soft {
1753 let (dev_alarm, dev_ts, dev_utag) = if let Some(ref dev) = instance.device {
1754 (dev.last_alarm(), dev.last_timestamp(), dev.last_utag())
1755 } else {
1756 (None, None, None)
1757 };
1758 if let Some((stat, sevr)) = dev_alarm {
1759 use crate::server::recgbl::rec_gbl_set_sevr;
1760 rec_gbl_set_sevr(
1761 &mut instance.common,
1762 stat,
1763 crate::server::record::AlarmSeverity::from_u16(sevr),
1764 );
1765 }
1766 if let Some(ts) = dev_ts {
1767 instance.common.time = ts;
1768 }
1769 // C device support writes `prec->utag` directly during
1770 // `read()` — the event-system pulse-id path, since
1771 // `epicsTimeStamp` carries no tag. Adopt the device's
1772 // userTag when it supplies one; read in the same `dev`
1773 // borrow as the timestamp above so the time/tag pair is a
1774 // single consistent device snapshot.
1775 if let Some(utag) = dev_utag {
1776 instance.common.utag = utag;
1777 }
1778 }
1779
1780 // pvalink `time=true` adopts the latched upstream timestamp
1781 // into the owning record. `external_link_time` returned
1782 // `None` unless the lset signalled the option, so a `Some`
1783 // here is the operator-requested remote timestamp: the remote
1784 // NT `timeStamp` while connected, or the disconnect-event time
1785 // while the subscription is down (pvxs `snap_time = e.time`,
1786 // adopted on the invalid read — `pvalink_lset.cpp:268-270`).
1787 // Apply BEFORE `apply_timestamp` so the upstream value
1788 // survives the soft-channel TSE=0 default (`apply_timestamp`
1789 // would otherwise stamp wall-clock-now on top).
1790 if let Some((secs, ns, utag)) = inp_link_remote_time {
1791 let secs = secs.max(0) as u64;
1792 let ns = ns.max(0) as u32;
1793 instance.common.time =
1794 std::time::UNIX_EPOCH + std::time::Duration::new(secs, ns.min(999_999_999));
1795 // adopt the upstream `timeStamp.userTag` alongside the
1796 // time, mirroring pvxs PR-added `precord->utag = snap_tag`
1797 // next to `precord->time = snap_time` in the `time=true`
1798 // branch. The tag is already widened without sign
1799 // extension by the lset; `0` when the source carries
1800 // none. `apply_timestamp` never touches `utag`, so this
1801 // survives regardless of the TSE branch below.
1802 instance.common.utag = utag;
1803 // TSE=-2 marks "device-set time" — `apply_timestamp`
1804 // honours this by leaving `common.time` untouched,
1805 // mirroring the device-support timestamp branch above.
1806 instance.common.tse = -2;
1807 }
1808
1809 // Transfer nsta/nsev -> sevr/stat, detect alarm change
1810 let alarm_result = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
1811
1812 // Apply timestamp based on TSE
1813 apply_timestamp(&mut instance.common, is_soft);
1814 // NOTE: UDF was already updated before `evaluate_alarms`
1815 // above — keyed on `value_is_undefined()` so a NaN result
1816 // keeps UDF true and UDF_ALARM is raised this cycle. Do
1817 // NOT clear UDF unconditionally here.
1818
1819 // IVOA check for output records with INVALID alarm
1820 let skip_out = if instance.common.sevr == crate::server::record::AlarmSeverity::Invalid
1821 {
1822 let ivoa = instance
1823 .record
1824 .get_field("IVOA")
1825 .and_then(|v| {
1826 if let EpicsValue::Short(s) = v {
1827 Some(s)
1828 } else {
1829 None
1830 }
1831 })
1832 .unwrap_or(0);
1833 match ivoa {
1834 1 => true, // Don't drive outputs
1835 2 => {
1836 // Set output to IVOV. Each record type knows
1837 // which field its OUT writeback consumes — see
1838 // [`Record::apply_invalid_output_value`]. The
1839 // earlier path special-cased `calcout`
1840 // (OVAL) and fell back to `set_val` (VAL) for
1841 // every other record. That hid a real bug:
1842 // ao/lso/bo/mbbo/busy left their OVAL/RVAL
1843 // staging field stale, so the OUT writeback —
1844 // which reads `OVAL.or(VAL)` — sent the
1845 // pre-IVOA value to the linked record. Per-type
1846 // overrides now apply IVOV to the field that
1847 // matches the C convention.
1848 if let Some(ivov) = instance.record.get_field("IVOV") {
1849 let _ = instance.record.apply_invalid_output_value(ivov);
1850 }
1851 false
1852 }
1853 _ => false, // Continue normally
1854 }
1855 } else {
1856 false
1857 };
1858
1859 // OUT stage: soft channel -> link put, non-soft -> device.write()
1860 // Must run BEFORE check_deadband_ext so MLST is not prematurely
1861 // updated for async writes that return early.
1862 let can_dev_write = instance.record.can_device_write();
1863 let is_soft_out =
1864 instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
1865 let record_should_output = instance.record.should_output();
1866 let out_info = if skip_out {
1867 None
1868 } else if !can_dev_write {
1869 // Non-output records (calcout, etc.) may still have a
1870 // soft OUT link (DB or external ca://`/`pva://`).
1871 // Write OVAL to OUT when the record says should_output().
1872 if record_should_output && instance.parsed_out.is_writable_out_link() {
1873 let oval = instance.record.get_field("OVAL");
1874 let val = instance.record.val();
1875 let out_val = oval.or(val);
1876 out_val.map(|v| (instance.parsed_out.clone(), v))
1877 } else {
1878 None
1879 }
1880 } else if is_soft_out {
1881 if !record_should_output {
1882 // epics-base 7.0.8 OOPT: gate the soft OUT-link
1883 // write on the record's `should_output()`. For
1884 // longout/calcout with OOPT != 0 this lets a
1885 // condition-not-met cycle silently skip the link
1886 // write without disturbing alarms / monitors.
1887 None
1888 } else if instance.parsed_out.is_writable_out_link() {
1889 let out_val = instance
1890 .record
1891 .get_field("OVAL")
1892 .or_else(|| instance.record.val());
1893 out_val.map(|v| (instance.parsed_out.clone(), v))
1894 } else {
1895 None
1896 }
1897 } else if !record_should_output {
1898 // OOPT gating for hardware outputs (longout DTYP=...).
1899 // Skip the device write when the OOPT predicate is
1900 // not satisfied; the record's val/timestamp/snapshot
1901 // path still runs so monitor consumers see the value
1902 // change even on a non-output cycle.
1903 None
1904 } else {
1905 if let Some(mut dev) = instance.device.take() {
1906 // Try async write_begin() first
1907 match dev.write_begin(&mut *instance.record) {
1908 Ok(Some(completion)) => {
1909 // Async write submitted -- set PACT, return early.
1910 // complete_async_record will handle deadband, snapshot,
1911 // notification, and FLNK when the write completes.
1912 instance
1913 .processing
1914 .store(true, std::sync::atomic::Ordering::Release);
1915 instance.device = Some(dev);
1916 let rec_name = instance.name.clone();
1917 let timeout = std::time::Duration::from_secs(5);
1918 let db = self.clone();
1919 tokio::spawn(async move {
1920 let _ =
1921 tokio::task::spawn_blocking(move || completion.wait(timeout))
1922 .await;
1923 let _ = db.complete_async_record(&rec_name).await;
1924 });
1925 return Ok(());
1926 }
1927 Ok(None) => {
1928 // No async support -- fall back to synchronous write
1929 if let Err(e) = dev.write(&mut *instance.record) {
1930 eprintln!("device write error on {}: {e}", instance.name);
1931 instance.common.stat =
1932 crate::server::recgbl::alarm_status::WRITE_ALARM;
1933 instance.common.sevr =
1934 crate::server::record::AlarmSeverity::Invalid;
1935 } else {
1936 // OOPT 7.0.8: notify the record so it can
1937 // latch transition state (e.g. longout.pval)
1938 // for the next cycle.
1939 instance.record.on_output_complete();
1940 }
1941 }
1942 Err(e) => {
1943 eprintln!("device write_begin error on {}: {e}", instance.name);
1944 instance.common.stat = crate::server::recgbl::alarm_status::WRITE_ALARM;
1945 instance.common.sevr = crate::server::record::AlarmSeverity::Invalid;
1946 }
1947 }
1948 instance.device = Some(dev);
1949 }
1950 None
1951 };
1952
1953 // Compute event mask (after OUT stage so async writes don't
1954 // update MLST/ALST prematurely before returning early)
1955 use crate::server::recgbl::EventMask;
1956 let mut event_mask = EventMask::NONE;
1957
1958 let (include_val, include_archive) = match instance.record.monitor_value_changed() {
1959 // lsi/lso post VALUE|LOG only when the string actually
1960 // changed (C `lsiRecord.c`/`lsoRecord.c` monitor: `len !=
1961 // olen || memcmp(oval, val, len)`); they have no MDEL/ADEL
1962 // deadband to express that, so the gate is explicit. The
1963 // MPST/APST `menuPost` "Always" override OR-adds DBE_VALUE /
1964 // DBE_LOG even on an unchanged cycle (C monitor: `if (mpst ==
1965 // menuPost_Always) events |= DBE_VALUE; if (apst ==
1966 // menuPost_Always) events |= DBE_LOG;`).
1967 Some(changed) => {
1968 let (val_always, archive_always) = instance.record.monitor_always_post();
1969 (changed || val_always, changed || archive_always)
1970 }
1971 None => {
1972 if instance.record.uses_monitor_deadband() {
1973 instance.check_deadband_ext()
1974 } else {
1975 // Binary records (bi/bo/busy/mbbi/mbbo): always post monitors
1976 (true, true)
1977 }
1978 }
1979 };
1980 if include_val {
1981 event_mask |= EventMask::VALUE;
1982 }
1983 if include_archive {
1984 event_mask |= EventMask::LOG;
1985 }
1986 if alarm_result.alarm_changed || alarm_result.amsg_changed {
1987 // C `recGbl.c:194/203` — amsg-only OR sevr-changed sets
1988 // `stat_mask = DBE_ALARM`, which raises `val_mask =
1989 // DBE_ALARM` at line 212. Without this, an alarm whose
1990 // sevr/stat is unchanged but whose amsg shifted (e.g.
1991 // device re-flagging the same severity with a different
1992 // human-readable cause) silently drops the AMSG event
1993 // and any DBE_ALARM-only subscribers stay stale.
1994 event_mask |= EventMask::ALARM;
1995 }
1996
1997 // Build snapshot
1998 let mut changed_fields = Vec::new();
1999 // C `recGblResetAlarms` returns `val_mask = DBE_ALARM`
2000 // when any alarm field moved — the record's VAL is posted
2001 // with DBE_ALARM even if the value deadband did not fire,
2002 // so a `DBE_ALARM`-only subscriber sees the value at the
2003 // moment the alarm changed.
2004 let val_on_alarm = alarm_result.alarm_changed || alarm_result.amsg_changed;
2005 if include_val || val_on_alarm {
2006 if let Some(val) = instance.record.val() {
2007 changed_fields.push(("VAL".to_string(), val));
2008 }
2009 }
2010 // Add subscribed fields that actually changed since last notification.
2011 let mut sub_updates: Vec<(String, EpicsValue)> = Vec::new();
2012 for (field, subs) in &instance.subscribers {
2013 if !subs.is_empty()
2014 && field != "VAL"
2015 && field != "SEVR"
2016 && field != "STAT"
2017 && field != "AMSG"
2018 && field != "UDF"
2019 {
2020 if let Some(val) = instance.resolve_field(field) {
2021 let changed = match instance.last_posted.get(field) {
2022 Some(prev) => prev != &val,
2023 None => true,
2024 };
2025 if changed {
2026 sub_updates.push((field.clone(), val));
2027 }
2028 }
2029 }
2030 }
2031 if !sub_updates.is_empty() {
2032 for (field, val) in &sub_updates {
2033 instance.last_posted.insert(field.clone(), val.clone());
2034 }
2035 changed_fields.extend(sub_updates);
2036 event_mask |= crate::server::recgbl::EventMask::VALUE;
2037 }
2038 // C `recGblResetAlarms` (recGbl.c:201-220) posts each
2039 // alarm field with its own per-field mask:
2040 // * SEVR — DBE_VALUE, ONLY when `prev_sevr != new_sevr`.
2041 // * STAT/AMSG — `stat_mask` = DBE_ALARM (on sevr- or
2042 // amsg-change) | DBE_VALUE (on stat-change).
2043 // * ACKS — DBE_VALUE when `stat_mask != 0`.
2044 // The pre-fix port pushed SEVR + STAT together on any
2045 // `alarm_changed`, over-posting SEVR on a stat-only
2046 // transition and collapsing the per-field mask into one
2047 // record-wide mask. Posting these via `notify_field` with
2048 // their individual masks restores C's granularity.
2049 let sevr_changed = instance.common.sevr != alarm_result.prev_sevr;
2050 let stat_changed = instance.common.stat != alarm_result.prev_stat;
2051 let stat_mask = {
2052 let mut m = EventMask::NONE;
2053 if sevr_changed || alarm_result.amsg_changed {
2054 m |= EventMask::ALARM;
2055 }
2056 if stat_changed {
2057 m |= EventMask::VALUE;
2058 }
2059 m
2060 };
2061 if !stat_mask.is_empty() {
2062 // C `val_mask = DBE_ALARM` — the value field carries
2063 // DBE_ALARM whenever any alarm field moved.
2064 event_mask |= EventMask::ALARM;
2065 }
2066 // Defer the SEVR/STAT/AMSG/ACKS posts to dedicated
2067 // `notify_field` calls (collected here, fired after the
2068 // snapshot notify below) so each gets its exact C mask.
2069 let mut alarm_posts: Vec<(&'static str, EventMask)> = Vec::new();
2070 if sevr_changed {
2071 alarm_posts.push(("SEVR", EventMask::VALUE));
2072 }
2073 if !stat_mask.is_empty() {
2074 alarm_posts.push(("STAT", stat_mask));
2075 alarm_posts.push(("AMSG", stat_mask));
2076 }
2077 // C parity (recGbl.c:216): ACKS is posted (DBE_VALUE) only
2078 // when `stat_mask != 0` AND recGblResetAlarms raised it.
2079 if alarm_result.acks_changed && !stat_mask.is_empty() {
2080 alarm_posts.push(("ACKS", EventMask::VALUE));
2081 }
2082 if !event_mask.is_empty() {
2083 changed_fields.push((
2084 "UDF".to_string(),
2085 EpicsValue::Char(if instance.common.udf { 1 } else { 0 }),
2086 ));
2087 }
2088 let snapshot = crate::server::record::ProcessSnapshot {
2089 changed_fields,
2090 event_mask,
2091 };
2092
2093 let flnk_name = if instance.record.should_fire_forward_link() {
2094 if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
2095 Some(l.record.clone())
2096 } else {
2097 None
2098 }
2099 } else {
2100 None
2101 };
2102
2103 // Put-notify completion is NOT fired here. Firing before the
2104 // OUT/FLNK/process-action tail (below) would report the
2105 // WRITE_NOTIFY done while the chain it triggers — including
2106 // an async FLNK target — is still running (C `dbNotify.c`
2107 // keeps the originating record in the waitList until the
2108 // chain settles). The originating record instead `leave`s
2109 // the wait-set at the END of this function, after every PP
2110 // target it drives has joined. See `complete_put_notify`
2111 // at the tail.
2112
2113 (snapshot, out_info, flnk_name, process_actions, alarm_posts)
2114 };
2115
2116 // 3. Notify subscribers (outside lock)
2117 {
2118 let instance = rec.read().await;
2119 instance.notify_from_snapshot(&snapshot);
2120 // Post the alarm fields (SEVR/STAT/AMSG/ACKS) with their
2121 // individual C masks — see recGblResetAlarms above.
2122 for &(field, mask) in &alarm_posts {
2123 instance.notify_field(field, mask);
2124 }
2125 }
2126
2127 // Snapshot source PUTF + put-notify wait-set for the C
2128 // `processTarget` / `dbNotifyAdd` invariants (see
2129 // `write_db_link_value` doc). Captured once here so every OUT /
2130 // multi-OUT / FLNK dispatch in this cycle propagates the same
2131 // bit and joins the same wait-set. The committed alarm is
2132 // captured the same way for `recGblInheritSevrMsg` MS-class
2133 // propagation into the OUT-link target.
2134 let (src_putf, src_notify, src_alarm) = {
2135 let guard = rec.read().await;
2136 (
2137 guard.common.putf,
2138 guard.notify.clone(),
2139 super::links::LinkAlarm {
2140 stat: guard.common.stat,
2141 sevr: guard.common.sevr,
2142 amsg: guard.common.amsg.clone(),
2143 },
2144 )
2145 };
2146
2147 // 4. OUT link — DB *or* external `ca://`/`pva://`. C
2148 // `dbLink.c::dbPutLink` (dbLink.c:434-448) routes every link
2149 // write through the link set's `putValue`, so the OUTPUT side
2150 // dispatches by scheme exactly as the INPUT side does (B
2151 // `resolve_external_pv`). An external link with no registered
2152 // lset fails gracefully inside `write_out_link_value`.
2153 if let Some((ref link, ref out_val)) = out_info {
2154 self.write_out_link_value(
2155 link,
2156 out_val.clone(),
2157 super::links::OutLinkSrc {
2158 putf: src_putf,
2159 notify: src_notify.as_ref(),
2160 alarm: &src_alarm,
2161 },
2162 visited,
2163 depth,
2164 )
2165 .await;
2166 // OOPT 7.0.8: latch the record's post-output state so the
2167 // next cycle's `should_output` sees the right pval.
2168 {
2169 let mut instance = rec.write().await;
2170 instance.record.on_output_complete();
2171 }
2172 }
2173
2174 // 7b. C record support performs a record's OUT/link writes BEFORE
2175 // its forward link: `transformRecord` calls `dbPutLink()`
2176 // (transformRecord.c:608-619) before `monitor()` +
2177 // `recGblFwdLink()`, `scalerRecord` writes COUT/COUTP
2178 // (scalerRecord.c:457-480) before its FLNK block, `throttleRecord`
2179 // writes the selected OUT link (throttleRecord.c:562-580) before
2180 // `recGblFwdLink()`, and `tableRecord` drives speed/drive links
2181 // (tableRecord.c:573-597) before its final FLNK. The
2182 // `ProcessAction::WriteDbLink` contract is documented as "before
2183 // FLNK", so split the requested actions: link writes run now;
2184 // delayed/reprocess and device-command actions (whose timing must
2185 // stay after the FLNK tail) run afterward. A downstream FLNK
2186 // target therefore reads the freshly written value, matching C.
2187 let (link_writes, deferred_actions): (Vec<_>, Vec<_>) =
2188 process_actions.into_iter().partition(|a| {
2189 matches!(
2190 a,
2191 crate::server::record::ProcessAction::WriteDbLink { .. }
2192 | crate::server::record::ProcessAction::WriteDbLinkNotify { .. }
2193 )
2194 });
2195 self.execute_process_actions(name, &rec, link_writes, visited, depth)
2196 .await;
2197
2198 // 4.5 - 7. Multi-output / event / generic-multi-out / FLNK /
2199 // CP / RPRO tail. Shared with the simulation-mode path so a
2200 // simulated record runs the exact same `recGblFwdLink`
2201 // equivalent (C `aiRecord.c:168`).
2202 self.run_forward_link_tail_with_putf(
2203 name,
2204 &rec,
2205 flnk_name.as_deref(),
2206 PutNotifyCtx {
2207 putf: src_putf,
2208 notify: src_notify.as_ref(),
2209 },
2210 visited,
2211 depth,
2212 )
2213 .await;
2214
2215 // 8. Execute the deferred ProcessActions after the FLNK tail:
2216 // `ReprocessAfter` schedules a later reprocess (the current
2217 // cycle's FLNK must proceed first) and `DeviceCommand` posts its
2218 // own monitors after this cycle's snapshot.
2219 self.execute_process_actions(name, &rec, deferred_actions, visited, depth)
2220 .await;
2221
2222 // 9. C `recGbl.c::recGblFwdLink:302` clears `putf = FALSE` at the
2223 // tail of every synchronous process cycle, NOT just on the
2224 // foreign-entry path. When this record was driven through an
2225 // OUT-link propagation (write_db_link_value set our putf), the
2226 // target record's own process cycle must clear it before
2227 // returning — same lifecycle as the source record's PUTF
2228 // (which `put_record_field_from_ca` separately clears at the
2229 // foreign-entry boundary, and the async branch clears in
2230 // `complete_async_record_inner`). Async-pending records skip
2231 // this clear: their FLNK / putf-clear happens later in
2232 // `complete_async_record_inner` once the device round-trip
2233 // completes.
2234 {
2235 let guard = rec.read().await;
2236 if !guard.is_processing() {
2237 drop(guard);
2238 let mut guard = rec.write().await;
2239 guard.common.putf = false;
2240 }
2241 }
2242
2243 // Put-notify completion: the record `leave`s the wait-set only
2244 // here, after its full OUT/FLNK/process-action tail has run — so
2245 // every PP target it drove has already joined (`enter`ed). Gated
2246 // on `is_put_complete`: a record reporting more work (e.g. motor
2247 // mid-move via `is_put_complete()==false`) keeps its membership
2248 // and leaves on the later cycle that completes the put — matching
2249 // the old fire site's gate. An async-pending record returned
2250 // earlier and is handled in `complete_async_record_inner`. The
2251 // completion oneshot fires on the `leave` that empties the set.
2252 {
2253 let mut guard = rec.write().await;
2254 if guard.record.is_put_complete() {
2255 complete_put_notify(&mut guard);
2256 }
2257 }
2258
2259 Ok(())
2260 }
2261
2262 /// Forward-link / CP / RPRO tail for the simulation-mode path.
2263 ///
2264 /// C `aiRecord.c:151-168`: a record in SIMM mode handles the value
2265 /// inside `readValue()`, then `process()` still runs `monitor` +
2266 /// `recGblFwdLink(prec)`. The simulation path in
2267 /// `process_record_with_links_inner` does its own monitor posting,
2268 /// so this drives the forward-link / CP / RPRO tail that
2269 /// `recGblFwdLink` would. `flnk_name` and `src_putf` are derived
2270 /// fresh from the record (a simulated cycle does not change FLNK,
2271 /// and SIOL reads/writes do not carry a foreign PUTF into the
2272 /// chain).
2273 async fn run_forward_link_tail(
2274 &self,
2275 name: &str,
2276 rec: &Arc<RwLock<RecordInstance>>,
2277 visited: &mut std::collections::HashSet<String>,
2278 depth: usize,
2279 ) {
2280 let (flnk_name, src_putf, src_notify) = {
2281 let instance = rec.read().await;
2282 let flnk = if instance.record.should_fire_forward_link() {
2283 if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
2284 Some(l.record.clone())
2285 } else {
2286 None
2287 }
2288 } else {
2289 None
2290 };
2291 (flnk, instance.common.putf, instance.notify.clone())
2292 };
2293 self.run_forward_link_tail_with_putf(
2294 name,
2295 rec,
2296 flnk_name.as_deref(),
2297 PutNotifyCtx {
2298 putf: src_putf,
2299 notify: src_notify.as_ref(),
2300 },
2301 visited,
2302 depth,
2303 )
2304 .await;
2305 }
2306
2307 /// Steps 4.5 - 7 of the process chain: multi-output dispatch,
2308 /// event-record posting, generic OUTA..OUTP links, FLNK forward
2309 /// link, CP-target dispatch, and RPRO reprocess. Shared by the
2310 /// main process path and the simulation-mode path so both run the
2311 /// identical `recGblFwdLink` equivalent.
2312 async fn run_forward_link_tail_with_putf(
2313 &self,
2314 name: &str,
2315 rec: &Arc<RwLock<RecordInstance>>,
2316 flnk_name: Option<&str>,
2317 src: PutNotifyCtx<'_>,
2318 visited: &mut std::collections::HashSet<String>,
2319 depth: usize,
2320 ) {
2321 // 4.5. Multi-output dispatch (fanout/dfanout/seq)
2322 self.dispatch_multi_output(rec, visited, depth).await;
2323
2324 // 4.55. event record: post the named software event.
2325 self.dispatch_event_record(rec).await;
2326
2327 // 4.6. Generic multi-output links (transform OUTA..OUTP -> A..P,
2328 // scalcout OUT->OVAL, epid OUTL).
2329 //
2330 // SINGLE-OWNER INVARIANT: a record type whose link groups are
2331 // dispatched by `dispatch_multi_output` (§4.5 above) MUST be
2332 // skipped here — otherwise its `LNKn`/`OUTn` would be written
2333 // twice per cycle. `sseq` previously also implemented the
2334 // `Record::multi_output_links` trait method, so this block
2335 // re-dispatched every selected `LNKn` after §4.5 already drove
2336 // it. The `multi_output_dispatch_owned` gate makes the
2337 // double-dispatch structurally impossible — not just removed
2338 // at the `SseqRecord` call site.
2339 {
2340 let multi_out = {
2341 let instance = rec.read().await;
2342 let links =
2343 if super::links::multi_output_dispatch_owned(instance.record.record_type()) {
2344 &[][..]
2345 } else {
2346 instance.record.multi_output_links()
2347 };
2348 if links.is_empty() {
2349 None
2350 } else {
2351 let mut pairs = Vec::new();
2352 for &(link_field, val_field) in links {
2353 let link_str = instance
2354 .record
2355 .get_field(link_field)
2356 .and_then(|v| {
2357 if let EpicsValue::String(s) = v {
2358 Some(s)
2359 } else {
2360 None
2361 }
2362 })
2363 .unwrap_or_default();
2364 if link_str.is_empty() {
2365 continue;
2366 }
2367 if let Some(val) = instance.record.get_field(val_field) {
2368 pairs.push((link_str, val));
2369 }
2370 }
2371 if pairs.is_empty() { None } else { Some(pairs) }
2372 }
2373 };
2374 if let Some(pairs) = multi_out {
2375 // Source committed alarm for `recGblInheritSevrMsg`
2376 // MS-class propagation into each OUT-link target —
2377 // captured once, same lifecycle as `src.putf`.
2378 let src_alarm = {
2379 let guard = rec.read().await;
2380 super::links::LinkAlarm {
2381 stat: guard.common.stat,
2382 sevr: guard.common.sevr,
2383 amsg: guard.common.amsg.clone(),
2384 }
2385 };
2386 for (link_str, val) in pairs {
2387 // `multi_output_links` carries record OUT links
2388 // (sseq `LNKn`, scalcout `OUTn` — all `DBF_OUTLINK`)
2389 // driven via `dbPutLink` → `dbDbPutValue`
2390 // (`dbDbLink.c:388`): a bare DB link is NPP, the
2391 // value is written but the target is NOT processed.
2392 // `parse_output_link_v2` applies the
2393 // OUT-link-correct NPP default; `parse_link_v2` would
2394 // wrongly default a bare link to ProcessPassive and
2395 // re-process the target. An external `ca://`/`pva://`
2396 // OUT link is routed through the link set's
2397 // `putValue` (C `dbLink.c::dbPutLink`,
2398 // dbLink.c:434-448).
2399 let parsed = crate::server::record::parse_output_link_v2(
2400 link_str.as_str_lossy().as_ref(),
2401 );
2402 self.write_out_link_value(
2403 &parsed,
2404 val,
2405 super::links::OutLinkSrc {
2406 putf: src.putf,
2407 notify: src.notify,
2408 alarm: &src_alarm,
2409 },
2410 visited,
2411 depth,
2412 )
2413 .await;
2414 }
2415 }
2416 }
2417
2418 // 5. FLNK -- only process if target is Passive (like C dbScanFwdLink).
2419 // FLNK goes through C `dbScanPassive` -> `processTarget`, which
2420 // propagates `src.putf` to the target the same way OUT links do.
2421 if let Some(flnk) = flnk_name {
2422 if let Some(target_rec) = self.get_record(flnk).await {
2423 let (target_scan, should_process) = {
2424 let mut tg = target_rec.write().await;
2425 let pact = tg.is_processing();
2426 let on_chain = visited.contains(flnk);
2427 let scan = tg.common.scan;
2428 if !pact {
2429 tg.common.putf = src.putf;
2430 // C `dbNotifyAdd` (dbDbLink.c:460) lives inside
2431 // `processTarget`, which `dbScanPassive` reaches
2432 // ONLY for a passive target (it returns early for
2433 // non-passive — dbDbLink.c:431). Gate the join on
2434 // the same passive condition as the process call
2435 // below: a non-passive FLNK target is dropped here
2436 // and must NOT join, or it would `enter` the
2437 // wait-set without ever processing to `leave` it,
2438 // hanging the completion forever.
2439 if scan == crate::server::record::ScanType::Passive {
2440 join_put_notify(&mut tg, src.notify);
2441 }
2442 } else if src.putf && !on_chain {
2443 tg.common.rpro = true;
2444 tg.common.putf = false;
2445 }
2446 (scan, !pact)
2447 };
2448 if should_process && target_scan == crate::server::record::ScanType::Passive {
2449 // recursive FLNK within one chain — gate
2450 // already held by the foreign entry record.
2451 let _ = self
2452 .process_record_with_links_recursive(flnk, visited, depth + 1)
2453 .await;
2454 }
2455 }
2456 }
2457
2458 // 5b. FLNK whose target is external (`pva://`/`ca://`): C
2459 // `dbScanFwdLink` dispatches it through the link set's
2460 // `scanForward` (pvalink `pvaScanForward`), a process-only trigger
2461 // of the remote target. The `flnk_name` above only ever names a
2462 // local DB target, so a non-DB FLNK is forwarded here through the
2463 // single owner.
2464 self.dispatch_external_forward_link(rec).await;
2465
2466 // 6. CP link targets -- process records that have CP input links from this record
2467 self.dispatch_cp_targets(name, visited, depth).await;
2468
2469 // 7. RPRO: if reprocess requested, clear flag and queue a
2470 // fresh process pass.
2471 //
2472 // C `recGblFwdLink` (recGbl.c:296-300) consumes RPRO via
2473 // `scanOnce(pdbc)` — the record is QUEUED on the scanOnce ring
2474 // buffer and reprocessed in a separate pass with a fresh lock
2475 // cycle AFTER the current process chain fully unwinds. It does
2476 // NOT recurse inline within the current link chain.
2477 //
2478 // Spawning a detached task is the Rust equivalent of the
2479 // scanOnce queue: the reprocess runs with a clean (empty)
2480 // `visited` set and starts at depth 0, so it cannot be
2481 // silently skipped by the current chain's cycle guard nor hit
2482 // the MAX_LINK_DEPTH / MAX_LINK_OPS budget the current chain
2483 // has already consumed.
2484 {
2485 let needs_rpro = {
2486 let mut instance = rec.write().await;
2487 if instance.common.rpro {
2488 instance.common.rpro = false;
2489 true
2490 } else {
2491 false
2492 }
2493 };
2494 if needs_rpro {
2495 let db = self.clone();
2496 let rpro_name = name.to_string();
2497 crate::runtime::task::spawn(async move {
2498 let mut fresh_visited = std::collections::HashSet::new();
2499 let _ = db
2500 .process_record_with_links(&rpro_name, &mut fresh_visited, 0)
2501 .await;
2502 });
2503 }
2504 }
2505 }
2506
2507 /// Fire a non-DB (external `pva://`/`ca://`) forward link (FLNK).
2508 ///
2509 /// C `recGblFwdLink` → `dbScanFwdLink` (`dbLink.c:475-480`) dispatches
2510 /// every FLNK uniformly through `plink->lset->scanForward`: a DB lset
2511 /// runs `scanOnce(target)` — handled directly by the local FLNK §5
2512 /// path — while the pvalink/calink lset runs `pvaScanForward`, a
2513 /// process-only trigger of the remote target. The DB-only `flnk_name`
2514 /// filter at the three `should_fire_forward_link` sites dropped every
2515 /// external FLNK; this is the single owner that forwards them, so the
2516 /// dispatch is not open-coded per site (each FLNK tail calls only
2517 /// this).
2518 ///
2519 /// On a non-retry, disconnected link the lset returns `Err`; pvxs
2520 /// raises `recGblSetSevrMsg(LINK_ALARM, INVALID_ALARM, "Disconn")` on
2521 /// the owning record (`pvxs/ioc/pvalink_lset.cpp:677-679`). This raises
2522 /// the same *pending* LINK/INVALID alarm via [`rec_gbl_set_sevr_msg`],
2523 /// promoted by the next `recGblResetAlarms` — exactly as the C late-set
2524 /// inside `recGblFwdLink` (after the record's own alarm/monitor stage)
2525 /// is.
2526 async fn dispatch_external_forward_link(&self, rec: &Arc<RwLock<RecordInstance>>) {
2527 let target = {
2528 let instance = rec.read().await;
2529 if !instance.record.should_fire_forward_link() {
2530 return;
2531 }
2532 match &instance.parsed_flnk {
2533 crate::server::record::ParsedLink::Pva(_)
2534 | crate::server::record::ParsedLink::PvaJson(_)
2535 | crate::server::record::ParsedLink::Ca(_) => instance
2536 .parsed_flnk
2537 .external_pv_name()
2538 .map(|s| s.to_string()),
2539 // A DB FLNK is processed by the local §5 scanOnce path;
2540 // every other kind (Constant/Hw/Calc/None) carries no
2541 // forward action.
2542 _ => None,
2543 }
2544 };
2545 let Some(target) = target else {
2546 return;
2547 };
2548 if let Err(e) = self.scan_forward_external_pv(&target).await {
2549 let _ = e;
2550 let mut instance = rec.write().await;
2551 crate::server::recgbl::rec_gbl_set_sevr_msg(
2552 &mut instance.common,
2553 crate::server::recgbl::alarm_status::LINK_ALARM,
2554 crate::server::record::AlarmSeverity::Invalid,
2555 "Disconn",
2556 );
2557 }
2558 }
2559
2560 /// Execute ReadDbLink actions before process().
2561 /// Reads linked PV values and writes them into record fields via put_field_internal.
2562 async fn execute_read_db_links(
2563 &self,
2564 _record_name: &str,
2565 rec: &Arc<crate::runtime::sync::RwLock<RecordInstance>>,
2566 actions: &[crate::server::record::ProcessAction],
2567 visited: &mut HashSet<String>,
2568 depth: usize,
2569 ) {
2570 use crate::server::record::ProcessAction;
2571 for action in actions {
2572 if let ProcessAction::ReadDbLink {
2573 link_field,
2574 target_field,
2575 } = action
2576 {
2577 let link_str = {
2578 let instance = rec.read().await;
2579 instance
2580 .record
2581 .get_field(link_field)
2582 .and_then(|v| {
2583 if let EpicsValue::String(s) = v {
2584 Some(s)
2585 } else {
2586 None
2587 }
2588 })
2589 .unwrap_or_default()
2590 };
2591 if link_str.is_empty() {
2592 continue;
2593 }
2594 let parsed = crate::server::record::parse_link_v2(link_str.as_str_lossy().as_ref());
2595 if let Some(value) = self.read_link_value(&parsed, visited, depth).await {
2596 let mut instance = rec.write().await;
2597 let _ = instance.record.put_field_internal(target_field, value);
2598 }
2599 }
2600 }
2601 }
2602
2603 /// Execute ProcessActions returned by a record's process() call.
2604 ///
2605 /// Actions are executed in order:
2606 /// - ReadDbLink: reads a linked PV value and writes it into a record field
2607 /// (bypasses read-only checks via put_field_internal)
2608 /// - WriteDbLink: writes a value to a linked PV
2609 /// - ReprocessAfter: schedules a delayed re-process via tokio::spawn
2610 async fn execute_process_actions(
2611 &self,
2612 record_name: &str,
2613 rec: &Arc<crate::runtime::sync::RwLock<RecordInstance>>,
2614 actions: Vec<crate::server::record::ProcessAction>,
2615 visited: &mut HashSet<String>,
2616 depth: usize,
2617 ) {
2618 use crate::server::record::ProcessAction;
2619
2620 for action in actions {
2621 match action {
2622 ProcessAction::ReadDbLink {
2623 link_field,
2624 target_field,
2625 } => {
2626 // 1. Get the link string from the record
2627 let link_str = {
2628 let instance = rec.read().await;
2629 instance
2630 .record
2631 .get_field(link_field)
2632 .and_then(|v| {
2633 if let EpicsValue::String(s) = v {
2634 Some(s)
2635 } else {
2636 None
2637 }
2638 })
2639 .unwrap_or_default()
2640 };
2641 if link_str.is_empty() {
2642 continue;
2643 }
2644 // 2. Parse and read the linked PV
2645 let parsed =
2646 crate::server::record::parse_link_v2(link_str.as_str_lossy().as_ref());
2647 if let Some(value) = self.read_link_value(&parsed, visited, depth).await {
2648 // 3. Write into the record field (internal put bypasses read-only)
2649 let mut instance = rec.write().await;
2650 let _ = instance.record.put_field_internal(target_field, value);
2651 }
2652 }
2653 ProcessAction::WriteDbLink { link_field, value } => {
2654 // 1. Get the link string (record fields → common fields)
2655 // and the source PUTF for processTarget propagation,
2656 // plus the committed alarm for `recGblInheritSevrMsg`
2657 // MS-class propagation into the OUT-link target.
2658 let (link_str, src_putf, src_notify, src_alarm) = {
2659 let instance = rec.read().await;
2660 let link = instance
2661 .resolve_field(link_field)
2662 .and_then(|v| {
2663 if let EpicsValue::String(s) = v {
2664 Some(s)
2665 } else {
2666 None
2667 }
2668 })
2669 .unwrap_or_default();
2670 (
2671 link,
2672 instance.common.putf,
2673 instance.notify.clone(),
2674 super::links::LinkAlarm {
2675 stat: instance.common.stat,
2676 sevr: instance.common.sevr,
2677 amsg: instance.common.amsg.clone(),
2678 },
2679 )
2680 };
2681 if link_str.is_empty() {
2682 continue;
2683 }
2684 // 2. Parse and write to the linked PV — DB *or*
2685 // external `ca://`/`pva://`. A record's `process()`
2686 // emits `WriteDbLink` to drive an OUT-link field
2687 // (transform `OUTn`, throttle/scaler `COUTP`, epid
2688 // `TRIG`/`OUTL`); that field may resolve to a CA/PVA
2689 // link, which C `dbPutLink` routes through the link
2690 // set's `putValue` identically to a DB link
2691 // (dbLink.c:434-448).
2692 let parsed =
2693 crate::server::record::parse_link_v2(link_str.as_str_lossy().as_ref());
2694 self.write_out_link_value(
2695 &parsed,
2696 value,
2697 super::links::OutLinkSrc {
2698 putf: src_putf,
2699 notify: src_notify.as_ref(),
2700 alarm: &src_alarm,
2701 },
2702 visited,
2703 depth,
2704 )
2705 .await;
2706 }
2707 ProcessAction::DeviceCommand { command, ref args } => {
2708 let mut instance = rec.write().await;
2709 if let Some(mut dev) = instance.device.take() {
2710 // `handle_command` runs after the process snapshot
2711 // was already built/notified, so any record field
2712 // it mutated needs an explicit monitor post. The
2713 // returned field names are posted with DBE_VALUE,
2714 // mirroring the C record's `db_post_events` calls
2715 // from inside `process()` (scalerRecord.c:425-430).
2716 let changed = dev
2717 .handle_command(&mut *instance.record, command, args)
2718 .unwrap_or_default();
2719 instance.device = Some(dev);
2720 for field in changed {
2721 instance.notify_field(field, crate::server::recgbl::EventMask::VALUE);
2722 }
2723 }
2724 }
2725 ProcessAction::ReprocessAfter(delay) => {
2726 // Owner-driven delayed re-entry, mirroring C
2727 // `callbackRequestDelayed` dispatching to
2728 // `(*prset->process)(prec)` directly (callback.c). Mint
2729 // a fresh token — which advances the record's generation
2730 // and so supersedes any prior pending re-entry for this
2731 // record (a newer ReprocessAfter replaces the older
2732 // timer) — then fire it after the delay. A newer mint or
2733 // an explicit `cancel_async_reentry` makes this fire a
2734 // structural no-op: the gate lives entirely in
2735 // `AsyncToken`, not in an inline generation compare here.
2736 let token = match self.mint_async_token(record_name).await {
2737 Some(t) => t,
2738 None => continue,
2739 };
2740 let db = self.clone();
2741 tokio::spawn(async move {
2742 tokio::time::sleep(delay).await;
2743 let _ = token.fire(&db).await;
2744 });
2745 }
2746 ProcessAction::WriteDbLinkNotify { link_field, value } => {
2747 // C `sseqRecord.c` WAITn put-callback dependency: write
2748 // the OUT link as a put-WITH-completion and re-enter THIS
2749 // record's process() once the downstream record (plus its
2750 // FLNK/OUT chain) finishes. Same OUT-link write a plain
2751 // WriteDbLink performs, wrapped in the c401e2f0 put-notify
2752 // wait-set + async re-entry primitive.
2753 let (link_str, src_putf, src_alarm) = {
2754 let instance = rec.read().await;
2755 let link = instance
2756 .resolve_field(link_field)
2757 .and_then(|v| {
2758 if let EpicsValue::String(s) = v {
2759 Some(s)
2760 } else {
2761 None
2762 }
2763 })
2764 .unwrap_or_default();
2765 (
2766 link,
2767 instance.common.putf,
2768 super::links::LinkAlarm {
2769 stat: instance.common.stat,
2770 sevr: instance.common.sevr,
2771 amsg: instance.common.amsg.clone(),
2772 },
2773 )
2774 };
2775 // Mint the re-entry token BEFORE issuing the put so a
2776 // synchronous downstream completion cannot fire the
2777 // oneshot before the waiter is wired. The mint supersedes
2778 // any prior pending re-entry for this record (newer
2779 // token), exactly like ReprocessAfter.
2780 let token = match self.mint_async_token(record_name).await {
2781 Some(t) => t,
2782 None => continue,
2783 };
2784 let (waitset, completion) = Self::new_put_notify();
2785 if !link_str.is_empty() {
2786 let parsed =
2787 crate::server::record::parse_link_v2(link_str.as_str_lossy().as_ref());
2788 self.write_out_link_value(
2789 &parsed,
2790 value,
2791 super::links::OutLinkSrc {
2792 putf: src_putf,
2793 notify: Some(&waitset),
2794 alarm: &src_alarm,
2795 },
2796 visited,
2797 depth,
2798 )
2799 .await;
2800 }
2801 // Release the initiator's own wait-set count (C
2802 // `dbProcessNotify` holds one count for the requester and
2803 // drops it after issuing the put). The set then drains —
2804 // and fires the completion — when the downstream
2805 // target(s) that joined via `join_put_notify` finish, or
2806 // immediately when the link was empty / the target
2807 // completed synchronously.
2808 waitset.leave();
2809 self.reprocess_on_notify(token, completion);
2810 }
2811 ProcessAction::CancelReprocess => {
2812 // C `callbackCancelDelayed` for `sseq` ABORT: advance the
2813 // record's re-entry generation so any pending DLYn timer
2814 // or WAITn notify re-entry becomes a structural no-op (the
2815 // AsyncToken gate), with no runtime is-aborted check on
2816 // the re-entry path.
2817 self.cancel_async_reentry(record_name).await;
2818 }
2819 }
2820 }
2821 }
2822
2823 /// Complete an asynchronous record's post-process steps.
2824 /// Call after device support signals completion (clears PACT, runs alarms, snapshot, OUT, FLNK).
2825 pub fn complete_async_record<'a>(
2826 &'a self,
2827 name: &'a str,
2828 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'a>> {
2829 Box::pin(async move {
2830 let mut visited = HashSet::new();
2831 self.complete_async_record_inner(name, &mut visited, 0)
2832 .await
2833 })
2834 }
2835
2836 async fn complete_async_record_inner(
2837 &self,
2838 name: &str,
2839 visited: &mut HashSet<String>,
2840 depth: usize,
2841 ) -> CaResult<()> {
2842 // Alias-aware entry — same pattern as
2843 // `process_record_with_links_inner`. `name` may arrive as an
2844 // alias from an async device-support callback that captured
2845 // the original record name; normalise to canonical so the
2846 // records-map lookup, the `visited` cycle set, and downstream
2847 // FLNK/OUT dispatches all see the same canonical name.
2848 let canonical_owned;
2849 let name: &str = if let Some(target) = self.resolve_alias(name).await {
2850 canonical_owned = target;
2851 &canonical_owned
2852 } else {
2853 name
2854 };
2855
2856 let rec = {
2857 let records = self.inner.records.read().await;
2858 records
2859 .get(name)
2860 .cloned()
2861 .ok_or_else(|| CaError::ChannelNotFound(name.to_string()))?
2862 };
2863
2864 // Seed the cycle guard with this record's own name — mirrors
2865 // the synchronous main path (`process_record_with_links_inner`
2866 // does `visited.insert(name)` before the body). Without this
2867 // the async-completion FLNK / OUT / CP dispatch can re-enter
2868 // the just-completed record: an async FLNK chain that loops
2869 // back (A async -> completes -> FLNK -> B -> FLNK -> A) would
2870 // re-process A unbounded, because PACT is cleared below before
2871 // the FLNK dispatch and nothing else blocks the re-entry.
2872 if !visited.insert(name.to_string()) {
2873 return Ok(()); // Cycle detected, skip
2874 }
2875
2876 let (snapshot, out_info, flnk_name, alarm_posts) = {
2877 let mut instance = rec.write().await;
2878
2879 // UDF update before alarm evaluation (C parity — see the
2880 // sync process path). A NaN/undefined value keeps UDF true
2881 // so `recGblCheckUDF` raises UDF_ALARM this cycle.
2882 if instance.record.clears_udf() {
2883 instance.common.udf = instance.record.value_is_undefined();
2884 }
2885 // Per-record alarm hook (C `checkAlarms()`).
2886 {
2887 let inst = &mut *instance;
2888 inst.record.check_alarms(&mut inst.common);
2889 }
2890
2891 // Evaluate alarms
2892 instance.evaluate_alarms();
2893
2894 let is_soft = instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
2895
2896 // Device support alarm/timestamp override
2897 if !is_soft {
2898 let (dev_alarm, dev_ts, dev_utag) = if let Some(ref dev) = instance.device {
2899 (dev.last_alarm(), dev.last_timestamp(), dev.last_utag())
2900 } else {
2901 (None, None, None)
2902 };
2903 if let Some((stat, sevr)) = dev_alarm {
2904 crate::server::recgbl::rec_gbl_set_sevr(
2905 &mut instance.common,
2906 stat,
2907 crate::server::record::AlarmSeverity::from_u16(sevr),
2908 );
2909 }
2910 if let Some(ts) = dev_ts {
2911 instance.common.time = ts;
2912 }
2913 // C device support writes `prec->utag` directly during
2914 // `read()` — the event-system pulse-id path, since
2915 // `epicsTimeStamp` carries no tag. Adopt the device's
2916 // userTag when it supplies one; read in the same `dev`
2917 // borrow as the timestamp above so the time/tag pair is a
2918 // single consistent device snapshot.
2919 if let Some(utag) = dev_utag {
2920 instance.common.utag = utag;
2921 }
2922 }
2923
2924 let alarm_result = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
2925
2926 apply_timestamp(&mut instance.common, is_soft);
2927 // UDF was already updated before `evaluate_alarms` above.
2928
2929 // Clear PACT
2930 instance
2931 .processing
2932 .store(false, std::sync::atomic::Ordering::Release);
2933
2934 // Put-notify completion is NOT fired here. The async device
2935 // round-trip has finished, but the OUT/FLNK/process-action
2936 // tail it drives (below) may itself reach an async target;
2937 // firing now would report WRITE_NOTIFY done while that chain
2938 // still runs. The originating record `leave`s the wait-set at
2939 // the END of this function, after every PP target it drives
2940 // has joined. See `complete_put_notify` at the tail.
2941
2942 use crate::server::recgbl::EventMask;
2943 let mut event_mask = EventMask::NONE;
2944 let (include_val, include_archive) = match instance.record.monitor_value_changed() {
2945 // lsi/lso post VALUE|LOG only when the string actually
2946 // changed (C `lsiRecord.c`/`lsoRecord.c` monitor: `len !=
2947 // olen || memcmp(oval, val, len)`); they have no MDEL/ADEL
2948 // deadband to express that, so the gate is explicit. The
2949 // MPST/APST `menuPost` "Always" override OR-adds DBE_VALUE /
2950 // DBE_LOG even on an unchanged cycle (C monitor: `if (mpst ==
2951 // menuPost_Always) events |= DBE_VALUE; if (apst ==
2952 // menuPost_Always) events |= DBE_LOG;`).
2953 Some(changed) => {
2954 let (val_always, archive_always) = instance.record.monitor_always_post();
2955 (changed || val_always, changed || archive_always)
2956 }
2957 None => {
2958 if instance.record.uses_monitor_deadband() {
2959 instance.check_deadband_ext()
2960 } else {
2961 // Binary records (bi/bo/busy/mbbi/mbbo): always post monitors
2962 (true, true)
2963 }
2964 }
2965 };
2966 if include_val {
2967 event_mask |= EventMask::VALUE;
2968 }
2969 if include_archive {
2970 event_mask |= EventMask::LOG;
2971 }
2972 if alarm_result.alarm_changed || alarm_result.amsg_changed {
2973 // C `recGbl.c:194/203` — same parity rule as the main
2974 // process path above (see comment there): amsg-only OR
2975 // sevr/stat change → DBE_ALARM.
2976 event_mask |= EventMask::ALARM;
2977 }
2978
2979 let mut changed_fields = Vec::new();
2980 if include_val {
2981 if let Some(val) = instance.record.val() {
2982 changed_fields.push(("VAL".to_string(), val));
2983 }
2984 }
2985 // C `recGblResetAlarms` (recGbl.c:201-220) posts each alarm
2986 // field with its OWN per-field mask, not the record-wide
2987 // `event_mask`. Mirror the synchronous link path
2988 // (`process_record_with_links_inner`) and `process_local`
2989 // exactly: SEVR=DBE_VALUE on a sevr change; STAT/AMSG share
2990 // `stat_mask` which carries DBE_ALARM when sevr OR amsg
2991 // moved and DBE_VALUE on a stat change; ACKS=DBE_VALUE only
2992 // when an alarm field moved AND recGblResetAlarms raised it.
2993 // Collapsing these into `changed_fields` would post them all
2994 // on one record-wide mask — losing C's per-field
2995 // granularity for `.SEVR`/`.STAT`-only subscribers.
2996 let sevr_changed = instance.common.sevr != alarm_result.prev_sevr;
2997 let stat_changed = instance.common.stat != alarm_result.prev_stat;
2998 let stat_mask = {
2999 let mut m = EventMask::NONE;
3000 if sevr_changed || alarm_result.amsg_changed {
3001 m |= EventMask::ALARM;
3002 }
3003 if stat_changed {
3004 m |= EventMask::VALUE;
3005 }
3006 m
3007 };
3008 let mut alarm_posts: Vec<(&'static str, EventMask)> = Vec::new();
3009 if sevr_changed {
3010 alarm_posts.push(("SEVR", EventMask::VALUE));
3011 }
3012 if !stat_mask.is_empty() {
3013 alarm_posts.push(("STAT", stat_mask));
3014 alarm_posts.push(("AMSG", stat_mask));
3015 // C `val_mask = DBE_ALARM` — the value field carries
3016 // DBE_ALARM whenever any alarm field moved.
3017 event_mask |= EventMask::ALARM;
3018 }
3019 // C parity (recGbl.c:216): ACKS is posted (DBE_VALUE) only
3020 // when an alarm field moved AND recGblResetAlarms raised it.
3021 if alarm_result.acks_changed && !stat_mask.is_empty() {
3022 alarm_posts.push(("ACKS", EventMask::VALUE));
3023 }
3024 if !event_mask.is_empty() {
3025 changed_fields.push((
3026 "UDF".to_string(),
3027 EpicsValue::Char(if instance.common.udf { 1 } else { 0 }),
3028 ));
3029 }
3030 // Add subscribed non-{VAL,SEVR,STAT,AMSG,UDF} fields that
3031 // actually changed since last notification — mirrors the
3032 // main-path snapshot gate (process_record_with_links_inner
3033 // L794-820). Without this, every async-completion cycle
3034 // re-sends every subscribed auxiliary field even when its
3035 // value is unchanged, multiplying the monitor traffic for
3036 // any record that pairs an async write with a sticky
3037 // metadata field.
3038 let mut sub_updates: Vec<(String, EpicsValue)> = Vec::new();
3039 for (field, subs) in &instance.subscribers {
3040 if !subs.is_empty()
3041 && field != "VAL"
3042 && field != "SEVR"
3043 && field != "STAT"
3044 && field != "AMSG"
3045 && field != "UDF"
3046 {
3047 if let Some(val) = instance.resolve_field(field) {
3048 let changed = match instance.last_posted.get(field) {
3049 Some(prev) => prev != &val,
3050 None => true,
3051 };
3052 if changed {
3053 sub_updates.push((field.clone(), val));
3054 }
3055 }
3056 }
3057 }
3058 if !sub_updates.is_empty() {
3059 for (field, val) in &sub_updates {
3060 instance.last_posted.insert(field.clone(), val.clone());
3061 }
3062 changed_fields.extend(sub_updates);
3063 event_mask |= crate::server::recgbl::EventMask::VALUE;
3064 }
3065 let snapshot = crate::server::record::ProcessSnapshot {
3066 changed_fields,
3067 event_mask,
3068 };
3069
3070 // IVOA check
3071 let skip_out = if instance.common.sevr == crate::server::record::AlarmSeverity::Invalid
3072 {
3073 let ivoa = instance
3074 .record
3075 .get_field("IVOA")
3076 .and_then(|v| {
3077 if let EpicsValue::Short(s) = v {
3078 Some(s)
3079 } else {
3080 None
3081 }
3082 })
3083 .unwrap_or(0);
3084 match ivoa {
3085 1 => true,
3086 2 => {
3087 // See the IVOA=2 comment in
3088 // `process_record_with_links_inner` — IVOA=2
3089 // delegates to the per-record
3090 // `apply_invalid_output_value` so OVAL/RVAL/VAL
3091 // get the C-convention values.
3092 if let Some(ivov) = instance.record.get_field("IVOV") {
3093 let _ = instance.record.apply_invalid_output_value(ivov);
3094 }
3095 false
3096 }
3097 _ => false,
3098 }
3099 } else {
3100 false
3101 };
3102
3103 let can_dev_write = instance.record.can_device_write();
3104 let is_soft_out =
3105 instance.common.dtyp.is_empty() || instance.common.dtyp == "Soft Channel";
3106 let record_should_output = instance.record.should_output();
3107 let out_info = if skip_out {
3108 None
3109 } else if !can_dev_write {
3110 // Non-output records (calcout, etc.) with soft OUT link
3111 // (DB or external `ca://`/`pva://`).
3112 if record_should_output && instance.parsed_out.is_writable_out_link() {
3113 let out_val = instance
3114 .record
3115 .get_field("OVAL")
3116 .or_else(|| instance.record.val());
3117 out_val.map(|v| (instance.parsed_out.clone(), v))
3118 } else {
3119 None
3120 }
3121 } else if is_soft_out {
3122 if instance.parsed_out.is_writable_out_link() {
3123 let out_val = instance
3124 .record
3125 .get_field("OVAL")
3126 .or_else(|| instance.record.val());
3127 out_val.map(|v| (instance.parsed_out.clone(), v))
3128 } else {
3129 None
3130 }
3131 } else {
3132 // Non-soft output: the async device write already completed
3133 // (that's why we're in complete_async_record). Don't re-do
3134 // write_begin -- it would start another async cycle.
3135 None
3136 };
3137
3138 let flnk_name = if instance.record.should_fire_forward_link() {
3139 if let crate::server::record::ParsedLink::Db(ref l) = instance.parsed_flnk {
3140 Some(l.record.clone())
3141 } else {
3142 None
3143 }
3144 } else {
3145 None
3146 };
3147
3148 (snapshot, out_info, flnk_name, alarm_posts)
3149 };
3150
3151 // Notify subscribers
3152 {
3153 let instance = rec.read().await;
3154 instance.notify_from_snapshot(&snapshot);
3155 // Post the alarm fields (SEVR/STAT/AMSG/ACKS) with their
3156 // individual C masks — see recGblResetAlarms above.
3157 for &(field, mask) in &alarm_posts {
3158 instance.notify_field(field, mask);
3159 }
3160 }
3161
3162 // Snapshot source PUTF + put-notify wait-set for processTarget /
3163 // dbNotifyAdd propagation (see `write_db_link_value` doc). For the
3164 // async-completion path PUTF would have been set when the put
3165 // landed on the record; it (and wait-set membership) must
3166 // propagate through the (now-completing) OUT / FLNK chain so an
3167 // async target reached here also defers WRITE_NOTIFY completion.
3168 // The committed alarm propagates the same way for
3169 // `recGblInheritSevrMsg` MS-class inheritance.
3170 let (src_putf, src_notify, src_alarm) = {
3171 let guard = rec.read().await;
3172 (
3173 guard.common.putf,
3174 guard.notify.clone(),
3175 super::links::LinkAlarm {
3176 stat: guard.common.stat,
3177 sevr: guard.common.sevr,
3178 amsg: guard.common.amsg.clone(),
3179 },
3180 )
3181 };
3182
3183 // OUT link — DB *or* external `ca://`/`pva://`. Same scheme
3184 // dispatch as the sync path (C `dbLink.c::dbPutLink`,
3185 // dbLink.c:434-448).
3186 if let Some((link, out_val)) = out_info {
3187 self.write_out_link_value(
3188 &link,
3189 out_val,
3190 super::links::OutLinkSrc {
3191 putf: src_putf,
3192 notify: src_notify.as_ref(),
3193 alarm: &src_alarm,
3194 },
3195 visited,
3196 depth,
3197 )
3198 .await;
3199 }
3200
3201 // Multi-output dispatch (fanout/dfanout/seq/sseq)
3202 self.dispatch_multi_output(&rec, visited, depth).await;
3203
3204 // event record: post the named software event.
3205 self.dispatch_event_record(&rec).await;
3206
3207 // Generic multi-output links (transform OUTA..OUTP -> A..P,
3208 // scalcout OUT->OVAL, epid OUTL).
3209 //
3210 // SINGLE-OWNER INVARIANT: skip any record type owned by
3211 // `dispatch_multi_output` (called above) so its `LNKn`/`OUTn`
3212 // is not dispatched twice — see the sync-path twin in
3213 // `run_forward_link_tail_with_putf` §4.6.
3214 {
3215 let multi_out = {
3216 let instance = rec.read().await;
3217 let links =
3218 if super::links::multi_output_dispatch_owned(instance.record.record_type()) {
3219 &[][..]
3220 } else {
3221 instance.record.multi_output_links()
3222 };
3223 if links.is_empty() {
3224 None
3225 } else {
3226 let mut pairs = Vec::new();
3227 for &(link_field, val_field) in links {
3228 let link_str = instance
3229 .record
3230 .get_field(link_field)
3231 .and_then(|v| {
3232 if let EpicsValue::String(s) = v {
3233 Some(s)
3234 } else {
3235 None
3236 }
3237 })
3238 .unwrap_or_default();
3239 if link_str.is_empty() {
3240 continue;
3241 }
3242 if let Some(val) = instance.record.get_field(val_field) {
3243 pairs.push((link_str, val));
3244 }
3245 }
3246 if pairs.is_empty() { None } else { Some(pairs) }
3247 }
3248 };
3249 if let Some(pairs) = multi_out {
3250 for (link_str, val) in pairs {
3251 // `multi_output_links` carries record OUT links
3252 // (sseq `LNKn`, scalcout `OUTn` — all `DBF_OUTLINK`):
3253 // a bare DB link is NPP (`dbDbLink.c:388`).
3254 // `parse_output_link_v2` applies the OUT-link-correct
3255 // NPP default; an external `ca://`/`pva://` link is
3256 // routed through the link set's `putValue` — see the
3257 // sync-path twin above.
3258 let parsed = crate::server::record::parse_output_link_v2(
3259 link_str.as_str_lossy().as_ref(),
3260 );
3261 self.write_out_link_value(
3262 &parsed,
3263 val,
3264 super::links::OutLinkSrc {
3265 putf: src_putf,
3266 notify: src_notify.as_ref(),
3267 alarm: &src_alarm,
3268 },
3269 visited,
3270 depth,
3271 )
3272 .await;
3273 }
3274 }
3275 }
3276
3277 // FLNK -- only process if target is Passive (C `dbScanFwdLink` ->
3278 // `dbScanPassive` -> `processTarget` propagates PUTF the same way
3279 // OUT links do).
3280 if let Some(ref flnk) = flnk_name {
3281 if let Some(target_rec) = self.get_record(flnk).await {
3282 let (target_scan, should_process) = {
3283 let mut tg = target_rec.write().await;
3284 let pact = tg.is_processing();
3285 let on_chain = visited.contains(flnk);
3286 let scan = tg.common.scan;
3287 if !pact {
3288 tg.common.putf = src_putf;
3289 // C `dbNotifyAdd` (dbDbLink.c:460) is reached only
3290 // inside `processTarget`, which `dbScanPassive`
3291 // calls solely for a passive target. Gate the join
3292 // on the same passive condition as the process
3293 // call below so a dropped (non-passive) target
3294 // never `enter`s the wait-set without `leave`ing.
3295 if scan == crate::server::record::ScanType::Passive {
3296 join_put_notify(&mut tg, src_notify.as_ref());
3297 }
3298 } else if src_putf && !on_chain {
3299 tg.common.rpro = true;
3300 tg.common.putf = false;
3301 }
3302 (scan, !pact)
3303 };
3304 if should_process && target_scan == crate::server::record::ScanType::Passive {
3305 // recursive FLNK within one chain — gate
3306 // already held by the foreign entry record.
3307 let _ = self
3308 .process_record_with_links_recursive(flnk, visited, depth + 1)
3309 .await;
3310 }
3311 }
3312 }
3313
3314 // FLNK whose target is external (`pva://`/`ca://`): forwarded
3315 // through the same single owner as the synchronous tail (C
3316 // `dbScanFwdLink` → lset `scanForward`). `flnk_name` above only
3317 // names a local DB target.
3318 self.dispatch_external_forward_link(&rec).await;
3319
3320 // CP link targets
3321 self.dispatch_cp_targets(name, visited, depth).await;
3322
3323 // RPRO: C `recGblFwdLink` consumes a pending reprocess via
3324 // `scanOnce` — queued, not recursed. Mirror the synchronous
3325 // path: spawn a fresh process pass (clean `visited`, depth 0).
3326 {
3327 let needs_rpro = {
3328 let mut guard = rec.write().await;
3329 if guard.common.rpro {
3330 guard.common.rpro = false;
3331 true
3332 } else {
3333 false
3334 }
3335 };
3336 if needs_rpro {
3337 let db = self.clone();
3338 let rpro_name = name.to_string();
3339 crate::runtime::task::spawn(async move {
3340 let mut fresh_visited = std::collections::HashSet::new();
3341 let _ = db
3342 .process_record_with_links(&rpro_name, &mut fresh_visited, 0)
3343 .await;
3344 });
3345 }
3346 }
3347
3348 // C `recGbl.c::recGblFwdLink:302` clears `putf = FALSE` after
3349 // the forward-link dispatch. The same clearing must happen
3350 // at the tail of the async-completion path (this is the moral
3351 // equivalent of the synchronous completion path in
3352 // `put_record_field_from_ca` which clears after
3353 // `process_record_with_links` returns). Without this, a
3354 // record that completed an async write triggered by a
3355 // CA put would keep `putf=1` forever, leaking into every
3356 // subsequent scan-driven process cycle.
3357 {
3358 let mut guard = rec.write().await;
3359 guard.common.putf = false;
3360 }
3361
3362 // Put-notify completion: the async device round-trip is done and
3363 // the full OUT/FLNK/process-action tail above has run, so every PP
3364 // target it drove has joined the wait-set. The originating record
3365 // now `leave`s; the completion oneshot fires on the `leave` that
3366 // empties the set (i.e. once every joined async target has also
3367 // completed). `complete_put_notify` `take`s the membership, so a
3368 // motor re-entering `complete_async_record_inner` over several
3369 // device cycles leaves exactly once — matching the old fire site,
3370 // which `take`d its oneshot.
3371 {
3372 let mut guard = rec.write().await;
3373 complete_put_notify(&mut guard);
3374 }
3375
3376 Ok(())
3377 }
3378
3379 /// Dispatch CP-link targets that take a CP/CPP input link from `name`.
3380 ///
3381 /// C parity (a4bc0db): the CP-driven dispatch is the moral equivalent of
3382 /// dbCaTask's CA_DBPROCESS handler invoking `db_process(prec)`. Before
3383 /// processing each target, set PUTF=true; if the target is already
3384 /// processing (async record mid-flight), set RPRO=true instead so the
3385 /// in-flight pass reprocesses on completion. Already-visited targets
3386 /// (current process chain) are skipped via the `visited` cycle guard.
3387 async fn dispatch_cp_targets(
3388 &self,
3389 name: &str,
3390 visited: &mut std::collections::HashSet<String>,
3391 depth: usize,
3392 ) {
3393 let cp_targets = self.get_cp_targets(name).await;
3394 for target in cp_targets {
3395 self.process_one_cp_target(&target, visited, depth).await;
3396 }
3397 }
3398
3399 /// Process a single CP/CPP target edge, applying the CPP passive gate
3400 /// and the PACT/RPRO pre-check. This is the single owner of the
3401 /// scan-time CP-dispatch decision, shared by the local-source path
3402 /// ([`Self::dispatch_cp_targets`]) and the cross-IOC path
3403 /// ([`Self::dispatch_external_cp_targets`]) so both honour the same
3404 /// `dbCa.c` semantics.
3405 async fn process_one_cp_target(
3406 &self,
3407 target: &super::CpTarget,
3408 visited: &mut std::collections::HashSet<String>,
3409 depth: usize,
3410 ) {
3411 if visited.contains(&target.record) {
3412 return;
3413 }
3414 let target_rec = {
3415 let records = self.inner.records.read().await;
3416 records.get(&target.record).cloned()
3417 };
3418 let mut skip = false;
3419 if let Some(ref t) = target_rec {
3420 let mut tg = t.write().await;
3421 if target.passive_only && tg.common.scan != crate::server::record::ScanType::Passive {
3422 // CPP gate (`dbCa.c:854,994,1072`): a CPP link adds
3423 // `CA_DBPROCESS` only when the link-holder's SCAN is
3424 // Passive. A non-Passive target is reached by its own
3425 // periodic/event scan, so skip it here — no process,
3426 // no RPRO. A CP link (`passive_only == false`) never
3427 // takes this branch and always processes.
3428 skip = true;
3429 } else if tg.processing.load(std::sync::atomic::Ordering::Acquire) {
3430 tg.common.rpro = true;
3431 skip = true;
3432 }
3433 // else (not processing): fall through and process below.
3434 // epics-base PR #3fb10b6: PUTF must remain false on
3435 // CP-driven targets — only the record directly receiving
3436 // the dbPut reports PUTF=1 to dbNotify/onChange observers,
3437 // so we deliberately do NOT set PUTF here.
3438 }
3439 if skip {
3440 return;
3441 }
3442 // recursive CP-target fan-out within one chain —
3443 // gate already held by the foreign entry record.
3444 let _ = self
3445 .process_record_with_links_recursive(&target.record, visited, depth + 1)
3446 .await;
3447 }
3448
3449 /// Process every holder of an EXTERNAL CP/CPP link to `external_pv` —
3450 /// the cross-IOC twin of [`Self::dispatch_cp_targets`]. Called by the
3451 /// calink/pvalink CA monitor callback on every remote change, this is
3452 /// the Rust equivalent of C `dbCa.c eventCallback` adding
3453 /// `CA_DBPROCESS` for a CP (or Passive CPP) link (`dbCa.c:993-994`)
3454 /// and the worker thread running `db_process(prec)` (`dbCa.c:1295`).
3455 /// A cross-IOC source never processes locally, so this callback is the
3456 /// only trigger; without it a `CP`/`CPP` link's holder never processes
3457 /// on a remote change.
3458 ///
3459 /// A fresh `visited` set and `depth = 0` start a new process chain —
3460 /// the monitor event is an independent external trigger, like a scan,
3461 /// not a continuation of an in-flight local chain.
3462 pub async fn dispatch_external_cp_targets(&self, external_pv: &str) {
3463 let targets = self.get_external_cp_targets(external_pv).await;
3464 if targets.is_empty() {
3465 return;
3466 }
3467 let mut visited = std::collections::HashSet::new();
3468 for target in targets {
3469 self.process_one_cp_target(&target, &mut visited, 0).await;
3470 }
3471 }
3472
3473 /// Write a simulation value to an output record's SIOL link,
3474 /// dispatching by link type and locality exactly as C `dbPutLink`
3475 /// (reached from `writeValue` for a SIMM-mode output record):
3476 ///
3477 /// - a **local DB** target uses the already-locked write — writing
3478 /// VAL is an internal step of this record's processing chain,
3479 /// which already holds the entry record's advisory write gate, so
3480 /// a SIOL pointing back at a chain record must not re-acquire the
3481 /// non-reentrant gate (same reasoning as `write_db_link_value`);
3482 /// - a **non-local DB** target (`dbInitLink` made it a CA link) and
3483 /// an explicit **`Ca`/`Pva`** link route through the lset put path;
3484 /// - constant / hardware / none SIOL targets are not writable — no-op
3485 /// (C `dbPutLink` -> `S_db_noLSET`).
3486 async fn write_sim_siol_value(
3487 &self,
3488 siol: &crate::server::record::ParsedLink,
3489 value: EpicsValue,
3490 ) {
3491 match siol {
3492 crate::server::record::ParsedLink::Db(link) => {
3493 let pv_name = if link.field == "VAL" {
3494 link.record.clone()
3495 } else {
3496 format!("{}.{}", link.record, link.field)
3497 };
3498 if self.has_name_no_resolve(&link.record).await {
3499 let _ = self.put_pv_already_locked(&pv_name, value).await;
3500 } else if let Err(e) = self
3501 .write_external_pv(&pv_name, value, crate::server::database::LinkPutOp::Plain)
3502 .await
3503 {
3504 eprintln!("SIOL simulation write to external PV '{pv_name}' failed: {e}");
3505 }
3506 }
3507 crate::server::record::ParsedLink::Ca(_)
3508 | crate::server::record::ParsedLink::Pva(_)
3509 | crate::server::record::ParsedLink::PvaJson(_) => {
3510 let name = siol
3511 .external_pv_name()
3512 .expect("Ca/Pva/PvaJson link carries a PV name");
3513 if let Err(e) = self
3514 .write_external_pv(&name, value, crate::server::database::LinkPutOp::Plain)
3515 .await
3516 {
3517 eprintln!("SIOL simulation write to external PV '{name}' failed: {e}");
3518 }
3519 }
3520 _ => {}
3521 }
3522 }
3523
3524 /// Check simulation mode for a record. Returns
3525 /// `SimOutcome::Simulated` when simulation handled the value (the
3526 /// caller must still run the forward-link tail), or
3527 /// `SimOutcome::NotSimulated` when normal processing should proceed.
3528 async fn check_simulation_mode(&self, rec: &Arc<RwLock<RecordInstance>>) -> SimOutcome {
3529 // Read SIML, SIMM, SIOL, SIMS from the record
3530 let (siml_link, siol_link, sims, _rtype, is_input) = {
3531 let instance = rec.read().await;
3532 let rtype = instance.record.record_type().to_string();
3533 // Every input record whose DBD declares SIML/SIOL/SIMM/SIMS.
3534 // `mbbi`/`mbbiDirect` are input records: `mbbiRecord.c:125-126`
3535 // (and mbbiDirectRecord.c) declare SIML+SIOL, and
3536 // `mbbiRecord.c:388-394` reads `dbGetLink(&prec->siol,
3537 // DBR_ULONG, &prec->sval)` then `rval = sval` — input
3538 // semantics. Omitting them sent a simulated mbbi down the
3539 // OUTPUT branch, which writes VAL out to SIOL instead of
3540 // reading the value in from it.
3541 let is_input = matches!(
3542 rtype.as_str(),
3543 "ai" | "bi"
3544 | "mbbi"
3545 | "mbbiDirect"
3546 | "longin"
3547 | "int64in"
3548 | "stringin"
3549 | "lsi"
3550 | "event"
3551 );
3552
3553 let siml = instance
3554 .record
3555 .get_field("SIML")
3556 .and_then(|v| {
3557 if let EpicsValue::String(s) = v {
3558 Some(s)
3559 } else {
3560 None
3561 }
3562 })
3563 .unwrap_or_default();
3564 let siol = instance
3565 .record
3566 .get_field("SIOL")
3567 .and_then(|v| {
3568 if let EpicsValue::String(s) = v {
3569 Some(s)
3570 } else {
3571 None
3572 }
3573 })
3574 .unwrap_or_default();
3575 let sims = instance
3576 .record
3577 .get_field("SIMS")
3578 .and_then(|v| {
3579 if let EpicsValue::Short(s) = v {
3580 Some(s)
3581 } else {
3582 None
3583 }
3584 })
3585 .unwrap_or(0);
3586
3587 if siml.is_empty() && siol.is_empty() {
3588 return SimOutcome::NotSimulated; // No simulation configured
3589 }
3590
3591 let siml_parsed = crate::server::record::parse_link_v2(siml.as_str_lossy().as_ref());
3592 let siol_parsed = crate::server::record::parse_link_v2(siol.as_str_lossy().as_ref());
3593
3594 (siml_parsed, siol_parsed, sims, rtype, is_input)
3595 };
3596
3597 // Read SIML -> update SIMM. C `dbGetLink(&prec->siml, DBR_USHORT,
3598 // &prec->simm, 0, 0)` reads the SIML link for any type; the
3599 // pre-fix port only read a `ParsedLink::Db` SIML, ignoring a
3600 // CA/PVA/constant simulation-mode source.
3601 if let Some(val) = self.read_link_value_no_process(&siml_link).await {
3602 let simm_val = val.to_f64().unwrap_or(0.0) as i16;
3603 let mut instance = rec.write().await;
3604 let _ = instance
3605 .record
3606 .put_field("SIMM", EpicsValue::Short(simm_val));
3607 }
3608
3609 // Check SIMM
3610 let simm = {
3611 let instance = rec.read().await;
3612 instance
3613 .record
3614 .get_field("SIMM")
3615 .and_then(|v| {
3616 if let EpicsValue::Short(s) = v {
3617 Some(s)
3618 } else {
3619 None
3620 }
3621 })
3622 .unwrap_or(0)
3623 };
3624
3625 if simm == 0 {
3626 return SimOutcome::NotSimulated; // NO simulation, proceed normally
3627 }
3628
3629 // epics-base 7.0.7 (SIMM menu):
3630 // 1 = YES — read/write via SIOL using the cooked VAL
3631 // 2 = RAW — read/write via SIOL using the raw RVAL when the
3632 // record carries one (ai/ao only); falls back to
3633 // VAL when no RVAL is present. Mirrors the C
3634 // implementation, which treats records lacking
3635 // a raw value as "YES" since there's nothing
3636 // else to copy.
3637 let raw_mode = simm == 2;
3638 let raw_field = if raw_mode { "RVAL" } else { "VAL" };
3639
3640 // SIMM=YES(1) / SIMM=RAW(2): read/write the SIOL link. C
3641 // `readValue`/`writeValue` for a SIMM-mode record go through
3642 // `dbGetLink`/`dbPutLink`, which dispatch by link type — a local
3643 // DB target, a CA target (a bare non-local name or an explicit
3644 // `CA`/`ca://` link), or a constant. The pre-fix port special-
3645 // cased a local `ParsedLink::Db` SIOL only, so a non-local or
3646 // external SIOL neither read nor wrote yet still returned
3647 // `Simulated` — the record froze with no value and no alarm.
3648 // Dispatch uniformly through the same link read/write owners as
3649 // every other link; the alarm/timestamp/notify tail below now
3650 // runs for every SIOL link type.
3651 {
3652 if is_input {
3653 // Input record: read from SIOL -> set VAL/RVAL. Uniform
3654 // across Db (with locality fallback) / Ca / Pva / constant
3655 // via `read_link_value_no_process` (C `dbGetLink`).
3656 if let Some(siol_val) = self.read_link_value_no_process(&siol_link).await {
3657 let mut instance = rec.write().await;
3658 let target_supports_raw =
3659 raw_mode && instance.record.get_field("RVAL").is_some();
3660 if target_supports_raw {
3661 // PR #ac92e3e follow-up: SIMM=RAW on records
3662 // with RVAL (ai/ao/etc.) writes the raw value
3663 // into RVAL and runs the record's own
3664 // process() so the LINR / ESLO / EOFF / ASLO
3665 // / AOFF conversion chain computes VAL. The
3666 // pre-fix path additionally called set_val
3667 // here, which overwrote VAL with the raw
3668 // count and silently bypassed conversion —
3669 // the visible failure mode was "SIMM=RAW
3670 // simulation returns counts instead of EGU".
3671 //
3672 // Coerce to RVAL's native DBR type before
3673 // put_field — ai.RVAL is Long, but SIOL on a
3674 // soft channel typically yields Double. Without
3675 // the coerce step the put_field rejects with
3676 // TypeMismatch and leaves RVAL at 0, so
3677 // process() computes VAL = 0*ESLO + EOFF
3678 // (the offset only), not the intended
3679 // RAW*ESLO + EOFF.
3680 let rval_type = instance
3681 .record
3682 .field_list()
3683 .iter()
3684 .find(|f| f.name == "RVAL")
3685 .map(|f| f.dbf_type)
3686 .unwrap_or(crate::types::DbFieldType::Long);
3687 // C parity (aiRecord.c:495): `rval = (long)floor(sval)`.
3688 // Rust `convert_to(Long)` truncates toward zero,
3689 // diverging for negative bipolar-ADC raw values
3690 // (sval=-1.5 → C: -2, Rust as-cast: -1).
3691 // Floor explicitly when narrowing a float to
3692 // an integer RVAL.
3693 let coerced = match (&siol_val, rval_type) {
3694 (EpicsValue::Double(d), crate::types::DbFieldType::Long) => {
3695 EpicsValue::Long(d.floor() as i32)
3696 }
3697 (EpicsValue::Double(d), crate::types::DbFieldType::Int64) => {
3698 EpicsValue::Int64(d.floor() as i64)
3699 }
3700 (EpicsValue::Float(d), crate::types::DbFieldType::Long) => {
3701 EpicsValue::Long((*d as f64).floor() as i32)
3702 }
3703 (EpicsValue::Float(d), crate::types::DbFieldType::Int64) => {
3704 EpicsValue::Int64((*d as f64).floor() as i64)
3705 }
3706 _ if siol_val.db_field_type() != rval_type => {
3707 siol_val.convert_to(rval_type)
3708 }
3709 _ => siol_val,
3710 };
3711 let _ = instance.record.put_field("RVAL", coerced);
3712 let ctx = instance.common.process_context();
3713 instance.record.set_process_context(&ctx);
3714 let _ = instance.record.process();
3715 } else {
3716 // Records without RVAL fall back to SIMM=YES
3717 // semantics: the SIOL value goes straight into
3718 // VAL; no conversion to run.
3719 let _ = instance.record.set_val(siol_val);
3720 }
3721 apply_timestamp(&mut instance.common, true);
3722 instance.common.udf = false;
3723
3724 // Simulation alarm + alarm tail. C `aiRecord.c`
3725 // (and every soft-input record): `readValue()` raises
3726 // `recGblSetSevr(prec, SIMM_ALARM, prec->sims)` —
3727 // MAXIMIZE into the pending nsta/nsev, not a direct
3728 // commit — and then `process()` ALWAYS runs
3729 // `checkAlarms` + `recGblResetAlarms` even for a
3730 // simulated value, so the sim VAL still trips its own
3731 // limit/state alarms and the SIMM severity maximizes
3732 // against them. Set SIMM_ALARM first so it wins
3733 // severity ties (C order: readValue before checkAlarms).
3734 let sev = crate::server::record::AlarmSeverity::from_u16(sims as u16);
3735 crate::server::recgbl::rec_gbl_set_sevr(
3736 &mut instance.common,
3737 crate::server::recgbl::alarm_status::SIMM_ALARM,
3738 sev,
3739 );
3740 {
3741 let inst = &mut *instance;
3742 inst.record.check_alarms(&mut inst.common);
3743 }
3744 instance.evaluate_alarms();
3745 let _ = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
3746
3747 // Build snapshot and notify
3748 let mut changed_fields = Vec::new();
3749 if let Some(val) = instance.record.val() {
3750 changed_fields.push(("VAL".to_string(), val));
3751 }
3752 changed_fields.push((
3753 "SEVR".to_string(),
3754 EpicsValue::Short(instance.common.sevr as i16),
3755 ));
3756 changed_fields.push((
3757 "STAT".to_string(),
3758 EpicsValue::Short(instance.common.stat as i16),
3759 ));
3760 let snapshot = crate::server::record::ProcessSnapshot {
3761 changed_fields,
3762 event_mask: crate::server::recgbl::EventMask::VALUE
3763 | crate::server::recgbl::EventMask::ALARM,
3764 };
3765 instance.notify_from_snapshot(&snapshot);
3766 }
3767 } else {
3768 // Output record: write VAL (or RVAL for SIMM=RAW) to
3769 // SIOL (skip device write).
3770 let out_val = {
3771 let instance = rec.read().await;
3772 if raw_mode {
3773 // RAW path: prefer RVAL when the record has
3774 // one. Otherwise fall through to VAL.
3775 instance
3776 .record
3777 .get_field(raw_field)
3778 .or_else(|| instance.record.val())
3779 } else {
3780 instance.record.val()
3781 }
3782 };
3783 if let Some(val) = out_val {
3784 // Write VAL to the SIOL target, dispatching by link
3785 // type/locality (C `dbPutLink`). A local DB target
3786 // uses the `_already_locked` write — writing VAL is an
3787 // internal step of this record's processing chain,
3788 // which already holds the entry record's advisory
3789 // write gate, so a SIOL that points back at a chain
3790 // record cannot dead-lock on a non-reentrant gate
3791 // (same reasoning as the OUT-link write in
3792 // `write_db_link_value`). A non-local or external
3793 // SIOL routes through the lset put path.
3794 self.write_sim_siol_value(&siol_link, val).await;
3795 }
3796
3797 let mut instance = rec.write().await;
3798 apply_timestamp(&mut instance.common, true);
3799 instance.common.udf = false;
3800
3801 // Simulation alarm + alarm tail (same C parity as the
3802 // input branch above): maximize SIMM_ALARM into the
3803 // pending state, then run the record's checkAlarms and
3804 // recGblResetAlarms so the sim output value trips its own
3805 // alarms and the SIMM severity maximizes against them.
3806 let sev = crate::server::record::AlarmSeverity::from_u16(sims as u16);
3807 crate::server::recgbl::rec_gbl_set_sevr(
3808 &mut instance.common,
3809 crate::server::recgbl::alarm_status::SIMM_ALARM,
3810 sev,
3811 );
3812 {
3813 let inst = &mut *instance;
3814 inst.record.check_alarms(&mut inst.common);
3815 }
3816 instance.evaluate_alarms();
3817 let _ = crate::server::recgbl::rec_gbl_reset_alarms(&mut instance.common);
3818
3819 // Notify subscribers of simulation output
3820 let mut changed_fields = Vec::new();
3821 if let Some(val) = instance.record.val() {
3822 changed_fields.push(("VAL".to_string(), val));
3823 }
3824 changed_fields.push((
3825 "SEVR".to_string(),
3826 EpicsValue::Short(instance.common.sevr as i16),
3827 ));
3828 changed_fields.push((
3829 "STAT".to_string(),
3830 EpicsValue::Short(instance.common.stat as i16),
3831 ));
3832 let snapshot = crate::server::record::ProcessSnapshot {
3833 changed_fields,
3834 event_mask: crate::server::recgbl::EventMask::VALUE
3835 | crate::server::recgbl::EventMask::ALARM,
3836 };
3837 instance.notify_from_snapshot(&snapshot);
3838 }
3839 }
3840
3841 SimOutcome::Simulated
3842 }
3843}