feat(storage): add asynchronous SQLite Web Storage areas and flush barriers

This commit is contained in:
ldm0
2026-11-01 04:03:38 +08:00
committed by Donough Liu
parent 5e4ce39c51
commit 03e09d240c
28 changed files with 2379 additions and 799 deletions
Generated
+5
View File
@@ -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",
]
+4 -8
View File
@@ -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(())
}
@@ -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
);
@@ -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())
);
@@ -756,8 +756,12 @@ impl BrowserContext {
serialized_origin: &str,
) -> Result<BrowserContextOriginStorageUsage, String> {
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}"))?;
+2 -2
View File
@@ -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"),
@@ -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"));
}
@@ -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::<Vec<_>>();
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::<Vec<_>>(),
Err(error) => return CommandOutputPlan::error(-32000, error.to_string()),
};
CommandOutputPlan::result(json!({ "entries": entries }))
}
DomStorageOperation::RemoveItem { key } => {
+23 -29
View File
@@ -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);
+1
View File
@@ -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"] }
@@ -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;
@@ -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());
}
@@ -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<R>(
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<R> {
let is_session = this_storage_kind(scope, storage);
let owner = storage_owner(scope, storage);
@@ -184,13 +184,64 @@ pub(super) fn with_storage_store<R>(
} 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<R, WebStorageMutationError>,
) -> Option<R> {
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");
}
@@ -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<Vec<u16>> {
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::<v8::String>::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::<v8::String>::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::<v8::String>::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::<v8::String>::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::<v8::String>::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::<v8::String>::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 {
@@ -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<u64, Entry>,
}
struct Entry {
store: SharedWebStorageStore,
key: String,
_lease: WebStorageAreaLease,
}
type Registry = Rc<RefCell<Leases>>;
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::<Registry>().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::<v8::BigInt>::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(&registry);
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(())
}
@@ -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<StorageUsageSnapshot, moli_web_storage::WebStorageMutationError> {
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>(
@@ -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::<Vec<_>>()
};
{
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);
}
@@ -2404,3 +2404,4 @@ mod xml_serializer_receivers;
mod hyperlink_audit_owner;
mod fetch_referrer;
mod sqlite_web_storage;
@@ -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");
}
+3 -2
View File
@@ -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
+2
View File
@@ -13,3 +13,5 @@ pub use connection::{
};
pub use runtime::{Runtime, RuntimeError, RuntimeHandle};
pub mod writeback;
+438
View File
@@ -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<str>);
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<rusqlite::Error> for WritebackError {
fn from(error: rusqlite::Error) -> Self {
Self::new(error)
}
}
type Result<T> = std::result::Result<T, WritebackError>;
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<usize>,
}
#[derive(Clone, Copy)]
pub enum FailureAdmission {
Retain,
Reject,
}
struct Pending<B> {
batch: B,
accepted: u64,
committed: u64,
operations: usize,
first_dirty: Option<tokio::time::Instant>,
failure: Option<WritebackError>,
closed: bool,
}
struct Shared<B> {
pending: Mutex<Pending<B>>,
changed: Notify,
}
trait Query: Send {
fn run(self: Box<Self>, connection: &mut Connection);
}
struct QueryJob<F, T> {
query: F,
reply: mpsc::Sender<Result<T>>,
}
impl<F, T> Query for QueryJob<F, T>
where
F: FnOnce(&mut Connection) -> Result<T> + Send,
T: Send,
{
fn run(self: Box<Self>, connection: &mut Connection) {
let _ = self.reply.send((self.query)(connection));
}
}
enum Message {
Query(Box<dyn Query>),
Flush {
through: u64,
reply: mpsc::Sender<Result<()>>,
},
Close,
}
pub struct WritebackClient<B> {
shared: Arc<Shared<B>>,
sender: async_mpsc::UnboundedSender<Message>,
runtime: RuntimeHandle,
}
impl<B> Clone for WritebackClient<B> {
fn clone(&self) -> Self {
Self {
shared: self.shared.clone(),
sender: self.sender.clone(),
runtime: self.runtime.clone(),
}
}
}
#[derive(Default)]
struct CloseReceipt {
receiver: Option<mpsc::Receiver<Result<()>>>,
result: Option<Result<()>>,
}
pub struct WritebackOwner<B> {
client: WritebackClient<B>,
finished: Mutex<CloseReceipt>,
}
impl<B: WriteBatch> WritebackOwner<B> {
/// 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<Self> {
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<B> {
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<B> Drop for WritebackOwner<B> {
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<B: WriteBatch> WritebackClient<B> {
/// Publish the native cache and accepted delta under the same admission
/// gate. The closure never enters V8 or performs SQL.
pub fn accept<T, E: From<WritebackError>>(
&self,
admission: FailureAdmission,
mutation: impl FnOnce() -> std::result::Result<(T, B), E>,
) -> std::result::Result<T, E> {
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<T: Send + 'static>(
&self,
query: impl FnOnce(&mut Connection) -> Result<T> + Send + 'static,
) -> Result<T> {
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<WritebackError> {
self.shared.pending.lock().failure.clone()
}
}
async fn commit<B: WriteBatch>(
connection: Connection,
shared: &Shared<B>,
runtime: &RuntimeHandle,
) -> Result<Connection> {
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<B: WriteBatch>(
mut connection: Connection,
mut receiver: async_mpsc::UnboundedReceiver<Message>,
shared: Arc<Shared<B>>,
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;
+254
View File
@@ -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<Option<(mpsc::Sender<()>, mpsc::Receiver<()>)>>,
fail: AtomicBool,
commits: AtomicUsize,
committed: Mutex<Option<oneshot::Sender<()>>>,
}
#[derive(Default)]
struct Batch {
values: BTreeMap<i64, i64>,
control: Option<Arc<Control>>,
}
impl Batch {
fn put(key: i64, value: i64, control: &Arc<Control>) -> 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<Batch> {
WritebackOwner::open(
path,
OpenMode::Create,
runtime.handle(),
long_policy(),
initialize,
)
.unwrap()
}
fn value(client: &WritebackClient<Batch>, 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<Batch>, tokio::task::JoinHandle<Result<()>>) {
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();
}
+5
View File
@@ -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
+57 -530
View File
@@ -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<String, Arc<RawValue>>;
@@ -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<u16>,
units: Arc<[u16]>,
}
#[derive(Debug, Default, Clone)]
@@ -69,16 +77,17 @@ struct JsonWebStorageBackend {
}
enum WebStorageBackend {
Memory(MemoryWebStorageBackend),
Json(JsonWebStorageBackend),
Memory(Mutex<MemoryWebStorageBackend>),
Json(Mutex<JsonWebStorageBackend>),
Sqlite(sqlite::SqliteBackend),
}
pub struct WebStorageStore {
backend: WebStorageBackend,
mutation_subscribers: Vec<WebStorageMutationSubscriber>,
mutation_subscribers: Mutex<Vec<WebStorageMutationSubscriber>>,
}
pub type SharedWebStorageStore = Arc<Mutex<WebStorageStore>>;
pub type SharedWebStorageStore = Arc<WebStorageStore>;
#[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<Path>) -> Result<SharedWebStorageStore> {
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::<Vec<_>>())
.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::<Vec<_>>())
.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<Vec<u16>>) -> 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<String> {
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<Vec<u16>> {
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<String> {
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<Vec<u16>> {
@@ -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<String> {
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<Vec<u16>> {
@@ -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<String> {
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<Vec<u16>> {
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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<String> {
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<Vec<u16>> {
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<String> {
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<Vec<u16>> {
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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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::<Vec<_>>();
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<String> {
self.origins.get(origin).and_then(|area| area.get_item(key))
}
fn get_item_utf16(&self, origin: &str, key: &[u16]) -> Option<Vec<u16>> {
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<String> {
self.origins.get(origin).and_then(|area| area.key(index))
}
fn key_utf16(&self, origin: &str, index: usize) -> Option<Vec<u16>> {
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<String> {
self.origins
.get(origin)
.map_or_else(Vec::new, MemoryWebStorageArea::sorted_keys)
}
fn sorted_keys_utf16(&self, origin: &str) -> Vec<Vec<u16>> {
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<String> {
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<String> {
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"));
+468
View File
@@ -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<Batch>,
}
struct Area {
state: Mutex<Option<MemoryWebStorageArea>>,
views: AtomicUsize,
}
pub struct WebStorageAreaLease {
area: Arc<Area>,
client: WritebackClient<Batch>,
}
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<String, Weak<Area>>,
insertions: usize,
}
pub(super) struct SqliteBackend {
pub(super) client: WritebackClient<Batch>,
cache: Mutex<Cache>,
pub(super) namespace: parking_lot::RwLock<()>,
}
#[derive(Default)]
pub(super) struct Batch(BTreeMap<String, AreaDelta>);
struct AreaDelta {
clear: bool,
changes: BTreeMap<WebStorageString, KeyChange>,
usage: usize,
count: usize,
_pin: Arc<Area>,
}
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<WritebackError> 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<u8> {
units.iter().flat_map(|u| u.to_le_bytes()).collect()
}
fn decode(bytes: Vec<u8>) -> std::result::Result<WebStorageString, WritebackError> {
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::<Vec<_>>(),
))
}
impl SqliteBackend {
fn area(&self, origin: &str) -> Arc<Area> {
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<MemoryWebStorageArea>,
) -> 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<u8>>(0)?, r.get::<_, Vec<u8>>(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<R>(
&self,
origin: &str,
read: impl FnOnce(Option<&MemoryWebStorageArea>) -> R,
) -> std::result::Result<R, WebStorageMutationError> {
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<R>(
&self,
origin: &str,
edit: impl FnOnce(
&mut MemoryWebStorageArea,
) -> std::result::Result<Edit<R>, WebStorageMutationError>,
publish: impl FnOnce(Edit<R>) -> R,
) -> std::result::Result<R, WebStorageMutationError> {
let _namespace = self.namespace.read();
self.edit_unfenced(origin, edit, publish)
}
pub(super) fn edit_unfenced<R>(
&self,
origin: &str,
edit: impl FnOnce(
&mut MemoryWebStorageArea,
) -> std::result::Result<Edit<R>, WebStorageMutationError>,
publish: impl FnOnce(Edit<R>) -> R,
) -> std::result::Result<R, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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::<Vec<_>>();
// 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<Vec<String>, 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::<std::result::Result<Vec<_>,_>>()?)
})?;
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<Option<WebStorageAreaLease>, 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<Option<String>> {
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;
+239
View File
@@ -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());
}
+499
View File
@@ -0,0 +1,499 @@
use super::*;
pub(super) struct Edit<R> {
pub(super) result: R,
pub(super) mutation: Option<WebStorageMutation>,
}
impl<R> Edit<R> {
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<Path>) -> Result<SharedWebStorageStore> {
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<R>(
&self,
origin: &str,
read: impl FnOnce(Option<&MemoryWebStorageArea>) -> R,
) -> std::result::Result<R, WebStorageMutationError> {
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<R>(
&self,
origin: &str,
edit: impl FnOnce(
&mut MemoryWebStorageArea,
) -> std::result::Result<Edit<R>, WebStorageMutationError>,
) -> std::result::Result<R, WebStorageMutationError> {
let publish = |edit: Edit<R>| {
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<Option<Vec<u16>>, 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<Vec<u16>> {
self.try_get_item_utf16(origin, key).ok().flatten()
}
pub fn get_item(&self, origin: &str, key: &str) -> Option<String> {
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<Option<WebStorageMutation>, 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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<Option<WebStorageMutation>, 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<bool, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<Option<WebStorageMutation>, 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<bool, WebStorageMutationError> {
// 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<bool, WebStorageMutationError> {
self.try_clear(origin)
}
pub fn try_key_utf16(
&self,
origin: &str,
index: usize,
) -> std::result::Result<Option<Vec<u16>>, WebStorageMutationError> {
self.read(origin, |area| area.and_then(|area| area.key_utf16(index)))
}
pub fn key_utf16(&self, origin: &str, index: usize) -> Option<Vec<u16>> {
self.try_key_utf16(origin, index).ok().flatten()
}
pub fn key(&self, origin: &str, index: usize) -> Option<String> {
self.key_utf16(origin, index)
.map(|v| String::from_utf16_lossy(&v))
}
pub fn try_items_utf16(
&self,
origin: &str,
) -> std::result::Result<Vec<(Vec<u16>, Vec<u16>)>, WebStorageMutationError> {
self.read(origin, |area| {
let Some(area) = area else {
return Vec::new();
};
let mut items = area.values.iter().collect::<Vec<_>>();
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<usize, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
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<Vec<Vec<u16>>, 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<Vec<u16>> {
self.try_sorted_keys_utf16(origin).unwrap_or_default()
}
pub fn sorted_keys(&self, origin: &str) -> Vec<String> {
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<usize, WebStorageMutationError> {
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<Vec<String>, 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<usize, WebStorageMutationError> {
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<bool, WebStorageMutationError> {
let publish = |keys: Vec<String>| {
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::<Vec<_>>();
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::<Vec<_>>();
let affected = keys.iter().map(String::as_str).collect::<Vec<_>>();
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(()),
}
}
}