settings_storage/
device_storage.rs

1// Copyright 2019 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::UpdateState;
6use crate::private::Sealed;
7use crate::stash_logger::StashInspectLogger;
8use crate::storage_factory::{DefaultLoader, NoneT};
9use anyhow::{Context, Error, format_err};
10use fidl_fuchsia_stash::{StoreAccessorProxy, Value};
11use fuchsia_async::{MonotonicDuration, MonotonicInstant, Task, Timer};
12use futures::channel::mpsc::UnboundedSender;
13use futures::future::OptionFuture;
14use futures::lock::{Mutex, MutexGuard};
15use futures::{FutureExt, StreamExt};
16use serde::Serialize;
17use serde::de::DeserializeOwned;
18use std::any::Any;
19use std::borrow::Cow;
20use std::collections::HashMap;
21use std::pin::pin;
22use std::rc::Rc;
23
24const SETTINGS_PREFIX: &str = "settings";
25
26/// Minimum amount of time between Flush calls to Stash, in milliseconds. The Flush call triggers
27/// file I/O which is slow. If we call flush too often, we can overwhelm Stash, which eventually
28/// causes the kernel to crash our service due to filling up the channel.
29const MIN_FLUSH_INTERVAL: MonotonicDuration = MonotonicDuration::from_millis(500);
30
31/// Stores device level settings in persistent storage.
32/// User level settings should not use this.
33pub struct DeviceStorage {
34    /// Map of [`DeviceStorageCompatible`] keys to their typed storage.
35    typed_storage_map: HashMap<&'static str, TypedStorage>,
36
37    typed_loader_map: HashMap<&'static str, Box<TypeErasedLoader>>,
38
39    /// If true, reads will be returned from the data in memory rather than reading from storage.
40    caching_enabled: bool,
41
42    /// If true, writes to the underlying storage will only occur at most every
43    /// MIN_WRITE_INTERVAL_MS.
44    debounce_writes: bool,
45
46    /// Handle used to write stash failures to inspect.
47    inspect_handle: Rc<Mutex<StashInspectLogger>>,
48}
49
50/// A wrapper for managing all communication and caching for one particular type of data being
51/// stored. The actual types are erased.
52struct TypedStorage {
53    /// Sender to communicate with task loop that handles flushes.
54    flush_sender: UnboundedSender<()>,
55
56    /// Cached storage managed through interior mutability.
57    cached_storage: Mutex<CachedStorage>,
58}
59
60/// `CachedStorage` abstracts over a cached value that's read from and written
61/// to some backing store.
62struct CachedStorage {
63    /// Cache for the most recently read or written value.
64    current_data: Option<Box<TypeErasedData>>,
65
66    /// Stash connection for this particular type's stash storage.
67    stash_proxy: StoreAccessorProxy,
68}
69
70/// Structs that can be stored in device storage
71///
72/// Structs that can be stored in device storage should derive the Serialize, Deserialize, and
73/// Clone traits, as well as provide constants.
74/// KEY should be unique the struct, usually the name of the struct itself.
75/// DEFAULT_VALUE will be the value returned when nothing has yet been stored.
76///
77/// Anything that implements this should not introduce breaking changes with the same key.
78/// Clients that want to make a breaking change should create a new structure with a new key and
79/// implement conversion/cleanup logic. Adding optional fields to a struct is not breaking, but
80/// removing fields, renaming fields, or adding non-optional fields are.
81///
82/// [`Storage`]: super::setting_handler::persist::Storage
83pub trait DeviceStorageCompatible: Serialize + DeserializeOwned + Clone + PartialEq + Any {
84    type Loader: DefaultDispatcher<Self>;
85
86    fn try_deserialize_from(value: &str) -> Result<Self, Error> {
87        Self::extract(value)
88    }
89
90    fn extract(value: &str) -> Result<Self, Error> {
91        serde_json::from_str(value).map_err(|e| format_err!("could not deserialize: {e:?}"))
92    }
93
94    fn serialize_to(&self) -> String {
95        serde_json::to_string(self).expect("value should serialize")
96    }
97
98    const KEY: &'static str;
99}
100
101/// Types that can be converted into a storable type.
102///
103/// This trait represents types that can be converted into a storable type. It's also important
104/// that the type it is transformed into can also be converted back into this type. This reverse
105/// conversion is used to populate the fields of the original type with the stored values plus
106/// defaulting the other fields that, e.g. might later be populated from hardware APIs.
107///
108/// # Example
109/// ```
110/// // Struct used in controllers.
111/// struct SomeSettingInfo {
112///     storable_field: u8,
113///     hardware_backed_field: String,
114/// }
115///
116/// // Struct only used for storage.
117/// #[derive(Serialize, Deserialize, PartialEq, Clone)]
118/// struct StorableSomeSettingInfo {
119///     storable_field: u8,
120/// }
121///
122/// // Impl compatible for the storable type.
123/// impl DeviceStorageCompatible for StorableSomeSettingInfo {
124///     const KEY: &'static str = "some_setting_info";
125///
126///     fn default_value() -> Self {
127///         Self { storable_field: 1, }
128///     }
129/// }
130///
131/// // Impl convertible for controller type.
132/// impl DeviceStorageConvertible for SomeSettingInfo {
133///     type Storable = StorableSomeSettingInfo;
134///     fn get_storable(&self) -> Cow<'_, Self::Storable> {
135///         Cow::Owned(Self {
136///             storable_field: self.storable_field,
137///             hardware_backed_field: String::new()
138///         })
139///     }
140/// }
141///
142/// // This impl helps us convert from the storable version to the
143/// // controller version of the struct. Hardware fields should be backed
144/// // by default or usable values.
145/// impl Into<SomeSettingInfo> for StorableSomeSettingInfo {
146///     fn into(self) -> SomeSettingInfo {
147///         SomeSettingInfo {
148///             storable_field: self.storable_field,
149///             hardware_backed_field: String::new(),
150///         }
151///     }
152/// }
153///
154/// ```
155pub trait DeviceStorageConvertible: Sized {
156    /// The type that will be used for storing the data.
157    type Storable: DeviceStorageCompatible + Into<Self>;
158
159    /// Convert `self` into its storable version.
160    // The reason we don't take ownership here is that the setting handler uses the original value
161    // to send a message on the message hub for when the change is written. Serializing also only
162    // borrows the data and doesn't need to own it. When `Storable` is `Self`, we only need to keep
163    // the borrow on self, but when the types differ, then we need to own the newly constructed
164    // type.
165    fn get_storable(&self) -> Cow<'_, Self::Storable>;
166}
167
168// Any type that is storage compatible is also storage convertible (it can convert to itself!).
169impl<T> DeviceStorageConvertible for T
170where
171    T: DeviceStorageCompatible,
172{
173    type Storable = T;
174
175    fn get_storable(&self) -> Cow<'_, Self::Storable> {
176        Cow::Borrowed(self)
177    }
178}
179
180type MappingFn = Box<dyn FnOnce(&dyn Any) -> String>;
181type TypeErasedData = dyn Any;
182type TypeErasedLoader = dyn Any;
183
184impl DeviceStorage {
185    /// Construct a device storage from the iteratable item, which will produce the keys for
186    /// storage, and from a generator that will produce a stash proxy given a particular key.
187    pub fn with_stash_proxy<I, G>(
188        iter: I,
189        stash_generator: G,
190        inspect_handle: Rc<Mutex<StashInspectLogger>>,
191    ) -> Self
192    where
193        I: IntoIterator<Item = (&'static str, Option<Box<TypeErasedLoader>>)>,
194        G: Fn() -> StoreAccessorProxy,
195    {
196        let mut typed_loader_map = HashMap::new();
197        let typed_storage_map = iter
198            .into_iter()
199            .map({
200                let inspect_handle = Rc::clone(&inspect_handle);
201                let typed_loader_map = &mut typed_loader_map;
202                move |(key, loader)| {
203                    if let Some(loader) = loader {
204                        let _ = typed_loader_map.insert(key, loader);
205                    }
206                    // Generate a separate stash proxy for each key.
207                    let (flush_sender, flush_receiver) = futures::channel::mpsc::unbounded::<()>();
208                    let stash_proxy = stash_generator();
209
210                    let storage = TypedStorage {
211                        flush_sender,
212                        cached_storage: Mutex::new(CachedStorage {
213                            current_data: None,
214                            stash_proxy: stash_proxy.clone(),
215                        }),
216                    };
217
218                    let inspect_handle = Rc::clone(&inspect_handle);
219                    // Each key has an independent flush queue.
220                    Task::local(async move {
221                        let mut next_allowed_flush = MonotonicInstant::now();
222                        let mut next_flush_timer = pin!(OptionFuture::from(None).fuse());
223                        let flush_requested = flush_receiver.fuse();
224                        futures::pin_mut!(flush_requested);
225                        loop {
226                            futures::select! {
227                                () = flush_requested.select_next_some() => {
228                                    next_flush_timer.set(OptionFuture::from(Some(Timer::new(
229                                        next_allowed_flush
230                                    )))
231                                    .fuse());
232                                },
233                                o = next_flush_timer => {
234                                    if let Some(()) = o {
235                                        DeviceStorage::stash_flush(
236                                            &stash_proxy,
237                                            Rc::clone(&inspect_handle),
238                                            key.to_string()).await;
239                                        next_allowed_flush = MonotonicInstant::now() + MIN_FLUSH_INTERVAL;
240                                    }
241                                }
242                                complete => break,
243                            }
244                        }
245                    })
246                    .detach();
247                    (key, storage)
248                }
249            })
250            .collect();
251        DeviceStorage {
252            caching_enabled: true,
253            debounce_writes: true,
254            typed_storage_map,
255            typed_loader_map,
256            inspect_handle,
257        }
258    }
259
260    /// Test-only
261    pub fn set_caching_enabled(&mut self, enabled: bool) {
262        self.caching_enabled = enabled;
263    }
264
265    /// Test-only
266    pub fn set_debounce_writes(&mut self, debounce: bool) {
267        self.debounce_writes = debounce;
268    }
269
270    /// Triggers a flush on the given stash proxy.
271    async fn stash_flush(
272        stash_proxy: &StoreAccessorProxy,
273        inspect_handle: Rc<Mutex<StashInspectLogger>>,
274        setting_key: String,
275    ) {
276        let flush_result = stash_proxy.flush().await;
277        match flush_result {
278            Ok(Err(err)) => {
279                Self::handle_flush_failure(inspect_handle, setting_key, format!("{err:?}")).await;
280            }
281            Err(err) => {
282                Self::handle_flush_failure(inspect_handle, setting_key, format!("{err:?}")).await;
283            }
284            _ => {}
285        }
286    }
287
288    async fn handle_flush_failure(
289        inspect_handle: Rc<Mutex<StashInspectLogger>>,
290        setting_key: String,
291        err: String,
292    ) {
293        log::error!("Failed to flush to stash: {:?}", err);
294
295        // Record the write failure to inspect.
296        inspect_handle.lock().await.record_flush_failure(setting_key);
297    }
298
299    async fn inner_write<T>(
300        &self,
301        new_value: &T,
302        immediate_flush: bool,
303    ) -> Result<UpdateState, Error>
304    where
305        T: DeviceStorageConvertible,
306    {
307        let storable = new_value.get_storable();
308        let key = T::Storable::KEY;
309        let serialized_value = storable.serialize_to();
310        let data_as_any = Box::new(storable.into_owned()) as Box<TypeErasedData>;
311        let mapping_fn: MappingFn = Box::new(|any: &dyn Any| {
312            // Attempt to downcast the `dyn Any` to its original type. If `T` was not its
313            // original type, then we want to panic because there's a compile-time issue
314            // with overlapping keys.
315            let value = any.downcast_ref::<T::Storable>().expect(
316                "Type mismatch even though keys match. Two different\
317                                    types have the same key value",
318            );
319            value.serialize_to()
320        });
321
322        let typed_storage = self
323            .typed_storage_map
324            .get(key)
325            .ok_or_else(|| format_err!("Invalid data keyed by {}", key))?;
326        let mut cached_storage = typed_storage.cached_storage.lock().await;
327        let mut maybe_init;
328        let cached_value = {
329            maybe_init = cached_storage
330                .current_data
331                .as_deref()
332                // Get the data as a shared reference so we don't move out of the option.
333                .map(mapping_fn);
334            if maybe_init.is_none() {
335                let stash_key = prefixed(key);
336                if let Some(stash_value) =
337                    cached_storage.stash_proxy.get_value(&stash_key).await.unwrap_or_else(|_| {
338                        panic!("failed to get value from stash for {stash_key:?}")
339                    })
340                {
341                    if let Value::Stringval(string_value) = &*stash_value {
342                        maybe_init = Some(string_value.clone());
343                    } else {
344                        panic!("Unexpected type for key found in stash");
345                    }
346                }
347            }
348            maybe_init.as_ref()
349        };
350
351        Ok(if cached_value != Some(&serialized_value) {
352            let serialized = Value::Stringval(serialized_value);
353            let key = prefixed(key);
354            cached_storage.stash_proxy.set_value(&key, serialized)?;
355            if immediate_flush {
356                DeviceStorage::stash_flush(
357                    &cached_storage.stash_proxy,
358                    Rc::clone(&self.inspect_handle),
359                    key,
360                )
361                .await;
362            } else {
363                typed_storage.flush_sender.unbounded_send(()).with_context(|| {
364                    format!("flush_sender failed to send flush message, associated key is {key}")
365                })?;
366            }
367            cached_storage.current_data = Some(data_as_any);
368            UpdateState::Updated
369        } else {
370            UpdateState::Unchanged
371        })
372    }
373
374    /// Write `new_value` to storage. The write will be persisted to disk at the normal debounce
375    /// interval which is configured for this storage instance.
376    pub async fn write<T>(&self, new_value: &T) -> Result<UpdateState, Error>
377    where
378        T: DeviceStorageConvertible,
379    {
380        self.inner_write(new_value, !self.debounce_writes).await
381    }
382
383    /// Write `new_value` to storage. The write will be persisted to disk immediately, overriding
384    /// the configured debounce timer for this storage instance.
385    pub async fn immediate_write<T>(&self, new_value: &T) -> Result<UpdateState, Error>
386    where
387        T: DeviceStorageConvertible,
388    {
389        self.inner_write(new_value, true).await
390    }
391
392    /// Test-only method to write directly to stash without touching the cache. This is used for
393    /// setting up data as if it existed on disk before the connection to stash was made.
394    pub async fn write_str(&self, key: &'static str, value: String) -> Result<(), Error> {
395        let typed_storage =
396            self.typed_storage_map.get(key).expect("Did not request an initialized key");
397        let cached_storage = typed_storage.cached_storage.lock().await;
398        cached_storage.stash_proxy.set_value(&prefixed(key), Value::Stringval(value))?;
399        typed_storage.flush_sender.unbounded_send(()).unwrap();
400        Ok(())
401    }
402
403    async fn get_inner(
404        &self,
405        key: &'static str,
406    ) -> (MutexGuard<'_, CachedStorage>, Option<Option<String>>) {
407        let typed_storage = self
408            .typed_storage_map
409            .get(key)
410            // TODO(https://fxbug.dev/42064613) Replace this with an error result.
411            .unwrap_or_else(|| panic!("Invalid data keyed by {key}"));
412        let cached_storage = typed_storage.cached_storage.lock().await;
413        let new = if cached_storage.current_data.is_none() || !self.caching_enabled {
414            let stash_key = prefixed(key);
415            if let Some(stash_value) = cached_storage
416                .stash_proxy
417                .get_value(&stash_key)
418                .await
419                .unwrap_or_else(|_| panic!("failed to get value from stash for {stash_key:?}"))
420            {
421                if let Value::Stringval(string_value) = *stash_value {
422                    Some(Some(string_value))
423                } else {
424                    panic!("Unexpected type for key found in stash");
425                }
426            } else {
427                Some(None)
428            }
429        } else {
430            None
431        };
432
433        (cached_storage, new)
434    }
435
436    /// Gets the latest value cached locally, or loads the value from storage.
437    /// Doesn't support multiple concurrent callers of the same struct.
438    pub async fn get<T>(&self) -> T::Storable
439    where
440        T: DeviceStorageConvertible,
441    {
442        let (mut cached_storage, update) = self.get_inner(T::Storable::KEY).await;
443        if let Some(update) = update {
444            cached_storage.current_data = Some(update.and_then(|string_value| {
445                T::Storable::try_deserialize_from(&string_value).map(|val| Box::new(val) as Box<TypeErasedData>).map_err(|e| log::error!(
446                    "Using default. Failed to deserialize type {}: {e:?}\nSource data: {string_value:?}",
447                    T::Storable::KEY
448                )).ok()
449            }).unwrap_or_else(|| Box::new(<<T::Storable as DeviceStorageCompatible>::Loader as DefaultDispatcher<T::Storable>>::get_default(self)) as Box<TypeErasedData>));
450        };
451
452        cached_storage
453            .current_data
454            .as_ref()
455            .expect("should always have a value")
456            .downcast_ref::<T::Storable>()
457            .expect(
458                "Type mismatch even though keys match. Two different types have the same key\
459                     value",
460            )
461            .clone()
462    }
463}
464
465pub trait DefaultDispatcher<T>: Sealed
466where
467    T: DeviceStorageCompatible,
468{
469    fn get_default(_: &DeviceStorage) -> T;
470}
471
472impl<T> DefaultDispatcher<T> for NoneT
473where
474    T: DeviceStorageCompatible<Loader = Self> + Default,
475{
476    fn get_default(_: &DeviceStorage) -> T {
477        T::default()
478    }
479}
480
481impl<T, L> DefaultDispatcher<T> for L
482where
483    T: DeviceStorageCompatible<Loader = L>,
484    L: DefaultLoader<Result = T> + 'static,
485{
486    fn get_default(storage: &DeviceStorage) -> T {
487        match storage.typed_loader_map.get(T::KEY) {
488            Some(loader) => match loader.downcast_ref::<T::Loader>() {
489                Some(loader) => loader.default_value(),
490                None => {
491                    panic!("Mismatch key and loader for key {}", T::KEY);
492                }
493            },
494            None => panic!("Missing loader for {}", T::KEY),
495        }
496    }
497}
498
499fn prefixed(input_string: &str) -> String {
500    format!("{SETTINGS_PREFIX}_{input_string}")
501}
502
503#[cfg(test)]
504mod tests {
505    use super::*;
506    use assert_matches::assert_matches;
507    use diagnostics_assertions::assert_data_tree;
508    use fidl_fuchsia_stash::{
509        FlushError, StoreAccessorMarker, StoreAccessorRequest, StoreAccessorRequestStream,
510    };
511    use fuchsia_async as fasync;
512    use fuchsia_async::TestExecutor;
513    use fuchsia_inspect::component;
514    use futures::prelude::*;
515    use serde::{Deserialize, Serialize};
516    use std::task::Poll;
517
518    const VALUE0: i32 = 3;
519    const VALUE1: i32 = 33;
520    const VALUE2: i32 = 128;
521
522    #[derive(PartialEq, Clone, Serialize, Deserialize, Debug)]
523    struct TestStruct {
524        value: i32,
525    }
526
527    const STORE_KEY: &str = "settings_testkey";
528
529    impl DeviceStorageCompatible for TestStruct {
530        type Loader = NoneT;
531        const KEY: &'static str = "testkey";
532    }
533
534    impl Default for TestStruct {
535        fn default() -> Self {
536            TestStruct { value: VALUE0 }
537        }
538    }
539
540    /// Advances `future` until `executor` finishes. Panics if the end result was a stall.
541    #[track_caller]
542    fn advance_executor<F>(executor: &mut TestExecutor, future: &mut F)
543    where
544        F: Future + Unpin,
545    {
546        assert!(executor.run_until_stalled(future).is_ready(), "TestExecutor stalled!");
547    }
548
549    /// Verifies that a SetValue call was sent to stash with the given value.
550    async fn verify_stash_set(stash_stream: &mut StoreAccessorRequestStream, expected_value: i32) {
551        match stash_stream.next().await.unwrap() {
552            Ok(StoreAccessorRequest::SetValue { key, val, control_handle: _ }) => {
553                assert_eq!(key, STORE_KEY);
554                if let Value::Stringval(string_value) = val {
555                    let input_value = TestStruct::try_deserialize_from(&string_value)
556                        .expect("deserialization should succeed");
557                    assert_eq!(input_value.value, expected_value);
558                } else {
559                    panic!("Unexpected type for key found in stash");
560                }
561            }
562            request => panic!("Unexpected request: {request:?}"),
563        }
564    }
565
566    /// Verifies that a SetValue call was sent to stash with the given value.
567    async fn validate_stash_get_and_respond(
568        stash_stream: &mut StoreAccessorRequestStream,
569        response: String,
570    ) {
571        match stash_stream.next().await.unwrap() {
572            Ok(StoreAccessorRequest::GetValue { key, responder }) => {
573                assert_eq!(key, STORE_KEY);
574                responder.send(Some(Value::Stringval(response))).expect("unable to send response");
575            }
576            request => panic!("Unexpected request: {request:?}"),
577        }
578    }
579
580    /// Verifies that a Flush call was sent to stash.
581    async fn verify_stash_flush(stash_stream: &mut StoreAccessorRequestStream) {
582        match stash_stream.next().await.unwrap() {
583            Ok(StoreAccessorRequest::Flush { responder }) => {
584                let _ = responder.send(Ok(()));
585            } // expected
586            request => panic!("Unexpected request: {request:?}"),
587        }
588    }
589
590    /// Verifies that a Flush call was sent to stash, and send back a failure.
591    async fn fail_stash_flush(stash_stream: &mut StoreAccessorRequestStream) {
592        match stash_stream.next().await.unwrap() {
593            Ok(StoreAccessorRequest::Flush { responder }) => {
594                let _ = responder.send(Err(FlushError::CommitFailed));
595            } // expected
596            request => panic!("Unexpected request: {request:?}"),
597        }
598    }
599
600    #[fuchsia::test(allow_stalls = false)]
601    async fn test_get() {
602        let (stash_proxy, mut stash_stream) =
603            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
604
605        fasync::Task::local(async move {
606            let value_to_get = TestStruct { value: VALUE1 };
607
608            #[allow(clippy::single_match)]
609            while let Some(req) = stash_stream.try_next().await.unwrap() {
610                #[allow(unreachable_patterns)]
611                match req {
612                    StoreAccessorRequest::GetValue { key, responder } => {
613                        assert_eq!(key, STORE_KEY);
614                        let response = Value::Stringval(value_to_get.serialize_to());
615
616                        responder.send(Some(response)).unwrap();
617                    }
618                    _ => {}
619                }
620            }
621        })
622        .detach();
623
624        let storage = DeviceStorage::with_stash_proxy(
625            vec![(TestStruct::KEY, None)],
626            move || stash_proxy.clone(),
627            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
628        );
629        let result = storage.get::<TestStruct>().await;
630
631        assert_eq!(result.value, VALUE1);
632    }
633
634    #[fuchsia::test(allow_stalls = false)]
635    async fn test_get_default() {
636        let (stash_proxy, mut stash_stream) =
637            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
638
639        fasync::Task::local(async move {
640            #[allow(clippy::single_match)]
641            while let Some(req) = stash_stream.try_next().await.unwrap() {
642                #[allow(unreachable_patterns)]
643                match req {
644                    StoreAccessorRequest::GetValue { key: _, responder } => {
645                        responder.send(None).unwrap();
646                    }
647                    _ => {}
648                }
649            }
650        })
651        .detach();
652
653        let storage = DeviceStorage::with_stash_proxy(
654            vec![(TestStruct::KEY, None)],
655            move || stash_proxy.clone(),
656            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
657        );
658        let result = storage.get::<TestStruct>().await;
659
660        assert_eq!(result.value, VALUE0);
661    }
662
663    // For an invalid stash value, the get() method should return the default value.
664    #[fuchsia::test(allow_stalls = false)]
665    async fn test_invalid_stash() {
666        let (stash_proxy, mut stash_stream) =
667            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
668
669        fasync::Task::local(async move {
670            #[allow(clippy::single_match)]
671            while let Some(req) = stash_stream.try_next().await.unwrap() {
672                #[allow(unreachable_patterns)]
673                match req {
674                    StoreAccessorRequest::GetValue { key: _, responder } => {
675                        let response = Value::Stringval("invalid value".to_string());
676                        responder.send(Some(response)).unwrap();
677                    }
678                    _ => {}
679                }
680            }
681        })
682        .detach();
683
684        let storage = DeviceStorage::with_stash_proxy(
685            vec![(TestStruct::KEY, None)],
686            move || stash_proxy.clone(),
687            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
688        );
689
690        let result = storage.get::<TestStruct>().await;
691
692        assert_eq!(result.value, VALUE0);
693    }
694
695    // Verifies that stash flush failures are written to inspect.
696    #[fuchsia::test]
697    fn test_flush_fail_writes_to_inspect() {
698        let written_value = VALUE2;
699        let mut executor = TestExecutor::new_with_fake_time();
700
701        let (stash_proxy, mut stash_stream) =
702            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
703
704        let inspector = component::inspector();
705        let logger_handle = Rc::new(Mutex::new(StashInspectLogger::new(inspector.root())));
706        let storage = DeviceStorage::with_stash_proxy(
707            vec![(TestStruct::KEY, None)],
708            move || stash_proxy.clone(),
709            logger_handle,
710        );
711
712        // Write to device storage.
713        let value_to_write = TestStruct { value: written_value };
714        let write_future = storage.write(&value_to_write);
715        futures::pin_mut!(write_future);
716
717        // Initial cache check is done if no read was ever performed.
718        assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Pending);
719
720        {
721            let respond_future = validate_stash_get_and_respond(
722                &mut stash_stream,
723                serde_json::to_string(&TestStruct::default()).unwrap(),
724            );
725            futures::pin_mut!(respond_future);
726            advance_executor(&mut executor, &mut respond_future);
727        }
728
729        // Write request finishes immediately.
730        assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Ready(Ok(_)));
731
732        // Set request is received immediately on write.
733        {
734            let set_value_future = verify_stash_set(&mut stash_stream, written_value);
735            futures::pin_mut!(set_value_future);
736            advance_executor(&mut executor, &mut set_value_future);
737        }
738
739        // Start listening for the flush request.
740        let flush_future = fail_stash_flush(&mut stash_stream);
741        futures::pin_mut!(flush_future);
742
743        // Flush is received without a wait. Due to the way time works with executors, if there was
744        // a delay, the test would stall since time never advances.
745        advance_executor(&mut executor, &mut flush_future);
746
747        // Queue up a second write to guarantee that CachedStorage has written the failure to
748        // inspect.
749        {
750            let value_to_write = TestStruct { value: VALUE1 };
751            let write_future = storage.write(&value_to_write);
752            futures::pin_mut!(write_future);
753            assert_matches!(
754                executor.run_until_stalled(&mut write_future),
755                Poll::Ready(Result::Ok(_))
756            );
757        }
758
759        // Run all background tasks until stalled.
760        let _ = executor.run_until_stalled(&mut future::pending::<()>());
761
762        assert_data_tree!(@executor executor, inspector, root: {
763            stash_failures: {
764                testkey: {
765                    count: 1u64,
766                }
767            }
768        });
769    }
770
771    // Test that an initial write to DeviceStorage causes a SetValue and Flush to Stash
772    // without any wait.
773    #[fuchsia::test]
774    fn test_first_write_flushes_immediately() {
775        let written_value = VALUE2;
776        let mut executor = TestExecutor::new_with_fake_time();
777
778        let (stash_proxy, mut stash_stream) =
779            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
780
781        let storage = DeviceStorage::with_stash_proxy(
782            vec![(TestStruct::KEY, None)],
783            move || stash_proxy.clone(),
784            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
785        );
786
787        // Write to device storage.
788        let value_to_write = TestStruct { value: written_value };
789        let write_future = storage.write(&value_to_write);
790        futures::pin_mut!(write_future);
791
792        // Initial cache check is done if no read was ever performed.
793        assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Pending);
794
795        {
796            let respond_future = validate_stash_get_and_respond(
797                &mut stash_stream,
798                serde_json::to_string(&TestStruct::default()).unwrap(),
799            );
800            futures::pin_mut!(respond_future);
801            advance_executor(&mut executor, &mut respond_future);
802        }
803
804        // Write request finishes immediately.
805        assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Ready(Ok(_)));
806
807        // Set request is received immediately on write.
808        {
809            let set_value_future = verify_stash_set(&mut stash_stream, written_value);
810            futures::pin_mut!(set_value_future);
811            advance_executor(&mut executor, &mut set_value_future);
812        }
813
814        // Start listening for the flush request.
815        let flush_future = verify_stash_flush(&mut stash_stream);
816        futures::pin_mut!(flush_future);
817
818        // Flush is received without a wait. Due to the way time works with executors, if there was
819        // a delay, the test would stall since time never advances.
820        advance_executor(&mut executor, &mut flush_future);
821    }
822
823    #[derive(Default, Copy, Clone, PartialEq, Serialize, Deserialize)]
824    struct WrongStruct;
825
826    impl DeviceStorageCompatible for WrongStruct {
827        type Loader = NoneT;
828        const KEY: &'static str = "WRONG_STRUCT";
829    }
830
831    // Test that an initial write to DeviceStorage causes a SetValue and Flush to Stash
832    // without any wait.
833    #[fuchsia::test(allow_stalls = false)]
834    async fn test_write_with_mismatch_type_returns_error() {
835        let (stash_proxy, mut stream) =
836            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
837
838        let spawned = fasync::Task::local(async move {
839            while let Some(request) = stream.next().await {
840                match request {
841                    Ok(StoreAccessorRequest::GetValue { key, responder }) => {
842                        assert_eq!(key, STORE_KEY);
843                        let _ = responder.send(Some(Value::Stringval(
844                            serde_json::to_string(&TestStruct { value: VALUE2 }).unwrap(),
845                        )));
846                    }
847                    Ok(StoreAccessorRequest::SetValue { key, .. }) => {
848                        assert_eq!(key, STORE_KEY);
849                    }
850                    _ => panic!("Unexpected request {request:?}"),
851                }
852            }
853        });
854
855        let storage = DeviceStorage::with_stash_proxy(
856            vec![(TestStruct::KEY, None)],
857            move || stash_proxy.clone(),
858            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
859        );
860
861        // Write successfully to storage once.
862        let result = storage.write(&TestStruct { value: VALUE2 }).await;
863        assert!(result.is_ok());
864
865        // Write to device storage again with a different type to validate that the type can't
866        // be changed.
867        let result = storage.write(&WrongStruct).await;
868        assert_matches!(result, Err(e) if e.to_string() == "Invalid data keyed by WRONG_STRUCT");
869
870        drop(storage);
871        spawned.await;
872    }
873
874    // Test that multiple writes to DeviceStorage will cause a SetValue each time, but will only
875    // Flush to Stash at an interval.
876    #[fuchsia::test]
877    fn test_multiple_write_debounce() {
878        // Custom executor for this test so that we can advance the clock arbitrarily and verify the
879        // state of the executor at any given point.
880        let mut executor = TestExecutor::new_with_fake_time();
881        let start_time = MonotonicInstant::from_nanos(0);
882        executor.set_fake_time(start_time);
883
884        let (stash_proxy, mut stash_stream) =
885            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
886
887        let storage = DeviceStorage::with_stash_proxy(
888            vec![(TestStruct::KEY, None)],
889            move || stash_proxy.clone(),
890            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
891        );
892
893        let first_value = VALUE1;
894        let second_value = VALUE2;
895
896        // First write finishes immediately.
897        {
898            let value_to_write = TestStruct { value: first_value };
899            let write_future = storage.write(&value_to_write);
900            futures::pin_mut!(write_future);
901
902            // Initial cache check is done if no read was ever performed.
903            assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Pending);
904
905            {
906                let respond_future = validate_stash_get_and_respond(
907                    &mut stash_stream,
908                    serde_json::to_string(&TestStruct::default()).unwrap(),
909                );
910                futures::pin_mut!(respond_future);
911                advance_executor(&mut executor, &mut respond_future);
912            }
913
914            assert_matches!(
915                executor.run_until_stalled(&mut write_future),
916                Poll::Ready(Result::Ok(_))
917            );
918        }
919
920        // First set request is received immediately on write.
921        {
922            let set_value_future = verify_stash_set(&mut stash_stream, first_value);
923            futures::pin_mut!(set_value_future);
924            advance_executor(&mut executor, &mut set_value_future);
925        }
926
927        // First flush request is received.
928        {
929            let flush_future = verify_stash_flush(&mut stash_stream);
930            futures::pin_mut!(flush_future);
931            advance_executor(&mut executor, &mut flush_future);
932        }
933
934        // Now we repeat the process with a second write request, which will need to advance the
935        // fake time due to the timer.
936
937        // Second write finishes immediately.
938        {
939            let value_to_write = TestStruct { value: second_value };
940            let write_future = storage.write(&value_to_write);
941            futures::pin_mut!(write_future);
942            assert_matches!(
943                executor.run_until_stalled(&mut write_future),
944                Poll::Ready(Result::Ok(_))
945            );
946        }
947
948        // Second set request finishes immediately on write.
949        {
950            let set_value_future = verify_stash_set(&mut stash_stream, second_value);
951            futures::pin_mut!(set_value_future);
952            advance_executor(&mut executor, &mut set_value_future);
953        }
954
955        // Start waiting for flush request.
956        let flush_future = verify_stash_flush(&mut stash_stream);
957        futures::pin_mut!(flush_future);
958
959        // TextExecutor stalls due to waiting on timer to finish.
960        assert_matches!(executor.run_until_stalled(&mut flush_future), Poll::Pending);
961
962        // Advance time to 1ms before the flush triggers.
963        executor
964            .set_fake_time(start_time + (MIN_FLUSH_INTERVAL - MonotonicDuration::from_millis(1)));
965
966        // TextExecutor is still waiting on the time to finish.
967        assert_matches!(executor.run_until_stalled(&mut flush_future), Poll::Pending);
968
969        // Advance time so that the flush will trigger.
970        executor.set_fake_time(start_time + MIN_FLUSH_INTERVAL);
971
972        // Stash receives a flush request after one timer cycle and the future terminates.
973        advance_executor(&mut executor, &mut flush_future);
974    }
975
976    // This mod includes structs to only be used by
977    // test_device_compatible_migration tests.
978    mod test_device_compatible_migration {
979        use super::*;
980        use serde::{Deserialize, Serialize};
981
982        pub(crate) const DEFAULT_V1_VALUE: i32 = 1;
983        pub(crate) const DEFAULT_CURRENT_VALUE: i32 = 2;
984        pub(crate) const DEFAULT_CURRENT_VALUE_2: i32 = 3;
985
986        #[derive(PartialEq, Clone, Serialize, Deserialize, Debug)]
987        pub(crate) struct V1 {
988            pub value: i32,
989        }
990
991        impl DeviceStorageCompatible for V1 {
992            type Loader = NoneT;
993            const KEY: &'static str = "testkey";
994        }
995
996        impl Default for V1 {
997            fn default() -> Self {
998                Self { value: DEFAULT_V1_VALUE }
999            }
1000        }
1001
1002        #[derive(PartialEq, Clone, Serialize, Deserialize, Debug)]
1003        pub(crate) struct Current {
1004            pub value: i32,
1005            pub value_2: i32,
1006        }
1007
1008        impl From<V1> for Current {
1009            fn from(v1: V1) -> Self {
1010                Current { value: v1.value, value_2: DEFAULT_CURRENT_VALUE_2 }
1011            }
1012        }
1013
1014        impl DeviceStorageCompatible for Current {
1015            type Loader = NoneT;
1016            const KEY: &'static str = "testkey2";
1017
1018            fn try_deserialize_from(value: &str) -> Result<Self, Error> {
1019                Self::extract(value).or_else(|_| V1::extract(value).map(Self::from))
1020            }
1021        }
1022
1023        impl Default for Current {
1024            fn default() -> Self {
1025                Self { value: DEFAULT_CURRENT_VALUE, value_2: DEFAULT_CURRENT_VALUE_2 }
1026            }
1027        }
1028    }
1029
1030    #[fuchsia::test]
1031    fn test_device_compatible_custom_migration() {
1032        // Create an initial struct based on the first version.
1033        let initial = test_device_compatible_migration::V1::default();
1034        // Serialize.
1035        let initial_serialized = initial.serialize_to();
1036
1037        // Deserialize using the second version.
1038        let current =
1039            test_device_compatible_migration::Current::try_deserialize_from(&initial_serialized)
1040                .expect("deserialization should succeed");
1041        // Assert values carried over from first version and defaults are used for rest.
1042        assert_eq!(current.value, test_device_compatible_migration::DEFAULT_V1_VALUE);
1043        assert_eq!(current.value_2, test_device_compatible_migration::DEFAULT_CURRENT_VALUE_2);
1044    }
1045
1046    #[fuchsia::test(allow_stalls = false)]
1047    async fn test_corrupt_get_returns_default() {
1048        let (stash_proxy, mut stash_stream) =
1049            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
1050
1051        fasync::Task::local(async move {
1052            #[allow(clippy::single_match)]
1053            while let Some(req) = stash_stream.try_next().await.unwrap() {
1054                #[allow(unreachable_patterns)]
1055                match req {
1056                    StoreAccessorRequest::GetValue { key, responder } => {
1057                        assert_eq!(
1058                            key,
1059                            format!("settings_{}", test_device_compatible_migration::Current::KEY)
1060                        );
1061                        let response = Value::Stringval("bad json".to_string());
1062                        responder.send(Some(response)).unwrap();
1063                    }
1064                    _ => {}
1065                }
1066            }
1067        })
1068        .detach();
1069
1070        let storage = DeviceStorage::with_stash_proxy(
1071            vec![(test_device_compatible_migration::Current::KEY, None)],
1072            move || stash_proxy.clone(),
1073            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
1074        );
1075        let current = storage.get::<test_device_compatible_migration::Current>().await;
1076
1077        assert_eq!(current.value, test_device_compatible_migration::DEFAULT_CURRENT_VALUE);
1078        assert_eq!(current.value_2, test_device_compatible_migration::DEFAULT_CURRENT_VALUE_2);
1079    }
1080
1081    #[fuchsia::test]
1082    fn test_write_without_debounce() {
1083        let mut executor = TestExecutor::new_with_fake_time();
1084
1085        let (stash_proxy, mut stash_stream) =
1086            fidl::endpoints::create_proxy_and_stream::<StoreAccessorMarker>();
1087
1088        let storage = DeviceStorage::with_stash_proxy(
1089            vec![(TestStruct::KEY, None)],
1090            move || stash_proxy.clone(),
1091            Rc::new(Mutex::new(StashInspectLogger::new(component::inspector().root()))),
1092        );
1093
1094        let first_value = VALUE1;
1095
1096        // Write first value to initialize cache.
1097        {
1098            let value_to_write = TestStruct { value: first_value };
1099            let write_future = storage.write(&value_to_write);
1100            futures::pin_mut!(write_future);
1101
1102            assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Pending);
1103
1104            {
1105                let respond_future = validate_stash_get_and_respond(
1106                    &mut stash_stream,
1107                    serde_json::to_string(&TestStruct::default()).unwrap(),
1108                );
1109                futures::pin_mut!(respond_future);
1110                advance_executor(&mut executor, &mut respond_future);
1111            }
1112
1113            assert_matches!(
1114                executor.run_until_stalled(&mut write_future),
1115                Poll::Ready(Result::Ok(_))
1116            );
1117        }
1118        {
1119            let set_value_future = verify_stash_set(&mut stash_stream, first_value);
1120            futures::pin_mut!(set_value_future);
1121            advance_executor(&mut executor, &mut set_value_future);
1122        }
1123        {
1124            let flush_future = verify_stash_flush(&mut stash_stream);
1125            futures::pin_mut!(flush_future);
1126            advance_executor(&mut executor, &mut flush_future);
1127        }
1128
1129        // immediate_write should immediately trigger a SetValue and a Flush without timer delay.
1130        let second_value = VALUE2;
1131        {
1132            let value_to_write = TestStruct { value: second_value };
1133            let write_future = storage.immediate_write(&value_to_write);
1134            futures::pin_mut!(write_future);
1135
1136            // Stash proxy set_value and flush happen immediately within write().
1137            assert_matches!(executor.run_until_stalled(&mut write_future), Poll::Pending);
1138
1139            {
1140                let set_value_future = verify_stash_set(&mut stash_stream, second_value);
1141                futures::pin_mut!(set_value_future);
1142                advance_executor(&mut executor, &mut set_value_future);
1143            }
1144
1145            {
1146                let flush_future = verify_stash_flush(&mut stash_stream);
1147                futures::pin_mut!(flush_future);
1148                advance_executor(&mut executor, &mut flush_future);
1149            }
1150
1151            assert_matches!(
1152                executor.run_until_stalled(&mut write_future),
1153                Poll::Ready(Result::Ok(_))
1154            );
1155        }
1156    }
1157}
settings_storage/device_storage.rs

settings_storage/
device_storage.rs
