mirror of
https://github.com/spacedriveapp/spacedrive.git
synced 2026-05-18 21:36:56 -04:00
refactor: Remove obsolete implementation progress and TODO documentation
- Deleted the `IMPLEMENTATION_PROGRESS.md` and `IMPLEMENTATION_TODO.md` files as they are no longer needed following the completion of the initial implementation phases. - Updated the sync module to include a new `transport` module for network communication, enhancing the overall architecture. - Cleaned up the `transaction` module by removing deprecated methods related to the old leader-based approach, streamlining the codebase for the leaderless sync model. - Added a new endpoint method in the `NetworkingService` for improved network communication handling.
This commit is contained in:
Jamie Pine
@@ -1,353 +0,0 @@
|
||||
# Leaderless Sync - Implementation Progress
|
||||
|
||||
**Date**: 2025-10-09
|
||||
**Status**: Phase 1 Started - Model Apply Functions ✅ (2/7 models)
|
||||
|
||||
---
|
||||
|
||||
## ✅ What We Just Implemented
|
||||
|
||||
### 1. Location Model (Device-Owned) ✅
|
||||
|
||||
**File**: `core/src/infra/db/entities/location.rs`
|
||||
|
||||
**Added**:
|
||||
- `apply_state_change()` method for idempotent state-based replication
|
||||
- Comprehensive documentation on device-owned sync strategy
|
||||
- Unit tests verifying Syncable trait implementation
|
||||
|
||||
**Implementation Details**:
|
||||
```rust
|
||||
impl Model {
|
||||
pub async fn apply_state_change(
|
||||
data: serde_json::Value,
|
||||
db: &DatabaseConnection,
|
||||
) -> Result<(), sea_orm::DbErr>
|
||||
}
|
||||
```
|
||||
|
||||
**Key Features**:
|
||||
- ✅ Idempotent upsert by UUID
|
||||
- ✅ No HLC needed (device-owned = no conflicts)
|
||||
- ✅ Proper field exclusions (id, scan_state, error_message, timestamps)
|
||||
- ✅ Foreign key handling (device_id, entry_id)
|
||||
- ✅ Local state reset (scan_state → "pending")
|
||||
|
||||
**Lines Added**: ~65 lines implementation + 5 lines tests
|
||||
|
||||
---
|
||||
|
||||
### 2. Tag Model (Shared Resource) ✅
|
||||
|
||||
**File**: `core/src/infra/db/entities/tag.rs`
|
||||
|
||||
**Added**:
|
||||
- `apply_shared_change()` method with union merge conflict resolution
|
||||
- Complete Syncable trait implementation
|
||||
- Documentation on polymorphic naming and conflict resolution
|
||||
- Unit tests for sync behavior and polymorphic naming
|
||||
|
||||
**Implementation Details**:
|
||||
```rust
|
||||
impl Model {
|
||||
pub async fn apply_shared_change(
|
||||
entry: SharedChangeEntry,
|
||||
db: &DatabaseConnection,
|
||||
) -> Result<(), sea_orm::DbErr>
|
||||
}
|
||||
|
||||
impl Syncable for Model {
|
||||
const SYNC_MODEL: &'static str = "tag";
|
||||
fn sync_id(&self) -> Uuid { self.uuid }
|
||||
fn version(&self) -> i64 { 1 }
|
||||
fn exclude_fields() -> Option<&'static [&'static str]> { ... }
|
||||
}
|
||||
```
|
||||
|
||||
**Key Features**:
|
||||
- ✅ Union merge: Multiple tags with same canonical_name preserved
|
||||
- ✅ Polymorphic naming via namespace differentiation
|
||||
- ✅ Last-writer-wins for properties (HLC ordering)
|
||||
- ✅ Delete handling via tombstone records
|
||||
- ✅ Proper field exclusions (id, timestamps)
|
||||
- ✅ Created-by attribution tracking
|
||||
|
||||
**Lines Added**: ~135 lines implementation + 75 lines tests
|
||||
|
||||
---
|
||||
|
||||
## 📊 Current Progress
|
||||
|
||||
### Task #2: Model-Specific Apply Functions
|
||||
**Status**: 2/7 models complete (29%)
|
||||
|
||||
| Model | Type | Status | Complexity | Next Step |
|
||||
|-------|------|--------|------------|-----------|
|
||||
| **location** | Device-owned | ✅ Done | Low | - |
|
||||
| **tag** | Shared | ✅ Done | Medium | - |
|
||||
| **entry** | Device-owned | ⚠️ Todo | Low | Similar to location |
|
||||
| **volume** | Device-owned | ⚠️ Todo | Low | Similar to location |
|
||||
| **device** | Device-owned (self) | ⚠️ Todo | Medium | Special broadcast |
|
||||
| **album** | Shared | ⚠️ Todo | Medium | Similar to tag |
|
||||
| **user_metadata** | Mixed | ⚠️ Todo | High | Context-dependent |
|
||||
|
||||
**Total Implemented**: ~200 lines
|
||||
**Remaining**: ~150 lines (5 models × 30 lines average)
|
||||
|
||||
---
|
||||
|
||||
## 🎯 What Works Now
|
||||
|
||||
### Location (Device-Owned)
|
||||
```rust
|
||||
// Receiving a location state change
|
||||
let location_data = json!({
|
||||
"uuid": "abc-123",
|
||||
"device_id": 1,
|
||||
"entry_id": 42,
|
||||
"name": "Photos",
|
||||
"index_mode": "deep",
|
||||
"total_file_count": 10000,
|
||||
"total_byte_size": 5000000000
|
||||
});
|
||||
|
||||
location::Model::apply_state_change(location_data, &db).await?;
|
||||
// ✅ Location inserted or updated idempotently
|
||||
// ✅ Local state reset (scan_state = "pending")
|
||||
// ✅ Timestamps regenerated locally
|
||||
```
|
||||
|
||||
### Tag (Shared Resource)
|
||||
```rust
|
||||
// Receiving a tag shared change
|
||||
let tag_entry = SharedChangeEntry {
|
||||
hlc: HLC::new(...),
|
||||
model_type: "tag".to_string(),
|
||||
record_uuid: Uuid::new_v4(),
|
||||
change_type: ChangeType::Insert,
|
||||
data: json!({
|
||||
"uuid": "tag-uuid",
|
||||
"canonical_name": "vacation",
|
||||
"display_name": "Vacation",
|
||||
"namespace": "travel",
|
||||
"tag_type": "standard",
|
||||
"privacy_level": "normal",
|
||||
"search_weight": 100
|
||||
}),
|
||||
};
|
||||
|
||||
tag::Model::apply_shared_change(tag_entry, &db).await?;
|
||||
// ✅ Tag inserted or updated by UUID
|
||||
// ✅ Different UUIDs with same name = different tags (union merge)
|
||||
// ✅ Same UUID = last-writer-wins for properties
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🚀 Next Steps (Priority Order)
|
||||
|
||||
### Immediate Next: Complete Remaining Models (1-2 hours)
|
||||
|
||||
**Easy Wins** (Copy & adapt from location):
|
||||
1. **Entry** (~30 lines) - Device-owned, similar structure
|
||||
2. **Volume** (~30 lines) - Device-owned, simpler than location
|
||||
3. **Device** (~40 lines) - Device-owned (self), special handling
|
||||
|
||||
**Medium Effort** (Copy & adapt from tag):
|
||||
4. **Album** (~35 lines) - Shared, union merge like tag
|
||||
5. **UserMetadata** (~50 lines) - Mixed strategy based on scope
|
||||
|
||||
### After Models Complete: Registry Function Pointers (Task #3)
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
// In infra/sync/registry.rs
|
||||
pub struct SyncableModelRegistration {
|
||||
pub model_type: &'static str,
|
||||
pub table_name: &'static str,
|
||||
pub is_device_owned: bool,
|
||||
|
||||
// NEW: Function pointers for polymorphic dispatch
|
||||
pub apply_fn: ApplyFn, // ← Add this
|
||||
}
|
||||
|
||||
// Register models with apply functions
|
||||
registry.register("location", "locations", true, location::Model::apply_state_change);
|
||||
registry.register("tag", "tag", false, tag::Model::apply_shared_change);
|
||||
```
|
||||
|
||||
**Effort**: ~50 lines in registry + ~10 lines per model registration
|
||||
**Complexity**: Medium (async function pointers)
|
||||
|
||||
### Then: Network Message Integration (Task #1)
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
// In service/sync/peer.rs
|
||||
impl PeerSync {
|
||||
async fn broadcast_state_change(&self, change: StateChangeMessage) {
|
||||
let partners = query_sync_partners(self.library_id).await?;
|
||||
|
||||
for partner in partners {
|
||||
self.networking.send_message(
|
||||
partner.remote_device_id,
|
||||
"sync",
|
||||
SyncMessage::StateChange(change.clone())
|
||||
).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~100 lines
|
||||
**Complexity**: Medium (networking integration)
|
||||
|
||||
---
|
||||
|
||||
## 🧪 Testing Strategy
|
||||
|
||||
### Unit Tests (Already Added) ✅
|
||||
- ✅ `location::tests::test_location_syncable()` - Verifies Syncable trait
|
||||
- ✅ `tag::tests::test_tag_syncable()` - Verifies Syncable trait
|
||||
- ✅ `tag::tests::test_tag_polymorphic_naming()` - Verifies union merge
|
||||
|
||||
### Integration Tests (Next Step)
|
||||
**File**: `core/tests/sync/model_apply_test.rs` (create this)
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn test_location_apply_idempotent() {
|
||||
let db = setup_test_db().await;
|
||||
|
||||
let location_json = json!({ /* ... */ });
|
||||
|
||||
// Apply twice
|
||||
location::Model::apply_state_change(location_json.clone(), &db).await.unwrap();
|
||||
location::Model::apply_state_change(location_json, &db).await.unwrap();
|
||||
|
||||
// Verify only one record
|
||||
let count = location::Entity::find().count(&db).await.unwrap();
|
||||
assert_eq!(count, 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_tag_union_merge() {
|
||||
let db = setup_test_db().await;
|
||||
|
||||
// Two devices create "vacation" tag with different UUIDs
|
||||
let tag1_entry = SharedChangeEntry { /* Device A */ };
|
||||
let tag2_entry = SharedChangeEntry { /* Device B */ };
|
||||
|
||||
tag::Model::apply_shared_change(tag1_entry, &db).await.unwrap();
|
||||
tag::Model::apply_shared_change(tag2_entry, &db).await.unwrap();
|
||||
|
||||
// Verify both tags exist
|
||||
let tags = tag::Entity::find().all(&db).await.unwrap();
|
||||
assert_eq!(tags.len(), 2);
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📝 Code Quality
|
||||
|
||||
### ✅ Verified
|
||||
- ✅ **Build**: `cargo check --lib` passes
|
||||
- ✅ **Lints**: No clippy warnings
|
||||
- ✅ **Format**: Follows Spacedrive code style (tabs, no emojis)
|
||||
- ✅ **Documentation**: Comprehensive doc comments
|
||||
- ✅ **Error Handling**: Proper Result types with context
|
||||
- ✅ **Tests**: Unit tests for core behavior
|
||||
|
||||
### 📋 Checklist for Remaining Models
|
||||
- [ ] Follow location pattern for device-owned
|
||||
- [ ] Follow tag pattern for shared
|
||||
- [ ] Add comprehensive doc comments
|
||||
- [ ] Add unit tests
|
||||
- [ ] Verify with `cargo check`
|
||||
- [ ] Test idempotency
|
||||
|
||||
---
|
||||
|
||||
## 🎓 Key Learnings
|
||||
|
||||
### Device-Owned Sync (Location)
|
||||
1. **No HLC needed** - Only owner modifies, no conflicts
|
||||
2. **State-based** - Just broadcast current state
|
||||
3. **Idempotent upsert** - `ON CONFLICT (uuid) DO UPDATE`
|
||||
4. **Reset local state** - scan_state, error_message not synced
|
||||
|
||||
### Shared Resource Sync (Tag)
|
||||
1. **Union merge** - Preserve all tags with different UUIDs
|
||||
2. **Polymorphic naming** - Same name, different contexts (namespace)
|
||||
3. **Last-writer-wins** - Properties updated on same UUID
|
||||
4. **HLC ordering** - Applied in causal order (handled upstream)
|
||||
5. **Tombstone deletes** - Explicit ChangeType::Delete records
|
||||
|
||||
### General Pattern
|
||||
1. **Deserialize** - JSON → Model via serde
|
||||
2. **Build ActiveModel** - Set() for synced fields, NotSet for id
|
||||
3. **Upsert** - insert().on_conflict(uuid).update_columns()
|
||||
4. **Error handling** - Map serde/db errors to DbErr
|
||||
|
||||
---
|
||||
|
||||
## 📈 Estimated Timeline
|
||||
|
||||
**Remaining Work**: ~3-4 days for full MVP
|
||||
|
||||
| Phase | Tasks | Effort | Status |
|
||||
|-------|-------|--------|--------|
|
||||
| **Phase 1a** | Remaining 5 models | 2-3 hours | ⚠️ In Progress (40% done) |
|
||||
| **Phase 1b** | Registry function pointers | 2 hours | ⚠️ Blocked |
|
||||
| **Phase 1c** | Network message integration | 3 hours | ⚠️ Blocked |
|
||||
| **Phase 1d** | TransactionManager integration | 3 hours | ⚠️ Blocked |
|
||||
| **Phase 2** | Protocol handler wiring | 4 hours | ⚠️ Blocked |
|
||||
| **Phase 2b** | Background tasks | 2 hours | ⚠️ Blocked |
|
||||
| **Phase 2c** | Backfill network requests | 2 hours | ⚠️ Blocked |
|
||||
| **Phase 3** | Testing & polish | 4 hours | ⚠️ Not started |
|
||||
|
||||
**Total**: ~22 hours (~3 days of focused work)
|
||||
|
||||
---
|
||||
|
||||
## 🔗 References
|
||||
|
||||
- **Architecture Doc**: `/docs/core/sync.md`
|
||||
- **Implementation TODO**: `/core/src/infra/sync/IMPLEMENTATION_TODO.md`
|
||||
- **Leaderless Design**: `/docs/core/sync/leaderless-architecture.md`
|
||||
- **Location Entity**: `/core/src/infra/db/entities/location.rs`
|
||||
- **Tag Entity**: `/core/src/infra/db/entities/tag.rs`
|
||||
- **PeerLog**: `/core/src/infra/sync/peer_log.rs`
|
||||
- **Syncable Trait**: `/core/src/infra/sync/syncable.rs`
|
||||
|
||||
---
|
||||
|
||||
## 🎯 Success Criteria for MVP
|
||||
|
||||
- [x] Location apply works ✅
|
||||
- [x] Tag apply works ✅
|
||||
- [ ] All 7 models can apply sync changes
|
||||
- [ ] Registry can dispatch to apply functions
|
||||
- [ ] Network messages send/receive
|
||||
- [ ] TransactionManager triggers broadcasts
|
||||
- [ ] Integration test: 2 devices sync location
|
||||
- [ ] Integration test: 2 devices sync tag with conflict
|
||||
|
||||
**Current**: 2/8 success criteria met (25%)
|
||||
|
||||
---
|
||||
|
||||
## 💡 Next Command to Run
|
||||
|
||||
```bash
|
||||
# Continue implementing remaining models
|
||||
cd /Users/jamespine/Projects/spacedrive/core
|
||||
|
||||
# Start with entry (similar to location)
|
||||
# Open: src/infra/db/entities/entry.rs
|
||||
# Add: impl Model { pub async fn apply_state_change(...) }
|
||||
```
|
||||
|
||||
Or ask: "Implement entry model next" to continue!
|
||||
|
||||
@@ -1,782 +0,0 @@
|
||||
# Leaderless Sync - Implementation TODO Map
|
||||
|
||||
**Status**: Architecture Complete ✅, Implementation Stubs Remaining
|
||||
**Last Updated**: 2025-10-08
|
||||
**Build**: Compiles successfully
|
||||
|
||||
---
|
||||
|
||||
## 🎯 Critical Path (Must Have for MVP)
|
||||
|
||||
### 1. Network Message Integration (HIGH PRIORITY)
|
||||
|
||||
**Location**: `service/sync/peer.rs`
|
||||
|
||||
**Current State**: Methods exist but don't send to network
|
||||
```rust
|
||||
// Line 135, 182
|
||||
// TODO: Send to all sync_partners via network protocol
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
impl PeerSync {
|
||||
async fn broadcast_state_change(&self, change: StateChangeMessage) {
|
||||
// Get sync_partners from database
|
||||
let partners = query_sync_partners(self.library_id).await?;
|
||||
|
||||
// Send to each via NetworkingService
|
||||
for partner in partners {
|
||||
networking.send_message(
|
||||
partner.remote_device_id,
|
||||
"sync",
|
||||
SyncMessage::StateChange { ... }
|
||||
).await?;
|
||||
}
|
||||
}
|
||||
|
||||
async fn broadcast_shared_change(&self, entry: SharedChangeEntry) {
|
||||
// Same pattern for SharedChange messages
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Dependencies**:
|
||||
- Access to NetworkingService in PeerSync
|
||||
- `sync_partners` table querying
|
||||
- Message serialization already done ✅
|
||||
|
||||
**Effort**: ~100 lines
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
### 2. Model-Specific Apply Functions (HIGH PRIORITY)
|
||||
|
||||
**Location**: `service/sync/protocol_handler.rs` + individual model files
|
||||
|
||||
**Current State**: Generic registry dispatch, but models don't implement apply yet
|
||||
```rust
|
||||
// Line 260, 237
|
||||
// TODO: Deserialize and upsert based on model_type
|
||||
// TODO: Deserialize and merge based on model_type
|
||||
```
|
||||
|
||||
**What's Needed** (per model):
|
||||
|
||||
**Example for Location (Device-Owned)**:
|
||||
```rust
|
||||
// In infra/db/entities/location.rs
|
||||
impl Model {
|
||||
pub async fn apply_state_change(
|
||||
data: serde_json::Value,
|
||||
db: &DatabaseConnection
|
||||
) -> Result<()> {
|
||||
let location: Self = serde_json::from_value(data)?;
|
||||
|
||||
// Idempotent upsert
|
||||
location::ActiveModel::from(location)
|
||||
.insert_or_update(db)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Example for Tag (Shared, with conflict resolution)**:
|
||||
```rust
|
||||
// In infra/db/entities/tag.rs
|
||||
impl Model {
|
||||
pub async fn apply_shared_change(
|
||||
entry: SharedChangeEntry,
|
||||
db: &DatabaseConnection
|
||||
) -> Result<()> {
|
||||
let tag: Self = serde_json::from_value(entry.data)?;
|
||||
|
||||
// Semantic tags use random UUIDs, not deterministic
|
||||
// Tags with same canonical_name are allowed (polymorphic naming)
|
||||
|
||||
// Check if this exact tag UUID exists
|
||||
if let Some(existing) = find_by_uuid(tag.uuid, db).await? {
|
||||
// Update existing tag (last-writer-wins for properties)
|
||||
update_tag(existing, tag, entry.hlc, db).await?;
|
||||
} else {
|
||||
// Insert new tag (preserves all tags via union merge)
|
||||
tag.insert(db).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Models to Implement**:
|
||||
- ✅ location (device-owned, state-based) - **IMPLEMENTED**
|
||||
- ⚠️ entry (device-owned, state-based)
|
||||
- ⚠️ volume (device-owned, state-based)
|
||||
- ⚠️ device (special: each broadcasts its own)
|
||||
- ✅ tag (shared, log-based, union merge) - **IMPLEMENTED**
|
||||
- ⚠️ album (shared, log-based, union merge)
|
||||
- ⚠️ user_metadata (shared, log-based, LWW)
|
||||
|
||||
**Effort**: ~50 lines per model (~350 lines total)
|
||||
**Complexity**: Medium-High (need conflict resolution)
|
||||
|
||||
**Status**: 2/7 models implemented (location, tag)
|
||||
|
||||
---
|
||||
|
||||
### 3. Registry Function Pointers (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `infra/sync/registry.rs`
|
||||
|
||||
**Current State**: Registry tracks metadata but can't call apply functions
|
||||
```rust
|
||||
// Line 28-29
|
||||
// TODO: Function pointers for serialize/deserialize/apply
|
||||
// Will be implemented when we wire up full sync
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
type ApplyFn = fn(serde_json::Value, &DatabaseConnection) -> BoxFuture<'static, Result<()>>;
|
||||
|
||||
pub struct SyncableModelRegistration {
|
||||
pub model_type: &'static str,
|
||||
pub table_name: &'static str,
|
||||
pub is_device_owned: bool,
|
||||
|
||||
// NEW:
|
||||
pub apply_fn: ApplyFn, // Polymorphic dispatch!
|
||||
}
|
||||
|
||||
// Then in apply_sync_entry:
|
||||
pub async fn apply_sync_entry(
|
||||
model_type: &str,
|
||||
data: serde_json::Value,
|
||||
) -> Result<()> {
|
||||
let registry = SYNCABLE_REGISTRY.read().unwrap();
|
||||
let registration = registry.get(model_type)
|
||||
.ok_or_else(|| anyhow!("Unknown model: {}", model_type))?;
|
||||
|
||||
// Call the registered function!
|
||||
(registration.apply_fn)(data, db).await
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~50 lines in registry, ~20 lines per model registration
|
||||
**Complexity**: Medium (async function pointers are tricky)
|
||||
|
||||
---
|
||||
|
||||
### 4. TransactionManager HLC Integration (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `infra/sync/transaction.rs`
|
||||
|
||||
**Current State**: Methods exist but are stubs
|
||||
```rust
|
||||
// Line 118-150
|
||||
// TODO: Implement
|
||||
// 1. Verify model.is_device_owned()
|
||||
// 2. Generate HLC
|
||||
// 3. Write to peer_log
|
||||
// 4. Emit event for broadcast
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
pub async fn commit_device_owned<M>(
|
||||
&self,
|
||||
library: &Library,
|
||||
model: M,
|
||||
) -> Result<M>
|
||||
where
|
||||
M: Syncable,
|
||||
{
|
||||
// 1. Verify classification
|
||||
if !model.is_device_owned() {
|
||||
return Err(TxError::InvalidModel("Expected device-owned".into()));
|
||||
}
|
||||
|
||||
// 2. Write to database (no log!)
|
||||
let saved = model.insert(library.db()).await?;
|
||||
|
||||
// 3. Emit event → PeerSync picks up and broadcasts
|
||||
self.event_bus.emit(Event::StateChanged {
|
||||
library_id: library.id(),
|
||||
model_type: M::SYNC_MODEL,
|
||||
record_uuid: saved.sync_id(),
|
||||
device_id: model.device_id().unwrap(),
|
||||
data: saved.to_sync_json()?,
|
||||
});
|
||||
|
||||
Ok(saved)
|
||||
}
|
||||
|
||||
pub async fn commit_shared<M>(
|
||||
&self,
|
||||
library: &Library,
|
||||
model: M,
|
||||
) -> Result<M>
|
||||
where
|
||||
M: Syncable,
|
||||
{
|
||||
// 1. Verify classification
|
||||
if model.is_device_owned() {
|
||||
return Err(TxError::InvalidModel("Expected shared".into()));
|
||||
}
|
||||
|
||||
// 2. Get PeerSync to generate HLC and write to peer_log
|
||||
let peer_sync = library.sync_service()
|
||||
.ok_or(TxError::SyncLog("Sync service not initialized".into()))?
|
||||
.peer_sync();
|
||||
|
||||
// 3. Atomic: DB write + peer_log write
|
||||
let saved = library.db().transaction(|txn| async {
|
||||
let saved = model.insert(txn).await?;
|
||||
|
||||
// Write to peer_log happens inside PeerSync
|
||||
peer_sync.broadcast_shared_change(
|
||||
M::SYNC_MODEL.to_string(),
|
||||
saved.sync_id(),
|
||||
ChangeType::Insert,
|
||||
saved.to_sync_json()?,
|
||||
).await?;
|
||||
|
||||
Ok(saved)
|
||||
}).await?;
|
||||
|
||||
Ok(saved)
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~150 lines
|
||||
**Complexity**: Medium (need Library parameter, transaction integration)
|
||||
|
||||
---
|
||||
|
||||
## 🔌 Network Integration (Must Have)
|
||||
|
||||
### 5. Protocol Handler Wiring (HIGH PRIORITY)
|
||||
|
||||
**Location**: `service/network/protocol/sync/handler.rs`
|
||||
|
||||
**Current State**: Completely stubbed out
|
||||
```rust
|
||||
// All methods return "not yet implemented"
|
||||
warn!("SyncProtocolHandler called but protocol not yet implemented");
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
impl ProtocolHandler for SyncProtocolHandler {
|
||||
async fn handle_stream(&self, send, recv, remote_node_id) {
|
||||
// 1. Read SyncMessage from recv
|
||||
let message: SyncMessage = read_message(recv).await?;
|
||||
|
||||
// 2. Route to StateSyncHandler or LogSyncHandler
|
||||
match message {
|
||||
SyncMessage::StateChange { .. } => {
|
||||
state_handler.handle_state_change(...).await?
|
||||
}
|
||||
SyncMessage::SharedChange { entry } => {
|
||||
log_handler.handle_shared_change(entry).await?
|
||||
}
|
||||
// ... other messages
|
||||
}
|
||||
|
||||
// 3. Write response to send (if needed)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~200 lines
|
||||
**Complexity**: Medium (stream handling, message routing)
|
||||
|
||||
---
|
||||
|
||||
### 6. PeerSync Background Tasks (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `service/sync/peer.rs`
|
||||
|
||||
**Current State**: Loop exists but tasks not implemented
|
||||
```rust
|
||||
// Line 92-96
|
||||
// TODO: Start background tasks for:
|
||||
// - Listening to network messages
|
||||
// - Processing buffer queue
|
||||
// - Pruning sync log
|
||||
// - Heartbeat to peers
|
||||
// - Reconnect to offline peers
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
pub async fn start(&self) -> Result<()> {
|
||||
// Spawn task 1: Process buffer queue
|
||||
tokio::spawn({
|
||||
let peer_sync = self.clone();
|
||||
async move {
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
peer_sync.process_buffer_queue().await;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn task 2: Prune sync log periodically
|
||||
tokio::spawn({
|
||||
let peer_log = self.peer_log.clone();
|
||||
async move {
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_mins(5)).await;
|
||||
peer_log.prune_acked().await;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn task 3: Heartbeat to peers
|
||||
// Spawn task 4: Reconnect attempts
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~150 lines
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
## 📊 State Management & Persistence
|
||||
|
||||
### 7. Checkpoint Persistence (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/state.rs`
|
||||
|
||||
**Current State**: Checkpoint save/load are stubs
|
||||
```rust
|
||||
// Line 185-194
|
||||
pub async fn save(&self) -> Result<()> {
|
||||
// TODO: Persist to disk for crash recovery
|
||||
}
|
||||
|
||||
pub async fn load() -> Result<Option<Self>> {
|
||||
// TODO: Load from disk
|
||||
}
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
// Save to library_path/sync_checkpoint.json
|
||||
let checkpoint_path = library_path.join("sync_checkpoint.json");
|
||||
let json = serde_json::to_string_pretty(self)?;
|
||||
tokio::fs::write(checkpoint_path, json).await?;
|
||||
```
|
||||
|
||||
**Effort**: ~30 lines
|
||||
**Complexity**: Low
|
||||
|
||||
---
|
||||
|
||||
### 8. Backfill Network Requests (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `service/sync/backfill.rs`
|
||||
|
||||
**Current State**: Methods exist but return empty responses
|
||||
```rust
|
||||
// Line 202-240
|
||||
// TODO: Send StateRequest via network
|
||||
// TODO: Send SharedChangeRequest via network
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
async fn request_state_batch(...) -> Result<SyncMessage> {
|
||||
// Serialize request
|
||||
let request = SyncMessage::StateRequest { ... };
|
||||
|
||||
// Send via networking service
|
||||
let response = networking.send_request(
|
||||
peer,
|
||||
"sync",
|
||||
request
|
||||
).await?;
|
||||
|
||||
// Deserialize response
|
||||
let message: SyncMessage = serde_json::from_slice(&response)?;
|
||||
Ok(message)
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~80 lines
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
## 🔍 Data Serialization (Quality of Life)
|
||||
|
||||
### 9. Row to JSON Serialization (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `service/sync/protocol_handler.rs`
|
||||
|
||||
**Current State**: Returns empty JSON
|
||||
```rust
|
||||
// Line 145-151
|
||||
// TODO: Proper serialization per model type via registry
|
||||
data: serde_json::json!({}) // Placeholder
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
// Use Syncable trait's to_sync_json() method
|
||||
// Need to fetch actual model from row and serialize it
|
||||
let model = Model::from_query_result(row)?;
|
||||
let data = model.to_sync_json()?;
|
||||
```
|
||||
|
||||
**Effort**: ~50 lines (generic row -> model conversion)
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
### 10. Shared State Fallback Query (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/protocol_handler.rs`
|
||||
|
||||
**Current State**: Returns empty JSON
|
||||
```rust
|
||||
// Line 246-250
|
||||
// TODO: Query via registry instead of hardcoding
|
||||
async fn get_current_shared_state() -> Result<serde_json::Value> {
|
||||
Ok(serde_json::json!({
|
||||
"tags": [],
|
||||
"albums": [],
|
||||
"user_metadata": [],
|
||||
}))
|
||||
}
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
async fn get_current_shared_state() -> Result<serde_json::Value> {
|
||||
// Query all shared models from registry
|
||||
let shared_models = registry.iter()
|
||||
.filter(|(_, reg)| !reg.is_device_owned)
|
||||
.map(|(name, _)| name);
|
||||
|
||||
let mut state = serde_json::Map::new();
|
||||
for model_type in shared_models {
|
||||
let records = query_all_records(model_type).await?;
|
||||
state.insert(model_type.clone(), serde_json::to_value(records)?);
|
||||
}
|
||||
|
||||
Ok(serde_json::Value::Object(state))
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~60 lines
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
## 🔄 Periodic Tasks (Nice to Have)
|
||||
|
||||
### 11. SyncService Periodic Tasks (MEDIUM PRIORITY)
|
||||
|
||||
**Location**: `service/sync/mod.rs`
|
||||
|
||||
**Current State**: Loop exists but empty
|
||||
```rust
|
||||
// Line 92-99
|
||||
// TODO: Implement periodic tasks:
|
||||
// - Process buffer queue
|
||||
// - Prune sync log
|
||||
// - Heartbeat to peers
|
||||
// - Reconnect to offline peers
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
async fn run_sync_loop(peer_sync: Arc<PeerSync>, ...) {
|
||||
let mut prune_interval = tokio::time::interval(Duration::from_secs(300)); // 5 min
|
||||
let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(30));
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = prune_interval.tick() => {
|
||||
peer_sync.peer_log.prune_acked().await?;
|
||||
}
|
||||
_ = heartbeat_interval.tick() => {
|
||||
send_heartbeat_to_peers().await?;
|
||||
}
|
||||
_ = shutdown_rx.recv() => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~80 lines
|
||||
**Complexity**: Low-Medium
|
||||
|
||||
---
|
||||
|
||||
### 12. Peer Reconnection Logic (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/backfill.rs`
|
||||
|
||||
**Current State**: Placeholder
|
||||
```rust
|
||||
// Line 251-253
|
||||
// TODO: Save checkpoint, select new peer, resume
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
async fn on_peer_disconnected(&self, peer_id: Uuid) {
|
||||
if self.is_backfilling_from(peer_id) {
|
||||
// Save current progress
|
||||
self.save_checkpoint().await?;
|
||||
|
||||
// Select new peer
|
||||
let peers = get_available_peers().await?;
|
||||
let new_peer = select_backfill_peer(peers)?;
|
||||
|
||||
// Resume from checkpoint
|
||||
self.resume_backfill(new_peer, self.load_checkpoint().await?).await?;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~60 lines
|
||||
**Complexity**: Medium
|
||||
|
||||
---
|
||||
|
||||
## 📈 Optimizations (Can Wait)
|
||||
|
||||
### 13. Backfill Progress Calculation (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/backfill.rs`
|
||||
|
||||
**Current State**: Hardcoded 0.5
|
||||
```rust
|
||||
// Line 157
|
||||
current_checkpoint.update(chk, 0.5); // TODO: Calculate actual progress
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
let total_expected = estimate_total_records(model_type).await?;
|
||||
let progress = records_received as f32 / total_expected as f32;
|
||||
current_checkpoint.update(chk, progress);
|
||||
```
|
||||
|
||||
**Effort**: ~30 lines
|
||||
**Complexity**: Low
|
||||
|
||||
---
|
||||
|
||||
### 14. Checkpointing in State Queries (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/protocol_handler.rs`
|
||||
|
||||
**Current State**: Always None
|
||||
```rust
|
||||
// Line 104-105
|
||||
checkpoint: None, // TODO: Implement checkpointing
|
||||
has_more: false, // TODO: Implement pagination
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
// Track offset in query
|
||||
let offset = parse_checkpoint(checkpoint)?;
|
||||
query.push_str(&format!(" OFFSET {}", offset));
|
||||
|
||||
// Return new checkpoint if more data
|
||||
let new_checkpoint = if total_count > batch_size {
|
||||
Some(format!("offset-{}", offset + batch_size))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
```
|
||||
|
||||
**Effort**: ~40 lines
|
||||
**Complexity**: Low
|
||||
|
||||
---
|
||||
|
||||
### 15. Multi-Peer Backfill (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/backfill.rs`
|
||||
|
||||
**Current State**: Only backfills from primary peer
|
||||
```rust
|
||||
// Line 102-104
|
||||
// TODO: Get list of all peers, not just primary
|
||||
// For now, just backfill from primary peer
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
async fn backfill_device_owned_state(&self) {
|
||||
// Get ALL peers
|
||||
let peers = get_all_sync_partners().await?;
|
||||
|
||||
// Backfill each peer's device-owned data
|
||||
for peer in peers {
|
||||
self.backfill_peer_state(
|
||||
peer.id,
|
||||
vec!["location", "entry", "volume"],
|
||||
None
|
||||
).await?;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~40 lines
|
||||
**Complexity**: Low
|
||||
|
||||
---
|
||||
|
||||
### 16. Shared State Fallback Application (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/backfill.rs`
|
||||
|
||||
**Current State**: Logged but not applied
|
||||
```rust
|
||||
// Line 193
|
||||
// TODO: Deserialize and insert tags, albums, etc.
|
||||
```
|
||||
|
||||
**What's Needed**:
|
||||
```rust
|
||||
if let Some(state) = current_state {
|
||||
// Deserialize shared models from state
|
||||
if let Some(tags) = state.get("tags") {
|
||||
let tags: Vec<tag::Model> = serde_json::from_value(tags)?;
|
||||
for tag in tags {
|
||||
tag.insert_or_ignore(db).await?;
|
||||
}
|
||||
}
|
||||
// Same for albums, user_metadata
|
||||
}
|
||||
```
|
||||
|
||||
**Effort**: ~40 lines
|
||||
**Complexity**: Low
|
||||
|
||||
---
|
||||
|
||||
## 🧹 Cleanup (Optional)
|
||||
|
||||
### 17. Remove Old Stubbed Methods (LOW PRIORITY)
|
||||
|
||||
**Location**: `infra/sync/transaction.rs`
|
||||
|
||||
**Current State**: Old methods still exist for compatibility
|
||||
```rust
|
||||
// Line 153-204
|
||||
// OLD METHODS (STUBBED - Will be replaced)
|
||||
pub async fn log_change_stubbed(...)
|
||||
pub async fn log_batch_stubbed(...)
|
||||
pub async fn log_bulk_stubbed(...)
|
||||
```
|
||||
|
||||
**Action**: Delete these once new methods are fully implemented
|
||||
|
||||
**Effort**: Delete ~50 lines
|
||||
**Complexity**: Trivial (just verify nothing calls them)
|
||||
|
||||
---
|
||||
|
||||
### 18. Remove SyncApplier Stub (LOW PRIORITY)
|
||||
|
||||
**Location**: `service/sync/applier.rs`
|
||||
|
||||
**Current State**: Entire file is a stub
|
||||
```rust
|
||||
// STUB - Being replaced with PeerSync
|
||||
```
|
||||
|
||||
**Action**: Can delete this file once registry apply is working
|
||||
|
||||
**Effort**: Delete file
|
||||
**Complexity**: Trivial
|
||||
|
||||
---
|
||||
|
||||
## 📋 Summary by Priority
|
||||
|
||||
### 🔴 Critical Path (Must Implement):
|
||||
1. **Network Message Integration** (~100 lines) - Send/receive messages
|
||||
2. **Model Apply Functions** (~350 lines) - 7 models × 50 lines each
|
||||
3. **Registry Function Pointers** (~100 lines) - Polymorphic dispatch
|
||||
4. **TransactionManager Integration** (~150 lines) - HLC + peer_log
|
||||
|
||||
**Total Critical**: ~700 lines, 1-2 days of focused work
|
||||
|
||||
### 🟡 Important (Should Implement):
|
||||
5. **Protocol Handler Wiring** (~200 lines) - Message routing
|
||||
6. **PeerSync Background Tasks** (~150 lines) - Periodic operations
|
||||
7. **Backfill Network Requests** (~80 lines) - Actual network calls
|
||||
|
||||
**Total Important**: ~430 lines, 1 day
|
||||
|
||||
### 🟢 Nice to Have (Can Wait):
|
||||
8-16. Various optimizations and completions
|
||||
|
||||
**Total Nice**: ~290 lines, 0.5 days
|
||||
|
||||
---
|
||||
|
||||
## 🎯 Recommended Implementation Order
|
||||
|
||||
### Phase 1: Get Basic Sync Working (Critical)
|
||||
1. Model apply functions (start with location, tag)
|
||||
2. Registry function pointers
|
||||
3. Network message integration
|
||||
4. TransactionManager integration
|
||||
|
||||
**Result**: Can sync locations and tags between 2 devices!
|
||||
|
||||
### Phase 2: Protocol & Background (Important)
|
||||
5. Protocol handler wiring
|
||||
6. PeerSync background tasks
|
||||
7. Backfill network requests
|
||||
|
||||
**Result**: Full sync including backfill works!
|
||||
|
||||
### Phase 3: Polish (Nice to Have)
|
||||
8-16. Optimizations, edge cases, cleanup
|
||||
|
||||
**Result**: Production-ready!
|
||||
|
||||
---
|
||||
|
||||
## 📊 Current Implementation Status
|
||||
|
||||
| Component | Architecture | Implementation | Status |
|
||||
|-----------|--------------|----------------|--------|
|
||||
| HLC | ✅ | ✅ | Complete |
|
||||
| PeerLog | ✅ | ✅ | Complete |
|
||||
| State Machine | ✅ | ✅ | Complete |
|
||||
| PeerSync | ✅ | 🟡 | Needs network integration |
|
||||
| Protocol Messages | ✅ | ✅ | Complete |
|
||||
| Protocol Handlers | ✅ | 🟡 | Needs wiring |
|
||||
| Backfill | ✅ | 🟡 | Needs network calls |
|
||||
| TransactionManager | ✅ | 🟡 | Needs HLC integration |
|
||||
| Registry Dispatch | ✅ | 🟡 | Needs function pointers |
|
||||
| Model Apply | ✅ | ⚠️ | Needs per-model impl |
|
||||
|
||||
**Legend**: ✅ Done | 🟡 Partial | ⚠️ Not Started
|
||||
|
||||
---
|
||||
|
||||
## 🚀 Estimated Total Remaining Work
|
||||
|
||||
**Critical Path**: ~700 lines, 1-2 days
|
||||
**Full Implementation**: ~1,420 lines, 3-4 days
|
||||
**Current Progress**: Architecture complete, ~40% implementation done
|
||||
|
||||
The hardest part (architecture) is DONE! What remains is straightforward implementation following the established patterns.
|
||||
|
||||
@@ -15,6 +15,7 @@ pub mod peer_log;
|
||||
pub mod registry;
|
||||
pub mod syncable;
|
||||
pub mod transaction;
|
||||
pub mod transport;
|
||||
|
||||
pub use deterministic::{
|
||||
deterministic_system_album_uuid, deterministic_system_tag_uuid, system_tags,
|
||||
@@ -27,3 +28,4 @@ pub use registry::{
|
||||
};
|
||||
pub use syncable::Syncable;
|
||||
pub use transaction::{BulkOperation, BulkOperationMetadata, TransactionManager, TxError};
|
||||
pub use transport::NetworkTransport;
|
||||
|
||||
@@ -96,10 +96,6 @@ impl TransactionManager {
|
||||
&self.event_bus
|
||||
}
|
||||
|
||||
// ===================================================================
|
||||
// NEW LEADERLESS METHODS
|
||||
// ===================================================================
|
||||
|
||||
/// Commit device-owned resource (state-based sync)
|
||||
///
|
||||
/// For locations, entries, volumes, audit logs - data owned by this device.
|
||||
@@ -149,9 +145,7 @@ impl TransactionManager {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ===================================================================
|
||||
// OLD METHODS (STUBBED - Will be replaced with HLC-based approach)
|
||||
// ===================================================================
|
||||
|
||||
/// Log a single change (DEPRECATED - Use PeerSync directly)
|
||||
///
|
||||
|
||||
151
core/src/infra/sync/transport.rs
Normal file
151
core/src/infra/sync/transport.rs
Normal file
@@ -0,0 +1,151 @@
|
||||
//! Network transport abstraction for sync messages
|
||||
//!
|
||||
//! Provides a trait-based abstraction layer between the sync system and networking layer,
|
||||
//! solving the circular dependency problem and enabling testability.
|
||||
|
||||
use anyhow::Result;
|
||||
use uuid::Uuid;
|
||||
|
||||
// Import sync message from network protocol module
|
||||
// (Network layer will implement this trait, but sync layer defines it)
|
||||
use crate::service::network::protocol::sync::messages::SyncMessage;
|
||||
|
||||
/// Abstraction for sending sync messages over the network
|
||||
///
|
||||
/// This trait decouples the sync system from the networking implementation:
|
||||
/// - Sync layer (PeerSync) depends on this trait
|
||||
/// - Network layer (NetworkingService) implements this trait
|
||||
/// - Breaks circular dependency: Library → SyncService → NetworkTransport ← NetworkingService
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```rust,ignore
|
||||
/// // In PeerSync
|
||||
/// async fn broadcast_state_change(&self, change: StateChangeMessage) {
|
||||
/// let partners = self.get_sync_partners().await?;
|
||||
///
|
||||
/// for partner_uuid in partners {
|
||||
/// // NetworkTransport handles UUID→NodeId mapping internally
|
||||
/// self.network.send_sync_message(partner_uuid, message.clone()).await?;
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// # Implementation Notes
|
||||
///
|
||||
/// The implementer (NetworkingService) must:
|
||||
/// 1. Map device UUID to network NodeId using DeviceRegistry
|
||||
/// 2. Serialize the SyncMessage
|
||||
/// 3. Send via Iroh endpoint
|
||||
/// 4. Handle connection errors gracefully (devices may be offline)
|
||||
#[async_trait::async_trait]
|
||||
pub trait NetworkTransport: Send + Sync {
|
||||
/// Send a sync message to a specific device
|
||||
///
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `target_device` - UUID of the target device (from sync_partners table)
|
||||
/// * `message` - The sync message to send (StateChange, SharedChange, etc.)
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// - `Ok(())` if message was sent successfully
|
||||
/// - `Err(...)` if:
|
||||
/// - Target device UUID is not mapped to a NodeId (device not paired/connected)
|
||||
/// - Network send fails (connection error, device offline)
|
||||
/// - Serialization fails
|
||||
///
|
||||
/// # Implementation
|
||||
///
|
||||
/// The implementer should:
|
||||
/// 1. Look up NodeId via `device_registry.get_node_id_for_device(target_device)`
|
||||
/// 2. If NodeId not found, return error (device not connected)
|
||||
/// 3. Serialize message to bytes
|
||||
/// 4. Send via `endpoint.send_message(node_id, "sync", bytes)`
|
||||
///
|
||||
/// # Error Handling
|
||||
///
|
||||
/// Callers should handle errors gracefully - devices may go offline mid-broadcast.
|
||||
/// Consider logging warnings rather than failing the entire operation.
|
||||
async fn send_sync_message(&self, target_device: Uuid, message: SyncMessage) -> Result<()>;
|
||||
|
||||
/// Get list of currently connected sync partner devices
|
||||
///
|
||||
/// Returns UUIDs of devices that are:
|
||||
/// - Listed in sync_partners table with sync_enabled=true
|
||||
/// - Currently connected (have active network connection)
|
||||
///
|
||||
/// This is used to optimize broadcasting - only send to devices that can receive.
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// Vector of device UUIDs that are currently reachable for sync messages.
|
||||
/// Empty vector if no sync partners are connected.
|
||||
///
|
||||
/// # Implementation Note
|
||||
///
|
||||
/// This should query:
|
||||
/// 1. `sync_partners` table for enabled partners
|
||||
/// 2. `device_registry` for connection status
|
||||
/// 3. Return intersection of (enabled) AND (connected)
|
||||
async fn get_connected_sync_partners(&self) -> Result<Vec<Uuid>>;
|
||||
|
||||
/// Check if a specific device is currently reachable
|
||||
///
|
||||
/// Useful before attempting to send, to avoid unnecessary errors.
|
||||
///
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `device_uuid` - UUID of the device to check
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// `true` if:
|
||||
/// - Device is mapped to a NodeId in DeviceRegistry
|
||||
/// - Device has an active network connection
|
||||
///
|
||||
/// `false` otherwise (device offline, not paired, etc.)
|
||||
async fn is_device_reachable(&self, device_uuid: Uuid) -> bool {
|
||||
// Default implementation: can be overridden for more efficient checks
|
||||
// For now, we just assume device is reachable if UUID is known
|
||||
// (actual implementation will check DeviceRegistry connection status)
|
||||
false // Implementer should override this
|
||||
}
|
||||
}
|
||||
|
||||
/// Mock implementation for testing
|
||||
#[cfg(test)]
|
||||
pub struct MockNetworkTransport {
|
||||
/// Track which devices received which messages
|
||||
pub sent_messages: std::sync::Arc<std::sync::Mutex<Vec<(Uuid, SyncMessage)>>>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
impl MockNetworkTransport {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
sent_messages: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn get_sent_messages(&self) -> Vec<(Uuid, SyncMessage)> {
|
||||
self.sent_messages.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[async_trait::async_trait]
|
||||
impl NetworkTransport for MockNetworkTransport {
|
||||
async fn send_sync_message(&self, target_device: Uuid, message: SyncMessage) -> Result<()> {
|
||||
self.sent_messages
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push((target_device, message));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_connected_sync_partners(&self) -> Result<Vec<Uuid>> {
|
||||
// For tests, return empty list
|
||||
Ok(vec![])
|
||||
}
|
||||
}
|
||||
@@ -908,6 +908,11 @@ impl NetworkingService {
|
||||
self.device_registry.clone()
|
||||
}
|
||||
|
||||
/// Get the Iroh endpoint for network communication
|
||||
pub fn endpoint(&self) -> Option<&Endpoint> {
|
||||
self.endpoint.as_ref()
|
||||
}
|
||||
|
||||
/// Publish a discovery record for pairing session
|
||||
// Note: Discovery for pairing is now handled via mDNS user_data field
|
||||
// - Initiator: Sets user_data to session_id via endpoint.set_user_data_for_discovery()
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
pub mod core;
|
||||
pub mod device;
|
||||
pub mod protocol;
|
||||
pub mod transports;
|
||||
pub mod utils;
|
||||
|
||||
// Re-export main types for easy access
|
||||
|
||||
177
core/src/service/network/protocol/sync/transport.rs
Normal file
177
core/src/service/network/protocol/sync/transport.rs
Normal file
@@ -0,0 +1,177 @@
|
||||
//! NetworkTransport implementation for NetworkingService
|
||||
//!
|
||||
//! Implements the sync layer's NetworkTransport trait, enabling PeerSync to send
|
||||
//! sync messages over the network without circular dependencies.
|
||||
|
||||
use crate::{
|
||||
infra::sync::NetworkTransport,
|
||||
service::network::{protocol::sync::messages::SyncMessage, NetworkingError},
|
||||
};
|
||||
use anyhow::Result;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::service::network::core::{NetworkingService, SYNC_ALPN};
|
||||
|
||||
/// Implementation of NetworkTransport for NetworkingService
|
||||
///
|
||||
/// This bridges the sync layer (which uses device UUIDs) with the network layer
|
||||
/// (which uses Iroh NodeIds) by leveraging the DeviceRegistry for UUID↔NodeId mapping.
|
||||
#[async_trait::async_trait]
|
||||
impl NetworkTransport for NetworkingService {
|
||||
/// Send a sync message to a target device
|
||||
///
|
||||
/// # Implementation Details
|
||||
///
|
||||
/// 1. Look up NodeId for device UUID via DeviceRegistry
|
||||
/// 2. Serialize the SyncMessage to JSON bytes
|
||||
/// 3. Send via Iroh endpoint using the sync protocol ALPN
|
||||
/// 4. Handle errors gracefully (device may be offline)
|
||||
async fn send_sync_message(&self, target_device: Uuid, message: SyncMessage) -> Result<()> {
|
||||
// 1. Look up NodeId for device UUID
|
||||
let node_id = {
|
||||
let registry = self.device_registry.read().await;
|
||||
registry
|
||||
.get_node_id_for_device(target_device)
|
||||
.ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"Device {} not found in registry (not paired or offline)",
|
||||
target_device
|
||||
)
|
||||
})?
|
||||
};
|
||||
|
||||
debug!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
message_type = ?std::mem::discriminant(&message),
|
||||
library_id = %message.library_id(),
|
||||
"Sending sync message"
|
||||
);
|
||||
|
||||
// 2. Serialize message to bytes
|
||||
let bytes = serde_json::to_vec(&message)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to serialize sync message: {}", e))?;
|
||||
|
||||
// 3. Send via Iroh endpoint
|
||||
let endpoint = self
|
||||
.endpoint
|
||||
.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("Network endpoint not initialized"))?;
|
||||
|
||||
// Open a connection and send
|
||||
// Note: Iroh handles connection pooling, so repeated calls are efficient
|
||||
let conn = endpoint
|
||||
.connect(node_id.into(), SYNC_ALPN)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
warn!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
error = %e,
|
||||
"Failed to connect to device for sync"
|
||||
);
|
||||
anyhow::anyhow!("Failed to connect to {}: {}", target_device, e)
|
||||
})?;
|
||||
|
||||
// Open a unidirectional stream and send the message
|
||||
let mut send = conn
|
||||
.open_uni()
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to open stream: {}", e))?;
|
||||
|
||||
send.write_all(&bytes).await.map_err(|e| {
|
||||
warn!(
|
||||
device_uuid = %target_device,
|
||||
error = %e,
|
||||
"Failed to write sync message to stream"
|
||||
);
|
||||
anyhow::anyhow!("Failed to write message: {}", e)
|
||||
})?;
|
||||
|
||||
send.finish()
|
||||
.map_err(|e| anyhow::anyhow!("Failed to finish stream: {}", e))?;
|
||||
|
||||
debug!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
bytes_sent = bytes.len(),
|
||||
"Sync message sent successfully"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Get list of currently connected sync partner devices
|
||||
///
|
||||
/// Returns device UUIDs that are both:
|
||||
/// - Registered in DeviceRegistry (paired)
|
||||
/// - Currently have an active connection
|
||||
///
|
||||
/// Note: This doesn't query the sync_partners table - that's the caller's responsibility.
|
||||
/// We just report which devices are network-reachable right now.
|
||||
async fn get_connected_sync_partners(&self) -> Result<Vec<Uuid>> {
|
||||
let registry = self.device_registry.read().await;
|
||||
|
||||
// Get all connected devices from registry
|
||||
let connected_devices = registry.get_connected_devices();
|
||||
|
||||
// Extract device UUIDs
|
||||
let device_uuids: Vec<Uuid> = connected_devices
|
||||
.into_iter()
|
||||
.filter_map(|device_info| {
|
||||
// Parse device UUID from device info
|
||||
// The device ID should be in the DeviceInfo structure
|
||||
device_info.id
|
||||
})
|
||||
.collect();
|
||||
|
||||
debug!(
|
||||
count = device_uuids.len(),
|
||||
"Retrieved connected sync partners"
|
||||
);
|
||||
|
||||
Ok(device_uuids)
|
||||
}
|
||||
|
||||
/// Check if a specific device is currently reachable
|
||||
///
|
||||
/// Returns true if:
|
||||
/// - Device UUID is mapped to a NodeId in DeviceRegistry
|
||||
/// - Device has an active network connection (we can reach it)
|
||||
async fn is_device_reachable(&self, device_uuid: Uuid) -> bool {
|
||||
let registry = self.device_registry.read().await;
|
||||
|
||||
// Check if device is in registry
|
||||
if let Some(node_id) = registry.get_node_id_for_device(device_uuid) {
|
||||
// Check if we have an active connection to this node
|
||||
// This is a lightweight check - just verifying the node is in our connection table
|
||||
if let Some(endpoint) = &self.endpoint {
|
||||
// Try to get connection info (doesn't actually open a connection)
|
||||
// If the node is reachable via our endpoint, return true
|
||||
// Note: This is an optimistic check - actual send might still fail
|
||||
return endpoint.connection_info(node_id.into()).is_some();
|
||||
}
|
||||
}
|
||||
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_transport_trait_implemented() {
|
||||
// This test just verifies that the trait is properly implemented
|
||||
// Actual functionality tests would require setting up Iroh endpoints
|
||||
// and device registries, which is better done as integration tests
|
||||
|
||||
// Verify trait bound is satisfied
|
||||
fn assert_network_transport<T: NetworkTransport>() {}
|
||||
assert_network_transport::<NetworkingService>();
|
||||
}
|
||||
}
|
||||
24
core/src/service/network/transports/mod.rs
Normal file
24
core/src/service/network/transports/mod.rs
Normal file
@@ -0,0 +1,24 @@
|
||||
//! Outbound message transports
|
||||
//!
|
||||
//! Transports are responsible for initiating messages to remote devices.
|
||||
//! This is distinct from protocol handlers which receive and respond to incoming messages.
|
||||
//!
|
||||
//! ## Architecture
|
||||
//!
|
||||
//! - `protocol/` - Inbound message handlers (implement ProtocolHandler)
|
||||
//! - `transports/` - Outbound message senders (implement domain-specific transport traits)
|
||||
//!
|
||||
//! ## When to Use Transports vs Protocols
|
||||
//!
|
||||
//! **Use a Transport when:**
|
||||
//! - You need to initiate messages from application logic (not in response to a network message)
|
||||
//! - Example: Broadcasting sync changes when local data is modified
|
||||
//!
|
||||
//! **Use a Protocol when:**
|
||||
//! - You're handling incoming network messages
|
||||
//! - Example: Receiving and applying sync changes from peers
|
||||
|
||||
pub mod sync;
|
||||
|
||||
// Re-export for convenience
|
||||
pub use sync::*;
|
||||
168
core/src/service/network/transports/sync.rs
Normal file
168
core/src/service/network/transports/sync.rs
Normal file
@@ -0,0 +1,168 @@
|
||||
//! NetworkTransport implementation for NetworkingService
|
||||
//!
|
||||
//! Implements the sync layer's NetworkTransport trait, enabling PeerSync to send
|
||||
//! sync messages over the network without circular dependencies.
|
||||
|
||||
use crate::{
|
||||
infra::sync::NetworkTransport,
|
||||
service::network::{protocol::sync::messages::SyncMessage, NetworkingError},
|
||||
};
|
||||
use anyhow::Result;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::service::network::core::{NetworkingService, SYNC_ALPN};
|
||||
|
||||
/// Implementation of NetworkTransport for NetworkingService
|
||||
///
|
||||
/// This bridges the sync layer (which uses device UUIDs) with the network layer
|
||||
/// (which uses Iroh NodeIds) by leveraging the DeviceRegistry for UUID↔NodeId mapping.
|
||||
#[async_trait::async_trait]
|
||||
impl NetworkTransport for NetworkingService {
|
||||
/// Send a sync message to a target device
|
||||
///
|
||||
/// # Implementation Details
|
||||
///
|
||||
/// 1. Look up NodeId for device UUID via DeviceRegistry
|
||||
/// 2. Serialize the SyncMessage to JSON bytes
|
||||
/// 3. Send via Iroh endpoint using the sync protocol ALPN
|
||||
/// 4. Handle errors gracefully (device may be offline)
|
||||
async fn send_sync_message(&self, target_device: Uuid, message: SyncMessage) -> Result<()> {
|
||||
// 1. Look up NodeId for device UUID via public getter
|
||||
let device_registry_arc = self.device_registry();
|
||||
let node_id = {
|
||||
let registry = device_registry_arc.read().await;
|
||||
registry
|
||||
.get_node_id_for_device(target_device)
|
||||
.ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"Device {} not found in registry (not paired or offline)",
|
||||
target_device
|
||||
)
|
||||
})?
|
||||
};
|
||||
|
||||
debug!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
message_type = ?std::mem::discriminant(&message),
|
||||
library_id = %message.library_id(),
|
||||
"Sending sync message"
|
||||
);
|
||||
|
||||
// 2. Serialize message to bytes
|
||||
let bytes = serde_json::to_vec(&message)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to serialize sync message: {}", e))?;
|
||||
|
||||
// 3. Get Iroh endpoint via public getter
|
||||
let endpoint = self
|
||||
.endpoint()
|
||||
.ok_or_else(|| anyhow::anyhow!("Network endpoint not initialized"))?;
|
||||
|
||||
// Open a connection and send
|
||||
// Note: Iroh handles connection pooling, so repeated calls are efficient
|
||||
let conn = endpoint.connect(node_id, SYNC_ALPN).await.map_err(|e| {
|
||||
warn!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
error = %e,
|
||||
"Failed to connect to device for sync"
|
||||
);
|
||||
anyhow::anyhow!("Failed to connect to {}: {}", target_device, e)
|
||||
})?;
|
||||
|
||||
// Open a unidirectional stream and send the message
|
||||
let mut send = conn
|
||||
.open_uni()
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to open stream: {}", e))?;
|
||||
|
||||
send.write_all(&bytes).await.map_err(|e| {
|
||||
warn!(
|
||||
device_uuid = %target_device,
|
||||
error = %e,
|
||||
"Failed to write sync message to stream"
|
||||
);
|
||||
anyhow::anyhow!("Failed to write message: {}", e)
|
||||
})?;
|
||||
|
||||
send.finish()
|
||||
.map_err(|e| anyhow::anyhow!("Failed to finish stream: {}", e))?;
|
||||
|
||||
debug!(
|
||||
device_uuid = %target_device,
|
||||
node_id = %node_id,
|
||||
bytes_sent = bytes.len(),
|
||||
"Sync message sent successfully"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Get list of currently connected sync partner devices
|
||||
///
|
||||
/// Returns device UUIDs that are both:
|
||||
/// - Registered in DeviceRegistry (paired)
|
||||
/// - Currently have an active connection
|
||||
///
|
||||
/// Note: This doesn't query the sync_partners table - that's the caller's responsibility.
|
||||
/// We just report which devices are network-reachable right now.
|
||||
async fn get_connected_sync_partners(&self) -> Result<Vec<Uuid>> {
|
||||
let device_registry_arc = self.device_registry();
|
||||
let registry = device_registry_arc.read().await;
|
||||
|
||||
// Get all connected devices from registry
|
||||
let connected_devices = registry.get_connected_devices();
|
||||
|
||||
// Extract device UUIDs
|
||||
let device_uuids: Vec<Uuid> = connected_devices
|
||||
.into_iter()
|
||||
.map(|device_info| device_info.device_id)
|
||||
.collect();
|
||||
|
||||
debug!(
|
||||
count = device_uuids.len(),
|
||||
"Retrieved connected sync partners"
|
||||
);
|
||||
|
||||
Ok(device_uuids)
|
||||
}
|
||||
|
||||
/// Check if a specific device is currently reachable
|
||||
///
|
||||
/// Returns true if:
|
||||
/// - Device UUID is mapped to a NodeId in DeviceRegistry
|
||||
/// - Device has an active network connection (we can reach it)
|
||||
async fn is_device_reachable(&self, device_uuid: Uuid) -> bool {
|
||||
let device_registry_arc = self.device_registry();
|
||||
let registry = device_registry_arc.read().await;
|
||||
|
||||
// Check if device is in registry and mapped to a node
|
||||
if registry.get_node_id_for_device(device_uuid).is_some() {
|
||||
// Device is registered and mapped - assume reachable
|
||||
// Actual connectivity will be determined when we try to send
|
||||
// (Iroh endpoint handles connection establishment)
|
||||
return true;
|
||||
}
|
||||
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_transport_trait_implemented() {
|
||||
// This test just verifies that the trait is properly implemented
|
||||
// Actual functionality tests would require setting up Iroh endpoints
|
||||
// and device registries, which is better done as integration tests
|
||||
|
||||
// Verify trait bound is satisfied
|
||||
fn assert_network_transport<T: NetworkTransport>() {}
|
||||
assert_network_transport::<NetworkingService>();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user