From 03e09d240c7bdbcafd7ceaa462b62ffefea64f8c Mon Sep 17 00:00:00 2001 From: ldm0 Date: Fri, 30 Oct 2026 22:36:39 +0800 Subject: [PATCH] feat(storage): add asynchronous SQLite Web Storage areas and flush barriers --- Cargo.lock | 5 + moli-core/src/runtime/storage_partition.rs | 12 +- .../conn/dispatch_tests/browser_context.rs | 7 +- .../src/conn/page_state/page_targets.rs | 19 +- .../src/conn/state/browser_context.rs | 12 +- moli-protocol/src/conn/state/tests.rs | 4 +- .../src/conn/tests/resource_runtime.rs | 2 +- .../src/domains/dom_storage/commands.rs | 23 +- moli-protocol/src/domains/storage/tests.rs | 52 +- moli-renderer-v8/Cargo.toml | 1 + .../src/context_bootstrap/web_storage.rs | 2 + .../web_storage/callbacks.rs | 24 +- .../context_bootstrap/web_storage/helpers.rs | 168 +++-- .../web_storage/interceptors.rs | 148 ++--- .../context_bootstrap/web_storage/leases.rs | 87 +++ .../window_runtime/navigator.rs | 45 +- .../src/native_bridge/context_host/popups.rs | 4 +- moli-renderer-v8/src/script_vm/tests/mod.rs | 1 + .../src/script_vm/tests/sqlite_web_storage.rs | 65 ++ moli-sqlite/Cargo.toml | 5 +- moli-sqlite/src/lib.rs | 2 + moli-sqlite/src/writeback.rs | 438 +++++++++++++ moli-sqlite/src/writeback/tests.rs | 254 ++++++++ moli-web-storage/Cargo.toml | 5 + moli-web-storage/src/lib.rs | 587 ++---------------- moli-web-storage/src/sqlite.rs | 468 ++++++++++++++ moli-web-storage/src/sqlite/tests.rs | 239 +++++++ moli-web-storage/src/store.rs | 499 +++++++++++++++ 28 files changed, 2379 insertions(+), 799 deletions(-) create mode 100644 moli-renderer-v8/src/context_bootstrap/web_storage/leases.rs create mode 100644 moli-renderer-v8/src/script_vm/tests/sqlite_web_storage.rs create mode 100644 moli-sqlite/src/writeback.rs create mode 100644 moli-sqlite/src/writeback/tests.rs create mode 100644 moli-web-storage/src/sqlite.rs create mode 100644 moli-web-storage/src/sqlite/tests.rs create mode 100644 moli-web-storage/src/store.rs diff --git a/Cargo.lock b/Cargo.lock index 043f7bb790..94daa4105d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3444,6 +3444,7 @@ dependencies = [ "stylo_dom", "stylo_static_prefs", "stylo_traits", + "tempfile", "thiserror", "tokio", "tracing", @@ -3521,6 +3522,7 @@ version = "0.1.0" dependencies = [ "parking_lot", "rusqlite", + "tempfile", "tokio", ] @@ -3702,10 +3704,13 @@ version = "0.1.0" dependencies = [ "anyhow", "moli-browser-profile", + "moli-sqlite", "moli-storage-key", "parking_lot", + "rusqlite", "serde", "serde_json", + "tempfile", "typed-num", ] diff --git a/moli-core/src/runtime/storage_partition.rs b/moli-core/src/runtime/storage_partition.rs index 8b7eba3d79..ca4f435801 100644 --- a/moli-core/src/runtime/storage_partition.rs +++ b/moli-core/src/runtime/storage_partition.rs @@ -749,23 +749,19 @@ mod tests { { let partition = StoragePartitionState::open(Some(&profile_dir.path))?; let local_store = partition.web_storage_store(); - local_store - .lock() - .set_item(origin, "local", "profile-backed"); + local_store.set_item(origin, "local", "profile-backed"); let session_store = partition.session_storage_store(); - session_store - .lock() - .set_item(origin, "session", "memory-only"); + session_store.set_item(origin, "session", "memory-only"); } let reopened = StoragePartitionState::open(Some(&profile_dir.path))?; let local_store = reopened.web_storage_store(); let session_store = reopened.session_storage_store(); assert_eq!( - local_store.lock().get_item(origin, "local").as_deref(), + local_store.get_item(origin, "local").as_deref(), Some("profile-backed") ); - assert_eq!(session_store.lock().get_item(origin, "session"), None); + assert_eq!(session_store.get_item(origin, "session"), None); Ok(()) } diff --git a/moli-protocol/src/conn/dispatch_tests/browser_context.rs b/moli-protocol/src/conn/dispatch_tests/browser_context.rs index 818b1b7c81..42b7f11974 100644 --- a/moli-protocol/src/conn/dispatch_tests/browser_context.rs +++ b/moli-protocol/src/conn/dispatch_tests/browser_context.rs @@ -5,11 +5,7 @@ async fn devtools_browser_context_create_uses_ephemeral_storage_partition() { let initial_storage_owner = StoragePartitionState::open(None).expect("memory partition"); let initial_shared_storage = initial_storage_owner.shared_storage_handles(); let initial_local_storage = initial_shared_storage.web_storage_store(); - assert!(initial_local_storage.lock().set_item( - "https://example.com", - "profile-key", - "profile-value" - )); + assert!(initial_local_storage.set_item("https://example.com", "profile-key", "profile-value")); let initial_storage_partition = CdpInitialStoragePartition::from_storage_partition( vec![stored_cookie_for_dispatch_test("sid", "seeded")], &initial_storage_owner, @@ -56,7 +52,6 @@ async fn devtools_browser_context_create_uses_ephemeral_storage_partition() { assert_eq!( created_context .web_storage_store_for_test() - .lock() .get_item("https://example.com", "profile-key"), None ); diff --git a/moli-protocol/src/conn/page_state/page_targets.rs b/moli-protocol/src/conn/page_state/page_targets.rs index 421e097511..b18606be0e 100644 --- a/moli-protocol/src/conn/page_state/page_targets.rs +++ b/moli-protocol/src/conn/page_state/page_targets.rs @@ -1371,11 +1371,11 @@ mod tests { context.set_active_target_id("TID-first"); let first_storage = context.page_storage_handles(); { - let mut local_storage = first_storage.web_storage_store.lock(); + let local_storage = &first_storage.web_storage_store; assert!(local_storage.set_item("https://same.test", "local", "shared")); } { - let mut session_storage = first_storage.session_storage_store.lock(); + let session_storage = &first_storage.session_storage_store; assert!(session_storage.set_item("https://same.test", "session", "first")); } @@ -1401,14 +1401,12 @@ mod tests { assert_eq!( second_storage .web_storage_store - .lock() .get_item("https://same.test", "local"), Some("shared".to_owned()) ); assert_eq!( second_storage .session_storage_store - .lock() .get_item("https://same.test", "session"), None ); @@ -1419,7 +1417,7 @@ mod tests { let mut context = BrowserContext::new("BC-popup-storage".to_owned()); context.set_active_target_id("TID-opener"); let opener_storage = context.page_storage_handles(); - assert!(opener_storage.session_storage_store.lock().set_item( + assert!(opener_storage.session_storage_store.set_item( "https://same.test", "session", "opener" @@ -1446,11 +1444,10 @@ mod tests { assert_eq!( popup_storage .session_storage_store - .lock() .get_item("https://same.test", "session"), Some("opener".to_owned()) ); - assert!(popup_storage.session_storage_store.lock().set_item( + assert!(popup_storage.session_storage_store.set_item( "https://same.test", "session", "popup" @@ -1458,7 +1455,6 @@ mod tests { assert_eq!( opener_storage .session_storage_store - .lock() .get_item("https://same.test", "session"), Some("opener".to_owned()) ); @@ -1469,11 +1465,7 @@ mod tests { let mut context = BrowserContext::new("BC-deactivated-storage".to_owned()); context.set_active_target_id("TID-first"); let first_session_storage = context.page_storage_handles().session_storage_store.clone(); - assert!( - first_session_storage - .lock() - .set_item("https://same.test", "session", "first") - ); + assert!(first_session_storage.set_item("https://same.test", "session", "first")); context.stage_foreground_target( "TID-second".to_owned(), @@ -1497,7 +1489,6 @@ mod tests { assert_eq!( first_target_storage .session_storage_store - .lock() .get_item("https://same.test", "session"), Some("first".to_owned()) ); diff --git a/moli-protocol/src/conn/state/browser_context.rs b/moli-protocol/src/conn/state/browser_context.rs index 44b1d3e790..b7dc6e5fe6 100644 --- a/moli-protocol/src/conn/state/browser_context.rs +++ b/moli-protocol/src/conn/state/browser_context.rs @@ -756,8 +756,12 @@ impl BrowserContext { serialized_origin: &str, ) -> Result { let local_storage_usage = { - let store = self.storage_partition.web_storage_store().lock(); - usize_to_u64_saturating(store.usage_bytes_for_origin_areas(serialized_origin)) + let store = self.storage_partition.web_storage_store(); + usize_to_u64_saturating( + store + .try_usage_bytes_for_origin_areas(serialized_origin) + .map_err(|error| error.to_string())?, + ) }; let indexed_db_usage = moli_core::storage::indexed_db_origins_with_prefix_usage_bytes( self.storage_partition.indexed_db_manager(), @@ -1370,7 +1374,7 @@ impl BrowserContext { let serialized_origin = origin.origin().ascii_serialization(); if options.local_storage { - let mut store = self.storage_partition.web_storage_store().lock(); + let store = self.storage_partition.web_storage_store(); store .try_clear_origin_areas(&serialized_origin) .map_err(|error| format!("FailedToClearLocalStorage: {error}"))?; @@ -1420,7 +1424,7 @@ impl BrowserContext { let serialized_storage_key = storage_key.serialized_storage_key(); if options.local_storage { - let mut store = self.storage_partition.web_storage_store().lock(); + let store = self.storage_partition.web_storage_store(); store .try_clear_origin(&serialized_storage_key) .map_err(|error| format!("FailedToClearLocalStorage: {error}"))?; diff --git a/moli-protocol/src/conn/state/tests.rs b/moli-protocol/src/conn/state/tests.rs index b761b973ea..f6204c6d72 100644 --- a/moli-protocol/src/conn/state/tests.rs +++ b/moli-protocol/src/conn/state/tests.rs @@ -1454,7 +1454,7 @@ fn browser_context_clears_origin_site_data_through_partition_owner() { ); assert_eq!(context.snapshot_cookies().len(), 2); { - let mut store = context.web_storage_store_for_test().lock(); + let store = context.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "1")); assert!(store.set_item(&sibling_storage_key, "local", "2")); } @@ -1475,7 +1475,7 @@ fn browser_context_clears_origin_site_data_through_partition_owner() { let cookies = context.snapshot_cookies(); assert_eq!(cookies.len(), 1); assert_eq!(cookies[0].name, "sibling"); - let mut store = context.web_storage_store_for_test().lock(); + let store = context.web_storage_store_for_test(); assert_eq!(store.get_item(&storage_key, "local"), None); assert_eq!( store.get_item(&sibling_storage_key, "local"), diff --git a/moli-protocol/src/conn/tests/resource_runtime.rs b/moli-protocol/src/conn/tests/resource_runtime.rs index 55620e7751..43017f0dd1 100644 --- a/moli-protocol/src/conn/tests/resource_runtime.rs +++ b/moli-protocol/src/conn/tests/resource_runtime.rs @@ -428,7 +428,7 @@ fn browser_context_storage_usage_reads_storage_partition_owner() { moli_storage_key::MoliStorageKey::first_party_from_url(&Url::parse(&origin).unwrap(), None) .serialized_storage_key(); { - let mut store = browser_context.web_storage_store_for_test().lock(); + let store = browser_context.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "owner")); } diff --git a/moli-protocol/src/domains/dom_storage/commands.rs b/moli-protocol/src/domains/dom_storage/commands.rs index 659f6b81b7..54c20e6112 100644 --- a/moli-protocol/src/domains/dom_storage/commands.rs +++ b/moli-protocol/src/domains/dom_storage/commands.rs @@ -128,11 +128,9 @@ fn enable_command_output_plan( }; let subscription = handles .web_storage_store - .lock() .subscribe_mutations(WebStorageAreaKind::Local); handles .session_storage_store - .lock() .add_mutation_subscription(WebStorageAreaKind::Session, &subscription); let enabled = conn.with_target_devtools_session_state_for_session_mut(session_id, |state| { @@ -321,19 +319,22 @@ fn complete_storage_operation( } else { handles.session_storage_store }; - let mut store = store.lock(); match operation { DomStorageOperation::Clear => mutation_output_plan(store.try_clear(storage_key)), DomStorageOperation::GetItems => { - let entries = store - .sorted_keys(storage_key) - .into_iter() - .filter_map(|key| { - let value = store.get_item(storage_key, &key)?; - Some(vec![key, value]) - }) - .collect::>(); + let entries = match store.try_items_utf16(storage_key) { + Ok(entries) => entries + .into_iter() + .map(|(key, value)| { + vec![ + String::from_utf16_lossy(&key), + String::from_utf16_lossy(&value), + ] + }) + .collect::>(), + Err(error) => return CommandOutputPlan::error(-32000, error.to_string()), + }; CommandOutputPlan::result(json!({ "entries": entries })) } DomStorageOperation::RemoveItem { key } => { diff --git a/moli-protocol/src/domains/storage/tests.rs b/moli-protocol/src/domains/storage/tests.rs index 8c231dde93..c965a2e923 100644 --- a/moli-protocol/src/domains/storage/tests.rs +++ b/moli-protocol/src/domains/storage/tests.rs @@ -2218,10 +2218,10 @@ async fn storage_clear_data_for_origin_clears_local_storage_area() { ); { ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert!(store.set_item(&first_party_key, "local", "1")); assert!(store.set_item(&partitioned_origin_top_a, "local", "1a")); @@ -2243,10 +2243,10 @@ async fn storage_clear_data_for_origin_clears_local_storage_area() { ctx.expect_result(12_525, json!({}), None); ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert_eq!(store.len(&first_party_key), 0); assert_eq!(store.get_item(&first_party_key, "local"), None); @@ -2285,10 +2285,10 @@ async fn storage_clear_data_for_storage_key_clears_site_data_for_matching_origin let sibling_storage_key = first_party_storage_key_for_origin(&sibling_origin); { ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert!(store.set_item(&storage_key, "local", "1")); assert!(store.set_item(&sibling_storage_key, "local", "2")); @@ -2348,7 +2348,7 @@ async fn storage_clear_data_for_storage_key_clears_site_data_for_matching_origin ctx.with_fixture(|conn| { let bc = conn.browser_context.as_ref().unwrap(); { - let mut store = bc.web_storage_store_for_test().lock(); + let store = bc.web_storage_store_for_test(); assert_eq!(store.get_item(&storage_key, "local"), None); assert_eq!( store.get_item(&sibling_storage_key, "local"), @@ -2400,10 +2400,10 @@ async fn storage_clear_data_for_partitioned_storage_key_clears_exact_partition_o ); { ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert!(store.set_item(&first_party_key, "local", "must-stay")); assert!(store.set_item(&partitioned_key, "local", "must-clear")); @@ -2458,7 +2458,7 @@ async fn storage_clear_data_for_partitioned_storage_key_clears_exact_partition_o ctx.with_fixture(|conn| { let bc = conn.browser_context.as_ref().unwrap(); { - let mut store = bc.web_storage_store_for_test().lock(); + let store = bc.web_storage_store_for_test(); assert_eq!( store.get_item(&first_party_key, "local"), Some("must-stay".to_owned()) @@ -2535,14 +2535,14 @@ async fn storage_get_usage_and_quota_reports_local_storage_usage_for_origin() { ); { ctx.with_fixture(|conn| { - let mut session_store = { + let session_store = { let bc = conn.browser_context.as_ref().unwrap(); - let mut store = bc.web_storage_store_for_test().lock(); + let store = bc.web_storage_store_for_test(); assert!(store.set_item(&first_party_key, "local", "abc")); assert!(store.set_item(&partitioned_origin, "partitioned", "de")); assert!(store.set_item(&partitioned_sibling, "sibling", "ignored")); - bc.session_storage_store_for_test().lock() + bc.session_storage_store_for_test() }; assert!(session_store.set_item(&first_party_key, "session", "session-only")); assert!(session_store.set_item(&partitioned_origin, "session", "also-ignored")); @@ -2840,7 +2840,7 @@ async fn storage_get_usage_and_quota_targets_command_session_browser_context() { let mut active = BrowserContext::new_with_page_for_test("BID-usage-active", "TID-usage-active"); active.attach_active_session("SID-usage-active"); { - let mut store = active.web_storage_store_for_test().lock(); + let store = active.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "aa")); } @@ -2848,7 +2848,7 @@ async fn storage_get_usage_and_quota_targets_command_session_browser_context() { BrowserContext::new_with_page_for_test("BID-usage-inactive", "TID-usage-inactive"); inactive.attach_active_session("SID-usage-inactive"); { - let mut store = inactive.web_storage_store_for_test().lock(); + let store = inactive.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "bbbb")); } @@ -3101,7 +3101,7 @@ async fn storage_clear_data_for_origin_targets_command_session_browser_context() ); active.attach_active_session("SID-session-clear-active"); { - let mut store = active.web_storage_store_for_test().lock(); + let store = active.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "active")); } @@ -3111,7 +3111,7 @@ async fn storage_clear_data_for_origin_targets_command_session_browser_context() ); inactive.attach_active_session("SID-session-clear-inactive"); { - let mut store = inactive.web_storage_store_for_test().lock(); + let store = inactive.web_storage_store_for_test(); assert!(store.set_item(&storage_key, "local", "inactive")); } @@ -3131,26 +3131,20 @@ async fn storage_clear_data_for_origin_targets_command_session_browser_context() ctx.expect_result(12_527, json!({}), Some("SID-session-clear-inactive")); ctx.with_fixture(|conn| { - let mut active_store = { + let active_store = { conn.browser_context .as_ref() .unwrap() .web_storage_store_for_test() - .lock() }; assert_eq!( active_store.get_item(&storage_key, "local"), Some("active".to_owned()) ); - drop(active_store); }); ctx.with_fixture(|conn| { - let mut inactive_store = { - conn.inactive_browser_contexts[0] - .web_storage_store_for_test() - .lock() - }; + let inactive_store = { conn.inactive_browser_contexts[0].web_storage_store_for_test() }; assert_eq!(inactive_store.get_item(&storage_key, "local"), None); }); } @@ -3172,10 +3166,10 @@ async fn storage_clear_data_for_origin_clears_cookies_and_local_storage_together let storage_key = first_party_storage_key_for_origin(&origin); { ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert!(store.set_item(&storage_key, "local", "1")); }); @@ -3207,10 +3201,10 @@ async fn storage_clear_data_for_origin_clears_cookies_and_local_storage_together { ctx.with_fixture(|conn| { - let mut store = { + let store = { let bc = conn.browser_context.as_ref().unwrap(); - bc.web_storage_store_for_test().lock() + bc.web_storage_store_for_test() }; assert_eq!(store.len(&storage_key), 0); assert_eq!(store.get_item(&storage_key, "local"), None); diff --git a/moli-renderer-v8/Cargo.toml b/moli-renderer-v8/Cargo.toml index 8abc158efc..183180e2ae 100644 --- a/moli-renderer-v8/Cargo.toml +++ b/moli-renderer-v8/Cargo.toml @@ -139,6 +139,7 @@ v8 = { version = "152.2.0", features = ["v8_enable_pointer_compression"] } v8 = "152.2.0" [dev-dependencies] +tempfile = "3.23.0" lopdf = { version = "0.45.0", default-features = false } fastrand = "2.3.0" moli-fetch = { path = "../moli-fetch", features = ["test-support"] } diff --git a/moli-renderer-v8/src/context_bootstrap/web_storage.rs b/moli-renderer-v8/src/context_bootstrap/web_storage.rs index bf2f535312..c8fc6a2232 100644 --- a/moli-renderer-v8/src/context_bootstrap/web_storage.rs +++ b/moli-renderer-v8/src/context_bootstrap/web_storage.rs @@ -47,3 +47,5 @@ pub use store::{ new_shared_json_web_storage_store, new_shared_web_storage_store, web_storage_area_key_for_storage_key, web_storage_partitioned_area_key, }; + +mod leases; diff --git a/moli-renderer-v8/src/context_bootstrap/web_storage/callbacks.rs b/moli-renderer-v8/src/context_bootstrap/web_storage/callbacks.rs index 1efff9a293..833e8a8360 100644 --- a/moli-renderer-v8/src/context_bootstrap/web_storage/callbacks.rs +++ b/moli-renderer-v8/src/context_bootstrap/web_storage/callbacks.rs @@ -1,4 +1,4 @@ -use super::helpers::{storage_clear, storage_put_utf16, storage_remove_utf16, with_storage_store}; +use super::helpers::{storage_clear, storage_put_utf16, storage_remove_utf16, with_storage_result}; use crate::util::{v8_string_from_utf16_units, v8_value_to_dom_string_u16}; use crate::webidl; @@ -39,8 +39,8 @@ pub(super) fn storage_get_item_callback<'s>( rv.set(v8::null(scope).into()); return; }; - let value = with_storage_store(scope, args.this(), |store, origin| { - store.get_item_utf16(origin, &key) + let value = with_storage_result(scope, args.this(), |store, origin| { + store.try_get_item_utf16(origin, &key) }) .flatten(); match value { @@ -94,9 +94,9 @@ pub(super) fn storage_remove_item_callback<'s>( storage_remove_utf16(scope, args.this(), &key); } -pub(super) fn storage_clear_callback( - scope: &mut v8::PinScope<'_, '_>, - args: v8::FunctionCallbackArguments<'_>, +pub(super) fn storage_clear_callback<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, _rv: v8::ReturnValue<'_, v8::Value>, ) { storage_clear(scope, args.this()); @@ -111,8 +111,8 @@ pub(super) fn storage_key_callback<'s>( rv.set(v8::null(scope).into()); return; }; - let value = with_storage_store(scope, args.this(), |store, origin| { - store.key_utf16(origin, parsed.index as usize) + let value = with_storage_result(scope, args.this(), |store, origin| { + store.try_key_utf16(origin, parsed.index as usize) }) .flatten(); match value { @@ -125,12 +125,12 @@ pub(super) fn storage_key_callback<'s>( } } -pub(super) fn storage_length_getter_callback( - scope: &mut v8::PinScope<'_, '_>, - args: v8::FunctionCallbackArguments<'_>, +pub(super) fn storage_length_getter_callback<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::FunctionCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) { let len = - with_storage_store(scope, args.this(), |store, origin| store.len(origin)).unwrap_or(0); + with_storage_result(scope, args.this(), |store, origin| store.try_len(origin)).unwrap_or(0); rv.set(v8::Integer::new_from_unsigned(scope, len as u32).into()); } diff --git a/moli-renderer-v8/src/context_bootstrap/web_storage/helpers.rs b/moli-renderer-v8/src/context_bootstrap/web_storage/helpers.rs index 147a8bbb7f..26a5650af6 100644 --- a/moli-renderer-v8/src/context_bootstrap/web_storage/helpers.rs +++ b/moli-renderer-v8/src/context_bootstrap/web_storage/helpers.rs @@ -163,10 +163,10 @@ pub(super) fn this_storage_kind( .is_some_and(|s| s.to_rust_string_lossy(scope) == "session") } -pub(super) fn with_storage_store( - scope: &mut v8::PinScope<'_, '_>, - storage: v8::Local<'_, v8::Object>, - f: impl FnOnce(&mut WebStorageStore, &str) -> R, +pub(super) fn with_storage_store<'s, R>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, + f: impl FnOnce(&WebStorageStore, &str) -> R, ) -> Option { let is_session = this_storage_kind(scope, storage); let owner = storage_owner(scope, storage); @@ -184,13 +184,64 @@ pub(super) fn with_storage_store( } else { host.web_storage_store() }; - let mut guard = store.lock(); - Some(f(&mut guard, &area_key)) + super::leases::ensure_storage_area_lease(scope, storage, &store, &area_key)?; + Some(f(&store, &area_key)) } -pub(super) fn storage_put_utf16( +pub(super) fn with_storage_result<'s, R>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, + f: impl FnOnce(&WebStorageStore, &str) -> Result, +) -> Option { + match with_storage_store(scope, storage, f)? { + Ok(value) => Some(value), + Err(WebStorageMutationError::QuotaExceeded) => { + throw_storage_quota_exceeded(scope); + None + } + Err(WebStorageMutationError::Persistence(message)) => { + throw_storage_persistence_error(scope, &message); + None + } + } +} + +fn queue_storage_change( scope: &mut v8::PinScope<'_, '_>, - storage: v8::Local<'_, v8::Object>, + owner: WebStorageOwner, + is_session: bool, + change: super::store::WebStorageMutation, +) { + use super::store::WebStorageMutation; + let (key, old, new) = match change { + WebStorageMutation::ItemAdded { key, value, .. } => ( + Some(key.as_units().to_vec()), + None, + Some(value.as_units().to_vec()), + ), + WebStorageMutation::ItemUpdated { + key, + old_value, + new_value, + .. + } => ( + Some(key.as_units().to_vec()), + Some(old_value.as_units().to_vec()), + Some(new_value.as_units().to_vec()), + ), + WebStorageMutation::ItemRemoved { key, old_value, .. } => ( + Some(key.as_units().to_vec()), + Some(old_value.as_units().to_vec()), + None, + ), + WebStorageMutation::ItemsCleared { .. } => (None, None, None), + }; + queue_storage_event(scope, owner, is_session, key, old, new); +} + +pub(super) fn storage_put_utf16<'s>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, key: &[u16], value: &[u16], ) -> bool { @@ -199,44 +250,19 @@ pub(super) fn storage_put_utf16( } let is_session = this_storage_kind(scope, storage); let owner = storage_owner(scope, storage); - let previous = with_storage_store(scope, storage, |store, origin| { - store.get_item_utf16(origin, key) - }) - .flatten(); - if previous.as_deref() == Some(value) { - return true; - } - let result = with_storage_store(scope, storage, |store, origin| { - store.try_set_item_utf16(origin, key, value) - }); - let Some(result) = result else { + let Some(change) = with_storage_result(scope, storage, |store, origin| { + store.try_set_change_utf16(origin, key, value) + }) else { return false; }; - let inserted = match result { - Ok(inserted) => inserted, - Err(WebStorageMutationError::QuotaExceeded) => { - throw_storage_quota_exceeded(scope); - return false; - } - Err(WebStorageMutationError::Persistence(message)) => { - throw_storage_persistence_error(scope, &message); - return false; - } - }; - queue_storage_event( - scope, - owner, - is_session, - Some(key.to_vec()), - previous, - Some(value.to_vec()), - ); - inserted + if let Some(change) = change { + queue_storage_change(scope, owner, is_session, change); + } + true } - -pub(super) fn storage_remove_utf16( - scope: &mut v8::PinScope<'_, '_>, - storage: v8::Local<'_, v8::Object>, +pub(super) fn storage_remove_utf16<'s>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, key: &[u16], ) -> bool { if storage_internal_name_utf16(key) { @@ -244,52 +270,24 @@ pub(super) fn storage_remove_utf16( } let is_session = this_storage_kind(scope, storage); let owner = storage_owner(scope, storage); - let previous = with_storage_store(scope, storage, |store, origin| { - store.get_item_utf16(origin, key) - }) - .flatten(); - if previous.is_none() { - return false; - } - let result = with_storage_store(scope, storage, |store, origin| { - store.try_remove_item_utf16(origin, key) - }); - let Some(result) = result else { + let Some(Some(change)) = with_storage_result(scope, storage, |store, origin| { + store.try_remove_change_utf16(origin, key) + }) else { return false; }; - let removed = match result { - Ok(removed) => removed, - Err(WebStorageMutationError::QuotaExceeded) => false, - Err(WebStorageMutationError::Persistence(message)) => { - throw_storage_persistence_error(scope, &message); - return false; - } - }; - if removed { - queue_storage_event(scope, owner, is_session, Some(key.to_vec()), previous, None); - } - removed + queue_storage_change(scope, owner, is_session, change); + true } - -pub(super) fn storage_clear(scope: &mut v8::PinScope<'_, '_>, storage: v8::Local<'_, v8::Object>) { +pub(super) fn storage_clear<'s>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, +) { let is_session = this_storage_kind(scope, storage); let owner = storage_owner(scope, storage); - let had_items = - with_storage_store(scope, storage, |store, origin| store.len(origin)).unwrap_or(0) > 0; - let result = with_storage_store(scope, storage, |store, origin| store.try_clear(origin)); - let Some(result) = result else { - return; - }; - let cleared = match result { - Ok(cleared) => cleared, - Err(WebStorageMutationError::QuotaExceeded) => false, - Err(WebStorageMutationError::Persistence(message)) => { - throw_storage_persistence_error(scope, &message); - return; - } - }; - if had_items && cleared { - queue_storage_event(scope, owner, is_session, None, None, None); + if let Some(Some(change)) = with_storage_result(scope, storage, |store, origin| { + store.try_clear_change(origin) + }) { + queue_storage_change(scope, owner, is_session, change); } } @@ -471,7 +469,7 @@ fn throw_storage_quota_exceeded(scope: &mut v8::PinScope<'_, '_>) { scope.throw_exception(exception); } -fn throw_storage_persistence_error(scope: &mut v8::PinScope<'_, '_>, message: &str) { +pub(super) fn throw_storage_persistence_error(scope: &mut v8::PinScope<'_, '_>, message: &str) { let message = format!("Failed to persist localStorage data: {message}"); crate::context_bootstrap::throw_dom_exception_value(scope, &message, "UnknownError"); } diff --git a/moli-renderer-v8/src/context_bootstrap/web_storage/interceptors.rs b/moli-renderer-v8/src/context_bootstrap/web_storage/interceptors.rs index 720538f007..07b2c9166f 100644 --- a/moli-renderer-v8/src/context_bootstrap/web_storage/interceptors.rs +++ b/moli-renderer-v8/src/context_bootstrap/web_storage/interceptors.rs @@ -2,7 +2,7 @@ use super::helpers::{ remove_storage_prototype_index_descriptor, set_storage_prototype_index_descriptor, storage_internal_name_utf16, storage_key_is_shadowed_by_prototype, storage_prototype_index_descriptor, storage_put_utf16, storage_remove_utf16, - with_storage_store, + with_storage_result, }; use super::*; use crate::util::{throw_type_error, v8_string_from_utf16_units, v8_string_to_u16_string}; @@ -26,8 +26,8 @@ struct StorageAccessorPropertyDescriptorDeclaration<'scope> { configurable: bool, } -fn storage_value_to_dom_string_utf16( - scope: &mut v8::PinScope<'_, '_>, +fn storage_value_to_dom_string_utf16<'s>( + scope: &mut v8::PinScope<'s, '_>, value: v8::Local<'_, v8::Value>, ) -> Option> { if value.is_symbol() { @@ -39,10 +39,10 @@ fn storage_value_to_dom_string_utf16( .map(|value| v8_string_to_u16_string(scope, value).into_vec()) } -pub(super) fn storage_named_getter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_getter<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -54,8 +54,8 @@ pub(super) fn storage_named_getter( { return v8::Intercepted::kNo; } - let value = with_storage_store(scope, args.holder(), |store, origin| { - store.get_item_utf16(origin, &key) + let value = with_storage_result(scope, args.holder(), |store, origin| { + store.try_get_item_utf16(origin, &key) }) .flatten(); let Some(value) = value else { @@ -68,11 +68,11 @@ pub(super) fn storage_named_getter( v8::Intercepted::kYes } -pub(super) fn storage_named_setter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_setter<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, value: v8::Local<'_, v8::Value>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, _rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -89,10 +89,10 @@ pub(super) fn storage_named_setter( v8::Intercepted::kYes } -pub(super) fn storage_named_query( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_query<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Integer>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -104,8 +104,8 @@ pub(super) fn storage_named_query( { return v8::Intercepted::kNo; } - let contains = with_storage_store(scope, args.holder(), |store, origin| { - store.contains_key_utf16(origin, &key) + let contains = with_storage_result(scope, args.holder(), |store, origin| { + store.try_contains_key_utf16(origin, &key) }) .unwrap_or(false); if !contains { @@ -115,10 +115,10 @@ pub(super) fn storage_named_query( v8::Intercepted::kYes } -pub(super) fn storage_named_deleter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_deleter<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -133,13 +133,13 @@ pub(super) fn storage_named_deleter( v8::Intercepted::kYes } -pub(super) fn storage_named_enumerator( - scope: &mut v8::PinScope<'_, '_>, - args: v8::PropertyCallbackArguments<'_>, +pub(super) fn storage_named_enumerator<'s>( + scope: &mut v8::PinScope<'s, '_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Array>, ) { - let keys = with_storage_store(scope, args.holder(), |store, origin| { - store.sorted_keys_utf16(origin) + let keys = with_storage_result(scope, args.holder(), |store, origin| { + store.try_sorted_keys_utf16(origin) }) .unwrap_or_default(); let array = v8::Array::new(scope, keys.len() as i32); @@ -151,10 +151,10 @@ pub(super) fn storage_named_enumerator( rv.set(array); } -pub(super) fn storage_named_descriptor( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_descriptor<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -166,8 +166,8 @@ pub(super) fn storage_named_descriptor( { return v8::Intercepted::kNo; } - let value = with_storage_store(scope, args.holder(), |store, origin| { - store.get_item_utf16(origin, &key) + let value = with_storage_result(scope, args.holder(), |store, origin| { + store.try_get_item_utf16(origin, &key) }) .flatten(); let Some(value) = value else { @@ -183,11 +183,11 @@ pub(super) fn storage_named_descriptor( v8::Intercepted::kYes } -pub(super) fn storage_named_definer( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_named_definer<'s>( + scope: &mut v8::PinScope<'s, '_>, key: v8::Local<'_, v8::Name>, desc: &v8::PropertyDescriptor, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, _rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Ok(key_string) = v8::Local::::try_from(key) else { @@ -204,10 +204,10 @@ pub(super) fn storage_named_definer( v8::Intercepted::kYes } -pub(super) fn storage_indexed_getter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_getter<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { if storage_prototype_index_getter_value(scope, index, args.holder(), &mut rv) { @@ -219,11 +219,11 @@ pub(super) fn storage_indexed_getter( storage_named_getter(scope, key.into(), args, rv) } -pub(super) fn storage_indexed_setter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_setter<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, value: v8::Local<'_, v8::Value>, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Some(key) = v8_string(scope, &index.to_string()) else { @@ -232,10 +232,10 @@ pub(super) fn storage_indexed_setter( storage_named_setter(scope, key.into(), value, args, rv) } -pub(super) fn storage_indexed_query( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_query<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, rv: v8::ReturnValue<'_, v8::Integer>, ) -> v8::Intercepted { if storage_prototype_index_descriptor(scope, index).is_some() { @@ -249,10 +249,10 @@ pub(super) fn storage_indexed_query( storage_named_query(scope, key.into(), args, rv) } -pub(super) fn storage_indexed_descriptor( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_descriptor<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { if storage_prototype_index_descriptor(scope, index).is_some() { @@ -264,10 +264,10 @@ pub(super) fn storage_indexed_descriptor( storage_named_descriptor(scope, key.into(), args, rv) } -pub(super) fn storage_indexed_deleter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_deleter<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Some(key) = v8_string(scope, &index.to_string()) else { @@ -276,11 +276,11 @@ pub(super) fn storage_indexed_deleter( storage_named_deleter(scope, key.into(), args, rv) } -pub(super) fn storage_indexed_definer( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_indexed_definer<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, desc: &v8::PropertyDescriptor, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Some(key) = v8_string(scope, &index.to_string()) else { @@ -289,10 +289,10 @@ pub(super) fn storage_indexed_definer( storage_named_definer(scope, key.into(), desc, args, rv) } -pub(super) fn storage_prototype_indexed_getter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_prototype_indexed_getter<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - args: v8::PropertyCallbackArguments<'_>, + args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { if storage_prototype_index_getter_value(scope, index, args.holder(), &mut rv) { @@ -305,7 +305,7 @@ pub(super) fn storage_prototype_indexed_setter<'s>( scope: &mut v8::PinScope<'s, '_>, index: u32, value: v8::Local<'s, v8::Value>, - _args: v8::PropertyCallbackArguments<'_>, + _args: v8::PropertyCallbackArguments<'s>, _rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Some(descriptor) = storage_value_property_descriptor(scope, value) else { @@ -317,10 +317,10 @@ pub(super) fn storage_prototype_indexed_setter<'s>( v8::Intercepted::kYes } -pub(super) fn storage_prototype_indexed_query( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_prototype_indexed_query<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - _args: v8::PropertyCallbackArguments<'_>, + _args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Integer>, ) -> v8::Intercepted { let Some(descriptor) = storage_prototype_index_descriptor(scope, index) else { @@ -330,21 +330,21 @@ pub(super) fn storage_prototype_indexed_query( v8::Intercepted::kYes } -pub(super) fn storage_prototype_indexed_deleter( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_prototype_indexed_deleter<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - _args: v8::PropertyCallbackArguments<'_>, + _args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { rv.set_bool(remove_storage_prototype_index_descriptor(scope, index)); v8::Intercepted::kYes } -pub(super) fn storage_prototype_indexed_definer( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_prototype_indexed_definer<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, desc: &v8::PropertyDescriptor, - _args: v8::PropertyCallbackArguments<'_>, + _args: v8::PropertyCallbackArguments<'s>, _rv: v8::ReturnValue<'_, v8::Boolean>, ) -> v8::Intercepted { let Some(descriptor) = storage_descriptor_object_from_property_descriptor(scope, desc) else { @@ -356,10 +356,10 @@ pub(super) fn storage_prototype_indexed_definer( v8::Intercepted::kYes } -pub(super) fn storage_prototype_indexed_descriptor( - scope: &mut v8::PinScope<'_, '_>, +pub(super) fn storage_prototype_indexed_descriptor<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, - _args: v8::PropertyCallbackArguments<'_>, + _args: v8::PropertyCallbackArguments<'s>, mut rv: v8::ReturnValue<'_, v8::Value>, ) -> v8::Intercepted { let Some(descriptor) = storage_prototype_index_descriptor(scope, index) else { @@ -412,8 +412,8 @@ fn storage_descriptor_local_value<'s>( v8::Local::new(scope, &value) } -fn storage_prototype_index_getter_value( - scope: &mut v8::PinScope<'_, '_>, +fn storage_prototype_index_getter_value<'s>( + scope: &mut v8::PinScope<'s, '_>, index: u32, receiver: v8::Local<'_, v8::Object>, rv: &mut v8::ReturnValue<'_, v8::Value>, @@ -445,8 +445,8 @@ fn storage_prototype_index_getter_value( true } -fn property_attributes_for_descriptor( - scope: &mut v8::PinScope<'_, '_>, +fn property_attributes_for_descriptor<'s>( + scope: &mut v8::PinScope<'s, '_>, descriptor: v8::Local<'_, v8::Object>, ) -> v8::PropertyAttribute { let mut attributes = v8::PropertyAttribute::NONE; @@ -469,8 +469,8 @@ fn property_attributes_for_descriptor( attributes } -fn descriptor_bool( - scope: &mut v8::PinScope<'_, '_>, +fn descriptor_bool<'s>( + scope: &mut v8::PinScope<'s, '_>, descriptor: v8::Local<'_, v8::Object>, key: &'static str, ) -> bool { @@ -479,8 +479,8 @@ fn descriptor_bool( .is_some_and(|value| value.boolean_value(scope)) } -fn descriptor_has_own( - scope: &mut v8::PinScope<'_, '_>, +fn descriptor_has_own<'s>( + scope: &mut v8::PinScope<'s, '_>, descriptor: v8::Local<'_, v8::Object>, key: &'static str, ) -> bool { diff --git a/moli-renderer-v8/src/context_bootstrap/web_storage/leases.rs b/moli-renderer-v8/src/context_bootstrap/web_storage/leases.rs new file mode 100644 index 0000000000..09a85784fd --- /dev/null +++ b/moli-renderer-v8/src/context_bootstrap/web_storage/leases.rs @@ -0,0 +1,87 @@ +//! A live Storage wrapper pins only its own area. Native dirty/in-flight +//! batches keep the same area alive independently of the wrapper's lifetime. +use crate::util::{get_private_value, set_private_value}; +use moli_web_storage::{SharedWebStorageStore, WebStorageAreaLease}; +use std::{cell::RefCell, collections::HashMap, rc::Rc}; + +const LEASE_SLOT: &str = "__moliWebStorageAreaLease"; +#[derive(Default)] +struct Leases { + next: u64, + entries: HashMap, +} +struct Entry { + store: SharedWebStorageStore, + key: String, + _lease: WebStorageAreaLease, +} +type Registry = Rc>; + +pub(super) fn ensure_storage_area_lease<'s>( + scope: &mut v8::PinScope<'s, '_>, + storage: v8::Local<'s, v8::Object>, + store: &SharedWebStorageStore, + key: &str, +) -> Option<()> { + let registry = scope.get_slot::().cloned().unwrap_or_else(|| { + let registry = Registry::default(); + scope.set_slot(registry.clone()); + registry + }); + let id = get_private_value(scope, storage, LEASE_SLOT) + .and_then(|value| v8::Local::::try_from(value).ok()) + .map(|id| id.u64_value().0); + if let Some(id) = id + && registry + .borrow() + .entries + .get(&id) + .is_some_and(|entry| std::sync::Arc::ptr_eq(&entry.store, store) && entry.key == key) + { + return Some(()); + } + let lease = match store.lease_area(key) { + Ok(Some(lease)) => lease, + Ok(None) => return Some(()), + Err(error) => { + super::helpers::throw_storage_persistence_error(scope, &error.to_string()); + return None; + } + }; + if let Some(id) = id { + let old = registry.borrow_mut().entries.insert( + id, + Entry { + store: store.clone(), + key: key.to_owned(), + _lease: lease, + }, + ); + drop(old); + return Some(()); + } + let id = { + let mut registry = registry.borrow_mut(); + registry.next = registry.next.checked_add(1)?; + let id = registry.next; + registry.entries.insert( + id, + Entry { + store: store.clone(), + key: key.to_owned(), + _lease: lease, + }, + ); + id + }; + let marker = v8::BigInt::new_from_u64(scope, id); + set_private_value(scope, storage, LEASE_SLOT, marker.into()); + let weak = Rc::downgrade(®istry); + crate::v8_finalizer::track_context_owned_v8_finalizer(scope, storage, move || { + if let Some(registry) = weak.upgrade() { + let retired = registry.borrow_mut().entries.remove(&id); + drop(retired); + } + }); + Some(()) +} diff --git a/moli-renderer-v8/src/context_bootstrap/window_runtime/navigator.rs b/moli-renderer-v8/src/context_bootstrap/window_runtime/navigator.rs index 64c69cfe00..27c374fdf5 100644 --- a/moli-renderer-v8/src/context_bootstrap/window_runtime/navigator.rs +++ b/moli-renderer-v8/src/context_bootstrap/window_runtime/navigator.rs @@ -706,7 +706,18 @@ pub(in crate::context_bootstrap) fn navigator_storage_estimate_callback<'s>( else { return; }; - let usage = storage_manager_usage_bytes(scope, &context); + let usage = match storage_manager_usage_bytes(scope, &context) { + Ok(usage) => usage, + Err(error) => { + let error = crate::context_bootstrap::new_dom_exception_value( + scope, + &error.to_string(), + "UnknownError", + ); + let _ = resolver.reject(scope, error); + return; + } + }; let estimate = build_storage_estimate_object_with_quota( scope, usage.total, @@ -864,27 +875,23 @@ fn storage_manager_popup_id<'s>( fn storage_manager_usage_bytes( scope: &mut v8::PinScope<'_, '_>, context: &StorageManagerOwnerContext, -) -> StorageUsageSnapshot { - let web_storage_usage = context - .host_ptr - .zip(context.area_key.as_deref()) - .map(|(host_ptr, area_key)| { +) -> Result { + let web_storage_usage = + if let Some((host_ptr, area_key)) = context.host_ptr.zip(context.area_key.as_deref()) { let host = unsafe { &*host_ptr }; - let local_usage = { - let store = host.web_storage_store(); - store.lock().usage_bytes(area_key) - }; - let session_usage = if let Some(popup_id) = context.popup_id { + let local = host.web_storage_store().try_usage_bytes(area_key)?; + let session = if let Some(popup_id) = context.popup_id { host.lightweight_popup_session_storage_store(popup_id) - .map(|store| store.lock().usage_bytes(area_key)) + .map(|store| store.try_usage_bytes(area_key)) + .transpose()? .unwrap_or(0) } else { - let store = host.session_storage_store(); - store.lock().usage_bytes(area_key) + host.session_storage_store().try_usage_bytes(area_key)? }; - local_usage.saturating_add(session_usage) as u64 - }) - .unwrap_or(0); + local.saturating_add(session) as u64 + } else { + 0 + }; let locator = moli_storage_service::StorageBucketLocator::default_bucket(&context.storage_key); let bucket_usage = storage_bucket_quota_owner_for_locator(scope, &locator) .and_then(|owner| owner.usage_snapshot().ok()); @@ -905,7 +912,7 @@ fn storage_manager_usage_bytes( 0, ) }); - StorageUsageSnapshot { + Ok(StorageUsageSnapshot { total: web_storage_usage .saturating_add(indexed_db_usage) .saturating_add(cache_storage_usage) @@ -914,7 +921,7 @@ fn storage_manager_usage_bytes( indexed_db: indexed_db_usage, cache_storage: cache_storage_usage, opfs: opfs_usage, - } + }) } pub(in crate::context_bootstrap) fn storage_bucket_manager_open_callback<'s>( diff --git a/moli-renderer-v8/src/native_bridge/context_host/popups.rs b/moli-renderer-v8/src/native_bridge/context_host/popups.rs index 299888510f..4637eff619 100644 --- a/moli-renderer-v8/src/native_bridge/context_host/popups.rs +++ b/moli-renderer-v8/src/native_bridge/context_host/popups.rs @@ -1252,7 +1252,7 @@ impl JsContextHost { return target_store; } let entries = { - let mut source = source_store.lock(); + let source = &source_store; source .sorted_keys_utf16(source_scope.area_key()) .into_iter() @@ -1264,7 +1264,7 @@ impl JsContextHost { .collect::>() }; { - let mut target = target_store.lock(); + let target = &target_store; for (key, value) in entries { let _ = target.set_item_utf16(target_scope.area_key(), &key, &value); } diff --git a/moli-renderer-v8/src/script_vm/tests/mod.rs b/moli-renderer-v8/src/script_vm/tests/mod.rs index 8e2f33f456..7e7ee0706c 100644 --- a/moli-renderer-v8/src/script_vm/tests/mod.rs +++ b/moli-renderer-v8/src/script_vm/tests/mod.rs @@ -2404,3 +2404,4 @@ mod xml_serializer_receivers; mod hyperlink_audit_owner; mod fetch_referrer; +mod sqlite_web_storage; diff --git a/moli-renderer-v8/src/script_vm/tests/sqlite_web_storage.rs b/moli-renderer-v8/src/script_vm/tests/sqlite_web_storage.rs new file mode 100644 index 0000000000..5adac10353 --- /dev/null +++ b/moli-renderer-v8/src/script_vm/tests/sqlite_web_storage.rs @@ -0,0 +1,65 @@ +use super::*; + +#[test] +fn sqlite_storage_wrappers_share_utf16_data_without_sharing_session_storage() { + let directory = tempfile::tempdir().unwrap(); + let runtime = moli_sqlite::Runtime::new().unwrap(); + let (local, owner) = moli_web_storage::new_sqlite_web_storage_store( + directory.path().join("localstorage.sqlite"), + moli_sqlite::OpenMode::Create, + runtime.handle(), + ) + .unwrap(); + let first_handles = + crate::RendererWebStorageHandles::new(local.clone(), crate::new_shared_web_storage_store()); + let second_handles = + crate::RendererWebStorageHandles::new(local, crate::new_shared_web_storage_store()); + let mut first = new_storage_test_vm("https://sqlite-storage.test/a"); + first.set_web_storage_handles(&first_handles); + let mut second = new_storage_test_vm("https://sqlite-storage.test/b"); + second.set_web_storage_handles(&second_handles); + first.eval("localStorage.setItem('key', String.fromCharCode(0xd800, 0, 0xdc00)); sessionStorage.setItem('private','first'); localStorage['property']='shared';").unwrap(); + assert_eq!(second.eval("JSON.stringify([Array.from({length:3},(_,i)=>localStorage.getItem('key').charCodeAt(i)),localStorage.property,localStorage.length,sessionStorage.getItem('private')])").unwrap(),"[[55296,0,56320],\"shared\",2,null]"); + second.eval("delete localStorage.property").unwrap(); + assert_eq!( + first + .eval("String(localStorage.getItem('property'))") + .unwrap(), + "null" + ); + drop(first); + drop(second); + owner.close().unwrap(); +} +#[test] +fn retired_sqlite_storage_reports_unknown_error_through_all_read_surfaces() { + let directory = tempfile::tempdir().unwrap(); + let runtime = moli_sqlite::Runtime::new().unwrap(); + let (local, owner) = moli_web_storage::new_sqlite_web_storage_store( + directory.path().join("localstorage.sqlite"), + moli_sqlite::OpenMode::Create, + runtime.handle(), + ) + .unwrap(); + let handles = + crate::RendererWebStorageHandles::new(local, crate::new_shared_web_storage_store()); + let mut vm = new_storage_test_vm("https://closed-sqlite.test/"); + vm.set_web_storage_handles(&handles); + vm.eval("localStorage.setItem('entry','value');globalThis.storage=localStorage;") + .unwrap(); + owner.close().unwrap(); + let result = vm + .eval( + r#"JSON.stringify([ + ()=>storage.getItem('entry'),()=>storage.key(0),()=>storage.length, + ()=>storage.entry,()=>('entry' in storage),()=>Object.keys(storage), + ()=>storage.setItem('entry','changed'),()=>storage.removeItem('entry'),()=>storage.clear() + ].map(read=>{try {read();return 'missing error';} catch(error) {return error.name;}}))"#, + ) + .unwrap(); + assert_eq!( + result, + "[\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\",\"UnknownError\"]" + ); + assert_eq!(vm.eval("typeof storage.key").unwrap(), "function"); +} diff --git a/moli-sqlite/Cargo.toml b/moli-sqlite/Cargo.toml index f536ad0f46..fff4784600 100644 --- a/moli-sqlite/Cargo.toml +++ b/moli-sqlite/Cargo.toml @@ -8,10 +8,11 @@ edition = "2024" parking_lot = "0.12.5" rusqlite = { version = "0.40.2", features = ["bundled"] } -tokio = { version = "1.53.1", features = ["rt-multi-thread", "sync", "time"] } +tokio = { version = "1.53.1", features = ["rt-multi-thread", "sync", "time", "macros"] } [dev-dependencies] -tokio = { version = "1.53.1", features = ["macros"] } +tempfile = "3.23.0" +tokio = { version = "1.53.1", features = ["macros", "test-util"] } [lints] workspace = true diff --git a/moli-sqlite/src/lib.rs b/moli-sqlite/src/lib.rs index f43e55430c..4c3df1757b 100644 --- a/moli-sqlite/src/lib.rs +++ b/moli-sqlite/src/lib.rs @@ -13,3 +13,5 @@ pub use connection::{ }; pub use runtime::{Runtime, RuntimeError, RuntimeHandle}; + +pub mod writeback; diff --git a/moli-sqlite/src/writeback.rs b/moli-sqlite/src/writeback.rs new file mode 100644 index 0000000000..e84f1ec1ad --- /dev/null +++ b/moli-sqlite/src/writeback.rs @@ -0,0 +1,438 @@ +//! A file owns one FIFO; the partition owns the executor. Logical mutation +//! acceptance never waits for SQLite. Pending data is moved, not copied, into +//! an immutable commit, so its acknowledgement cannot clear newer changes. +use crate::{ConnectionOptions, OpenMode, RuntimeHandle}; +use parking_lot::Mutex; +use rusqlite::Connection; +use std::{ + fmt, + path::PathBuf, + sync::{Arc, mpsc}, + time::Duration, +}; +use tokio::sync::{Notify, mpsc as async_mpsc}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WritebackError(Arc); +impl WritebackError { + pub fn new(message: impl fmt::Display) -> Self { + Self(message.to_string().into()) + } +} +impl fmt::Display for WritebackError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} +impl std::error::Error for WritebackError {} +impl From for WritebackError { + fn from(error: rusqlite::Error) -> Self { + Self::new(error) + } +} +type Result = std::result::Result; + +pub trait WriteBatch: Default + Send + 'static { + fn is_empty(&self) -> bool; + /// Compose chronological deltas: self happened before newer. + fn merge(&mut self, newer: Self); + /// An error can leave commit publication uncertain. Do not elide a later + /// delete merely because an older put was originally a new row. + fn mark_commit_uncertain(&mut self) {} + /// Must be atomic and idempotent; a failed batch can be explicitly retried. + fn write(&self, connection: &mut Connection) -> Result<()>; +} +#[derive(Clone, Copy)] +pub struct WritebackPolicy { + pub delay: Duration, + pub early_commit_operations: Option, +} +#[derive(Clone, Copy)] +pub enum FailureAdmission { + Retain, + Reject, +} +struct Pending { + batch: B, + accepted: u64, + committed: u64, + operations: usize, + first_dirty: Option, + failure: Option, + closed: bool, +} +struct Shared { + pending: Mutex>, + changed: Notify, +} +trait Query: Send { + fn run(self: Box, connection: &mut Connection); +} +struct QueryJob { + query: F, + reply: mpsc::Sender>, +} +impl Query for QueryJob +where + F: FnOnce(&mut Connection) -> Result + Send, + T: Send, +{ + fn run(self: Box, connection: &mut Connection) { + let _ = self.reply.send((self.query)(connection)); + } +} +enum Message { + Query(Box), + Flush { + through: u64, + reply: mpsc::Sender>, + }, + Close, +} +pub struct WritebackClient { + shared: Arc>, + sender: async_mpsc::UnboundedSender, + runtime: RuntimeHandle, +} +impl Clone for WritebackClient { + fn clone(&self) -> Self { + Self { + shared: self.shared.clone(), + sender: self.sender.clone(), + runtime: self.runtime.clone(), + } + } +} +#[derive(Default)] +struct CloseReceipt { + receiver: Option>>, + result: Option>, +} +pub struct WritebackOwner { + client: WritebackClient, + finished: Mutex, +} +impl WritebackOwner { + /// Native initialization, before opening V8. Every connection operation, + /// including schema initialization and final close, runs in blocking work. + pub fn open( + path: PathBuf, + mode: OpenMode, + runtime: RuntimeHandle, + policy: WritebackPolicy, + initialize: impl FnOnce(&mut Connection) -> Result<()> + Send + 'static, + ) -> Result { + if runtime.is_current() { + return Err(WritebackError::new( + "cannot open a database from its SQL executor", + )); + } + let (sender, receiver) = async_mpsc::unbounded_channel(); + let shared = Arc::new(Shared { + pending: Mutex::new(Pending { + batch: B::default(), + accepted: 0, + committed: 0, + operations: 0, + first_dirty: None, + failure: None, + closed: false, + }), + changed: Notify::new(), + }); + let client = WritebackClient { + shared: shared.clone(), + sender, + runtime: runtime.clone(), + }; + let (ready_tx, ready_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + let worker = runtime.clone(); + runtime + .spawn(async move { + let opened = worker + .blocking(move || { + let mut connection = + crate::open_disk(&path, mode, ConnectionOptions::default()) + .map_err(WritebackError::new)?; + initialize(&mut connection)?; + Ok(connection) + }) + .await + .map_err(WritebackError::new) + .and_then(|r| r); + let result = match opened { + Ok(connection) => { + let _ = ready_tx.send(Ok(())); + run(connection, receiver, shared.clone(), worker, policy).await + } + Err(error) => { + let _ = ready_tx.send(Err(error.clone())); + Err(error) + } + }; + let mut pending = shared.pending.lock(); + pending.closed = true; + if let Err(error) = &result { + pending.failure = Some(error.clone()); + } + drop(pending); + let _ = done_tx.send(result); + }) + .map_err(WritebackError::new)?; + ready_rx + .recv() + .map_err(|_| WritebackError::new("database initialization stopped"))??; + Ok(Self { + client, + finished: Mutex::new(CloseReceipt { + receiver: Some(done_rx), + result: None, + }), + }) + } + pub fn client(&self) -> WritebackClient { + self.client.clone() + } + pub fn close(&self) -> Result<()> { + if self.client.runtime.is_current() { + return Err(WritebackError::new( + "cannot close a database from its SQL executor", + )); + } + let mut finished = self.finished.lock(); + if let Some(result) = &finished.result { + return result.clone(); + } + let receipt = finished.receiver.take().expect("unobserved close receipt"); + self.client.shared.pending.lock().closed = true; + let _ = self.client.sender.send(Message::Close); + let result = receipt + .recv() + .map_err(|_| WritebackError::new("database stopped before its close receipt")) + .and_then(|r| r); + finished.result = Some(result.clone()); + result + } +} +impl Drop for WritebackOwner { + fn drop(&mut self) { + // The native partition closes explicitly and observes the receipt. + // Drop still seals admission if startup unwinds before that boundary. + self.client.shared.pending.lock().closed = true; + let _ = self.client.sender.send(Message::Close); + if !self.client.runtime.is_current() + && let Some(receipt) = self.finished.get_mut().receiver.take() + { + let _ = receipt.recv(); + } + } +} +impl WritebackClient { + /// Publish the native cache and accepted delta under the same admission + /// gate. The closure never enters V8 or performs SQL. + pub fn accept>( + &self, + admission: FailureAdmission, + mutation: impl FnOnce() -> std::result::Result<(T, B), E>, + ) -> std::result::Result { + let mut pending = self.shared.pending.lock(); + if pending.closed { + return Err(WritebackError::new("database is closed").into()); + } + if let FailureAdmission::Reject = admission + && let Some(error) = &pending.failure + { + return Err(error.clone().into()); + } + let next = pending + .accepted + .checked_add(1) + .ok_or_else(|| E::from(WritebackError::new("database sequence exhausted")))?; + let (result, batch) = mutation()?; + if !batch.is_empty() { + pending.batch.merge(batch); + pending.accepted = next; + pending.operations = pending.operations.saturating_add(1); + pending + .first_dirty + .get_or_insert_with(tokio::time::Instant::now); + self.shared.changed.notify_one(); + } + Ok(result) + } + pub fn record(&self, batch: B, admission: FailureAdmission) -> Result<()> { + self.accept(admission, || Ok(((), batch))) + } + /// Only the cold-load native bridge uses this on a V8 owner. Completion + /// comes directly from the SQL worker; it cannot depend on a JS task. + pub fn query( + &self, + query: impl FnOnce(&mut Connection) -> Result + Send + 'static, + ) -> Result { + if self.runtime.is_current() { + return Err(WritebackError::new( + "cannot synchronously query the SQL executor from itself", + )); + } + let (tx, rx) = mpsc::channel(); + { + let pending = self.shared.pending.lock(); + if pending.closed { + return Err(WritebackError::new("database is closed")); + } + self.sender + .send(Message::Query(Box::new(QueryJob { query, reply: tx }))) + .map_err(WritebackError::new)?; + } + rx.recv() + .map_err(|_| WritebackError::new("database stopped before its query reply"))? + } + pub fn request_flush(&self) { + let pending = self.shared.pending.lock(); + if pending.closed { + return; + } + let (reply, _) = mpsc::channel(); + let _ = self.sender.send(Message::Flush { + through: pending.accepted, + reply, + }); + } + pub fn flush(&self) -> Result<()> { + if self.runtime.is_current() { + return Err(WritebackError::new( + "cannot synchronously flush the SQL executor from itself", + )); + } + let (tx, rx) = mpsc::channel(); + { + let pending = self.shared.pending.lock(); + if pending.closed { + return Err(WritebackError::new("database is closed")); + } + self.sender + .send(Message::Flush { + through: pending.accepted, + reply: tx, + }) + .map_err(WritebackError::new)?; + } + rx.recv() + .map_err(|_| WritebackError::new("database stopped before its flush receipt"))? + } + pub fn ensure_open(&self) -> Result<()> { + if self.shared.pending.lock().closed { + Err(WritebackError::new("database is closed")) + } else { + Ok(()) + } + } + pub fn failure(&self) -> Option { + self.shared.pending.lock().failure.clone() + } +} + +async fn commit( + connection: Connection, + shared: &Shared, + runtime: &RuntimeHandle, +) -> Result { + let taken = { + let mut pending = shared.pending.lock(); + if pending.batch.is_empty() { + pending.committed = pending.accepted; + pending.operations = 0; + pending.first_dirty = None; + pending.failure = None; + return Ok(connection); + } + let batch = std::mem::take(&mut pending.batch); + let operations = std::mem::take(&mut pending.operations); + let since = pending.first_dirty.take(); + (batch, pending.accepted, operations, since) + }; + let (batch, through, operations, since) = taken; + let (connection, batch, result) = runtime + .blocking(move || { + let mut connection = connection; + let result = batch.write(&mut connection); + (connection, batch, result) + }) + .await + .map_err(WritebackError::new)?; + let mut pending = shared.pending.lock(); + match result { + Ok(()) => { + pending.committed = through; + pending.failure = None; + } + Err(error) => { + let mut batch = batch; + batch.mark_commit_uncertain(); + batch.merge(std::mem::take(&mut pending.batch)); + pending.batch = batch; + pending.operations = pending.operations.saturating_add(operations); + pending.first_dirty = since.or(pending.first_dirty); + pending.failure = Some(error); + } + } + Ok(connection) +} +async fn run( + mut connection: Connection, + mut receiver: async_mpsc::UnboundedReceiver, + shared: Arc>, + runtime: RuntimeHandle, + policy: WritebackPolicy, +) -> Result<()> { + loop { + let deadline = { + let pending = shared.pending.lock(); + if pending.failure.is_some() { + None + } else if policy + .early_commit_operations + .is_some_and(|limit| pending.operations >= limit) + { + Some(tokio::time::Instant::now()) + } else { + pending.first_dirty.map(|since| since + policy.delay) + } + }; + tokio::select! { + message = receiver.recv() => match message { + Some(Message::Query(query)) => { + connection = runtime.blocking(move || { query.run(&mut connection); connection }).await.map_err(WritebackError::new)?; + } + Some(Message::Flush {through, reply}) => { + if shared.pending.lock().committed < through { + connection = commit(connection, &shared, &runtime).await?; + } + let result = { + let pending=shared.pending.lock(); + if pending.committed>=through {Ok(())} + else {Err(pending.failure.clone().unwrap_or_else(||WritebackError::new("flush did not cover its accepted sequence")))} + }; + let _ = reply.send(result); + } + Some(Message::Close) | None => { + connection = commit(connection, &shared, &runtime).await?; + let result = shared.pending.lock().failure.clone().map_or(Ok(()), Err); + let closed = runtime.blocking(move || connection.close().map_err(|(_, error)| WritebackError::new(error))).await.map_err(WritebackError::new)?; + return result.and(closed); + } + }, + _ = shared.changed.notified() => {}, + _ = async { + match deadline { + Some(deadline) => tokio::time::sleep_until(deadline).await, + None => std::future::pending().await, + } + } => { connection = commit(connection, &shared, &runtime).await?; } + } + } +} + +#[cfg(test)] +mod tests; diff --git a/moli-sqlite/src/writeback/tests.rs b/moli-sqlite/src/writeback/tests.rs new file mode 100644 index 0000000000..b725ae319a --- /dev/null +++ b/moli-sqlite/src/writeback/tests.rs @@ -0,0 +1,254 @@ +use super::*; +use crate::Runtime; +use std::{ + collections::BTreeMap, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, + thread, +}; +use tokio::sync::oneshot; + +#[derive(Default)] +struct Control { + gate: Mutex, mpsc::Receiver<()>)>>, + fail: AtomicBool, + commits: AtomicUsize, + committed: Mutex>>, +} +#[derive(Default)] +struct Batch { + values: BTreeMap, + control: Option>, +} +impl Batch { + fn put(key: i64, value: i64, control: &Arc) -> Self { + Self { + values: BTreeMap::from([(key, value)]), + control: Some(control.clone()), + } + } +} +impl WriteBatch for Batch { + fn is_empty(&self) -> bool { + self.values.is_empty() + } + fn merge(&mut self, newer: Self) { + self.values.extend(newer.values); + self.control = newer.control.or(self.control.take()); + } + fn write(&self, connection: &mut Connection) -> Result<()> { + let control = self.control.as_ref().unwrap(); + let gate = control.gate.lock().take(); + if let Some((started, release)) = gate { + started.send(()).unwrap(); + release.recv().unwrap(); + } + if control.fail.swap(false, Ordering::AcqRel) { + return Err(WritebackError::new("injected failure")); + } + let tx = connection.transaction()?; + for (key, value) in &self.values { + tx.execute("INSERT INTO data VALUES (?1,?2) ON CONFLICT(key) DO UPDATE SET value=excluded.value",rusqlite::params![key,value])?; + } + tx.commit()?; + control.commits.fetch_add(1, Ordering::AcqRel); + if let Some(done) = control.committed.lock().take() { + let _ = done.send(()); + } + Ok(()) + } +} +fn initialize(connection: &mut Connection) -> Result<()> { + connection.execute_batch("CREATE TABLE data(key INTEGER PRIMARY KEY,value INTEGER)")?; + Ok(()) +} +fn long_policy() -> WritebackPolicy { + WritebackPolicy { + delay: Duration::from_secs(3600), + early_commit_operations: None, + } +} +fn owner(runtime: &Runtime, path: PathBuf) -> WritebackOwner { + WritebackOwner::open( + path, + OpenMode::Create, + runtime.handle(), + long_policy(), + initialize, + ) + .unwrap() +} +fn value(client: &WritebackClient, key: i64) -> i64 { + client + .query(move |connection| { + Ok(connection.query_row("SELECT value FROM data WHERE key=?1", [key], |r| r.get(0))?) + }) + .unwrap() +} + +#[test] +fn an_in_flight_acknowledgement_cannot_clear_a_newer_batch_and_another_file_can_progress() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let first = owner(&runtime, root.path().join("first.sqlite")); + let second = owner(&runtime, root.path().join("second.sqlite")); + let control = Arc::new(Control::default()); + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + *control.gate.lock() = Some((started_tx, release_rx)); + let client = first.client(); + client + .record(Batch::put(1, 10, &control), FailureAdmission::Retain) + .unwrap(); + client.request_flush(); + started_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + client + .record(Batch::put(1, 20, &control), FailureAdmission::Retain) + .unwrap(); + let other = Arc::new(Control::default()); + second + .client() + .record(Batch::put(2, 42, &other), FailureAdmission::Retain) + .unwrap(); + second.client().flush().unwrap(); + assert_eq!(value(&second.client(), 2), 42); + release_tx.send(()).unwrap(); + assert_eq!(value(&client, 1), 10); + client.flush().unwrap(); + assert_eq!(value(&client, 1), 20); + assert_eq!(control.commits.load(Ordering::Acquire), 2); + first.close().unwrap(); + assert!( + client + .record(Batch::put(1, 99, &control), FailureAdmission::Retain) + .is_err() + ); + second + .client() + .record(Batch::put(2, 43, &other), FailureAdmission::Retain) + .unwrap(); + second.close().unwrap(); +} +#[test] +fn failed_batch_merges_before_newer_changes_and_retry_is_explicit() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let owner = owner(&runtime, root.path().join("failure.sqlite")); + let client = owner.client(); + let control = Arc::new(Control::default()); + control.fail.store(true, Ordering::Release); + client + .record(Batch::put(1, 10, &control), FailureAdmission::Retain) + .unwrap(); + assert!(client.flush().is_err()); + let called = AtomicBool::new(false); + assert!( + client + .accept(FailureAdmission::Reject, || { + called.store(true, Ordering::Release); + Ok::<_, WritebackError>(((), Batch::put(1, 999, &control))) + }) + .is_err() + ); + assert!(!called.load(Ordering::Acquire)); + client + .record(Batch::put(1, 20, &control), FailureAdmission::Retain) + .unwrap(); + client + .record(Batch::put(2, 30, &control), FailureAdmission::Retain) + .unwrap(); + assert!(client.failure().is_some()); + client.flush().unwrap(); + assert!(client.failure().is_none()); + assert_eq!(value(&client, 1), 20); + assert_eq!(value(&client, 2), 30); + let retained = client.clone(); + thread::spawn(move || owner.close().unwrap()) + .join() + .unwrap(); + assert!(retained.query(|_| Ok(())).is_err()); +} +fn virtual_client( + runtime: &Runtime, + policy: WritebackPolicy, +) -> (WritebackClient, tokio::task::JoinHandle>) { + let mut connection = Connection::open_in_memory().unwrap(); + initialize(&mut connection).unwrap(); + let shared = Arc::new(Shared { + pending: Mutex::new(Pending { + batch: Batch::default(), + accepted: 0, + committed: 0, + operations: 0, + first_dirty: None, + failure: None, + closed: false, + }), + changed: Notify::new(), + }); + let (sender, receiver) = async_mpsc::unbounded_channel(); + let client = WritebackClient { + shared: shared.clone(), + sender, + runtime: runtime.handle(), + }; + let task = tokio::spawn(run(connection, receiver, shared, runtime.handle(), policy)); + (client, task) +} +#[tokio::test(start_paused = true)] +async fn the_first_dirty_deadline_is_not_extended_by_later_writes() { + let runtime = Runtime::new().unwrap(); + let (client, task) = virtual_client( + &runtime, + WritebackPolicy { + delay: Duration::from_secs(5), + early_commit_operations: None, + }, + ); + let control = Arc::new(Control::default()); + let (done_tx, done_rx) = oneshot::channel(); + *control.committed.lock() = Some(done_tx); + client + .record(Batch::put(1, 10, &control), FailureAdmission::Retain) + .unwrap(); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(2500)).await; + client + .record(Batch::put(1, 20, &control), FailureAdmission::Retain) + .unwrap(); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(2499)).await; + tokio::task::yield_now().await; + assert_eq!(control.commits.load(Ordering::Acquire), 0); + tokio::time::advance(Duration::from_millis(1)).await; + done_rx.await.unwrap(); + assert_eq!(control.commits.load(Ordering::Acquire), 1); + client.shared.pending.lock().closed = true; + client.sender.send(Message::Close).unwrap(); + task.await.unwrap().unwrap(); +} +#[tokio::test(start_paused = true)] +async fn early_commit_counts_accepted_operations_even_when_the_same_key_is_coalesced() { + let runtime = Runtime::new().unwrap(); + let (client, task) = virtual_client( + &runtime, + WritebackPolicy { + delay: Duration::from_secs(30), + early_commit_operations: Some(512), + }, + ); + let control = Arc::new(Control::default()); + let (done_tx, done_rx) = oneshot::channel(); + *control.committed.lock() = Some(done_tx); + let start = tokio::time::Instant::now(); + for value in 0..512 { + client + .record(Batch::put(1, value, &control), FailureAdmission::Retain) + .unwrap(); + } + assert_eq!(client.shared.pending.lock().batch.values.len(), 1); + done_rx.await.unwrap(); + assert_eq!(tokio::time::Instant::now(), start); + client.shared.pending.lock().closed = true; + client.sender.send(Message::Close).unwrap(); + task.await.unwrap().unwrap(); +} diff --git a/moli-web-storage/Cargo.toml b/moli-web-storage/Cargo.toml index 80caff4c17..9ef241beb2 100644 --- a/moli-web-storage/Cargo.toml +++ b/moli-web-storage/Cargo.toml @@ -7,11 +7,16 @@ edition = "2024" [dependencies] anyhow = "1.0.100" moli-browser-profile = { path = "../moli-browser-profile" } +moli-sqlite = { path = "../moli-sqlite" } +rusqlite = { version = "0.40.2", features = ["bundled"] } moli-storage-key = { path = "../moli-storage-key" } parking_lot = "0.12.5" serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.145", features = ["raw_value"] } typed-num = { version = "0.3.0", default-features = false, features = ["serde"] } +[dev-dependencies] +tempfile = "3.23.0" + [lints] workspace = true diff --git a/moli-web-storage/src/lib.rs b/moli-web-storage/src/lib.rs index 0dd5939f2f..7b8c643773 100644 --- a/moli-web-storage/src/lib.rs +++ b/moli-web-storage/src/lib.rs @@ -29,6 +29,14 @@ fn utf16_units_contain_unpaired_surrogate(units: &[u16]) -> bool { char::decode_utf16(units.iter().copied()).any(|decoded| decoded.is_err()) } +mod sqlite; +mod store; +pub use sqlite::{SqliteWebStorageOwner, WebStorageAreaLease, new_sqlite_web_storage_store}; +pub use store::{ + deep_clone_shared_web_storage_store, new_shared_json_web_storage_store, + new_shared_web_storage_store, +}; + const WEB_STORAGE_QUOTA_BYTES: usize = 5 * 1024 * 1024; type WebStorageJsonVersion = Num<1>; type SerializedWebStorageAreas = BTreeMap>; @@ -47,7 +55,7 @@ fn web_storage_area_key_prefix_for_origin(origin: &str) -> String { #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct WebStorageString { - units: Vec, + units: Arc<[u16]>, } #[derive(Debug, Default, Clone)] @@ -69,16 +77,17 @@ struct JsonWebStorageBackend { } enum WebStorageBackend { - Memory(MemoryWebStorageBackend), - Json(JsonWebStorageBackend), + Memory(Mutex), + Json(Mutex), + Sqlite(sqlite::SqliteBackend), } pub struct WebStorageStore { backend: WebStorageBackend, - mutation_subscribers: Vec, + mutation_subscribers: Mutex>, } -pub type SharedWebStorageStore = Arc>; +pub type SharedWebStorageStore = Arc; #[derive(Debug, Clone, PartialEq, Eq)] pub enum WebStorageMutation { @@ -142,60 +151,6 @@ impl fmt::Display for WebStorageMutationError { impl std::error::Error for WebStorageMutationError {} -pub fn new_shared_web_storage_store() -> SharedWebStorageStore { - Arc::new(Mutex::new(WebStorageStore::default())) -} - -pub fn deep_clone_shared_web_storage_store( - source: &SharedWebStorageStore, -) -> SharedWebStorageStore { - let memory = match &source.lock().backend { - WebStorageBackend::Memory(memory) => memory.clone(), - WebStorageBackend::Json(json) => json.memory.clone(), - }; - Arc::new(Mutex::new(WebStorageStore { - backend: WebStorageBackend::Memory(memory), - mutation_subscribers: Vec::new(), - })) -} - -pub fn new_shared_json_web_storage_store(path: impl AsRef) -> Result { - let backend = JsonWebStorageBackend::open(path.as_ref())?; - Ok(Arc::new(Mutex::new(WebStorageStore { - backend: WebStorageBackend::Json(backend), - mutation_subscribers: Vec::new(), - }))) -} - -impl Default for WebStorageStore { - fn default() -> Self { - Self { - backend: WebStorageBackend::Memory(MemoryWebStorageBackend::default()), - mutation_subscribers: Vec::new(), - } - } -} - -impl fmt::Debug for WebStorageStore { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match &self.backend { - WebStorageBackend::Memory(memory) => f - .debug_struct("WebStorageStore") - .field("backend", &"memory") - .field("origins", &memory.origins.keys().collect::>()) - .field("mutation_subscribers", &self.mutation_subscribers.len()) - .finish(), - WebStorageBackend::Json(json) => f - .debug_struct("WebStorageStore") - .field("backend", &"json") - .field("path", &json.path) - .field("origins", &json.memory.origins.keys().collect::>()) - .field("mutation_subscribers", &self.mutation_subscribers.len()) - .finish(), - } - } -} - impl fmt::Debug for WebStorageMutationSubscription { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("WebStorageMutationSubscription") @@ -225,7 +180,7 @@ impl WebStorageMutationSubscription { impl WebStorageString { pub fn from_utf16_units(units: impl Into>) -> Self { Self { - units: units.into(), + units: units.into().into(), } } @@ -246,11 +201,20 @@ impl WebStorageString { } fn usage_bytes(&self) -> usize { - if self.contains_unpaired_surrogate() { - return self.units.len().saturating_mul(2); + let mut bytes = 0usize; + for decoded in char::decode_utf16(self.units.iter().copied()) { + let Ok(character) = decoded else { + return self.units.len().saturating_mul(2); + }; + bytes = bytes.saturating_add(character.len_utf8()); } - String::from_utf16(&self.units) - .map_or_else(|_| self.units.len().saturating_mul(2), |value| value.len()) + bytes + } +} + +impl std::borrow::Borrow<[u16]> for WebStorageString { + fn borrow(&self) -> &[u16] { + self.as_units() } } @@ -261,43 +225,8 @@ impl From<&str> for WebStorageString { } impl MemoryWebStorageArea { - fn get_item(&self, key: &str) -> Option { - self.get_item_utf16(utf16_units(key).as_slice()) - .map(|value| string_from_utf16_units_lossy(&value)) - } - fn get_item_utf16(&self, key: &[u16]) -> Option> { - self.values - .get(&WebStorageString::from_utf16_units(key.to_vec())) - .map(|value| value.as_units().to_vec()) - } - - fn set_item_utf16(&mut self, key: &[u16], value: &[u16]) -> bool { - let key = WebStorageString::from_utf16_units(key.to_vec()); - let value = WebStorageString::from_utf16_units(value.to_vec()); - let next_size = updated_storage_size(self.size, self.values.get(&key), value.usage_bytes()); - if next_size > WEB_STORAGE_QUOTA_BYTES { - return false; - } - self.size = next_size; - self.values.insert(key, value); - true - } - - fn remove_item_utf16(&mut self, key: &[u16]) -> bool { - let Some(previous) = self - .values - .remove(&WebStorageString::from_utf16_units(key.to_vec())) - else { - return false; - }; - self.size = self.size.saturating_sub(previous.usage_bytes()); - true - } - - fn key(&self, index: usize) -> Option { - self.key_utf16(index) - .map(|key| string_from_utf16_units_lossy(&key)) + self.values.get(key).map(|value| value.as_units().to_vec()) } fn key_utf16(&self, index: usize) -> Option> { @@ -310,24 +239,8 @@ impl MemoryWebStorageArea { self.values.len() } - fn contains_key(&self, key: &str) -> bool { - self.contains_key_utf16(&utf16_units(key)) - } - fn contains_key_utf16(&self, key: &[u16]) -> bool { - self.values - .contains_key(&WebStorageString::from_utf16_units(key.to_vec())) - } - - fn usage_bytes(&self) -> usize { - self.size - } - - fn sorted_keys(&self) -> Vec { - self.sorted_keys_utf16() - .into_iter() - .map(|key| string_from_utf16_units_lossy(&key)) - .collect() + self.values.contains_key(key) } fn sorted_keys_utf16(&self) -> Vec> { @@ -349,396 +262,17 @@ fn updated_storage_size( .saturating_add(new_value_size) } -impl WebStorageStore { - pub fn subscribe_mutations( - &mut self, - area_kind: WebStorageAreaKind, - ) -> WebStorageMutationSubscription { - let queue = Arc::new(Mutex::new(VecDeque::new())); - let subscription = WebStorageMutationSubscription { queue }; - self.add_mutation_subscription(area_kind, &subscription); - subscription - } - - pub fn add_mutation_subscription( - &mut self, - area_kind: WebStorageAreaKind, - subscription: &WebStorageMutationSubscription, - ) { - self.mutation_subscribers.retain(|subscriber| { - let Some(queue) = subscriber.queue.upgrade() else { - return false; - }; - !(subscriber.area_kind == area_kind && Arc::ptr_eq(&queue, &subscription.queue)) - }); - self.mutation_subscribers - .push(WebStorageMutationSubscriber { - area_kind, - queue: Arc::downgrade(&subscription.queue), - }); - } - - pub fn get_item(&mut self, origin: &str, key: &str) -> Option { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.get_item(origin, key), - WebStorageBackend::Json(json) => json.memory.get_item(origin, key), - } - } - - pub fn get_item_utf16(&mut self, origin: &str, key: &[u16]) -> Option> { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.get_item_utf16(origin, key), - WebStorageBackend::Json(json) => json.memory.get_item_utf16(origin, key), - } - } - - pub fn set_item(&mut self, origin: &str, key: &str, value: &str) -> bool { - self.try_set_item(origin, key, value).unwrap_or(false) - } - - pub fn try_set_item( - &mut self, - origin: &str, - key: &str, - value: &str, - ) -> std::result::Result { - self.try_set_item_utf16(origin, &utf16_units(key), &utf16_units(value)) - } - - pub fn set_item_utf16(&mut self, origin: &str, key: &[u16], value: &[u16]) -> bool { - self.try_set_item_utf16(origin, key, value).unwrap_or(false) - } - - pub fn try_set_item_utf16( - &mut self, - origin: &str, - key: &[u16], - value: &[u16], - ) -> std::result::Result { - let previous = self.get_item_utf16(origin, key); - if previous.as_deref() == Some(value) { - return Ok(true); - } - let updated = match &mut self.backend { - WebStorageBackend::Memory(memory) => { - if memory.set_item_utf16(origin, key, value) { - Ok(true) - } else { - Err(WebStorageMutationError::QuotaExceeded) - } - } - WebStorageBackend::Json(json) => json.update(&[origin], |memory| { - memory.set_item_utf16(origin, key, value) - }), - }?; - if updated { - let key = WebStorageString::from_utf16_units(key.to_vec()); - let value = WebStorageString::from_utf16_units(value.to_vec()); - let mutation = match previous { - Some(previous) => WebStorageMutation::ItemUpdated { - area_key: origin.to_owned(), - key, - old_value: WebStorageString::from_utf16_units(previous), - new_value: value, - }, - None => WebStorageMutation::ItemAdded { - area_key: origin.to_owned(), - key, - value, - }, - }; - self.publish_mutation(mutation); - } - Ok(updated) - } - - pub fn remove_item(&mut self, origin: &str, key: &str) -> bool { - self.try_remove_item(origin, key).unwrap_or(false) - } - - pub fn try_remove_item( - &mut self, - origin: &str, - key: &str, - ) -> std::result::Result { - self.try_remove_item_utf16(origin, &utf16_units(key)) - } - - pub fn remove_item_utf16(&mut self, origin: &str, key: &[u16]) -> bool { - self.try_remove_item_utf16(origin, key).unwrap_or(false) - } - - pub fn try_remove_item_utf16( - &mut self, - origin: &str, - key: &[u16], - ) -> std::result::Result { - let Some(previous) = self.get_item_utf16(origin, key) else { - return Ok(false); - }; - let removed = match &mut self.backend { - WebStorageBackend::Memory(memory) => Ok(memory.remove_item_utf16(origin, key)), - WebStorageBackend::Json(json) => { - json.update(&[origin], |memory| memory.remove_item_utf16(origin, key)) - } - }?; - if removed { - self.publish_mutation(WebStorageMutation::ItemRemoved { - area_key: origin.to_owned(), - key: WebStorageString::from_utf16_units(key.to_vec()), - old_value: WebStorageString::from_utf16_units(previous), - }); - } - Ok(removed) - } - - pub fn clear(&mut self, origin: &str) { - let _ = self.try_clear(origin); - } - - pub fn try_clear( - &mut self, - origin: &str, - ) -> std::result::Result { - let had_items = self.len(origin) != 0; - let cleared = match &mut self.backend { - WebStorageBackend::Memory(memory) => Ok(memory.clear_origin(origin)), - WebStorageBackend::Json(json) => { - json.update(&[origin], |memory| memory.clear_origin(origin)) - } - }?; - if cleared && had_items { - self.publish_mutation(WebStorageMutation::ItemsCleared { - area_key: origin.to_owned(), - }); - } - Ok(cleared) - } - - pub fn key(&mut self, origin: &str, index: usize) -> Option { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.key(origin, index), - WebStorageBackend::Json(json) => json.memory.key(origin, index), - } - } - - pub fn key_utf16(&mut self, origin: &str, index: usize) -> Option> { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.key_utf16(origin, index), - WebStorageBackend::Json(json) => json.memory.key_utf16(origin, index), - } - } - - pub fn len(&mut self, origin: &str) -> usize { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.len(origin), - WebStorageBackend::Json(json) => json.memory.len(origin), - } - } - - pub fn contains_key(&mut self, origin: &str, key: &str) -> bool { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.contains_key(origin, key), - WebStorageBackend::Json(json) => json.memory.contains_key(origin, key), - } - } - - pub fn contains_key_utf16(&mut self, origin: &str, key: &[u16]) -> bool { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.contains_key_utf16(origin, key), - WebStorageBackend::Json(json) => json.memory.contains_key_utf16(origin, key), - } - } - - pub fn sorted_keys(&mut self, origin: &str) -> Vec { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.sorted_keys(origin), - WebStorageBackend::Json(json) => json.memory.sorted_keys(origin), - } - } - - pub fn sorted_keys_utf16(&mut self, origin: &str) -> Vec> { - match &mut self.backend { - WebStorageBackend::Memory(memory) => memory.sorted_keys_utf16(origin), - WebStorageBackend::Json(json) => json.memory.sorted_keys_utf16(origin), - } - } - - pub fn usage_bytes(&self, origin: &str) -> usize { - match &self.backend { - WebStorageBackend::Memory(memory) => memory.usage_bytes(origin), - WebStorageBackend::Json(json) => json.memory.usage_bytes(origin), - } - } - - pub fn usage_bytes_for_origin_areas(&self, origin: &str) -> usize { - match &self.backend { - WebStorageBackend::Memory(memory) => memory.usage_bytes_for_origin_areas(origin), - WebStorageBackend::Json(json) => json.memory.usage_bytes_for_origin_areas(origin), - } - } - - pub fn clear_origin(&mut self, origin: &str) { - self.clear(origin); - } - - pub fn try_clear_origin( - &mut self, - origin: &str, - ) -> std::result::Result { - self.try_clear(origin) - } - - pub fn clear_origin_areas(&mut self, origin: &str) { - let _ = self.try_clear_origin_areas(origin); - } - - pub fn try_clear_origin_areas( - &mut self, - origin: &str, - ) -> std::result::Result { - let (area_keys, nonempty_area_keys) = match &self.backend { - WebStorageBackend::Memory(memory) => ( - memory.area_keys_for_origin(origin), - memory.nonempty_area_keys_for_origin(origin), - ), - WebStorageBackend::Json(json) => ( - json.memory.area_keys_for_origin(origin), - json.memory.nonempty_area_keys_for_origin(origin), - ), - }; - if area_keys.is_empty() { - return Ok(false); - } - let cleared = match &mut self.backend { - WebStorageBackend::Memory(memory) => Ok(memory.clear_origin_areas(origin)), - WebStorageBackend::Json(json) => { - let affected_areas = area_keys.iter().map(String::as_str).collect::>(); - json.update(&affected_areas, |memory| memory.clear_origin_areas(origin)) - } - }?; - if cleared { - for area_key in nonempty_area_keys { - self.publish_mutation(WebStorageMutation::ItemsCleared { area_key }); - } - } - Ok(cleared) - } - - fn publish_mutation(&mut self, mutation: WebStorageMutation) { - self.mutation_subscribers.retain(|subscriber| { - let Some(queue) = subscriber.queue.upgrade() else { - return false; - }; - queue.lock().push_back(WebStorageMutationRecord { - area_kind: subscriber.area_kind, - mutation: mutation.clone(), - }); - true - }); - } -} - impl MemoryWebStorageBackend { - fn get_item(&self, origin: &str, key: &str) -> Option { - self.origins.get(origin).and_then(|area| area.get_item(key)) - } - - fn get_item_utf16(&self, origin: &str, key: &[u16]) -> Option> { - self.origins - .get(origin) - .and_then(|area| area.get_item_utf16(key)) - } - - fn set_item_utf16(&mut self, origin: &str, key: &[u16], value: &[u16]) -> bool { - self.origins - .entry(origin.to_owned()) - .or_default() - .set_item_utf16(key, value) - } - - fn remove_item_utf16(&mut self, origin: &str, key: &[u16]) -> bool { - let (removed, clear_origin) = { - let Some(area) = self.origins.get_mut(origin) else { - return false; - }; - let removed = area.remove_item_utf16(key); - (removed, area.len() == 0) - }; - if clear_origin { - self.origins.remove(origin); - } - removed - } - fn clear_origin(&mut self, origin: &str) -> bool { self.origins.remove(origin).is_some() } - fn key(&self, origin: &str, index: usize) -> Option { - self.origins.get(origin).and_then(|area| area.key(index)) - } - - fn key_utf16(&self, origin: &str, index: usize) -> Option> { - self.origins - .get(origin) - .and_then(|area| area.key_utf16(index)) - } - fn len(&self, origin: &str) -> usize { self.origins .get(origin) .map_or(0, MemoryWebStorageArea::len) } - fn contains_key(&self, origin: &str, key: &str) -> bool { - self.origins - .get(origin) - .is_some_and(|area| area.contains_key(key)) - } - - fn contains_key_utf16(&self, origin: &str, key: &[u16]) -> bool { - self.origins - .get(origin) - .is_some_and(|area| area.contains_key_utf16(key)) - } - - fn sorted_keys(&self, origin: &str) -> Vec { - self.origins - .get(origin) - .map_or_else(Vec::new, MemoryWebStorageArea::sorted_keys) - } - - fn sorted_keys_utf16(&self, origin: &str) -> Vec> { - self.origins - .get(origin) - .map_or_else(Vec::new, MemoryWebStorageArea::sorted_keys_utf16) - } - - fn usage_bytes(&self, origin: &str) -> usize { - self.origins - .get(origin) - .map_or(0, MemoryWebStorageArea::usage_bytes) - } - - fn usage_bytes_for_origin_areas(&self, origin: &str) -> usize { - let area_prefix = web_storage_area_key_prefix_for_origin(origin); - self.origins - .iter() - .filter(|(area_key, _)| area_key.starts_with(&area_prefix)) - .map(|(_, area)| area.usage_bytes()) - .sum() - } - - fn clear_origin_areas(&mut self, origin: &str) -> bool { - let area_keys = self.area_keys_for_origin(origin); - let removed = !area_keys.is_empty(); - for area_key in area_keys { - self.origins.remove(&area_key); - } - removed - } - fn area_keys_for_origin(&self, origin: &str) -> Vec { let area_prefix = web_storage_area_key_prefix_for_origin(origin); self.origins @@ -747,13 +281,6 @@ impl MemoryWebStorageBackend { .cloned() .collect() } - - fn nonempty_area_keys_for_origin(&self, origin: &str) -> Vec { - self.area_keys_for_origin(origin) - .into_iter() - .filter(|area_key| self.len(area_key) > 0) - .collect() - } } impl JsonWebStorageBackend { @@ -968,7 +495,7 @@ mod tests { #[test] fn memory_web_storage_preserves_sorted_key_and_quota_behavior() { - let mut store = WebStorageStore::default(); + let store = WebStorageStore::default(); let area_key = first_party_area_key("https://a.test"); assert!(store.set_item(&area_key, "b", "2")); assert!(store.set_item(&area_key, "a", "1")); @@ -990,7 +517,7 @@ mod tests { #[test] fn mutation_subscription_reports_only_successful_observable_changes() { - let mut store = WebStorageStore::default(); + let store = WebStorageStore::default(); let area_key = first_party_area_key("https://a.test"); let subscription = store.subscribe_mutations(WebStorageAreaKind::Local); @@ -1054,13 +581,14 @@ mod tests { assert!(subscription.is_empty()); let empty_area_key = first_party_area_key("https://empty.test"); - match &mut store.backend { + match &store.backend { WebStorageBackend::Memory(memory) => { memory + .lock() .origins .insert(empty_area_key.clone(), MemoryWebStorageArea::default()); } - WebStorageBackend::Json(_) => panic!("default store should use the memory backend"), + _ => panic!("default store should use the memory backend"), } assert!( store @@ -1071,7 +599,7 @@ mod tests { panic!("default store should use the memory backend"); }; assert!( - !memory.origins.contains_key(&empty_area_key), + !memory.lock().origins.contains_key(&empty_area_key), "clear must preserve the existing area cleanup semantics" ); assert!( @@ -1081,14 +609,14 @@ mod tests { drop(subscription); assert!(store.set_item(&area_key, "after-drop", "4")); - assert!(store.mutation_subscribers.is_empty()); + assert!(store.mutation_subscribers.lock().is_empty()); } #[test] fn one_mutation_subscription_preserves_cross_store_order_and_deduplicates_registration() { let area_key = first_party_area_key("https://a.test"); - let mut local = WebStorageStore::default(); - let mut session = WebStorageStore::default(); + let local = WebStorageStore::default(); + let session = WebStorageStore::default(); let subscription = local.subscribe_mutations(WebStorageAreaKind::Local); local.add_mutation_subscription(WebStorageAreaKind::Local, &subscription); session.add_mutation_subscription(WebStorageAreaKind::Session, &subscription); @@ -1118,17 +646,17 @@ mod tests { let a_area_key = first_party_area_key("https://a.test"); let b_area_key = first_party_area_key("https://b.test"); { - let mut source = source.lock(); + let source = source.as_ref(); assert!(source.set_item(&a_area_key, "shared", "source")); assert!(source.set_item(&b_area_key, "other", "copied")); } let subscription = source - .lock() + .as_ref() .subscribe_mutations(WebStorageAreaKind::Session); let cloned = deep_clone_shared_web_storage_store(&source); { - let mut cloned = cloned.lock(); + let cloned = cloned.as_ref(); assert_eq!( cloned.get_item(&a_area_key, "shared"), Some("source".to_owned()) @@ -1145,7 +673,7 @@ mod tests { "a deep-cloned sessionStorage namespace must not retain opener mutation subscribers" ); { - let mut source = source.lock(); + let source = source.as_ref(); assert_eq!( source.get_item(&a_area_key, "shared"), Some("source".to_owned()) @@ -1153,7 +681,7 @@ mod tests { assert_eq!(source.get_item(&a_area_key, "clone-only"), None); assert!(source.set_item(&b_area_key, "source-only", "yes")); } - assert_eq!(cloned.lock().get_item(&b_area_key, "source-only"), None); + assert_eq!(cloned.as_ref().get_item(&b_area_key, "source-only"), None); } #[test] @@ -1164,7 +692,7 @@ mod tests { { let store = new_shared_json_web_storage_store(&temp.path) .expect("json web storage should open"); - let mut store = store.lock(); + let store = store.as_ref(); assert!(store.set_item(&a_area_key, "b", "2")); assert!(store.set_item(&a_area_key, "a", "1")); assert!(store.set_item(&b_area_key, "a", "other")); @@ -1173,7 +701,7 @@ mod tests { let store = new_shared_json_web_storage_store(&temp.path).expect("json web storage should reopen"); - let mut store = store.lock(); + let store = store.as_ref(); assert_eq!(store.get_item(&a_area_key, "a"), Some("1".to_owned())); assert_eq!(store.get_item(&b_area_key, "a"), Some("other".to_owned())); assert_eq!(store.len(&a_area_key), 2); @@ -1189,7 +717,7 @@ mod tests { let other_area_key = first_party_area_key("https://b.test"); let shared = new_shared_json_web_storage_store(&temp.path).expect("json web storage should open"); - let mut store = shared.lock(); + let store = shared.as_ref(); assert!(store.set_item(&area_key, "a", "1")); assert!(store.set_item(&other_area_key, "b", "2")); fs::remove_file(&temp.path).expect("storage file should be removed"); @@ -1213,7 +741,7 @@ mod tests { assert!(store.set_item(&other_area_key, "b", "recovered")); let reopened = new_shared_json_web_storage_store(&temp.path) .expect("json web storage should reopen after recovery"); - let mut reopened = reopened.lock(); + let reopened = reopened.as_ref(); assert_eq!(reopened.get_item(&area_key, "a"), Some("1".to_owned())); assert_eq!( reopened.get_item(&other_area_key, "b"), @@ -1233,7 +761,7 @@ mod tests { { let shared = new_shared_json_web_storage_store(&temp.path).expect("storage should open"); - let mut store = shared.lock(); + let store = shared.as_ref(); assert!(store.set_item(&first_party, "existing", "first")); assert!(store.set_item(&partitioned, "existing", "partitioned")); assert!(store.set_item_utf16(&unrelated, &surrogate_key, &surrogate_value)); @@ -1241,7 +769,7 @@ mod tests { { let shared = new_shared_json_web_storage_store(&temp.path).expect("storage should reopen"); - let mut store = shared.lock(); + let store = shared.as_ref(); assert!(store.set_item(&first_party, "existing", "updated")); assert!(store.set_item(&first_party, "new", "added")); assert!(store.remove_item(&first_party, "existing")); @@ -1257,7 +785,7 @@ mod tests { } let shared = new_shared_json_web_storage_store(&temp.path) .expect("storage should reopen after clearing"); - let mut store = shared.lock(); + let store = shared.as_ref(); assert_eq!(store.len(&first_party), 0); assert_eq!(store.len(&partitioned), 0); assert_eq!( @@ -1265,15 +793,14 @@ mod tests { Some(surrogate_value.to_vec()) ); assert!(store.set_item(&first_party, "new", "restored")); - drop(store); let reopened = new_shared_json_web_storage_store(&temp.path) .expect("storage should reopen after restoring"); assert_eq!( - reopened.lock().get_item(&first_party, "new"), + reopened.as_ref().get_item(&first_party, "new"), Some("restored".to_owned()) ); assert_eq!( - reopened.lock().get_item_utf16(&unrelated, &surrogate_key), + reopened.as_ref().get_item_utf16(&unrelated, &surrogate_key), Some(surrogate_value.to_vec()) ); } @@ -1287,7 +814,7 @@ mod tests { { let store = new_shared_json_web_storage_store(&temp.path) .expect("json web storage should open"); - let mut store = store.lock(); + let store = store.as_ref(); assert!(store.set_item_utf16(&area_key, &key, &value)); } @@ -1307,14 +834,14 @@ mod tests { let store = new_shared_json_web_storage_store(&temp.path).expect("json web storage should reopen"); - let mut store = store.lock(); + let store = store.as_ref(); assert_eq!(store.get_item_utf16(&area_key, &key), Some(value.clone())); assert_eq!(store.key_utf16(&area_key, 0), Some(key)); } #[test] fn web_storage_usage_tracks_value_bytes_by_storage_key() { - let mut store = WebStorageStore::default(); + let store = WebStorageStore::default(); let a_area_key = first_party_area_key("https://a.test"); let b_area_key = first_party_area_key("https://b.test"); assert_eq!(store.usage_bytes(&a_area_key), 0); @@ -1346,7 +873,7 @@ mod tests { "https://sibling.example.test", "https://top-a.example.test", ); - let mut store = WebStorageStore::default(); + let store = WebStorageStore::default(); assert!(store.set_item(&first_party, "fp", "a")); assert!(store.set_item(&top_a, "a", "bb")); assert!(store.set_item(&top_b, "b", "ccc")); diff --git a/moli-web-storage/src/sqlite.rs b/moli-web-storage/src/sqlite.rs new file mode 100644 index 0000000000..9498c229ba --- /dev/null +++ b/moli-web-storage/src/sqlite.rs @@ -0,0 +1,468 @@ +use super::{store::Edit, *}; +use moli_sqlite::writeback::{ + FailureAdmission, WriteBatch, WritebackClient, WritebackError, WritebackOwner, WritebackPolicy, +}; +use moli_sqlite::{OpenMode, RuntimeHandle}; +use rusqlite::{Connection, OptionalExtension, params}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +const APPLICATION_ID: i32 = 0x4d4c4c53; // MLLS +pub struct SqliteWebStorageOwner { + owner: WritebackOwner, +} +struct Area { + state: Mutex>, + views: AtomicUsize, +} +pub struct WebStorageAreaLease { + area: Arc, + client: WritebackClient, +} +impl Drop for WebStorageAreaLease { + fn drop(&mut self) { + if self.area.views.fetch_sub(1, Ordering::AcqRel) == 1 { + self.client.request_flush(); + } + } +} +#[derive(Default)] +struct Cache { + areas: HashMap>, + insertions: usize, +} +pub(super) struct SqliteBackend { + pub(super) client: WritebackClient, + cache: Mutex, + pub(super) namespace: parking_lot::RwLock<()>, +} +#[derive(Default)] +pub(super) struct Batch(BTreeMap); +struct AreaDelta { + clear: bool, + changes: BTreeMap, + usage: usize, + count: usize, + _pin: Arc, +} +enum KeyChange { + Put { + value: WebStorageString, + existed: bool, + }, + Delete, +} +impl WriteBatch for Batch { + fn is_empty(&self) -> bool { + self.0.is_empty() + } + fn merge(&mut self, newer: Self) { + for (key, mut next) in newer.0 { + if let Some(mut previous) = self.0.remove(&key) + && !next.clear + { + for (key, change) in next.changes { + match change { + KeyChange::Put { value, existed } => { + let existed = match previous.changes.remove(&key) { + Some(KeyChange::Put { existed, .. }) => existed, + Some(KeyChange::Delete) => true, + None => existed && !previous.clear, + }; + previous + .changes + .insert(key, KeyChange::Put { value, existed }); + } + KeyChange::Delete => { + if !matches!( + previous.changes.remove(&key), + Some(KeyChange::Put { existed: false, .. }) + ) && !previous.clear + { + previous.changes.insert(key, KeyChange::Delete); + } + } + } + } + next.changes = previous.changes; + next.clear = previous.clear; + } + self.0.insert(key, next); + } + } + fn mark_commit_uncertain(&mut self) { + for delta in self.0.values_mut() { + for change in delta.changes.values_mut() { + if let KeyChange::Put { existed, .. } = change { + *existed = true; + } + } + } + } + fn write(&self, connection: &mut Connection) -> std::result::Result<(), WritebackError> { + let tx = connection.transaction()?; + for (key, delta) in &self.0 { + if delta.count == 0 { + tx.execute("DELETE FROM areas WHERE storage_key=?1", [key])?; + continue; + } + let area: i64 = tx.query_row( + "INSERT INTO areas(storage_key,usage_bytes,entry_count) VALUES (?1,?2,?3) ON CONFLICT(storage_key) DO UPDATE SET usage_bytes=excluded.usage_bytes,entry_count=excluded.entry_count RETURNING area_id", + params![key, i64::try_from(delta.usage).map_err(WritebackError::new)?, i64::try_from(delta.count).map_err(WritebackError::new)?], |r| r.get(0))?; + if delta.clear { + tx.execute("DELETE FROM entries WHERE area_id=?1", [area])?; + } + for (key, value) in &delta.changes { + match value { + KeyChange::Put { value, .. } => { + tx.execute("INSERT INTO entries(area_id,key,value) VALUES (?1,?2,?3) ON CONFLICT(area_id,key) DO UPDATE SET value=excluded.value", params![area, encode(key.as_units()), encode(value.as_units())])?; + } + KeyChange::Delete => { + tx.execute( + "DELETE FROM entries WHERE area_id=?1 AND key=?2", + params![area, encode(key.as_units())], + )?; + } + } + } + } + tx.commit()?; + Ok(()) + } +} +impl From for WebStorageMutationError { + fn from(error: WritebackError) -> Self { + Self::Persistence(error.to_string()) + } +} +pub fn new_sqlite_web_storage_store( + path: PathBuf, + mode: OpenMode, + runtime: RuntimeHandle, +) -> Result<(SharedWebStorageStore, SqliteWebStorageOwner)> { + let owner = WritebackOwner::open( + path, + mode, + runtime, + WritebackPolicy { + delay: std::time::Duration::from_secs(5), + early_commit_operations: None, + }, + initialize, + )?; + let backend = SqliteBackend { + client: owner.client(), + cache: Mutex::new(Cache::default()), + namespace: parking_lot::RwLock::new(()), + }; + let store = Arc::new(WebStorageStore { + backend: WebStorageBackend::Sqlite(backend), + mutation_subscribers: Mutex::new(Vec::new()), + }); + Ok((store, SqliteWebStorageOwner { owner })) +} +fn initialize(connection: &mut Connection) -> std::result::Result<(), WritebackError> { + connection.pragma_update(None, "foreign_keys", true)?; + let application: i32 = connection.pragma_query_value(None, "application_id", |r| r.get(0))?; + let version: i32 = connection.pragma_query_value(None, "user_version", |r| r.get(0))?; + if application == 0 && version == 0 { + let tables: i64 = connection.query_row( + "SELECT count(*) FROM sqlite_master WHERE name NOT LIKE 'sqlite_%'", + [], + |r| r.get(0), + )?; + if tables != 0 { + return Err(WritebackError::new("unidentified localStorage database")); + } + let tx = connection.transaction()?; + tx.execute_batch("CREATE TABLE areas(area_id INTEGER PRIMARY KEY,storage_key TEXT NOT NULL UNIQUE,usage_bytes INTEGER NOT NULL CHECK(usage_bytes>=0),entry_count INTEGER NOT NULL CHECK(entry_count>=0)); + CREATE TABLE entries(area_id INTEGER NOT NULL REFERENCES areas(area_id) ON DELETE CASCADE,key BLOB NOT NULL CHECK(length(key)%2=0),value BLOB NOT NULL CHECK(length(value)%2=0),PRIMARY KEY(area_id,key)) WITHOUT ROWID; + CREATE TABLE migration(source_fingerprint TEXT NOT NULL,complete INTEGER NOT NULL CHECK(complete=1));")?; + tx.pragma_update(None, "application_id", APPLICATION_ID)?; + tx.pragma_update(None, "user_version", 1)?; + tx.commit()?; + } else if application != APPLICATION_ID || version != 1 { + return Err(WritebackError::new(format!( + "unsupported localStorage database identity/version {application}/{version}" + ))); + } + // Verify the actual schema even when a database claims the current version. + connection.prepare("SELECT area_id,storage_key,usage_bytes,entry_count FROM areas LIMIT 0")?; + connection.prepare("SELECT area_id,key,value FROM entries LIMIT 0")?; + Ok(()) +} +fn encode(units: &[u16]) -> Vec { + units.iter().flat_map(|u| u.to_le_bytes()).collect() +} +fn decode(bytes: Vec) -> std::result::Result { + if !bytes.len().is_multiple_of(2) { + return Err(WritebackError::new("invalid UTF-16LE localStorage row")); + } + Ok(WebStorageString::from_utf16_units( + bytes + .chunks_exact(2) + .map(|p| u16::from_le_bytes([p[0], p[1]])) + .collect::>(), + )) +} +impl SqliteBackend { + fn area(&self, origin: &str) -> Arc { + let mut cache = self.cache.lock(); + if let Some(area) = cache.areas.get(origin).and_then(Weak::upgrade) { + return area; + } + cache.insertions += 1; + if cache.insertions >= 64 { + cache.areas.retain(|_, area| area.strong_count() != 0); + cache.insertions = 0; + } + let area = Arc::new(Area { + state: Mutex::new(None), + views: AtomicUsize::new(0), + }); + cache.areas.insert(origin.to_owned(), Arc::downgrade(&area)); + area + } + fn load( + &self, + origin: &str, + state: &mut Option, + ) -> std::result::Result<(), WebStorageMutationError> { + self.client.ensure_open()?; + if state.is_some() { + return Ok(()); + } + let origin = origin.to_owned(); + *state = Some(self.client.query(move |connection| { + let metadata: Option<(i64, i64, i64)> = connection + .query_row( + "SELECT area_id,usage_bytes,entry_count FROM areas WHERE storage_key=?1", + [&origin], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .optional()?; + let Some((id, usage, count)) = metadata else { + return Ok(MemoryWebStorageArea::default()); + }; + let mut statement = + connection.prepare("SELECT key,value FROM entries WHERE area_id=?1")?; + let rows = statement.query_map([id], |r| { + Ok((r.get::<_, Vec>(0)?, r.get::<_, Vec>(1)?)) + })?; + let mut values = HashMap::new(); + for row in rows { + let (key, value) = row?; + values.insert(decode(key)?, decode(value)?); + } + let size = values.values().map(WebStorageString::usage_bytes).sum(); + if usize::try_from(usage).ok() != Some(size) + || usize::try_from(count).ok() != Some(values.len()) + { + return Err(WritebackError::new( + "localStorage area metadata disagrees with its entries", + )); + } + Ok(MemoryWebStorageArea { values, size }) + })?); + Ok(()) + } + pub(super) fn read( + &self, + origin: &str, + read: impl FnOnce(Option<&MemoryWebStorageArea>) -> R, + ) -> std::result::Result { + let _namespace = self.namespace.read(); + let area = self.area(origin); + let mut state = area.state.lock(); + self.load(origin, &mut state)?; + Ok(read(state.as_ref())) + } + pub(super) fn edit( + &self, + origin: &str, + edit: impl FnOnce( + &mut MemoryWebStorageArea, + ) -> std::result::Result, WebStorageMutationError>, + publish: impl FnOnce(Edit) -> R, + ) -> std::result::Result { + let _namespace = self.namespace.read(); + self.edit_unfenced(origin, edit, publish) + } + pub(super) fn edit_unfenced( + &self, + origin: &str, + edit: impl FnOnce( + &mut MemoryWebStorageArea, + ) -> std::result::Result, WebStorageMutationError>, + publish: impl FnOnce(Edit) -> R, + ) -> std::result::Result { + let area = self.area(origin); + let mut state = area.state.lock(); + self.load(origin, &mut state)?; + let data = state.as_mut().expect("loaded area"); + let result = self.client.accept(FailureAdmission::Reject, || { + let result = edit(data)?; + let mut batch = Batch::default(); + if let Some(mutation) = &result.mutation { + let mut delta = AreaDelta { + clear: false, + changes: BTreeMap::new(), + usage: data.size, + count: data.values.len(), + _pin: area.clone(), + }; + match mutation { + WebStorageMutation::ItemAdded { key, value, .. } => { + delta.changes.insert( + key.clone(), + KeyChange::Put { + value: value.clone(), + existed: false, + }, + ); + } + WebStorageMutation::ItemUpdated { key, new_value, .. } => { + delta.changes.insert( + key.clone(), + KeyChange::Put { + value: new_value.clone(), + existed: true, + }, + ); + } + WebStorageMutation::ItemRemoved { key, .. } => { + delta.changes.insert(key.clone(), KeyChange::Delete); + } + WebStorageMutation::ItemsCleared { .. } => delta.clear = true, + } + batch.0.insert(origin.to_owned(), delta); + } + Ok::<_, WebStorageMutationError>((result, batch)) + })?; + Ok(publish(result)) + } + pub(super) fn clear_origin_areas( + &self, + origin: &str, + publish: impl Fn(WebStorageMutation), + ) -> std::result::Result { + let _namespace = self.namespace.write(); + let areas = self + .area_keys_for_origin(origin)? + .into_iter() + .map(|key| { + let area = self.area(&key); + (key, area) + }) + .collect::>(); + // Finish all cold reads before changing any map. The namespace fence + // prevents a concurrent cold load or new area from restoring old data. + for (key, area) in &areas { + self.load(key, &mut area.state.lock())?; + } + let mutations = self.client.accept(FailureAdmission::Reject, || { + let mut batch = Batch::default(); + let mut mutations = Vec::new(); + for (key, area) in &areas { + let mut state = area.state.lock(); + let data = state.as_mut().expect("loaded area"); + if data.values.is_empty() { + continue; + } + data.values.clear(); + data.size = 0; + batch.0.insert( + key.clone(), + AreaDelta { + clear: true, + changes: BTreeMap::new(), + usage: 0, + count: 0, + _pin: area.clone(), + }, + ); + mutations.push(WebStorageMutation::ItemsCleared { + area_key: key.clone(), + }); + } + Ok::<_, WebStorageMutationError>((mutations, batch)) + })?; + let changed = !mutations.is_empty(); + for mutation in mutations { + publish(mutation); + } + Ok(changed) + } + pub(super) fn area_keys_for_origin( + &self, + origin: &str, + ) -> std::result::Result, WebStorageMutationError> { + let prefix = web_storage_area_key_prefix_for_origin(origin); + let query_origin = origin.to_owned(); + let query_prefix = prefix.clone(); + let mut keys = self.client.query(move |connection| { + let mut statement=connection.prepare("SELECT storage_key FROM areas WHERE storage_key=?1 OR substr(storage_key,1,length(?2))=?2")?; + Ok(statement.query_map(params![query_origin,query_prefix],|r|r.get::<_,String>(0))?.collect::,_>>()?) + })?; + keys.extend( + self.cache + .lock() + .areas + .iter() + .filter(|(_, area)| area.strong_count() != 0) + .map(|(key, _)| key.clone()), + ); + keys.retain(|key| key == origin || key.starts_with(&prefix)); + keys.sort(); + keys.dedup(); + Ok(keys) + } +} +impl WebStorageStore { + pub fn lease_area( + &self, + origin: &str, + ) -> std::result::Result, WebStorageMutationError> { + let WebStorageBackend::Sqlite(sqlite) = &self.backend else { + return Ok(None); + }; + sqlite.client.ensure_open()?; + let area = sqlite.area(origin); + area.views.fetch_add(1, Ordering::AcqRel); + Ok(Some(WebStorageAreaLease { + area, + client: sqlite.client.clone(), + })) + } +} +impl SqliteWebStorageOwner { + pub fn flush(&self) -> Result<()> { + self.owner.client().flush()?; + Ok(()) + } + pub fn close(&self) -> Result<()> { + self.owner.close()?; + Ok(()) + } + pub fn checkpoint(&self) -> Result<()> { + self.flush()?; + self.owner.client().query(|connection| { + connection.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")?; + Ok(()) + })?; + Ok(()) + } + pub fn migration_fingerprint(&self) -> Result> { + Ok(self.owner.client().query(|connection| { + Ok(connection + .query_row( + "SELECT source_fingerprint FROM migration WHERE complete=1", + [], + |r| r.get(0), + ) + .optional()?) + })?) + } +} + +#[cfg(test)] +mod tests; diff --git a/moli-web-storage/src/sqlite/tests.rs b/moli-web-storage/src/sqlite/tests.rs new file mode 100644 index 0000000000..016187bcd1 --- /dev/null +++ b/moli-web-storage/src/sqlite/tests.rs @@ -0,0 +1,239 @@ +use super::*; +use moli_sqlite::Runtime; +use std::{sync::mpsc, thread, time::Duration}; + +fn open(root: &Path, runtime: &Runtime) -> (SharedWebStorageStore, SqliteWebStorageOwner) { + new_sqlite_web_storage_store( + root.join("localstorage.sqlite"), + OpenMode::Create, + runtime.handle(), + ) + .unwrap() +} +fn backend(store: &WebStorageStore) -> &SqliteBackend { + let WebStorageBackend::Sqlite(backend) = &store.backend else { + panic!("SQLite fixture"); + }; + backend +} +fn resident(store: &WebStorageStore) -> usize { + backend(store) + .cache + .lock() + .areas + .values() + .filter(|area| area.strong_count() > 0) + .count() +} + +#[test] +fn utf16_blobs_empty_strings_and_partitioned_areas_survive_close_and_reopen() { + let root = tempfile::tempdir().unwrap(); + let mut runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let a = web_storage_partitioned_area_key("https://a.test", "https://top.test"); + let b = web_storage_partitioned_area_key("https://a.test", "https://other.test"); + let key = [0xd800, 0, 0xdc00]; + let value = [0xdc00, 0, 0xd800]; + store.try_set_item_utf16(&a, &key, &value).unwrap(); + store.try_set_item(&a, "", "").unwrap(); + store.try_set_item(&b, "private", "other").unwrap(); + owner.close().unwrap(); + assert!(store.try_get_item_utf16(&a, &key).is_err()); + drop(store); + drop(owner); + let (store, owner) = open(root.path(), &runtime); + assert_eq!(resident(&store), 0); + let _view = store.lease_area(&a).unwrap(); + assert_eq!( + store.try_get_item_utf16(&a, &key).unwrap(), + Some(value.to_vec()) + ); + assert_eq!(store.try_get_item_utf16(&a, &[]).unwrap(), Some(vec![])); + assert_eq!(store.try_len(&a).unwrap(), 2); + assert_eq!(store.try_get_item_utf16(&b, &key).unwrap(), None); + assert_eq!(store.get_item(&b, "private").as_deref(), Some("other")); + owner.close().unwrap(); + runtime.shutdown().unwrap(); +} +#[test] +fn warmed_reads_and_writes_do_not_wait_for_the_database() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let key = "https://warm.test"; + let _view = store.lease_area(key).unwrap(); + store.try_set_item(key, "value", "before").unwrap(); + owner.flush().unwrap(); + let client = backend(&store).client.clone(); + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let blocked = thread::spawn(move || { + client.query(move |_| { + started_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + Ok(()) + }) + }); + started_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + store + .try_set_item(key, "value", "accepted while SQL is blocked") + .unwrap(); + assert_eq!( + store.get_item(key, "value").as_deref(), + Some("accepted while SQL is blocked") + ); + release_tx.send(()).unwrap(); + blocked.join().unwrap().unwrap(); + owner.flush().unwrap(); +} +#[test] +fn asynchronous_failure_keeps_accepted_values_and_rejects_later_mutations_until_explicit_flush() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let key = "https://failure.test"; + let _view = store.lease_area(key).unwrap(); + let events = store.subscribe_mutations(WebStorageAreaKind::Local); + backend(&store).client.query(|connection| {connection.execute_batch("CREATE TRIGGER fail_entries BEFORE INSERT ON entries BEGIN SELECT RAISE(FAIL,'injected persistence failure'); END;")?;Ok(())}).unwrap(); + store.try_set_item(key, "key", "accepted").unwrap(); + assert!(owner.flush().is_err()); + assert_eq!(store.get_item(key, "key").as_deref(), Some("accepted")); + assert_eq!(events.drain().len(), 1); + assert!(matches!( + store.try_set_item(key, "key", "rejected"), + Err(WebStorageMutationError::Persistence(_)) + )); + assert_eq!(store.get_item(key, "key").as_deref(), Some("accepted")); + assert!(events.is_empty()); + backend(&store) + .client + .query(|connection| { + connection.execute_batch("DROP TRIGGER fail_entries")?; + Ok(()) + }) + .unwrap(); + owner.flush().unwrap(); + store.try_set_item(key, "key", "recovered").unwrap(); + owner.close().unwrap(); + let (store, owner) = open(root.path(), &runtime); + assert_eq!(store.get_item(key, "key").as_deref(), Some("recovered")); + owner.close().unwrap(); +} +#[test] +fn clear_then_put_composes_across_areas_and_logical_events_remain_individual() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let _a = store.lease_area("a").unwrap(); + let _b = store.lease_area("b").unwrap(); + let events = store.subscribe_mutations(WebStorageAreaKind::Local); + store.try_set_item("a", "old", "gone").unwrap(); + store.try_set_item("b", "b", "separate").unwrap(); + owner.flush().unwrap(); + events.drain(); + store.try_set_item("a", "k", "one").unwrap(); + store.try_set_item("a", "k", "two").unwrap(); + store.try_clear("a").unwrap(); + store.try_set_item("a", "new", "kept").unwrap(); + assert_eq!(events.drain().len(), 4); + owner.close().unwrap(); + let (store, owner) = open(root.path(), &runtime); + assert_eq!(store.sorted_keys("a"), vec!["new"]); + assert_eq!(store.get_item("a", "new").as_deref(), Some("kept")); + assert_eq!(store.get_item("b", "b").as_deref(), Some("separate")); + owner.close().unwrap(); +} +#[test] +fn last_view_release_flushes_and_clean_areas_can_be_unloaded() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let view = store.lease_area("area").unwrap(); + store.try_set_item("area", "k", "v").unwrap(); + assert_eq!(resident(&store), 1); + drop(view); + // The last-view flush is ahead of this query in the database FIFO. + assert_eq!( + backend(&store) + .client + .query(|connection| Ok(connection.query_row( + "SELECT count(*) FROM entries", + [], + |r| r.get::<_, i64>(0) + )?)) + .unwrap(), + 1 + ); + assert_eq!(resident(&store), 0); + let _view = store.lease_area("area").unwrap(); + assert_eq!(store.get_item("area", "k").as_deref(), Some("v")); + owner.close().unwrap(); +} +#[test] +fn malformed_persisted_area_is_reported_instead_of_appearing_empty() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + store.try_set_item("area", "k", "v").unwrap(); + owner.flush().unwrap(); + assert_eq!(resident(&store), 0); + backend(&store) + .client + .query(|connection| { + connection.execute("UPDATE areas SET entry_count=99", [])?; + Ok(()) + }) + .unwrap(); + assert!(matches!( + store.try_get_item_utf16("area", &[107]), + Err(WebStorageMutationError::Persistence(_)) + )); + owner.close().unwrap(); +} +#[test] +fn quota_rejection_and_same_value_writes_do_not_publish_mutations() { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new().unwrap(); + let (store, owner) = open(root.path(), &runtime); + let _view = store.lease_area("area").unwrap(); + let events = store.subscribe_mutations(WebStorageAreaKind::Local); + store.try_set_item("area", "k", "v").unwrap(); + events.drain(); + store.try_set_item("area", "k", "v").unwrap(); + assert_eq!( + store.try_set_item("area", "k", &"x".repeat(WEB_STORAGE_QUOTA_BYTES + 1)), + Err(WebStorageMutationError::QuotaExceeded) + ); + assert_eq!(store.get_item("area", "k").as_deref(), Some("v")); + assert!(events.is_empty()); + owner.close().unwrap(); +} +#[test] +fn transient_keys_do_not_accumulate_delete_history_in_a_batch() { + let pin = Arc::new(Area { + state: Mutex::new(None), + views: AtomicUsize::new(0), + }); + let mut batch = Batch::default(); + for i in 0..10_000 { + let key = WebStorageString::from_utf8(&i.to_string()); + for change in [ + KeyChange::Put { + value: WebStorageString::from_utf8("v"), + existed: false, + }, + KeyChange::Delete, + ] { + let delta = AreaDelta { + clear: false, + changes: BTreeMap::from([(key.clone(), change)]), + usage: 0, + count: 0, + _pin: pin.clone(), + }; + batch.merge(Batch(BTreeMap::from([("area".into(), delta)]))); + } + } + assert!(batch.0["area"].changes.is_empty()); +} diff --git a/moli-web-storage/src/store.rs b/moli-web-storage/src/store.rs new file mode 100644 index 0000000000..0ef5971c5e --- /dev/null +++ b/moli-web-storage/src/store.rs @@ -0,0 +1,499 @@ +use super::*; + +pub(super) struct Edit { + pub(super) result: R, + pub(super) mutation: Option, +} +impl Edit { + fn unchanged(result: R) -> Self { + Self { + result, + mutation: None, + } + } +} + +pub fn new_shared_web_storage_store() -> SharedWebStorageStore { + Arc::new(WebStorageStore::default()) +} +pub fn deep_clone_shared_web_storage_store( + source: &SharedWebStorageStore, +) -> SharedWebStorageStore { + // This API implements browsing-context sessionStorage cloning. A local + // SQLite store is not a sessionStorage namespace and cannot be copied here. + let memory = match &source.backend { + WebStorageBackend::Memory(memory) => memory.lock().clone(), + WebStorageBackend::Json(json) => json.lock().memory.clone(), + WebStorageBackend::Sqlite(_) => { + panic!("sessionStorage cloning requires an in-memory namespace") + } + }; + Arc::new(WebStorageStore { + backend: WebStorageBackend::Memory(Mutex::new(memory)), + mutation_subscribers: Mutex::new(Vec::new()), + }) +} +pub fn new_shared_json_web_storage_store(path: impl AsRef) -> Result { + let backend = JsonWebStorageBackend::open(path.as_ref())?; + Ok(Arc::new(WebStorageStore { + backend: WebStorageBackend::Json(Mutex::new(backend)), + mutation_subscribers: Mutex::new(Vec::new()), + })) +} +impl Default for WebStorageStore { + fn default() -> Self { + Self { + backend: WebStorageBackend::Memory(Mutex::new(MemoryWebStorageBackend::default())), + mutation_subscribers: Mutex::new(Vec::new()), + } + } +} +impl fmt::Debug for WebStorageStore { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let kind = match &self.backend { + WebStorageBackend::Memory(_) => "memory", + WebStorageBackend::Json(_) => "legacy-json", + WebStorageBackend::Sqlite(_) => "sqlite", + }; + f.debug_struct("WebStorageStore") + .field("backend", &kind) + .finish_non_exhaustive() + } +} +impl WebStorageStore { + fn read( + &self, + origin: &str, + read: impl FnOnce(Option<&MemoryWebStorageArea>) -> R, + ) -> std::result::Result { + match &self.backend { + WebStorageBackend::Memory(memory) => Ok(read(memory.lock().origins.get(origin))), + WebStorageBackend::Json(json) => Ok(read(json.lock().memory.origins.get(origin))), + WebStorageBackend::Sqlite(sqlite) => sqlite.read(origin, read), + } + } + fn edit( + &self, + origin: &str, + edit: impl FnOnce( + &mut MemoryWebStorageArea, + ) -> std::result::Result, WebStorageMutationError>, + ) -> std::result::Result { + let publish = |edit: Edit| { + if let Some(mutation) = edit.mutation { + self.publish_mutation(mutation); + } + edit.result + }; + match &self.backend { + WebStorageBackend::Memory(memory) => { + let mut memory = memory.lock(); + let result = edit(memory.origins.entry(origin.to_owned()).or_default())?; + if memory + .origins + .get(origin) + .is_some_and(|area| area.values.is_empty()) + { + memory.origins.remove(origin); + } + Ok(publish(result)) + } + WebStorageBackend::Json(json) => { + let mut json = json.lock(); + let mut area = json.memory.origins.get(origin).cloned().unwrap_or_default(); + let result = edit(&mut area)?; + if result.mutation.is_some() { + json.update(&[origin], |memory| { + if area.values.is_empty() { + memory.origins.remove(origin); + } else { + memory.origins.insert(origin.to_owned(), area); + } + true + })?; + } + Ok(publish(result)) + } + WebStorageBackend::Sqlite(sqlite) => sqlite.edit(origin, edit, publish), + } + } + pub fn subscribe_mutations( + &self, + area_kind: WebStorageAreaKind, + ) -> WebStorageMutationSubscription { + let subscription = WebStorageMutationSubscription { + queue: Arc::new(Mutex::new(VecDeque::new())), + }; + self.add_mutation_subscription(area_kind, &subscription); + subscription + } + pub fn add_mutation_subscription( + &self, + area_kind: WebStorageAreaKind, + subscription: &WebStorageMutationSubscription, + ) { + let mut subscribers = self.mutation_subscribers.lock(); + subscribers.retain(|subscriber| { + subscriber.queue.upgrade().is_some_and(|queue| { + !(subscriber.area_kind == area_kind && Arc::ptr_eq(&queue, &subscription.queue)) + }) + }); + subscribers.push(WebStorageMutationSubscriber { + area_kind, + queue: Arc::downgrade(&subscription.queue), + }); + } + fn publish_mutation(&self, mutation: WebStorageMutation) { + self.mutation_subscribers.lock().retain(|subscriber| { + let Some(queue) = subscriber.queue.upgrade() else { + return false; + }; + queue.lock().push_back(WebStorageMutationRecord { + area_kind: subscriber.area_kind, + mutation: mutation.clone(), + }); + true + }); + } + pub fn try_get_item_utf16( + &self, + origin: &str, + key: &[u16], + ) -> std::result::Result>, WebStorageMutationError> { + self.read(origin, |area| { + area.and_then(|area| area.get_item_utf16(key)) + }) + } + pub fn get_item_utf16(&self, origin: &str, key: &[u16]) -> Option> { + self.try_get_item_utf16(origin, key).ok().flatten() + } + pub fn get_item(&self, origin: &str, key: &str) -> Option { + self.get_item_utf16(origin, &utf16_units(key)) + .map(|v| String::from_utf16_lossy(&v)) + } + pub fn try_set_change_utf16( + &self, + origin: &str, + key: &[u16], + value: &[u16], + ) -> std::result::Result, WebStorageMutationError> { + self.edit(origin, |area| { + let key = WebStorageString::from_utf16_units(key.to_vec()); + let value = WebStorageString::from_utf16_units(value.to_vec()); + let previous = area.values.get(&key); + if previous == Some(&value) { + return Ok(Edit::unchanged(None)); + } + let size = updated_storage_size(area.size, previous, value.usage_bytes()); + if size > WEB_STORAGE_QUOTA_BYTES { + return Err(WebStorageMutationError::QuotaExceeded); + } + let previous = area.values.insert(key.clone(), value.clone()); + area.size = size; + let mutation = match previous { + Some(old_value) => WebStorageMutation::ItemUpdated { + area_key: origin.to_owned(), + key, + old_value, + new_value: value, + }, + None => WebStorageMutation::ItemAdded { + area_key: origin.to_owned(), + key, + value, + }, + }; + Ok(Edit { + result: Some(mutation.clone()), + mutation: Some(mutation), + }) + }) + } + pub fn try_set_item_utf16( + &self, + origin: &str, + key: &[u16], + value: &[u16], + ) -> std::result::Result { + self.try_set_change_utf16(origin, key, value).map(|_| true) + } + pub fn try_set_item( + &self, + origin: &str, + key: &str, + value: &str, + ) -> std::result::Result { + self.try_set_item_utf16(origin, &utf16_units(key), &utf16_units(value)) + } + pub fn set_item(&self, origin: &str, key: &str, value: &str) -> bool { + self.try_set_item(origin, key, value).unwrap_or(false) + } + pub fn set_item_utf16(&self, origin: &str, key: &[u16], value: &[u16]) -> bool { + self.try_set_item_utf16(origin, key, value).unwrap_or(false) + } + pub fn try_remove_change_utf16( + &self, + origin: &str, + key: &[u16], + ) -> std::result::Result, WebStorageMutationError> { + self.edit(origin, |area| { + let key = WebStorageString::from_utf16_units(key.to_vec()); + let Some(old_value) = area.values.remove(&key) else { + return Ok(Edit::unchanged(None)); + }; + area.size -= old_value.usage_bytes(); + let mutation = WebStorageMutation::ItemRemoved { + area_key: origin.to_owned(), + key, + old_value, + }; + Ok(Edit { + result: Some(mutation.clone()), + mutation: Some(mutation), + }) + }) + } + pub fn try_remove_item_utf16( + &self, + origin: &str, + key: &[u16], + ) -> std::result::Result { + self.try_remove_change_utf16(origin, key) + .map(|change| change.is_some()) + } + pub fn try_remove_item( + &self, + origin: &str, + key: &str, + ) -> std::result::Result { + self.try_remove_item_utf16(origin, &utf16_units(key)) + } + pub fn remove_item(&self, origin: &str, key: &str) -> bool { + self.try_remove_item(origin, key).unwrap_or(false) + } + pub fn remove_item_utf16(&self, origin: &str, key: &[u16]) -> bool { + self.try_remove_item_utf16(origin, key).unwrap_or(false) + } + pub fn try_clear_change( + &self, + origin: &str, + ) -> std::result::Result, WebStorageMutationError> { + self.edit(origin, |area| { + if area.values.is_empty() { + return Ok(Edit::unchanged(None)); + } + area.values.clear(); + area.size = 0; + let mutation = WebStorageMutation::ItemsCleared { + area_key: origin.to_owned(), + }; + Ok(Edit { + result: Some(mutation.clone()), + mutation: Some(mutation), + }) + }) + } + pub fn try_clear(&self, origin: &str) -> std::result::Result { + // Legacy native callers distinguish removal of an existing empty area. + match &self.backend { + WebStorageBackend::Memory(memory) => { + let mut memory = memory.lock(); + let removed = memory.origins.remove(origin); + if removed.as_ref().is_some_and(|area| !area.values.is_empty()) { + self.publish_mutation(WebStorageMutation::ItemsCleared { + area_key: origin.to_owned(), + }); + } + Ok(removed.is_some()) + } + WebStorageBackend::Json(json) => { + let mut json = json.lock(); + let had_items = json.memory.len(origin) > 0; + let removed = json.update(&[origin], |memory| memory.clear_origin(origin))?; + if removed && had_items { + self.publish_mutation(WebStorageMutation::ItemsCleared { + area_key: origin.to_owned(), + }); + } + Ok(removed) + } + WebStorageBackend::Sqlite(_) => { + self.try_clear_change(origin).map(|change| change.is_some()) + } + } + } + pub fn clear(&self, origin: &str) { + let _ = self.try_clear(origin); + } + pub fn clear_origin(&self, origin: &str) { + self.clear(origin); + } + pub fn try_clear_origin( + &self, + origin: &str, + ) -> std::result::Result { + self.try_clear(origin) + } + pub fn try_key_utf16( + &self, + origin: &str, + index: usize, + ) -> std::result::Result>, WebStorageMutationError> { + self.read(origin, |area| area.and_then(|area| area.key_utf16(index))) + } + pub fn key_utf16(&self, origin: &str, index: usize) -> Option> { + self.try_key_utf16(origin, index).ok().flatten() + } + pub fn key(&self, origin: &str, index: usize) -> Option { + self.key_utf16(origin, index) + .map(|v| String::from_utf16_lossy(&v)) + } + pub fn try_items_utf16( + &self, + origin: &str, + ) -> std::result::Result, Vec)>, WebStorageMutationError> { + self.read(origin, |area| { + let Some(area) = area else { + return Vec::new(); + }; + let mut items = area.values.iter().collect::>(); + items.sort_by_key(|(key, _)| *key); + items + .into_iter() + .map(|(key, value)| (key.as_units().to_vec(), value.as_units().to_vec())) + .collect() + }) + } + pub fn try_len(&self, origin: &str) -> std::result::Result { + self.read(origin, |area| area.map_or(0, |a| a.values.len())) + } + pub fn len(&self, origin: &str) -> usize { + self.try_len(origin).unwrap_or(0) + } + pub fn try_contains_key_utf16( + &self, + origin: &str, + key: &[u16], + ) -> std::result::Result { + self.read(origin, |area| { + area.is_some_and(|area| area.contains_key_utf16(key)) + }) + } + pub fn contains_key_utf16(&self, origin: &str, key: &[u16]) -> bool { + self.try_contains_key_utf16(origin, key).unwrap_or(false) + } + pub fn contains_key(&self, origin: &str, key: &str) -> bool { + self.contains_key_utf16(origin, &utf16_units(key)) + } + pub fn try_sorted_keys_utf16( + &self, + origin: &str, + ) -> std::result::Result>, WebStorageMutationError> { + self.read(origin, |area| { + area.map_or_else(Vec::new, |area| area.sorted_keys_utf16()) + }) + } + pub fn sorted_keys_utf16(&self, origin: &str) -> Vec> { + self.try_sorted_keys_utf16(origin).unwrap_or_default() + } + pub fn sorted_keys(&self, origin: &str) -> Vec { + self.sorted_keys_utf16(origin) + .into_iter() + .map(|v| String::from_utf16_lossy(&v)) + .collect() + } + pub fn try_usage_bytes( + &self, + origin: &str, + ) -> std::result::Result { + self.read(origin, |area| area.map_or(0, |a| a.size)) + } + pub fn usage_bytes(&self, origin: &str) -> usize { + self.try_usage_bytes(origin).unwrap_or(0) + } + fn area_keys_for_origin( + &self, + origin: &str, + ) -> std::result::Result, WebStorageMutationError> { + match &self.backend { + WebStorageBackend::Memory(memory) => Ok(memory.lock().area_keys_for_origin(origin)), + WebStorageBackend::Json(json) => Ok(json.lock().memory.area_keys_for_origin(origin)), + WebStorageBackend::Sqlite(sqlite) => sqlite.area_keys_for_origin(origin), + } + } + pub fn try_usage_bytes_for_origin_areas( + &self, + origin: &str, + ) -> std::result::Result { + self.area_keys_for_origin(origin)? + .iter() + .try_fold(0usize, |sum, key| { + self.try_usage_bytes(key) + .map(|usage| sum.saturating_add(usage)) + }) + } + pub fn usage_bytes_for_origin_areas(&self, origin: &str) -> usize { + self.try_usage_bytes_for_origin_areas(origin).unwrap_or(0) + } + pub fn try_clear_origin_areas( + &self, + origin: &str, + ) -> std::result::Result { + let publish = |keys: Vec| { + let changed = !keys.is_empty(); + for area_key in keys { + self.publish_mutation(WebStorageMutation::ItemsCleared { area_key }); + } + changed + }; + match &self.backend { + WebStorageBackend::Memory(memory) => { + let mut memory = memory.lock(); + let keys = memory.area_keys_for_origin(origin); + let nonempty = keys + .iter() + .filter(|key| memory.len(key) > 0) + .cloned() + .collect::>(); + for key in &keys { + memory.origins.remove(key); + } + publish(nonempty); + Ok(!keys.is_empty()) + } + WebStorageBackend::Json(json) => { + let mut json = json.lock(); + let keys = json.memory.area_keys_for_origin(origin); + if keys.is_empty() { + return Ok(false); + } + let nonempty = keys + .iter() + .filter(|key| json.memory.len(key) > 0) + .cloned() + .collect::>(); + let affected = keys.iter().map(String::as_str).collect::>(); + json.update(&affected, |memory| { + for key in &keys { + memory.origins.remove(key); + } + true + })?; + publish(nonempty); + Ok(true) + } + WebStorageBackend::Sqlite(sqlite) => { + sqlite.clear_origin_areas(origin, |mutation| self.publish_mutation(mutation)) + } + } + } + pub fn clear_origin_areas(&self, origin: &str) { + let _ = self.try_clear_origin_areas(origin); + } + pub fn flush(&self) -> std::result::Result<(), WebStorageMutationError> { + match &self.backend { + WebStorageBackend::Sqlite(sqlite) => sqlite.client.flush().map_err(Into::into), + _ => Ok(()), + } + } +}