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