Skip to main content

epics_base_rs/server/database/
field_io.rs

1use std::collections::HashSet;
2
3use crate::error::{CaError, CaResult};
4use crate::server::snapshot::Snapshot;
5use crate::types::EpicsValue;
6
7use super::PvDatabase;
8
9/// Coerce a write `value` to a record field's stored `target` type. A
10/// `DBR_STRING` write to a `DBF_MENU` field resolves the label against THAT
11/// field's own menu first (C `dbConvert` `putStringMenu`: an exact label,
12/// then a numeric index) — never the field-blind global table that
13/// `EpicsValue::convert_to` would otherwise consult, which mis-maps a
14/// menu-specific label (e.g. `SELM "Specified"`). Every other value, and a
15/// string that names no choice, falls through to the single value-coercion
16/// owner `EpicsValue::convert_to`.
17fn coerce_write_value(
18    record: &dyn crate::server::record::Record,
19    field: &str,
20    target: crate::types::DbFieldType,
21    value: EpicsValue,
22) -> EpicsValue {
23    if let EpicsValue::String(s) = &value {
24        if let Some(choices) = record
25            .menu_field_choices(field)
26            .or_else(|| crate::server::record::shared_menu_choices(field))
27        {
28            let s = s.as_str_lossy();
29            if let Some(resolved) =
30                crate::server::record::resolve_menu_field_string(choices, target, &s)
31            {
32                return resolved;
33            }
34        }
35    }
36    value.convert_to(target)
37}
38
39impl PvDatabase {
40    /// Get a PV value synchronously from a blocking thread.
41    ///
42    /// Uses `block_in_place` + `Handle::block_on` to bridge the async
43    /// `get_pv` call. Safe to call from std::threads spawned within
44    /// a tokio runtime context.
45    pub fn get_pv_blocking(&self, name: &str) -> CaResult<EpicsValue> {
46        let db = self.clone();
47        let name = name.to_string();
48        if crate::runtime::task::RuntimeHandle::try_current().is_ok() {
49            crate::__tokio::task::block_in_place(|| {
50                crate::runtime::task::RuntimeHandle::current().block_on(db.get_pv(&name))
51            })
52        } else {
53            Err(CaError::InvalidValue(
54                "no runtime for get_pv_blocking".into(),
55            ))
56        }
57    }
58
59    /// Get the current value of a PV or record field.
60    /// Uses resolve_field for records (3-level priority).
61    pub async fn get_pv(&self, name: &str) -> CaResult<EpicsValue> {
62        let (base, field) = super::parse_pv_name(name);
63        let field = field.to_ascii_uppercase();
64
65        // Check simple PVs first (exact match)
66        if let Some(pv) = self.inner.simple_pvs.read().await.get(name) {
67            return Ok(pv.get().await);
68        }
69
70        // Records — alias-aware via `get_record` (epics-base PR #336).
71        if let Some(rec) = self.get_record(base).await {
72            let instance = rec.read().await;
73            return instance
74                .resolve_field(&field)
75                .ok_or_else(|| CaError::ChannelNotFound(name.to_string()));
76        }
77
78        Err(CaError::ChannelNotFound(name.to_string()))
79    }
80
81    /// Set a PV value or record field, notifying subscribers.
82    /// Tries record put_field first, then put_common_field as fallback.
83    ///
84    /// Acquires the record's advisory write gate.
85    pub async fn put_pv(&self, name: &str, value: EpicsValue) -> CaResult<()> {
86        self.put_pv_inner(name, value, true).await
87    }
88
89    /// `put_pv` variant for a caller already holding the
90    /// record's advisory write gate (QSRV atomic group PUT). See
91    /// [`Self::put_record_field_from_ca_already_locked`].
92    pub async fn put_pv_already_locked(&self, name: &str, value: EpicsValue) -> CaResult<()> {
93        self.put_pv_inner(name, value, false).await
94    }
95
96    /// C `IOCSource::doPreProcessing` gate (pvxs `iocsource.cpp:363-375`).
97    ///
98    /// Reject an *external* put (a PVA/CA client put routed through QSRV)
99    /// that C refuses before any write: a put to a `DISP=1` record's
100    /// non-DISP field (`S_db_putDisabled`) or to a read-only / `SPC_NOMOD`
101    /// field (`S_db_noMod`). No value is written — this is a precondition
102    /// check only. It mirrors the two gates inside
103    /// [`Self::put_record_field_from_ca`] (the Passive route) so the QSRV
104    /// `Force`/`Inhibit` routes — which go through [`Self::put_pv`] — enforce
105    /// the same preconditions. `put_pv` itself is the internal `dbPut`
106    /// analogue and deliberately does not gate DISP (internal
107    /// link/processing puts must bypass it), so the gate lives at the
108    /// external put boundary, exactly as C places `doPreProcessing` in the
109    /// source layer rather than in `dbPut`.
110    pub async fn check_external_put_preconditions(
111        &self,
112        record_name: &str,
113        field: &str,
114    ) -> CaResult<()> {
115        let field_upper = field.to_ascii_uppercase();
116        // A missing record is not a DISP/read-only precondition violation:
117        // stay silent and let the downstream put report the not-found (for
118        // QSRV, inside its own `asTrapWrite` bracket). C's `doPreProcessing`
119        // only runs against an established channel — the record is
120        // guaranteed present there — and a `BridgeChannel` likewise always
121        // binds a real record in production.
122        let Some(rec) = self.get_record(record_name).await else {
123            return Ok(());
124        };
125        let instance = rec.read().await;
126        // DISP=1 blocks a put to any field except DISP itself
127        // (C: `precord->disp && pfield != &precord->disp` → S_db_putDisabled).
128        if instance.common.disp && field_upper != "DISP" {
129            return Err(CaError::PutDisabled(field_upper));
130        }
131        // Read-only / SPC_NOMOD field
132        // (C: `special == SPC_ATTRIBUTE` → S_db_noMod) — same `read_only`
133        // flag the Passive route checks, so all three modes gate uniformly.
134        let is_read_only = instance
135            .record
136            .field_list()
137            .iter()
138            .find(|f| f.name.eq_ignore_ascii_case(&field_upper))
139            .is_some_and(|f| f.read_only);
140        if is_read_only {
141            return Err(CaError::ReadOnlyField(field_upper));
142        }
143        Ok(())
144    }
145
146    async fn put_pv_inner(
147        &self,
148        name: &str,
149        value: EpicsValue,
150        acquire_gate: bool,
151    ) -> CaResult<()> {
152        let (base, field) = super::parse_pv_name(name);
153        let field = field.to_ascii_uppercase();
154
155        // Check simple PVs first
156        if let Some(pv) = self.inner.simple_pvs.read().await.get(name) {
157            pv.set(value).await;
158            return Ok(());
159        }
160
161        // Records — alias-aware (epics-base PR #336).
162        if let Some(rec) = self.get_record(base).await {
163            // `base` may be an alias; resolve to the canonical record
164            // name so scan-index updates target the right entry.
165            let canonical_base: String = self
166                .resolve_alias(base)
167                .await
168                .unwrap_or_else(|| base.to_string());
169            // advisory write gate (`dbScanLock` analogue) so a
170            // plain `put_pv` to a backing record cannot interleave
171            // with an atomic group transaction holding the same gate.
172            // Skipped when the caller already owns the gate.
173            let _record_gate = if acquire_gate {
174                Some(self.lock_record(&canonical_base).await)
175            } else {
176                None
177            };
178            let mut instance = rec.write().await;
179
180            // Coerce value to field's native type
181            let value = {
182                let target_type = instance
183                    .record
184                    .field_list()
185                    .iter()
186                    .find(|f| f.name.eq_ignore_ascii_case(&field))
187                    .map(|f| f.dbf_type);
188                if let Some(target) = target_type {
189                    if value.db_field_type() != target {
190                        // C EPICS dbPut (12cfd41): nRequest=0 into a scalar
191                        // field must NOT silently coerce. `convert_to` on an
192                        // empty array calls `to_f64().unwrap_or(0.0)` and
193                        // would produce a scalar zero — the same garbage-
194                        // value bug the C fix raised LINK_ALARM for.
195                        if value.is_empty_array() {
196                            return Err(CaError::InvalidValue(format!(
197                                "empty array cannot be coerced to scalar field {field}"
198                            )));
199                        }
200                        coerce_write_value(&*instance.record, &field, target, value)
201                    } else {
202                        value
203                    }
204                } else {
205                    value
206                }
207            };
208
209            // Capture the pre-put value so the metadata-cache
210            // invalidation (and the downstream `DBE_PROPERTY`
211            // emission) can be skipped when the put is a no-op —
212            // epics-base faac1df1.
213            let prev_value = instance.record.get_field(&field);
214
215            // put_pv is C EPICS dbPut: write value + special/on_put.
216            // Does NOT post monitor events (use put_pv_and_post for that).
217            // Does NOT clear UDF or trigger processing.
218            use crate::server::record::CommonFieldPutResult;
219            let common_result = match instance.record.put_field(&field, value.clone()) {
220                Ok(()) => {
221                    instance.record.on_put(&field);
222                    let _ = instance.record.special(&field, true);
223                    CommonFieldPutResult::NoChange
224                }
225                Err(CaError::FieldNotFound(_)) => instance.put_common_field(&field, value)?,
226                Err(e) => return Err(e),
227            };
228
229            // Invalidate metadata cache only if the metadata-class
230            // field's value actually changed (faac1df1).
231            instance.notify_field_written_if_changed(&field, prev_value.as_ref());
232
233            // Update scan index if SCAN or PHAS changed
234            match common_result {
235                CommonFieldPutResult::ScanChanged {
236                    old_scan,
237                    new_scan,
238                    phas,
239                } => {
240                    drop(instance);
241                    self.update_scan_index(&canonical_base, old_scan, new_scan, phas, phas)
242                        .await;
243                }
244                CommonFieldPutResult::PhasChanged {
245                    scan: s,
246                    old_phas,
247                    new_phas,
248                } => {
249                    drop(instance);
250                    self.update_scan_index(&canonical_base, s, s, old_phas, new_phas)
251                        .await;
252                }
253                CommonFieldPutResult::NoChange => {}
254            }
255
256            // mirror the CA-write path's ASG-field notifier so
257            // restore scripts / autosave / admin tools that go via
258            // `put_pv` (not `put_record_field_from_ca`) also trigger
259            // per-client `reeval_access_rights`. C `dbAccess.c::
260            // dbPutSpecial` invokes the SPC_AS callback from dbPut
261            // regardless of caller entry path.
262            if field == "ASG" {
263                crate::server::access_security::notify_asg_field_changed();
264            }
265
266            return Ok(());
267        }
268
269        Err(CaError::ChannelNotFound(name.to_string()))
270    }
271
272    /// Write a value and post monitor events if changed.
273    /// Equivalent to C EPICS `dbPut` + `db_post_events(DBE_VALUE|DBE_LOG)`.
274    ///
275    /// Use for readback/status mirror PVs that are written by sequencer-style
276    /// code and need to be visible to CA monitors without triggering record
277    /// processing. Clears UDF/UDF_ALARM on primary field write.
278    ///
279    /// `origin`: writer ID for self-write filtering. Subscribers with the
280    /// same `ignore_origin` will skip this event. Pass 0 to disable.
281    pub async fn put_pv_and_post(&self, name: &str, value: EpicsValue) -> CaResult<()> {
282        self.put_pv_and_post_with_origin(name, value, 0).await
283    }
284
285    /// Push a monitor event holding the simple PV's *current* value
286    /// but with explicit alarm severity/status. Used by the gateway
287    /// to surface upstream-disconnect to downstream monitor
288    /// subscribers without dropping the shadow PV (which would force
289    /// downstream clients into ECA_DISCONN reconnect storms on every
290    /// transient hiccup). Returns `ChannelNotFound` for record-backed
291    /// PVs — those carry their own `common.sevr/stat` in record
292    /// processing.
293    pub async fn post_alarm(&self, name: &str, severity: u16, status: u16) -> CaResult<()> {
294        if let Some(pv) = self.inner.simple_pvs.read().await.get(name).cloned() {
295            pv.post_alarm(severity, status).await;
296            return Ok(());
297        }
298        Err(crate::error::CaError::ChannelNotFound(name.to_string()))
299    }
300
301    /// Propagate a full upstream snapshot (value + alarm status/severity +
302    /// IOC timestamp) to a simple shadow PV and fan out to downstream
303    /// monitor subscribers. Used by the CA gateway forwarding task to avoid
304    /// discarding the upstream alarm and timestamp decoded from the incoming
305    /// `DBR_TIME_*` frame. Returns `ChannelNotFound` for record-backed PVs
306    /// (those carry their own alarm engine and are not shadow PVs).
307    pub async fn put_pv_and_post_snapshot(&self, name: &str, snapshot: Snapshot) -> CaResult<()> {
308        if let Some(pv) = self.inner.simple_pvs.read().await.get(name).cloned() {
309            pv.set_snapshot(snapshot).await;
310            return Ok(());
311        }
312        Err(CaError::ChannelNotFound(name.to_string()))
313    }
314
315    /// Install upstream `DBR_CTRL_*` metadata (display / control limits,
316    /// enum labels) on a shadow simple PV WITHOUT posting an event.
317    ///
318    /// The CA gateway calls this once on upstream connect, after its initial
319    /// `DBR_CTRL_*` get, so a later downstream `DBR_CTRL_*` / `DBR_GR_*` read
320    /// returns the real limits instead of zeroed ones. No `DBE_PROPERTY`
321    /// monitor event fires — nothing has *changed* yet, this only seeds the
322    /// attribute cache. Mirrors C `gatePvData::getCB` → `runDataCB` →
323    /// `vc->setPvData(dd)` (`gatePv.cc:1693-1695`), which seeds the property
324    /// cache from the initial control get in both cache modes before any
325    /// monitor is enabled.
326    ///
327    /// Returns `ChannelNotFound` for record-backed PVs — those own their own
328    /// metadata via record processing and are not gateway shadow PVs.
329    pub async fn set_pv_metadata(&self, name: &str, snapshot: &Snapshot) -> CaResult<()> {
330        if let Some(pv) = self.inner.simple_pvs.read().await.get(name).cloned() {
331            pv.set_metadata(metadata_from_snapshot(snapshot));
332            return Ok(());
333        }
334        Err(CaError::ChannelNotFound(name.to_string()))
335    }
336
337    /// Refresh a shadow simple PV's upstream metadata AND post a
338    /// `DBE_PROPERTY` monitor event carrying `snapshot` to downstream
339    /// property subscribers.
340    ///
341    /// `snapshot` is the decoded upstream `DBR_CTRL_*` property event: it
342    /// carries the control value and the upstream `status` / `severity`,
343    /// and (because control DBR structs carry no timestamp) an undefined
344    /// timestamp the caller must NOT replace with a fresh wall-clock. The
345    /// gateway's property monitor calls this on every upstream
346    /// `DBE_PROPERTY` event, mirroring C `gatePvData::propEventCB` →
347    /// `runDataCB` + `setPvData` + `runValueDataCB` +
348    /// `vcPostEvent(propertyEventMask())` (`gatePv.cc:1571-1607`): the
349    /// attribute cache is refreshed and a property event is posted with the
350    /// upstream alarm state preserved (`setStatSevr`) and the undefined
351    /// control-DBR timestamp left as-is (`gatePv.cc:1594-1595`).
352    ///
353    /// Returns `ChannelNotFound` for record-backed PVs.
354    pub async fn post_pv_property(&self, name: &str, snapshot: Snapshot) -> CaResult<()> {
355        if let Some(pv) = self.inner.simple_pvs.read().await.get(name).cloned() {
356            pv.set_metadata(metadata_from_snapshot(&snapshot));
357            pv.post_property(snapshot).await;
358            return Ok(());
359        }
360        Err(CaError::ChannelNotFound(name.to_string()))
361    }
362
363    /// Like `put_pv_and_post` but with explicit origin tag.
364    pub async fn put_pv_and_post_with_origin(
365        &self,
366        name: &str,
367        value: EpicsValue,
368        origin: u64,
369    ) -> CaResult<()> {
370        let (base, field) = super::parse_pv_name(name);
371        let field = field.to_ascii_uppercase();
372
373        // Simple-PV path: PVs registered via `add_pv` (e.g. CA gateway
374        // shadow PVs, IOCsh stats PVs) are stored in `simple_pvs`,
375        // not `records`. Without this branch the function would
376        // silently return `ChannelNotFound` for every gateway-mirrored
377        // PV — `ProcessVariable::set` already does the
378        // notify-subscribers fan-out internally so all we need here is
379        // to delegate. The `origin` tag is a no-op for simple PVs
380        // because they don't yet plumb origin through `set`.
381        if let Some(pv) = self.inner.simple_pvs.read().await.get(name).cloned() {
382            let _ = origin; // simple PVs don't currently honor origin tagging
383            pv.set(value).await;
384            return Ok(());
385        }
386
387        if let Some(rec) = self.get_record(base).await {
388            // `put_pv_and_post` is a public record-write API —
389            // it must take the same advisory write gate
390            // (`dbScanLock` analogue) as `put_pv` /
391            // `put_record_field_from_ca`, or a gateway/sequencer
392            // write through this helper can still land between the
393            // member writes of a QSRV atomic group or a pvalink
394            // atomic scan epoch holding `lock_records`. `base` is
395            // alias-resolved to the canonical record name so an alias
396            // and its target share one gate. Held until return.
397            let canonical_base: String = self
398                .resolve_alias(base)
399                .await
400                .unwrap_or_else(|| base.to_string());
401            let _record_gate = self.lock_record(&canonical_base).await;
402
403            let mut instance = rec.write().await;
404
405            // Type coercion
406            let value = {
407                let target_type = instance
408                    .record
409                    .field_list()
410                    .iter()
411                    .find(|f| f.name.eq_ignore_ascii_case(&field))
412                    .map(|f| f.dbf_type);
413                if let Some(target) = target_type {
414                    if value.db_field_type() != target {
415                        // C EPICS dbPut (12cfd41): empty-array → scalar
416                        // coercion would produce silent zero; reject.
417                        if value.is_empty_array() {
418                            return Err(CaError::InvalidValue(format!(
419                                "empty array cannot be coerced to scalar field {field}"
420                            )));
421                        }
422                        coerce_write_value(&*instance.record, &field, target, value)
423                    } else {
424                        value
425                    }
426                } else {
427                    value
428                }
429            };
430
431            let old_value = instance.record.get_field(&field);
432            let old_stat = instance.common.stat;
433            let old_sevr = instance.common.sevr;
434            // Snapshot side-effect-prone fields BEFORE the put. The
435            // array-family records (waveform/aai/aao/subArray) update
436            // NORD as a side-effect of put_field("VAL"); other record
437            // types return None for "NORD" and the comparison reduces
438            // to None==None → unchanged.
439            let old_nord = if field == "VAL" {
440                instance.record.get_field("NORD")
441            } else {
442                None
443            };
444
445            // Write value + special/on_put
446            match instance.record.put_field(&field, value.clone()) {
447                Ok(()) => {
448                    instance.record.on_put(&field);
449                    let _ = instance.record.special(&field, true);
450                    // Clear UDF/UDF_ALARM on primary field write
451                    if field == instance.record.primary_field() {
452                        instance.common.udf = false;
453                        if instance.common.stat == crate::server::recgbl::alarm_status::UDF_ALARM {
454                            instance.common.stat = 0;
455                            instance.common.sevr = crate::server::record::AlarmSeverity::NoAlarm;
456                        }
457                    }
458                }
459                Err(CaError::FieldNotFound(_)) => {
460                    instance.put_common_field(&field, value)?;
461                }
462                Err(e) => return Err(e),
463            }
464
465            // Invalidate metadata cache only if a metadata-class
466            // field actually changed value (faac1df1 — DBE_PROPERTY
467            // fires on real changes, not no-op writes).
468            instance.notify_field_written_if_changed(&field, old_value.as_ref());
469
470            // Post monitor events if value or alarm changed
471            let new_value = instance.record.get_field(&field);
472            let value_changed = old_value != new_value;
473            let alarm_changed =
474                old_stat != instance.common.stat || old_sevr != instance.common.sevr;
475            let new_nord = if field == "VAL" {
476                instance.record.get_field("NORD")
477            } else {
478                None
479            };
480            let nord_changed = field == "VAL" && old_nord != new_nord && new_nord.is_some();
481            if value_changed || alarm_changed || nord_changed {
482                // Update timestamp so the snapshot carries current time
483                instance.common.time = crate::runtime::general_time::get_current();
484                instance.cleanup_subscribers();
485                if value_changed || alarm_changed {
486                    instance.notify_field_with_origin(
487                        &field,
488                        crate::server::recgbl::EventMask::VALUE
489                            | crate::server::recgbl::EventMask::LOG
490                            | crate::server::recgbl::EventMask::ALARM,
491                        origin,
492                    );
493                }
494                // Surface the implicit NORD update to NORD subscribers
495                // for waveform/aai/aao/subArray. Without this, a CA
496                // gateway forwarding upstream waveform monitors via
497                // put_pv_and_post would update VAL on the shadow PV
498                // but leave downstream NORD subscribers stuck at their
499                // last seen length — a frozen-element-count bug that
500                // surfaces in PyDM image views and similar consumers
501                // that compute height = element_count / width.
502                if nord_changed {
503                    instance.notify_field_with_origin(
504                        "NORD",
505                        crate::server::recgbl::EventMask::VALUE
506                            | crate::server::recgbl::EventMask::LOG,
507                        origin,
508                    );
509                }
510            }
511
512            // same SPC_AS parity as `put_pv` / `put_pv_no_process`
513            // / the CA-write path — a gateway mirroring `.ASG` via
514            // `put_pv_and_post` must still trigger per-client
515            // re-eval.
516            if field == "ASG" {
517                crate::server::access_security::notify_asg_field_changed();
518            }
519
520            return Ok(());
521        }
522
523        Err(CaError::ChannelNotFound(name.to_string()))
524    }
525
526    /// CA client's unified entry point for record field put.
527    /// Handles DISP/PROC/PACT/LCNT checks, field put, device write, and Passive process.
528    ///
529    /// Acquires the record's advisory write gate
530    /// (`dbScanLock` analogue) for the duration of the write.
531    pub async fn put_record_field_from_ca(
532        &self,
533        record_name: &str,
534        field: &str,
535        value: EpicsValue,
536    ) -> CaResult<Option<crate::runtime::sync::oneshot::Receiver<()>>> {
537        self.put_record_field_from_ca_inner(record_name, field, value, true, true)
538            .await
539    }
540
541    /// Variant for a caller that already owns the target
542    /// record's advisory write gate — the QSRV atomic group PUT,
543    /// which acquired every member-record gate up-front via
544    /// [`Self::lock_records`]. The per-record `tokio::sync::Mutex`
545    /// gate is NOT reentrant, so the atomic group path MUST use this
546    /// `_already_locked` entry to avoid dead-locking on its own
547    /// `ManyRecordWriteGuard`.
548    pub async fn put_record_field_from_ca_already_locked(
549        &self,
550        record_name: &str,
551        field: &str,
552        value: EpicsValue,
553    ) -> CaResult<Option<crate::runtime::sync::oneshot::Receiver<()>>> {
554        self.put_record_field_from_ca_inner(record_name, field, value, false, true)
555            .await
556    }
557
558    /// Fire-and-forget variant — C `dbPutField` semantics: the put
559    /// processes the record but creates NO put-notify wait-set (C
560    /// builds a `putNotify` only in `dbPutNotify`, i.e. for
561    /// WRITE_NOTIFY). A caller that does not await the returned
562    /// receiver MUST use this entry: parking a wait-set whose receiver
563    /// is dropped occupies `RecordInstance::notify` until the record's
564    /// async work ends (a motor's whole motion), failing every
565    /// legitimate WRITE_NOTIFY on the record with ECA_PUTCBINPROG in
566    /// the meantime.
567    pub async fn put_record_field_from_ca_no_notify(
568        &self,
569        record_name: &str,
570        field: &str,
571        value: EpicsValue,
572    ) -> CaResult<()> {
573        self.put_record_field_from_ca_inner(record_name, field, value, true, false)
574            .await
575            .map(|_| ())
576    }
577
578    /// Fire-and-forget + caller-held gate: see
579    /// [`Self::put_record_field_from_ca_no_notify`] and
580    /// [`Self::put_record_field_from_ca_already_locked`].
581    pub async fn put_record_field_from_ca_no_notify_already_locked(
582        &self,
583        record_name: &str,
584        field: &str,
585        value: EpicsValue,
586    ) -> CaResult<()> {
587        self.put_record_field_from_ca_inner(record_name, field, value, false, false)
588            .await
589            .map(|_| ())
590    }
591
592    /// Process a record UNCONDITIONALLY with a put-notify wait-set, returning
593    /// the completion receiver — the QSRV `record[process=true,block=true]`
594    /// (Force + block) barrier.
595    ///
596    /// C `dbProcessNotify`: pvxs routes a blocking forced put through
597    /// `dbProcessNotify` (`singlesource.cpp:360-369`), whose completion fires
598    /// only after the record's whole processing chain — including async device
599    /// work (a motor move, an asyn-backed AO) — settles. The value is written
600    /// by the caller's preceding [`Self::put_pv`] (the `dbPut` analogue, no
601    /// process); this entry then mints the wait-set, registers it into the
602    /// record's `notify` slot so PACT records join it, and runs the full
603    /// unconditional [`Self::process_record_with_links`] cycle (C `dbProcess`,
604    /// the Force analogue). A fully synchronous chain returns `Ok(None)` (the
605    /// wait-set already drained); an async record returns `Ok(Some(rx))` for
606    /// the caller to await. A concurrent put-callback already in flight on the
607    /// record is rejected with `PutCallbackInProgress`, matching the PROC path.
608    pub async fn process_record_with_notify(
609        &self,
610        record_name: &str,
611    ) -> CaResult<Option<crate::runtime::sync::oneshot::Receiver<()>>> {
612        let (completion_tx, completion_rx) = crate::runtime::sync::oneshot::channel();
613        let notify = crate::server::record::NotifyWaitSet::new(completion_tx);
614        {
615            // Collect-then-act: clone the handle under a brief map read, drop
616            // the map lock before taking the per-record write lock.
617            let rec_arc = {
618                let recs = self.inner.records.read().await;
619                recs.get(record_name).cloned()
620            };
621            let Some(rec_arc) = rec_arc else {
622                return Err(CaError::ChannelNotFound(record_name.to_string()));
623            };
624            let mut guard = rec_arc.write().await;
625            if guard.notify.is_some() {
626                return Err(CaError::PutCallbackInProgress(record_name.to_string()));
627            }
628            guard.notify = Some(notify.clone());
629        }
630        let mut visited = HashSet::new();
631        self.process_record_with_links(record_name, &mut visited, 0)
632            .await?;
633        // The wait-set fires the oneshot only after the whole FLNK/OUT chain
634        // (sync + async) settles. Already-completed ⟹ fully synchronous ⟹
635        // report immediate success; otherwise hand the receiver back to await
636        // the deferred async completion.
637        if notify.completed() {
638            Ok(None)
639        } else {
640            Ok(Some(completion_rx))
641        }
642    }
643
644    async fn put_record_field_from_ca_inner(
645        &self,
646        record_name: &str,
647        field: &str,
648        value: EpicsValue,
649        acquire_gate: bool,
650        want_notify: bool,
651    ) -> CaResult<Option<crate::runtime::sync::oneshot::Receiver<()>>> {
652        let field = field.to_ascii_uppercase();
653
654        // Get record Arc — alias-aware (epics-base PR #336) so a CA
655        // client that connects via an alias name can put fields on
656        // the canonical record.
657        let rec = self
658            .get_record(record_name)
659            .await
660            .ok_or_else(|| CaError::ChannelNotFound(record_name.to_string()))?;
661        // Normalise to the canonical name for the rest of this
662        // function — every subsequent call (PACT/LCNT lookup,
663        // `process_record_with_links`, `update_scan_index`) uses the
664        // raw records map and would miss when `record_name` is an
665        // alias. Resolve once up front.
666        let canonical_owned;
667        let record_name: &str = if let Some(target) = self.resolve_alias(record_name).await {
668            canonical_owned = target;
669            &canonical_owned
670        } else {
671            record_name
672        };
673
674        // take the record's advisory write gate — the
675        // `dbScanLock(precord)` analogue. While a QSRV atomic group
676        // PUT/GET holds this record's gate via `lock_records`, this
677        // plain write blocks here, so a direct backing-record write
678        // can no longer land between member writes of an atomic group
679        // transaction. Held until the function returns. Skipped when
680        // the caller (atomic group PUT) already owns the gate — the
681        // gate `Mutex` is not reentrant.
682        let _record_gate = if acquire_gate {
683            Some(self.lock_record(record_name).await)
684        } else {
685            None
686        };
687
688        // Special field intercepts (read lock, then drop)
689        {
690            let instance = rec.read().await;
691            match field.as_str() {
692                "PACT" => return Err(CaError::ReadOnlyField("PACT".into())),
693                "LCNT" => return Err(CaError::ReadOnlyField("LCNT".into())),
694                "PUTF" => return Err(CaError::ReadOnlyField("PUTF".into())),
695                _ => {}
696            }
697
698            // PROC intercept: trigger processing regardless of DISP.
699            // Falls through to the put_notify_tx registration below
700            // so async records (motor, asyn-backed AO) signal real
701            // completion; otherwise WRITE_NOTIFY would return ECA_NORMAL
702            // before the device move actually finished.
703            if field == "PROC" {
704                // C `dbPutField` (dbAccess.c:1265) matches the proc field by
705                // pointer with NO value check: any write to PROC — including
706                // 0 — processes the record (when !pact). The standard
707                // `caput REC.PROC 0` / `dbpf REC.PROC 0` force-process idiom
708                // must therefore not be skipped for a zero value.
709                drop(instance);
710                // Continue to the put-notify setup + process below
711                // by jumping past the field-write step (the value
712                // itself isn't stored; PROC is a trigger). A
713                // fire-and-forget caller parks nothing — C `dbPutField`
714                // on PROC processes the record with no putNotify.
715                let parked = if want_notify {
716                    let (completion_tx, completion_rx) = crate::runtime::sync::oneshot::channel();
717                    let notify = crate::server::record::NotifyWaitSet::new(completion_tx);
718                    {
719                        // Collect-then-act: clone the handle under a brief map
720                        // read, drop the map lock before the per-record write.
721                        let rec_arc = {
722                            let recs = self.inner.records.read().await;
723                            recs.get(record_name).cloned()
724                        };
725                        if let Some(rec_arc) = rec_arc {
726                            let mut guard = rec_arc.write().await;
727                            if guard.notify.is_some() {
728                                return Err(CaError::PutCallbackInProgress(
729                                    record_name.to_string(),
730                                ));
731                            }
732                            guard.notify = Some(notify.clone());
733                        }
734                    }
735                    Some((notify, completion_rx))
736                } else {
737                    None
738                };
739                let mut visited = HashSet::new();
740                // this PROC trigger already holds `record_name`'s
741                // advisory write gate — either `_record_gate` above, or
742                // the QSRV atomic group's `lock_records` epoch when
743                // entered via `put_record_field_from_ca_already_locked`.
744                // The gate `Mutex` is not reentrant, so the processing
745                // call MUST use the `_already_locked` variant.
746                let _ = self
747                    .process_record_with_links_already_locked(record_name, &mut visited, 0)
748                    .await;
749                // The wait-set fires the oneshot only after the whole
750                // FLNK/OUT chain (sync + async) settles. If it has
751                // already completed the chain was fully synchronous —
752                // report immediate success; otherwise hand the receiver
753                // to the CA layer to await the deferred completion.
754                return match parked {
755                    Some((notify, completion_rx)) => {
756                        if notify.completed() {
757                            Ok(None)
758                        } else {
759                            Ok(Some(completion_rx))
760                        }
761                    }
762                    None => Ok(None),
763                };
764            }
765
766            // DISP check: block CA puts to non-DISP fields when DISP=1
767            if instance.common.disp && field != "DISP" {
768                return Err(CaError::PutDisabled(field));
769            }
770        }
771
772        // Normal field put (write lock)
773        let common_result = {
774            let mut instance = rec.write().await;
775            instance.common.putf = true;
776
777            // Coerce value to the field's native DBR type (e.g. String → Double for ao.VAL).
778            // This matches C EPICS db_put_field() which converts from the CA client's type
779            // to the record field's native type.
780            let value = {
781                let target_type = instance
782                    .record
783                    .field_list()
784                    .iter()
785                    .find(|f| f.name.eq_ignore_ascii_case(&field))
786                    .map(|f| f.dbf_type);
787                if let Some(target) = target_type {
788                    if value.db_field_type() != target {
789                        // C EPICS dbPut (12cfd41): empty-array → scalar
790                        // coercion would produce silent zero; reject.
791                        if value.is_empty_array() {
792                            instance.common.putf = false;
793                            return Err(CaError::InvalidValue(format!(
794                                "empty array cannot be coerced to scalar field {field}"
795                            )));
796                        }
797                        coerce_write_value(&*instance.record, &field, target, value)
798                    } else {
799                        value
800                    }
801                } else {
802                    value
803                }
804            };
805
806            // SPC_NOMOD: reject writes to read-only fields (C EPICS S_db_noMod)
807            let is_read_only = instance
808                .record
809                .field_list()
810                .iter()
811                .find(|f| f.name.eq_ignore_ascii_case(&field))
812                .is_some_and(|f| f.read_only);
813            if is_read_only {
814                instance.common.putf = false;
815                return Err(CaError::ReadOnlyField(field));
816            }
817
818            // Pre-write special hook (C EPICS dbPutSpecial pass=0)
819            if let Err(e) = instance.record.special(&field, false) {
820                instance.common.putf = false;
821                return Err(e);
822            }
823
824            // Capture pre-put value for faac1df1 idempotent-write suppression.
825            let prev_value = instance.record.get_field(&field);
826
827            // Try record-specific field first; fall back to common on FieldNotFound.
828            // For record-owned fields, call on_put() and special() after successful put,
829            // matching what put_common_field() does for common fields.
830            use crate::server::record::CommonFieldPutResult;
831            // Snapshot alarm-ack state so the post block can replicate C
832            // putAckt/putAcks (dbAccess.c:1285-1315), which post only when
833            // `ackt`/`acks` actually change.
834            let ackt_before = instance.common.ackt;
835            let acks_before = instance.common.acks;
836            let common_result = match instance.record.put_field(&field, value.clone()) {
837                Ok(()) => {
838                    instance.record.on_put(&field);
839                    let _ = instance.record.special(&field, true);
840                    // C `dbAccess.c::dbPut:1410-1411` clears
841                    // `precord->udf = FALSE` synchronously when the
842                    // put target is the record-type's primary value
843                    // field (`dbIsValueField`). The clear happens
844                    // BEFORE `dbProcess` runs, so any reader between
845                    // the put and the process-cycle's own clear sees
846                    // the new value with a consistent UDF=false.
847                    //
848                    // Rust's processing path also clears UDF via
849                    // `clears_udf()` in process/complete_async_record,
850                    // but that runs AFTER the put lock drops and the
851                    // process re-acquires — leaving a small window
852                    // where another reader can observe (new VAL,
853                    // udf=true). For async records the window spans
854                    // the entire device round trip. Clear here to
855                    // close the window. The same clear already exists
856                    // in `put_pv_and_post` (line 256-262); mirror it.
857                    if field == instance.record.primary_field() {
858                        instance.common.udf = false;
859                        if instance.common.stat == crate::server::recgbl::alarm_status::UDF_ALARM {
860                            instance.common.stat = 0;
861                            instance.common.sevr = crate::server::record::AlarmSeverity::NoAlarm;
862                        }
863                    }
864                    CommonFieldPutResult::NoChange
865                }
866                Err(CaError::FieldNotFound(_)) => instance.put_common_field(&field, value)?,
867                Err(e) => {
868                    instance.common.putf = false;
869                    return Err(e);
870                }
871            };
872
873            // Invalidate metadata cache only if the metadata-class
874            // field's value actually changed (faac1df1).
875            instance.notify_field_written_if_changed(&field, prev_value.as_ref());
876
877            // C `dbAccess.c::dbPutField:1276` sets `precord->putf = TRUE`
878            // immediately before calling `dbProcess`, and the flag stays
879            // TRUE through the entire process cycle. It is cleared only
880            // in `recGblFwdLink` (recGbl.c:302) after FLNK fires, OR in
881            // the disable-alarm bail (dbAccess.c:576). The Rust port
882            // previously cleared `putf` here — BEFORE the
883            // `process_record_with_links` call below — so any code
884            // path (TPRO trace, async-completion logic, monitor on
885            // .PUTF) observing the bit during the process cycle saw
886            // `putf=0` and could not distinguish put-driven vs
887            // scan-driven processing.
888            //
889            // DO NOT clear `putf` here. The clearing now happens after
890            // the process call returns (synchronous completion) or in
891            // `complete_async_record` (async completion).
892
893            instance.cleanup_subscribers();
894            // C `dbPut:1408-1414` posts DBE_VALUE|DBE_LOG for the put field
895            // unless `(isValueField && pfldDes->process_passive)` — the
896            // immediate post is suppressed for the value field ONLY when that
897            // field is `pp(TRUE)`, because then the reprocess cycle
898            // (`dbPutField:1265-1268`) re-posts it via the deadband snapshot.
899            // For a value field that is NOT `pp` (calc/calcout/aSub VAL), C
900            // posts here and does not reprocess; the port must do the same,
901            // because the `should_process` gate below skips the cycle for a
902            // non-`pp` value field — without this post a direct VAL put would
903            // fire no monitor at all.
904            if field.eq_ignore_ascii_case("ACKT") || field.eq_ignore_ascii_case("ACKS") {
905                // Alarm-acknowledge fields are C `dbPut`'s DBR_PUT_ACKT/ACKS
906                // special handlers (`dbAccess.c:1285-1315`), NOT a plain
907                // common-field put. They post with DBE_VALUE|DBE_ALARM (never
908                // DBE_LOG), and post a record-wide DBE_ALARM
909                // (`db_post_events(precord, NULL, DBE_ALARM)`) so an
910                // alarm-mask monitor on ANY field observes the ack — but only
911                // when the ack state actually changed, so the generic
912                // DBE_VALUE|DBE_LOG post below is fully suppressed here.
913                use crate::server::recgbl::EventMask;
914                let ack_mask = EventMask::VALUE | EventMask::ALARM;
915                if field.eq_ignore_ascii_case("ACKT") {
916                    // putAckt: post only on a real ackt change; re-post ACKS
917                    // when turning ACKT off lowered it (C:1294-1297).
918                    if instance.common.ackt != ackt_before {
919                        instance.notify_field(&field, ack_mask);
920                        if instance.common.acks != acks_before {
921                            instance.notify_field("ACKS", ack_mask);
922                        }
923                        instance.notify_record_alarm();
924                    }
925                } else {
926                    // putAcks: post only when the write actually cleared ACKS
927                    // (C:1309-1313); a too-low ack severity posts nothing.
928                    if instance.common.acks != acks_before {
929                        instance.notify_field(&field, ack_mask);
930                        instance.notify_record_alarm();
931                    }
932                }
933            } else {
934                // Suppress the immediate value-field post only when this put
935                // will itself drive a reprocess (the cycle re-posts the field).
936                // `process_passive_fields()` is total/fail-safe: a put to a
937                // non-pp field — including any field of an unmodeled type
938                // (`&[]`) — does not reprocess, so it is not suppressed here.
939                let suppress_value_field_post = field == instance.record.primary_field()
940                    && instance
941                        .record
942                        .process_passive_fields()
943                        .iter()
944                        .any(|f| f.eq_ignore_ascii_case(&field));
945                if !suppress_value_field_post {
946                    instance.notify_field(
947                        &field,
948                        crate::server::recgbl::EventMask::VALUE
949                            | crate::server::recgbl::EventMask::LOG,
950                    );
951                }
952
953                // Fields a `special()` changed as a side effect of this put
954                // (e.g. compress RES reset zeroing NUSE/VAL) get their monitors
955                // posted here, mirroring the explicit `db_post_events` a C
956                // `special()` makes — these fields are not pp(TRUE), so no
957                // process cycle would otherwise post them. Each post carries
958                // VALUE|LOG unless the record names the field in
959                // `value_only_change_fields()` — a record whose C `special()`
960                // posts the field with a literal `DBE_VALUE` (e.g. table SET,
961                // tableRecord.c:659) gets the LOG bit stripped, honoring the
962                // same value-only contract as the change-detection path.
963                let side_effect_value_only = instance.record.value_only_change_fields();
964                for sf in instance.record.monitor_side_effect_fields(&field) {
965                    use crate::server::recgbl::EventMask;
966                    let mask = if side_effect_value_only
967                        .iter()
968                        .any(|f| f.eq_ignore_ascii_case(sf))
969                    {
970                        EventMask::VALUE
971                    } else {
972                        EventMask::VALUE | EventMask::LOG
973                    };
974                    instance.notify_field(sf, mask);
975                }
976            }
977
978            common_result
979        };
980        // ASG-field change re-evaluation hook. C
981        // `asDbLib.c:107-110,144` `asSpcAsCallback` invokes
982        // `asChangeGroup` → `asAddMemberPvt` → `asComputePvt` for
983        // every `ASGCLIENT` on `dbPut record.ASG NEW_ASG`. Pre-fix
984        // Rust mutated `common.asg` directly with no notification,
985        // so the wire ACCESS_RIGHTS the client saw still reflected
986        // the OLD ASG until something else triggered re-eval. Now we
987        // fire a process-wide notifier that the CA server folds into
988        // its per-client `reeval_access_rights` path.
989        if field == "ASG" {
990            crate::server::access_security::notify_asg_field_changed();
991        }
992        // record lock released
993
994        // Update scan index if SCAN or PHAS changed
995        match common_result {
996            crate::server::record::CommonFieldPutResult::ScanChanged {
997                old_scan,
998                new_scan,
999                phas,
1000            } => {
1001                self.update_scan_index(record_name, old_scan, new_scan, phas, phas)
1002                    .await;
1003            }
1004            crate::server::record::CommonFieldPutResult::PhasChanged {
1005                scan: s,
1006                old_phas,
1007                new_phas,
1008            } => {
1009                self.update_scan_index(record_name, s, s, old_phas, new_phas)
1010                    .await;
1011            }
1012            crate::server::record::CommonFieldPutResult::NoChange => {}
1013        }
1014
1015        // C `dbAccess.c::dbPutField:1263-1268` re-processes the
1016        // record on a put only when the put field is `pp(TRUE)` AND the
1017        // record is Passive (`SCAN == 0`). (The `PROC` field has its own
1018        // always-process intercept above, matching C's
1019        // `pfield == &precord->proc`; alarm-ack fields like ACKT/ACKS are
1020        // not `pp(TRUE)` so they fall out here, matching C's
1021        // `dbrType < DBR_PUT_ACKT`.) Processing on every put would
1022        // double-process scanned records and spuriously process puts to
1023        // non-`pp` fields (extra FLNK / monitors / device writes /
1024        // timestamps). `process_passive_fields()` is total and fail-safe: an
1025        // unmodeled type returns `&[]` (and warns once), so it processes on
1026        // `PROC` only — spurious processing is opt-in (a type must declare its
1027        // pp set), never the default.
1028        let should_process = {
1029            let instance = rec.read().await;
1030            instance.common.scan == crate::server::record::ScanType::Passive
1031                && instance.record.processes_after_put(&field)
1032        };
1033
1034        if !should_process {
1035            // No processing cycle. C never sets `putf` on this path, so
1036            // clear the flag the field-put set at entry, and report
1037            // immediate (synchronous) completion to a WRITE_NOTIFY caller.
1038            // Collect-then-act: clone the handle under a brief map read, drop
1039            // the map lock before the per-record write.
1040            let rec_arc = {
1041                let recs = self.inner.records.read().await;
1042                recs.get(record_name).cloned()
1043            };
1044            if let Some(rec_arc) = rec_arc {
1045                let mut guard = rec_arc.write().await;
1046                if !guard.is_processing() {
1047                    guard.common.putf = false;
1048                }
1049            }
1050            return Ok(None);
1051        }
1052
1053        // Set up the put-notify wait-set BEFORE processing. The wait-set
1054        // fires `completion_tx` only after the originating record AND
1055        // every FLNK/OUT chain target it triggers (sync or async) has
1056        // completed — C `dbNotify.c` `processNotify`/`dbNotifyCompletion`.
1057        // Refuse a second concurrent WRITE_NOTIFY on the same record:
1058        // C EPICS returns S_db_Blocked / ECA_PUTCBINPROG, and silently
1059        // overwriting the wait-set would drop the prior Sender, waking
1060        // the prior caller's rx with RecvError that the CA dispatcher
1061        // treats as success.
1062        //
1063        // A fire-and-forget put parks NOTHING — C builds a `putNotify`
1064        // only in `dbPutNotify`; `dbPutField` processes the record with
1065        // no notify state at all. It therefore neither conflicts with
1066        // nor disturbs a WRITE_NOTIFY already parked on the record.
1067        let parked = if want_notify {
1068            let (completion_tx, completion_rx) = crate::runtime::sync::oneshot::channel();
1069            let notify = crate::server::record::NotifyWaitSet::new(completion_tx);
1070            {
1071                // Collect-then-act: clone the handle under a brief map read,
1072                // drop the map lock before the per-record write.
1073                let rec_arc = {
1074                    let recs = self.inner.records.read().await;
1075                    recs.get(record_name).cloned()
1076                };
1077                if let Some(rec_arc) = rec_arc {
1078                    let mut guard = rec_arc.write().await;
1079                    if guard.notify.is_some() {
1080                        return Err(CaError::PutCallbackInProgress(record_name.to_string()));
1081                    }
1082                    guard.notify = Some(notify.clone());
1083                }
1084            }
1085            Some((notify, completion_rx))
1086        } else {
1087            None
1088        };
1089
1090        // When a CA put writes directly to VAL on an INPUT record whose
1091        // VAL is the engineering value, the built-in `RVAL → VAL`
1092        // `convert()` must be suppressed for the put-driven process —
1093        // re-deriving VAL from a stale RVAL would clobber the value the
1094        // operator just wrote (the soft ai preset-NaN case, processing.rs
1095        // ~line 677). The framework expresses this by calling
1096        // `set_device_did_compute(true)`.
1097        //
1098        // This MUST be gated on `soft_channel_skips_convert()`. Output
1099        // records (mbbo/mbbo_direct/bo/ao) implement
1100        // `set_device_did_compute` as "skip the VAL → RVAL output
1101        // convert" — the OPPOSITE direction. C `mbboRecord.c::process`
1102        // (line 217), `mbboDirectRecord.c::process` (line 198) and
1103        // `boRecord.c::process` (line 207) call `convert()`
1104        // unconditionally on every non-pact process; a CA VAL-put on an
1105        // output record MUST recompute RVAL/ORAW. Suppressing it there
1106        // left RVAL/ORAW/ORBV stale. Output records return the default
1107        // `false` from `soft_channel_skips_convert()`, so this gate
1108        // matches the identical gates in processing.rs (line 694) and
1109        // record_instance.rs (line 1381).
1110        if field == "VAL" {
1111            // Collect-then-act: clone the handle under a brief map read, drop
1112            // the map lock before the per-record write.
1113            let rec_arc = {
1114                let recs = self.inner.records.read().await;
1115                recs.get(record_name).cloned()
1116            };
1117            if let Some(rec_arc) = rec_arc {
1118                let mut guard = rec_arc.write().await;
1119                if guard.record.soft_channel_skips_convert() {
1120                    guard.record.set_device_did_compute(true);
1121                }
1122            }
1123        }
1124
1125        // Process the record after field put.
1126        {
1127            let mut visited = HashSet::new();
1128            // `record_name`'s advisory write gate is already
1129            // held by this `put` (the `_record_gate` taken above, or
1130            // the QSRV atomic group's `lock_records` epoch via
1131            // `put_record_field_from_ca_already_locked`). The gate
1132            // `Mutex` is not reentrant — use the `_already_locked`
1133            // processing entry.
1134            let _ = self
1135                .process_record_with_links_already_locked(record_name, &mut visited, 0)
1136                .await;
1137        }
1138
1139        // Is the ORIGINATING record itself still async-pending? Its
1140        // wait-set membership is taken + `leave`d at its own completion
1141        // (sync-end, or later in `complete_async_record_inner`), so a
1142        // lingering `notify` on its instance means its device round-trip
1143        // is still in flight. This gates only the originating record's
1144        // PUTF clear — independent of whether downstream chain targets
1145        // are still pending.
1146        //
1147        // A fire-and-forget put parked nothing, and a `notify` it sees
1148        // on the instance belongs to some other caller's WRITE_NOTIFY —
1149        // not evidence about THIS put. Fall through to the guarded
1150        // clear; its `!is_processing()` gate already preserves PUTF
1151        // across an async-pending device round-trip.
1152        let originating_pending = want_notify && {
1153            let rec = self.inner.records.read().await;
1154            if let Some(rec_arc) = rec.get(record_name) {
1155                rec_arc.read().await.notify.is_some()
1156            } else {
1157                false
1158            }
1159        };
1160
1161        // C `recGbl.c::recGblFwdLink:302` clears `putf = FALSE` after
1162        // the forward-link dispatch — the marker only lives for the
1163        // duration of the put's processing cycle. For SYNCHRONOUS
1164        // completions (PACT was cleared by the time
1165        // `process_record_with_links` returns) clear it here. For
1166        // async-pending records, the clearing happens later in
1167        // `complete_async_record_inner` (which runs FLNK as part of
1168        // the completion path) so the PUTF marker survives the
1169        // device-write round trip.
1170        if !originating_pending {
1171            // Collect-then-act: clone the handle under a brief map read, drop
1172            // the map lock before the per-record write.
1173            let rec_arc = {
1174                let recs = self.inner.records.read().await;
1175                recs.get(record_name).cloned()
1176            };
1177            if let Some(rec_arc) = rec_arc {
1178                let mut guard = rec_arc.write().await;
1179                if !guard.is_processing() {
1180                    guard.common.putf = false;
1181                }
1182            }
1183        }
1184
1185        // CA completion gates on the WHOLE chain, not just the
1186        // originating record: the put-notify must not report
1187        // done until every FLNK/OUT target it drove — including an async
1188        // FLNK target that the originating record's sync cycle merely
1189        // kicked off — has settled. `completed()` is true iff the
1190        // wait-set drained to zero during this call (fully synchronous
1191        // chain); otherwise the receiver fires later from the last
1192        // chain member's `leave`.
1193        match parked {
1194            Some((notify, completion_rx)) => {
1195                if notify.completed() {
1196                    Ok(None)
1197                } else {
1198                    Ok(Some(completion_rx))
1199                }
1200            }
1201            None => Ok(None),
1202        }
1203    }
1204
1205    /// Put a PV value without triggering process (for restore).
1206    pub async fn put_pv_no_process(&self, name: &str, value: EpicsValue) -> CaResult<()> {
1207        let (base, field) = super::parse_pv_name(name);
1208        let field = field.to_ascii_uppercase();
1209
1210        if let Some(pv) = self.inner.simple_pvs.read().await.get(name) {
1211            pv.set(value).await;
1212            return Ok(());
1213        }
1214
1215        // Records — alias-aware (epics-base PR #336).
1216        if let Some(rec) = self.get_record(base).await {
1217            // `put_pv_no_process` is a public record-write API
1218            // (autosave restore). It must take the advisory write gate
1219            // (`dbScanLock` analogue) so an autosave restore cannot
1220            // land between the member writes of a QSRV atomic group or
1221            // a pvalink atomic scan epoch holding `lock_records`.
1222            // `base` is alias-resolved so an alias and its target
1223            // share one gate. Held until return.
1224            let canonical_base: String = self
1225                .resolve_alias(base)
1226                .await
1227                .unwrap_or_else(|| base.to_string());
1228            let _record_gate = self.lock_record(&canonical_base).await;
1229
1230            let mut instance = rec.write().await;
1231            let prev_value = instance.record.get_field(&field);
1232            match instance.record.put_field(&field, value.clone()) {
1233                Ok(()) => {}
1234                Err(CaError::FieldNotFound(_)) => {
1235                    instance.put_common_field(&field, value)?;
1236                }
1237                Err(e) => return Err(e),
1238            }
1239            // Invalidate metadata cache only if the metadata-class
1240            // field actually changed (faac1df1).
1241            instance.notify_field_written_if_changed(&field, prev_value.as_ref());
1242            // same SPC_AS parity as `put_pv` / the CA-write
1243            // path — autosave-style restores writing `.ASG` at IOC
1244            // startup must still trigger per-client re-eval.
1245            if field == "ASG" {
1246                crate::server::access_security::notify_asg_field_changed();
1247            }
1248            return Ok(());
1249        }
1250
1251        Err(CaError::ChannelNotFound(name.to_string()))
1252    }
1253}
1254
1255/// Project a decoded `DBR_CTRL_*` / `DBR_GR_*` snapshot's metadata fields
1256/// (display / control limits, enum labels) into the shadow-PV
1257/// [`PvMetadata`](crate::server::pv::PvMetadata) the CA gateway installs.
1258/// A non-metadata (TIME/STS) snapshot carries `None` in all three, which
1259/// clears the shadow metadata — but the gateway only ever feeds this a
1260/// control-class snapshot, matching C `setPvData` replacing the attribute
1261/// gdd wholesale from the control get/event.
1262fn metadata_from_snapshot(snapshot: &Snapshot) -> crate::server::pv::PvMetadata {
1263    crate::server::pv::PvMetadata {
1264        display: snapshot.display.clone(),
1265        control: snapshot.control.clone(),
1266        enums: snapshot.enums.clone(),
1267    }
1268}
1269
1270#[cfg(test)]
1271mod tests {
1272    use super::super::PvDatabase;
1273    use crate::types::EpicsValue;
1274
1275    /// Regression: prior to fixing B1, `put_pv_and_post` walked only
1276    /// `inner.records` and returned `ChannelNotFound` for everything
1277    /// `add_pv`-registered. The CA gateway's monitor forwarder uses
1278    /// `add_pv` then expects `put_pv_and_post` to fan-out to
1279    /// downstream subscribers — without the simple-PV branch, every
1280    /// upstream event was silently dropped and the gateway delivered
1281    /// no monitors.
1282    #[tokio::test]
1283    async fn put_pv_and_post_handles_simple_pv() {
1284        let db = PvDatabase::new();
1285        db.add_pv("gw:test", EpicsValue::Double(0.0)).await.unwrap();
1286
1287        // Should NOT return ChannelNotFound.
1288        db.put_pv_and_post("gw:test", EpicsValue::Double(42.0))
1289            .await
1290            .expect("simple PV put_pv_and_post must succeed");
1291
1292        // Value actually landed.
1293        let pv = db.find_pv("gw:test").await.expect("PV exists");
1294        assert!(matches!(pv.get().await, EpicsValue::Double(v) if v == 42.0));
1295    }
1296
1297    /// Regression: `get_pv`, `put_pv`, `put_pv_and_post`,
1298    /// and `put_pv_no_process` all bypassed `get_record` and walked
1299    /// `self.inner.records` directly, so alias names from epics-base
1300    /// PR #336 silently returned `ChannelNotFound`. A later fix closed
1301    /// `get_record` but the same defect was hiding in field_io.rs.
1302    /// All four CA-server-and-bridge entry points must accept aliases.
1303    #[tokio::test]
1304    async fn field_io_entry_points_accept_aliases() {
1305        use crate::server::records::ai::AiRecord;
1306
1307        let db = PvDatabase::new();
1308        db.add_record("CANON", Box::new(AiRecord::new(0.0)))
1309            .await
1310            .unwrap();
1311        db.add_alias("ALT", "CANON").await.unwrap();
1312
1313        // get_pv via alias
1314        db.put_pv("CANON.VAL", EpicsValue::Double(1.5))
1315            .await
1316            .unwrap();
1317        let v = db.get_pv("ALT.VAL").await.unwrap();
1318        assert!(matches!(v, EpicsValue::Double(x) if x == 1.5));
1319
1320        // put_pv via alias
1321        db.put_pv("ALT.VAL", EpicsValue::Double(7.0)).await.unwrap();
1322        let v = db.get_pv("CANON.VAL").await.unwrap();
1323        assert!(matches!(v, EpicsValue::Double(x) if x == 7.0));
1324
1325        // put_pv_and_post via alias
1326        db.put_pv_and_post("ALT.VAL", EpicsValue::Double(11.0))
1327            .await
1328            .unwrap();
1329        let v = db.get_pv("CANON.VAL").await.unwrap();
1330        assert!(matches!(v, EpicsValue::Double(x) if x == 11.0));
1331
1332        // put_pv_no_process via alias
1333        db.put_pv_no_process("ALT.VAL", EpicsValue::Double(13.0))
1334            .await
1335            .unwrap();
1336        let v = db.get_pv("ALT.VAL").await.unwrap();
1337        assert!(matches!(v, EpicsValue::Double(x) if x == 13.0));
1338    }
1339
1340    /// A `DBR_STRING` menu label written to a `DBF_MENU` field resolves
1341    /// against THAT field's own menu (C `dbConvert` `putStringMenu`), not the
1342    /// field-blind global table that `EpicsValue::convert_to` would consult.
1343    /// Covers the `put_pv` (`put_pv_inner`) and `put_pv_and_post` coercion
1344    /// sites; the CA field-put path (`put_record_field_from_ca_inner`) shares
1345    /// the identical `coerce_write_value` helper.
1346    #[tokio::test]
1347    async fn write_path_menu_label_resolves_against_field_menu() {
1348        use crate::server::records::sel::SelRecord;
1349
1350        let db = PvDatabase::new();
1351        db.add_record("SEL", Box::new(SelRecord::default()))
1352            .await
1353            .unwrap();
1354
1355        // put_pv (put_pv_inner): "Specified" is selSELM index 0, NOT the
1356        // menuFanout index 1 the global table would have returned.
1357        db.put_pv("SEL.SELM", EpicsValue::String("Specified".into()))
1358            .await
1359            .unwrap();
1360        assert_eq!(db.get_pv("SEL.SELM").await.unwrap(), EpicsValue::Enum(0));
1361
1362        // put_pv_and_post: a later choice, proving the whole menu.
1363        db.put_pv_and_post("SEL.SELM", EpicsValue::String("High Signal".into()))
1364            .await
1365            .unwrap();
1366        assert_eq!(db.get_pv("SEL.SELM").await.unwrap(), EpicsValue::Enum(1));
1367
1368        // A bare numeric string still resolves (C epicsParseUInt16 fallback).
1369        db.put_pv("SEL.SELM", EpicsValue::String("2".into()))
1370            .await
1371            .unwrap();
1372        assert_eq!(db.get_pv("SEL.SELM").await.unwrap(), EpicsValue::Enum(2));
1373    }
1374
1375    /// `set_pv_metadata` installs the upstream `DBR_CTRL_*` metadata on a
1376    /// shadow simple PV WITHOUT posting any event (the CA gateway's
1377    /// connect-time seed). A later GET-class read must then see the
1378    /// installed limits/units, and a `DBE_PROPERTY` subscriber must NOT
1379    /// have received anything (nothing *changed* yet). An unknown / record
1380    /// name is rejected with `ChannelNotFound`.
1381    #[tokio::test]
1382    async fn set_pv_metadata_installs_without_posting() {
1383        use crate::error::CaError;
1384        use crate::server::snapshot::{DisplayInfo, Snapshot};
1385        use crate::types::DbFieldType;
1386        use std::time::SystemTime;
1387
1388        let db = PvDatabase::new();
1389        db.add_pv("gw:meta", EpicsValue::Double(0.0)).await.unwrap();
1390
1391        // A DBE_PROPERTY subscriber attached BEFORE the seed — it must stay
1392        // empty, because seeding metadata is not a property *change*.
1393        const DBE_PROPERTY: u16 = 8;
1394        let pv = db.find_pv("gw:meta").await.expect("PV exists");
1395        let mut prop_rx = pv
1396            .add_subscriber(1, DbFieldType::Double, DBE_PROPERTY)
1397            .await
1398            .expect("subscriber added");
1399
1400        // Build a CTRL-class snapshot carrying display metadata.
1401        let mut ctrl = Snapshot::new(EpicsValue::Double(0.0), 0, 0, SystemTime::UNIX_EPOCH);
1402        ctrl.display = Some(DisplayInfo {
1403            units: "mm".into(),
1404            precision: 3,
1405            upper_disp_limit: 10.0,
1406            lower_disp_limit: -10.0,
1407            ..Default::default()
1408        });
1409
1410        db.set_pv_metadata("gw:meta", &ctrl)
1411            .await
1412            .expect("simple PV set_pv_metadata must succeed");
1413
1414        // The metadata landed on the shadow PV.
1415        let installed = pv.metadata();
1416        assert_eq!(
1417            installed.display.expect("display metadata installed").units,
1418            "mm"
1419        );
1420
1421        // No event was posted (seed != change).
1422        assert!(
1423            prop_rx.try_recv().is_err(),
1424            "set_pv_metadata must not post a DBE_PROPERTY event"
1425        );
1426
1427        // Unknown / non-simple PV is rejected.
1428        assert!(matches!(
1429            db.set_pv_metadata("no:such:pv", &ctrl).await,
1430            Err(CaError::ChannelNotFound(_))
1431        ));
1432    }
1433
1434    /// `post_pv_property` refreshes the shadow metadata AND posts a
1435    /// `DBE_PROPERTY` event carrying the supplied snapshot's metadata,
1436    /// upstream status/severity, and (undefined control-DBR) timestamp — to
1437    /// `DBE_PROPERTY` subscribers only. This is the DB-routing layer the
1438    /// gateway's property monitor drives on every upstream `DBE_PROPERTY`
1439    /// event. An unknown / record name is rejected with `ChannelNotFound`.
1440    #[tokio::test]
1441    async fn post_pv_property_refreshes_and_posts_property_event() {
1442        use crate::error::CaError;
1443        use crate::server::snapshot::{DisplayInfo, Snapshot};
1444        use crate::types::{DbFieldType, WallTime};
1445
1446        const DBE_PROPERTY: u16 = 8;
1447        const DBE_VALUE: u16 = 1;
1448        const MAJOR: u16 = 2;
1449        const HIGH: u16 = 3;
1450
1451        let db = PvDatabase::new();
1452        db.add_pv("gw:prop", EpicsValue::Double(0.0)).await.unwrap();
1453        let pv = db.find_pv("gw:prop").await.expect("PV exists");
1454
1455        let mut prop_rx = pv
1456            .add_subscriber(1, DbFieldType::Double, DBE_PROPERTY)
1457            .await
1458            .expect("property subscriber added");
1459        let mut val_rx = pv
1460            .add_subscriber(2, DbFieldType::Double, DBE_VALUE)
1461            .await
1462            .expect("value subscriber added");
1463
1464        // Upstream CTRL event: metadata + MAJOR/HIGH alarm + a fixed past
1465        // timestamp that is unmistakably not a fresh wall clock.
1466        let upstream_ts = WallTime::from_unix(2_000_000, 0);
1467        let mut ctrl = Snapshot::new(EpicsValue::Double(5.0), HIGH, MAJOR, upstream_ts);
1468        ctrl.display = Some(DisplayInfo {
1469            units: "V".into(),
1470            precision: 1,
1471            ..Default::default()
1472        });
1473
1474        db.post_pv_property("gw:prop", ctrl)
1475            .await
1476            .expect("simple PV post_pv_property must succeed");
1477
1478        // The metadata was refreshed on the shadow PV.
1479        assert_eq!(
1480            pv.metadata().display.expect("metadata refreshed").units,
1481            "V"
1482        );
1483
1484        // The DBE_PROPERTY subscriber received the metadata-bearing event,
1485        // with the upstream alarm and timestamp preserved.
1486        let ev = prop_rx
1487            .try_recv()
1488            .expect("DBE_PROPERTY subscriber receives the property event");
1489        assert_eq!(
1490            ev.snapshot.display.expect("event carries metadata").units,
1491            "V"
1492        );
1493        assert_eq!(
1494            ev.snapshot.alarm.severity, MAJOR,
1495            "upstream severity preserved"
1496        );
1497        assert_eq!(ev.snapshot.alarm.status, HIGH, "upstream status preserved");
1498        assert_eq!(
1499            ev.snapshot.timestamp, upstream_ts,
1500            "control-DBR timestamp preserved, not a fresh wall clock"
1501        );
1502
1503        // The DBE_VALUE-only subscriber must NOT receive a property event.
1504        assert!(
1505            val_rx.try_recv().is_err(),
1506            "DBE_VALUE-only subscriber must not receive a property post"
1507        );
1508
1509        // Unknown / non-simple PV is rejected.
1510        let again = Snapshot::new(EpicsValue::Double(0.0), 0, 0, WallTime::UNIX_EPOCH);
1511        assert!(matches!(
1512            db.post_pv_property("no:such:pv", again).await,
1513            Err(CaError::ChannelNotFound(_))
1514        ));
1515    }
1516
1517    /// Regression: `put_record_field_from_ca` (the CA
1518    /// server's main put fast path) must accept aliases. Pre-fix it
1519    /// only consulted `inner.records` directly. Also exercises the
1520    /// canonical-name normalisation that protects subsequent
1521    /// `process_record_with_links` / `update_scan_index` calls.
1522    #[tokio::test]
1523    async fn put_record_field_from_ca_accepts_alias() {
1524        use crate::server::records::ai::AiRecord;
1525
1526        let db = PvDatabase::new();
1527        db.add_record("CANON", Box::new(AiRecord::new(0.0)))
1528            .await
1529            .unwrap();
1530        db.add_alias("ALT", "CANON").await.unwrap();
1531
1532        // Put VAL via the alias name.
1533        let _ = db
1534            .put_record_field_from_ca("ALT", "VAL", EpicsValue::Double(2.5))
1535            .await
1536            .expect("put via alias must succeed");
1537
1538        // Read back via canonical to confirm the value landed on the
1539        // right record.
1540        let v = db.get_pv("CANON.VAL").await.unwrap();
1541        assert!(matches!(v, EpicsValue::Double(x) if x == 2.5));
1542    }
1543
1544    /// Regression: a DBR_PUT_ACKT alarm-acknowledge put posts a record-wide
1545    /// DBE_ALARM (C `dbAccess.c:1299` putAckt
1546    /// `db_post_events(precord, NULL, DBE_ALARM)`), so an alarm-mask monitor
1547    /// on ANY field is notified — and a DBE_VALUE-only monitor is not.
1548    /// Pre-fix the ack field posted only itself with DBE_VALUE|DBE_LOG, so no
1549    /// alarm-mask subscriber observed the acknowledgement, and the post fired
1550    /// on every put regardless of whether `ackt` changed.
1551    #[tokio::test]
1552    async fn alarm_ack_put_posts_record_wide_dbe_alarm() {
1553        use crate::server::recgbl::EventMask;
1554        use crate::server::records::ai::AiRecord;
1555        use crate::types::DbFieldType;
1556
1557        let db = PvDatabase::new();
1558        db.add_record("A:REC", Box::new(AiRecord::new(1.0)))
1559            .await
1560            .unwrap();
1561        let rec = db.get_record("A:REC").await.expect("record exists");
1562
1563        let (mut alarm_rx, mut value_rx) = {
1564            let mut inst = rec.write().await;
1565            let a = inst
1566                .add_subscriber("VAL", 1, DbFieldType::Double, EventMask::ALARM.bits())
1567                .expect("alarm subscriber");
1568            let v = inst
1569                .add_subscriber("VAL", 2, DbFieldType::Double, EventMask::VALUE.bits())
1570                .expect("value subscriber");
1571            (a, v)
1572        };
1573
1574        // DBR_PUT_ACKT arrives as Short. ACKT defaults YES (true), so writing
1575        // 0 (disable transient acknowledgement) is a real change.
1576        db.put_record_field_from_ca_no_notify("A:REC", "ACKT", EpicsValue::Short(0))
1577            .await
1578            .expect("ackt put");
1579
1580        // The alarm-mask monitor on VAL receives the record-wide DBE_ALARM.
1581        assert!(
1582            alarm_rx.try_recv().is_ok(),
1583            "DBE_ALARM subscriber must receive the record-wide alarm post"
1584        );
1585        // The DBE_VALUE-only monitor on VAL must NOT: VAL's value is unchanged.
1586        assert!(
1587            value_rx.try_recv().is_err(),
1588            "DBE_VALUE-only subscriber must not receive the alarm post"
1589        );
1590
1591        // Re-putting the same ACKT value is a no-op: C putAckt returns early
1592        // on an unchanged ackt, so no further alarm post fires.
1593        db.put_record_field_from_ca_no_notify("A:REC", "ACKT", EpicsValue::Short(0))
1594            .await
1595            .expect("ackt re-put");
1596        assert!(
1597            alarm_rx.try_recv().is_err(),
1598            "unchanged ACKT must post nothing"
1599        );
1600    }
1601
1602    /// `post_property_fields` writes each field through the internal put and
1603    /// posts a `DBE_PROPERTY` monitor — the C
1604    /// `db_post_events(precord, &precord->val, DBE_PROPERTY)` that asyn's
1605    /// runtime enum re-propagation drives (devAsynInt32.c callbackEnum). A
1606    /// `DBE_VALUE`-only subscriber on the same field must NOT receive it:
1607    /// re-keying enum strings is a property change, not a value change.
1608    #[tokio::test]
1609    async fn post_property_fields_writes_and_posts_dbe_property_only() {
1610        use crate::server::recgbl::EventMask;
1611        use crate::server::records::mbbi::MbbiRecord;
1612        use crate::types::DbFieldType;
1613
1614        let db = PvDatabase::new();
1615        db.add_record("M:ENUM", Box::new(MbbiRecord::new(0)))
1616            .await
1617            .unwrap();
1618        let rec = db.get_record("M:ENUM").await.expect("record exists");
1619
1620        let (mut prop_rx, mut val_rx) = {
1621            let mut inst = rec.write().await;
1622            let p = inst
1623                .add_subscriber("ZRST", 1, DbFieldType::String, EventMask::PROPERTY.bits())
1624                .expect("property subscriber");
1625            let v = inst
1626                .add_subscriber("ZRST", 2, DbFieldType::String, EventMask::VALUE.bits())
1627                .expect("value subscriber");
1628            (p, v)
1629        };
1630
1631        let posted = db
1632            .post_property_fields(
1633                "M:ENUM",
1634                vec![("ZRST".to_string(), EpicsValue::String("LABEL".into()))],
1635            )
1636            .await
1637            .expect("post_property_fields succeeds");
1638        assert_eq!(posted, vec!["ZRST".to_string()]);
1639
1640        // The field landed on the record.
1641        assert_eq!(
1642            db.get_pv("M:ENUM.ZRST").await.unwrap(),
1643            EpicsValue::String("LABEL".into())
1644        );
1645
1646        // The DBE_PROPERTY subscriber received the event; the DBE_VALUE-only
1647        // subscriber did not (mask 0x08 vs 0x01, no intersection).
1648        assert!(
1649            prop_rx.try_recv().is_ok(),
1650            "DBE_PROPERTY subscriber must receive the property post"
1651        );
1652        assert!(
1653            val_rx.try_recv().is_err(),
1654            "DBE_VALUE-only subscriber must not receive a property post"
1655        );
1656    }
1657
1658    /// Regression: a direct CA put to a record whose value field VAL is NOT
1659    /// `pp(TRUE)` (calc / calcout / aSub) must still fire a DBE_VALUE monitor.
1660    /// C `dbAccess.c::dbPut:1408-1414` posts the value field immediately
1661    /// unless it is `pp(TRUE)`. The port previously suppressed the immediate
1662    /// post for every `VAL` and — with the `should_process` gate — skipped
1663    /// the reprocess cycle for a non-`pp` VAL, so the operator's write fired
1664    /// no monitor at all. calc's VAL is not in its `pp` field set, so the
1665    /// immediate post is the only event that can fire.
1666    #[tokio::test]
1667    async fn ca_put_to_non_pp_val_posts_monitor() {
1668        use crate::server::database::db_access::DbSubscription;
1669        use crate::server::records::calc::CalcRecord;
1670
1671        let db = PvDatabase::new();
1672        db.add_record("CALC1", Box::new(CalcRecord::new("0")))
1673            .await
1674            .unwrap();
1675
1676        let mut sub = DbSubscription::subscribe(&db, "CALC1.VAL")
1677            .await
1678            .expect("subscribe to CALC1.VAL");
1679
1680        db.put_record_field_from_ca("CALC1", "VAL", EpicsValue::Double(5.0))
1681            .await
1682            .expect("CA put to CALC1.VAL must succeed");
1683
1684        let got = tokio::time::timeout(std::time::Duration::from_secs(1), sub.recv_f64())
1685            .await
1686            .expect("a DBE_VALUE monitor must fire for a direct VAL put to a non-pp record");
1687        assert_eq!(got, Some(5.0));
1688    }
1689
1690    /// Record-backed consumer half of the source-coalesced stale-tail rule.
1691    ///
1692    /// `DbSubscription::next_event` shares the `coalesce_consume` ordering
1693    /// rule with `PvSubscription`: a record-field monitor whose bounded
1694    /// queue overflows mid-burst must converge on the newest value and
1695    /// never step back to an older queued one. Each non-pp `VAL` put posts
1696    /// exactly one DBE_VALUE monitor with the put value and does NOT
1697    /// reprocess (see `ca_put_to_non_pp_val_posts_monitor`), so 80 distinct
1698    /// puts produce a strictly increasing 1..=80 stream; with no consumer
1699    /// draining, 1..=64 fill the queue and the newest (80) lands in the
1700    /// coalesce slot.
1701    ///
1702    /// Before the fix `next_event` returned the coalesced `80` and then
1703    /// replayed the stale tail `1..=64` (`80, 1, 2, ...`) — value time
1704    /// going backwards.
1705    #[tokio::test]
1706    async fn r0604_db_overflow_never_delivers_newest_then_old() {
1707        use crate::server::database::db_access::DbSubscription;
1708        use crate::server::records::calc::CalcRecord;
1709
1710        let db = PvDatabase::new();
1711        db.add_record("CALC1", Box::new(CalcRecord::new("0")))
1712            .await
1713            .unwrap();
1714        let mut sub = DbSubscription::subscribe(&db, "CALC1.VAL")
1715            .await
1716            .expect("subscribe to CALC1.VAL");
1717
1718        for i in 1..=80u32 {
1719            db.put_record_field_from_ca("CALC1", "VAL", EpicsValue::Double(i as f64))
1720                .await
1721                .expect("CA put to CALC1.VAL must succeed");
1722        }
1723
1724        // Drain every immediately-available delivery; the recv past the
1725        // last event has nothing queued and times out, ending collection.
1726        let mut seq = Vec::new();
1727        while let Ok(Some(v)) =
1728            tokio::time::timeout(std::time::Duration::from_millis(200), sub.recv_f64()).await
1729        {
1730            seq.push(v);
1731        }
1732        assert!(!seq.is_empty(), "consumer must observe at least one value");
1733        for w in seq.windows(2) {
1734            assert!(
1735                w[0] <= w[1],
1736                "record monitor delivery stepped backward {} -> {} (sequence {seq:?})",
1737                w[0],
1738                w[1],
1739            );
1740        }
1741        assert_eq!(
1742            *seq.last().unwrap(),
1743            80.0,
1744            "record consumer must converge on the newest produced value (sequence {seq:?})"
1745        );
1746    }
1747}