Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522use std::{ fs, path::Path, sync::{Arc, Mutex},};
use tempfile::TempDir;use time::{Duration, OffsetDateTime};use tokio::time::sleep;
use crate::{ app::manifest::{build_manifest, build_manifest_with_stages}, config::{AppaConfig, CONFIG_VERSION, ConfiguredFolder}, iroh::{FolderSession, NodeHost},};
use super::{AppPaths, AppaService, INVITATION_PROTOCOL_VERSION, Invite};
const INITIAL_SYNC_ATTEMPTS: usize = 3;const INITIAL_SYNC_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(100);
#[test]fn bounds_exponential_retry_delays() { assert_eq!( super::run::retry_delay(1), std::time::Duration::from_secs(2) ); assert_eq!( super::run::retry_delay(2), std::time::Duration::from_secs(4) ); assert_eq!( super::run::retry_delay(6), std::time::Duration::from_secs(64) ); assert_eq!( super::run::retry_delay(100), std::time::Duration::from_secs(5 * 60) );}
#[test]fn applies_configuration() -> anyhow::Result<()> { let data_directory = TempDir::new()?; let folder_directory = TempDir::new()?; let config_path = data_directory.path().join("appa.toml"); let config = AppaConfig { version: CONFIG_VERSION, folders: vec![ConfiguredFolder { path: folder_directory.path().to_owned(), name: Some("notes".to_owned()), mode: crate::domain::FolderMode::ReceiveOnly, }], }; fs::write(&config_path, toml::to_string(&config)?)?; let appa = AppaService::open_at(AppPaths::from_data_directory( data_directory.path().join("state"), )?)?;
appa.validate_folder_inventory(&config_path)?; let folders = appa.import_folder_inventory(&config_path)?;
assert_eq!(folders[0].mode, crate::domain::FolderMode::ReceiveOnly); assert!( appa.active_member_ids(folders[0].id)? .contains(&appa.device_id()?) ); Ok(())}
#[test]fn revoking_a_member_updates_the_signed_roster_and_capability() -> anyhow::Result<()> { let data_directory = TempDir::new()?; let folder_directory = TempDir::new()?; let appa = AppaService::open_at(AppPaths::from_data_directory( data_directory.path().join("state"), )?)?; let folder = appa.register_folder(folder_directory.path())?; let member_device_id = iroh::SecretKey::generate().public().to_string(); appa.enroll_discovered_member(folder.id, &member_device_id)?; let old_capability = folder.capability;
appa.revoke_member(folder_directory.path(), &member_device_id)?;
assert!( !appa .active_member_ids(folder.id)? .contains(&member_device_id) ); assert_ne!( appa.require_folder(folder_directory.path())?.capability, old_capability ); appa.load_roster(folder.id)?.validate()}
#[tokio::test]async fn publishes_imported_files_in_bounded_stages() -> anyhow::Result<()> { let data_directory = TempDir::new()?; let folder_directory = TempDir::new()?; for file_number in 0..257 { fs::write( folder_directory.path().join(format!("file-{file_number}")), "small file", )?; } let appa = AppaService::open_at(AppPaths::from_data_directory( data_directory.path().join("state"), )?)?; let folder = appa.register_folder(folder_directory.path())?; let node = appa.load_node().await?; let context = super::SyncContext { folder: &folder, node: &node, state_store: &appa.state_store, now: OffsetDateTime::UNIX_EPOCH, }; let stages = Arc::new(Mutex::new(Vec::new())); let captured_stages = Arc::clone(&stages);
let manifest = build_manifest_with_stages(&context, move |manifest| { let captured_stages = Arc::clone(&captured_stages); async move { captured_stages.lock().expect("stages").push(manifest); Ok(()) } }) .await? .manifest;
{ let stages = stages.lock().expect("stages"); assert_eq!(stages.len(), 2); assert_eq!(stages[0].entries.len(), 256); assert_eq!(stages[1].entries.len(), 257); } assert_eq!(manifest.entries.len(), 257); node.shutdown().await?; Ok(())}
#[tokio::test]async fn synchronizes_and_deletes_a_file_between_two_devices() -> anyhow::Result<()> { let source_data = TempDir::new()?; let target_data = TempDir::new()?; let source_folder = TempDir::new()?; let target_folder = TempDir::new()?; let source_file = source_folder.path().join("note.txt"); let target_file = target_folder.path().join("note.txt"); let source_empty_directory = source_folder.path().join("empty"); let target_empty_directory = target_folder.path().join("empty"); fs::write(&source_file, "shared from Appa")?; fs::create_dir(&source_empty_directory)?;
let source = AppaService::open_at(AppPaths::from_data_directory( source_data.path().to_owned(), )?)?; let target = AppaService::open_at(AppPaths::from_data_directory( target_data.path().to_owned(), )?)?; let source_config = source.register_folder(source_folder.path())?; source.enroll_discovered_member(source_config.id, &target.device_id()?)?; let source_node = source.load_node().await?; source_node.wait_until_online().await?; let source_context = super::SyncContext { folder: &source_config, node: &source_node, state_store: &source.state_store, now: OffsetDateTime::UNIX_EPOCH, }; let source_manifest = build_manifest(&source_context).await?.manifest; source_node .register_folder( FolderSession { folder_id: source_config.id, capability: source_config.capability.clone(), active_member_ids: source.active_member_ids(source_config.id)?, roster: source .state_store .load_roster(source_config.id)? .expect("roster"), peer_endpoints: vec![], audit_events: vec![], }, source_manifest.clone(), ) .await?; source.state_store.save_manifest(&source_manifest)?; source_node.publish_manifest(source_manifest).await?;
let invite = Invite { protocol_version: INVITATION_PROTOCOL_VERSION, folder_id: source_config.id, folder_name: source_config.name.clone(), inviter_device_id: source_node.endpoint_address().id.to_string(), inviter_endpoint: source_node.endpoint_address(), capability: source_config.capability.clone(), roster: source .state_store .load_roster(source_config.id)? .expect("roster"), expires_at: OffsetDateTime::now_utc() + Duration::hours(1), signature: None, }; target .state_store .import_invite(target_folder.path(), &invite)?;
assert_eq!( sync_until_file_materializes(&target, target_folder.path(), &target_file).await?, 1 ); assert_eq!(fs::read_to_string(&target_file)?, "shared from Appa"); assert!(target_empty_directory.is_dir()); source.save_discovered_peers(&source_config, &source_node)?; assert_eq!(source.state_store.peers(source_config.id)?.len(), 1); assert!( source .active_member_ids(source_config.id)? .contains(&target.device_id()?) );
fs::remove_file(&source_file)?; fs::remove_dir(&source_empty_directory)?; let deleted_manifest = build_manifest(&source_context).await?.manifest; source.state_store.save_manifest(&deleted_manifest)?; source_node.publish_manifest(deleted_manifest).await?;
target.sync_once(target_folder.path()).await?; assert!(!target_file.exists()); assert!(!target_empty_directory.exists()); source_node.shutdown().await?; Ok(())}
#[tokio::test]async fn synchronizes_audit_events_written_before_the_author_was_revoked() -> anyhow::Result<()> { let source_data = TempDir::new()?; let target_data = TempDir::new()?; let source_folder = TempDir::new()?; let target_folder = TempDir::new()?; let source = AppaService::open_at(AppPaths::from_data_directory( source_data.path().to_owned(), )?)?; let target = AppaService::open_at(AppPaths::from_data_directory( target_data.path().to_owned(), )?)?; let initial_folder = source.register_folder(source_folder.path())?; let former_member = iroh::SecretKey::generate(); source.enroll_discovered_member(initial_folder.id, &former_member.public().to_string())?; let historical_roster = source .state_store .load_roster(initial_folder.id)? .expect("historical roster"); let historical_event = crate::domain::AuditEvent::create( initial_folder.id, former_member.public().to_string(), 1, None, crate::domain::AuditEventKind::ManifestCommitted, Some("manifest-before-revocation".to_owned()), historical_roster.hash()?, historical_roster, OffsetDateTime::UNIX_EPOCH, ) .sign(&former_member)?; source.state_store.append_audit_event(&historical_event)?; source.enroll_discovered_member(initial_folder.id, &target.device_id()?)?; source.revoke_member(source_folder.path(), &former_member.public().to_string())?; let source_config = source.require_folder(source_folder.path())?; let source_node = source.load_node().await?; source_node.wait_until_online().await?; source.sync_with_node(&source_config, &source_node).await?; let invite = Invite { protocol_version: INVITATION_PROTOCOL_VERSION, folder_id: source_config.id, folder_name: source_config.name.clone(), inviter_device_id: source_node.endpoint_address().id.to_string(), inviter_endpoint: source_node.endpoint_address(), capability: source_config.capability.clone(), roster: source .state_store .load_roster(source_config.id)? .expect("current roster"), expires_at: OffsetDateTime::now_utc() + Duration::hours(1), signature: None, }; target .state_store .import_invite(target_folder.path(), &invite)?;
target.sync_once(target_folder.path()).await?;
let target_events = target.state_store.audit_events(source_config.id)?; let historical_event_hash = historical_event.hash()?; assert!(target_events.iter().any(|event| { event.author_device_id == former_member.public().to_string() && event.hash().is_ok_and(|hash| hash == historical_event_hash) })); source_node.shutdown().await?; Ok(())}
#[tokio::test]async fn learns_and_contacts_a_peer_introduced_during_the_same_sync_pass() -> anyhow::Result<()> { let source_data = TempDir::new()?; let source_folder = TempDir::new()?; let bridge_data = TempDir::new()?; let third_data = TempDir::new()?; let source = AppaService::open_at(AppPaths::from_data_directory( source_data.path().to_owned(), )?)?; let folder = source.register_folder(source_folder.path())?; let bridge_identity = iroh::SecretKey::generate(); let third_identity = iroh::SecretKey::generate(); source.enroll_discovered_member(folder.id, &bridge_identity.public().to_string())?; source.enroll_discovered_member(folder.id, &third_identity.public().to_string())?; let roster = source .state_store .load_roster(folder.id)? .expect("shared roster"); let bridge_node = NodeHost::load_with_lan_discovery(bridge_data.path(), bridge_identity, false).await?; let third_node = NodeHost::load_with_lan_discovery(third_data.path(), third_identity, false).await?; bridge_node .register_folder( FolderSession { folder_id: folder.id, capability: folder.capability.clone(), active_member_ids: roster.active_device_ids(), roster, peer_endpoints: vec![third_node.endpoint_address()], audit_events: vec![], }, crate::domain::Manifest::empty(folder.id), ) .await?; let source_node = source.load_node().await?; source_node.wait_until_online().await?; bridge_node.wait_until_online().await?; third_node.wait_until_online().await?; source .state_store .save_peer(folder.id, &bridge_node.endpoint_address())?;
source.sync_with_node(&folder, &source_node).await?;
assert!( source .state_store .peers(folder.id)? .iter() .any(|peer| peer.id == third_node.endpoint_address().id) ); source_node.shutdown().await?; bridge_node.shutdown().await?; third_node.shutdown().await?; Ok(())}
async fn sync_until_file_materializes( appa: &AppaService, folder: &Path, expected_file: &Path,) -> anyhow::Result<usize> { for attempt in 0..INITIAL_SYNC_ATTEMPTS { let synchronized_files = appa.sync_once(folder).await?; if expected_file.exists() { return Ok(synchronized_files); } if attempt + 1 < INITIAL_SYNC_ATTEMPTS { sleep(INITIAL_SYNC_RETRY_DELAY).await; } } anyhow::bail!( "{} did not materialize after {INITIAL_SYNC_ATTEMPTS} sync attempts", expected_file.display() )}
#[tokio::test]async fn preserves_both_versions_after_offline_edits() -> anyhow::Result<()> { let source_data = TempDir::new()?; let target_data = TempDir::new()?; let source_folder = TempDir::new()?; let target_folder = TempDir::new()?; let source_file = source_folder.path().join("note.txt"); let target_file = target_folder.path().join("note.txt"); fs::write(&source_file, "original")?;
let source = AppaService::open_at(AppPaths::from_data_directory( source_data.path().to_owned(), )?)?; let target = AppaService::open_at(AppPaths::from_data_directory( target_data.path().to_owned(), )?)?; let source_config = source.register_folder(source_folder.path())?; source.enroll_discovered_member(source_config.id, &target.device_id()?)?; let source_node = source.load_node().await?; source_node.wait_until_online().await?; let source_context = super::SyncContext { folder: &source_config, node: &source_node, state_store: &source.state_store, now: OffsetDateTime::UNIX_EPOCH, }; let source_manifest = build_manifest(&source_context).await?.manifest; source_node .register_folder( FolderSession { folder_id: source_config.id, capability: source_config.capability.clone(), active_member_ids: source.active_member_ids(source_config.id)?, roster: source .state_store .load_roster(source_config.id)? .expect("roster"), peer_endpoints: vec![], audit_events: vec![], }, source_manifest.clone(), ) .await?; source.state_store.save_manifest(&source_manifest)?; source_node.publish_manifest(source_manifest).await?;
let invite = Invite { protocol_version: INVITATION_PROTOCOL_VERSION, folder_id: source_config.id, folder_name: source_config.name.clone(), inviter_device_id: source_node.endpoint_address().id.to_string(), inviter_endpoint: source_node.endpoint_address(), capability: source_config.capability.clone(), roster: source .state_store .load_roster(source_config.id)? .expect("roster"), expires_at: OffsetDateTime::now_utc() + Duration::hours(1), signature: None, }; let target_config = target .state_store .import_invite(target_folder.path(), &invite)?; target.sync_once(target_folder.path()).await?;
fs::write(&source_file, "fortree edit")?; fs::write(&target_file, "petalburg edit")?; let target_node = target.load_node().await?; target_node.wait_until_online().await?; let target_context = super::SyncContext { folder: &target_config, node: &target_node, state_store: &target.state_store, now: OffsetDateTime::UNIX_EPOCH, }; let target_manifest = build_manifest(&target_context).await?.manifest; target_node .register_folder( FolderSession { folder_id: target_config.id, capability: target_config.capability.clone(), active_member_ids: target.active_member_ids(target_config.id)?, roster: source .state_store .load_roster(source_config.id)? .expect("roster"), peer_endpoints: vec![], audit_events: vec![], }, target_manifest.clone(), ) .await?; target.state_store.save_manifest(&target_manifest)?; target_node.publish_manifest(target_manifest).await?; source .state_store .save_peer(source_config.id, &target_node.endpoint_address())?;
source.sync_with_node(&source_config, &source_node).await?;
let manifest = source.state_store.load_manifest(source_config.id)?; assert_eq!( manifest .entries .values() .filter(|entry| entry.kind == crate::domain::EntryKind::File) .count(), 2 ); let contents = fs::read_dir(source_folder.path())? .map(|entry| fs::read_to_string(entry?.path())) .collect::<Result<Vec<_>, _>>()?; assert!(contents.contains(&"fortree edit".to_owned())); assert!(contents.contains(&"petalburg edit".to_owned())); assert!( manifest .entries .keys() .any(|path| path.contains(" (conflict-")) );
for _ in 0..3 { target.sync_with_node(&target_config, &target_node).await?; source.sync_with_node(&source_config, &source_node).await?; } let settled_manifest = source.state_store.load_manifest(source_config.id)?; assert_eq!( settled_manifest .entries .keys() .filter(|path| path.contains(" (conflict-")) .count(), 1 ); assert_eq!(source.conflicts(source_folder.path())?.len(), 1);
target_node.shutdown().await?; source_node.shutdown().await?; Ok(())}