Library Sync

Uninitialized → Backfilling → CatchingUp → Ready ⇄ Paused

Uninitialized
    → [peer available] → Backfilling
    → [already has data] → Ready

Backfilling
    → [complete] → CatchingUp
    → [peer disconnected] → save checkpoint, select new peer

CatchingUp
    → [buffer empty] → Ready
    → [5 consecutive failures] → Uninitialized (escalate to full backfill)

Ready
    → [offline] → Paused
    → [watermarks stale] → CatchingUp

Paused
    → [online] → Ready or CatchingUp

Attempt 1: Wait 10s, retry
Attempt 2: Wait 20s, retry
Attempt 3: Wait 40s, retry
Attempt 4: Wait 80s, retry
Attempt 5: Wait 160s (capped), retry
After 5 failures: Reset to Uninitialized, trigger full backfill

StateRequest {
    model_types: ["location", "entry"],
    since: Some(last_state_watermark),  // Only records newer than this
    checkpoint: Some("2025-10-21T19:10:00.456Z|uuid"),  // Resume cursor
    batch_size: config.batching.backfill_batch_size,
}

1. Device A inserts tag in database
2. Device A generates HLC timestamp
3. Device A appends to sync log
4. Device A broadcasts SharedChange message
5. Other devices apply in HLC order
6. After acknowledgment, prune from log

1. Device A inserts all tags in database
2. Device A generates HLC for each and appends to sync log
3. Device A broadcasts single SharedChangeBatch message
4. Other devices apply all entries in HLC order
5. After acknowledgment, prune from log

pub struct HLC {
    /// Physical time component (milliseconds since Unix epoch)
    pub timestamp: u64,

    /// Logical counter for events within the same millisecond
    pub counter: u64,

    /// Device that generated this HLC (for deterministic ordering)
    pub device_id: Uuid,
}

fn generate(last: Option<HLC>, device_id: Uuid) -> HLC {
    let physical = now_millis();
    let (timestamp, counter) = match last {
        Some(prev) if prev.timestamp >= physical => {
            // Clock hasn't advanced, increment counter
            (prev.timestamp, prev.counter + 1)
        }
        Some(prev) => {
            // Clock advanced, reset counter
            (physical, 0)
        }
        None => (physical, 0),
    };
    HLC { timestamp, counter, device_id }
}

fn update(&mut self, received: HLC) {
    let physical = now_millis();
    let max_ts = max(self.timestamp, max(received.timestamp, physical));

    self.counter = if max_ts == self.timestamp && max_ts == received.timestamp {
        max(self.counter, received.counter) + 1
    } else if max_ts == self.timestamp {
        self.counter + 1
    } else if max_ts == received.timestamp {
        received.counter + 1
    } else {
        0  // Physical time advanced
    };
    self.timestamp = max_ts;
}

Device A updates tag with HLC(timestamp_a, 0, device-a)
Device B updates same tag with HLC(timestamp_b, 0, device-b)

If timestamp_b > timestamp_a: Device B's version wins
If timestamps equal: Higher device_id breaks the tie (deterministic)

Device A creates tag "Vacation" with UUID-A
Device B creates tag "Vacation" with UUID-B

After sync: Both tags exist (different UUIDs, same name)
Tags can be disambiguated by namespace or merged by user

-- Device-owned tables
CREATE TABLE locations (
    id INTEGER PRIMARY KEY,
    uuid TEXT UNIQUE,
    device_id INTEGER,  -- Owner
    path TEXT,
    name TEXT
);

CREATE TABLE entries (
    id INTEGER PRIMARY KEY,
    uuid TEXT UNIQUE,
    location_id INTEGER,  -- Inherits ownership
    name TEXT,
    kind INTEGER,
    size_bytes INTEGER
);

-- Shared resource tables
CREATE TABLE tags (
    id INTEGER PRIMARY KEY,
    uuid TEXT UNIQUE,
    canonical_name TEXT
    -- No device_id (anyone can modify)
);

-- Shared resource changes pending acknowledgment
CREATE TABLE shared_changes (
    hlc TEXT PRIMARY KEY,
    model_type TEXT NOT NULL,
    record_uuid TEXT NOT NULL,
    change_type TEXT NOT NULL,  -- insert/update/delete
    data TEXT NOT NULL,         -- JSON payload
    created_at TEXT NOT NULL    -- When this change was logged
);

-- Peer acknowledgment tracking (outgoing - for pruning our log)
-- Tracks which of our changes each peer has acknowledged receiving
CREATE TABLE peer_acks (
    peer_device_id TEXT PRIMARY KEY,
    last_acked_hlc TEXT NOT NULL,
    acked_at TEXT NOT NULL
);

-- Per-resource watermarks for device-owned incremental sync
CREATE TABLE device_resource_watermarks (
    device_uuid TEXT NOT NULL,
    peer_device_uuid TEXT NOT NULL,
    resource_type TEXT NOT NULL,    -- "location", "entry", "volume", etc.
    last_watermark TEXT NOT NULL,   -- RFC3339 timestamp
    updated_at TEXT NOT NULL,
    PRIMARY KEY (device_uuid, peer_device_uuid, resource_type)
);

-- Per-peer watermarks for shared resource incremental sync (incoming)
-- Tracks the maximum HLC we've received from each peer
CREATE TABLE peer_received_watermarks (
    device_uuid TEXT NOT NULL,
    peer_device_uuid TEXT NOT NULL,
    max_received_hlc TEXT NOT NULL, -- Maximum HLC received from this peer
    updated_at TEXT NOT NULL,
    PRIMARY KEY (device_uuid, peer_device_uuid)
);

-- Resumable backfill checkpoints
CREATE TABLE backfill_checkpoints (
    id INTEGER PRIMARY KEY,
    peer_device_uuid TEXT NOT NULL,
    model_type TEXT NOT NULL,
    resume_token TEXT,              -- timestamp|uuid cursor
    progress REAL,                  -- 0.0 to 1.0
    completed_models TEXT,          -- JSON array of completed model types
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

// 1. Simple models without FK relationships (shared resources)
// Use sync_model() - no DB connection needed
let tag = tag::ActiveModel { ... }.insert(db).await?;
library.sync_model(&tag, ChangeType::Insert).await?;

// 2. Models with FK relationships (needs UUID lookup)
// Use sync_model_with_db() - requires DB connection for FK conversion
let location = location::ActiveModel { ... }.insert(db).await?;
library.sync_model_with_db(&location, ChangeType::Insert, db.conn()).await?;

// 3. Bulk operations (1000+ records)
// Use sync_models_batch() - batches FK lookups and network broadcasts
let entries: Vec<entry::Model> = bulk_insert_entries(db).await?;
library.sync_models_batch(&entries, ChangeType::Insert, db.conn()).await?;

impl Syncable for YourModel {
    /// Stable model identifier used in sync logs (must never change)
    const SYNC_MODEL: &'static str = "your_model";

    /// Get the globally unique ID for this resource
    fn sync_id(&self) -> Uuid {
        self.uuid
    }

    /// Version number for optimistic concurrency control
    fn version(&self) -> i64 {
        self.version
    }

    /// Fields to exclude from sync (platform-specific data)
    fn exclude_fields() -> Option<&'static [&'static str]> {
        Some(&["id", "created_at", "updated_at"])
    }

    /// Declare sync dependencies on other models
    fn sync_depends_on() -> &'static [&'static str] {
        &["parent_model"]  // Models that must sync first
    }

    /// Declare foreign key mappings for automatic UUID conversion
    fn foreign_key_mappings() -> Vec<FKMapping> {
        vec![
            FKMapping::new("device_id", "devices"),
            FKMapping::new("parent_id", "your_models"),
        ]
    }
}

// Register with sync system - choose based on ownership model:

// For shared resources (any device can modify):
crate::register_syncable_shared!(Model, "your_model", "your_table");

// For shared resources with closure table rebuild after backfill:
crate::register_syncable_shared!(Model, "tag_relationship", "tag_relationship", with_rebuild);

// For device-owned data:
crate::register_syncable_device_owned!(Model, "your_model", "your_table");

// With deletion support:
crate::register_syncable_device_owned!(Model, "your_model", "your_table", with_deletion);

// With deletion + post-backfill rebuild (for models with closure tables):
crate::register_syncable_device_owned!(Model, "entry", "entries", with_deletion, with_rebuild);

impl Syncable for YourModel {
    // ... other trait methods ...

    async fn apply_shared_change(
        entry: SharedChangeEntry,
        db: &DatabaseConnection,
    ) -> Result<(), sea_orm::DbErr> {
        match entry.change_type {
            ChangeType::Insert | ChangeType::Update => {
                // Option 1: Default LWW - just upsert
                let active = deserialize_to_active_model(&entry.data)?;
                Entity::insert(active)
                    .on_conflict(/* upsert on uuid */)
                    .exec(db).await?;

                // Option 2: Field-level merge
                if let Some(existing) = Entity::find_by_uuid(uuid).one(db).await? {
                    let merged = merge_fields(existing, incoming, entry.hlc);
                    merged.update(db).await?;
                }

                // Option 3: Domain-specific rules
                // e.g., keep longer description, union tags, etc.
            }
            ChangeType::Delete => {
                Entity::delete_by_uuid(uuid).exec(db).await?;
            }
        }
        Ok(())
    }
}

Sync Order During Backfill:
1. Shared resources (tags, collections, content_identities)
2. Devices
3. Locations (needs devices)
4. Volumes (needs devices)
5. Entries (needs locations and content_identities)

// Before: 3000 queries for 1000 records with 3 FKs each
// After: 3 queries total (one per FK type)
let result = batch_map_sync_json_to_local(records, fk_mappings, db).await?;

// Records with missing FK references are returned separately for retry
for (record, fk_field, missing_uuid) in result.failed {
    // Buffer for retry when dependency arrives
}

// Record fails FK resolution - parent doesn't exist yet
let error = "Foreign key lookup failed: parent_uuid abc-123 not found";
let missing_uuid = extract_missing_dependency_uuid(&error);

// Track the waiting record
dependency_tracker.add_dependency(missing_uuid, buffered_update);

// Later, when parent record arrives and is applied...
let waiting = dependency_tracker.resolve(parent_uuid);
for update in waiting {
    // Retry applying - FK should resolve now
    apply_update(update).await?;
}

fn score(&self) -> i32 {
    let mut score = 0;

    // Prefer online peers
    if self.is_online { score += 100; }

    // Prefer peers with complete state
    if self.has_complete_state { score += 50; }

    // Prefer low latency (measured RTT)
    score -= (self.latency_ms / 10) as i32;

    // Prefer less busy peers
    score -= (self.active_syncs * 10) as i32;

    score
}

// System tags have consistent UUIDs everywhere
let system_tag_uuid = deterministic_system_tag_uuid("system");
// Always: 550e8400-e29b-41d4-a716-446655440000 (example)

// Library-scoped defaults
let default_uuid = deterministic_library_default_uuid(library_id, "default_collection");

StateResponse {
    records: [...],           // New and updated entries
    deleted_uuids: [uuid1],   // Root UUID only (cascade handles children)
}

SharedChangeResponse {
    entries: [...],              // Recent changes from peer log
    current_state: {
        tags: [...],             // Complete snapshot
        content_identities: [...],
        collections: [...],
    },
    has_more: bool,              // True if snapshot exceeds batch limit
}

1. Device B checks entry watermark for Device A: 2025-10-20 14:30:00
2. Device B sends StateRequest(model_types: ["entry"], since: 2025-10-20 14:30:00) to Device A
3. Device A queries: SELECT * FROM entries WHERE updated_at >= '2025-10-20 14:30:00'
4. Device A responds with StateResponse containing 3 new entries
5. Device B applies changes and updates entry watermark for Device A

checkpoint: "2025-10-21T19:10:00.456Z|abc-123-uuid"

WHERE (updated_at > cursor_timestamp)
   OR (updated_at = cursor_timestamp AND uuid > cursor_uuid)
ORDER BY updated_at, uuid
LIMIT {configured_batch_size}

SharedChangeRequest {
    since_hlc: Some(last_hlc),  // Resume from this HLC
    limit: config.batching.backfill_batch_size,
}

// Device-owned state change
StateChange {
    library_id: Uuid,
    model_type: String,      // "location", "entry", etc.
    record_uuid: Uuid,
    device_id: Uuid,         // Owner device
    data: serde_json::Value, // Record as JSON
    timestamp: DateTime<Utc>,
}

// Batch of device-owned changes
StateBatch {
    library_id: Uuid,
    model_type: String,
    device_id: Uuid,
    records: Vec<StateRecord>,  // [{uuid, data, timestamp}, ...]
}

// Request device-owned state
StateRequest {
    library_id: Uuid,
    model_types: Vec<String>,
    device_id: Option<Uuid>,    // Specific device or all
    since: Option<DateTime>,    // Incremental sync
    checkpoint: Option<String>, // Resume cursor
    batch_size: usize,
}

// Response with device-owned state
StateResponse {
    library_id: Uuid,
    model_type: String,
    device_id: Uuid,
    records: Vec<StateRecord>,
    deleted_uuids: Vec<Uuid>,   // Tombstones
    checkpoint: Option<String>, // Next page cursor
    has_more: bool,
}

// Shared resource change (HLC-ordered)
SharedChange {
    library_id: Uuid,
    entry: SharedChangeEntry,
}

SharedChangeEntry {
    hlc: HLC,                   // Ordering key
    model_type: String,
    record_uuid: Uuid,
    change_type: ChangeType,    // Insert, Update, Delete
    data: serde_json::Value,
}

// Heartbeat with sync progress
Heartbeat {
    library_id: Uuid,
    device_id: Uuid,
    timestamp: DateTime<Utc>,
    state_watermark: Option<DateTime>,  // Last state sync
    shared_watermark: Option<HLC>,      // Last shared change
}

// Watermark exchange for reconnection
WatermarkExchangeRequest {
    library_id: Uuid,
    device_id: Uuid,
    my_state_watermark: Option<DateTime>,
    my_shared_watermark: Option<HLC>,
}

WatermarkExchangeResponse {
    library_id: Uuid,
    device_id: Uuid,
    state_watermark: Option<DateTime>,
    shared_watermark: Option<HLC>,
    needs_state_catchup: bool,
    needs_shared_catchup: bool,
}

UPDATE devices SET
    is_online = true,
    last_seen_at = NOW()
WHERE uuid = 'peer-device-id';

1. Broadcast fails (peer unreachable, timeout, etc.)
2. Message queued with next_retry = now + 5s
3. Background task checks queue every sync_loop_interval
4. Ready messages retried in order
5. Success: remove from queue
6. Failure: re-queue with doubled delay
7. After 5 attempts: drop and log warning

# Check pending changes
sqlite3 sync.db "SELECT COUNT(*) FROM shared_changes"

# Verify peer connections
sd sync status

# Monitor sync activity
RUST_LOG=sd_core::sync=debug cargo run

/// HLC parsing failures
HLCError::ParseError(String)

/// Peer log database errors
PeerLogError {
    ConnectionError(String),  // Can't open sync.db
    QueryError(String),       // SQL query failed
    SerializationError(String), // JSON encode/decode failed
    ParseError(String),       // Invalid data format
}

/// Watermark tracking errors
WatermarkError {
    QueryError(String),
    ParseError(String),
}

/// Checkpoint persistence errors
CheckpointError {
    QueryError(String),
    ParseError(String),
}

ApplyError {
    UnknownModel(String),           // Model not registered
    MissingFkLookup(String),        // FK mapper not configured
    WrongSyncType { model, expected, got }, // Device-owned vs shared mismatch
    MissingApplyFunction(String),   // No apply handler
    MissingQueryFunction(String),   // No query handler
    MissingDeletionHandler(String), // No deletion handler
    DatabaseError(String),          // DB operation failed
}

DependencyError {
    CircularDependency(String),     // A → B → A detected
    UnknownDependency(String, String), // Depends on unregistered model
    NoModels,                       // Empty registry
}

TxError {
    Database(DbErr),           // SeaORM error
    SyncLog(String),           // Peer log write failed
    Serialization(serde_json::Error), // JSON error
    InvalidModel(String),      // Model validation failed
}

HistogramMetric {
    count: AtomicU64,   // Number of samples
    sum: AtomicU64,     // Sum for average
    min: AtomicU64,     // Minimum value
    max: AtomicU64,     // Maximum value
}

// Methods
histogram.avg()   // Average latency
histogram.min()   // Best case
histogram.max()   // Worst case
histogram.count() // Sample count

let snapshot = sync_service.metrics().snapshot().await;

// Filter by time range
let recent = snapshot.filter_since(one_hour_ago);

// Filter by peer
let alice_metrics = snapshot.filter_by_peer(alice_device_id);

// Filter by model
let entry_metrics = snapshot.filter_by_model("entry");

MetricsHistory {
    capacity: 1000,  // Max snapshots retained
    snapshots: VecDeque<SyncMetricsSnapshot>,
}

// Query methods
history.get_snapshots_since(timestamp)
history.get_snapshots_range(start, end)
history.get_latest_snapshot()

enum SyncEvent {
    // Device-owned state change ready to broadcast
    StateChange {
        library_id: Uuid,
        model_type: String,
        record_uuid: Uuid,
        device_id: Uuid,
        data: serde_json::Value,
        timestamp: DateTime<Utc>,
    },

    // Shared resource change ready to broadcast
    SharedChange {
        library_id: Uuid,
        entry: SharedChangeEntry,
    },

    // Metrics snapshot available
    MetricsUpdated {
        library_id: Uuid,
        metrics: SyncMetricsSnapshot,
    },
}

1. Event arrives on SyncEventBus
2. Add to batch buffer
3. If buffer.len() >= 100 OR 50ms elapsed:
   4. Flush batch as single network message
5. Reset buffer and timer

#[model(
    table_name = "album",
    sync_strategy = "shared"
)]
struct Album {
    #[primary_key]
    id: Uuid,
    title: String,
    #[metadata]
    metadata_id: i32,
}

SyncConfig {
    batching: BatchingConfig {
        backfill_batch_size: 10_000,           // Records per backfill request
        state_broadcast_batch_size: 1_000,     // Device-owned records per broadcast
        shared_broadcast_batch_size: 100,      // Shared records per broadcast
        max_snapshot_size: 100_000,            // Max records in state snapshot
        realtime_batch_max_entries: 100,       // Max entries before flush
        realtime_batch_flush_interval_ms: 50,  // Auto-flush interval (ms)
    },
    retention: RetentionConfig {
        strategy: AcknowledgmentBased,
        tombstone_max_retention_days: 7,       // Hard limit for tombstone pruning
        peer_log_max_retention_days: 7,        // Hard limit for peer log pruning
        force_full_sync_threshold_days: 25,    // Force full sync if watermark older
    },
    network: NetworkConfig {
        message_timeout_secs: 30,              // Timeout for sync messages
        backfill_request_timeout_secs: 60,     // Timeout for backfill requests
        sync_loop_interval_secs: 5,            // Sync loop check interval
        connection_check_interval_secs: 10,    // How often to check peer connectivity
    },
    monitoring: MonitoringConfig {
        pruning_interval_secs: 3600,           // How often to prune sync.db (1 hour)
        enable_metrics: true,                  // Enable sync metrics collection
        metrics_log_interval_secs: 300,        // Persist metrics every 5 minutes
    },
}

# Use a preset
sd sync config set --preset aggressive

# Customize individual settings
sd sync config set --batch-size 5000 --retention-days 14

# Per-library configuration
sd library "Photos" sync config set --preset mobile