diff --git a/CHANGELOG.md b/CHANGELOG.md index 924dc59..9405294 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice. ## [Unreleased] +### Added + +- **A change to a synced folder can be staged, reviewed, and published on + purpose.** `fabric sync stage ` makes a staged copy under the fabric + home, outside every synced folder, and records the published file's hash as + its base. `fabric sync staged` lists staged files with their state, and + `fabric sync publish` writes the reviewed bytes into the folder, refusing a + file whose published version moved since it was staged unless `--force`. + With the daemon running, a publish is one scan, one persist, and one + reconcile per peer for the whole set. No build that has shipped can publish a + staged file, because every daemon publishes only from the folder it walks + and its own include globs. + ### Changed - **The dormant `fabric/sync-ipc/1` local bridge now has a frozen contract.** diff --git a/README.md b/README.md index 464f736..ea67255 100644 --- a/README.md +++ b/README.md @@ -1082,6 +1082,10 @@ fabric sync ls fabric sync ls --json fabric sync rm fabric sync reload +fabric sync stage [--from ] [--entry ] +fabric sync staged [--entry ] [--json] +fabric sync publish ... | --all --entry [--force] +fabric sync discard ... [--entry ] ``` `fabric sync add` is a convenience writer for `syncs.toml`; the file can also be @@ -1115,6 +1119,46 @@ restart or removing and later re-adding the name starts a new counter epoch. `fabric sync ls --json` emits a stable array with all of those fields plus a Boolean `drift` for automation. +### Stage a change before it publishes + +In a synced folder the write is the publish. The moment bytes land on disk, +every peer receives them, so an edit made over several saves publishes each +save, and nothing can be reviewed before it crosses. Staging adds the missing +state: a change that exists, is complete, and has not been distributed. + +```sh +fabric sync stage ~/catalog/docs/handbook.md # prints the path to edit +$EDITOR ~/.local/share/fabric/staging/st2-declarations-default/docs/handbook.md +fabric sync staged # review: state, hashes, both paths +fabric sync publish ~/catalog/docs/handbook.md # one revision, one reconcile per peer +``` + +`stage` resolves the entry from the target path and the include globs, seeds +the staged copy from the published file when there is one, or from `--from`, +and records the published file's hash as the base. `staged` lists every staged +file with its state: `new` when nothing is published at the target, `edit` when +the published file is still the one it was staged against, and `stale` when +that file moved. `publish` refuses a stale file unless `--force`, so a change a +peer made under a staged edit is never overwritten by accident. `discard` +removes staged copies and touches nothing in the folder. + +A staged file lives under `/staging//`, never inside a +synced folder. That location is the whole guarantee. A daemon decides what to +publish from exactly two things, the folder it walks and the include globs in +its own `syncs.toml`, so no build that has shipped can publish a staged file, +including an older binary after a rollback. A daemon restart with staged files +publishes nothing. `stage` refuses when the staging tree would lie inside a +synced folder. + +With the daemon running, `publish` hands the reviewed bytes to it, and the +daemon writes them under the entry's operation guard: one scan, one persist, +and one reconcile carries the whole set to each peer. Without a daemon, or +with one older than this command, `publish` writes each file atomically into +the folder itself and says so; the next scan records them, and a set can then +reach a peer in more than one reconcile. A crash in the middle of a multi-file +publish leaves the written files to publish on the next start and the rest +still staged; publish again to finish. + ### Sync an st2 catalog safely An st2 catalog mixes declarative fleet data, durable bus data, and strictly diff --git a/src/control.rs b/src/control.rs index 7aee025..0940d90 100644 --- a/src/control.rs +++ b/src/control.rs @@ -102,9 +102,44 @@ pub enum ControlRequest { }, /// Report which process owns sync and whether its companion is present. SyncRuntimeStatus, + /// Publish staged files into one sync entry under its operation guard, so + /// the set becomes one scan, one persist, and one reconcile on each peer. + /// + /// Carries the bytes rather than a path: the daemon writes exactly what the + /// caller reviewed, and a staged file is small by the nature of the thing + /// being staged. + SyncPublish { + name: String, + files: Vec, + #[serde(default)] + force: bool, + }, Shutdown, } +/// One file of a `SyncPublish` request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SyncPublishFile { + /// The path inside the synced folder, in manifest form. + pub rel: String, + pub bytes: Vec, + #[serde(default)] + pub executable: bool, + /// The hex content hash of the published file when this was staged, or + /// `None` when there was no published file. The daemon refuses to publish + /// over a file that moved since, unless forced. + #[serde(default)] + pub base: Option, +} + +/// One file of a `SyncPublished` response. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SyncPublishedFile { + pub rel: String, + pub version: u64, + pub hash: String, +} + fn default_persist() -> bool { true } @@ -204,6 +239,9 @@ pub enum ControlResponse { SyncRuntimeStatus { runtime: SyncRuntimeStatus, }, + SyncPublished { + files: Vec, + }, Error { message: String, }, diff --git a/src/daemon.rs b/src/daemon.rs index 9c505a8..c8b2182 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -3480,6 +3480,41 @@ async fn process_control_request( ControlRequest::SyncRuntimeStatus => ControlResponse::SyncRuntimeStatus { runtime: state.sync_runtime_status().await, }, + ControlRequest::SyncPublish { name, files, force } => { + let Some(engine) = state.sync_engine() else { + anyhow::bail!("this daemon does not own sync, so it cannot publish"); + }; + let files = files + .into_iter() + .map(|file| { + let base = match file.base { + Some(hex) => Some( + crate::sync::manifest::ContentHash::from_hex(&hex).ok_or_else(|| { + anyhow::anyhow!("{}: the base is not a content hash", file.rel) + })?, + ), + None => None, + }; + Ok(crate::sync::staging::PublishFile { + rel: file.rel, + bytes: file.bytes, + executable: file.executable, + base, + }) + }) + .collect::>>()?; + let published = engine.publish_staged(&name, files, force).await?; + ControlResponse::SyncPublished { + files: published + .into_iter() + .map(|file| crate::control::SyncPublishedFile { + rel: file.rel, + version: file.version, + hash: file.hash.to_hex(), + }) + .collect(), + } + } ControlRequest::Shutdown => { state.cancel.cancel(); ControlResponse::Ok diff --git a/src/main.rs b/src/main.rs index e754731..e5bf70a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,6 +23,7 @@ use fabric::{ service::{self, ServiceInstallOptions}, shell::{self, ServerFrame}, sync::config::{SyncBook, SyncEntry, SyncPeers, SyncPolicy}, + sync::staging, telemetry::{PeerTelemetry, TelemetryWindow}, terminal::TerminalModeGuard, update, @@ -452,6 +453,51 @@ enum SyncCommands { Rm { name_or_folder: String }, /// Re-read syncs.toml into the running daemon (like reload-peers). Reload, + /// Stage a change to a synced file without publishing it. + /// + /// The staged copy lives under the fabric home, outside every synced + /// folder, so no daemon publishes it until `fabric sync publish`. + Stage { + /// The path inside a synced folder that the change is for. + target: String, + /// Seed the staged copy from this file instead of the published one. + #[arg(long)] + from: Option, + /// The sync entry, when the target lies inside more than one folder. + #[arg(long)] + entry: Option, + }, + /// List staged changes and whether their published file moved since. + Staged { + /// Only this sync entry. + #[arg(long)] + entry: Option, + /// Emit a JSON array for scripts. + #[arg(long)] + json: bool, + }, + /// Publish staged changes into their synced folder. + Publish { + /// The paths inside synced folders to publish. + targets: Vec, + /// Publish every staged file of --entry. + #[arg(long)] + all: bool, + /// The sync entry, for --all or to break a tie between folders. + #[arg(long)] + entry: Option, + /// Publish even if the published file changed since it was staged. + #[arg(long)] + force: bool, + }, + /// Remove staged changes without publishing them. + Discard { + /// The paths inside synced folders whose staged copies to remove. + targets: Vec, + /// The sync entry, to break a tie between folders. + #[arg(long)] + entry: Option, + }, } #[derive(Debug, Subcommand)] @@ -1226,6 +1272,117 @@ fn expose_request( async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { match command { + SyncCommands::Stage { + target, + from, + entry, + } => { + let book = SyncBook::load(home)?; + let target = absolutize(&target)?; + let from = from.as_deref().map(absolutize).transpose()?; + let staged = staging::stage(home, &book, &target, from.as_deref(), entry.as_deref())?; + println!("staged\t{}", staged.rel); + println!("entry\t{}", staged.entry); + println!("edit\t{}", staged.staged_path.display()); + println!("target\t{}", staged.target_path.display()); + println!("base\t{}", staged.base.as_deref().unwrap_or("new")); + } + SyncCommands::Staged { entry, json } => { + let book = SyncBook::load(home)?; + let files = staging::list(home, &book, entry.as_deref())?; + if json { + println!("{}", serde_json::to_string_pretty(&files)?); + return Ok(()); + } + if files.is_empty() { + println!("nothing staged"); + return Ok(()); + } + for file in files { + println!( + "{}\t{}\t{}\t{}B\thash={}\tbase={}\tedit={}\ttarget={}", + file.entry, + file.rel, + file.state(), + file.bytes, + &file.hash[..12], + file.base + .as_deref() + .map(|hex| &hex[..12]) + .unwrap_or("new"), + file.staged_path.display(), + file.target_path.display() + ); + } + } + SyncCommands::Publish { + targets, + all, + entry, + force, + } => { + let book = SyncBook::load(home)?; + let groups = group_targets_by_entry(&book, &targets, all, entry.as_deref())?; + for (name, rels) in groups { + let (configured, files) = staging::read_for_publish(home, &book, &name, &rels)?; + let rels: Vec = files.iter().map(|file| file.rel.clone()).collect(); + let request = ControlRequest::SyncPublish { + name: name.clone(), + files: files + .iter() + .map(|file| fabric::control::SyncPublishFile { + rel: file.rel.clone(), + bytes: file.bytes.clone(), + executable: file.executable, + base: file.base.map(|hash| hash.to_hex()), + }) + .collect(), + force, + }; + match send_control(home, request).await { + Ok(ControlResponse::SyncPublished { files }) => { + for file in files { + println!( + "published\t{name}\t{}\tversion={}\thash={}", + file.rel, + file.version, + &file.hash[..12] + ); + } + } + Ok(response) => bail!("unexpected daemon response: {response:?}"), + // Only when no daemon can take the request: it is down, or + // it predates SyncPublish. A refusal is a decision and is + // never retried around. + Err(error) if daemon_cannot_publish(&error) => { + let placed = staging::publish_locally(&configured, &files, force)?; + for (rel, hash) in placed { + println!( + "placed\t{name}\t{rel}\thash={}\tvia=folder", + &hash.to_hex()[..12] + ); + } + println!( + "note\tno daemon took the publish ({}); the next scan records the \ + files, and a set can cross to a peer in more than one reconcile", + first_line(&format!("{error:#}")) + ); + } + Err(error) => return Err(error), + } + staging::forget(home, &name, &rels)?; + } + } + SyncCommands::Discard { targets, entry } => { + let book = SyncBook::load(home)?; + let groups = group_targets_by_entry(&book, &targets, false, entry.as_deref())?; + for (name, rels) in groups { + staging::forget(home, &name, &rels)?; + for rel in rels { + println!("discarded\t{name}\t{rel}"); + } + } + } SyncCommands::Add { folder, name, @@ -1270,10 +1427,18 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { ) } }; + // Staged files are a local fact the daemon does not know: they live + // in the fabric home, outside every folder. Count them here so a + // staged change is never forgotten because nobody ran `staged`. + let staged_counts = staged_counts(home); if json { let entries: Vec<_> = entries .iter() - .map(|entry| SyncLsJsonEntry::from(entry).with_runtime(&runtime)) + .map(|entry| { + SyncLsJsonEntry::from(entry) + .with_runtime(&runtime) + .with_staged(staged_counts.get(&entry.name).copied().unwrap_or(0)) + }) .collect(); println!("{}", serde_json::to_string_pretty(&entries)?); return Ok(()); @@ -1286,9 +1451,10 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { println!("no sync entries"); } for entry in entries { + let staged = staged_counts.get(&entry.name).copied().unwrap_or(0); if runtime.owner == "unavailable" { println!( - "{}\t{}\t{}\tpeers={}\truntime=unavailable\tdrift=unknown\tstopped={}", + "{}\t{}\t{}\tpeers={}\truntime=unavailable\tdrift=unknown\tstopped={}\tstaged={staged}", entry.name, entry.folder, entry.policy, @@ -1304,7 +1470,7 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { && entry.scan_issues.is_empty() { println!( - "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=clean\tscan_issues=none\tstopped={}\taway={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", + "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=clean\tscan_issues=none\tstopped={}\taway={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}\tstaged={staged}", entry.name, entry.folder, entry.policy, @@ -1331,7 +1497,7 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { ); } else { println!( - "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=WARNING missing={} unexpected={} mismatched={}\tscan_issues={}\tstopped={}\taway={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", + "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=WARNING missing={} unexpected={} mismatched={}\tscan_issues={}\tstopped={}\taway={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}\tstaged={staged}", entry.name, entry.folder, entry.policy, @@ -1458,6 +1624,9 @@ struct SyncLsJsonEntry<'a> { /// `tombstones` can match while the state differs, so they cannot answer /// this. Empty from a daemon that predates the field. digest: &'a str, + /// Files staged for this entry under the fabric home and not yet + /// published. Counted locally; the daemon does not know them. + staged: usize, } impl<'a> From<&'a fabric::control::SyncEntryStatus> for SyncLsJsonEntry<'a> { @@ -1498,6 +1667,7 @@ impl<'a> From<&'a fabric::control::SyncEntryStatus> for SyncLsJsonEntry<'a> { content_bytes: entry.content_bytes, delta_fallbacks: entry.delta_fallbacks, digest: &entry.digest, + staged: 0, sync_passes: entry.sync_passes, full_scans: entry.full_scans, inbound_noop_transactions: entry.inbound_noop_transactions, @@ -1514,6 +1684,11 @@ impl<'a> From<&'a fabric::control::SyncEntryStatus> for SyncLsJsonEntry<'a> { } impl SyncLsJsonEntry<'_> { + fn with_staged(mut self, staged: usize) -> Self { + self.staged = staged; + self + } + fn with_runtime(mut self, runtime: &SyncRuntimeStatus) -> Self { self.runtime_owner = runtime.owner.clone(); self.companion = runtime.companion.clone(); @@ -2246,6 +2421,7 @@ mod sync_ls_tests { "inbound_guarded_transactions": 3, "sync_passes": 9, "scan_micros": 1500, + "staged": 0, "materialize_micros": 2500, "persist_micros": 3500, "reconcile_micros": 4500, @@ -2293,6 +2469,69 @@ mod sync_ls_tests { } } +/// Group target paths by the entry that would publish each, or take every +/// staged file of one entry for `--all`. +fn group_targets_by_entry( + book: &SyncBook, + targets: &[String], + all: bool, + entry: Option<&str>, +) -> Result>> { + let mut groups: BTreeMap> = BTreeMap::new(); + if all { + let Some(entry) = entry else { + bail!("--all needs --entry "); + }; + if book.get(entry).is_none() { + bail!("no sync entry named {entry:?}"); + } + groups.insert(entry.to_string(), Vec::new()); + return Ok(groups); + } + if targets.is_empty() { + bail!("give one or more target paths, or --all --entry "); + } + for target in targets { + let resolved = staging::resolve_target(book, &absolutize(target)?, entry)?; + groups + .entry(resolved.entry.name) + .or_default() + .push(resolved.rel); + } + Ok(groups) +} + +/// True only when no daemon can take a request: it is not running, or it is +/// an older build that does not know the request type. +fn daemon_cannot_publish(error: &anyhow::Error) -> bool { + let detail = format!("{error:#}"); + detail.contains("is not running") || detail.contains("unknown variant") +} + +fn first_line(text: &str) -> &str { + text.lines().next().unwrap_or(text) +} + +/// Staged files per entry, from the staging tree under the fabric home. An +/// unreadable tree prints one line and counts as nothing, so `sync ls` still +/// answers about the daemon. +fn staged_counts(home: &FabricHome) -> BTreeMap { + let listed = SyncBook::load(home).and_then(|book| staging::list(home, &book, None)); + match listed { + Ok(files) => { + let mut counts = BTreeMap::new(); + for file in files { + *counts.entry(file.entry).or_insert(0) += 1; + } + counts + } + Err(error) => { + eprintln!("fabric: could not read the staging tree: {error:#}"); + BTreeMap::new() + } + } +} + fn absolutize(folder: &str) -> Result { let path = PathBuf::from(folder); if path.is_absolute() { diff --git a/src/sync/engine.rs b/src/sync/engine.rs index ee2efd3..554cff2 100644 --- a/src/sync/engine.rs +++ b/src/sync/engine.rs @@ -544,6 +544,14 @@ impl Drop for InboundWaiter { } } +/// One file after `SyncEngine::publish_staged` recorded it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PublishedFile { + pub rel: String, + pub version: u64, + pub hash: ContentHash, +} + /// One configured entry's live state. struct EntryState { config: SyncEntry, @@ -1138,6 +1146,115 @@ impl SyncEngine { Ok(()) } + /// Publish staged files into `name` under its operation guard. + /// + /// One guard hold, so the set is one scan, one persist, and one wake, and + /// a peer adopts all of it in one reconcile or none of it. Every base is + /// checked against the live manifest before anything is written, so a + /// refused set changes nothing. Each write goes through the engine write + /// path and its receipt journal, so the watcher acknowledges the daemon's + /// own writes without a second scan. The forward wake then makes the entry + /// loop push the set now rather than on the next tick. + /// + /// Atomic per file, not across a crash. A crash after some writes and + /// before the scan leaves those files in the folder, and the next start + /// records them; the rest are still staged. Publish again to finish. + pub async fn publish_staged( + &self, + name: &str, + files: Vec, + force: bool, + ) -> Result> { + let Some(entry) = self.entries.read().await.get(name).cloned() else { + anyhow::bail!("no sync entry named {name:?}"); + }; + if files.is_empty() { + return Ok(Vec::new()); + } + let _operation = entry.operation.lock().await; + // A local edit the watcher has reported but no scan has recorded yet + // is still a change to the published file. Bring the manifest up to + // date before comparing bases against it. + if !entry.work.is_clean() { + self.scan_entry(&entry).await?; + } + { + let node = entry.node.lock().await; + let mut refusals = Vec::new(); + for file in &files { + if !entry.config.includes(&file.rel) { + refusals.push(format!( + "{}: no include glob of sync {name:?} matches it", + file.rel + )); + continue; + } + let current = node + .manifest() + .get(&file.rel) + .and_then(|recorded| recorded.meta()) + .map(|meta| meta.hash); + if current != file.base && !force { + refusals.push(super::staging::refusal_line( + &file.rel, file.base, current, + )); + } + } + if !refusals.is_empty() { + anyhow::bail!("publish refused:\n{}", refusals.join("\n")); + } + } + + let generation = entry.work.mutation_generation.load(Ordering::Acquire); + let root = entry.config.folder.clone(); + let work = entry.work.clone(); + let files = tokio::task::spawn_blocking(move || { + for file in &files { + let path = root.join(&file.rel); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + write_atomic_with_mode(&path, &file.bytes, file.executable)?; + work.record_engine_write(&path, content_hash(&file.bytes), generation); + } + Ok::<_, anyhow::Error>(files) + }) + .await + .context("the publish blocking task stopped")??; + + self.scan_entry(&entry).await?; + self.persist_entry(&entry).await?; + entry.work.mark_generation_durable(generation); + let published = { + let node = entry.node.lock().await; + files + .iter() + .map(|file| { + let Some(meta) = node + .manifest() + .get(&file.rel) + .and_then(|recorded| recorded.meta()) + else { + anyhow::bail!("{}: the scan after the write did not record it", file.rel); + }; + Ok(PublishedFile { + rel: file.rel.clone(), + version: meta.version, + hash: meta.hash, + }) + }) + .collect::>>()? + }; + drop(_operation); + // Both, for the same two reasons as an inbound adoption: the wake + // makes the set reach peers now, and the generation makes the tick a + // backstop if the wake is missed. + entry.work.record_forward(); + entry.work.forward.notify_one(); + Ok(published) + } + /// The configured sync names. pub async fn names(&self) -> Vec { let mut names: Vec = self.entries.read().await.keys().cloned().collect(); @@ -4078,7 +4195,7 @@ pub(crate) const METADATA_ONLY_CHANGES_DO_NOT_PROPAGATE: () = (); /// syncs the attributes git syncs, and git deliberately does not track mtime. /// `FileMeta` still carries one, but it is informational only: see the note on /// that field, and on [`METADATA_ONLY_CHANGES_DO_NOT_PROPAGATE`]. -fn write_atomic_with_mode(path: &Path, bytes: &[u8], executable: bool) -> Result<()> { +pub(crate) fn write_atomic_with_mode(path: &Path, bytes: &[u8], executable: bool) -> Result<()> { write_atomic_inner(path, bytes, executable) } @@ -4322,7 +4439,7 @@ fn now_secs() -> i64 { } /// Make a sync name safe to use as a directory component for its manifest store. -fn sanitize_name(name: &str) -> String { +pub(crate) fn sanitize_name(name: &str) -> String { name.chars() .map(|c| { if c.is_alphanumeric() || c == '-' || c == '_' { @@ -11130,4 +11247,428 @@ mod tests { "the removed file was written back to disk" ); } + + // ---- staging: a change that exists, is complete, and is not yet published ---- + + /// Two engines over loopback, each with a bus entry named "bus" on its own + /// folder, converged on one seed file. The convergence matters: a later + /// absence on B is then a fact about staging, not about a pair that never + /// synced. + async fn staged_pair() -> ( + tempfile::TempDir, + tempfile::TempDir, + Arc>, + Arc>, + Arc, + Arc, + ) { + let dir_a = tempfile::tempdir().unwrap(); + let dir_b = tempfile::tempdir().unwrap(); + let root_a = dir_a.path().join("resources"); + let root_b = dir_b.path().join("resources"); + std::fs::create_dir_all(&root_a).unwrap(); + std::fs::create_dir_all(&root_b).unwrap(); + write_bus_sync(dir_a.path(), &root_a); + write_bus_sync(dir_b.path(), &root_b); + let ta = Arc::new(LoopbackTransport::default()); + let tb = Arc::new(LoopbackTransport::default()); + let a = SyncEngine::new( + FabricHome::new(dir_a.path()), + Author([1; 32]), + ta.clone(), + CancellationToken::new(), + ) + .await + .unwrap(); + let b = SyncEngine::new( + FabricHome::new(dir_b.path()), + Author([2; 32]), + tb.clone(), + CancellationToken::new(), + ) + .await + .unwrap(); + ta.add_peer("b", "bus", b.node_for("bus").await.unwrap()); + tb.add_peer("a", "bus", a.node_for("bus").await.unwrap()); + std::fs::write(root_a.join("seed.md"), b"seed").unwrap(); + a.sync_once("bus").await.unwrap(); + b.sync_once("bus").await.unwrap(); + assert_eq!(std::fs::read(root_b.join("seed.md")).unwrap(), b"seed"); + (dir_a, dir_b, a, b, ta, tb) + } + + async fn pass_both(a: &SyncEngine, b: &SyncEngine) { + a.sync_once("bus").await.unwrap(); + b.sync_once("bus").await.unwrap(); + } + + async fn manifest_holds(engine: &SyncEngine, rel: &str) -> bool { + engine + .node_for("bus") + .await + .unwrap() + .lock() + .await + .manifest() + .get(rel) + .is_some_and(Entry::is_present) + } + + async fn assert_peer_never_saw(engine: &SyncEngine, home: &Path, rel: &str) { + assert!( + !manifest_holds(engine, rel).await, + "the peer must not hold {rel} in its manifest" + ); + assert!( + !home.join("resources").join(rel).exists(), + "the peer must not hold {rel} on disk" + ); + } + + #[tokio::test] + async fn a_staged_file_never_enters_the_manifest_or_reaches_a_peer() { + let (dir_a, dir_b, a, b, _ta, _tb) = staged_pair().await; + let home_a = FabricHome::new(dir_a.path()); + let book_a = SyncBook::load(&home_a).unwrap(); + + let staged = crate::sync::staging::stage( + &home_a, + &book_a, + &dir_a.path().join("resources/draft.md"), + None, + None, + ) + .unwrap(); + std::fs::write(&staged.staged_path, b"not yet").unwrap(); + assert!( + !staged.staged_path.starts_with(dir_a.path().join("resources")), + "the staged copy must live outside the synced folder" + ); + + for _ in 0..3 { + pass_both(&a, &b).await; + } + assert!( + !manifest_holds(&a, "draft.md").await, + "the staged file entered the manifest of the machine that staged it" + ); + assert_peer_never_saw(&b, dir_b.path(), "draft.md").await; + assert_eq!( + std::fs::read(&staged.staged_path).unwrap(), + b"not yet", + "the staged copy must survive every pass untouched" + ); + } + + /// EXPECTED TO FAIL, FOREVER. THIS IS THE CONTROL FOR THE TEST ABOVE. + /// + /// An absence assertion has two possible causes: there was nothing to see, + /// or the assertion cannot see. The test above asserts that a peer never + /// holds a staged file. This test writes the same file INSIDE the synced + /// folder, where it must publish, and runs the same assertion. It must + /// panic, and `should_panic` pins the exact message, so the assertion is + /// known to see a leak. If this test ever passes, the staged-file test + /// above has stopped watching and proves nothing. + #[tokio::test] + #[should_panic(expected = "the peer must not hold draft.md in its manifest")] + async fn the_leak_control_sees_a_file_placed_inside_the_folder() { + let (dir_a, dir_b, a, b, _ta, _tb) = staged_pair().await; + std::fs::write(dir_a.path().join("resources/draft.md"), b"leaks").unwrap(); + for _ in 0..3 { + pass_both(&a, &b).await; + } + assert_peer_never_saw(&b, dir_b.path(), "draft.md").await; + } + + #[tokio::test] + async fn an_engine_restart_with_a_staged_file_publishes_nothing() { + let (dir_a, dir_b, a, b, ta, tb) = staged_pair().await; + let home_a = FabricHome::new(dir_a.path()); + let book_a = SyncBook::load(&home_a).unwrap(); + let staged = crate::sync::staging::stage( + &home_a, + &book_a, + &dir_a.path().join("resources/draft.md"), + None, + None, + ) + .unwrap(); + std::fs::write(&staged.staged_path, b"held across a restart").unwrap(); + + // Stop A and start it again over the same home, as a daemon restart + // does. The transports point at the old node, so rewire them. + drop(a); + drop(ta); + let ta = Arc::new(LoopbackTransport::default()); + let a = SyncEngine::new( + home_a.clone(), + Author([1; 32]), + ta.clone(), + CancellationToken::new(), + ) + .await + .unwrap(); + ta.add_peer("b", "bus", b.node_for("bus").await.unwrap()); + tb.peers.lock().unwrap().clear(); + tb.add_peer("a", "bus", a.node_for("bus").await.unwrap()); + + for _ in 0..3 { + pass_both(&a, &b).await; + } + assert!( + !manifest_holds(&a, "draft.md").await, + "the restart published the staged file on the machine that staged it" + ); + assert_peer_never_saw(&b, dir_b.path(), "draft.md").await; + assert_eq!( + std::fs::read(&staged.staged_path).unwrap(), + b"held across a restart" + ); + } + + #[tokio::test] + async fn publish_records_the_set_in_one_pass_and_the_peer_adopts_it_in_one_reconcile() { + let (dir_a, dir_b, a, b, _ta, _tb) = staged_pair().await; + let home_a = FabricHome::new(dir_a.path()); + let book_a = SyncBook::load(&home_a).unwrap(); + let root_a = dir_a.path().join("resources"); + for (rel, bytes) in [ + ("notes/one.md", &b"one"[..]), + ("notes/two.md", &b"two"[..]), + ("three.md", &b"three"[..]), + ] { + let staged = + crate::sync::staging::stage(&home_a, &book_a, &root_a.join(rel), None, None) + .unwrap(); + std::fs::write(&staged.staged_path, bytes).unwrap(); + } + let (_entry, files) = + crate::sync::staging::read_for_publish(&home_a, &book_a, "bus", &[]).unwrap(); + assert_eq!(files.len(), 3); + let rels: Vec = files.iter().map(|file| file.rel.clone()).collect(); + + let entry = a.entries.read().await.get("bus").cloned().unwrap(); + let scans_before = entry.work.full_scans.load(Ordering::Relaxed); + let persists_before = entry.work.persist_calls.load(Ordering::Relaxed); + let published = a.publish_staged("bus", files, false).await.unwrap(); + + assert_eq!(published.len(), 3); + for file in &published { + assert_eq!(file.version, 1, "{} must be a first version", file.rel); + } + assert_eq!( + entry.work.full_scans.load(Ordering::Relaxed) - scans_before, + 1, + "a publish of a clean entry costs exactly one scan" + ); + assert_eq!( + entry.work.persist_calls.load(Ordering::Relaxed) - persists_before, + 1, + "a publish persists exactly once" + ); + assert!( + entry.work.has_pending_forward(), + "a publish must wake the entry loop so peers receive it without waiting for a tick" + ); + assert_eq!(std::fs::read(root_a.join("notes/one.md")).unwrap(), b"one"); + assert_eq!(std::fs::read(root_a.join("three.md")).unwrap(), b"three"); + + // One pass on B is one reconcile with A, and it carries the whole set. + b.sync_once("bus").await.unwrap(); + let root_b = dir_b.path().join("resources"); + assert_eq!(std::fs::read(root_b.join("notes/one.md")).unwrap(), b"one"); + assert_eq!(std::fs::read(root_b.join("notes/two.md")).unwrap(), b"two"); + assert_eq!(std::fs::read(root_b.join("three.md")).unwrap(), b"three"); + + crate::sync::staging::forget(&home_a, "bus", &rels).unwrap(); + assert!( + crate::sync::staging::list(&home_a, &book_a, Some("bus")) + .unwrap() + .is_empty(), + "published files must leave the staging tree" + ); + } + + #[tokio::test] + async fn a_publish_after_the_published_file_changed_is_refused_and_force_publishes_the_next_version() + { + let (dir_a, _dir_b, a, _b, _ta, _tb) = staged_pair().await; + let home_a = FabricHome::new(dir_a.path()); + let book_a = SyncBook::load(&home_a).unwrap(); + let root_a = dir_a.path().join("resources"); + let staged = + crate::sync::staging::stage(&home_a, &book_a, &root_a.join("seed.md"), None, None) + .unwrap(); + assert_eq!( + staged.base.as_deref(), + Some(content_hash(b"seed").to_hex().as_str()), + "staging a published file records its hash as the base" + ); + std::fs::write(&staged.staged_path, b"staged edit").unwrap(); + + // The published file moves under the staged change. + std::fs::write(root_a.join("seed.md"), b"someone else edited").unwrap(); + a.sync_once("bus").await.unwrap(); + + let (_entry, files) = + crate::sync::staging::read_for_publish(&home_a, &book_a, "bus", &[]).unwrap(); + let error = a + .publish_staged("bus", files.clone(), false) + .await + .unwrap_err(); + let detail = format!("{error:#}"); + assert!( + detail.contains("publish refused") && detail.contains("seed.md"), + "a moved base must refuse by name: {detail}" + ); + assert_eq!( + std::fs::read(root_a.join("seed.md")).unwrap(), + b"someone else edited", + "a refused publish must change nothing" + ); + + let published = a.publish_staged("bus", files, true).await.unwrap(); + assert_eq!(published.len(), 1); + assert_eq!(published[0].version, 3, "seed v1, edit v2, forced publish v3"); + assert_eq!(std::fs::read(root_a.join("seed.md")).unwrap(), b"staged edit"); + } + + #[tokio::test] + async fn a_publish_write_is_acknowledged_by_the_watcher_without_a_rescan() { + use notify::event::{CreateKind, DataChange, ModifyKind, RenameMode}; + + let (dir_a, _dir_b, a, _b, _ta, _tb) = staged_pair().await; + let home_a = FabricHome::new(dir_a.path()); + let book_a = SyncBook::load(&home_a).unwrap(); + let root_a = dir_a.path().join("resources"); + let staged = + crate::sync::staging::stage(&home_a, &book_a, &root_a.join("quiet.md"), None, None) + .unwrap(); + std::fs::write(&staged.staged_path, b"quiet bytes").unwrap(); + let (_entry, files) = + crate::sync::staging::read_for_publish(&home_a, &book_a, "bus", &[]).unwrap(); + a.publish_staged("bus", files, false).await.unwrap(); + + // The watcher then reports the daemon's own atomic write. The receipt + // recorded during publish must acknowledge it, exactly as it does for + // a materialization, so a publish does not schedule a second pass. + let entry = a.entries.read().await.get("bus").cloned().unwrap(); + let work = entry.work.clone(); + let final_path = root_a.join("quiet.md"); + let temp_path = root_a.join("quiet.md.fabric-tmp"); + let scans_before = work.full_scans.load(Ordering::Relaxed); + let push = + |batch: &mut Option, paths: Vec, kind: notify::EventKind| { + let event = WatchEvent { + paths, + generation: work.record_mutation(), + engine_write_candidate: watcher_event_can_match_engine_write(&kind), + rename: watcher_event_is_rename(&kind), + }; + if let Some(batch) = batch { + batch.push(event); + } else { + *batch = Some(WatchEventBatch::new(event)); + } + }; + let mut batch = None; + push( + &mut batch, + vec![temp_path.clone()], + notify::EventKind::Create(CreateKind::File), + ); + push( + &mut batch, + vec![temp_path.clone()], + notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), + ); + push( + &mut batch, + vec![temp_path, final_path], + notify::EventKind::Modify(ModifyKind::Name(RenameMode::Both)), + ); + assert!( + work.acknowledge_engine_write_batch(&batch.unwrap()), + "the watcher must acknowledge a publish write from its receipt" + ); + assert_eq!( + work.mutation_generation.load(Ordering::Acquire), + work.durable_generation.load(Ordering::Acquire), + "an acknowledged publish leaves no periodic dirty work" + ); + assert_eq!(work.full_scans.load(Ordering::Relaxed), scans_before); + } + + #[tokio::test] + async fn stage_refuses_a_target_outside_the_include_and_a_staging_tree_inside_a_folder() { + let dir = tempfile::tempdir().unwrap(); + let home = FabricHome::new(dir.path()); + let root = dir.path().join("resources"); + std::fs::create_dir_all(&root).unwrap(); + let mut book = SyncBook::default(); + book.upsert(SyncEntry { + name: "bus".to_string(), + folder: root.clone(), + peers: SyncPeers::Wildcard("*".to_string()), + policy: SyncPolicy::Bus, + include: Some(vec!["*.md".to_string()]), + }); + book.save(&home).unwrap(); + + let error = crate::sync::staging::stage(&home, &book, &root.join("notes.txt"), None, None) + .unwrap_err(); + let detail = format!("{error:#}"); + assert!( + detail.contains("include") && detail.contains("notes.txt"), + "a target no include glob matches must be refused by name: {detail}" + ); + assert!( + crate::sync::staging::list(&home, &book, None) + .unwrap() + .is_empty(), + "a refused stage leaves nothing behind" + ); + + let error = crate::sync::staging::stage( + &home, + &book, + &dir.path().join("elsewhere/notes.md"), + None, + None, + ) + .unwrap_err(); + let detail = format!("{error:#}"); + assert!( + detail.contains("not inside any synced folder"), + "a target outside every folder must say so: {detail}" + ); + + // An entry whose folder is the fabric home itself would publish the + // staging tree. Staging must refuse rather than stage into a folder. + let mut wide = SyncBook::default(); + wide.upsert(SyncEntry { + name: "home".to_string(), + folder: dir.path().to_path_buf(), + peers: SyncPeers::Wildcard("*".to_string()), + policy: SyncPolicy::Bus, + include: None, + }); + let error = crate::sync::staging::stage( + &home, + &wide, + &dir.path().join("anything.md"), + None, + None, + ) + .unwrap_err(); + let detail = format!("{error:#}"); + assert!( + detail.contains("lies inside the synced folder"), + "a staging tree inside a folder must be refused: {detail}" + ); + assert!( + !crate::sync::staging::staging_root(&home).join("home").exists(), + "a refused stage must write nothing into the folder" + ); + } } diff --git a/src/sync/manifest.rs b/src/sync/manifest.rs index 67860f9..08d3efd 100644 --- a/src/sync/manifest.rs +++ b/src/sync/manifest.rs @@ -36,6 +36,20 @@ use serde::{Deserialize, Serialize}; pub struct ContentHash(pub [u8; 32]); impl ContentHash { + /// Parse the 64-character form that `to_hex` writes. `None` for any other + /// length or a non-hex character. + pub fn from_hex(hex: &str) -> Option { + if hex.len() != 64 || !hex.is_ascii() { + return None; + } + let mut out = [0u8; 32]; + for (index, chunk) in hex.as_bytes().chunks(2).enumerate() { + let pair = std::str::from_utf8(chunk).ok()?; + out[index] = u8::from_str_radix(pair, 16).ok()?; + } + Some(Self(out)) + } + pub fn to_hex(self) -> String { let mut s = String::with_capacity(64); for byte in self.0 { diff --git a/src/sync/mod.rs b/src/sync/mod.rs index af86849..97e1f9b 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -23,6 +23,7 @@ pub mod ipc; pub mod manifest; pub mod node; pub mod paths; +pub mod staging; pub mod wire; pub use config::{PolicyRules, SyncBook, SyncEntry, SyncPeers, SyncPolicy}; @@ -31,3 +32,4 @@ pub use engine::{PeerRef, SYNC_LOG_TARGET, SyncEngine, SyncStatus, SyncTransport pub use manifest::{FileMeta, Manifest, ManifestDiff}; pub use node::{Reconciled, SyncNode, content_hash}; pub use paths::{SyncOwnerLease, SyncOwnerLeaseState, SyncPaths}; +pub use staging::{PublishFile, StagedFile}; diff --git a/src/sync/staging.rs b/src/sync/staging.rs new file mode 100644 index 0000000..9a18110 --- /dev/null +++ b/src/sync/staging.rs @@ -0,0 +1,552 @@ +//! Stage a change beside a synced folder and publish it on purpose. +//! +//! In a synced folder the write is the publish: the moment bytes land on disk, +//! every peer receives them. There is no state in which a change exists, is +//! complete, and has not yet been distributed, so nothing can be reviewed +//! before it crosses. This module adds that state. +//! +//! A staged file lives under `/staging//`, never +//! inside any synced folder. That location is the whole guarantee. A fabric +//! daemon decides what to publish from exactly two things: the folder it walks +//! and the include globs in its own `syncs.toml`. Nothing else, not a config +//! key, not engine state, not a control request, reaches an old binary's scan. +//! A path outside every folder is therefore never published by any build that +//! has shipped, which is what a fleet that rolls one machine at a time needs. +//! +//! Publishing writes the staged bytes to the target path and forgets the +//! staged copy. [`crate::sync::SyncEngine::publish_staged`] does that under the +//! entry's operation guard so a set of files is one scan, one persist, and one +//! reconcile on each peer. [`publish_locally`] is the fallback when no daemon +//! answers or the daemon predates the request. + +use std::{ + collections::{BTreeMap, BTreeSet}, + fs, + path::{Path, PathBuf}, + time::{SystemTime, UNIX_EPOCH}, +}; + +use anyhow::{Context, Result, bail}; +use serde::{Deserialize, Serialize}; + +use crate::config::FabricHome; + +use super::{ + config::{SyncBook, SyncEntry}, + engine::{sanitize_name, write_atomic_with_mode}, + manifest::{ContentHash, Manifest}, + node::content_hash, +}; + +/// The directory under the fabric home that holds every staged file. +pub const STAGING_DIR: &str = "staging"; +/// The per-entry record of what each staged file was staged against. +const SIDECAR_NAME: &str = "staged.json"; +const TEMP_SUFFIX: &str = ".fabric-tmp"; + +/// Where every staged file of every entry lives. +pub fn staging_root(home: &FabricHome) -> PathBuf { + home.root().join(STAGING_DIR) +} + +/// Where one entry's staged files live. Two entry names that sanitize to the +/// same directory name would share a tree; `syncs.toml` names are chosen by a +/// person and the sanitizer keeps letters, digits, `-` and `_`, so that is a +/// naming collision to notice, not a case to handle here. +pub fn entry_staging_dir(home: &FabricHome, entry: &str) -> PathBuf { + staging_root(home).join(sanitize_name(entry)) +} + +/// A target path resolved to the one entry that would publish it. +#[derive(Debug, Clone)] +pub struct ResolvedTarget { + pub entry: SyncEntry, + /// The path inside the folder, in manifest form. + pub rel: String, + /// The published path: the entry folder joined with `rel`. + pub target: PathBuf, +} + +/// One staged file as `fabric sync staged` reports it. +#[derive(Debug, Clone, Serialize)] +pub struct StagedFile { + pub entry: String, + pub rel: String, + pub staged_path: PathBuf, + pub target_path: PathBuf, + pub bytes: u64, + /// Hex content hash of the staged bytes. + pub hash: String, + pub executable: bool, + /// Hex content hash of the published file when this was staged, or `None` + /// when there was none. + pub base: Option, + pub staged_at: i64, + /// Hex content hash of the file at the target path now, or `None` when + /// there is none. + pub published_now: Option, +} + +impl StagedFile { + /// True when the published file is not the one this was staged against. + pub fn changed_since_staging(&self) -> bool { + self.base != self.published_now + } + + /// `new` for a path with no published file, `edit` for a staged change to + /// a published file that has not moved, `stale` when it has. + pub fn state(&self) -> &'static str { + match (self.base.is_some(), self.changed_since_staging()) { + (_, true) => "stale", + (false, false) => "new", + (true, false) => "edit", + } + } +} + +/// The bytes and base of one staged file, ready to publish. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PublishFile { + pub rel: String, + pub bytes: Vec, + pub executable: bool, + pub base: Option, +} + +#[derive(Debug, Default, Serialize, Deserialize)] +struct Sidecar { + #[serde(default)] + files: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct SidecarRecord { + base: Option, + staged_at: i64, +} + +fn load_sidecar(dir: &Path) -> Result { + let path = dir.join(SIDECAR_NAME); + if !path.exists() { + return Ok(Sidecar::default()); + } + let raw = fs::read(&path).with_context(|| format!("failed to read {}", path.display()))?; + serde_json::from_slice(&raw).with_context(|| format!("failed to parse {}", path.display())) +} + +fn save_sidecar(dir: &Path, sidecar: &Sidecar) -> Result<()> { + fs::create_dir_all(dir).with_context(|| format!("failed to create {}", dir.display()))?; + let raw = serde_json::to_vec_pretty(sidecar)?; + write_atomic_with_mode(&dir.join(SIDECAR_NAME), &raw, false) +} + +fn now_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_secs() as i64) + .unwrap_or(0) +} + +fn is_executable(meta: &fs::Metadata) -> bool { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + meta.permissions().mode() & 0o111 != 0 + } + #[cfg(not(unix))] + { + let _ = meta; + false + } +} + +/// The bytes and executable bit of the regular file at `path`, `None` when +/// nothing is there, and an error for anything else there. Symlinks are not +/// followed: fabric does not sync them, so it does not stage them either. +fn read_regular(path: &Path) -> Result, bool)>> { + let meta = match fs::symlink_metadata(path) { + Ok(meta) => meta, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => { + return Err(error).with_context(|| format!("failed to stat {}", path.display())); + } + }; + if !meta.is_file() { + bail!("{} is not a regular file", path.display()); + } + let bytes = fs::read(path).with_context(|| format!("failed to read {}", path.display()))?; + Ok(Some((bytes, is_executable(&meta)))) +} + +/// `target` relative to `folder` in manifest form, by the configured spelling +/// of the folder or its canonical one. `None` when `target` is not inside it, +/// or is the folder itself. +fn rel_inside(folder: &Path, target: &Path) -> Option { + let rel = target.strip_prefix(folder).ok().or_else(|| { + let canonical = folder.canonicalize().ok()?; + target.strip_prefix(canonical).ok() + })?; + Manifest::normalize_path(&rel.to_string_lossy()) +} + +/// Resolve `target` to exactly one entry, by folder and then by include. +pub fn resolve_target( + book: &SyncBook, + target: &Path, + entry_hint: Option<&str>, +) -> Result { + if !target.is_absolute() { + bail!("the target must be an absolute path, got {}", target.display()); + } + let inside: Vec<(&SyncEntry, String)> = book + .entries() + .iter() + .filter_map(|entry| rel_inside(&entry.folder, target).map(|rel| (entry, rel))) + .collect(); + if inside.is_empty() { + let folders = book + .entries() + .iter() + .map(|entry| format!("{} ({})", entry.folder.display(), entry.name)) + .collect::>() + .join(", "); + bail!( + "{} is not inside any synced folder; configured folders: {}", + target.display(), + if folders.is_empty() { + "none".to_string() + } else { + folders + } + ); + } + let resolved = |entry: &SyncEntry, rel: &str| ResolvedTarget { + entry: entry.clone(), + rel: rel.to_string(), + target: entry.folder.join(rel), + }; + let not_included = |entry: &SyncEntry, rel: &str| { + format!( + "no include glob of sync {:?} matches {rel}; its include list is {:?}, so publishing \ + would replicate nothing", + entry.name, + entry.include.clone().unwrap_or_default() + ) + }; + if let Some(hint) = entry_hint { + let Some((entry, rel)) = inside.iter().find(|(entry, _)| entry.name == hint) else { + bail!( + "{} is not inside the folder of sync {hint:?}", + target.display() + ); + }; + if !entry.includes(rel) { + bail!("{}", not_included(entry, rel)); + } + return Ok(resolved(entry, rel)); + } + let included: Vec<&(&SyncEntry, String)> = inside + .iter() + .filter(|(entry, rel)| entry.includes(rel)) + .collect(); + match included.as_slice() { + [] => { + let detail = inside + .iter() + .map(|(entry, rel)| not_included(entry, rel)) + .collect::>() + .join("; "); + bail!("{detail}") + } + [(entry, rel)] => Ok(resolved(entry, rel)), + many => { + let names = many + .iter() + .map(|(entry, _)| entry.name.as_str()) + .collect::>() + .join(", "); + bail!( + "{} belongs to more than one sync ({names}); pass --entry ", + target.display() + ) + } + } +} + +/// Refuse when the staging tree lies inside any synced folder. A file staged +/// there would publish, which is the one thing staging exists to prevent. +pub fn ensure_staging_outside_every_folder(home: &FabricHome, book: &SyncBook) -> Result<()> { + let staging = staging_root(home); + for entry in book.entries() { + if staging == entry.folder || rel_inside(&entry.folder, &staging).is_some() { + bail!( + "the staging tree {} lies inside the synced folder {} of sync {:?}; a file staged \ + there would publish, so nothing was staged", + staging.display(), + entry.folder.display(), + entry.name + ); + } + } + Ok(()) +} + +fn describe( + home: &FabricHome, + entry: &SyncEntry, + rel: &str, + record: Option<&SidecarRecord>, +) -> Result { + let staged_path = entry_staging_dir(home, &entry.name).join(rel); + let Some((bytes, executable)) = read_regular(&staged_path)? else { + bail!("{rel} is not staged for sync {:?}", entry.name); + }; + let target_path = entry.folder.join(rel); + let published_now = + read_regular(&target_path)?.map(|(published, _)| content_hash(&published).to_hex()); + Ok(StagedFile { + entry: entry.name.clone(), + rel: rel.to_string(), + staged_path, + target_path, + bytes: bytes.len() as u64, + hash: content_hash(&bytes).to_hex(), + executable, + base: record.and_then(|record| record.base.clone()), + staged_at: record.map(|record| record.staged_at).unwrap_or(0), + published_now, + }) +} + +/// Stage a change to `target`: seed the staged copy from `from`, else from the +/// published file, else empty, and record the published hash as the base. +pub fn stage( + home: &FabricHome, + book: &SyncBook, + target: &Path, + from: Option<&Path>, + entry_hint: Option<&str>, +) -> Result { + ensure_staging_outside_every_folder(home, book)?; + let resolved = resolve_target(book, target, entry_hint)?; + let published = read_regular(&resolved.target)?; + let (bytes, executable) = match from { + Some(from) => { + let Some(source) = read_regular(from)? else { + bail!("{} does not exist", from.display()); + }; + source + } + None => published.clone().unwrap_or_default(), + }; + let base = published.as_ref().map(|(bytes, _)| content_hash(bytes)); + + let dir = entry_staging_dir(home, &resolved.entry.name); + let staged_path = dir.join(&resolved.rel); + if let Some(parent) = staged_path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + write_atomic_with_mode(&staged_path, &bytes, executable)?; + let mut sidecar = load_sidecar(&dir)?; + sidecar.files.insert( + resolved.rel.clone(), + SidecarRecord { + base: base.map(ContentHash::to_hex), + staged_at: now_secs(), + }, + ); + save_sidecar(&dir, &sidecar)?; + describe( + home, + &resolved.entry, + &resolved.rel, + sidecar.files.get(&resolved.rel), + ) +} + +/// Every regular file under `dir`, in manifest form, minus the sidecar and any +/// temp file an interrupted write left behind. +fn staged_rels(dir: &Path) -> Result> { + fn walk(root: &Path, dir: &Path, out: &mut BTreeSet) -> Result<()> { + for child in fs::read_dir(dir).with_context(|| format!("failed to read {}", dir.display()))? { + let child = child?; + let path = child.path(); + let file_type = child.file_type()?; + if file_type.is_dir() { + walk(root, &path, out)?; + continue; + } + if !file_type.is_file() { + continue; + } + let Ok(rel) = path.strip_prefix(root) else { + continue; + }; + let Some(norm) = Manifest::normalize_path(&rel.to_string_lossy()) else { + continue; + }; + if norm == SIDECAR_NAME || norm.ends_with(TEMP_SUFFIX) { + continue; + } + out.insert(norm); + } + Ok(()) + } + let mut out = BTreeSet::new(); + if dir.is_dir() { + walk(dir, dir, &mut out)?; + } + Ok(out) +} + +/// Every staged file, for one entry or for all. A file dropped into the tree +/// by hand, with no record, is listed with no base, so it reads as stale +/// against any published file and publishes only when forced. +pub fn list(home: &FabricHome, book: &SyncBook, entry: Option<&str>) -> Result> { + let mut out = Vec::new(); + for configured in book.entries() { + if entry.is_some_and(|wanted| wanted != configured.name) { + continue; + } + let dir = entry_staging_dir(home, &configured.name); + let sidecar = load_sidecar(&dir)?; + for rel in staged_rels(&dir)? { + out.push(describe(home, configured, &rel, sidecar.files.get(&rel))?); + } + } + Ok(out) +} + +/// Read the staged bytes of `rels` in `entry`, or of every staged file of the +/// entry when `rels` is empty. Refuses a path the include no longer matches +/// and a target that is now a directory, before anything is written. +pub fn read_for_publish( + home: &FabricHome, + book: &SyncBook, + entry: &str, + rels: &[String], +) -> Result<(SyncEntry, Vec)> { + let Some(configured) = book.get(entry) else { + bail!("no sync entry named {entry:?}"); + }; + let dir = entry_staging_dir(home, entry); + let sidecar = load_sidecar(&dir)?; + let rels: Vec = if rels.is_empty() { + staged_rels(&dir)?.into_iter().collect() + } else { + rels.to_vec() + }; + if rels.is_empty() { + bail!("nothing is staged for sync {entry:?}"); + } + let mut files = Vec::with_capacity(rels.len()); + for rel in rels { + if !configured.includes(&rel) { + bail!( + "no include glob of sync {entry:?} matches {rel}; publishing would replicate nothing" + ); + } + let Some((bytes, executable)) = read_regular(&dir.join(&rel))? else { + bail!("{rel} is not staged for sync {entry:?}"); + }; + let target = configured.folder.join(&rel); + if target.is_dir() { + bail!( + "{} is a directory; a staged file cannot replace it", + target.display() + ); + } + let base = match sidecar.files.get(&rel).and_then(|record| record.base.as_deref()) { + Some(hex) => Some( + ContentHash::from_hex(hex) + .with_context(|| format!("{rel}: the recorded base is not a content hash"))?, + ), + None => None, + }; + files.push(PublishFile { + rel, + bytes, + executable, + base, + }); + } + Ok((configured.clone(), files)) +} + +/// Forget staged copies after they were published or discarded. Empty +/// directories go with them, and an empty entry tree goes entirely. +pub fn forget(home: &FabricHome, entry: &str, rels: &[String]) -> Result<()> { + let dir = entry_staging_dir(home, entry); + let mut sidecar = load_sidecar(&dir)?; + for rel in rels { + let path = dir.join(rel); + match fs::remove_file(&path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error).with_context(|| format!("failed to remove {}", path.display())); + } + } + sidecar.files.remove(rel); + let mut parent = path.parent(); + while let Some(current) = parent { + if current == dir || fs::remove_dir(current).is_err() { + break; + } + parent = current.parent(); + } + } + if staged_rels(&dir)?.is_empty() { + let _ = fs::remove_dir_all(&dir); + return Ok(()); + } + save_sidecar(&dir, &sidecar) +} + +fn hex_or_absent(hash: Option) -> String { + hash.map(ContentHash::to_hex) + .unwrap_or_else(|| "no file".to_string()) +} + +/// The one line a refused publish prints for one file. +pub fn refusal_line(rel: &str, base: Option, current: Option) -> String { + format!( + "{rel}: the published file changed since it was staged (staged against {}, now {}); \ + re-stage it or pass --force", + hex_or_absent(base), + hex_or_absent(current) + ) +} + +/// Publish without a daemon: check every base against the folder, then write +/// each file atomically into it. The daemon's next scan records them, one at a +/// time if its watcher splits them. +pub fn publish_locally( + entry: &SyncEntry, + files: &[PublishFile], + force: bool, +) -> Result> { + let mut refusals = Vec::new(); + for file in files { + let current = + read_regular(&entry.folder.join(&file.rel))?.map(|(bytes, _)| content_hash(&bytes)); + if current != file.base && !force { + refusals.push(refusal_line(&file.rel, file.base, current)); + } + } + if !refusals.is_empty() { + bail!("publish refused:\n{}", refusals.join("\n")); + } + let mut out = Vec::with_capacity(files.len()); + for file in files { + let target = entry.folder.join(&file.rel); + if let Some(parent) = target.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + write_atomic_with_mode(&target, &file.bytes, file.executable)?; + out.push((file.rel.clone(), content_hash(&file.bytes))); + } + Ok(out) +} diff --git a/tests/sync_slice.rs b/tests/sync_slice.rs index 260d6c7..b99363c 100644 --- a/tests/sync_slice.rs +++ b/tests/sync_slice.rs @@ -235,6 +235,106 @@ async fn bus_update_beats_equal_version_delete_then_archive_survives_restart() - Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_staged_file_does_not_reach_a_peer_until_published() -> Result<()> { + use fabric::control::SyncPublishFile; + use fabric::sync::{SyncBook, staging}; + + let _guard = SYNC_SLICE_LOCK.lock().await; + let a_dir = TempDir::new()?; + let b_dir = TempDir::new()?; + let a_home = FabricHome::new(a_dir.path()); + let b_home = FabricHome::new(b_dir.path()); + let a_bus = a_dir.path().join("bus"); + let b_bus = b_dir.path().join("bus"); + std::fs::create_dir_all(&a_bus)?; + std::fs::create_dir_all(&b_bus)?; + write_sync(a_dir.path(), &a_bus, "bus"); + write_sync(b_dir.path(), &b_bus, "bus"); + + let node_a = FabricNode::start(a_home.clone()).await?; + let node_b = FabricNode::start(b_home.clone()).await?; + trust_peer(&a_home, &node_a, node_b.id(), "node-b", node_b.addr()).await?; + trust_peer(&b_home, &node_b, node_a.id(), "node-a", node_a.addr()).await?; + + std::fs::write(a_bus.join("seed.md"), b"seed")?; + reload_sync(&a_home).await?; + assert!( + wait_for_file(&b_bus.join("seed.md"), b"seed").await, + "the pair did not converge on the seed" + ); + + // Stage on A. The staged copy lives in A's fabric home, not in the folder. + let book_a = SyncBook::load(&a_home)?; + let staged = staging::stage(&a_home, &book_a, &a_bus.join("draft.md"), None, None)?; + std::fs::write(&staged.staged_path, b"held")?; + assert!(!staged.staged_path.starts_with(&a_bus)); + + // A control file written into the folder crosses in the same window. That + // proves the window carried a file, so the staged file's absence on B is + // about staging and not about a quiet pair. + reload_sync(&a_home).await?; + reload_sync(&b_home).await?; + std::fs::write(a_bus.join("control-one.md"), b"crosses")?; + assert!( + wait_for_file(&b_bus.join("control-one.md"), b"crosses").await, + "the control file did not cross, so this window proves nothing" + ); + assert!(!b_bus.join("draft.md").exists(), "the staged file reached the peer"); + assert_stays_missing(&b_bus.join("draft.md")).await; + + // Restart A with the file still staged. Nothing may publish it. + node_a.shutdown().await?; + let node_a = FabricNode::start(a_home.clone()).await?; + trust_peer(&a_home, &node_a, node_b.id(), "node-b", node_b.addr()).await?; + trust_peer(&b_home, &node_b, node_a.id(), "node-a", node_a.addr()).await?; + reload_sync(&a_home).await?; + std::fs::write(a_bus.join("control-two.md"), b"crosses again")?; + assert!( + wait_for_file(&b_bus.join("control-two.md"), b"crosses again").await, + "the second control file did not cross after the restart" + ); + assert!( + !b_bus.join("draft.md").exists(), + "the restart published the staged file" + ); + assert_stays_missing(&b_bus.join("draft.md")).await; + assert!(!a_bus.join("draft.md").exists()); + + // Publish through the daemon. The peer receives exactly the reviewed bytes. + let (_entry, files) = staging::read_for_publish(&a_home, &book_a, "shared", &[])?; + let request = ControlRequest::SyncPublish { + name: "shared".to_string(), + files: files + .iter() + .map(|file| SyncPublishFile { + rel: file.rel.clone(), + bytes: file.bytes.clone(), + executable: file.executable, + base: file.base.map(|hash| hash.to_hex()), + }) + .collect(), + force: false, + }; + let ControlResponse::SyncPublished { files: published } = send_control(&a_home, request).await? + else { + anyhow::bail!("unexpected response to SyncPublish"); + }; + assert_eq!(published.len(), 1); + assert_eq!(published[0].rel, "draft.md"); + assert_eq!(published[0].version, 1); + staging::forget(&a_home, "shared", &["draft.md".to_string()])?; + assert!( + wait_for_file(&b_bus.join("draft.md"), b"held").await, + "the published file did not reach the peer" + ); + assert!(!staged.staged_path.exists(), "a published file must leave the staging tree"); + + node_b.shutdown().await?; + node_a.shutdown().await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn production_status_exposes_exact_inbound_scan_ledger() -> Result<()> { let _guard = SYNC_SLICE_LOCK.lock().await;