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