feat(sync): store profile links, kept batches and profile names

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
+);

src-tauri/src/sync2/batches.rs

@@ -40,10 +40,13 @@ pub struct Batch {
 #[derive(Debug, Default)]
 pub struct Outgoing {
     pub batches: Vec<Batch>,
-    sent: Vec<(Collection, String, String)>,
+    sent: Vec<Sent>,
     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<String> {
     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<BTreeMap<String, (String, i64)>, 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<ApplyReport, SyncError> {
     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()));
             }
         }
     }

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",

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

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();

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);
+}