Author: Danny Milosavljevic Subject: Fix workspace persistence races during restoration. Date: 2026-03-01 License: expat In 0.225.10, workspace persistence has three separate ordering problems. First, newly allocated workspace IDs could be used for item persistence before a parent row had been inserted into the workspaces table. That made editor persistence race with workspace creation and could fail with a foreign key constraint error. Second, workspace layout persistence and per-item persistence run on separate asynchronous paths. That allowed the workspace layout to be saved with item IDs before the background item-serialization worker had finished writing the corresponding per-item rows. On restore, deserializers that require persisted item state could then fail with "No entry in database for item_id ...", which could drop restored state and break restoration features that depend on that persisted item state, including edited-window restoration. Third, restore-side cleanup assumed that deserialized items had already been saved under their new runtime item IDs after being added to panes. In reality, adding a deserialized item only enqueues that serialization work. Cleanup could therefore delete the old persisted rows before the new rows existed, leaving the next reopen with a pane/item layout that still referenced state that had just been deleted. This change fixes all three issues: - eagerly create the parent workspaces row as soon as a new workspace ID is allocated, before any item serializer can write child rows - keep item persistence on the existing serialization queue, but add an ordering barrier before saving workspace layout so the layout is never published ahead of the item state it references - during restore, wait for the queued serialization of newly added deserialized items before cleaning up stale persisted item IDs, and skip that cleanup pass if the barrier cannot be established or awaited so stale rows are kept rather than deleting still-needed state See also . diff --git a/crates/workspace/src/persistence.rs b/crates/workspace/src/persistence.rs index e0ad046e8d..aeb4c3ba39 100644 --- a/crates/workspace/src/persistence.rs +++ b/crates/workspace/src/persistence.rs @@ -2124,6 +2124,13 @@ impl WorkspaceDb { } } + query! { + pub(crate) async fn ensure_workspace_row(workspace_id: WorkspaceId) -> Result<()> { + INSERT OR IGNORE INTO workspaces(workspace_id) + VALUES (?1) + } + } + query! { pub(crate) async fn set_window_open_status(workspace_id: WorkspaceId, bounds: SerializedWindowBounds, display: Uuid) -> Result<()> { UPDATE workspaces diff --git a/crates/workspace/src/workspace.rs b/crates/workspace/src/workspace.rs index 4141952cc7..3ea631ed6f 100644 --- a/crates/workspace/src/workspace.rs +++ b/crates/workspace/src/workspace.rs @@ -1175,6 +1175,11 @@ enum WorkspaceLocation { None, } +enum SerializableItemMessage { + Item(Box), + Barrier(oneshot::Sender<()>), +} + type PromptForNewPath = Box< dyn Fn( &mut Workspace, @@ -1253,7 +1258,7 @@ pub struct Workspace { on_prompt_for_open_path: Option, terminal_provider: Option>, debugger_provider: Option>, - serializable_items_tx: UnboundedSender>, + serializable_items_tx: UnboundedSender, _items_serializer: Task>, session_id: Option, scheduled_tasks: Vec>, @@ -1554,7 +1559,7 @@ impl Workspace { } let (serializable_items_tx, serializable_items_rx) = - mpsc::unbounded::>(); + mpsc::unbounded::(); let _items_serializer = cx.spawn_in(window, async move |this, cx| { Self::serialize_items(&this, serializable_items_rx, cx).await }); @@ -1759,6 +1764,7 @@ impl Workspace { } else { DB.next_id().await.unwrap_or_else(|_| Default::default()) }; + DB.ensure_workspace_row(workspace_id).await?; let toolchains = DB.toolchains(workspace_id).await?; @@ -6036,8 +6042,16 @@ impl Workspace { window_id: Some(window.window_handle().window_id().as_u64()), user_toolchains, }; + let item_serialization_barrier = self.item_serialization_barrier(); window.spawn(cx, async move |_| { + let Some(item_serialization_barrier) = item_serialization_barrier.log_err() + else { + return; + }; + if item_serialization_barrier.await.log_err().is_none() { + return; + } persistence::DB.save_workspace(serialized_workspace).await; }) } @@ -6108,31 +6122,49 @@ impl Workspace { async fn serialize_items( this: &WeakEntity, - items_rx: UnboundedReceiver>, + mut items_rx: UnboundedReceiver, cx: &mut AsyncWindowContext, ) -> Result<()> { const CHUNK_SIZE: usize = 200; - let mut serializable_items = items_rx.ready_chunks(CHUNK_SIZE); + while let Some(message) = items_rx.next().await { + let mut unique_items = HashMap::default(); + let mut barrier = None; + let mut channel_closed = false; - while let Some(items_received) = serializable_items.next().await { - let unique_items = - items_received - .into_iter() - .fold(HashMap::default(), |mut acc, item| { - acc.entry(item.item_id()).or_insert(item); - acc - }); + Self::push_serializable_item_message(message, &mut unique_items, &mut barrier); - // We use into_iter() here so that the references to the items are moved into - // the tasks and not kept alive while we're sleeping. - for (_, item) in unique_items.into_iter() { - if let Ok(Some(task)) = this.update_in(cx, |workspace, window, cx| { - item.serialize(workspace, false, window, cx) - }) { - cx.background_spawn(async move { task.await.log_err() }) - .detach(); + while barrier.is_none() && unique_items.len() < CHUNK_SIZE { + match items_rx.try_next() { + Ok(Some(message)) => { + Self::push_serializable_item_message( + message, + &mut unique_items, + &mut barrier, + ); + } + Ok(None) => { + channel_closed = true; + break; + } + Err(_) => break, + } + } + + let batch_succeeded = Self::serialize_item_batch(this, unique_items, cx) + .await + .log_err() + .is_some(); + + if let Some(barrier) = barrier { + if batch_succeeded { + let _ = barrier.send(()); } + continue; + } + + if channel_closed { + break; } cx.background_executor() @@ -6143,12 +6175,54 @@ impl Workspace { Ok(()) } + fn push_serializable_item_message( + message: SerializableItemMessage, + unique_items: &mut HashMap>, + barrier: &mut Option>, + ) { + match message { + SerializableItemMessage::Item(item) => { + unique_items.entry(item.item_id()).or_insert(item); + } + SerializableItemMessage::Barrier(sender) => { + *barrier = Some(sender); + } + } + } + + async fn serialize_item_batch( + this: &WeakEntity, + unique_items: HashMap>, + cx: &mut AsyncWindowContext, + ) -> Result<()> { + let serialize_tasks = match this.update_in(cx, move |workspace, window, cx| { + unique_items + .into_values() + .filter_map(|item| item.serialize(workspace, false, window, cx)) + .collect::>() + }) { + Ok(serialize_tasks) => serialize_tasks, + Err(_) => return Ok(()), + }; + + try_join_all(serialize_tasks).await?; + Ok(()) + } + + fn item_serialization_barrier(&self) -> Result> { + let (sender, receiver) = oneshot::channel(); + self.serializable_items_tx + .unbounded_send(SerializableItemMessage::Barrier(sender)) + .map_err(|err| anyhow!("failed to send item serialization barrier: {err}"))?; + Ok(receiver) + } + pub(crate) fn enqueue_item_serialization( &mut self, item: Box, ) -> Result<()> { self.serializable_items_tx - .unbounded_send(item) + .unbounded_send(SerializableItemMessage::Item(item)) .map_err(|err| anyhow!("failed to send serializable item over channel: {err}")) } @@ -6252,24 +6326,42 @@ impl Workspace { // after loading the items, we might have different items and in order to avoid // the database filling up, we delete items that haven't been loaded now. // - // The items that have been loaded, have been saved after they've been added to the workspace. - let clean_up_tasks = workspace.update_in(cx, |_, window, cx| { - item_ids_by_kind - .into_iter() - .map(|(item_kind, loaded_items)| { - SerializableItemRegistry::cleanup( - item_kind, - serialized_workspace.id, - loaded_items, - window, - cx, - ) - .log_err() - }) - .collect::>() - })?; + // Newly deserialized items only enqueue serialization when they are added to their panes. + // Wait for that queued work to finish before deleting the old persisted item ids. + let item_serialization_barrier = workspace + .update_in(cx, |workspace, _, _| { + workspace.item_serialization_barrier().log_err() + }) + .log_err() + .flatten(); + + let item_serialization_flushed = if let Some(item_serialization_barrier) = + item_serialization_barrier + { + item_serialization_barrier.await.log_err().is_some() + } else { + false + }; - futures::future::join_all(clean_up_tasks).await; + if item_serialization_flushed { + let clean_up_tasks = workspace.update_in(cx, |_, window, cx| { + item_ids_by_kind + .into_iter() + .map(|(item_kind, loaded_items)| { + SerializableItemRegistry::cleanup( + item_kind, + serialized_workspace.id, + loaded_items, + window, + cx, + ) + .log_err() + }) + .collect::>() + })?; + + futures::future::join_all(clean_up_tasks).await; + } workspace .update_in(cx, |workspace, window, cx| { @@ -9001,6 +9093,7 @@ fn deserialize_remote_project( } else { persistence::DB.next_id().await? }; + persistence::DB.ensure_workspace_row(workspace_id).await?; Ok((workspace_id, serialized_workspace)) })