spvirit_server/
pvstore.rs

1//! The [`Source`] trait — an object-safe abstraction over any PV data source,
2//! and [`SourceRegistry`] — a dynamic, priority-ordered collection of sources.
3//!
4//! Protocol handlers use `SourceRegistry` to resolve PV names across multiple
5//! registered sources, allowing different backends (in-memory records, hardware
6//! drivers, proxies, etc.) to coexist in a single PVA server. basically what pvxs does with its provider registry.
7
8use std::collections::HashSet;
9use std::future::Future;
10use std::pin::Pin;
11use std::sync::Arc;
12
13use spvirit_codec::spvd_decode::{DecodedValue, StructureDesc};
14use spvirit_types::NtPayload;
15use tokio::sync::{RwLock, mpsc};
16use tracing::debug;
17
18// ---------------------------------------------------------------------------
19// PvInfo — metadata returned by Source::claim
20// ---------------------------------------------------------------------------
21
22/// Metadata about a PV as reported by the source that owns it.
23#[derive(Debug, Clone)]
24pub struct PvInfo {
25    /// Structure descriptor for the PV.
26    pub descriptor: StructureDesc,
27    /// Whether the PV accepts PUT operations.
28    pub writable: bool,
29}
30
31// ---------------------------------------------------------------------------
32// Source — the object-safe provider trait
33// ---------------------------------------------------------------------------
34
35/// Object-safe trait for a PV data provider.
36///
37/// A source is responsible for a set of PV names. The server's
38/// [`SourceRegistry`] iterates sources in priority order to find the first
39/// that *claims* a given name.
40///
41/// # Implementing a custom source
42///
43/// ```rust,ignore
44/// use spvirit_server::pvstore::{Source, PvInfo};
45///
46/// struct MySource { /* ... */ }
47///
48/// impl Source for MySource {
49///     fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
50///         Box::pin(async move { /* ... */ })
51///     }
52///     // ...other methods...
53/// }
54/// ```
55pub trait Source: Send + Sync {
56    /// Check whether this source owns `name` and, if so, return its metadata.
57    ///
58    /// Return `None` to let the registry try the next source.
59    fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>>;
60
61    /// Read the current value of a PV.
62    ///
63    /// Only called for PVs this source has previously claimed.
64    fn get(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<NtPayload>> + Send + '_>>;
65
66    /// Apply a PUT value to a PV.
67    ///
68    /// Returns the list of `(pv_name, updated_payload)` pairs for all PVs
69    /// that changed as a result (e.g. forward-link propagation).
70    fn put(
71        &self,
72        name: &str,
73        value: &DecodedValue,
74    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, NtPayload)>, String>> + Send + '_>>;
75
76    /// Subscribe to value-change notifications on a PV.
77    ///
78    /// Returns `None` if the PV does not support subscription.
79    fn subscribe(
80        &self,
81        name: &str,
82    ) -> Pin<Box<dyn Future<Output = Option<mpsc::Receiver<NtPayload>>> + Send + '_>>;
83
84    /// Execute an RPC call on a channel.
85    ///
86    /// `name` is the channel name, `args` is the decoded request structure.
87    /// Returns the response payload on success.
88    ///
89    /// The default implementation returns an error — override it in sources
90    /// that provide RPC endpoints.
91    fn rpc(
92        &self,
93        _name: &str,
94        _args: &DecodedValue,
95    ) -> Pin<Box<dyn Future<Output = Result<NtPayload, String>> + Send + '_>> {
96        Box::pin(async { Err("RPC not supported".to_string()) })
97    }
98
99    /// List all PV names provided by this source.
100    fn names(&self) -> Pin<Box<dyn Future<Output = Vec<String>> + Send + '_>>;
101}
102
103/// A source that owns a fixed, enumerable set of record names.
104///
105/// The distinction matters because the two tiers have different rules:
106/// **stores must be disjoint** — two stores claiming the same name is a
107/// configuration error caught at `build()` — while **sources may shadow**,
108/// which is a legitimate way to override a PV and only warrants a warning.
109/// `record_names` is synchronous because the disjointness check runs inside
110/// `PvaServerBuilder::build`, which is not async.
111///
112/// The returned list must be deterministic (sorted) so overlap diagnostics
113/// are stable between runs.
114pub trait StoreSource: Source {
115    fn record_names(&self) -> Vec<String>;
116}
117
118// ---------------------------------------------------------------------------
119// SourceEntry — one registered source with its priority
120// ---------------------------------------------------------------------------
121
122struct SourceEntry {
123    /// Human-readable label for debugging / logging.
124    label: String,
125    /// Lower values are checked first.
126    order: i32,
127    /// The actual source implementation.
128    source: Arc<dyn Source>,
129    /// True for entries registered via [`SourceRegistry::add_store`].
130    ///
131    /// Consumed by `PvaServerBuilder::build`'s disjointness check and by
132    /// `SourceRegistry::claim`'s shadow-warning.
133    is_store: bool,
134}
135
136// ---------------------------------------------------------------------------
137// SourceRegistry — ordered collection of sources
138// ---------------------------------------------------------------------------
139
140/// A dynamic, priority-ordered registry of [`Source`] providers.
141///
142/// PV name resolution iterates sources from lowest `order` to highest and
143/// delegates to the first source that claims the name.
144pub struct SourceRegistry {
145    sources: RwLock<Vec<SourceEntry>>,
146    /// PVs whose shadowing status has already been determined. Bounded by
147    /// the number of distinct names clients search for, and only ever grown
148    /// by a claim that a non-store source won.
149    shadow_checked: RwLock<HashSet<String>>,
150}
151
152impl SourceRegistry {
153    /// Create an empty registry.
154    pub fn new() -> Self {
155        Self {
156            sources: RwLock::new(Vec::new()),
157            shadow_checked: RwLock::new(HashSet::new()),
158        }
159    }
160
161    /// Register an ordinary source. Sources may shadow stores and each other.
162    ///
163    /// Lower `order` values are queried first.
164    pub async fn add(&self, label: impl Into<String>, order: i32, source: Arc<dyn Source>) {
165        self.insert(label.into(), order, source, false).await;
166    }
167
168    /// Register a store — a source whose record set is fixed and must not
169    /// overlap another store's. See [`StoreSource`].
170    pub async fn add_store(&self, label: impl Into<String>, order: i32, source: Arc<dyn Source>) {
171        self.insert(label.into(), order, source, true).await;
172    }
173
174    async fn insert(&self, label: String, order: i32, source: Arc<dyn Source>, is_store: bool) {
175        debug!(
176            "SourceRegistry: adding source '{}' at order {} (store: {})",
177            label, order, is_store
178        );
179        let mut sources = self.sources.write().await;
180        sources.push(SourceEntry {
181            label,
182            order,
183            source,
184            is_store,
185        });
186        sources.sort_by_key(|e| e.order);
187    }
188
189    /// Remove all sources with the given label.
190    pub async fn remove(&self, label: &str) {
191        debug!("SourceRegistry: removing source '{}'", label);
192        let mut sources = self.sources.write().await;
193        sources.retain(|e| e.label != label);
194    }
195
196    // ── Delegating operations ────────────────────────────────────────
197
198    /// Find the first source that claims `name` and return its metadata.
199    pub async fn claim(&self, name: &str) -> Option<PvInfo> {
200        let sources = self.sources.read().await;
201        for entry in sources.iter() {
202            if let Some(info) = entry.source.claim(name).await {
203                if !entry.is_store {
204                    self.warn_if_shadowing_a_store(&sources, &entry.label, name)
205                        .await;
206                }
207                return Some(info);
208            }
209        }
210        None
211    }
212
213    /// Log once if `winner` — an ordinary source — beat a store to `name`.
214    ///
215    /// Shadowing between sources is legal and often deliberate, so this
216    /// warns rather than failing. Overlap between two *stores* is the error
217    /// case, and `PvaServerBuilder::build` rejects it outright.
218    ///
219    /// Only `claim` calls this. `get`/`put`/`subscribe`/`rpc` re-resolve the
220    /// same name through the same ordered list, and a client always searches
221    /// before operating, so the first claim is where the diagnostic belongs.
222    async fn warn_if_shadowing_a_store(&self, sources: &[SourceEntry], winner: &str, name: &str) {
223        if self.shadow_checked.read().await.contains(name) {
224            return;
225        }
226        if !self.shadow_checked.write().await.insert(name.to_string()) {
227            // Another task got there between the read and the write.
228            return;
229        }
230        for entry in sources.iter().filter(|e| e.is_store) {
231            if entry.source.claim(name).await.is_some() {
232                tracing::warn!(
233                    "source '{winner}' shadows store '{}' for PV '{name}': the store's \
234                     value will never be served",
235                    entry.label
236                );
237                return;
238            }
239        }
240    }
241
242    /// Check whether any source claims the given PV name.
243    pub async fn has_pv(&self, name: &str) -> bool {
244        self.claim(name).await.is_some()
245    }
246
247    /// Get the value from the first source that claims the PV.
248    pub async fn get(&self, name: &str) -> Option<NtPayload> {
249        let sources = self.sources.read().await;
250        for entry in sources.iter() {
251            if entry.source.claim(name).await.is_some() {
252                return entry.source.get(name).await;
253            }
254        }
255        None
256    }
257
258    /// Get the structure descriptor from the first source that claims the PV.
259    pub async fn get_descriptor(&self, name: &str) -> Option<StructureDesc> {
260        self.claim(name).await.map(|info| info.descriptor)
261    }
262
263    /// Check if the PV is writable (via the first claiming source).
264    pub async fn is_writable(&self, name: &str) -> bool {
265        self.claim(name).await.is_some_and(|info| info.writable)
266    }
267
268    /// Delegate a PUT to the first source that claims the PV.
269    pub async fn put(
270        &self,
271        name: &str,
272        value: &DecodedValue,
273    ) -> Result<Vec<(String, NtPayload)>, String> {
274        let sources = self.sources.read().await;
275        for entry in sources.iter() {
276            if entry.source.claim(name).await.is_some() {
277                return entry.source.put(name, value).await;
278            }
279        }
280        Err(format!("PV '{}' not found", name))
281    }
282
283    /// Subscribe via the first source that claims the PV.
284    pub async fn subscribe(&self, name: &str) -> Option<mpsc::Receiver<NtPayload>> {
285        let sources = self.sources.read().await;
286        for entry in sources.iter() {
287            if entry.source.claim(name).await.is_some() {
288                return entry.source.subscribe(name).await;
289            }
290        }
291        None
292    }
293
294    /// Execute an RPC call via the first source that claims the channel.
295    pub async fn rpc(&self, name: &str, args: &DecodedValue) -> Result<NtPayload, String> {
296        let sources = self.sources.read().await;
297        for entry in sources.iter() {
298            if entry.source.claim(name).await.is_some() {
299                return entry.source.rpc(name, args).await;
300            }
301        }
302        Err(format!("RPC channel '{}' not found", name))
303    }
304
305    /// Collect all PV names from every registered source.
306    pub async fn names(&self) -> Vec<String> {
307        let sources = self.sources.read().await;
308        let mut seen = HashSet::new();
309        let mut all = Vec::new();
310        for entry in sources.iter() {
311            for name in entry.source.names().await {
312                if seen.insert(name.clone()) {
313                    all.push(name);
314                }
315            }
316        }
317        all.sort();
318        all
319    }
320}
321
322impl Default for SourceRegistry {
323    fn default() -> Self {
324        Self::new()
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331
332    /// A minimal [`Source`] over a fixed name list, for registry tests.
333    struct StubSource {
334        names: Vec<String>,
335        claims: std::sync::atomic::AtomicUsize,
336    }
337
338    impl StubSource {
339        fn new(names: &[&str]) -> Self {
340            Self {
341                names: names.iter().map(|s| s.to_string()).collect(),
342                claims: std::sync::atomic::AtomicUsize::new(0),
343            }
344        }
345
346        fn claim_count(&self) -> usize {
347            self.claims.load(std::sync::atomic::Ordering::SeqCst)
348        }
349    }
350
351    impl Source for StubSource {
352        fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
353            self.claims.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
354            let claimed = self.names.iter().any(|n| n == name);
355            Box::pin(async move {
356                claimed.then(|| PvInfo {
357                    descriptor: StructureDesc::default(),
358                    writable: true,
359                })
360            })
361        }
362
363        fn get(&self, _name: &str) -> Pin<Box<dyn Future<Output = Option<NtPayload>> + Send + '_>> {
364            Box::pin(async { None })
365        }
366
367        fn put(
368            &self,
369            _name: &str,
370            _value: &DecodedValue,
371        ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, NtPayload)>, String>> + Send + '_>>
372        {
373            Box::pin(async { Ok(vec![]) })
374        }
375
376        fn subscribe(
377            &self,
378            _name: &str,
379        ) -> Pin<Box<dyn Future<Output = Option<mpsc::Receiver<NtPayload>>> + Send + '_>> {
380            Box::pin(async { None })
381        }
382
383        fn names(&self) -> Pin<Box<dyn Future<Output = Vec<String>> + Send + '_>> {
384            let names = self.names.clone();
385            Box::pin(async move { names })
386        }
387    }
388
389    #[tokio::test]
390    async fn stores_are_recorded_as_stores_and_sources_are_not() {
391        let reg = SourceRegistry::new();
392        reg.add_store("builtin", 0, Arc::new(StubSource::new(&["A"]))).await;
393        reg.add("custom", 10, Arc::new(StubSource::new(&["B"]))).await;
394        let flags: Vec<(String, bool)> = reg
395            .sources
396            .read()
397            .await
398            .iter()
399            .map(|e| (e.label.clone(), e.is_store))
400            .collect();
401        assert_eq!(
402            flags,
403            vec![("builtin".to_string(), true), ("custom".to_string(), false)]
404        );
405    }
406
407    #[tokio::test]
408    async fn a_store_added_late_still_sorts_by_order() {
409        let reg = SourceRegistry::new();
410        reg.add("custom", 10, Arc::new(StubSource::new(&["B"]))).await;
411        reg.add_store("builtin", 0, Arc::new(StubSource::new(&["A"]))).await;
412        let labels: Vec<String> = reg
413            .sources
414            .read()
415            .await
416            .iter()
417            .map(|e| e.label.clone())
418            .collect();
419        assert_eq!(labels, vec!["builtin".to_string(), "custom".to_string()]);
420    }
421
422    /// A source registered ahead of a store wins the claim — that is the
423    /// documented behaviour, and it stays. The registry just says so.
424    #[tokio::test]
425    async fn a_source_shadowing_a_store_still_wins_the_claim() {
426        let reg = SourceRegistry::new();
427        reg.add("override", -1, Arc::new(StubSource::new(&["PV:X"]))).await;
428        reg.add_store("builtin", 0, Arc::new(StubSource::new(&["PV:X"]))).await;
429        assert!(reg.claim("PV:X").await.is_some());
430    }
431
432    /// The warning is emitted at most once per PV, so a client that searches
433    /// in a loop does not flood the log.
434    #[tokio::test]
435    async fn the_shadow_check_runs_once_per_pv() {
436        let reg = SourceRegistry::new();
437        let store = Arc::new(StubSource::new(&["PV:X"]));
438        reg.add("override", -1, Arc::new(StubSource::new(&["PV:X"]))).await;
439        reg.add_store("builtin", 0, store.clone()).await;
440        let before = store.claim_count();
441        for _ in 0..5 {
442            reg.claim("PV:X").await;
443        }
444        assert_eq!(
445            store.claim_count() - before,
446            1,
447            "the shadowed store must be consulted exactly once"
448        );
449    }
450
451    /// A source claiming a name no store owns is the ordinary case and must
452    /// not cost a scan of every store on every search after the first.
453    #[tokio::test]
454    async fn an_unshadowed_source_claim_is_also_checked_only_once() {
455        let reg = SourceRegistry::new();
456        let store = Arc::new(StubSource::new(&["PV:OTHER"]));
457        reg.add("plain", -1, Arc::new(StubSource::new(&["PV:X"]))).await;
458        reg.add_store("builtin", 0, store.clone()).await;
459        let before = store.claim_count();
460        for _ in 0..5 {
461            reg.claim("PV:X").await;
462        }
463        assert_eq!(store.claim_count() - before, 1);
464    }
465
466    /// A store winning its own claim triggers no check at all.
467    #[tokio::test]
468    async fn a_store_winning_its_own_claim_consults_nothing_else() {
469        let reg = SourceRegistry::new();
470        let other = Arc::new(StubSource::new(&["PV:X"]));
471        reg.add_store("builtin", 0, Arc::new(StubSource::new(&["PV:X"]))).await;
472        reg.add_store("second", 5, other.clone()).await;
473        let before = other.claim_count();
474        reg.claim("PV:X").await;
475        assert_eq!(other.claim_count() - before, 0);
476    }
477
478    /// An `io::Write` sink over a shared buffer, for capturing `tracing`
479    /// output during a test. Only ever touched synchronously by the
480    /// subscriber's own formatting call — never held across an `.await`.
481    #[derive(Clone, Default)]
482    struct CaptureWriter(Arc<std::sync::Mutex<Vec<u8>>>);
483
484    impl std::io::Write for CaptureWriter {
485        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
486            self.0.lock().unwrap().extend_from_slice(buf);
487            Ok(buf.len())
488        }
489
490        fn flush(&mut self) -> std::io::Result<()> {
491            Ok(())
492        }
493    }
494
495    /// The counter-based tests above prove the shadow check runs at most
496    /// once per name, but they infer "warned" from how many times the
497    /// store's `claim` ran — which can't tell a real warning apart from a
498    /// silent scan. This test observes the actual `tracing::warn!` event.
499    #[tokio::test]
500    async fn the_shadow_warning_is_emitted_once_not_just_counted() {
501        let buffer = CaptureWriter::default();
502        let writer = buffer.clone();
503        let subscriber = tracing_subscriber::fmt()
504            .with_max_level(tracing::Level::WARN)
505            .with_ansi(false)
506            .without_time()
507            .with_writer(move || writer.clone())
508            .finish();
509
510        let _subscriber_guard = tracing::subscriber::set_default(subscriber);
511
512        let reg = SourceRegistry::new();
513        reg.add("override", -1, Arc::new(StubSource::new(&["PV:X"]))).await;
514        reg.add_store("builtin", 0, Arc::new(StubSource::new(&["PV:X"]))).await;
515
516        reg.claim("PV:X").await;
517        reg.claim("PV:X").await;
518
519        let captured = String::from_utf8(buffer.0.lock().unwrap().clone()).unwrap();
520        let warnings: Vec<&str> = captured.lines().filter(|l| !l.is_empty()).collect();
521        assert_eq!(
522            warnings.len(),
523            1,
524            "expected exactly one warning event, got: {captured:?}"
525        );
526        assert!(warnings[0].contains("PV:X"), "missing PV name: {captured:?}");
527        assert!(warnings[0].contains("override"), "missing source label: {captured:?}");
528        assert!(warnings[0].contains("builtin"), "missing store label: {captured:?}");
529    }
530}
spvirit_server/pvstore.rs

spvirit_server/
pvstore.rs
