spvirit_server/
simple_store.rs

1//! A simple in-memory [`Source`] implementation backed by `RecordInstance`.
2//!
3//! Used by [`PvaServer`](crate::pva_server::PvaServer) to serve PVs without
4//! requiring an external database.
5
6use std::collections::{HashMap, HashSet};
7use std::sync::Arc;
8
9use tokio::sync::{RwLock, mpsc};
10use tracing::debug;
11
12use std::future::Future;
13use std::pin::Pin;
14
15use spvirit_codec::spvd_decode::{DecodedValue, FieldDesc, FieldType, StructureDesc, TypeCode};
16use spvirit_types::{NtPayload, ScalarArrayValue, ScalarValue};
17
18use crate::apply::{
19    apply_alarm_update, apply_control_update, apply_display_update, apply_scalar_array_put,
20    apply_value_update,
21};
22use crate::monitor::MonitorRegistry;
23use crate::pvstore::{PvInfo, Source};
24use crate::types::{RecordData, RecordInstance};
25
26/// Callback invoked after a PUT value is applied to a record.
27pub type OnPutCallback = Arc<dyn Fn(&str, &DecodedValue) + Send + Sync>;
28
29/// Callback invoked by the scan scheduler; returns the new value for the PV.
30pub type ScanCallback = Arc<dyn Fn(&str) -> ScalarValue + Send + Sync>;
31
32/// Callback that computes a derived PV value from its input values.
33pub type LinkCallback = Arc<dyn Fn(&[ScalarValue]) -> ScalarValue + Send + Sync>;
34
35/// Pre-apply PUT validator: `Err(msg)` rejects the PUT (error on the wire).
36pub(crate) type PutValidator = Arc<dyn Fn(&str, &DecodedValue) -> Result<(), String> + Send + Sync>;
37
38/// A link from one or more input PVs to a computed output PV.
39pub(crate) struct LinkDef {
40    pub output: String,
41    pub inputs: Vec<String>,
42    pub compute: LinkCallback,
43}
44
45struct PvEntry {
46    record: RecordInstance,
47    subscribers: Vec<mpsc::Sender<NtPayload>>,
48    /// Value carried by the last update posted to subscribers/monitors —
49    /// the reference point for the MDEL monitor deadband. `None` until the
50    /// first post (so the first change always posts).
51    last_posted: Option<f64>,
52}
53
54/// A simple in-memory PV store.
55pub struct SimplePvStore {
56    pvs: RwLock<HashMap<String, PvEntry>>,
57    on_put: HashMap<String, OnPutCallback>,
58    links: Vec<LinkDef>,
59    compute_alarms: bool,
60    registry: RwLock<Option<Arc<MonitorRegistry>>>,
61    validators: RwLock<HashMap<String, PutValidator>>,
62}
63
64impl SimplePvStore {
65    pub(crate) fn new(
66        records: HashMap<String, RecordInstance>,
67        on_put: HashMap<String, OnPutCallback>,
68        links: Vec<LinkDef>,
69        compute_alarms: bool,
70    ) -> Self {
71        let pvs = records
72            .into_iter()
73            .map(|(name, record)| {
74                let last_posted = initial_posted(&record);
75                (
76                    name,
77                    PvEntry {
78                        record,
79                        subscribers: Vec::new(),
80                        last_posted,
81                    },
82                )
83            })
84            .collect();
85        Self {
86            pvs: RwLock::new(pvs),
87            on_put,
88            links,
89            compute_alarms,
90            registry: RwLock::new(None),
91            validators: RwLock::new(HashMap::new()),
92        }
93    }
94
95    /// Attach the [`MonitorRegistry`] so that `set_value` can push updates
96    /// to PVAccess monitor clients.  Called automatically by [`PvaServer::run`].
97    pub async fn set_registry(&self, registry: Arc<MonitorRegistry>) {
98        *self.registry.write().await = Some(registry);
99    }
100
101    /// Register a pre-apply PUT validator for a PV.
102    pub(crate) async fn set_validator(&self, name: String, v: PutValidator) {
103        self.validators.write().await.insert(name, v);
104    }
105
106    /// Insert or replace a PV record at runtime.
107    pub async fn insert(&self, name: String, record: RecordInstance) {
108        let mut pvs = self.pvs.write().await;
109        let last_posted = initial_posted(&record);
110        pvs.insert(
111            name,
112            PvEntry {
113                record,
114                subscribers: Vec::new(),
115                last_posted,
116            },
117        );
118    }
119
120    /// Read the current [`ScalarValue`] of a PV.
121    pub async fn get_value(&self, name: &str) -> Option<ScalarValue> {
122        let pvs = self.pvs.read().await;
123        pvs.get(name).map(|e| e.record.current_value())
124    }
125
126    /// Read a clone of the full [`RecordInstance`] backing a PV.
127    ///
128    /// Used by [`RecordFieldSource`](crate::record_fields::RecordFieldSource)
129    /// to serve `<name>.<FIELD>` channels.
130    pub async fn get_record(&self, name: &str) -> Option<RecordInstance> {
131        let pvs = self.pvs.read().await;
132        pvs.get(name).map(|e| e.record.clone())
133    }
134
135    /// Read the full [`NtPayload`] of a PV.
136    pub async fn get_nt(&self, name: &str) -> Option<NtPayload> {
137        let pvs = self.pvs.read().await;
138        pvs.get(name).map(|e| e.record.to_ntpayload())
139    }
140
141    /// Write a [`ScalarValue`] to a PV (bypasses on_put).
142    pub async fn set_value(&self, name: &str, value: ScalarValue) -> bool {
143        if self.set_value_inner(name, value).await {
144            self.evaluate_links(name).await;
145            true
146        } else {
147            false
148        }
149    }
150
151    /// Write a [`ScalarArrayValue`] to an array PV (bypasses on_put).
152    pub async fn set_array_value(&self, name: &str, value: ScalarArrayValue) -> bool {
153        if self.set_array_value_inner(name, value).await {
154            self.evaluate_links(name).await;
155            true
156        } else {
157            false
158        }
159    }
160
161    /// Write a full [`NtPayload`] to a PV (bypasses on_put).
162    pub async fn put_nt(&self, name: &str, payload: NtPayload) -> bool {
163        if self.put_nt_inner(name, payload).await {
164            self.evaluate_links(name).await;
165            true
166        } else {
167            false
168        }
169    }
170
171    /// Explicitly set a record's alarm fields (severity/status/message),
172    /// independent of its value. Unlike [`SimplePvStore::set_value`], alarm
173    /// transitions always post — there is no MDEL deadband gating and no
174    /// link evaluation (alarm changes don't propagate links). Returns
175    /// `false` if the alarm state is unchanged (idempotent) or the record
176    /// doesn't support alarm fields (`Table`/`NdArray`/`Generic`) or doesn't
177    /// exist.
178    pub async fn set_alarm(&self, name: &str, severity: i32, status: i32, message: &str) -> bool {
179        let payload = {
180            let mut pvs = self.pvs.write().await;
181            let Some(entry) = pvs.get_mut(name) else {
182                return false;
183            };
184            let alarm = if let Some(nt) = entry.record.nt_scalar_mut() {
185                (
186                    &mut nt.alarm_severity,
187                    &mut nt.alarm_status,
188                    &mut nt.alarm_message,
189                )
190            } else {
191                match &mut entry.record.data {
192                    RecordData::NtEnum { nt, .. } => (
193                        &mut nt.alarm.severity,
194                        &mut nt.alarm.status,
195                        &mut nt.alarm.message,
196                    ),
197                    RecordData::Waveform { nt, .. }
198                    | RecordData::Aai { nt, .. }
199                    | RecordData::Aao { nt, .. }
200                    | RecordData::SubArray { nt, .. } => (
201                        &mut nt.alarm.severity,
202                        &mut nt.alarm.status,
203                        &mut nt.alarm.message,
204                    ),
205                    _ => return false,
206                }
207            };
208            let (sev, sta, msg) = alarm;
209            let changed = *sev != severity || *sta != status || msg.as_str() != message;
210            if !changed {
211                return false;
212            }
213            *sev = severity;
214            *sta = status;
215            *msg = message.to_string();
216
217            let payload = entry.record.to_ntpayload();
218            entry
219                .subscribers
220                .retain(|tx| tx.try_send(payload.clone()).is_ok());
221            payload
222        };
223
224        let reg = self.registry.read().await;
225        if let Some(registry) = reg.as_ref() {
226            registry.notify_monitors(name, &payload).await;
227        }
228        true
229    }
230
231    /// Core write logic — updates the value, notifies subscribers and monitors,
232    /// but does **not** trigger link evaluation (to avoid recursion).
233    async fn set_value_inner(&self, name: &str, value: ScalarValue) -> bool {
234        let payload = {
235            let mut pvs = self.pvs.write().await;
236            if let Some(entry) = pvs.get_mut(name) {
237                let prev_severity = entry.record.to_ntscalar().alarm_severity;
238                let changed = entry.record.set_scalar_value(value, self.compute_alarms);
239                if changed {
240                    if !should_post_update(entry, prev_severity) {
241                        // Changed, but within the MDEL monitor deadband:
242                        // the record holds the new value (GETs see it), just
243                        // no update is posted to subscribers/monitors.
244                        return true;
245                    }
246                    let payload = entry.record.to_ntpayload();
247                    entry
248                        .subscribers
249                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
250                    Some(payload)
251                } else {
252                    None
253                }
254            } else {
255                return false;
256            }
257        };
258
259        if let Some(payload) = payload {
260            // Notify PVAccess monitor clients (if the registry is attached).
261            let reg = self.registry.read().await;
262            if let Some(registry) = reg.as_ref() {
263                registry.notify_monitors(name, &payload).await;
264            }
265            true
266        } else {
267            false
268        }
269    }
270
271    /// Core array write logic — updates the value, notifies subscribers and monitors,
272    /// but does **not** trigger link evaluation (to avoid recursion).
273    async fn set_array_value_inner(&self, name: &str, value: ScalarArrayValue) -> bool {
274        let payload = {
275            let mut pvs = self.pvs.write().await;
276            if let Some(entry) = pvs.get_mut(name) {
277                let changed = entry.record.set_array_value(value);
278                if changed {
279                    let payload = entry.record.to_ntpayload();
280                    entry
281                        .subscribers
282                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
283                    Some(payload)
284                } else {
285                    None
286                }
287            } else {
288                return false;
289            }
290        };
291
292        if let Some(payload) = payload {
293            // Notify PVAccess monitor clients (if the registry is attached).
294            let reg = self.registry.read().await;
295            if let Some(registry) = reg.as_ref() {
296                registry.notify_monitors(name, &payload).await;
297            }
298            true
299        } else {
300            false
301        }
302    }
303
304    /// Core NtPayload write logic — updates the payload, notifies subscribers
305    /// and monitors, but does **not** trigger link evaluation.
306    async fn put_nt_inner(&self, name: &str, payload: NtPayload) -> bool {
307        let payload = {
308            let mut pvs = self.pvs.write().await;
309            if let Some(entry) = pvs.get_mut(name) {
310                let changed = entry.record.set_nt_payload(payload);
311                if changed {
312                    let payload = entry.record.to_ntpayload();
313                    entry
314                        .subscribers
315                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
316                    Some(payload)
317                } else {
318                    None
319                }
320            } else {
321                return false;
322            }
323        };
324
325        if let Some(payload) = payload {
326            // Notify PVAccess monitor clients (if the registry is attached).
327            let reg = self.registry.read().await;
328            if let Some(registry) = reg.as_ref() {
329                registry.notify_monitors(name, &payload).await;
330            }
331            true
332        } else {
333            false
334        }
335    }
336
337    /// Walk every link whose inputs include `changed_pv`, compute the output,
338    /// and propagate (BFS with cycle detection).
339    async fn evaluate_links(&self, changed_pv: &str) {
340        if self.links.is_empty() {
341            return;
342        }
343        let mut queue = vec![changed_pv.to_string()];
344        let mut visited = HashSet::new();
345
346        while let Some(pv) = queue.pop() {
347            if !visited.insert(pv.clone()) {
348                debug!("Circular link detected for PV '{}', skipping", pv);
349                continue;
350            }
351            for link in &self.links {
352                if !link.inputs.iter().any(|i| i == &pv) {
353                    continue;
354                }
355                // Gather current values of all inputs.
356                let values = {
357                    let pvs = self.pvs.read().await;
358                    link.inputs
359                        .iter()
360                        .map(|n| {
361                            pvs.get(n)
362                                .map(|e| e.record.current_value())
363                                .unwrap_or(ScalarValue::F64(0.0))
364                        })
365                        .collect::<Vec<_>>()
366                };
367                let new_val = (link.compute)(&values);
368                if self.set_value_inner(&link.output, new_val).await {
369                    queue.push(link.output.clone());
370                }
371            }
372        }
373    }
374
375    /// List all PV names.
376    pub async fn pv_names(&self) -> Vec<String> {
377        let pvs = self.pvs.read().await;
378        pvs.keys().cloned().collect()
379    }
380}
381
382impl Source for SimplePvStore {
383    fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
384        let name = name.to_string();
385        Box::pin(async move {
386            let pvs = self.pvs.read().await;
387            let entry = pvs.get(&name)?;
388            let descriptor = descriptor_for_payload(&entry.record.to_ntpayload());
389            Some(PvInfo {
390                descriptor,
391                writable: entry.record.writable(),
392            })
393        })
394    }
395
396    fn get(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<NtPayload>> + Send + '_>> {
397        let name = name.to_string();
398        Box::pin(async move {
399            let pvs = self.pvs.read().await;
400            pvs.get(&name).map(|e| e.record.to_ntpayload())
401        })
402    }
403
404    fn put(
405        &self,
406        name: &str,
407        value: &DecodedValue,
408    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, NtPayload)>, String>> + Send + '_>> {
409        let name = name.to_string();
410        let value = value.clone();
411        Box::pin(async move {
412            // Clone the validator out inside a tight scope so the read guard
413            // drops before the user callback runs — otherwise temporary
414            // lifetime extension holds the lock across the call, blocking
415            // concurrent set_validator for the duration of every PUT.
416            let validator = {
417                let guard = self.validators.read().await;
418                guard.get(&name).cloned()
419            };
420            if let Some(v) = validator {
421                v(&name, &value)?;
422            }
423
424            let result = {
425                let mut pvs = self.pvs.write().await;
426                let entry = pvs
427                    .get_mut(&name)
428                    .ok_or_else(|| format!("PV '{}' not found", name))?;
429
430                if !entry.record.writable() {
431                    return Err(format!("PV '{}' is not writable", name));
432                }
433
434                let prev_severity = entry.record.to_ntscalar().alarm_severity;
435                let changed = apply_put_to_record(&mut entry.record, &value, self.compute_alarms);
436                if !changed {
437                    return Ok(vec![]);
438                }
439
440                if should_post_update(entry, prev_severity) {
441                    let payload = entry.record.to_ntpayload();
442                    entry
443                        .subscribers
444                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
445                    Some((name.clone(), payload))
446                } else {
447                    // Within the MDEL monitor deadband — value applied, no post.
448                    None
449                }
450            }; // pvs lock dropped
451
452            // Fire on_put callback (non-blocking).
453            if let Some(cb) = self.on_put.get(&name) {
454                let cb = cb.clone();
455                let n = name.clone();
456                let v = value.clone();
457                tokio::spawn(async move { cb(&n, &v) });
458            }
459
460            // Propagate linked PV updates.
461            self.evaluate_links(&name).await;
462
463            Ok(result.into_iter().collect())
464        })
465    }
466
467    fn subscribe(
468        &self,
469        name: &str,
470    ) -> Pin<Box<dyn Future<Output = Option<mpsc::Receiver<NtPayload>>> + Send + '_>> {
471        let name = name.to_string();
472        Box::pin(async move {
473            let mut pvs = self.pvs.write().await;
474            let entry = pvs.get_mut(&name)?;
475            let (tx, rx) = mpsc::channel(64);
476            entry.subscribers.push(tx);
477            Some(rx)
478        })
479    }
480
481    fn names(&self) -> Pin<Box<dyn Future<Output = Vec<String>> + Send + '_>> {
482        Box::pin(async move {
483            let pvs = self.pvs.read().await;
484            pvs.keys().cloned().collect()
485        })
486    }
487}
488
489// ── Helpers ──────────────────────────────────────────────────────────────
490
491/// Numeric view of a scalar value, for deadband arithmetic.
492fn scalar_as_f64(v: &ScalarValue) -> Option<f64> {
493    Some(match v {
494        ScalarValue::I8(x) => *x as f64,
495        ScalarValue::I16(x) => *x as f64,
496        ScalarValue::I32(x) => *x as f64,
497        ScalarValue::I64(x) => *x as f64,
498        ScalarValue::U8(x) => *x as f64,
499        ScalarValue::U16(x) => *x as f64,
500        ScalarValue::U32(x) => *x as f64,
501        ScalarValue::U64(x) => *x as f64,
502        ScalarValue::F32(x) => *x as f64,
503        ScalarValue::F64(x) => *x,
504        ScalarValue::Bool(_) | ScalarValue::Str(_) => return None,
505    })
506}
507
508/// Initial MDEL deadband reference: the record's starting value (EPICS
509/// initialises MLST from the initial VAL, so the first small change is
510/// already subject to the deadband).
511fn initial_posted(record: &RecordInstance) -> Option<f64> {
512    match record.to_ntpayload() {
513        NtPayload::Scalar(nt) => scalar_as_f64(&nt.value),
514        _ => None,
515    }
516}
517
518/// MDEL monitor-deadband gate, called after a record changed.
519///
520/// Returns `true` when the update must be posted to subscribers/monitors
521/// (and records it as the new deadband reference point). An update is
522/// suppressed only when the record is a numeric scalar with MDEL > 0, the
523/// alarm severity did not change, and the value moved less than MDEL from
524/// the last *posted* value — EPICS monitor-deadband semantics.
525fn should_post_update(entry: &mut PvEntry, prev_severity: i32) -> bool {
526    let new_f = match entry.record.to_ntpayload() {
527        NtPayload::Scalar(nt) => match scalar_as_f64(&nt.value) {
528            Some(f) => f,
529            None => return true,
530        },
531        _ => return true,
532    };
533    let mdel = crate::record_fields::mdel_of(&entry.record);
534    let severity_changed = entry.record.to_ntscalar().alarm_severity != prev_severity;
535    let within_deadband = mdel > 0.0
536        && !severity_changed
537        && entry
538            .last_posted
539            .is_some_and(|last| (new_f - last).abs() < mdel);
540    if within_deadband {
541        return false;
542    }
543    entry.last_posted = Some(new_f);
544    true
545}
546
547/// Apply a decoded PUT value to a RecordInstance, returning whether it changed.
548fn apply_put_to_record(
549    record: &mut RecordInstance,
550    value: &DecodedValue,
551    compute_alarms: bool,
552) -> bool {
553    let fields = match value {
554        DecodedValue::Structure(f) => f,
555        other => {
556            // Bare scalar — wrap as value field.
557            return apply_put_to_record(
558                record,
559                &DecodedValue::Structure(vec![("value".to_string(), other.clone())]),
560                compute_alarms,
561            );
562        }
563    };
564
565    let mut changed = false;
566
567    match &mut record.data {
568        RecordData::Ai { nt, .. }
569        | RecordData::Ao { nt, .. }
570        | RecordData::Bi { nt, .. }
571        | RecordData::Bo { nt, .. }
572        | RecordData::StringIn { nt, .. }
573        | RecordData::StringOut { nt, .. } => {
574            for (name, val) in fields {
575                match name.as_str() {
576                    "value" => {
577                        changed |= apply_value_update(nt, val, compute_alarms);
578                    }
579                    "alarm" => {
580                        changed |= apply_alarm_update(nt, val);
581                    }
582                    "display" => {
583                        changed |= apply_display_update(nt, val);
584                    }
585                    "control" => {
586                        changed |= apply_control_update(nt, val);
587                    }
588                    _ => {}
589                }
590            }
591        }
592        RecordData::Waveform { nt, nord, .. }
593        | RecordData::Aai { nt, nord, .. }
594        | RecordData::Aao { nt, nord, .. }
595        | RecordData::SubArray { nt, nord, .. } => {
596            changed = apply_scalar_array_put(nt, nord, value);
597        }
598        RecordData::NtTable { .. } | RecordData::NtNdArray { .. } => {
599            // Table/NdArray PUT not supported via high-level API yet.
600            debug!("PUT to NtTable/NtNdArray not yet supported in SimplePvStore");
601        }
602        RecordData::NtEnum { nt, .. } => {
603            // Accept index updates for NtEnum PVs.
604            for (name, val) in fields {
605                if name == "value" {
606                    let idx = match val {
607                        DecodedValue::Int32(v) => Some(*v),
608                        DecodedValue::Int64(v) => Some(*v as i32),
609                        DecodedValue::Int16(v) => Some(*v as i32),
610                        DecodedValue::Int8(v) => Some(*v as i32),
611                        DecodedValue::Float64(v) => Some(*v as i32),
612                        _ => None,
613                    };
614                    if let Some(idx) = idx {
615                        if idx < 0 || (idx as usize) >= nt.choices.len() {
616                            // out-of-range index — reject, keep value
617                        } else if nt.index != idx {
618                            nt.index = idx;
619                            changed = true;
620                        }
621                    }
622                }
623            }
624        }
625        RecordData::Generic { .. } => {
626            debug!("PUT to Generic not yet supported in SimplePvStore");
627        }
628    }
629
630    changed
631}
632
633// ── NtPayload → StructureDesc ────────────────────────────────────────────
634
635pub fn descriptor_for_payload(payload: &NtPayload) -> StructureDesc {
636    match payload {
637        NtPayload::Scalar(nt) => nt_scalar_desc(&nt.value),
638        NtPayload::ScalarArray(arr) => nt_scalar_array_desc(&arr.value),
639        _ => StructureDesc::new(),
640    }
641}
642
643fn value_type_code(sv: &ScalarValue) -> TypeCode {
644    match sv {
645        ScalarValue::Bool(_) => TypeCode::Boolean,
646        ScalarValue::I8(_) => TypeCode::Int8,
647        ScalarValue::I16(_) => TypeCode::Int16,
648        ScalarValue::I32(_) => TypeCode::Int32,
649        ScalarValue::I64(_) => TypeCode::Int64,
650        ScalarValue::U8(_) => TypeCode::UInt8,
651        ScalarValue::U16(_) => TypeCode::UInt16,
652        ScalarValue::U32(_) => TypeCode::UInt32,
653        ScalarValue::U64(_) => TypeCode::UInt64,
654        ScalarValue::F32(_) => TypeCode::Float32,
655        ScalarValue::F64(_) => TypeCode::Float64,
656        ScalarValue::Str(_) => TypeCode::String,
657    }
658}
659
660fn array_type_code(sav: &ScalarArrayValue) -> TypeCode {
661    match sav {
662        ScalarArrayValue::Bool(_) => TypeCode::Boolean,
663        ScalarArrayValue::I8(_) => TypeCode::Int8,
664        ScalarArrayValue::I16(_) => TypeCode::Int16,
665        ScalarArrayValue::I32(_) => TypeCode::Int32,
666        ScalarArrayValue::I64(_) => TypeCode::Int64,
667        ScalarArrayValue::U8(_) => TypeCode::UInt8,
668        ScalarArrayValue::U16(_) => TypeCode::UInt16,
669        ScalarArrayValue::U32(_) => TypeCode::UInt32,
670        ScalarArrayValue::U64(_) => TypeCode::UInt64,
671        ScalarArrayValue::F32(_) => TypeCode::Float32,
672        ScalarArrayValue::F64(_) => TypeCode::Float64,
673        ScalarArrayValue::Str(_) => TypeCode::String,
674    }
675}
676
677fn nt_scalar_desc(sv: &ScalarValue) -> StructureDesc {
678    let tc = value_type_code(sv);
679    StructureDesc {
680        struct_id: Some("epics:nt/NTScalar:1.0".to_string()),
681        fields: vec![
682            FieldDesc {
683                name: "value".to_string(),
684                field_type: FieldType::Scalar(tc),
685            },
686            FieldDesc {
687                name: "alarm".to_string(),
688                field_type: FieldType::Structure(alarm_desc()),
689            },
690            FieldDesc {
691                name: "timeStamp".to_string(),
692                field_type: FieldType::Structure(timestamp_desc()),
693            },
694            FieldDesc {
695                name: "display".to_string(),
696                field_type: FieldType::Structure(display_desc()),
697            },
698            FieldDesc {
699                name: "control".to_string(),
700                field_type: FieldType::Structure(control_desc()),
701            },
702            FieldDesc {
703                name: "valueAlarm".to_string(),
704                field_type: FieldType::Structure(value_alarm_desc()),
705            },
706        ],
707    }
708}
709
710fn nt_scalar_array_desc(sav: &ScalarArrayValue) -> StructureDesc {
711    let tc = array_type_code(sav);
712    StructureDesc {
713        struct_id: Some("epics:nt/NTScalarArray:1.0".to_string()),
714        fields: vec![
715            FieldDesc {
716                name: "value".to_string(),
717                field_type: FieldType::ScalarArray(tc),
718            },
719            FieldDesc {
720                name: "alarm".to_string(),
721                field_type: FieldType::Structure(alarm_desc()),
722            },
723            FieldDesc {
724                name: "timeStamp".to_string(),
725                field_type: FieldType::Structure(timestamp_desc()),
726            },
727            FieldDesc {
728                name: "display".to_string(),
729                field_type: FieldType::Structure(display_desc()),
730            },
731            FieldDesc {
732                name: "control".to_string(),
733                field_type: FieldType::Structure(control_desc()),
734            },
735        ],
736    }
737}
738
739fn alarm_desc() -> StructureDesc {
740    StructureDesc {
741        struct_id: Some("alarm_t".to_string()),
742        fields: vec![
743            FieldDesc {
744                name: "severity".to_string(),
745                field_type: FieldType::Scalar(TypeCode::Int32),
746            },
747            FieldDesc {
748                name: "status".to_string(),
749                field_type: FieldType::Scalar(TypeCode::Int32),
750            },
751            FieldDesc {
752                name: "message".to_string(),
753                field_type: FieldType::String,
754            },
755        ],
756    }
757}
758
759fn timestamp_desc() -> StructureDesc {
760    StructureDesc {
761        struct_id: Some("time_t".to_string()),
762        fields: vec![
763            FieldDesc {
764                name: "secondsPastEpoch".to_string(),
765                field_type: FieldType::Scalar(TypeCode::Int64),
766            },
767            FieldDesc {
768                name: "nanoseconds".to_string(),
769                field_type: FieldType::Scalar(TypeCode::Int32),
770            },
771            FieldDesc {
772                name: "userTag".to_string(),
773                field_type: FieldType::Scalar(TypeCode::Int32),
774            },
775        ],
776    }
777}
778
779fn display_desc() -> StructureDesc {
780    StructureDesc {
781        struct_id: Some("display_t".to_string()),
782        fields: vec![
783            FieldDesc {
784                name: "limitLow".to_string(),
785                field_type: FieldType::Scalar(TypeCode::Float64),
786            },
787            FieldDesc {
788                name: "limitHigh".to_string(),
789                field_type: FieldType::Scalar(TypeCode::Float64),
790            },
791            FieldDesc {
792                name: "description".to_string(),
793                field_type: FieldType::String,
794            },
795            FieldDesc {
796                name: "units".to_string(),
797                field_type: FieldType::String,
798            },
799            FieldDesc {
800                name: "precision".to_string(),
801                field_type: FieldType::Scalar(TypeCode::Int32),
802            },
803            FieldDesc {
804                name: "form".to_string(),
805                field_type: FieldType::Structure(StructureDesc {
806                    struct_id: Some("enum_t".to_string()),
807                    fields: vec![
808                        FieldDesc {
809                            name: "index".to_string(),
810                            field_type: FieldType::Scalar(TypeCode::Int32),
811                        },
812                        FieldDesc {
813                            name: "choices".to_string(),
814                            field_type: FieldType::StringArray,
815                        },
816                    ],
817                }),
818            },
819        ],
820    }
821}
822
823fn control_desc() -> StructureDesc {
824    StructureDesc {
825        struct_id: Some("control_t".to_string()),
826        fields: vec![
827            FieldDesc {
828                name: "limitLow".to_string(),
829                field_type: FieldType::Scalar(TypeCode::Float64),
830            },
831            FieldDesc {
832                name: "limitHigh".to_string(),
833                field_type: FieldType::Scalar(TypeCode::Float64),
834            },
835            FieldDesc {
836                name: "minStep".to_string(),
837                field_type: FieldType::Scalar(TypeCode::Float64),
838            },
839        ],
840    }
841}
842
843fn value_alarm_desc() -> StructureDesc {
844    StructureDesc {
845        struct_id: Some("valueAlarm_t".to_string()),
846        fields: vec![
847            FieldDesc {
848                name: "active".to_string(),
849                field_type: FieldType::Scalar(TypeCode::Boolean),
850            },
851            FieldDesc {
852                name: "lowAlarmLimit".to_string(),
853                field_type: FieldType::Scalar(TypeCode::Float64),
854            },
855            FieldDesc {
856                name: "lowWarningLimit".to_string(),
857                field_type: FieldType::Scalar(TypeCode::Float64),
858            },
859            FieldDesc {
860                name: "highWarningLimit".to_string(),
861                field_type: FieldType::Scalar(TypeCode::Float64),
862            },
863            FieldDesc {
864                name: "highAlarmLimit".to_string(),
865                field_type: FieldType::Scalar(TypeCode::Float64),
866            },
867            FieldDesc {
868                name: "lowAlarmSeverity".to_string(),
869                field_type: FieldType::Scalar(TypeCode::Int32),
870            },
871            FieldDesc {
872                name: "lowWarningSeverity".to_string(),
873                field_type: FieldType::Scalar(TypeCode::Int32),
874            },
875            FieldDesc {
876                name: "highWarningSeverity".to_string(),
877                field_type: FieldType::Scalar(TypeCode::Int32),
878            },
879            FieldDesc {
880                name: "highAlarmSeverity".to_string(),
881                field_type: FieldType::Scalar(TypeCode::Int32),
882            },
883            FieldDesc {
884                name: "hysteresis".to_string(),
885                field_type: FieldType::Scalar(TypeCode::UInt8),
886            },
887        ],
888    }
889}
890
891#[cfg(test)]
892mod tests {
893    use super::*;
894    use crate::types::{DbCommonState, RecordType};
895    use spvirit_types::{
896        NdCodec, NdDimension, NtNdArray, NtPayload, NtScalar, NtScalarArray, NtTable,
897        NtTableColumn, ScalarArrayValue, ScalarValue,
898    };
899
900    fn make_ai(name: &str, val: f64) -> RecordInstance {
901        RecordInstance {
902            name: name.to_string(),
903            record_type: RecordType::Ai,
904            common: DbCommonState::default(),
905            data: RecordData::Ai {
906                nt: NtScalar::from_value(ScalarValue::F64(val)),
907                inp: None,
908                siml: None,
909                siol: None,
910                simm: false,
911            },
912            raw_fields: HashMap::new(),
913        }
914    }
915
916    fn make_mbbo(name: &str, choices: Vec<String>, initial: i32) -> RecordInstance {
917        RecordInstance {
918            name: name.to_string(),
919            record_type: RecordType::Mbbo,
920            common: DbCommonState::default(),
921            data: RecordData::NtEnum {
922                nt: spvirit_types::NtEnum::new(initial, choices),
923                inp: None,
924                out: None,
925                omsl: crate::types::OutputMode::Supervisory,
926            },
927            raw_fields: HashMap::new(),
928        }
929    }
930
931    fn make_ao(name: &str, val: f64) -> RecordInstance {
932        RecordInstance {
933            name: name.to_string(),
934            record_type: RecordType::Ao,
935            common: DbCommonState::default(),
936            data: RecordData::Ao {
937                nt: NtScalar::from_value(ScalarValue::F64(val)),
938                out: None,
939                dol: None,
940                omsl: crate::types::OutputMode::Supervisory,
941                drvl: None,
942                drvh: None,
943                oroc: None,
944                siml: None,
945                siol: None,
946                simm: false,
947            },
948            raw_fields: HashMap::new(),
949        }
950    }
951
952    #[tokio::test]
953    async fn mdel_deadband_suppresses_small_monitor_updates() {
954        let recs = crate::db::parse_db(
955            r#"
956record(ao, "DB:AO") {
957    field(VAL, "0.0")
958    field(MDEL, "0.5")
959}"#,
960        )
961        .expect("parse");
962        let store = SimplePvStore::new(recs, HashMap::new(), Vec::new(), false);
963        let mut rx = Source::subscribe(&store, "DB:AO")
964            .await
965            .expect("subscribed");
966
967        // |Δ| = 0.2 < MDEL 0.5 → value updates but no monitor post.
968        assert!(store.set_value("DB:AO", ScalarValue::F64(0.2)).await);
969        // |Δ| = 0.9 ≥ MDEL 0.5 → posted.
970        assert!(store.set_value("DB:AO", ScalarValue::F64(0.9)).await);
971
972        match rx.recv().await.expect("posted update") {
973            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(0.9)),
974            other => panic!("expected scalar, got {other:?}"),
975        }
976        // The suppressed 0.2 update must not be queued behind it.
977        assert!(rx.try_recv().is_err());
978        // GETs always see the latest value regardless of the deadband.
979        assert_eq!(store.get_value("DB:AO").await, Some(ScalarValue::F64(0.9)));
980    }
981
982    fn make_waveform(name: &str, value: ScalarArrayValue) -> RecordInstance {
983        let nelm = value.len();
984        RecordInstance {
985            name: name.to_string(),
986            record_type: RecordType::Waveform,
987            common: DbCommonState::default(),
988            data: RecordData::Waveform {
989                nt: NtScalarArray::from_value(value),
990                inp: None,
991                ftvl: "DOUBLE".to_string(),
992                nelm,
993                nord: nelm,
994            },
995            raw_fields: HashMap::new(),
996        }
997    }
998
999    fn make_nt_table(name: &str) -> RecordInstance {
1000        RecordInstance {
1001            name: name.to_string(),
1002            record_type: RecordType::NtTable,
1003            common: DbCommonState::default(),
1004            data: RecordData::NtTable {
1005                nt: NtTable {
1006                    labels: vec!["X".to_string(), "Y".to_string()],
1007                    columns: vec![
1008                        NtTableColumn {
1009                            name: "x".to_string(),
1010                            values: ScalarArrayValue::F64(vec![1.0, 2.0]),
1011                        },
1012                        NtTableColumn {
1013                            name: "y".to_string(),
1014                            values: ScalarArrayValue::F64(vec![10.0, 20.0]),
1015                        },
1016                    ],
1017                    descriptor: Some("table".to_string()),
1018                    alarm: None,
1019                    time_stamp: None,
1020                },
1021                inp: None,
1022                out: None,
1023                omsl: crate::types::OutputMode::Supervisory,
1024            },
1025            raw_fields: HashMap::new(),
1026        }
1027    }
1028
1029    fn make_nt_ndarray(name: &str) -> RecordInstance {
1030        RecordInstance {
1031            name: name.to_string(),
1032            record_type: RecordType::NtNdArray,
1033            common: DbCommonState::default(),
1034            data: RecordData::NtNdArray {
1035                nt: NtNdArray {
1036                    value: ScalarArrayValue::U8(vec![0; 4]),
1037                    codec: NdCodec {
1038                        name: "none".to_string(),
1039                        parameters: HashMap::new(),
1040                    },
1041                    compressed_size: 4,
1042                    uncompressed_size: 4,
1043                    dimension: vec![NdDimension {
1044                        size: 2,
1045                        offset: 0,
1046                        full_size: 2,
1047                        binning: 1,
1048                        reverse: false,
1049                    }],
1050                    unique_id: 1,
1051                    data_time_stamp: Default::default(),
1052                    attribute: vec![],
1053                    descriptor: Some("ndarray".to_string()),
1054                    alarm: None,
1055                    time_stamp: None,
1056                    display: None,
1057                },
1058                inp: None,
1059                out: None,
1060                omsl: crate::types::OutputMode::Supervisory,
1061            },
1062            raw_fields: HashMap::new(),
1063        }
1064    }
1065
1066    #[tokio::test]
1067    async fn has_pv_returns_true_for_existing() {
1068        let mut records = HashMap::new();
1069        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1070        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1071        assert!(store.claim("TEST:AI").await.is_some());
1072        assert!(store.claim("MISSING").await.is_none());
1073    }
1074
1075    #[tokio::test]
1076    async fn get_snapshot_returns_payload() {
1077        let mut records = HashMap::new();
1078        records.insert("TEST:AI".into(), make_ai("TEST:AI", 42.0));
1079        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1080        let snap = store.get("TEST:AI").await.unwrap();
1081        match snap {
1082            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(42.0)),
1083            _ => panic!("expected scalar"),
1084        }
1085    }
1086
1087    #[tokio::test]
1088    async fn put_value_updates_writable_record() {
1089        let mut records = HashMap::new();
1090        records.insert("TEST:AO".into(), make_ao("TEST:AO", 0.0));
1091        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1092
1093        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(99.5))]);
1094        let result = store.put("TEST:AO", &val).await.unwrap();
1095        assert_eq!(result.len(), 1);
1096        assert_eq!(result[0].0, "TEST:AO");
1097
1098        let snap = store.get("TEST:AO").await.unwrap();
1099        match snap {
1100            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(99.5)),
1101            _ => panic!("expected scalar"),
1102        }
1103    }
1104
1105    #[tokio::test]
1106    async fn put_wire_rejects_out_of_range_enum_index() {
1107        let mut records = HashMap::new();
1108        records.insert(
1109            "E".into(),
1110            make_mbbo("E", vec!["A".into(), "B".into(), "C".into()], 0),
1111        );
1112        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1113
1114        // Out-of-range index — must be a no-op (Ok, no changed PVs), index unchanged.
1115        let result = Source::put(&store, "E", &DecodedValue::Int32(7))
1116            .await
1117            .unwrap();
1118        assert!(result.is_empty());
1119        assert_eq!(store.get_value("E").await.unwrap(), ScalarValue::I32(0));
1120
1121        // In-range index — applied.
1122        let result = Source::put(&store, "E", &DecodedValue::Int32(2))
1123            .await
1124            .unwrap();
1125        assert_eq!(result.len(), 1);
1126        assert_eq!(store.get_value("E").await.unwrap(), ScalarValue::I32(2));
1127    }
1128
1129    #[tokio::test]
1130    async fn put_value_rejects_readonly() {
1131        let mut records = HashMap::new();
1132        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1133        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1134
1135        let val = DecodedValue::Float64(5.0);
1136        let err = store.put("TEST:AI", &val).await.unwrap_err();
1137        assert!(err.contains("not writable"));
1138    }
1139
1140    #[tokio::test]
1141    async fn set_value_bypasses_writable_check() {
1142        let mut records = HashMap::new();
1143        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1144        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1145        assert!(store.set_value("TEST:AI", ScalarValue::F64(10.0)).await);
1146        let val = store.get_value("TEST:AI").await.unwrap();
1147        assert_eq!(val, ScalarValue::F64(10.0));
1148    }
1149
1150    #[tokio::test]
1151    async fn set_array_value_updates_all_scalar_array_types() {
1152        let cases: Vec<ScalarArrayValue> = vec![
1153            ScalarArrayValue::Bool(vec![false, true]),
1154            ScalarArrayValue::I8(vec![1, 2]),
1155            ScalarArrayValue::I16(vec![1, 2]),
1156            ScalarArrayValue::I32(vec![1, 2]),
1157            ScalarArrayValue::I64(vec![1, 2]),
1158            ScalarArrayValue::U8(vec![1, 2]),
1159            ScalarArrayValue::U16(vec![1, 2]),
1160            ScalarArrayValue::U32(vec![1, 2]),
1161            ScalarArrayValue::U64(vec![1, 2]),
1162            ScalarArrayValue::F32(vec![1.0, 2.0]),
1163            ScalarArrayValue::F64(vec![1.0, 2.0]),
1164            ScalarArrayValue::Str(vec!["a".to_string(), "b".to_string()]),
1165        ];
1166
1167        for (idx, updated) in cases.into_iter().enumerate() {
1168            let pv = format!("TEST:WF:{idx}");
1169            let mut records = HashMap::new();
1170            records.insert(pv.clone(), make_waveform(&pv, updated.clone()));
1171            let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1172
1173            assert!(!store.set_array_value(&pv, updated.clone()).await);
1174
1175            let second = match updated {
1176                ScalarArrayValue::Bool(_) => ScalarArrayValue::Bool(vec![true, false]),
1177                ScalarArrayValue::I8(_) => ScalarArrayValue::I8(vec![3, 4]),
1178                ScalarArrayValue::I16(_) => ScalarArrayValue::I16(vec![3, 4]),
1179                ScalarArrayValue::I32(_) => ScalarArrayValue::I32(vec![3, 4]),
1180                ScalarArrayValue::I64(_) => ScalarArrayValue::I64(vec![3, 4]),
1181                ScalarArrayValue::U8(_) => ScalarArrayValue::U8(vec![3, 4]),
1182                ScalarArrayValue::U16(_) => ScalarArrayValue::U16(vec![3, 4]),
1183                ScalarArrayValue::U32(_) => ScalarArrayValue::U32(vec![3, 4]),
1184                ScalarArrayValue::U64(_) => ScalarArrayValue::U64(vec![3, 4]),
1185                ScalarArrayValue::F32(_) => ScalarArrayValue::F32(vec![3.0, 4.0]),
1186                ScalarArrayValue::F64(_) => ScalarArrayValue::F64(vec![3.0, 4.0]),
1187                ScalarArrayValue::Str(_) => {
1188                    ScalarArrayValue::Str(vec!["x".to_string(), "y".to_string()])
1189                }
1190            };
1191
1192            assert!(store.set_array_value(&pv, second.clone()).await);
1193            let snap = store.get(&pv).await.unwrap();
1194            match snap {
1195                NtPayload::ScalarArray(nt) => assert_eq!(nt.value, second),
1196                _ => panic!("expected scalar array"),
1197            }
1198        }
1199    }
1200
1201    #[tokio::test]
1202    async fn get_nt_returns_full_payload() {
1203        let mut records = HashMap::new();
1204        records.insert("TEST:AI".into(), make_ai("TEST:AI", 12.5));
1205        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1206
1207        let nt = store.get_nt("TEST:AI").await.unwrap();
1208        match nt {
1209            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(12.5)),
1210            _ => panic!("expected scalar payload"),
1211        }
1212    }
1213
1214    #[tokio::test]
1215    async fn put_nt_updates_scalar_array_table_and_ndarray() {
1216        let mut records = HashMap::new();
1217        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1218        records.insert(
1219            "TEST:WF".into(),
1220            make_waveform("TEST:WF", ScalarArrayValue::F64(vec![0.0, 0.0])),
1221        );
1222        records.insert("TEST:TBL".into(), make_nt_table("TEST:TBL"));
1223        records.insert("TEST:NDA".into(), make_nt_ndarray("TEST:NDA"));
1224        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1225
1226        assert!(
1227            store
1228                .put_nt(
1229                    "TEST:AI",
1230                    NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(5.0))),
1231                )
1232                .await
1233        );
1234        assert!(
1235            store
1236                .put_nt(
1237                    "TEST:WF",
1238                    NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::F64(vec![
1239                        3.0, 4.0
1240                    ],))),
1241                )
1242                .await
1243        );
1244
1245        let table = NtTable {
1246            labels: vec!["X".to_string(), "Y".to_string()],
1247            columns: vec![
1248                NtTableColumn {
1249                    name: "x".to_string(),
1250                    values: ScalarArrayValue::F64(vec![2.0, 3.0]),
1251                },
1252                NtTableColumn {
1253                    name: "y".to_string(),
1254                    values: ScalarArrayValue::F64(vec![20.0, 30.0]),
1255                },
1256            ],
1257            descriptor: Some("updated table".to_string()),
1258            alarm: None,
1259            time_stamp: None,
1260        };
1261        assert!(
1262            store
1263                .put_nt("TEST:TBL", NtPayload::Table(table.clone()))
1264                .await
1265        );
1266
1267        let ndarray = NtNdArray {
1268            value: ScalarArrayValue::U8(vec![1, 2, 3, 4]),
1269            codec: NdCodec {
1270                name: "none".to_string(),
1271                parameters: HashMap::new(),
1272            },
1273            compressed_size: 4,
1274            uncompressed_size: 4,
1275            dimension: vec![NdDimension {
1276                size: 4,
1277                offset: 0,
1278                full_size: 4,
1279                binning: 1,
1280                reverse: false,
1281            }],
1282            unique_id: 2,
1283            data_time_stamp: Default::default(),
1284            attribute: vec![],
1285            descriptor: Some("updated ndarray".to_string()),
1286            alarm: None,
1287            time_stamp: None,
1288            display: None,
1289        };
1290        assert!(
1291            store
1292                .put_nt("TEST:NDA", NtPayload::NdArray(ndarray.clone()))
1293                .await
1294        );
1295
1296        assert!(
1297            !store
1298                .put_nt(
1299                    "TEST:AI",
1300                    NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::F64(vec![
1301                        1.0
1302                    ]))),
1303                )
1304                .await
1305        );
1306
1307        match store.get_nt("TEST:TBL").await.unwrap() {
1308            NtPayload::Table(nt) => assert_eq!(nt, table),
1309            _ => panic!("expected table payload"),
1310        }
1311        match store.get_nt("TEST:NDA").await.unwrap() {
1312            NtPayload::NdArray(nt) => assert_eq!(nt, ndarray),
1313            _ => panic!("expected ndarray payload"),
1314        }
1315    }
1316
1317    #[tokio::test]
1318    async fn descriptor_matches_value_type() {
1319        let mut records = HashMap::new();
1320        records.insert("TEST:AI".into(), make_ai("TEST:AI", 0.0));
1321        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1322        let info = store.claim("TEST:AI").await.unwrap();
1323        assert_eq!(
1324            info.descriptor.struct_id.as_deref(),
1325            Some("epics:nt/NTScalar:1.0")
1326        );
1327        let desc = info.descriptor;
1328        let value_field = desc.field("value").unwrap();
1329        assert!(matches!(
1330            value_field.field_type,
1331            FieldType::Scalar(TypeCode::Float64)
1332        ));
1333    }
1334
1335    #[tokio::test]
1336    async fn subscribe_receives_updates() {
1337        let mut records = HashMap::new();
1338        records.insert("TEST:AO".into(), make_ao("TEST:AO", 0.0));
1339        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1340
1341        let mut rx = Source::subscribe(&store, "TEST:AO").await.unwrap();
1342
1343        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(7.7))]);
1344        store.put("TEST:AO", &val).await.unwrap();
1345
1346        let update = rx.recv().await.unwrap();
1347        match update {
1348            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(7.7)),
1349            _ => panic!("expected scalar"),
1350        }
1351    }
1352
1353    #[tokio::test]
1354    async fn on_put_callback_is_invoked() {
1355        use std::sync::atomic::{AtomicBool, Ordering};
1356
1357        let called = Arc::new(AtomicBool::new(false));
1358        let called2 = called.clone();
1359
1360        let mut records = HashMap::new();
1361        records.insert("CB:AO".into(), make_ao("CB:AO", 0.0));
1362
1363        let mut on_put = HashMap::new();
1364        let cb: OnPutCallback = Arc::new(move |_name, _val| {
1365            called2.store(true, Ordering::SeqCst);
1366        });
1367        on_put.insert("CB:AO".into(), cb);
1368
1369        let store = SimplePvStore::new(records, on_put, vec![], false);
1370        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(1.0))]);
1371        store.put("CB:AO", &val).await.unwrap();
1372
1373        // Give the spawned task time to run.
1374        tokio::task::yield_now().await;
1375        tokio::task::yield_now().await;
1376
1377        assert!(called.load(Ordering::SeqCst));
1378    }
1379
1380    #[tokio::test]
1381    async fn validator_rejects_put_before_apply() {
1382        let mut records = std::collections::HashMap::new();
1383        records.insert(
1384            "V".to_string(),
1385            crate::pva_server::make_output_record(
1386                "V",
1387                crate::types::RecordType::Ao,
1388                ScalarValue::F64(1.0),
1389            ),
1390        );
1391        let store =
1392            SimplePvStore::new(records, std::collections::HashMap::new(), Vec::new(), false);
1393        store
1394            .set_validator(
1395                "V".to_string(),
1396                std::sync::Arc::new(|_name, _val| Err("nope".to_string())),
1397            )
1398            .await;
1399
1400        let dv = DecodedValue::Float64(2.0);
1401        let res = Source::put(&store, "V", &dv).await;
1402        assert_eq!(res, Err("nope".to_string()));
1403        // value unchanged — validator ran BEFORE apply
1404        assert_eq!(store.get_value("V").await, Some(ScalarValue::F64(1.0)));
1405    }
1406
1407    #[tokio::test]
1408    async fn validator_allows_structure_wrapped_put_through() {
1409        // Real puts to scalar records arrive wrapped as a Structure with a
1410        // "value" field (see apply_put_to_record's bare-scalar-wrapping).
1411        // The validator itself only sees the raw DecodedValue as given to
1412        // `put`; this test documents that a validator returning Ok lets a
1413        // structure-wrapped put proceed and apply normally.
1414        let mut records = std::collections::HashMap::new();
1415        records.insert("W".to_string(), make_ao("W", 1.0));
1416        let store =
1417            SimplePvStore::new(records, std::collections::HashMap::new(), Vec::new(), false);
1418        store
1419            .set_validator("W".to_string(), std::sync::Arc::new(|_name, _val| Ok(())))
1420            .await;
1421
1422        let dv = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(5.0))]);
1423        let res = Source::put(&store, "W", &dv).await;
1424        assert!(res.is_ok());
1425        assert_eq!(store.get_value("W").await, Some(ScalarValue::F64(5.0)));
1426    }
1427}
spvirit_server/simple_store.rs

spvirit_server/
simple_store.rs
