From 81fd3adef19178a4c384465a29a77a3811ca00db Mon Sep 17 00:00:00 2001 From: Max Summe Date: Mon, 13 Apr 2026 16:09:05 -0700 Subject: [PATCH 1/4] Add Send + Sync supertraits to Value for Send futures The Value trait's lack of Send/Sync bounds caused futures returned by MapStorage methods to be !Send, since store_item_inner uses &dyn Value which requires Sync on the trait object for the reference to be Send. Closes #125 --- CHANGELOG.md | 2 ++ src/map.rs | 18 +++++++++++++++++- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ace197d..e3fbd18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ## Unreleased +- *Breaking:* Added `Send + Sync` supertraits to `Value` so that futures returned by `MapStorage` methods are `Send` + ## 7.2.0 - 24-03-26 - Added a RAM-buffered queue diff --git a/src/map.rs b/src/map.rs index 9d88ec9..33f054e 100644 --- a/src/map.rs +++ b/src/map.rs @@ -1143,7 +1143,7 @@ impl Key for () { /// /// It also carries a lifetime so that zero-copy deserialization is supported. /// Zero-copy serialization is not supported due to technical restrictions. -pub trait Value<'a> { +pub trait Value<'a>: Send + Sync { /// Serialize the value into the given buffer. If everything went ok, this function returns the length /// of the used part of the buffer. fn serialize_into(&self, buffer: &mut [u8]) -> Result; @@ -1887,4 +1887,20 @@ mod tests { Ok((Foo(123), 1)) ); } + + /// Compile-time check: the future returned by `store_item` must be `Send` + /// when all components are `Send`. See https://github.com/tweedegolf/sequential-storage/issues/125 + fn _assert_store_item_future_is_send() { + fn assert_send(_t: T) {} + + let mut storage = MapStorage::::new( + MockFlashBig::default(), + MapConfig::new(0x000..0x1000), + NoCache::new(), + ); + let mut data_buffer = AlignedBuf([0; 128]); + + assert_send(storage.store_item(&mut data_buffer, &0u8, &42u32)); + } + } From df6e1a2317e459767ef93fad7e7ea7befeaa609b Mon Sep 17 00:00:00 2001 From: Max Summe Date: Tue, 14 Apr 2026 09:19:36 -0700 Subject: [PATCH 2/4] Fix CI: require Send + Sync on postcard blanket impl, add fetch_item Send check - Add Send + Sync bounds to the PostcardValue blanket impl's where clause so types using postcard via the Value trait satisfy the new supertraits. - Extend the compile-time Send check to also cover fetch_item. - Remove trailing blank line flagged by rustfmt. --- src/map.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/map.rs b/src/map.rs index 33f054e..341a6d3 100644 --- a/src/map.rs +++ b/src/map.rs @@ -1238,7 +1238,7 @@ pub trait PostcardValue<'a>: Serialize + Deserialize<'a> {} #[cfg(feature = "postcard")] impl<'a, T> Value<'a> for T where - T: PostcardValue<'a>, + T: PostcardValue<'a> + Send + Sync, { fn serialize_into(&self, buffer: &mut [u8]) -> Result { Ok(postcard::to_slice(self, buffer).map(|s| s.len())?) @@ -1888,9 +1888,9 @@ mod tests { ); } - /// Compile-time check: the future returned by `store_item` must be `Send` - /// when all components are `Send`. See https://github.com/tweedegolf/sequential-storage/issues/125 - fn _assert_store_item_future_is_send() { + /// Compile-time check: the futures returned by `store_item` and `fetch_item` + /// must be `Send`. See https://github.com/tweedegolf/sequential-storage/issues/125 + fn _assert_public_futures_are_send() { fn assert_send(_t: T) {} let mut storage = MapStorage::::new( @@ -1901,6 +1901,6 @@ mod tests { let mut data_buffer = AlignedBuf([0; 128]); assert_send(storage.store_item(&mut data_buffer, &0u8, &42u32)); + assert_send(storage.fetch_item::(&mut data_buffer, &0u8)); } - } From e8475e8f0557c2bc5d9b630e47de24766a742682 Mon Sep 17 00:00:00 2001 From: Max Summe Date: Thu, 16 Apr 2026 16:11:29 -0700 Subject: [PATCH 3/4] Move Send+Sync bound from Value trait to store_item parameter --- example/Cargo.lock | 2 +- src/map.rs | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/example/Cargo.lock b/example/Cargo.lock index 46eb5c3..98e2b5e 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -578,7 +578,7 @@ checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" [[package]] name = "sequential-storage" -version = "7.1.0" +version = "7.2.0" dependencies = [ "defmt", "embedded-storage-async", diff --git a/src/map.rs b/src/map.rs index 341a6d3..9c76db8 100644 --- a/src/map.rs +++ b/src/map.rs @@ -369,7 +369,7 @@ impl, K: Key> MapStorage { /// /// The data buffer must be long enough to hold the longest serialized data of your [Key] + [Value] types combined, /// rounded up to flash word alignment. - pub async fn store_item<'d, V: Value<'d>>( + pub async fn store_item<'d, V: Value<'d> + Send + Sync>( &mut self, data_buffer: &mut [u8], key: &K, @@ -385,7 +385,7 @@ impl, K: Key> MapStorage { &mut self, data_buffer: &mut [u8], key: &K, - item: &dyn Value<'_>, + item: &(dyn Value<'_> + Send + Sync), ) -> Result<(), Error> { if self.inner.cache.is_dirty() { self.inner.cache.invalidate_cache_state(); @@ -1143,7 +1143,7 @@ impl Key for () { /// /// It also carries a lifetime so that zero-copy deserialization is supported. /// Zero-copy serialization is not supported due to technical restrictions. -pub trait Value<'a>: Send + Sync { +pub trait Value<'a> { /// Serialize the value into the given buffer. If everything went ok, this function returns the length /// of the used part of the buffer. fn serialize_into(&self, buffer: &mut [u8]) -> Result; @@ -1238,7 +1238,7 @@ pub trait PostcardValue<'a>: Serialize + Deserialize<'a> {} #[cfg(feature = "postcard")] impl<'a, T> Value<'a> for T where - T: PostcardValue<'a> + Send + Sync, + T: PostcardValue<'a>, { fn serialize_into(&self, buffer: &mut [u8]) -> Result { Ok(postcard::to_slice(self, buffer).map(|s| s.len())?) From 5dd764cc062245fce416f7fc74240bd432fad765 Mon Sep 17 00:00:00 2001 From: Max Summe Date: Thu, 16 Apr 2026 16:17:35 -0700 Subject: [PATCH 4/4] Add store_item_local for non-Sync Value types --- src/map.rs | 347 ++++++++++++++++++++++++++++++++--------------------- 1 file changed, 209 insertions(+), 138 deletions(-) diff --git a/src/map.rs b/src/map.rs index 9c76db8..deca771 100644 --- a/src/map.rs +++ b/src/map.rs @@ -139,6 +139,157 @@ pub struct MapStorage> { _phantom: PhantomData, } +/// Shared body for [`MapStorage::store_item_inner`] and [`MapStorage::store_item_inner_local`]. +/// The two differ only in the trait-object bound on `item` (Send+Sync vs bare); the logic +/// is identical, so it lives here as a macro to keep a single source of truth. +macro_rules! store_item_body { + ($this:ident, $data_buffer:ident, $key:ident, $item:ident) => {{ + if $this.inner.cache.is_dirty() { + $this.inner.cache.invalidate_cache_state(); + } + + let mut recursion_level = 0; + loop { + // Check if we're in an infinite recursion which happens when we don't have enough space to store the new data + if recursion_level == $this.inner.get_pages(0).count() { + $this.inner.cache.unmark_dirty(); + return Err(Error::FullStorage); + } + + // If there is a partial open page, we try to write in that first if there is enough space + let next_page_to_use = if let Some(partial_open_page) = $this + .inner + .find_first_page(0, PageState::PartialOpen) + .await? + { + // We found a partial open page, but at this point it's relatively cheap to do a consistency check + if !$this + .inner + .get_page_state($this.inner.next_page(partial_open_page)) + .await? + .is_open() + { + // Oh oh, the next page which serves as the buffer page is not open. We're corrupt. + // This likely happened because of an unexpected shutdown during data migration from the + // then new buffer page to the new partial open page. + // The repair function should be able to repair this. + return Err(Error::Corrupted { + #[cfg(feature = "_test")] + backtrace: std::backtrace::Backtrace::capture(), + }); + } + + // We've got to search where the free space is since the page starts with items present already + + let page_data_start_address = + calculate_page_address::($this.flash_range(), partial_open_page) + + S::WORD_SIZE as u32; + let page_data_end_address = + calculate_page_end_address::($this.flash_range(), partial_open_page) + - S::WORD_SIZE as u32; + + let key_len = $key.serialize_into($data_buffer)?; + let item_data_length = key_len + + $item + .serialize_into(&mut $data_buffer[key_len..]) + .map_err(Error::SerializationError)?; + + if item_data_length > u16::MAX as usize + || item_data_length + > calculate_page_size::() + .saturating_sub(ItemHeader::data_address::(0) as usize) + { + $this.inner.cache.unmark_dirty(); + return Err(Error::ItemTooBig); + } + + let free_spot_address = $this + .inner + .find_next_free_item_spot( + page_data_start_address, + page_data_end_address, + item_data_length as u32, + ) + .await?; + + if let Some(free_spot_address) = free_spot_address { + $this + .inner + .cache + .notice_key_location($key, free_spot_address, true); + Item::write_new( + &mut $this.inner.flash, + $this.inner.flash_range.clone(), + &mut $this.inner.cache, + free_spot_address, + &$data_buffer[..item_data_length], + ) + .await?; + + $this.inner.cache.unmark_dirty(); + return Ok(()); + } + + // The item doesn't fit here, so we need to close this page and move to the next + $this.inner.close_page(partial_open_page).await?; + Some($this.inner.next_page(partial_open_page)) + } else { + None + }; + + // If we get here, there was no partial page found or the partial page has now been closed because the item didn't fit. + // If there was a partial page, then we need to look at the next page. It's supposed to be open since it was the previous empty buffer page. + // The new buffer page has to be emptied if it was closed. + // If there was no partial page, we just use the first open page. + + if let Some(next_page_to_use) = next_page_to_use { + let next_page_state = $this.inner.get_page_state(next_page_to_use).await?; + + if !next_page_state.is_open() { + // What was the previous buffer page was not open... + return Err(Error::Corrupted { + #[cfg(feature = "_test")] + backtrace: std::backtrace::Backtrace::capture(), + }); + } + + // Since we're gonna write data here, let's already partially close the page + // This could be done after moving the data, but this is more robust in the + // face of shutdowns and cancellations + $this.inner.partial_close_page(next_page_to_use).await?; + + let next_buffer_page = $this.inner.next_page(next_page_to_use); + let next_buffer_page_state = + $this.inner.get_page_state(next_buffer_page).await?; + + if !next_buffer_page_state.is_open() { + $this + .migrate_items($data_buffer, next_buffer_page, next_page_to_use) + .await?; + } + } else { + // There's no partial open page, so we just gotta turn the first open page into a partial open one + let Some(first_open_page) = + $this.inner.find_first_page(0, PageState::Open).await? + else { + // Uh oh, no open pages. + // Something has gone wrong. + // We should never get here. + return Err(Error::Corrupted { + #[cfg(feature = "_test")] + backtrace: std::backtrace::Backtrace::capture(), + }); + }; + + $this.inner.partial_close_page(first_open_page).await?; + } + + // If we get here, we just freshly partially closed a new page, so the next loop iteration should succeed. + recursion_level += 1; + } + }}; +} + impl, K: Key> MapStorage { /// Create a new map instance /// @@ -369,6 +520,10 @@ impl, K: Key> MapStorage { /// /// The data buffer must be long enough to hold the longest serialized data of your [Key] + [Value] types combined, /// rounded up to flash word alignment. + /// + /// The returned future is `Send`, so this method is usable from multi-threaded executors. + /// If your `Value` type isn't `Send + Sync` (e.g. it contains a [`core::cell::RefCell`]), + /// use [`Self::store_item_local`] instead. pub async fn store_item<'d, V: Value<'d> + Send + Sync>( &mut self, data_buffer: &mut [u8], @@ -381,151 +536,39 @@ impl, K: Key> MapStorage { ) } + /// Like [`Self::store_item`], but works with `Value` types that aren't `Send + Sync`. + /// + /// The returned future is **not** `Send`, so this method is only suitable for + /// single-threaded executors. In exchange, `V` may contain non-`Sync` interior + /// mutability (e.g. [`core::cell::RefCell`]). + pub async fn store_item_local<'d, V: Value<'d>>( + &mut self, + data_buffer: &mut [u8], + key: &K, + item: &V, + ) -> Result<(), Error> { + run_with_auto_repair!( + function = self.store_item_inner_local(data_buffer, key, item).await, + repair = self.try_repair(data_buffer).await? + ) + } + async fn store_item_inner( &mut self, data_buffer: &mut [u8], key: &K, item: &(dyn Value<'_> + Send + Sync), ) -> Result<(), Error> { - if self.inner.cache.is_dirty() { - self.inner.cache.invalidate_cache_state(); - } - - let mut recursion_level = 0; - loop { - // Check if we're in an infinite recursion which happens when we don't have enough space to store the new data - if recursion_level == self.inner.get_pages(0).count() { - self.inner.cache.unmark_dirty(); - return Err(Error::FullStorage); - } - - // If there is a partial open page, we try to write in that first if there is enough space - let next_page_to_use = if let Some(partial_open_page) = self - .inner - .find_first_page(0, PageState::PartialOpen) - .await? - { - // We found a partial open page, but at this point it's relatively cheap to do a consistency check - if !self - .inner - .get_page_state(self.inner.next_page(partial_open_page)) - .await? - .is_open() - { - // Oh oh, the next page which serves as the buffer page is not open. We're corrupt. - // This likely happened because of an unexpected shutdown during data migration from the - // then new buffer page to the new partial open page. - // The repair function should be able to repair this. - return Err(Error::Corrupted { - #[cfg(feature = "_test")] - backtrace: std::backtrace::Backtrace::capture(), - }); - } - - // We've got to search where the free space is since the page starts with items present already - - let page_data_start_address = - calculate_page_address::(self.flash_range(), partial_open_page) - + S::WORD_SIZE as u32; - let page_data_end_address = - calculate_page_end_address::(self.flash_range(), partial_open_page) - - S::WORD_SIZE as u32; - - let key_len = key.serialize_into(data_buffer)?; - let item_data_length = key_len - + item - .serialize_into(&mut data_buffer[key_len..]) - .map_err(Error::SerializationError)?; - - if item_data_length > u16::MAX as usize - || item_data_length - > calculate_page_size::() - .saturating_sub(ItemHeader::data_address::(0) as usize) - { - self.inner.cache.unmark_dirty(); - return Err(Error::ItemTooBig); - } - - let free_spot_address = self - .inner - .find_next_free_item_spot( - page_data_start_address, - page_data_end_address, - item_data_length as u32, - ) - .await?; - - if let Some(free_spot_address) = free_spot_address { - self.inner - .cache - .notice_key_location(key, free_spot_address, true); - Item::write_new( - &mut self.inner.flash, - self.inner.flash_range.clone(), - &mut self.inner.cache, - free_spot_address, - &data_buffer[..item_data_length], - ) - .await?; - - self.inner.cache.unmark_dirty(); - return Ok(()); - } - - // The item doesn't fit here, so we need to close this page and move to the next - self.inner.close_page(partial_open_page).await?; - Some(self.inner.next_page(partial_open_page)) - } else { - None - }; - - // If we get here, there was no partial page found or the partial page has now been closed because the item didn't fit. - // If there was a partial page, then we need to look at the next page. It's supposed to be open since it was the previous empty buffer page. - // The new buffer page has to be emptied if it was closed. - // If there was no partial page, we just use the first open page. - - if let Some(next_page_to_use) = next_page_to_use { - let next_page_state = self.inner.get_page_state(next_page_to_use).await?; - - if !next_page_state.is_open() { - // What was the previous buffer page was not open... - return Err(Error::Corrupted { - #[cfg(feature = "_test")] - backtrace: std::backtrace::Backtrace::capture(), - }); - } - - // Since we're gonna write data here, let's already partially close the page - // This could be done after moving the data, but this is more robust in the - // face of shutdowns and cancellations - self.inner.partial_close_page(next_page_to_use).await?; - - let next_buffer_page = self.inner.next_page(next_page_to_use); - let next_buffer_page_state = self.inner.get_page_state(next_buffer_page).await?; - - if !next_buffer_page_state.is_open() { - self.migrate_items(data_buffer, next_buffer_page, next_page_to_use) - .await?; - } - } else { - // There's no partial open page, so we just gotta turn the first open page into a partial open one - let Some(first_open_page) = self.inner.find_first_page(0, PageState::Open).await? - else { - // Uh oh, no open pages. - // Something has gone wrong. - // We should never get here. - return Err(Error::Corrupted { - #[cfg(feature = "_test")] - backtrace: std::backtrace::Backtrace::capture(), - }); - }; - - self.inner.partial_close_page(first_open_page).await?; - } + store_item_body!(self, data_buffer, key, item) + } - // If we get here, we just freshly partially closed a new page, so the next loop iteration should succeed. - recursion_level += 1; - } + async fn store_item_inner_local( + &mut self, + data_buffer: &mut [u8], + key: &K, + item: &dyn Value<'_>, + ) -> Result<(), Error> { + store_item_body!(self, data_buffer, key, item) } /// Fully remove an item. Additional calls to fetch with the same key will return None until @@ -1903,4 +1946,32 @@ mod tests { assert_send(storage.store_item(&mut data_buffer, &0u8, &42u32)); assert_send(storage.fetch_item::(&mut data_buffer, &0u8)); } + + /// Compile-time check: `store_item_local` accepts non-`Sync` value types + /// (like those containing a `RefCell`). The returned future is intentionally + /// not required to be `Send`. + fn _assert_store_item_local_accepts_non_sync() { + use core::cell::RefCell; + + struct NotSync(RefCell); + impl<'a> Value<'a> for NotSync { + fn serialize_into(&self, buffer: &mut [u8]) -> Result { + ::serialize_into(&self.0.borrow(), buffer) + } + fn deserialize_from(buffer: &'a [u8]) -> Result<(Self, usize), SerializationError> { + let (v, n) = ::deserialize_from(buffer)?; + Ok((NotSync(RefCell::new(v)), n)) + } + } + + let mut storage = MapStorage::::new( + MockFlashBig::default(), + MapConfig::new(0x000..0x1000), + NoCache::new(), + ); + let mut data_buffer = AlignedBuf([0; 128]); + let item = NotSync(RefCell::new(42)); + + let _fut = storage.store_item_local(&mut data_buffer, &0u8, &item); + } }