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