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