diff --git a/docs/reference/lore-cli-commands.md b/docs/reference/lore-cli-commands.md index d0e74551..6eb27a91 100644 --- a/docs/reference/lore-cli-commands.md +++ b/docs/reference/lore-cli-commands.md @@ -25,6 +25,7 @@ This page is generated from `lore --markdown-help` (CLI `0.8.2-nightly+31`). Eve * [`lore repository verify`↴](#lore-repository-verify) * [`lore repository verify state`↴](#lore-repository-verify-state) * [`lore repository verify fragment`↴](#lore-repository-verify-fragment) +* [`lore repository push-content`↴](#lore-repository-push-content) * [`lore repository dump`↴](#lore-repository-dump) * [`lore repository gc`↴](#lore-repository-gc) * [`lore repository store`↴](#lore-repository-store) @@ -219,8 +220,9 @@ This page is generated from `lore --markdown-help` (CLI `0.8.2-nightly+31`). Eve * `--compress-limit ` — Set maximum number of parallel compress operations * `--search-limit ` — Set maximum number of revisions to search when matching or finding revisions * `--search-nearest` — Set to search for nearest match when matching revisions -* `--gc` — Set to run automatic garbage collection on local store in background +* `--no-gc` — Prevent automatic incremental garbage collection for this command; it otherwise runs in the background on writes. `lore repository gc` always runs a full pass regardless * `--sync-data` — Force sync data to storage media during flush +* `--cache` — Cache fragment payloads fetched from remote in the local store * `--non-interactive` — Disable interactive prompts (e.g., per-link commit messages) @@ -240,6 +242,7 @@ Repository commands * `clone` — Clone a remote repository into the given path * `delete` — Delete a repository * `verify` — Verify repository state consistency +* `push-content` — Push locally-stored content that is not yet durable on the remote * `dump` — Dump repository state information * `gc` — Run a full garbage collection pass on the local repository store * `store` — Access the repository data store @@ -417,6 +420,22 @@ Verify a specific fragment in the local store +## `lore repository push-content` + +Push locally-stored content that is not yet durable on the remote + +**Usage:** `lore repository push-content [OPTIONS] [ADDRESSES]...` + +###### **Arguments:** + +* `` — Fragment addresses to push, as `hash` or `hash-context` (the format printed in `Address not found` errors) + +###### **Options:** + +* `--scan` — Scan the local store for all content of this repository that was never confirmed durable on the remote (e.g. after a commit whose upload timed out) and push all of it + + + ## `lore repository dump` Dump repository state information diff --git a/lore-client/src/cli/commands/repository.rs b/lore-client/src/cli/commands/repository.rs index e725e7ab..3aad6060 100644 --- a/lore-client/src/cli/commands/repository.rs +++ b/lore-client/src/cli/commands/repository.rs @@ -316,6 +316,19 @@ pub struct RepositoryVerifyFragmentArgs { heal: bool, } +#[derive(Args)] +pub struct RepositoryPushContentArgs { + /// Fragment addresses to push, as `hash` or `hash-context` (the format + /// printed in `Address not found` errors) + addresses: Vec, + + /// Scan the local store for all content of this repository that was never + /// confirmed durable on the remote (e.g. after a commit whose upload + /// timed out) and push all of it + #[clap(long, action)] + scan: bool, +} + #[derive(Subcommand)] pub enum RepositoryCommands { /// Show current repository status. @@ -345,6 +358,10 @@ pub enum RepositoryCommands { /// Verify repository state consistency Verify(RepositoryVerifyArgs), + /// Push locally-stored content that is not yet durable on the remote + #[command(name = "push-content")] + PushContent(RepositoryPushContentArgs), + /// Dump repository state information Dump(RepositoryDumpArgs), @@ -1142,6 +1159,64 @@ pub fn handle_repository_verify_state( runtime().block_on(repository::verify_state(globals, verify_args, callback)) as u8 } +pub fn handle_repository_push_content( + globals: LoreGlobalArgs, + args: &RepositoryPushContentArgs, +) -> u8 { + if !args.scan && args.addresses.is_empty() { + println!( + "{}No addresses given{}: pass one or more `hash-context` addresses or use --scan", + CommonStyles::FAILURE, + anstyle::Reset + ); + return 1; + } + + let push_args = lore::interface::LoreRepositoryPushContentArgs { + addresses: LoreArray::from_vec( + args.addresses + .iter() + .map(|address| LoreString::from(address.as_str())) + .collect(), + ), + scan: args.scan.into(), + }; + + let _spinner = ProgressBar::new_spinner("Pushing content to remote..."); + + let callback = output_formatter().unwrap_or(Some( + (Box::new(move |event: &LoreEvent| match event { + LoreEvent::StorageUploadItemComplete(data) => { + if data.error_code != lore_revision::event::LoreErrorCode::None { + println!( + "{}Failed{} to push fragment ({:?})", + CommonStyles::FAILURE, + anstyle::Reset, + data.error_code + ); + } else if data.already_durable != 0 { + println!("Fragment {} already durable", data.address); + } else { + println!( + "{}Pushed{} fragment {}", + CommonStyles::SUCCESS, + anstyle::Reset, + data.address + ); + } + } + LoreEvent::Complete(_) => {} + LoreEvent::Maintenance(data) => { + util::handle_maintenance_event(data); + } + _ => (), + }) as EventCallbackFn) + .with_defaults(), + )); + + runtime().block_on(repository::push_content(globals, push_args, callback)) as u8 +} + pub fn handle_repository_verify_fragment( globals: LoreGlobalArgs, args: &RepositoryVerifyFragmentArgs, @@ -1669,6 +1744,7 @@ pub fn handle_repository_commands(cmd: &RepositoryCommands, globals: LoreGlobalA RepositoryCommands::Delete(args) => handle_repository_delete(globals, args), RepositoryCommands::Clone(args) => handle_repository_clone(globals, args), RepositoryCommands::Verify(args) => handle_repository_verify(globals, args), + RepositoryCommands::PushContent(args) => handle_repository_push_content(globals, args), RepositoryCommands::Dump(args) => handle_repository_dump( globals, args.revision.as_deref().unwrap_or(""), diff --git a/lore-revision/src/repository.rs b/lore-revision/src/repository.rs index 84c27a66..6f120079 100644 --- a/lore-revision/src/repository.rs +++ b/lore-revision/src/repository.rs @@ -11,6 +11,7 @@ pub mod delete; pub mod dump; pub mod info; pub mod list; +pub mod push_content; pub mod status; pub mod store; pub mod verify; diff --git a/lore-revision/src/repository/push_content.rs b/lore-revision/src/repository/push_content.rs new file mode 100644 index 00000000..bcf1312e --- /dev/null +++ b/lore-revision/src/repository/push_content.rs @@ -0,0 +1,219 @@ +// SPDX-FileCopyrightText: 2026 Epic Games, Inc. +// SPDX-License-Identifier: MIT +//! `repository push-content` — push locally-stored, not-yet-durable content to +//! the remote store. +//! +//! A commit whose remote upload fails (e.g. a timeout) keeps the payload in +//! the local store with `PayloadStoredDurable` clear. Other clients then fail +//! to sync the published revision with `Address not found`. This command +//! re-uploads such payloads from the machine that still holds them: +//! +//! - `scan == true`: enumerate every non-durable entry for this repository's +//! partition in the local store and upload each one. +//! - otherwise: upload the explicitly given addresses (`hash` or +//! `hash-context` strings, matching the format printed in errors). +//! +//! Each item emits a [`LoreStorageUploadItemCompleteEventData`] event; the +//! call fails if any item fails. + +use std::str::FromStr; +use std::sync::Arc; + +use lore_base::lore_spawn; +use lore_error_set::prelude::*; +use lore_storage::concurrency::acquire_fragment_memory_permit; +use lore_storage::options::ReadOptions; +use lore_storage::read::load_fragment; +use lore_storage::store_types::StoreMatch; +use lore_storage::write::store_fragment; +use lore_transport::StorageSession; +use tokio::task::JoinSet; + +use super::RepositoryContext; +use super::RepositoryError; +use crate::event; +use crate::fragment::FragmentFlags; +use crate::interface::LoreString; +use crate::lore::Address; +use crate::lore::Context; +use crate::lore::Hash; +use crate::lore::Partition; +use crate::lore::execution_context; +use crate::lore_info; +use crate::store::event::LoreStorageUploadItemCompleteEventData; + +/// Arguments for `push_content` from the library layer. +pub struct PushContentArgs { + /// Explicit addresses to push, as `hash` or `hash-context` strings. + pub addresses: Vec, + /// Scan the local store for all non-durable entries instead. + pub scan: bool, +} + +pub async fn push_content( + repository: Arc, + args: PushContentArgs, +) -> Result<(), RepositoryError> { + let items: Vec<(Partition, Address)> = if args.scan { + scan_non_durable(&repository).await? + } else { + let mut items = Vec::with_capacity(args.addresses.len()); + for address in &args.addresses { + items.push((repository.id, parse_address(address.as_str())?)); + } + items + }; + + if items.is_empty() { + lore_info!("No non-durable content found; nothing to push"); + return Ok(()); + } + lore_info!("Pushing {} content fragment(s) to remote", items.len()); + + let remote = repository + .remote() + .await + .forward::("Push content failed: no remote connection")?; + let correlation_id = execution_context().globals().correlation_id.to_string(); + let session = remote + .session(repository.id, &correlation_id) + .await + .forward::("Push content failed: failed to connect")?; + + let mut tasks: JoinSet> = JoinSet::new(); + for (index, (partition, address)) in items.into_iter().enumerate() { + let repository = repository.clone(); + let session = session.clone(); + lore_spawn!(tasks, async move { + push_item(repository, index as u64, partition, address, session).await + }); + } + + let mut result: Result<(), RepositoryError> = Ok(()); + while let Some(task_result) = tasks.join_next().await { + let item_result = task_result + .unwrap_or_else(|err| Err(RepositoryError::internal(format!("Task failure: {err}")))); + if result.is_ok() && item_result.is_err() { + result = item_result; + } + } + result +} + +/// Enumerate every non-durable entry for this repository's partition in the +/// local immutable store. +async fn scan_non_durable( + repository: &Arc, +) -> Result, RepositoryError> { + let store = repository.immutable_store(); + if !store.is_local() { + return Err(RepositoryError::internal( + "Push content failed: scan requires a local store", + )); + } + // Same downcast as `verify_fragment_local`: the local store is the + // concrete `LocalImmutableStore` behind the trait object. + let concrete_store: Arc = (store + as Arc) + .downcast::() + .map_err(|_not_local| { + RepositoryError::internal("Push content failed: store is not a local ImmutableStore") + })?; + Ok(concrete_store + .non_durable_entries(Some(repository.id)) + .await) +} + +/// Parse a `hash` or `hash-context` address string (the format printed in +/// `Address not found` errors). +fn parse_address(text: &str) -> Result { + let (hash_text, context_text) = match text.split_once('-') { + Some((hash, context)) => (hash, Some(context)), + None => (text, None), + }; + let hash = Hash::from_str(hash_text).internal("Invalid address")?; + let context = match context_text { + Some(context) => Context::from_str(context).internal("Invalid address")?, + None => Context::default(), + }; + Ok(Address { hash, context }) +} + +async fn push_item( + repository: Arc, + id: u64, + partition: Partition, + address: Address, + session: Arc, +) -> Result<(), RepositoryError> { + let store = repository.immutable_store(); + + // Skip entries that are already durable (e.g. healed since the scan). + let query = store + .clone() + .query(partition, address, StoreMatch::MatchFull) + .await; + if let Ok(qr) = &query + && qr.match_made == StoreMatch::MatchFull + && qr.fragment.flags & FragmentFlags::PayloadStoredDurable != 0 + { + emit_complete(id, address, 1, event::LoreErrorCode::None); + return Ok(()); + } + + // `no_remote()` is load-bearing: we must not pull from a third party to + // satisfy a missing-local-payload push. + let load = load_fragment( + store.clone(), + partition, + address, + ReadOptions::default().no_remote(), + None, + ) + .await; + let (fragment, payload) = match load { + Ok(pair) => pair, + Err(err) => { + emit_complete(id, address, 0, event::LoreErrorCode::AddressNotFound); + return Err(RepositoryError::internal(format!( + "Push content failed: No local payload for {address}: {err}" + ))); + } + }; + + let permit = acquire_fragment_memory_permit(payload.len()).await; + match store_fragment( + store, + partition, + address, + fragment, + payload, + true, + Some(session), + None, + permit, + ) + .await + { + Ok(_) => { + emit_complete(id, address, 0, event::LoreErrorCode::None); + Ok(()) + } + Err(err) => { + emit_complete(id, address, 0, event::LoreErrorCode::Internal); + Err(RepositoryError::internal(format!( + "Push content failed: Failed to upload {address}: {err}" + ))) + } + } +} + +fn emit_complete(id: u64, address: Address, already_durable: u8, error_code: event::LoreErrorCode) { + event::LoreEvent::StorageUploadItemComplete(LoreStorageUploadItemCompleteEventData { + id, + address, + already_durable, + error_code, + }) + .send(); +} diff --git a/lore-revision/src/repository/verify.rs b/lore-revision/src/repository/verify.rs index 24f94ed8..15896f2b 100644 --- a/lore-revision/src/repository/verify.rs +++ b/lore-revision/src/repository/verify.rs @@ -4,6 +4,7 @@ use std::future::Future; use std::pin::Pin; use std::str::FromStr; use std::sync::Arc; +use std::sync::Mutex; use lore_base::lore_spawn; use lore_error_set::prelude::*; @@ -15,6 +16,7 @@ use tokio::task::JoinSet; use super::RepositoryContext; use super::RepositoryError; use crate::event; +use crate::fragment::FragmentFlags; use crate::hash; use crate::interface::LoreArray; use crate::interface::LoreString; @@ -134,6 +136,16 @@ pub struct VerifyFragmentArgs { pub heal: bool, } +/// A file whose content could not be confirmed durable from the local store +/// alone: the local entry is missing or lacks `PayloadStoredDurable`. Resolved +/// against the remote store after the node walk. +struct DurabilitySuspect { + address: Address, + path: String, +} + +type DurabilitySuspects = Arc>>; + pub async fn verify( repository: Arc, path: Option, @@ -177,7 +189,20 @@ pub async fn verify( return Err(RepositoryError::internal("Invalid repository path")); } - verify_node(repository.clone(), state.clone(), is_staged, node_id).await?; + let suspects: DurabilitySuspects = Arc::new(Mutex::new(Vec::new())); + verify_node( + repository.clone(), + state.clone(), + is_staged, + node_id, + suspects.clone(), + ) + .await?; + + // Resolve content whose durability could not be confirmed locally against + // the remote store. Catches revisions published with content the remote + // never received (e.g. a commit whose upload timed out). + verify_content_durability(repository.clone(), suspects, heal).await?; // Check case uniqueness of each node state::verify_node_name_case(repository.clone(), state.clone(), node_id) @@ -226,10 +251,11 @@ async fn verify_node( state: Arc, is_staged: bool, node_id: NodeID, + suspects: DurabilitySuspects, ) -> Result<(), RepositoryError> { let mut tasks = JoinSet::new(); let mut result = verify_node_single( - repository, state, is_staged, node_id, None, &mut tasks, None, + repository, state, is_staged, node_id, None, &mut tasks, None, suspects, ) .await; @@ -261,6 +287,7 @@ async fn verify_node_single( expected_parent: Option, tasks: &mut JoinSet>, cycle: Option<&mut SiblingCycleGuard>, + suspects: DurabilitySuspects, ) -> Result { let block_index = NodeBlock::index(node_id); let node_index = Node::index(node_id); @@ -335,7 +362,34 @@ async fn verify_node_single( } if node.is_file() { - // ... + // Confirm the file's content is durable from the local store's + // point of view; anything unconfirmed (missing entry, or an entry + // whose upload never succeeded) is checked against the remote store + // after the walk. + if node.size > 0 && !node.address.hash.is_zero() { + let query = repository + .immutable_store() + .query(repository.id, node.address, StoreMatch::MatchFull) + .await; + let durable_locally = matches!( + &query, + Ok(qr) if qr.match_made == StoreMatch::MatchFull + && qr.fragment.flags & FragmentFlags::PayloadStoredDurable != 0 + ); + if !durable_locally { + let path = state + .node_path(repository.clone(), node_id) + .await + .unwrap_or_default(); + suspects + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(DurabilitySuspect { + address: node.address, + path, + }); + } + } } else if node.is_link() { // TODO(vri): UCS-19231 - Links: Handle link nodes in repository verify } else if node.is_directory() @@ -344,9 +398,17 @@ async fn verify_node_single( lore_spawn!(tasks, { let repository = repository.clone(); let state = state.clone(); + let suspects = suspects.clone(); async move { - verify_node_recurse(repository.clone(), state.clone(), is_staged, child, node_id) - .await + verify_node_recurse( + repository.clone(), + state.clone(), + is_staged, + child, + node_id, + suspects, + ) + .await } }); } @@ -360,6 +422,7 @@ async fn verify_node_and_siblings( is_staged: bool, mut node_id: NodeID, expected_parent: NodeID, + suspects: DurabilitySuspects, ) -> Result { lore_trace!("Verify node {node_id} and siblings"); @@ -375,6 +438,7 @@ async fn verify_node_and_siblings( Some(expected_parent), &mut tasks, Some(&mut cycle), + suspects.clone(), ) .await; @@ -406,6 +470,7 @@ fn verify_node_recurse( is_staged: bool, node: NodeID, expected_parent: NodeID, + suspects: DurabilitySuspects, ) -> Pin> + Send>> { Box::pin(verify_node_and_siblings( repository, @@ -413,9 +478,78 @@ fn verify_node_recurse( is_staged, node, expected_parent, + suspects, )) } +/// Check every durability suspect against the remote store: a file's content +/// that is not confirmed durable locally must exist on the remote, otherwise +/// the published revision references content other clients cannot fetch. +/// +/// Skipped when no remote is available (offline/local repository). +async fn verify_content_durability( + repository: Arc, + suspects: DurabilitySuspects, + heal: bool, +) -> Result<(), RepositoryError> { + let suspects = std::mem::take(&mut *suspects.lock().unwrap_or_else(|e| e.into_inner())); + if suspects.is_empty() { + return Ok(()); + } + + let Ok(remote) = repository.remote().await else { + lore_info!( + "Skipping content durability check for {} file(s): no remote available", + suspects.len() + ); + return Ok(()); + }; + let correlation_id = execution_context().globals().correlation_id.to_string(); + let storage = remote + .session(repository.id, &correlation_id) + .await + .forward::("Repository verification failed: Failed to connect")?; + + lore_debug!( + "Checking {} file content address(es) against the remote store", + suspects.len() + ); + + let mut missing = Vec::new(); + for suspect in &suspects { + match storage.verify(&suspect.address, heal).await { + Ok(_) => {} + Err(err) => { + let error_msg = match &err { + ProtocolError::NotFound(_) => "Fragment not found".to_string(), + _ => format!("Verification failed: {err}"), + }; + event::LoreEvent::RepositoryVerifyFragmentRemote( + LoreRepositoryVerifyFragmentRemoteEventData { + address_hash: suspect.address.hash, + address_context: suspect.address.context, + corrupted: 0, + healed: 0, + error: LoreString::from(format!("{error_msg}: {}", suspect.path)), + }, + ) + .send(); + missing.push(suspect); + } + } + } + + if let Some(first) = missing.first() { + return Err(RepositoryError::internal(format!( + "Repository verification failed: {} file(s) reference content missing from the remote store, e.g. {} ({}); run `repository push-content --scan` on the machine that committed them", + missing.len(), + first.path, + first.address, + ))); + } + Ok(()) +} + pub async fn verify_fragment( repository: Arc, args: VerifyFragmentArgs, diff --git a/lore-server/src/server.rs b/lore-server/src/server.rs index 989fb956..26372af7 100644 --- a/lore-server/src/server.rs +++ b/lore-server/src/server.rs @@ -1065,6 +1065,11 @@ async fn create_local_store( implicit_durable_stored: true, /* Server mode, consider all fragments as durably stored */ flush_background, flush_delay_seconds: settings.flush_delay_seconds as u64, + /* Server mode: this store is the authoritative persistence layer for its content, + so a `PayloadStoredDurable` claim must be backed by an actual fsync. Runs a + background heartbeat that fsyncs on this interval; durable writes wait for it + to confirm (bounded worst case ~2 ticks) rather than fsync-per-write. */ + durable_flush_tick_ms: 15, target_capacity_percentage: settings.target_capacity_percentage.unwrap_or(default_settings.target_capacity_percentage), target_size_percentage: settings.target_size_percentage.unwrap_or(default_settings.target_size_percentage), compaction_parallel_groups: settings.compaction_parallel_groups.unwrap_or(default_settings.compaction_parallel_groups), diff --git a/lore-storage/src/local/immutable_store.rs b/lore-storage/src/local/immutable_store.rs index e4438cc2..0f1336d0 100644 --- a/lore-storage/src/local/immutable_store.rs +++ b/lore-storage/src/local/immutable_store.rs @@ -42,6 +42,7 @@ use opentelemetry::metrics::Gauge; use opentelemetry::metrics::Histogram; use smallvec::SmallVec; use tokio::sync::Mutex; +use tokio::sync::Notify; use tokio::sync::OwnedRwLockReadGuard; use tokio::sync::RwLock; use tokio::sync::Semaphore; @@ -192,6 +193,71 @@ impl ImmutableStoreBucket { } } +/// Lets writers that just stored a payload wait for a durable (fsynced) copy, backed by +/// a continuously-running background ticker (see `durable_flush_loop`) rather than a +/// per-write sleep-then-fsync task. The ticker fsyncs every group's packstore on a fixed +/// interval and bumps `generation` each time, regardless of whether anything was dirty. +/// +/// # Why `wait_past` waits for +2, not +1 +/// +/// The ticker loop is single-threaded and its ticks never overlap (one +/// `tokio::time::interval` loop). A writer sets its packfile's dirty bit, then snapshots +/// `generation` via `current_generation()`. If a tick was already in flight at that +/// moment (started before the write, so its per-packfile dirty check ran too early to see +/// it), that tick can still complete and bump `generation` once — without having actually +/// synced the writer's bytes. Waiting for only +1 past the snapshot could observe exactly +/// that non-covering bump and return early, recreating the false-durability bug this type +/// exists to prevent. +/// +/// Because ticks are sequential, at most one tick can be "already in flight and +/// non-covering" at snapshot time. The *next* tick after that one is guaranteed to start +/// after the snapshot was taken, so it will observe the dirty bit (already set before the +/// snapshot) and sync it. Waiting for the generation to advance by 2 past the snapshot +/// therefore always waits past a tick that is guaranteed to cover the write, regardless of +/// how the write and the ticker happen to interleave. +pub struct FlushBarrier { + /// Bumped once after each completed tick (whether or not it found anything dirty). A + /// waiter that captured the generation right after its write knows its data is durable + /// once this counter has advanced by 2 past that value (see type-level doc for why 2). + generation: AtomicU64, + done: Notify, +} + +impl Default for FlushBarrier { + fn default() -> Self { + Self { + generation: AtomicU64::new(0), + done: Notify::new(), + } + } +} + +impl FlushBarrier { + /// Snapshot the current generation right after writing dirty bytes; pass the result + /// to `wait_past` to be woken once a tick guaranteed to have synced them completes. + pub fn current_generation(&self) -> u64 { + self.generation.load(atomic::Ordering::Acquire) + } + + /// Wait until a background tick guaranteed to cover a write snapshotted at `since` has + /// completed. See the type-level doc for why this waits for +2 rather than +1. + pub async fn wait_past(&self, since: u64) { + loop { + let notified = self.done.notified(); + if self.generation.load(atomic::Ordering::Acquire) >= since + 2 { + return; + } + notified.await; + } + } + + /// Mark a tick complete, bumping the generation and waking all waiters. + pub fn complete(&self) { + self.generation.fetch_add(1, atomic::Ordering::AcqRel); + self.done.notify_waiters(); + } +} + pub struct ImmutableStoreGroup { /// Per-slot lazily-initialized bucket. Empty `OnceLock` at construction; first /// `bucket()` call materializes the `Arc>`. Use @@ -222,6 +288,10 @@ pub struct ImmutableStoreGroup { pub committed_level: std::sync::atomic::AtomicUsize, pub packstore: crate::PackStore, pub flush: Mutex>, + /// Wakes writers waiting for a durable (fsynced) copy of freshly-written payload + /// bytes, driven by the store-wide `durable_flush_loop` background ticker rather than + /// a per-write task. See `FlushBarrier`'s docs for the wait invariant. + pub flush_barrier: FlushBarrier, } impl ImmutableStoreGroup { @@ -284,6 +354,16 @@ pub struct ImmutableStoreSettings { pub flush_background: bool, /// Flush delay in seconds pub flush_delay_seconds: u64, + /// Interval (milliseconds) of the background `durable_flush_loop` heartbeat that + /// fsyncs every group's packstore. When a `store()` call writes payload bytes under a + /// `PayloadStoredDurable` claim, it waits for this ticker to confirm (with a bounded + /// worst case of ~2 ticks — see `FlushBarrier`'s docs) before returning, so the claim + /// is never made ahead of an actual `fsync`. `0` disables the ticker entirely (no + /// background task is spawned, and durable writes return without waiting, matching + /// pre-fix behavior) — appropriate for clients that only cache a remote-durable copy. + /// A store acting as the authoritative persistence layer for its content (e.g. + /// `lore-server`) must set this to a small positive value. + pub durable_flush_tick_ms: u64, /// Eviction target capacity as a percentage of the max capacity (0-100) pub target_capacity_percentage: usize, /// Compaction target size as a percentage of the max size (0-100) @@ -311,6 +391,7 @@ impl Default for ImmutableStoreSettings { implicit_durable_stored: false, flush_background: false, flush_delay_seconds: DEFAULT_FLUSH_DELAY_SECONDS, + durable_flush_tick_ms: 0, target_capacity_percentage: 70, target_size_percentage: 70, compaction_parallel_groups: 8, @@ -890,6 +971,28 @@ impl ImmutableStoreGroup { } } +/// Background heartbeat that fsyncs every group's packstore on a fixed interval and wakes +/// `FlushBarrier` waiters — see `FlushBarrier`'s docs for why writers wait for 2 completed +/// ticks, not 1. Mirrors the weak-ref self-cancelling pattern used by +/// `lore-storage::maintenance::spawn_gc`'s evictor/compactor loops: the store has no +/// explicit shutdown hook, so this task simply exits once the store's `Arc` is dropped. +async fn durable_flush_loop(weak_ref: Weak, tick: Duration) { + let mut ticker = tokio::time::interval(tick); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + ticker.tick().await; + let Some(store) = weak_ref.upgrade() else { + break; + }; + for group in &store.group { + // Cheap when clean: flush_all only does I/O for packfiles whose dirty flag is + // still set, via a lock-free compare_exchange per packfile. + group.packstore.flush_all(true).await; + group.flush_barrier.complete(); + } + } +} + impl LocalImmutableStore { /// Set the automatic-GC caps (from create options) on this store's load-driven GC /// counters. Caps of 0 leave the corresponding trigger disabled — which is how @@ -1086,6 +1189,7 @@ impl LocalImmutableStore { Some(store.gc_counters.clone()), ), flush: Mutex::new(JoinSet::new()), + flush_barrier: FlushBarrier::default(), })); } @@ -1094,6 +1198,14 @@ impl LocalImmutableStore { // the weak self-ref the load hooks need to fire a pass. let dyn_store: Arc = store.clone(); store.gc_counters.set_store(&dyn_store); + + if store.settings.durable_flush_tick_ms > 0 { + let weak = Arc::downgrade(&store); + let tick = Duration::from_millis(store.settings.durable_flush_tick_ms); + drop(lore_base::lore_spawn!(async move { + durable_flush_loop(weak, tick).await; + })); + } if let Some(path) = immutable_path.as_deref() { let mut old_packpath = path.clone(); old_packpath.push("pack"); @@ -1525,6 +1637,14 @@ impl LocalImmutableStore { )?; pack_file = packref.id; pack_offset = packref.offset; + + // These bytes are not yet fsynced; see `durable_flush_tick_ms`'s doc. + if self.settings.durable_flush_tick_ms > 0 + && (fragment_flags & FragmentFlags::PayloadStoredDurable) != 0 + { + let since = group.flush_barrier.current_generation(); + group.flush_barrier.wait_past(since).await; + } } } else { if !self.settings.allow_partial_fragment { @@ -3885,6 +4005,42 @@ impl crate::immutable_store::ImmutableStore for LocalImmutableStore { } impl LocalImmutableStore { + /// Enumerate every entry whose payload is stored locally but has not been + /// confirmed durable on the remote — i.e. `PayloadStoredLocal` set, + /// `PayloadStoredDurable` clear, and not obliterated. Optionally filtered by + /// partition. + /// + /// Deserializes all buckets, so this is a full-store scan. + pub async fn non_durable_entries( + &self, + partition: Option, + ) -> Vec<(Partition, Address)> { + let _ = self.deserialize_all_buckets().await; + + let mut found = Vec::new(); + for group in self.group.iter() { + let active_buckets = group.bucket_count.load(atomic::Ordering::Relaxed); + for bucket_index in 0..active_buckets { + let bucket = group.bucket(bucket_index).read().await; + for entry in bucket.entry.iter() { + if let Some(partition) = partition + && entry.partition != partition + { + continue; + } + let flags = entry.data.flags; + if flags & FragmentFlags::PayloadStoredLocal != 0 + && flags & FragmentFlags::PayloadStoredDurable == 0 + && flags & FragmentFlags::PayloadObliterated == 0 + { + found.push((entry.partition, entry.address)); + } + } + } + } + found + } + pub async fn verify_fragment( self: Arc, address: Address, @@ -4663,6 +4819,93 @@ mod tests { assert_eq!(bytes.as_ref(), payload.as_ref()); } + #[tokio::test] + async fn durable_put_waits_for_batched_fsync_before_returning() { + // Regression test for the false-durability data-loss bug: a store acting as the + // authoritative persistence layer (durable_flush_tick_ms > 0, matching + // lore-server's configuration) must not let a `PayloadStoredDurable` put return + // until the background durable-flush ticker has confirmed that write with a real + // fsync (i.e. `flush_barrier`'s generation has advanced by 2 — see its docs). + // A short tick keeps this test fast while still exercising real ticks. + let store = LocalImmutableStore::new( + None, + ImmutableStoreSettings { + initial_fan_out_level: 1, + durable_flush_tick_ms: 5, + ..Default::default() + }, + ) + .await + .unwrap(); + + let payload = Bytes::from(vec![0xABu8; 64]); + let hash = Hash::hash_buffer(payload.as_ref()); + let context = Context::from([0x11u8; 16]); + let address = Address { context, hash }; + let partition = Partition::from([0x22u8; 16]); + let group_index = address.hash.data()[0] as usize; + + let fragment = Fragment { + flags: FragmentFlags::PayloadStoredDurable.bits(), + size_payload: payload.len() as u32, + size_content: payload.len() as u64, + }; + + let generation_before = store.group[group_index].flush_barrier.current_generation(); + + store + .clone() + .store(partition, address, fragment, Some(payload), false) + .await + .unwrap(); + + assert!( + store.group[group_index].flush_barrier.current_generation() > generation_before, + "a durable store() call must wait for a batched fsync to complete before returning" + ); + } + + #[tokio::test] + async fn non_durable_batch_window_skips_fsync_wait() { + // Default (durable_flush_tick_ms: 0, matching clients caching remote-durable + // content) must not spawn the ticker or pay any fsync-wait cost — preserves prior + // behavior. + let store = LocalImmutableStore::new( + None, + ImmutableStoreSettings { + initial_fan_out_level: 1, + ..Default::default() + }, + ) + .await + .unwrap(); + + let payload = Bytes::from(vec![0xCDu8; 64]); + let hash = Hash::hash_buffer(payload.as_ref()); + let context = Context::from([0x33u8; 16]); + let address = Address { context, hash }; + let partition = Partition::from([0x44u8; 16]); + let group_index = address.hash.data()[0] as usize; + + let fragment = Fragment { + flags: FragmentFlags::PayloadStoredDurable.bits(), + size_payload: payload.len() as u32, + size_content: payload.len() as u64, + }; + + store + .clone() + .store(partition, address, fragment, Some(payload), false) + .await + .unwrap(); + + assert_eq!( + store.group[group_index].flush_barrier.current_generation(), + 0, + "durable_batch_window_ms: 0 must not schedule/await a batched fsync" + ); + } + #[test] fn immutable_store_settings_default_includes_fan_out_fields() { let s = ImmutableStoreSettings::default(); diff --git a/lore-storage/src/maintenance.rs b/lore-storage/src/maintenance.rs index b64814cb..d4f2a85b 100644 --- a/lore-storage/src/maintenance.rs +++ b/lore-storage/src/maintenance.rs @@ -741,6 +741,7 @@ mod tests { ), packstore: PackStore::new(Some(tempdir.to_path_buf()), 1, None), flush: tokio::sync::Mutex::new(JoinSet::new()), + flush_barrier: crate::local::immutable_store::FlushBarrier::default(), }); // Buffer lengths are primes to ensure test actually verify the correct thing diff --git a/lore-storage/src/write.rs b/lore-storage/src/write.rs index 724a2615..c9c3a44c 100644 --- a/lore-storage/src/write.rs +++ b/lore-storage/src/write.rs @@ -166,13 +166,24 @@ pub async fn wait_if_in_flight(partition: Partition, address: Address) { } /// Result of a [`store_fragment`] operation. +#[derive(Debug)] pub struct StoreResult { pub address: Address, pub fragment: Fragment, pub deduplicated: bool, } -/// Put a fragment to a remote session with retry on `SlowDown`. +/// Bound on retries for `Disconnected` in `remote_put_retry`. Mirrors +/// `MAX_STALE_SESSION_RETRIES` on the read path: transient timeouts and QUIC +/// reconnects typically recover within a retry or two once the session is +/// re-established, while a genuinely unreachable server should surface +/// quickly rather than looping through the full backoff schedule. +const MAX_PUT_DISCONNECT_RETRIES: u32 = 5; + +/// Put a fragment to a remote session with retry on `SlowDown` and bounded +/// retry on `Disconnected` (transient timeouts/disconnects map to +/// `Disconnected`; the session is invalidated so the next attempt +/// re-establishes it). /// /// Takes an owned `Arc` so callers can spawn this into a /// background task (the returned future must be `'static`). @@ -183,6 +194,7 @@ async fn remote_put_retry( payload: Option, ) -> Result<(), StorageError> { let mut retry = store_retry(); + let mut disconnect_retries = 0; loop { match session.put(address, fragment, payload.clone()).await { Ok(_) => return Ok(()), @@ -191,6 +203,15 @@ async fn remote_put_retry( return Err(StorageError::from(SlowDown)); } } + Err(ref e) + if e.is_disconnected() && disconnect_retries < MAX_PUT_DISCONNECT_RETRIES => + { + disconnect_retries += 1; + session.invalidate().await; + if !retry.wait().await { + return Err(crate::error::protocol_error_to_storage(e.clone(), address)); + } + } Err(err) => return Err(crate::error::protocol_error_to_storage(err, address)), } } @@ -408,7 +429,13 @@ async fn store_fragment_dispatched( // Follower path: drop buffer and permit, register on the tracker. drop(buffer); drop(permit); - tracker.register_follower(follower_future(store.clone(), partition, address, token)); + tracker.register_follower(follower_future( + store.clone(), + partition, + address, + token, + remote_session.is_some(), + )); return Ok(StoreResult { address, fragment, @@ -565,11 +592,16 @@ async fn leader_body( } } - // Remote upload if session provided and not already durable + // Remote upload if session provided and not already durable. A failed + // upload must not be silently downgraded to a local-only write; the error + // must propagate. The payload is still written to the local store below + // (non-durable, cached) so the bytes remain locally recoverable. + let mut upload_error: Option = None; if !stored_durable && let Some(session) = remote_session.clone() { - stored_durable = remote_put_retry(session, address, fragment, Some(buffer.clone())) - .await - .is_ok(); + match remote_put_retry(session, address, fragment, Some(buffer.clone())).await { + Ok(()) => stored_durable = true, + Err(err) => upload_error = Some(err), + } } if stored_durable { @@ -588,6 +620,9 @@ async fn leader_body( drop(permit); drop(guard); + if let Some(err) = upload_error { + return Err(err); + } Ok((address, fragment)) } @@ -956,10 +991,13 @@ pub async fn hash_file( /// terminal store state for `address`. /// /// Returns `Ok((address, fragment))` if the store now holds a full-match entry -/// with either [`PayloadStoredDurable`](FragmentFlags::PayloadStoredDurable) or -/// [`PayloadStoredLocal`](FragmentFlags::PayloadStoredLocal) set. Returns an -/// internal error if no terminal entry exists — that means the leader errored -/// out and we have nothing to dedup against. +/// with the flags the follower's write required: when `require_durable` is set +/// (the follower's own write was remote-coupled), only +/// [`PayloadStoredDurable`](FragmentFlags::PayloadStoredDurable) counts — +/// a leader whose upload failed must not turn the follower into a false +/// success. Otherwise [`PayloadStoredLocal`](FragmentFlags::PayloadStoredLocal) +/// also satisfies. Returns an internal error if no terminal entry exists — +/// that means the leader errored out and we have nothing to dedup against. /// /// The follower holds no memory permit and no buffer; the caller is expected /// to have dropped both before invoking this future. @@ -968,15 +1006,18 @@ pub async fn follower_future( partition: Partition, address: Address, token: CancellationToken, + require_durable: bool, ) -> Result<(Address, Fragment), StorageError> { token.cancelled().await; + let required_flags = if require_durable { + FragmentFlags::PayloadStoredDurable.bits() + } else { + FragmentFlags::PayloadStoredDurable.bits() | FragmentFlags::PayloadStoredLocal.bits() + }; match store.query(partition, address, StoreMatch::MatchFull).await { Ok(result) if result.match_made == StoreMatch::MatchFull - && (result.fragment.flags - & (FragmentFlags::PayloadStoredDurable.bits() - | FragmentFlags::PayloadStoredLocal.bits())) - != 0 => + && (result.fragment.flags & required_flags) != 0 => { Ok((address, result.fragment)) } @@ -1057,7 +1098,7 @@ mod tests { let token = CancellationToken::new(); token.cancel(); - let result = follower_future(store, partition, address, token).await; + let result = follower_future(store, partition, address, token, false).await; let (addr, frag) = result.expect("follower should observe terminal entry"); assert_eq!(addr, address); assert_ne!( @@ -1074,7 +1115,7 @@ mod tests { let token = CancellationToken::new(); token.cancel(); - let err = follower_future(store, partition, address, token) + let err = follower_future(store, partition, address, token, false) .await .expect_err("follower should fail when no terminal entry"); let msg = format!("{err:?}"); @@ -1101,6 +1142,7 @@ mod tests { partition, address, token.clone(), + true, )); // Follower is waiting on the token. Write the entry AFTER spawn, THEN cancel. @@ -1473,6 +1515,137 @@ mod tests { assert_eq!(query.match_made, StoreMatch::MatchNone); } + /// A session whose every operation fails with `Disconnected` — the + /// pending resolver never succeeds, simulating a remote that times out + /// or is unreachable during upload. + fn failing_session() -> Arc { + Arc::new(StorageSession::pending(|| async { + Err(lore_transport::ProtocolError::from( + lore_base::error::Disconnected, + )) + })) + } + + #[tokio::test] + async fn upload_failure_fails_inline_store_but_keeps_local_payload() { + let (_dir, store) = make_test_store().await; + let (partition, address, fragment, buffer) = make_input(0x70); + + let err = store_fragment( + store.clone(), + partition, + address, + fragment, + buffer, + false, + Some(failing_session()), + None, + None, + ) + .await + .expect_err("failed remote upload must fail the store operation"); + assert!( + matches!(err, StorageError::Disconnected(_)), + "expected Disconnected, got: {err:?}" + ); + + // The payload must still be written locally (non-durable, cached). + let query = store + .query(partition, address, StoreMatch::MatchFull) + .await + .expect("query after failed upload"); + assert_eq!(query.match_made, StoreMatch::MatchFull); + assert_eq!( + query.fragment.flags & FragmentFlags::PayloadStoredDurable.bits(), + 0, + "durable flag must be clear after failed upload" + ); + assert_ne!( + query.fragment.flags & FragmentFlags::PayloadStoredLocal.bits(), + 0, + "payload must remain cached locally after failed upload" + ); + } + + #[tokio::test] + async fn upload_failure_surfaces_through_tracker_await_all() { + let (_dir, store) = make_test_store().await; + let (partition, address, fragment, buffer) = make_input(0x71); + let tracker = Arc::new(WriteTracker::new()); + + // Dispatch path returns Ok immediately; the leader's failed upload + // must be recorded in the tracker so a commit draining it fails + // before publishing the branch. + store_fragment( + store.clone(), + partition, + address, + fragment, + buffer, + false, + Some(failing_session()), + Some(tracker.clone()), + None, + ) + .await + .expect("dispatch returns Ok — upload runs in the leader task"); + + let err = tracker + .await_all() + .await + .expect_err("failed upload must surface through await_all"); + assert!( + matches!(err, StorageError::Disconnected(_)), + "expected Disconnected, got: {err:?}" + ); + + // Local payload retained for later repair. + let query = store + .query(partition, address, StoreMatch::MatchFull) + .await + .expect("query after failed upload"); + assert_eq!(query.match_made, StoreMatch::MatchFull); + assert_eq!( + query.fragment.flags & FragmentFlags::PayloadStoredDurable.bits(), + 0 + ); + } + + #[tokio::test] + async fn follower_requiring_durable_rejects_local_only_entry() { + let (_dir, store) = make_test_store().await; + let (partition, address) = make_address(0x72); + let payload = vec![0x72; 64]; + let fragment = Fragment { + flags: FragmentFlags::PayloadStoredLocal.bits(), + size_payload: payload.len() as u32, + size_content: payload.len() as u64, + }; + store + .clone() + .put( + partition, + address, + fragment, + Some(Bytes::from(payload)), + false, + ) + .await + .expect("put local-only terminal entry"); + + let token = CancellationToken::new(); + token.cancel(); + // A remote-coupled follower must not treat a local-only entry (the + // leader's upload failed) as success. + follower_future(store.clone(), partition, address, token.clone(), true) + .await + .expect_err("local-only entry must not satisfy a durable follower"); + // A local-only follower is satisfied by the same entry. + follower_future(store, partition, address, token, false) + .await + .expect("local-only follower accepts local entry"); + } + #[tokio::test(flavor = "multi_thread")] async fn concurrent_writers_of_same_address_dedup_through_tracker() { // Two concurrent store_fragment calls for the same (partition, address) diff --git a/lore/src/interface.rs b/lore/src/interface.rs index e38a71c7..7a6d2052 100644 --- a/lore/src/interface.rs +++ b/lore/src/interface.rs @@ -4587,6 +4587,7 @@ pub extern "C" fn lore_repository_store_immutable_query_async( pub type LoreRepositoryVerifyStateArgs = crate::repository::LoreRepositoryVerifyStateArgs; pub type LoreRepositoryVerifyFragmentArgs = crate::repository::LoreRepositoryVerifyFragmentArgs; +pub type LoreRepositoryPushContentArgs = crate::repository::LoreRepositoryPushContentArgs; /// Verify the integrity of the repository's stored fragments. /// diff --git a/lore/src/remote/command.rs b/lore/src/remote/command.rs index cb40961f..477aebff 100644 --- a/lore/src/remote/command.rs +++ b/lore/src/remote/command.rs @@ -93,6 +93,7 @@ pub enum LoreCommand { RepositoryStoreImmutableQuery(crate::repository::LoreRepositoryStoreImmutableQueryArgs), RepositoryVerifyState(crate::repository::LoreRepositoryVerifyStateArgs), RepositoryVerifyFragment(crate::repository::LoreRepositoryVerifyFragmentArgs), + RepositoryPushContent(crate::repository::LoreRepositoryPushContentArgs), RevisionAmend(crate::revision::LoreRevisionAmendArgs), RevisionCommit(crate::revision::LoreRevisionCommitArgs), RevisionInfo(crate::revision::LoreRevisionInfoArgs), diff --git a/lore/src/repository.rs b/lore/src/repository.rs index 2109813b..1d3c34f8 100644 --- a/lore/src/repository.rs +++ b/lore/src/repository.rs @@ -1067,6 +1067,65 @@ async fn verify_fragment_impl( lore_revision::repository::verify::verify_fragment(repository, core_args).await } +/// Arguments for pushing locally-stored, not-yet-durable content to the remote store. +#[repr(C)] +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, LoreArgs)] +#[handler(push_content_local)] +pub struct LoreRepositoryPushContentArgs { + /// Addresses to push, as `hash` or `hash-context` strings; ignored when `scan` is set + pub addresses: LoreArray, + /// Scan the local store for all non-durable entries of this repository and push them + pub scan: u8, +} + +/// Pushes locally-stored content that has not been confirmed durable on the remote +/// (e.g. after a commit whose upload timed out) back to the remote store. +/// +/// # Events +/// +/// ## Standard Events +/// +/// These events are emitted by all interface functions: +/// +/// | Event | Description | +/// |-------|-------------| +/// | [`LoreEvent::Log`](crate::interface::LoreEvent::Log) | Diagnostic messages throughout execution | +/// | [`LoreEvent::Error`](crate::interface::LoreEvent::Error) | Emitted for a non-fatal error during the operation | +/// | [`LoreEvent::Complete`](crate::interface::LoreEvent::Complete) | Always emitted at the end; `status` is `0` on success or the error code on failure | +/// | [`LoreEvent::End`](crate::interface::LoreEvent::End) | Always emitted after `Complete` to signal callback termination | +/// +/// ## Push Content Events +/// +/// | Event | Description | +/// |-------|-------------| +/// | [`LoreEvent::StorageUploadItemComplete`](crate::interface::LoreEvent::StorageUploadItemComplete) | Emitted for each pushed item with its outcome | +pub async fn push_content( + globals: LoreGlobalArgs, + args: LoreRepositoryPushContentArgs, + callback: LoreEventCallback, +) -> i32 { + dispatch_call(globals, args, callback, push_content_local).await +} + +async fn push_content_local( + globals: LoreGlobalArgs, + args: LoreRepositoryPushContentArgs, + callback: LoreEventCallback, +) -> i32 { + repository_call_read(globals, callback, args, push_content, push_content_impl).await +} + +async fn push_content_impl( + repository: Arc, + args: LoreRepositoryPushContentArgs, +) -> Result<(), RepositoryError> { + let core_args = lore_revision::repository::push_content::PushContentArgs { + addresses: args.addresses.as_slice().to_vec(), + scan: args.scan != 0, + }; + lore_revision::repository::push_content::push_content(repository, core_args).await +} + /// Arguments for querying the local immutable store by fragment address. #[repr(C)] #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, LoreArgs)] diff --git a/scripts/test/lore.py b/scripts/test/lore.py index 81ee1df9..a8239f34 100644 --- a/scripts/test/lore.py +++ b/scripts/test/lore.py @@ -473,6 +473,19 @@ def repository_verify_fragment( **kwargs, ) + def repository_push_content( + self, + addresses: list[str] | None = None, + scan: bool = False, + **kwargs: Unpack[GlobalOptions], + ): + return self.run( + ["repository", "push-content"] + + (addresses if addresses else []) + + (["--scan"] if scan else []), + **kwargs, + ) + @overload def branch_list( self, *, archived: bool = False, **kwargs: Unpack[GlobalOptionsParseable] diff --git a/scripts/test/test_push_content.py b/scripts/test/test_push_content.py new file mode 100644 index 00000000..21eb0908 --- /dev/null +++ b/scripts/test/test_push_content.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: 2026 Epic Games, Inc. +# SPDX-License-Identifier: MIT +import logging +import os + +import pytest + +from lore import Lore + +logger = logging.getLogger(__name__) + + +@pytest.mark.smoke +def test_push_content_scan_with_nothing_pending_is_a_no_op(new_lore_repo): + """--scan with no locally-cached, non-durable content should not error.""" + repo: Lore = new_lore_repo() + + test_file = "test.txt" + with repo.open_file(test_file, "w+b") as f: + f.write(os.urandom(1000)) + + repo.stage(scan=True) + repo.commit("Test commit") + repo.push() + + # Everything just committed and pushed is already durable on the remote, + # so a scan should find nothing to push. + repo.repository_push_content(scan=True) + + +@pytest.mark.smoke +def test_push_content_explicit_address_already_durable_is_a_no_op(new_lore_repo): + """An explicit address that's already durable on the remote should not error.""" + repo: Lore = new_lore_repo() + + test_file = "test.txt" + with repo.open_file(test_file, "w+b") as f: + f.write(os.urandom(1000)) + + repo.stage(scan=True) + repo.commit("Test commit") + repo.push() + + file_info = repo.file_info(test_file)[0] + address = f"{file_info.hash}-{file_info.context}" + + repo.repository_push_content(addresses=[address])