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}