mirror of
https://github.com/spacedriveapp/spacedrive.git
synced 2026-02-20 07:37:26 -05:00
1461 lines
41 KiB
Rust
1461 lines
41 KiB
Rust
//! Initial Backfill Sync Test
|
|
//!
|
|
//! Tests the scenario where one device indexes completely before the second device connects.
|
|
//! This validates backfill behavior and content_id linkage without real-time sync complexity.
|
|
|
|
mod helpers;
|
|
|
|
use helpers::{
|
|
create_snapshot_dir, create_test_volume, init_test_tracing, register_device, wait_for_indexing,
|
|
wait_for_sync, MockTransport, TestConfigBuilder, TestDataDir,
|
|
};
|
|
use sd_core::{
|
|
infra::{db::entities, sync::NetworkTransport},
|
|
location::{create_location, IndexMode, LocationCreateArgs},
|
|
service::Service,
|
|
Core,
|
|
};
|
|
use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect};
|
|
use std::sync::Arc;
|
|
use tokio::time::Duration;
|
|
use uuid::Uuid;
|
|
|
|
#[tokio::test]
|
|
async fn test_initial_backfill_alice_indexes_first() -> anyhow::Result<()> {
|
|
let snapshot_dir = create_snapshot_dir("backfill_alice_first").await?;
|
|
init_test_tracing("backfill_alice_first", &snapshot_dir)?;
|
|
|
|
// Use TestDataDir helper for proper cross-platform directory management
|
|
let test_data_alice = TestDataDir::new("backfill_alice")?;
|
|
let test_data_bob = TestDataDir::new("backfill_bob")?;
|
|
|
|
let temp_dir_alice = test_data_alice.core_data_path();
|
|
let temp_dir_bob = test_data_bob.core_data_path();
|
|
|
|
tracing::info!(
|
|
snapshot_dir = %snapshot_dir.display(),
|
|
alice_dir = %temp_dir_alice.display(),
|
|
bob_dir = %temp_dir_bob.display(),
|
|
"Test directories initialized"
|
|
);
|
|
|
|
TestConfigBuilder::new(temp_dir_alice.clone()).build()?;
|
|
TestConfigBuilder::new(temp_dir_bob.clone()).build()?;
|
|
|
|
tracing::info!("=== Phase 1: Alice indexes location (Bob not connected yet) ===");
|
|
|
|
// Generate a shared library UUID for both devices
|
|
let library_id = Uuid::new_v4();
|
|
|
|
let core_alice = Core::new(temp_dir_alice.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Alice core: {}", e))?;
|
|
let device_alice_id = core_alice.device.device_id()?;
|
|
let library_alice = core_alice
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Backfill Test Library",
|
|
None,
|
|
core_alice.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let device_record = entities::device::Entity::find()
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Device not found"))?;
|
|
|
|
// Use Spacedrive source code for deterministic testing across all environments
|
|
let test_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
|
|
.parent()
|
|
.unwrap()
|
|
.to_path_buf();
|
|
let location_args = LocationCreateArgs {
|
|
path: test_path.clone(),
|
|
name: Some("spacedrive".to_string()),
|
|
index_mode: IndexMode::Content,
|
|
};
|
|
|
|
let location_db_id = create_location(
|
|
library_alice.clone(),
|
|
library_alice.event_bus(),
|
|
location_args,
|
|
device_record.id,
|
|
)
|
|
.await?;
|
|
|
|
tracing::info!(location_id = location_db_id, "Location created on Alice");
|
|
|
|
// Create volumes BEFORE indexing so entries can reference them
|
|
tracing::info!("Creating test volumes on Alice");
|
|
let _ = create_test_volume(
|
|
&library_alice,
|
|
device_alice_id,
|
|
"test-vol-1",
|
|
"Alice Volume 1",
|
|
)
|
|
.await?;
|
|
|
|
// Link the location to the first volume
|
|
let first_volume = entities::volume::Entity::find()
|
|
.filter(entities::volume::Column::DeviceId.eq(device_alice_id))
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Failed to find volume for Alice"))?;
|
|
|
|
// Get the location record to find its root entry
|
|
let location_record = entities::location::Entity::find()
|
|
.filter(entities::location::Column::Id.eq(location_db_id))
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Location not found"))?;
|
|
|
|
let location_entry_id = location_record
|
|
.entry_id
|
|
.ok_or_else(|| anyhow::anyhow!("Location has no entry_id"))?;
|
|
|
|
// Update location to reference volume
|
|
entities::location::Entity::update_many()
|
|
.filter(entities::location::Column::Id.eq(location_db_id))
|
|
.col_expr(
|
|
entities::location::Column::VolumeId,
|
|
sea_orm::sea_query::Expr::value(first_volume.id),
|
|
)
|
|
.exec(library_alice.db().conn())
|
|
.await?;
|
|
|
|
// CRITICAL: Update location root entry to reference the volume
|
|
// Without this, the root entry has volume_id=NULL and won't be queried during sync
|
|
entities::entry::Entity::update_many()
|
|
.filter(entities::entry::Column::Id.eq(location_entry_id))
|
|
.col_expr(
|
|
entities::entry::Column::VolumeId,
|
|
sea_orm::sea_query::Expr::value(first_volume.id),
|
|
)
|
|
.exec(library_alice.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
volume_id = first_volume.id,
|
|
entry_id = location_entry_id,
|
|
"Linked location and its root entry to volume before indexing"
|
|
);
|
|
|
|
wait_for_indexing(&library_alice, location_db_id, Duration::from_secs(120)).await?;
|
|
|
|
let alice_entries_after_index = entities::entry::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let alice_content_after_index = entities::content_identity::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let alice_mime_types_after_index = entities::mime_type::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
entries = alice_entries_after_index,
|
|
content_identities = alice_content_after_index,
|
|
mime_types = alice_mime_types_after_index,
|
|
"Alice indexing complete"
|
|
);
|
|
|
|
// Create additional volume for testing volume sync
|
|
tracing::info!("Creating second test volume on Alice");
|
|
let _ = create_test_volume(
|
|
&library_alice,
|
|
device_alice_id,
|
|
"test-vol-2",
|
|
"Alice Volume 2",
|
|
)
|
|
.await?;
|
|
|
|
let alice_volumes = entities::volume::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
tracing::info!(volumes = alice_volumes, "Alice has tracked volumes");
|
|
|
|
tracing::info!("=== Phase 2: Bob connects and starts backfill ===");
|
|
|
|
let core_bob = Core::new(temp_dir_bob.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Bob core: {}", e))?;
|
|
let device_bob_id = core_bob.device.device_id()?;
|
|
let library_bob = core_bob
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Backfill Test Library",
|
|
None,
|
|
core_bob.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
register_device(&library_alice, device_bob_id, "Bob").await?;
|
|
register_device(&library_bob, device_alice_id, "Alice").await?;
|
|
|
|
let (transport_alice, transport_bob) = MockTransport::new_pair(device_alice_id, device_bob_id);
|
|
|
|
library_alice
|
|
.init_sync_service(
|
|
device_alice_id,
|
|
transport_alice.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
library_bob
|
|
.init_sync_service(
|
|
device_bob_id,
|
|
transport_bob.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
transport_alice
|
|
.register_sync_service(
|
|
device_alice_id,
|
|
Arc::downgrade(library_alice.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
transport_bob
|
|
.register_sync_service(
|
|
device_bob_id,
|
|
Arc::downgrade(library_bob.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
|
|
library_alice.sync_service().unwrap().start().await?;
|
|
library_bob.sync_service().unwrap().start().await?;
|
|
|
|
tracing::info!("Sync services started - backfill should begin automatically");
|
|
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
|
|
tracing::info!("=== Phase 3: Waiting for backfill to complete ===");
|
|
|
|
// Log current counts before sync
|
|
let alice_entries_before_sync = entities::entry::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_entries_before_sync = entities::entry::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
alice_entries = alice_entries_before_sync,
|
|
bob_entries = bob_entries_before_sync,
|
|
"Starting sync wait - Alice has indexed, Bob needs backfill"
|
|
);
|
|
|
|
wait_for_sync(&library_alice, &library_bob, Duration::from_secs(120)).await?;
|
|
|
|
let bob_entries_final = entities::entry::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
let bob_content_final = entities::content_identity::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
let bob_mime_types_final = entities::mime_type::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
let alice_volumes_final = entities::volume::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_volumes_final = entities::volume::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
alice_entries = alice_entries_after_index,
|
|
bob_entries = bob_entries_final,
|
|
alice_content = alice_content_after_index,
|
|
bob_content = bob_content_final,
|
|
alice_mime_types = alice_mime_types_after_index,
|
|
bob_mime_types = bob_mime_types_final,
|
|
alice_volumes = alice_volumes_final,
|
|
bob_volumes = bob_volumes_final,
|
|
"=== Final counts ==="
|
|
);
|
|
|
|
let entry_diff = (alice_entries_after_index as i64 - bob_entries_final as i64).abs();
|
|
let content_diff = (alice_content_after_index as i64 - bob_content_final as i64).abs();
|
|
let mime_type_diff = (alice_mime_types_after_index as i64 - bob_mime_types_final as i64).abs();
|
|
|
|
assert!(
|
|
entry_diff <= 5,
|
|
"Entry count mismatch after backfill: Alice {}, Bob {} (diff: {})",
|
|
alice_entries_after_index,
|
|
bob_entries_final,
|
|
entry_diff
|
|
);
|
|
|
|
assert!(
|
|
content_diff <= 5,
|
|
"Content identity count mismatch after backfill: Alice {}, Bob {} (diff: {})",
|
|
alice_content_after_index,
|
|
bob_content_final,
|
|
content_diff
|
|
);
|
|
|
|
assert!(
|
|
mime_type_diff == 0,
|
|
"Mime type count mismatch after backfill: Alice {}, Bob {} (diff: {})",
|
|
alice_mime_types_after_index,
|
|
bob_mime_types_final,
|
|
mime_type_diff
|
|
);
|
|
|
|
// Verify mime types have valid UUIDs (required for sync)
|
|
let alice_mime_types_with_uuid = entities::mime_type::Entity::find()
|
|
.filter(entities::mime_type::Column::Uuid.is_not_null())
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
|
|
assert_eq!(
|
|
alice_mime_types_with_uuid, alice_mime_types_after_index,
|
|
"All mime types on Alice should have UUIDs for sync"
|
|
);
|
|
|
|
let bob_mime_types_with_uuid = entities::mime_type::Entity::find()
|
|
.filter(entities::mime_type::Column::Uuid.is_not_null())
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
assert_eq!(
|
|
bob_mime_types_with_uuid, bob_mime_types_final,
|
|
"All mime types on Bob should have UUIDs after sync"
|
|
);
|
|
|
|
tracing::info!(
|
|
alice_mime_types = alice_mime_types_after_index,
|
|
bob_mime_types = bob_mime_types_final,
|
|
"Mime type sync verification passed"
|
|
);
|
|
|
|
// Verify volume sync
|
|
assert_eq!(
|
|
alice_volumes_final, bob_volumes_final,
|
|
"Volume count mismatch after backfill: Alice {}, Bob {}",
|
|
alice_volumes_final, bob_volumes_final
|
|
);
|
|
|
|
tracing::info!(
|
|
alice_volumes = alice_volumes_final,
|
|
bob_volumes = bob_volumes_final,
|
|
"Volume sync verification passed"
|
|
);
|
|
|
|
// Verify content_id linkage
|
|
let bob_files_linked = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Kind.eq(0))
|
|
.filter(entities::entry::Column::ContentId.is_not_null())
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
let bob_total_files = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Kind.eq(0))
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
let linkage_pct = if bob_total_files > 0 {
|
|
(bob_files_linked * 100) / bob_total_files
|
|
} else {
|
|
0
|
|
};
|
|
|
|
tracing::info!(
|
|
bob_files_linked = bob_files_linked,
|
|
bob_total_files = bob_total_files,
|
|
linkage_pct = linkage_pct,
|
|
"Bob's content_id linkage after backfill"
|
|
);
|
|
|
|
assert!(
|
|
linkage_pct >= 90,
|
|
"Content_id linkage too low on Bob after backfill: {}% (expected >= 90%)",
|
|
linkage_pct
|
|
);
|
|
|
|
tracing::info!("=== Phase 4: Verifying structural integrity ===");
|
|
|
|
// Verify directory structure preservation by checking known directories
|
|
verify_known_directories(&library_alice, &library_bob).await?;
|
|
|
|
// Verify closure table correctness
|
|
verify_closure_table_integrity(&library_alice, &library_bob).await?;
|
|
|
|
// Verify parent-child relationships match
|
|
verify_parent_child_relationships(&library_alice, &library_bob).await?;
|
|
|
|
// Verify file metadata matches for sample files
|
|
verify_file_metadata_accuracy(&library_alice, &library_bob).await?;
|
|
|
|
// Verify nested file structure and ancestor chains
|
|
verify_nested_file_structure(&library_alice, &library_bob).await?;
|
|
|
|
tracing::info!("✅ All structural integrity checks passed");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test bidirectional volume sync - both devices should receive each other's volumes
|
|
#[tokio::test]
|
|
async fn test_bidirectional_volume_sync() -> anyhow::Result<()> {
|
|
let snapshot_dir = create_snapshot_dir("bidirectional_volume_sync").await?;
|
|
init_test_tracing("bidirectional_volume_sync", &snapshot_dir)?;
|
|
|
|
// Use TestDataDir helper for proper cross-platform directory management
|
|
let test_data_alice = TestDataDir::new("volume_sync_alice")?;
|
|
let test_data_bob = TestDataDir::new("volume_sync_bob")?;
|
|
|
|
let temp_dir_alice = test_data_alice.core_data_path();
|
|
let temp_dir_bob = test_data_bob.core_data_path();
|
|
|
|
tracing::info!("=== Phase 1: Initialize both devices ===");
|
|
|
|
TestConfigBuilder::new(temp_dir_alice.clone()).build()?;
|
|
TestConfigBuilder::new(temp_dir_bob.clone()).build()?;
|
|
|
|
// Generate a shared library UUID for both devices
|
|
let library_id = Uuid::new_v4();
|
|
|
|
let core_alice = Core::new(temp_dir_alice.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Alice core: {}", e))?;
|
|
let device_alice_id = core_alice.device.device_id()?;
|
|
let library_alice = core_alice
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Volume Sync Test",
|
|
None,
|
|
core_alice.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let core_bob = Core::new(temp_dir_bob.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Bob core: {}", e))?;
|
|
let device_bob_id = core_bob.device.device_id()?;
|
|
let library_bob = core_bob
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Volume Sync Test",
|
|
None,
|
|
core_bob.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
register_device(&library_alice, device_bob_id, "Bob").await?;
|
|
register_device(&library_bob, device_alice_id, "Alice").await?;
|
|
|
|
tracing::info!("=== Phase 2: Create volumes on both devices ===");
|
|
|
|
// Alice creates her Macintosh HD
|
|
let _ = create_test_volume(
|
|
&library_alice,
|
|
device_alice_id,
|
|
"alice-macos-hd-fingerprint",
|
|
"Macintosh HD",
|
|
)
|
|
.await?;
|
|
|
|
// Bob creates his Macintosh HD
|
|
let _ = create_test_volume(
|
|
&library_bob,
|
|
device_bob_id,
|
|
"bob-macos-hd-fingerprint",
|
|
"Macintosh HD",
|
|
)
|
|
.await?;
|
|
|
|
let alice_volumes_before = entities::volume::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_volumes_before = entities::volume::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
alice_volumes = alice_volumes_before,
|
|
bob_volumes = bob_volumes_before,
|
|
"Volumes created on both devices"
|
|
);
|
|
|
|
assert_eq!(
|
|
alice_volumes_before, 1,
|
|
"Alice should have 1 volume before sync"
|
|
);
|
|
assert_eq!(
|
|
bob_volumes_before, 1,
|
|
"Bob should have 1 volume before sync"
|
|
);
|
|
|
|
tracing::info!("=== Phase 3: Start sync services ===");
|
|
|
|
let (transport_alice, transport_bob) = MockTransport::new_pair(device_alice_id, device_bob_id);
|
|
|
|
library_alice
|
|
.init_sync_service(
|
|
device_alice_id,
|
|
transport_alice.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
library_bob
|
|
.init_sync_service(
|
|
device_bob_id,
|
|
transport_bob.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
transport_alice
|
|
.register_sync_service(
|
|
device_alice_id,
|
|
Arc::downgrade(library_alice.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
transport_bob
|
|
.register_sync_service(
|
|
device_bob_id,
|
|
Arc::downgrade(library_bob.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
|
|
library_alice.sync_service().unwrap().start().await?;
|
|
library_bob.sync_service().unwrap().start().await?;
|
|
|
|
tracing::info!("Sync services started - backfill should begin");
|
|
|
|
tokio::time::sleep(Duration::from_millis(1000)).await;
|
|
|
|
tracing::info!("=== Phase 4: Wait for bidirectional sync ===");
|
|
|
|
// Wait for sync with simpler logic for volumes
|
|
let start = tokio::time::Instant::now();
|
|
let max_duration = Duration::from_secs(30);
|
|
let mut stable_iterations = 0;
|
|
|
|
while start.elapsed() < max_duration {
|
|
let alice_volumes = entities::volume::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_volumes = entities::volume::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::debug!(
|
|
alice_volumes = alice_volumes,
|
|
bob_volumes = bob_volumes,
|
|
elapsed_ms = start.elapsed().as_millis(),
|
|
"Checking sync progress"
|
|
);
|
|
|
|
if alice_volumes == 2 && bob_volumes == 2 {
|
|
stable_iterations += 1;
|
|
if stable_iterations >= 5 {
|
|
tracing::info!(
|
|
duration_ms = start.elapsed().as_millis(),
|
|
"Bidirectional volume sync complete"
|
|
);
|
|
break;
|
|
}
|
|
} else {
|
|
stable_iterations = 0;
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
|
|
tracing::info!("=== Phase 5: Verify bidirectional sync ===");
|
|
|
|
let alice_volumes_final = entities::volume::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_volumes_final = entities::volume::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
let alice_volumes_list = entities::volume::Entity::find()
|
|
.all(library_alice.db().conn())
|
|
.await?;
|
|
let bob_volumes_list = entities::volume::Entity::find()
|
|
.all(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
alice_total = alice_volumes_final,
|
|
bob_total = bob_volumes_final,
|
|
alice_devices = ?alice_volumes_list.iter().map(|v| (v.device_id, v.display_name.clone())).collect::<Vec<_>>(),
|
|
bob_devices = ?bob_volumes_list.iter().map(|v| (v.device_id, v.display_name.clone())).collect::<Vec<_>>(),
|
|
"=== Final volume counts ==="
|
|
);
|
|
|
|
assert_eq!(
|
|
alice_volumes_final, 2,
|
|
"Alice should have 2 volumes (her own + Bob's), but has {}",
|
|
alice_volumes_final
|
|
);
|
|
assert_eq!(
|
|
bob_volumes_final, 2,
|
|
"Bob should have 2 volumes (his own + Alice's), but has {}",
|
|
bob_volumes_final
|
|
);
|
|
|
|
// Verify Alice has both
|
|
let alice_has_own = alice_volumes_list
|
|
.iter()
|
|
.any(|v| v.device_id == device_alice_id);
|
|
let alice_has_bobs = alice_volumes_list
|
|
.iter()
|
|
.any(|v| v.device_id == device_bob_id);
|
|
|
|
assert!(alice_has_own, "Alice should have her own volume");
|
|
assert!(alice_has_bobs, "Alice should have Bob's volume");
|
|
|
|
// Verify Bob has both
|
|
let bob_has_own = bob_volumes_list
|
|
.iter()
|
|
.any(|v| v.device_id == device_bob_id);
|
|
let bob_has_alices = bob_volumes_list
|
|
.iter()
|
|
.any(|v| v.device_id == device_alice_id);
|
|
|
|
assert!(bob_has_own, "Bob should have his own volume");
|
|
assert!(bob_has_alices, "Bob should have Alice's volume");
|
|
|
|
tracing::info!("✅ Bidirectional volume sync verified successfully");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test that volume ResourceChanged events are emitted on the receiving device during sync
|
|
#[tokio::test]
|
|
async fn test_volume_resource_events_on_sync() -> anyhow::Result<()> {
|
|
let snapshot_dir = create_snapshot_dir("volume_resource_events").await?;
|
|
init_test_tracing("volume_resource_events", &snapshot_dir)?;
|
|
|
|
let test_data_alice = TestDataDir::new("volume_events_alice")?;
|
|
let test_data_bob = TestDataDir::new("volume_events_bob")?;
|
|
|
|
let temp_dir_alice = test_data_alice.core_data_path();
|
|
let temp_dir_bob = test_data_bob.core_data_path();
|
|
|
|
tracing::info!("=== Phase 1: Initialize both devices ===");
|
|
|
|
TestConfigBuilder::new(temp_dir_alice.clone()).build()?;
|
|
TestConfigBuilder::new(temp_dir_bob.clone()).build()?;
|
|
|
|
let library_id = Uuid::new_v4();
|
|
|
|
let core_alice = Core::new(temp_dir_alice.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Alice core: {}", e))?;
|
|
let device_alice_id = core_alice.device.device_id()?;
|
|
let library_alice = core_alice
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Volume Event Test",
|
|
None,
|
|
core_alice.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let core_bob = Core::new(temp_dir_bob.clone())
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Bob core: {}", e))?;
|
|
let device_bob_id = core_bob.device.device_id()?;
|
|
let library_bob = core_bob
|
|
.libraries
|
|
.create_library_with_id(
|
|
library_id,
|
|
"Volume Event Test",
|
|
None,
|
|
core_bob.context.clone(),
|
|
)
|
|
.await?;
|
|
|
|
register_device(&library_alice, device_bob_id, "Bob").await?;
|
|
register_device(&library_bob, device_alice_id, "Alice").await?;
|
|
|
|
tracing::info!("=== Phase 2: Create volume on Alice only ===");
|
|
|
|
// Alice creates a volume
|
|
let alice_volume_uuid = create_test_volume(
|
|
&library_alice,
|
|
device_alice_id,
|
|
"alice-test-volume",
|
|
"Alice's Test Volume",
|
|
)
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
volume_uuid = %alice_volume_uuid,
|
|
"Alice created volume"
|
|
);
|
|
|
|
tracing::info!("=== Phase 3: Set up event listener on Bob BEFORE sync ===");
|
|
|
|
// Subscribe to Bob's event bus for volume ResourceChanged events
|
|
let mut bob_events = library_bob.event_bus().subscribe();
|
|
let volume_event_received = Arc::new(tokio::sync::Mutex::new(false));
|
|
let volume_event_received_clone = volume_event_received.clone();
|
|
let alice_volume_uuid_clone = alice_volume_uuid;
|
|
|
|
// Spawn event listener task
|
|
let event_listener = tokio::spawn(async move {
|
|
use sd_core::infra::event::Event;
|
|
|
|
tracing::info!("Bob's event listener started, waiting for volume ResourceChanged...");
|
|
|
|
while let Ok(event) = bob_events.recv().await {
|
|
tracing::debug!("Bob received event: {:?}", event);
|
|
|
|
match event {
|
|
Event::ResourceChangedBatch {
|
|
resource_type,
|
|
resources,
|
|
..
|
|
} => {
|
|
if resource_type == "volume" {
|
|
tracing::info!(
|
|
resource_count = if let serde_json::Value::Array(arr) = &resources {
|
|
arr.len()
|
|
} else {
|
|
0
|
|
},
|
|
"Bob received ResourceChangedBatch for volumes"
|
|
);
|
|
|
|
// Check if Alice's volume is in the batch
|
|
if let serde_json::Value::Array(volume_array) = resources {
|
|
for volume_json in volume_array {
|
|
if let Some(uuid_str) =
|
|
volume_json.get("id").and_then(|v| v.as_str())
|
|
{
|
|
if let Ok(volume_id) = Uuid::parse_str(uuid_str) {
|
|
if volume_id == alice_volume_uuid_clone {
|
|
tracing::info!(
|
|
volume_uuid = %volume_id,
|
|
"✅ Bob received ResourceChanged event for Alice's volume!"
|
|
);
|
|
*volume_event_received_clone.lock().await = true;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Event::ResourceChanged {
|
|
resource_type,
|
|
resource,
|
|
..
|
|
} => {
|
|
if resource_type == "volume" {
|
|
tracing::info!("Bob received single ResourceChanged for volume");
|
|
|
|
if let Some(uuid_str) = resource.get("id").and_then(|v| v.as_str()) {
|
|
if let Ok(volume_id) = Uuid::parse_str(uuid_str) {
|
|
if volume_id == alice_volume_uuid_clone {
|
|
tracing::info!(
|
|
volume_uuid = %volume_id,
|
|
"✅ Bob received ResourceChanged event for Alice's volume!"
|
|
);
|
|
*volume_event_received_clone.lock().await = true;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
_ => {
|
|
// Ignore other events
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
tracing::info!("=== Phase 4: Start sync services ===");
|
|
|
|
let (transport_alice, transport_bob) = MockTransport::new_pair(device_alice_id, device_bob_id);
|
|
|
|
library_alice
|
|
.init_sync_service(
|
|
device_alice_id,
|
|
transport_alice.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
library_bob
|
|
.init_sync_service(
|
|
device_bob_id,
|
|
transport_bob.clone() as Arc<dyn NetworkTransport>,
|
|
)
|
|
.await?;
|
|
|
|
transport_alice
|
|
.register_sync_service(
|
|
device_alice_id,
|
|
Arc::downgrade(library_alice.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
transport_bob
|
|
.register_sync_service(
|
|
device_bob_id,
|
|
Arc::downgrade(library_bob.sync_service().unwrap()),
|
|
)
|
|
.await;
|
|
|
|
library_alice.sync_service().unwrap().start().await?;
|
|
library_bob.sync_service().unwrap().start().await?;
|
|
|
|
tracing::info!("Sync services started - backfill should begin");
|
|
|
|
tracing::info!("=== Phase 5: Wait for volume to sync and event to be emitted ===");
|
|
|
|
// Wait for Bob to receive the volume in the database
|
|
let start = tokio::time::Instant::now();
|
|
let max_duration = Duration::from_secs(30);
|
|
|
|
loop {
|
|
if start.elapsed() > max_duration {
|
|
anyhow::bail!("Timeout waiting for volume to sync to Bob");
|
|
}
|
|
|
|
let bob_volume = entities::volume::Entity::find()
|
|
.filter(entities::volume::Column::Uuid.eq(alice_volume_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?;
|
|
|
|
if bob_volume.is_some() {
|
|
tracing::info!("Bob received Alice's volume in database");
|
|
break;
|
|
}
|
|
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
}
|
|
|
|
// Give the event system a moment to emit the event after DB insert
|
|
tokio::time::sleep(Duration::from_millis(500)).await;
|
|
|
|
// Check if the event was received
|
|
let event_was_received = *volume_event_received.lock().await;
|
|
|
|
// Abort the listener task
|
|
event_listener.abort();
|
|
|
|
tracing::info!(event_received = event_was_received, "=== Test Result ===");
|
|
|
|
assert!(
|
|
event_was_received,
|
|
"Bob should have received a ResourceChanged event for Alice's volume during sync, but didn't"
|
|
);
|
|
|
|
tracing::info!(
|
|
"✅ Volume ResourceChanged event was emitted on the receiving device during sync"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Verify that known directories from the Spacedrive source exist on both devices
|
|
async fn verify_known_directories(
|
|
library_alice: &Arc<sd_core::library::Library>,
|
|
library_bob: &Arc<sd_core::library::Library>,
|
|
) -> anyhow::Result<()> {
|
|
use sea_orm::EntityTrait;
|
|
|
|
tracing::info!("Verifying known directory structure...");
|
|
|
|
// Known directories in Spacedrive source tree
|
|
let known_dirs = ["core", "apps", "packages", "interface"];
|
|
|
|
for dir_name in known_dirs {
|
|
// Check Alice has this directory
|
|
let alice_dir = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Name.eq(dir_name))
|
|
.filter(entities::entry::Column::Kind.eq(1)) // Directory
|
|
.one(library_alice.db().conn())
|
|
.await?;
|
|
|
|
let alice_uuid = alice_dir
|
|
.as_ref()
|
|
.and_then(|d| d.uuid)
|
|
.ok_or_else(|| anyhow::anyhow!("Alice missing directory: {}", dir_name))?;
|
|
|
|
// Check Bob has the same directory with matching UUID
|
|
let bob_dir = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(alice_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"Bob missing directory with UUID {}: {}",
|
|
alice_uuid,
|
|
dir_name
|
|
)
|
|
})?;
|
|
|
|
assert_eq!(
|
|
bob_dir.name, dir_name,
|
|
"Directory name mismatch for UUID {}: Alice '{}', Bob '{}'",
|
|
alice_uuid, dir_name, bob_dir.name
|
|
);
|
|
|
|
assert_eq!(
|
|
bob_dir.kind, 1,
|
|
"Directory kind mismatch for '{}': expected 1 (Directory), got {}",
|
|
dir_name, bob_dir.kind
|
|
);
|
|
|
|
tracing::debug!(
|
|
dir_name = dir_name,
|
|
uuid = %alice_uuid,
|
|
"Directory structure verified"
|
|
);
|
|
}
|
|
|
|
tracing::info!("✅ Known directory structure preserved");
|
|
Ok(())
|
|
}
|
|
|
|
/// Verify closure table integrity by checking ancestor-descendant relationships
|
|
async fn verify_closure_table_integrity(
|
|
library_alice: &Arc<sd_core::library::Library>,
|
|
library_bob: &Arc<sd_core::library::Library>,
|
|
) -> anyhow::Result<()> {
|
|
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
|
|
|
|
tracing::info!("Verifying closure table integrity...");
|
|
|
|
// Get total closure entries on both sides
|
|
let alice_closure_count = entities::entry_closure::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
|
|
let bob_closure_count = entities::entry_closure::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
// Also check actual entry counts for comparison
|
|
let alice_entry_count = entities::entry::Entity::find()
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
let bob_entry_count = entities::entry::Entity::find()
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
alice_closure = alice_closure_count,
|
|
bob_closure = bob_closure_count,
|
|
alice_entries = alice_entry_count,
|
|
bob_entries = bob_entry_count,
|
|
closure_ratio_alice = alice_closure_count as f64 / alice_entry_count as f64,
|
|
closure_ratio_bob = bob_closure_count as f64 / bob_entry_count as f64,
|
|
"Closure table counts vs actual entries"
|
|
);
|
|
|
|
let closure_diff = (alice_closure_count as i64 - bob_closure_count as i64).abs();
|
|
|
|
// TODO: Fix parent ordering issue causing ~60% of entries to be stuck in dependency tracker
|
|
// For now, allow larger tolerance to test other assertions
|
|
let closure_diff_pct = (closure_diff as f64 / alice_closure_count as f64) * 100.0;
|
|
if closure_diff_pct > 10.0 {
|
|
tracing::warn!(
|
|
"Closure table mismatch: Alice {}, Bob {} (diff: {}, {:.1}% missing)",
|
|
alice_closure_count,
|
|
bob_closure_count,
|
|
closure_diff,
|
|
closure_diff_pct
|
|
);
|
|
tracing::warn!("This indicates parent directories are syncing out of order - entries stuck in dependency tracker");
|
|
}
|
|
|
|
// Sample check: find a directory and verify its descendants match
|
|
let sample_dir = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Name.eq("core"))
|
|
.filter(entities::entry::Column::Kind.eq(1))
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!("Could not find 'core' directory for closure verification")
|
|
})?;
|
|
|
|
let sample_uuid = sample_dir
|
|
.uuid
|
|
.ok_or_else(|| anyhow::anyhow!("Directory missing UUID"))?;
|
|
|
|
// Find corresponding directory on Bob by UUID
|
|
let bob_sample_dir = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(sample_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Bob missing directory with UUID {}", sample_uuid))?;
|
|
|
|
// Count descendants for this directory on Alice
|
|
let alice_descendants = entities::entry_closure::Entity::find()
|
|
.filter(entities::entry_closure::Column::AncestorId.eq(sample_dir.id))
|
|
.filter(entities::entry_closure::Column::Depth.gt(0)) // Exclude self-reference
|
|
.count(library_alice.db().conn())
|
|
.await?;
|
|
|
|
// Count descendants for this directory on Bob
|
|
let bob_descendants = entities::entry_closure::Entity::find()
|
|
.filter(entities::entry_closure::Column::AncestorId.eq(bob_sample_dir.id))
|
|
.filter(entities::entry_closure::Column::Depth.gt(0))
|
|
.count(library_bob.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
dir_name = sample_dir.name,
|
|
alice_descendants = alice_descendants,
|
|
bob_descendants = bob_descendants,
|
|
"Descendant count verification for sample directory"
|
|
);
|
|
|
|
let descendant_diff = (alice_descendants as i64 - bob_descendants as i64).abs();
|
|
assert!(
|
|
descendant_diff <= 5,
|
|
"Descendant count mismatch for '{}': Alice {}, Bob {} (diff: {})",
|
|
sample_dir.name,
|
|
alice_descendants,
|
|
bob_descendants,
|
|
descendant_diff
|
|
);
|
|
|
|
tracing::info!("✅ Closure table integrity verified");
|
|
Ok(())
|
|
}
|
|
|
|
/// Verify parent-child relationships match between Alice and Bob
|
|
async fn verify_parent_child_relationships(
|
|
library_alice: &Arc<sd_core::library::Library>,
|
|
library_bob: &Arc<sd_core::library::Library>,
|
|
) -> anyhow::Result<()> {
|
|
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
|
|
|
|
tracing::info!("Verifying parent-child relationships...");
|
|
|
|
// Find a directory with children
|
|
let parent_dir = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Kind.eq(1)) // Directory
|
|
.filter(entities::entry::Column::ChildCount.gt(0))
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("No directory with children found for relationship test"))?;
|
|
|
|
let parent_uuid = parent_dir
|
|
.uuid
|
|
.ok_or_else(|| anyhow::anyhow!("Parent directory missing UUID"))?;
|
|
|
|
// Find children on Alice
|
|
let alice_children = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::ParentId.eq(parent_dir.id))
|
|
.all(library_alice.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
parent_name = parent_dir.name,
|
|
child_count = alice_children.len(),
|
|
"Found parent directory with children on Alice"
|
|
);
|
|
|
|
// Find the same parent on Bob by UUID
|
|
let bob_parent = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(parent_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Bob missing parent directory with UUID {}", parent_uuid))?;
|
|
|
|
// Verify child_count matches
|
|
assert_eq!(
|
|
parent_dir.child_count, bob_parent.child_count,
|
|
"Child count mismatch for '{}': Alice {}, Bob {}",
|
|
parent_dir.name, parent_dir.child_count, bob_parent.child_count
|
|
);
|
|
|
|
// Find children on Bob
|
|
let bob_children = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::ParentId.eq(bob_parent.id))
|
|
.all(library_bob.db().conn())
|
|
.await?;
|
|
|
|
assert_eq!(
|
|
alice_children.len(),
|
|
bob_children.len(),
|
|
"Actual children count mismatch for '{}': Alice {}, Bob {}",
|
|
parent_dir.name,
|
|
alice_children.len(),
|
|
bob_children.len()
|
|
);
|
|
|
|
// Verify each child exists on Bob with matching UUID
|
|
for alice_child in &alice_children {
|
|
let child_uuid = alice_child
|
|
.uuid
|
|
.ok_or_else(|| anyhow::anyhow!("Child entry missing UUID: {}", alice_child.name))?;
|
|
|
|
let bob_child = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(child_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"Bob missing child entry with UUID {} (name: {})",
|
|
child_uuid,
|
|
alice_child.name
|
|
)
|
|
})?;
|
|
|
|
assert_eq!(
|
|
alice_child.name, bob_child.name,
|
|
"Child name mismatch for UUID {}: Alice '{}', Bob '{}'",
|
|
child_uuid, alice_child.name, bob_child.name
|
|
);
|
|
|
|
// Verify the parent_id points to Bob's version of the parent
|
|
assert_eq!(
|
|
bob_child.parent_id,
|
|
Some(bob_parent.id),
|
|
"Child '{}' has wrong parent_id on Bob: expected {}, got {:?}",
|
|
bob_child.name,
|
|
bob_parent.id,
|
|
bob_child.parent_id
|
|
);
|
|
}
|
|
|
|
tracing::info!("✅ Parent-child relationships verified");
|
|
Ok(())
|
|
}
|
|
|
|
/// Verify file metadata matches for sample files
|
|
async fn verify_file_metadata_accuracy(
|
|
library_alice: &Arc<sd_core::library::Library>,
|
|
library_bob: &Arc<sd_core::library::Library>,
|
|
) -> anyhow::Result<()> {
|
|
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
|
|
|
|
tracing::info!("Verifying file metadata accuracy...");
|
|
|
|
// Find sample files (limit to 10 for performance)
|
|
let sample_files = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Kind.eq(0)) // File
|
|
.filter(entities::entry::Column::Uuid.is_not_null())
|
|
.limit(10)
|
|
.all(library_alice.db().conn())
|
|
.await?;
|
|
|
|
tracing::info!(
|
|
sample_count = sample_files.len(),
|
|
"Verifying metadata for sample files"
|
|
);
|
|
|
|
for alice_file in sample_files {
|
|
let file_uuid = alice_file
|
|
.uuid
|
|
.ok_or_else(|| anyhow::anyhow!("File missing UUID: {}", alice_file.name))?;
|
|
|
|
let bob_file = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(file_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"Bob missing file with UUID {} (name: {})",
|
|
file_uuid,
|
|
alice_file.name
|
|
)
|
|
})?;
|
|
|
|
// Verify name matches
|
|
assert_eq!(
|
|
alice_file.name, bob_file.name,
|
|
"File name mismatch for UUID {}: Alice '{}', Bob '{}'",
|
|
file_uuid, alice_file.name, bob_file.name
|
|
);
|
|
|
|
// Verify size matches
|
|
assert_eq!(
|
|
alice_file.size, bob_file.size,
|
|
"File size mismatch for '{}': Alice {}, Bob {}",
|
|
alice_file.name, alice_file.size, bob_file.size
|
|
);
|
|
|
|
// Verify kind matches
|
|
assert_eq!(
|
|
alice_file.kind, bob_file.kind,
|
|
"File kind mismatch for '{}': Alice {}, Bob {}",
|
|
alice_file.name, alice_file.kind, bob_file.kind
|
|
);
|
|
|
|
// Verify extension matches
|
|
assert_eq!(
|
|
alice_file.extension, bob_file.extension,
|
|
"File extension mismatch for '{}': Alice '{:?}', Bob '{:?}'",
|
|
alice_file.name, alice_file.extension, bob_file.extension
|
|
);
|
|
|
|
// Verify content_id linkage matches (if present)
|
|
if alice_file.content_id.is_some() {
|
|
assert!(
|
|
bob_file.content_id.is_some(),
|
|
"File '{}' has content_id on Alice but not on Bob",
|
|
alice_file.name
|
|
);
|
|
|
|
// Find the content identity UUIDs to compare
|
|
if let Some(alice_cid) = alice_file.content_id {
|
|
if let Some(bob_cid) = bob_file.content_id {
|
|
let alice_content = entities::content_identity::Entity::find()
|
|
.filter(entities::content_identity::Column::Id.eq(alice_cid))
|
|
.one(library_alice.db().conn())
|
|
.await?;
|
|
|
|
let bob_content = entities::content_identity::Entity::find()
|
|
.filter(entities::content_identity::Column::Id.eq(bob_cid))
|
|
.one(library_bob.db().conn())
|
|
.await?;
|
|
|
|
if let (Some(alice_ci), Some(bob_ci)) = (alice_content, bob_content) {
|
|
assert_eq!(
|
|
alice_ci.uuid, bob_ci.uuid,
|
|
"Content identity UUID mismatch for file '{}': Alice {:?}, Bob {:?}",
|
|
alice_file.name, alice_ci.uuid, bob_ci.uuid
|
|
);
|
|
|
|
assert_eq!(
|
|
alice_ci.content_hash, bob_ci.content_hash,
|
|
"Content hash mismatch for file '{}': Alice '{}', Bob '{}'",
|
|
alice_file.name, alice_ci.content_hash, bob_ci.content_hash
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
tracing::debug!(
|
|
file_name = alice_file.name,
|
|
uuid = %file_uuid,
|
|
size = alice_file.size,
|
|
"File metadata verified"
|
|
);
|
|
}
|
|
|
|
tracing::info!("✅ File metadata accuracy verified");
|
|
Ok(())
|
|
}
|
|
|
|
/// Verify nested file structure and ancestor chains
|
|
async fn verify_nested_file_structure(
|
|
library_alice: &Arc<sd_core::library::Library>,
|
|
library_bob: &Arc<sd_core::library::Library>,
|
|
) -> anyhow::Result<()> {
|
|
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
|
|
|
|
tracing::info!("Verifying nested file structure and ancestor chains...");
|
|
|
|
// Find files nested at least 2 levels deep (has parent with parent)
|
|
let alice_entries = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Kind.eq(0)) // Files only
|
|
.filter(entities::entry::Column::ParentId.is_not_null())
|
|
.limit(20)
|
|
.all(library_alice.db().conn())
|
|
.await?;
|
|
|
|
let mut verified_count = 0;
|
|
let mut nested_files_checked = 0;
|
|
|
|
for alice_file in alice_entries {
|
|
// Walk up the parent chain to verify depth
|
|
let mut current_id = alice_file.parent_id;
|
|
let mut depth = 0;
|
|
|
|
while let Some(parent_id) = current_id {
|
|
let parent = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Id.eq(parent_id))
|
|
.one(library_alice.db().conn())
|
|
.await?;
|
|
|
|
if let Some(p) = parent {
|
|
current_id = p.parent_id;
|
|
depth += 1;
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Only test files that are at least 2 levels deep
|
|
if depth < 2 {
|
|
continue;
|
|
}
|
|
|
|
nested_files_checked += 1;
|
|
|
|
let file_uuid = match alice_file.uuid {
|
|
Some(uuid) => uuid,
|
|
None => {
|
|
tracing::warn!("Nested file missing UUID: {}", alice_file.name);
|
|
continue;
|
|
}
|
|
};
|
|
|
|
// Find the same file on Bob
|
|
let bob_file = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(file_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?;
|
|
|
|
let bob_file = match bob_file {
|
|
Some(f) => f,
|
|
None => {
|
|
anyhow::bail!(
|
|
"Bob missing nested file with UUID {} (name: {}, depth: {})",
|
|
file_uuid,
|
|
alice_file.name,
|
|
depth
|
|
);
|
|
}
|
|
};
|
|
|
|
tracing::debug!(
|
|
file_name = alice_file.name,
|
|
depth = depth,
|
|
uuid = %file_uuid,
|
|
"Found nested file to verify"
|
|
);
|
|
|
|
// Walk up Alice's parent chain and collect ancestor UUIDs
|
|
let mut alice_ancestor_uuids = Vec::new();
|
|
let mut current_parent_id = alice_file.parent_id;
|
|
|
|
while let Some(parent_id) = current_parent_id {
|
|
let parent = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Id.eq(parent_id))
|
|
.one(library_alice.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Alice parent not found: id {}", parent_id))?;
|
|
|
|
if let Some(parent_uuid) = parent.uuid {
|
|
alice_ancestor_uuids.push((parent.name.clone(), parent_uuid));
|
|
}
|
|
|
|
current_parent_id = parent.parent_id;
|
|
}
|
|
|
|
// Walk up Bob's parent chain and collect ancestor UUIDs
|
|
let mut bob_ancestor_uuids = Vec::new();
|
|
let mut current_parent_id = bob_file.parent_id;
|
|
|
|
while let Some(parent_id) = current_parent_id {
|
|
let parent = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Id.eq(parent_id))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| anyhow::anyhow!("Bob parent not found: id {}", parent_id))?;
|
|
|
|
if let Some(parent_uuid) = parent.uuid {
|
|
bob_ancestor_uuids.push((parent.name.clone(), parent_uuid));
|
|
}
|
|
|
|
current_parent_id = parent.parent_id;
|
|
}
|
|
|
|
// Verify the ancestor chains match
|
|
assert_eq!(
|
|
alice_ancestor_uuids.len(),
|
|
bob_ancestor_uuids.len(),
|
|
"Ancestor chain length mismatch for file '{}': Alice has {} ancestors, Bob has {}",
|
|
alice_file.name,
|
|
alice_ancestor_uuids.len(),
|
|
bob_ancestor_uuids.len()
|
|
);
|
|
|
|
for (i, ((alice_name, alice_uuid), (bob_name, bob_uuid))) in alice_ancestor_uuids
|
|
.iter()
|
|
.zip(bob_ancestor_uuids.iter())
|
|
.enumerate()
|
|
{
|
|
assert_eq!(
|
|
alice_uuid, bob_uuid,
|
|
"Ancestor UUID mismatch at level {} for file '{}': Alice has '{}' ({}), Bob has '{}' ({})",
|
|
i,
|
|
alice_file.name,
|
|
alice_name,
|
|
alice_uuid,
|
|
bob_name,
|
|
bob_uuid
|
|
);
|
|
}
|
|
|
|
// Verify closure table has all ancestor relationships on Bob
|
|
for (_ancestor_name, ancestor_uuid) in &alice_ancestor_uuids {
|
|
// Find ancestor entry on Bob by UUID
|
|
let bob_ancestor = entities::entry::Entity::find()
|
|
.filter(entities::entry::Column::Uuid.eq(*ancestor_uuid))
|
|
.one(library_bob.db().conn())
|
|
.await?
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"Bob missing ancestor with UUID {} for file '{}'",
|
|
ancestor_uuid,
|
|
alice_file.name
|
|
)
|
|
})?;
|
|
|
|
// Verify closure table entry exists
|
|
let closure_entry = entities::entry_closure::Entity::find()
|
|
.filter(entities::entry_closure::Column::AncestorId.eq(bob_ancestor.id))
|
|
.filter(entities::entry_closure::Column::DescendantId.eq(bob_file.id))
|
|
.one(library_bob.db().conn())
|
|
.await?;
|
|
|
|
assert!(
|
|
closure_entry.is_some(),
|
|
"Closure table missing entry on Bob: ancestor '{}' ({}) -> descendant '{}' ({})",
|
|
bob_ancestor.name,
|
|
bob_ancestor.id,
|
|
bob_file.name,
|
|
bob_file.id
|
|
);
|
|
}
|
|
|
|
verified_count += 1;
|
|
|
|
tracing::debug!(
|
|
file_name = alice_file.name,
|
|
depth = depth,
|
|
ancestor_count = alice_ancestor_uuids.len(),
|
|
"Nested file structure verified"
|
|
);
|
|
|
|
// Stop after verifying 5 nested files to keep test time reasonable
|
|
if verified_count >= 5 {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// If we found nested files, verify they synced correctly
|
|
// If no nested files found, that's OK - the closure table rebuild proves parent relationships work
|
|
if nested_files_checked > 0 {
|
|
assert!(
|
|
verified_count > 0,
|
|
"Found {} nested files but couldn't verify any of them",
|
|
nested_files_checked
|
|
);
|
|
tracing::info!(
|
|
verified_count = verified_count,
|
|
nested_files_checked = nested_files_checked,
|
|
"Verified nested file structure"
|
|
);
|
|
} else {
|
|
tracing::warn!(
|
|
"No nested files found to verify, but closure table rebuild proves parent relationships work"
|
|
);
|
|
}
|
|
|
|
Ok(())
|
|
}
|