diff --git a/src-tauri/migrations/0036_sync2_profiles.sql b/src-tauri/migrations/0036_sync2_profiles.sql new file mode 100644 index 00000000..05fbb350 --- /dev/null +++ b/src-tauri/migrations/0036_sync2_profiles.sql @@ -0,0 +1,57 @@ +-- FLOW-SYNC/2 profiles (spec/protocol.md, Profiles): each synced profile has a link, and devices +-- keep what other profiles need until they run. Kept in Default's database, like the group. + +-- Which local profile syncs which link. A group joined before this starts with the link Default +-- joined (`sync2_group.link`), filled in by the app, which knows Default's id. +CREATE TABLE IF NOT EXISTS sync2_profile_links ( + profile TEXT PRIMARY KEY, + link TEXT NOT NULL UNIQUE +); + +-- What was sent, now per link. `link` is empty for the profiles collection. +CREATE TABLE IF NOT EXISTS sync2_sent_links ( + link TEXT NOT NULL, + collection TEXT NOT NULL, + record_key TEXT NOT NULL, + hash TEXT NOT NULL, + sent_at INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (link, collection, record_key) +); +INSERT OR IGNORE INTO sync2_sent_links (link, collection, record_key, hash, sent_at) + SELECT g.link, s.collection, s.record_key, s.hash, s.sent_at + FROM sync2_sent s, sync2_group g WHERE g.link != ''; + +-- Batches for a linked profile that isn't running, applied in `id` order once it runs. `part` +-- numbers the blobs of a snapshot read for that link; it is 0 for a batch. +CREATE TABLE IF NOT EXISTS sync2_kept ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + link TEXT NOT NULL, + sender TEXT NOT NULL, + seq INTEGER NOT NULL, + part INTEGER NOT NULL DEFAULT 0, + collection TEXT NOT NULL, + head TEXT NOT NULL, + records BLOB NOT NULL, + UNIQUE (link, sender, seq, part) +); + +-- The profiles collection: each link's name, merged last-writer-wins by `hlc`. +CREATE TABLE IF NOT EXISTS sync2_profiles ( + link TEXT PRIMARY KEY, + name TEXT NOT NULL, + hlc TEXT NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0 +); + +-- A link read from a snapshot already includes its batches up to `seq` of each sender. +CREATE TABLE IF NOT EXISTS sync2_floors ( + link TEXT NOT NULL, + did TEXT NOT NULL, + seq INTEGER NOT NULL, + PRIMARY KEY (link, did) +); + +-- Links a profile joined that are still to be read from their snapshot. +CREATE TABLE IF NOT EXISTS sync2_reading ( + link TEXT PRIMARY KEY +); diff --git a/src-tauri/src/sync2/batches.rs b/src-tauri/src/sync2/batches.rs index 15c4baa2..b81f18e3 100644 --- a/src-tauri/src/sync2/batches.rs +++ b/src-tauri/src/sync2/batches.rs @@ -40,10 +40,13 @@ pub struct Batch { #[derive(Debug, Default)] pub struct Outgoing { pub batches: Vec, - sent: Vec<(Collection, String, String)>, + sent: Vec, now_ms: i64, } +/// A record as uploaded or received: its link, collection, key and hash. +type Sent = (String, Collection, String, String); + fn record_key(collection: Collection, record: &Value) -> Option { let field = |name: &str| record.get(name)?.as_str().map(str::to_owned); match collection { @@ -81,13 +84,16 @@ fn record_hash(collection: Collection, record: &Value, line: &[u8]) -> String { async fn sent_hashes( pool: &SqlitePool, + link: &str, collection: Collection, ) -> Result, SyncError> { - let rows: Vec<(String, String, i64)> = - sqlx::query_as("SELECT record_key, hash, sent_at FROM sync2_sent WHERE collection = ?") - .bind(collection.key()) - .fetch_all(pool) - .await?; + let rows: Vec<(String, String, i64)> = sqlx::query_as( + "SELECT record_key, hash, sent_at FROM sync2_sent_links WHERE link = ? AND collection = ?", + ) + .bind(link) + .bind(collection.key()) + .fetch_all(pool) + .await?; Ok(rows .into_iter() .map(|(key, hash, at)| (key, (hash, at))) @@ -130,7 +136,7 @@ pub async fn outgoing_batches( }; for exported in exported { let collection = exported.collection; - let sent = sent_hashes(pool, collection).await?; + let sent = sent_hashes(pool, link, collection).await?; let mut changed = Vec::new(); for line in exported .ndjson @@ -147,7 +153,7 @@ pub async fn outgoing_batches( && previous.is_some_and(|(_, at)| now_ms - at < BRAIN_INTERVAL_MS); if previous.map(|(h, _)| h) != Some(&hash) && !too_soon { changed.push(line.to_vec()); - outgoing.sent.push((collection, key, hash)); + outgoing.sent.push((link.to_owned(), collection, key, hash)); } } for records in changed.chunks(RECORDS_PER_BATCH) { @@ -167,14 +173,11 @@ pub async fn mark_sent(pool: &SqlitePool, outgoing: &Outgoing) -> Result<(), Syn /// The record of what was sent is the device's, in Default's database like the rest of the group's /// state, while `pool` may be the running profile's. -async fn remember( - pool: &SqlitePool, - records: &[(Collection, String, String)], - now_ms: i64, -) -> Result<(), SyncError> { +async fn remember(pool: &SqlitePool, records: &[Sent], now_ms: i64) -> Result<(), SyncError> { let mut tx = crate::db::device::pool(pool).begin().await?; - for (collection, key, hash) in records { - sqlx::query("INSERT INTO sync2_sent (collection, record_key, hash, sent_at) VALUES (?, ?, ?, ?) ON CONFLICT (collection, record_key) DO UPDATE SET hash = excluded.hash, sent_at = excluded.sent_at") + for (link, collection, key, hash) in records { + sqlx::query("INSERT INTO sync2_sent_links (link, collection, record_key, hash, sent_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT (link, collection, record_key) DO UPDATE SET hash = excluded.hash, sent_at = excluded.sent_at") + .bind(link) .bind(collection.key()) .bind(key) .bind(hash) @@ -224,9 +227,9 @@ pub async fn apply_batches( now_ms: i64, ) -> Result { let mut staged = Vec::new(); - let mut received = BTreeSet::new(); + let mut received: BTreeMap<(&str, String), BTreeSet<&str>> = BTreeMap::new(); for batch in batches { - let Some(collection) = batch.collection else { + let (Some(collection), Some(link)) = (batch.collection, batch.link.as_deref()) else { continue; }; let lines: Vec<&[u8]> = batch @@ -237,7 +240,10 @@ pub async fn apply_batches( for line in &lines { let record: Value = serde_json::from_slice(line)?; if let Some(key) = record_key(collection, &record) { - received.insert((collection.key(), key)); + received + .entry((collection.key(), key)) + .or_default() + .insert(link); } } let ndjson = lines.join(&b'\n'); @@ -267,10 +273,15 @@ pub async fn apply_batches( .filter(|l| !l.is_empty()) { let record: Value = serde_json::from_slice(line)?; - if let Some(key) = record_key(collection, &record) - && received.contains(&(collection.key(), key.clone())) - { - known.push((collection, key, record_hash(collection, &record, line))); + let Some(key) = record_key(collection, &record) else { + continue; + }; + let Some(links) = received.get(&(collection.key(), key.clone())) else { + continue; + }; + let hash = record_hash(collection, &record, line); + for link in links { + known.push(((*link).to_owned(), collection, key.clone(), hash.clone())); } } } diff --git a/src-tauri/src/sync2/state.rs b/src-tauri/src/sync2/state.rs index 065d88ed..f2f344fa 100644 --- a/src-tauri/src/sync2/state.rs +++ b/src-tauri/src/sync2/state.rs @@ -179,6 +179,12 @@ pub async fn forget_group(pool: &SqlitePool) -> Result<(), SyncError> { "DELETE FROM sync2_roster", "DELETE FROM sync2_streams", "DELETE FROM sync2_sent", + "DELETE FROM sync2_sent_links", + "DELETE FROM sync2_profile_links", + "DELETE FROM sync2_kept", + "DELETE FROM sync2_profiles", + "DELETE FROM sync2_floors", + "DELETE FROM sync2_reading", "DELETE FROM sync2_snapshots", "DELETE FROM sync2_acks", "DELETE FROM sync2_manifests", diff --git a/src-tauri/tests/profiles/store_inventory.tsv b/src-tauri/tests/profiles/store_inventory.tsv index 69b913c7..54b218da 100644 --- a/src-tauri/tests/profiles/store_inventory.tsv +++ b/src-tauri/tests/profiles/store_inventory.tsv @@ -9,7 +9,7 @@ table sync_identity DEVICE - The sync device id; stays in Default's file where o table sync_peers DEFAULT_ONLY sync Only Default syncs table sync_log DEFAULT_ONLY sync Only Default syncs table sync_changes DEFAULT_ONLY sync Only Default syncs; local edits are not recorded elsewhere -table sync2_sent DEVICE - The group's state, in Default's file whichever profile runs: the FLOW-SYNC/2 upload ledger +table sync2_sent DEAD - Replaced by sync2_sent_links; still cleared when a group is forgotten table sync2_group DEVICE - The group's state, in Default's file whichever profile runs: the FLOW-SYNC/2 group and log positions table sync2_roster DEVICE - The group's state, in Default's file whichever profile runs: the verified FLOW-SYNC/2 roster table sync2_streams DEVICE - The group's state, in Default's file whichever profile runs: how far each device's log was read @@ -24,6 +24,12 @@ table sync2_held_objects DEVICE - The group's state, in Default's file whichever table sync2_live_seen DEVICE - The group's state, in Default's file whichever profile runs: the highest live counter accepted from each device table sync2_consent DEVICE - The group's state, in Default's file whichever profile runs: which devices may control playback here table sync2_pending DEVICE - The group's state, in Default's file whichever profile runs: objects written while the server was out of reach +table sync2_sent_links DEVICE - The group's state, in Default's file whichever profile runs: the FLOW-SYNC/2 upload ledger, per link +table sync2_profile_links DEVICE - The group's state, in Default's file whichever profile runs: which local profile syncs which link +table sync2_kept DEVICE - The group's state, in Default's file whichever profile runs: batches kept for a linked profile that isn't running +table sync2_profiles DEVICE - The group's state, in Default's file whichever profile runs: the profiles collection, each link's name +table sync2_floors DEVICE - The group's state, in Default's file whichever profile runs: how far a link read from a snapshot already goes +table sync2_reading DEVICE - The group's state, in Default's file whichever profile runs: links still to be read from their snapshot table watch_history PROFILE library Video and music history (is_music) table notifications PROFILE library New-upload inbox written by the poll loop for this profile's subscriptions table recommendation_events PROFILE private FlowNeuro event log diff --git a/src-tauri/tests/sync2_device_state.rs b/src-tauri/tests/sync2_device_state.rs index c0e792e5..c1c1aef8 100644 --- a/src-tauri/tests/sync2_device_state.rs +++ b/src-tauri/tests/sync2_device_state.rs @@ -14,7 +14,7 @@ async fn sent_rows(path: &std::path::Path) -> i64 { let pool = SqlitePool::connect(&format!("sqlite://{}", path.display())) .await .unwrap(); - let rows = sqlx::query_scalar("SELECT COUNT(*) FROM sync2_sent") + let rows = sqlx::query_scalar("SELECT COUNT(*) FROM sync2_sent_links") .fetch_one(&pool) .await .unwrap(); diff --git a/src-tauri/tests/sync2_link_storage.rs b/src-tauri/tests/sync2_link_storage.rs new file mode 100644 index 00000000..8dbb0c3f --- /dev/null +++ b/src-tauri/tests/sync2_link_storage.rs @@ -0,0 +1,74 @@ +//! Migration 0036 carries a group's upload ledger over to the ledger kept per link. + +use sqlx::SqlitePool; +use sqlx::migrate::Migrator; +use sqlx::sqlite::SqlitePoolOptions; + +const BEFORE_PROFILES: usize = 35; + +async fn migrated_to(count: usize) -> SqlitePool { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + let mut migrator: Migrator = sqlx::migrate!("./migrations"); + migrator.migrations = migrator.migrations[..count].to_vec().into(); + migrator.run(&pool).await.unwrap(); + pool +} + +async fn group_with_link(pool: &SqlitePool, link: &str) { + sqlx::query( + "INSERT INTO sync2_group (id, group_id, kind, store_url, link, own_seq, own_prev, cursor) + VALUES (1, 'g', 'server', 'https://s', ?, 0, '', 0)", + ) + .bind(link) + .execute(pool) + .await + .unwrap(); + sqlx::query("INSERT INTO sync2_sent (collection, record_key, hash, sent_at) VALUES ('likes', 'video:a', 'h', 7)") + .execute(pool) + .await + .unwrap(); +} + +async fn finish(pool: &SqlitePool) { + sqlx::migrate!("./migrations").run(pool).await.unwrap(); +} + +#[tokio::test] +async fn a_linked_group_keeps_what_it_sent_under_its_link() { + let pool = migrated_to(BEFORE_PROFILES).await; + group_with_link(&pool, "link-a").await; + finish(&pool).await; + + let sent: Vec<(String, String, String, String, i64)> = + sqlx::query_as("SELECT link, collection, record_key, hash, sent_at FROM sync2_sent_links") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + sent, + [( + "link-a".to_owned(), + "likes".to_owned(), + "video:a".to_owned(), + "h".to_owned(), + 7 + )] + ); +} + +#[tokio::test] +async fn a_restored_group_without_a_link_carries_nothing_over() { + let pool = migrated_to(BEFORE_PROFILES).await; + group_with_link(&pool, "").await; + finish(&pool).await; + + let sent: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync2_sent_links") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(sent, 0); +}