diff --git a/AGENTS.md b/AGENTS.md index 201ab4a..0603d6a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,6 +25,11 @@ Impure shell around pure decision cores: - Each tick: `local_snapshot` + `index_snapshot` → pure `reconcile()` → `Vec` {Import, WriteFile, DeleteLocal, Tombstone, Merge} → execute → settle into `SyncState` (atomic temp+rename to `data_dir/state.toml`). +- The `Wiremap` (ssync-core `wiremap.rs`) owns the adapters, the per-agent excludes, and + the path-map resolver: every wire-key ↔ local-path translation and every freeze verdict + (excluded key #14, dropped-agent guard, failed mapping #49) lives behind it — Engine + never touches adapters directly, and the freeze invariants are unit-tested there + without iroh. - Import path: file → age-encrypt (subprocess) → `Node::publish` (blob + index entry keyed `{agent}/{relative_path}`; the ONLY write path — temp-tag prevents the GC race). - Export path: index entry → `Node::blob` (local miss → bounded fetch from peers behind diff --git a/crates/ssync-core/src/cleanup.rs b/crates/ssync-core/src/cleanup.rs index d7e70e7..7c3a081 100644 --- a/crates/ssync-core/src/cleanup.rs +++ b/crates/ssync-core/src/cleanup.rs @@ -69,7 +69,7 @@ pub fn plan(adapters: &[Box], filter: &Filter) -> Result>, + wiremap: Wiremap, identity: AgeIdentity, node: Node, /// What we last materialised per key; feeds [`reconcile`]. @@ -66,11 +67,6 @@ pub struct Engine { rotation_pending: bool, /// Import failures in the current pass; a rotation only settles at zero. import_errors: usize, - /// Per-agent session exclusion patterns (issue #14); a matching key is - /// invisible to reconcile from both sides, freezing it everywhere. - excludes: std::collections::HashMap>, - /// Wire↔local path-map translation (issue #13, map #42); default = inert. - resolver: Resolver, } impl Engine { @@ -85,7 +81,7 @@ impl Engine { ) -> Self { let recipients_fp = Hash::new(identity.recipients().join("\n")).to_string(); Self { - adapters, + wiremap: Wiremap::new(adapters), identity, node, state: SyncState::default(), @@ -99,8 +95,6 @@ impl Engine { recipients_fp, rotation_pending: false, import_errors: 0, - excludes: Default::default(), - resolver: Resolver::default(), } } @@ -122,12 +116,12 @@ impl Engine { /// Per-agent `exclude` patterns from config (`[[agents]]` tables). pub fn set_excludes(&mut self, excludes: std::collections::HashMap>) { - self.excludes = excludes; + self.wiremap.set_excludes(excludes); } /// The `[[path_map]]` + `canonical_home` from config (issue #13). pub fn set_path_map(&mut self, map: PathMap, canonical_home: Option) { - self.resolver = Resolver::new(map, canonical_home); + self.wiremap.set_path_map(map, canonical_home); } /// Live index keys (wire form) — status/debug surface; the two-node @@ -146,84 +140,6 @@ impl Engine { self.node.shutdown().await } - /// The adapter owning an index key (matching `{agent}/` prefix), if any — - /// peers may sync agents this node does not have configured. - fn adapter_of_key(&self, key: &str) -> Option<&dyn Adapter> { - self.adapter_index_of_key(key) - .map(|i| self.adapters[i].as_ref()) - } - - /// The adapter whose session root contains `path`. - fn adapter_of_path(&self, path: &Path) -> Option<&dyn Adapter> { - self.adapters - .iter() - .map(|a| a.as_ref()) - .find(|a| path.starts_with(a.session_root())) - } - - /// The iroh-docs index key for a session: `{agent}/{relative_path}`. The - /// relative path is machine-independent and carries the write-back location, - /// so the exporter can reconstruct where the file belongs on any peer. - fn index_key(&self, id: &SessionIdentity) -> String { - format!("{}/{}", id.agent, id.relative_path.display()) - } - - /// The wire key for a local file (issue #13): translated through the - /// path map when its project dir is mapped, today's relative path - /// otherwise. `Err` = unresolvable this tick; the caller skips the file - /// (logged once) and a later pass retries. - fn mapped_key(&mut self, id: &SessionIdentity, path: &Path) -> Result { - let Some(idx) = self.adapter_index_of_path(path) else { - return Ok(self.index_key(id)); - }; - let rel = id.relative_path.to_string_lossy(); - let adapter = self.adapters[idx].as_ref(); - match self.resolver.wire_rel(adapter, &rel)? { - None => Ok(self.index_key(id)), - Some(mapped) => Ok(format!("{}/{mapped}", id.agent)), - } - } - - /// Local destination + on-disk bytes for a wire key's plaintext, through - /// the path map. `Ok(None)` = not resolvable yet — the caller returns - /// false and a later tick retries. - fn localize(&mut self, key: &str, plaintext: Vec) -> Result)>> { - let Some(idx) = self.adapter_index_of_key(key) else { - return Ok(None); - }; - let adapter = self.adapters[idx].as_ref(); - let Some(rel) = key - .strip_prefix(adapter.agent()) - .and_then(|r| r.strip_prefix('/')) - else { - return Ok(None); - }; - self.resolver.localize(adapter, rel, plaintext) - } - - /// [`dest_of`](Self::dest_of) through the path map (for actions without - /// plaintext in hand: DeleteLocal). `None` = nothing local to touch. - fn local_dest_of(&self, key: &str) -> Option { - let adapter = self.adapter_of_key(key)?; - let rel = key.strip_prefix(adapter.agent())?.strip_prefix('/')?; - self.resolver.local_dest_of(adapter, rel) - } - - /// Index of the adapter whose session root contains `path`. - fn adapter_index_of_path(&self, path: &Path) -> Option { - self.adapters - .iter() - .position(|a| path.starts_with(a.session_root())) - } - - /// Index of the adapter owning an index key (matching `{agent}/` prefix). - fn adapter_index_of_key(&self, key: &str) -> Option { - self.adapters.iter().position(|a| { - key.strip_prefix(a.agent()) - .is_some_and(|r| r.starts_with('/')) - }) - } - /// Read → encrypt → blob → upsert index. Dedups on *plaintext* (age /// ciphertext is randomized), or the exporter's own write-back echoes. /// `force` (recipient-set rotation) bypasses the dedup: the plaintext is @@ -238,7 +154,7 @@ impl Engine { let plaintext = tokio::fs::read(path) .await .with_context(|| format!("reading session file {}", path.display()))?; - let plaintext = self.canonical_plaintext(key, plaintext)?; + let plaintext = self.wiremap.canonical_plaintext(key, plaintext)?; if !force && let Some(w) = winner && self.get_plain(w).await.as_deref() == Some(&plaintext) @@ -250,23 +166,6 @@ impl Engine { Ok(ImportOutcome::Published(hash)) } - /// The wire form of a local file's bytes: header cwd mapped to canonical - /// (issue #13). A machine-local path must never reach the index — an - /// unmappable header is a per-key error, not a pass-through (#49). - fn canonical_plaintext(&self, key: &str, bytes: Vec) -> Result> { - let Some(adapter) = self.adapter_of_key(key) else { - return Ok(bytes); - }; - self.resolver.canonical_plaintext(adapter, bytes) - } - - /// Identify a path via the adapter whose session root contains it. - fn identify(&self, path: &Path) -> Result { - self.adapter_of_path(path) - .ok_or_else(|| anyhow!("{} is under no configured session root", path.display()))? - .identify(path) - } - /// The divergence verdict for a key, decrypting whatever the cache cannot /// answer. Decryption stops at the first unavailable blob; the resulting /// short set reads as incomplete in [`Divergence::verdict`]. @@ -297,13 +196,10 @@ impl Engine { continue; // deleted — never resurrect from stale live entries }; let key = String::from_utf8(rec.key).context("index key not utf-8")?; - if self.excluded_key(&key) { + if self.wiremap.excluded(&key) { continue; // frozen keys neither count nor report conflicts } - match self - .dest_of(&key) - .and_then(|dest| self.adapter_of_key(&key)?.identify(&dest).ok()) - { + match self.wiremap.session_identity_of_key(&key) { Some(id) => sessions.insert((id.agent, id.project_id, id.session_id)), // unconfigured agent or unparseable path: the key is the session None => sessions.insert((key.clone(), String::new(), String::new())), @@ -311,7 +207,7 @@ impl Engine { if rec.versions.len() <= 1 { continue; } - let Some(rel) = self.relative_of(&key).map(str::to_string) else { + let Some(rel) = self.wiremap.relative_of(&key).map(str::to_string) else { continue; }; let diverged = match self.divergence.cached(&key, &rec.versions) { @@ -352,7 +248,7 @@ impl Engine { /// it, so a settled key costs one lookup. Returns the relative path when /// it published. async fn merge_one(&self, key: &str) -> Result> { - let Some(rel) = self.relative_of(key).map(str::to_string) else { + let Some(rel) = self.wiremap.relative_of(key).map(str::to_string) else { return Ok(None); }; let Some(rec) = self.node.index_record(key).await? else { @@ -376,43 +272,6 @@ impl Engine { Ok(Some(rel)) } - /// Decode an index key back to its session-root-relative path (the inverse of - /// `index_key`): strip the `{agent}/` prefix of a configured adapter. - fn relative_of<'a>(&self, key: &'a str) -> Option<&'a str> { - let adapter = self.adapter_of_key(key)?; - key.strip_prefix(adapter.agent())?.strip_prefix('/') - } - - /// Whether a key falls under its agent's `exclude` patterns (issue #14). - /// Filtered out of BOTH reconcile inputs, so the key is frozen: never - /// imported, exported, tombstoned, or merged — here or on write-back. - fn excluded_key(&self, key: &str) -> bool { - let Some(adapter) = self.adapter_of_key(key) else { - return false; - }; - let Some(patterns) = self.excludes.get(adapter.agent()) else { - return false; - }; - self.relative_of(key) - .is_some_and(|rel| exclude::is_excluded(patterns, rel)) - } - - /// The absolute session-dir path an index key maps to on this machine, or - /// `None` when no configured adapter owns the key. - fn dest_of(&self, key: &str) -> Option { - let adapter = self.adapter_of_key(key)?; - let rel = key.strip_prefix(adapter.agent())?.strip_prefix('/')?; - Some(adapter.session_root().join(rel)) - } - - /// Session files under every configured adapter's root. - fn all_session_files(&self) -> Vec { - self.adapters - .iter() - .flat_map(|a| session_files(a.session_root(), a.as_ref())) - .collect() - } - /// Write the status snapshot; `announce` additionally logs conflicts (off /// for the periodic liveness refresh, which would repeat them every tick). async fn write_status(&mut self, path: &Path, announce: bool) { @@ -431,13 +290,13 @@ impl Engine { /// Snapshot every configured session dir as `reconcile` input. fn local_snapshot(&mut self) -> Vec { - self.resolver.begin_pass(); + self.wiremap.begin_pass(); let mut out = Vec::new(); - for path in self.all_session_files() { + for path in self.wiremap.session_files() { let Some(stamp) = file_stamp_micros(&path) else { continue; }; - let id = match self.identify(&path) { + let id = match self.wiremap.identify(&path) { Ok(id) => id, Err(e) => { // per-key errors are logged, never silently dropped @@ -447,20 +306,16 @@ impl Engine { continue; } }; - let key = match self.mapped_key(&id, &path) { - Ok(k) => k, - Err(e) => { - // an unresolvable mapping FREEZES the agent's tombstones - // this pass (#49): a skip that read as deletion would - // propagate mesh-wide (same class as the #14 fix) - let dir = path.parent().unwrap_or(&path).to_path_buf(); - if self.resolver.note_failure(&id.agent, dir) { - eprintln!("ssync: skipping {}: {e:#}", path.display()); + let key = match self.wiremap.key_of(&id, &path) { + KeyLookup::Key(key) => key, + KeyLookup::Skipped(announcement) => { + if let Some(error) = announcement { + eprintln!("ssync: skipping {}: {error:#}", path.display()); } continue; } }; - if self.excluded_key(&key) { + if self.wiremap.excluded(&key) { continue; } out.push(LocalFile { key, path, stamp }); @@ -474,13 +329,10 @@ impl Engine { let mut out = std::collections::HashMap::new(); for rec in self.node.index_records().await? { let key = String::from_utf8(rec.key).context("index key not utf-8")?; - // No adapter = frozen, exactly like an excluded key: a dropped - // `[[agents]]` entry must never tombstone peers' sessions (the - // key would read as "materialised here, now gone"). - if self.adapter_of_key(&key).is_none() || self.excluded_key(&key) { + if self.wiremap.frozen(&key) { continue; } - let merge_allowed = self.adapter_of_key(&key).is_some_and(|a| a.append_only()); + let merge_allowed = self.wiremap.append_only(&key); out.insert( key, IndexEntry { @@ -526,12 +378,8 @@ impl Engine { }; let mut changed = false; for action in reconcile(&self.state, &local, &index) { - // withhold tombstones for agents whose mapping failed this pass: - // the "file gone" reading is an artifact of the skip, not a delete if let Action::Tombstone { key } = &action - && self - .adapter_of_key(key) - .is_some_and(|a| self.resolver.agent_failed(a.agent())) + && self.wiremap.tombstone_withheld(key) { continue; } @@ -616,7 +464,7 @@ impl Engine { return false; } }; - let (dest, plaintext) = match self.localize(key, plaintext) { + let (dest, plaintext) = match self.wiremap.localize(key, plaintext) { Ok(Some(pair)) => pair, Ok(None) => return false, // unresolved yet — retried Err(e) => { @@ -634,14 +482,14 @@ impl Engine { } Action::DeleteLocal { key } => { self.state.settle_delete(key); - let Some(dest) = self.local_dest_of(key) else { + let Some(dest) = self.wiremap.local_dest_of(key) else { return false; }; let existed = dest.exists(); let _ = tokio::fs::remove_file(&dest).await; // sweep the emptied artifact dir the deletion may leave behind - if let Some(adapter) = self.adapter_of_key(key) { - cleanup::remove_empty_parents(&dest, adapter.session_root()); + if let Some(root) = self.wiremap.session_root_of(key) { + cleanup::remove_empty_parents(&dest, root); } existed } @@ -696,10 +544,10 @@ impl Engine { let mut watcher = notify::recommended_watcher(move |res| { let _ = tx.send(res); })?; - for adapter in &self.adapters { + for root in self.wiremap.roots() { watcher - .watch(adapter.session_root(), RecursiveMode::Recursive) - .with_context(|| format!("watching {}", adapter.session_root().display()))?; + .watch(root, RecursiveMode::Recursive) + .with_context(|| format!("watching {}", root.display()))?; } // Doc events are drained behind the Node seam (Node::signals): only @@ -732,9 +580,7 @@ impl Engine { tokio::select! { Some(res) = rx.recv() => { if let Ok(event) = res - && event.paths.iter().any(|p| { - self.adapter_of_path(p).is_some_and(|a| a.is_session_file(p)) - }) { + && event.paths.iter().any(|p| self.wiremap.is_session_file(p)) { deadline = Some(tokio::time::Instant::now() + DEBOUNCE); } } @@ -798,26 +644,6 @@ fn file_stamp_micros(path: &Path) -> Option<(u64, u64)> { Some((mtime, m.len())) } -/// Recursively collect session files under `root` accepted by `adapter`. -fn session_files(root: &Path, adapter: &dyn Adapter) -> Vec { - let mut out = Vec::new(); - let mut stack = vec![root.to_path_buf()]; - while let Some(dir) = stack.pop() { - let Ok(entries) = std::fs::read_dir(&dir) else { - continue; - }; - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - stack.push(path); - } else if adapter.is_session_file(&path) { - out.push(path); - } - } - } - out -} - /// Conflicts to announce this pass: those in `current` not yet logged. The /// caller replaces its logged set with `current` afterwards, so a conflict /// that resolved and re-diverged announces again. @@ -882,9 +708,13 @@ mod tests { assert!(engine.tick_once().await, "first tick must import"); assert!(!engine.tick_once().await, "second tick must be a no-op"); + let id = engine.wiremap.identify(&session_path).unwrap(); + let KeyLookup::Key(key) = engine.wiremap.key_of(&id, &session_path) else { + panic!("session key lookup skipped"); + }; let rec = engine .node - .index_record(engine.index_key(&engine.identify(&session_path).unwrap())) + .index_record(key) .await .unwrap() .expect("session indexed"); diff --git a/crates/ssync-core/src/search.rs b/crates/ssync-core/src/search.rs index 32ffdcb..d484e0d 100644 --- a/crates/ssync-core/src/search.rs +++ b/crates/ssync-core/src/search.rs @@ -42,7 +42,7 @@ pub fn search(adapters: &[Box], query: &str, agent: Option<&str>) - if agent.is_some_and(|a| a != adapter.agent()) { continue; } - for path in crate::session_files(adapter.session_root(), adapter.as_ref()) { + for path in crate::wiremap::session_files(adapter.session_root(), adapter.as_ref()) { let Ok(id) = adapter.identify(&path) else { continue; }; diff --git a/crates/ssync-core/src/wiremap.rs b/crates/ssync-core/src/wiremap.rs new file mode 100644 index 0000000..1bde621 --- /dev/null +++ b/crates/ssync-core/src/wiremap.rs @@ -0,0 +1,365 @@ +//! The wire-map: every translation between wire keys (`{agent}/{relative_path}` +//! index keys) and local session paths, plus the freeze verdicts that keep a +//! skipped key from ever reading as a deletion. Owns the adapters, the +//! per-agent excludes (#14), and the path-map [`Resolver`] (#13/#49); the +//! engine holds one `Wiremap` and never touches adapters directly. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +use anyhow::{Result, anyhow}; +use ssync_adapters::{Adapter, SessionIdentity}; + +use crate::exclude; +use crate::pathmap::{PathMap, Resolver}; + +/// Result of resolving a local session to its wire key. +pub(crate) enum KeyLookup { + Key(String), + /// Retriable per-key skip (#49); `Some` is the first announcement. + Skipped(Option), +} + +/// Wire key ↔ local path translation and freeze state for one engine. +pub(crate) struct Wiremap { + adapters: Vec>, + /// Per-agent session exclusion patterns (issue #14); a matching key is + /// invisible to reconcile from both sides, freezing it everywhere. + excludes: HashMap>, + /// Wire↔local path-map translation (issue #13, map #42); default = inert. + resolver: Resolver, +} + +impl Wiremap { + pub fn new(adapters: Vec>) -> Self { + Self { + adapters, + excludes: HashMap::new(), + resolver: Resolver::default(), + } + } + + /// Per-agent `exclude` patterns from config (`[[agents]]` tables). + pub fn set_excludes(&mut self, excludes: HashMap>) { + self.excludes = excludes; + } + + /// The `[[path_map]]` + `canonical_home` from config (issue #13). + pub fn set_path_map(&mut self, map: PathMap, canonical_home: Option) { + self.resolver = Resolver::new(map, canonical_home); + } + + /// Start a snapshot pass: last pass's mapping failures no longer freeze. + pub fn begin_pass(&mut self) { + self.resolver.begin_pass(); + } + + /// Every configured session root (watch targets). + pub fn roots(&self) -> impl Iterator { + self.adapters.iter().map(|a| a.session_root()) + } + + /// Whether `path` is a session file of the adapter whose root contains it. + pub fn is_session_file(&self, path: &Path) -> bool { + self.adapter_of_path(path) + .is_some_and(|a| a.is_session_file(path)) + } + + /// Session files under every configured adapter's root. + pub fn session_files(&self) -> Vec { + self.adapters + .iter() + .flat_map(|a| session_files(a.session_root(), a.as_ref())) + .collect() + } + + /// Identify a path via the adapter whose session root contains it. + pub fn identify(&self, path: &Path) -> Result { + self.adapter_of_path(path) + .ok_or_else(|| anyhow!("{} is under no configured session root", path.display()))? + .identify(path) + } + + /// The wire key for a local file (issue #13): translated through the path + /// map when its project dir is mapped, today's relative path otherwise. + /// Mapping failures are retriable per-key skips (#49). + pub fn key_of(&mut self, id: &SessionIdentity, path: &Path) -> KeyLookup { + let Some(idx) = self.adapter_index_of_path(path) else { + return KeyLookup::Key(index_key(id)); + }; + let rel = id.relative_path.to_string_lossy(); + match self.resolver.wire_rel(self.adapters[idx].as_ref(), &rel) { + Ok(None) => KeyLookup::Key(index_key(id)), + Ok(Some(mapped)) => KeyLookup::Key(format!("{}/{mapped}", id.agent)), + Err(error) => { + let dir = path.parent().unwrap_or(path).to_path_buf(); + let announcement = self.resolver.note_failure(&id.agent, dir).then_some(error); + KeyLookup::Skipped(announcement) + } + } + } + + /// Local destination + on-disk bytes for a wire key's plaintext, through + /// the path map. `Ok(None)` = not resolvable yet — the caller returns + /// false and a later tick retries. + pub fn localize( + &mut self, + key: &str, + plaintext: Vec, + ) -> Result)>> { + let Some((idx, rel)) = self.key_parts(key) else { + return Ok(None); + }; + let adapter = self.adapters[idx].as_ref(); + self.resolver.localize(adapter, rel, plaintext) + } + + /// Local destination through the path map for actions without plaintext + /// in hand (DeleteLocal). `None` = nothing local to touch. + pub fn local_dest_of(&self, key: &str) -> Option { + let (idx, rel) = self.key_parts(key)?; + self.resolver + .local_dest_of(self.adapters[idx].as_ref(), rel) + } + + /// The wire form of a local file's bytes: header cwd mapped to canonical + /// (issue #13). A machine-local path must never reach the index — an + /// unmappable header is a per-key error, not a pass-through (#49). + pub fn canonical_plaintext(&self, key: &str, bytes: Vec) -> Result> { + let Some(adapter) = self.adapter_of_key(key) else { + return Ok(bytes); + }; + self.resolver.canonical_plaintext(adapter, bytes) + } + + /// The session root of the adapter owning `key` (empty-parent sweep after + /// DeleteLocal). + pub fn session_root_of(&self, key: &str) -> Option<&Path> { + self.adapter_of_key(key).map(|a| a.session_root()) + } + + /// Decode an index key back to its session-root-relative path (the + /// inverse of the key encoding): strip the `{agent}/` prefix of a + /// configured adapter. + pub fn relative_of<'a>(&self, key: &'a str) -> Option<&'a str> { + self.key_parts(key).map(|(_, rel)| rel) + } + + /// The session identity an index key resolves to on this machine, if a + /// configured adapter can parse its destination path. + pub fn session_identity_of_key(&self, key: &str) -> Option { + let (idx, rel) = self.key_parts(key)?; + let adapter = self.adapters[idx].as_ref(); + adapter.identify(&adapter.session_root().join(rel)).ok() + } + + /// Whether a key falls under its agent's `exclude` patterns (issue #14). + /// Filtered out of BOTH reconcile inputs, so the key is frozen: never + /// imported, exported, tombstoned, or merged — here or on write-back. + pub fn excluded(&self, key: &str) -> bool { + let Some((idx, rel)) = self.key_parts(key) else { + return false; + }; + self.excluded_parts(self.adapters[idx].agent(), rel) + } + + /// Whether a key is invisible to reconcile: excluded (#14) or owned by no + /// configured adapter (dropped-agent guard — a removed `[[agents]]` entry + /// must never tombstone peers' sessions). + pub fn frozen(&self, key: &str) -> bool { + let Some((idx, rel)) = self.key_parts(key) else { + return true; + }; + self.excluded_parts(self.adapters[idx].agent(), rel) + } + + /// Whether a tombstone for this key must be withheld this pass: its + /// agent's mapping failed (#49), so "file gone" is an artifact of the + /// skip, not a delete. + pub fn tombstone_withheld(&self, key: &str) -> bool { + self.adapter_of_key(key) + .is_some_and(|a| self.resolver.agent_failed(a.agent())) + } + + /// Whether the key's format merges (append-only line union) rather than + /// newest-wins. `false` for keys of unconfigured agents. + pub fn append_only(&self, key: &str) -> bool { + self.adapter_of_key(key).is_some_and(|a| a.append_only()) + } + + /// The adapter owning an index key (matching `{agent}/` prefix), if any — + /// peers may sync agents this node does not have configured. + fn adapter_of_key(&self, key: &str) -> Option<&dyn Adapter> { + let (idx, _) = self.key_parts(key)?; + Some(self.adapters[idx].as_ref()) + } + + /// The adapter whose session root contains `path`. + fn adapter_of_path(&self, path: &Path) -> Option<&dyn Adapter> { + self.adapter_index_of_path(path) + .map(|i| self.adapters[i].as_ref()) + } + + fn adapter_index_of_path(&self, path: &Path) -> Option { + self.adapters + .iter() + .position(|a| path.starts_with(a.session_root())) + } + + fn key_parts<'a>(&self, key: &'a str) -> Option<(usize, &'a str)> { + self.adapters.iter().enumerate().find_map(|(idx, adapter)| { + key.strip_prefix(adapter.agent()) + .and_then(|rest| rest.strip_prefix('/')) + .map(|rel| (idx, rel)) + }) + } + + fn excluded_parts(&self, agent: &str, rel: &str) -> bool { + self.excludes + .get(agent) + .is_some_and(|patterns| exclude::is_excluded(patterns, rel)) + } +} + +/// The iroh-docs index key for a session: `{agent}/{relative_path}`. The +/// relative path is machine-independent and carries the write-back location, +/// so the exporter can reconstruct where the file belongs on any peer. +fn index_key(id: &SessionIdentity) -> String { + format!("{}/{}", id.agent, id.relative_path.display()) +} + +/// Recursively collect session files under `root` accepted by `adapter`. +pub(crate) fn session_files(root: &Path, adapter: &dyn Adapter) -> Vec { + let mut out = Vec::new(); + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&dir) else { + continue; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + stack.push(path); + } else if adapter.is_session_file(&path) { + out.push(path); + } + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use ssync_adapters::blob_store::BlobStoreAdapter; + use ssync_adapters::pi::PiAdapter; + + fn scratch(tag: &str) -> PathBuf { + let p = std::env::temp_dir().join(format!("ssync-wiremap-{tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&p); + std::fs::create_dir_all(&p).unwrap(); + p + } + + fn pi_map(root: &Path) -> Wiremap { + Wiremap::new(vec![Box::new(PiAdapter::new("pi", root))]) + } + + fn id(agent: &str, rel: &str) -> SessionIdentity { + SessionIdentity { + agent: agent.into(), + session_id: "s".into(), + project_id: "p".into(), + relative_path: rel.into(), + } + } + + #[test] + fn key_and_dest_round_trip_without_map() { + let root = scratch("roundtrip"); + let mut wm = pi_map(&root); + let rel = "--proj--/s.jsonl"; + let KeyLookup::Key(key) = wm.key_of(&id("pi", rel), &root.join(rel)) else { + panic!("key lookup skipped"); + }; + assert_eq!(key, format!("pi/{rel}")); + assert_eq!(wm.local_dest_of(&key), Some(root.join(rel))); + assert_eq!(wm.relative_of(&key), Some(rel)); + } + + #[test] + fn excluded_keys_are_frozen_but_foreign_agents_are_not_excluded() { + let root = scratch("exclude"); + let mut wm = pi_map(&root); + wm.set_excludes(HashMap::from([( + "pi".to_string(), + vec!["*secret*".to_string()], + )])); + assert!(wm.excluded("pi/--proj--/secret.jsonl")); + assert!(wm.frozen("pi/--proj--/secret.jsonl")); + assert!(!wm.excluded("pi/--proj--/s.jsonl")); + // patterns bind per agent; a key of an unconfigured agent is not + // excluded (it freezes via the dropped-agent guard instead) + assert!(!wm.excluded("ghost/--proj--/secret.jsonl")); + } + + #[test] + fn dropped_agent_keys_are_frozen() { + let root = scratch("dropped"); + let wm = pi_map(&root); + assert!(wm.frozen("ghost/--proj--/s.jsonl")); + assert!(!wm.frozen("pi/--proj--/s.jsonl")); + assert_eq!(wm.local_dest_of("ghost/--proj--/s.jsonl"), None); + } + + #[test] + fn mapping_failure_withholds_tombstones_and_announces_once() { + let root = scratch("freeze"); + let mut wm = pi_map(&root); + wm.set_path_map( + PathMap::new(vec![("/data".into(), "/canon".into())]).unwrap(), + None, + ); + // project dir exists but holds no session header to resolve the cwd + let dir = root.join("--proj--"); + std::fs::create_dir_all(&dir).unwrap(); + let rel = "--proj--/s.jsonl"; + let sid = id("pi", rel); + let path = root.join(rel); + + wm.begin_pass(); + let KeyLookup::Skipped(first) = wm.key_of(&sid, &path) else { + panic!("mapping unexpectedly succeeded"); + }; + assert!(first.is_some(), "first failure logs"); + assert!(wm.tombstone_withheld("pi/--proj--/other.jsonl")); + let KeyLookup::Skipped(second) = wm.key_of(&sid, &path) else { + panic!("mapping unexpectedly succeeded"); + }; + assert!(second.is_none(), "repeat failure stays quiet"); + // the freeze lasts one pass; the log-once memory does not reset + wm.begin_pass(); + assert!(!wm.tombstone_withheld("pi/--proj--/other.jsonl")); + } + + #[test] + fn append_only_follows_the_owning_adapter() { + let root = scratch("append"); + let blob_root = scratch("append-blobs"); + let wm = Wiremap::new(vec![ + Box::new(PiAdapter::new("pi", &root)), + Box::new(BlobStoreAdapter::new("omp-blobs", &blob_root)), + ]); + assert!(wm.append_only("pi/--proj--/s.jsonl")); + assert!(!wm.append_only("omp-blobs/abc123")); + assert!(!wm.append_only("ghost/x")); + } + + #[test] + fn is_session_file_requires_an_owning_root() { + let root = scratch("owning"); + let wm = pi_map(&root); + assert!(wm.is_session_file(&root.join("--proj--/s.jsonl"))); + assert!(!wm.is_session_file(Path::new("/elsewhere/--proj--/s.jsonl"))); + } +} diff --git a/crates/ssync-net/src/lib.rs b/crates/ssync-net/src/lib.rs index 7020911..f74f2e9 100644 --- a/crates/ssync-net/src/lib.rs +++ b/crates/ssync-net/src/lib.rs @@ -689,6 +689,17 @@ impl Node { mod tests { use super::*; + async fn direct_addr_of(node: &Node) -> EndpointAddr { + for _ in 0..40 { + let addr = node.endpoint_addr(); + if !addr.addrs.is_empty() { + return addr; + } + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + panic!("node {} has no direct addresses", node.endpoint_addr().id); + } + #[test] fn parse_peer_addrs_trims_and_skips_junk() { let id1 = node_id_of(&generate_key_bytes()); @@ -793,22 +804,8 @@ mod tests { let secret = generate_key_bytes(); a.open_shared_namespace(secret).await.unwrap(); b.open_shared_namespace(secret).await.unwrap(); - // without a relay fallback an addr is dialable only once the bound - // socket addresses show up; poll instead of assuming bind ordering. - let addr_of = |n: &Node| { - let addr = n.endpoint_addr(); - (!addr.addrs.is_empty()).then_some(addr) - }; - let (mut addr_a, mut addr_b) = (addr_of(&a), addr_of(&b)); - for _ in 0..40 { - if addr_a.is_some() && addr_b.is_some() { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(250)).await; - (addr_a, addr_b) = (addr_of(&a), addr_of(&b)); - } - let addr_a = addr_a.expect("lan-only node a has no direct addrs"); - let addr_b = addr_b.expect("lan-only node b has no direct addrs"); + let addr_a = direct_addr_of(&a).await; + let addr_b = direct_addr_of(&b).await; a.sync_with(vec![addr_b]).await.unwrap(); b.sync_with(vec![addr_a]).await.unwrap(); @@ -877,21 +874,22 @@ mod tests { #[tokio::test] async fn peer_paths_observe_active_direct_path_once_connected() { let (da, db) = (tempdir("paths-a"), tempdir("paths-b")); - let mut a = Node::spawn(&da, SecretKey::generate()).await.unwrap(); - let mut b = Node::spawn(&db, SecretKey::generate()).await.unwrap(); + let mut a = Node::spawn_lan_only(&da, SecretKey::generate()) + .await + .unwrap(); + let mut b = Node::spawn_lan_only(&db, SecretKey::generate()) + .await + .unwrap(); let secret = generate_key_bytes(); a.open_shared_namespace(secret).await.unwrap(); b.open_shared_namespace(secret).await.unwrap(); - let (addr_a, addr_b) = (a.endpoint_addr(), b.endpoint_addr()); + let addr_a = direct_addr_of(&a).await; + let addr_b = direct_addr_of(&b).await; let id_a = addr_a.id; a.sync_with(vec![addr_b]).await.unwrap(); b.sync_with(vec![addr_a]).await.unwrap(); a.publish("pi/p/s", b"ciphertext".to_vec()).await.unwrap(); - // poll: the connection (and its path classification) settles async. - // Direct = localhost only; Mixed allowed because on a networked dev - // machine the n0 relay may also be active — the invariant is that the - // direct localhost path is observed, not that the relay is absent. let mut kind = PathKind::Unknown; for _ in 0..40 { tokio::time::sleep(std::time::Duration::from_millis(250)).await; @@ -902,9 +900,10 @@ mod tests { } } } - assert!( - matches!(kind, PathKind::Direct | PathKind::Mixed), - "in-process peers must show an active direct path, got {kind}" + assert_eq!( + kind, + PathKind::Direct, + "in-process peers must show an active direct path" ); a.shutdown().await.unwrap(); b.shutdown().await.unwrap(); @@ -928,15 +927,20 @@ mod tests { #[tokio::test] async fn blob_fetches_missed_download_from_peers() { let (da, db) = (tempdir("blob-fetch-a"), tempdir("blob-fetch-b")); - let mut a = Node::spawn(&da, SecretKey::generate()).await.unwrap(); - let mut b = Node::spawn(&db, SecretKey::generate()).await.unwrap(); + let mut a = Node::spawn_lan_only(&da, SecretKey::generate()) + .await + .unwrap(); + let mut b = Node::spawn_lan_only(&db, SecretKey::generate()) + .await + .unwrap(); let secret = generate_key_bytes(); a.open_shared_namespace(secret).await.unwrap(); b.open_shared_namespace(secret).await.unwrap(); // model iroh-docs' missed live download (never retried upstream): // index entries sync, content does not b.disable_auto_download().await.unwrap(); - let (addr_a, addr_b) = (a.endpoint_addr(), b.endpoint_addr()); + let addr_a = direct_addr_of(&a).await; + let addr_b = direct_addr_of(&b).await; a.sync_with(vec![addr_b]).await.unwrap(); b.sync_with(vec![addr_a]).await.unwrap(); let hash = a.publish("pi/p/s", b"ciphertext".to_vec()).await.unwrap();