From 4270029db529fec96a959eb9524c458f45e3b8b6 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 31 Oct 2026 18:54:15 +0200 Subject: [PATCH 1/2] refactor(sync): keep linked devices state per device --- src-tauri/src/commands/profiles.rs | 3 +- src-tauri/src/lib.rs | 8 +-- src-tauri/src/sync2/actions.rs | 34 +++++++---- src-tauri/src/sync2/batches.rs | 4 +- src-tauri/src/sync2/live_hub.rs | 13 ++-- src-tauri/src/sync2/service.rs | 62 +++++++++++++------- src-tauri/tests/profiles/store_inventory.tsv | 30 +++++----- src-tauri/tests/sync2_device_state.rs | 58 ++++++++++++++++++ 8 files changed, 153 insertions(+), 59 deletions(-) create mode 100644 src-tauri/tests/sync2_device_state.rs diff --git a/src-tauri/src/commands/profiles.rs b/src-tauri/src/commands/profiles.rs index d6c1ea12..be9bd7be 100644 --- a/src-tauri/src/commands/profiles.rs +++ b/src-tauri/src/commands/profiles.rs @@ -75,10 +75,11 @@ pub async fn profiles_delete( #[tauri::command] pub async fn profiles_prepare_switch( id: String, + app: AppHandle, service: State<'_, ProfileService>, sync: State<'_, Arc>, ) -> Result<(), ErrorResponse> { - let sync_busy = sync.is_busy().await; + let sync_busy = sync.is_busy().await || crate::sync2::service::is_syncing(&app); service .prepare_switch(&id, sync_busy) .map_err(ErrorResponse::from) diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 4b082204..1401a9a8 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -497,11 +497,9 @@ pub fn run() { app.manage(sync2::link_task::LinkTask::default()); app.manage(sync2::lan_peers::LanPeers::default()); app.manage(sync2::live_hub::LiveHub::default()); - if active_profile.is_default { - sync2::service::spawn(app.handle().clone()); - sync2::lan_runtime::spawn(app.handle().clone()); - sync2::live::spawn(app.handle().clone()); - } + sync2::service::spawn(app.handle().clone()); + sync2::lan_runtime::spawn(app.handle().clone()); + sync2::live::spawn(app.handle().clone()); // Manage the Shorts feed service (prefetch buffer + session de-dup) app.manage(ShortsService::new()); diff --git a/src-tauri/src/sync2/actions.rs b/src-tauri/src/sync2/actions.rs index a1e6f4a9..6611c550 100644 --- a/src-tauri/src/sync2/actions.rs +++ b/src-tauri/src/sync2/actions.rs @@ -42,18 +42,22 @@ const NAME_LIMIT: usize = 64; /// This installation as the group sees it. #[derive(Clone)] pub struct Local { + /// The group's state on this device, in Default's database whichever profile runs. pub pool: SqlitePool, + /// The running profile's database, which batches are exported from and applied to. + pub data: SqlitePool, + pub profile: ActiveProfile, pub device: MemberSecrets, pub meta: DeviceMeta, } -/// Like FLOW-SYNC/1, only the Default profile syncs until profiles get their own links. +/// The database holding the group's state: Default's, whichever profile runs. +pub fn state_pool(app: &AppHandle) -> SqlitePool { + crate::db::device::pool(app.state::().inner()).clone() +} + pub async fn local(app: &AppHandle) -> Result { - let profile = app.state::(); - if !profile.is_default { - return Err(SyncError::NotDefaultProfile(profile.default_name.clone()).into()); - } - let pool = app.state::().inner().clone(); + let pool = state_pool(app); let meta = DeviceMeta { app: ledger::get_or_create_device_id(&pool).await?, name: ledger::device_name(&pool).await?, @@ -61,6 +65,8 @@ pub async fn local(app: &AppHandle) -> Result { }; Ok(Local { device: DEVICE_KEYS.get().await?, + data: app.state::().inner().clone(), + profile: app.state::().inner().clone(), pool, meta, }) @@ -97,8 +103,12 @@ pub async fn open_session<'a>( } /// Checks that this device is in no group, and gives it fresh keys for the one it is about to -/// create or join. +/// create or join. Like FLOW-SYNC/1, only the Default profile links until profiles choose links. pub async fn start_membership(local: &mut Local) -> Result<()> { + if !local.profile.is_default { + let name = local.profile.default_name.clone(); + return Err(SyncError::NotDefaultProfile(name).into()); + } if load_group(&local.pool).await?.is_some() { return Err(EngineError::AlreadyLinked); } @@ -168,7 +178,7 @@ pub struct LinkedStatus { /// What can be shown without reaching the server. pub async fn status(app: &AppHandle) -> Result { - let pool = app.state::().inner().clone(); + let pool = state_pool(app); let saved = get_setting(&pool, RECOVERY_SAVED).await?; let record = load_group(&pool).await?; Ok(LinkedStatus { @@ -355,7 +365,7 @@ pub async fn retry_keychain(app: &AppHandle) -> Result<()> { } pub async fn recovery_saved(app: &AppHandle) -> Result<()> { - set_recovery_saved(app.state::().inner(), true).await + set_recovery_saved(&state_pool(app), true).await } pub async fn rename(app: &AppHandle, id: &str, name: &str) -> Result<()> { @@ -417,7 +427,7 @@ pub async fn leave(app: &AppHandle) -> Result<()> { /// [`leave`], when this device is in a group. pub async fn leave_if_linked(app: &AppHandle) -> Result<()> { - match load_group(app.state::().inner()).await? { + match load_group(&state_pool(app)).await? { Some(_) => leave(app).await, None => Ok(()), } @@ -447,14 +457,14 @@ pub async fn depart( /// Forgets the group and this device's keys without telling the group, for a device it already /// shut out. pub async fn forget(app: &AppHandle) -> Result<()> { - forget_with_keys(app.state::().inner()).await + forget_with_keys(&state_pool(app)).await } /// Clears history, subscriptions and playlists. When this profile syncs, the caller first leaves /// the group ([`leave_if_linked`]), and the deletions recorded here are dropped, so a reset never /// reaches the other devices; linking again brings their data back. pub async fn reset_local_data(pool: &SqlitePool, syncs: bool) -> Result<()> { - if syncs && load_group(pool).await?.is_some() { + if syncs && load_group(crate::db::device::pool(pool)).await?.is_some() { return Err(EngineError::AlreadyLinked); } crate::db::watch_history::clear_watch_history(pool).await?; diff --git a/src-tauri/src/sync2/batches.rs b/src-tauri/src/sync2/batches.rs index 1ae8229a..15c4baa2 100644 --- a/src-tauri/src/sync2/batches.rs +++ b/src-tauri/src/sync2/batches.rs @@ -165,12 +165,14 @@ pub async fn mark_sent(pool: &SqlitePool, outgoing: &Outgoing) -> Result<(), Syn remember(pool, &outgoing.sent, outgoing.now_ms).await } +/// 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> { - let mut tx = pool.begin().await?; + 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") .bind(collection.key()) diff --git a/src-tauri/src/sync2/live_hub.rs b/src-tauri/src/sync2/live_hub.rs index b38cedfc..37afe78a 100644 --- a/src-tauri/src/sync2/live_hub.rs +++ b/src-tauri/src/sync2/live_hub.rs @@ -10,13 +10,14 @@ use sqlx::SqlitePool; use tauri::{AppHandle, Emitter, Manager}; use tokio::sync::broadcast; -use super::actions::{local, open_session}; +use super::actions::{local, open_session, state_pool}; use super::engine::{EngineError, Result, now_ms}; use super::lan_peers::nearby; use super::live_message::{Ability, Answer, DeviceKind, LiveMessage, Now}; use super::live_seal::{LiveKeys, Received}; use super::live_state::{accept_live_seq, consents, next_live_seq}; use super::state::load_group; +use crate::profiles::ActiveProfile; /// The devices' presence and what they play, whenever either changes. pub const EVENT_LIVE: &str = "sync2://live"; @@ -98,7 +99,7 @@ pub async fn forget_keys(app: &AppHandle) { async fn keys(app: &AppHandle) -> Result<(LiveKeys, SqlitePool)> { let mut cached = hub(app).keys.lock().await; - let pool = app.state::().inner().clone(); + let pool = state_pool(app); if let Some(keys) = cached.as_ref() { return Ok((keys.clone(), pool)); } @@ -138,10 +139,12 @@ pub async fn announce(app: &AppHandle) -> Result<()> { Ok(()) } -/// The profile link this device syncs, which `now` and `play` name. +/// The profile link the running profile syncs, which `now` and `play` name. async fn own_link(app: &AppHandle) -> Result> { - let pool = app.state::().inner().clone(); - Ok(load_group(&pool) + if !app.state::().is_default { + return Ok(None); + } + Ok(load_group(&state_pool(app)) .await? .map(|record| record.link) .filter(|link| !link.is_empty())) diff --git a/src-tauri/src/sync2/service.rs b/src-tauri/src/sync2/service.rs index a5c05d0f..2e324282 100644 --- a/src-tauri/src/sync2/service.rs +++ b/src-tauri/src/sync2/service.rs @@ -12,10 +12,12 @@ use tauri::{AppHandle, Emitter, Manager}; use tokio::sync::Mutex; use super::actions::{ - default_link, local, login_refused, open_session, set_login_refused, stop_if_shut_out, stopped, + Local, default_link, local, login_refused, open_session, set_login_refused, state_pool, + stop_if_shut_out, stopped, }; use super::batches::{IncomingBatch, apply_batches}; -use super::engine::{EngineError, Result, now_ms}; +use super::engine::{EngineError, Result, Session, now_ms}; +use super::group_store::GroupStore; use super::live_hub::forget_keys; use super::names::device_names; use super::server::StoreError; @@ -78,7 +80,7 @@ pub fn last_sync(app: &AppHandle) -> Option { pub async fn sync_now(app: &AppHandle) -> Result> { let state = app.state::(); let _running = state.running.lock().await; - let pool = app.state::().inner().clone(); + let pool = state_pool(app); if load_group(&pool).await?.is_none() || stopped(&pool).await?.is_some() || login_refused(&pool).await? @@ -112,14 +114,42 @@ async fn run(app: &AppHandle) -> Result> { let Some(record) = load_group(&local.pool).await? else { return Ok(Vec::new()); }; - let pool = &local.pool; - let mut session = open_session(pool, &record, &local.device).await?; + let mut session = open_session(&local.pool, &record, &local.device).await?; + // Only the linked profile's data is exchanged; other profiles keep the group going. + let received = if local.profile.is_default { + let Some(received) = exchange(app, &local, &mut session).await? else { + return Ok(Vec::new()); + }; + received + } else { + session.store().upload_pending().await?; + Vec::new() + }; + session.collect_garbage(&local.pool).await?; + // A server without join requests (older than flow-sync#41) shouldn't fail the sync. + if let Some(server) = session.store().server() + && let Ok(requests) = server.join_requests().await + && !requests.is_empty() + { + let _ = app.emit(EVENT_JOIN_REQUESTS, requests); + } + Ok(received) +} + +/// Pushes the running profile's changes and applies everyone else's. `None` while a restored +/// device can't choose its link yet. +async fn exchange( + app: &AppHandle, + local: &Local, + session: &mut Session<'_, GroupStore>, +) -> Result>> { + let (pool, data) = (&local.pool, &local.data); if !session.adopt_link(pool, &default_link(app)).await? { - return Ok(Vec::new()); + return Ok(None); } flush_brains(app).await; - let mut exported = export_collections(pool, &local.meta.app, &Collection::ALL).await?; + let mut exported = export_collections(data, &local.meta.app, &Collection::ALL).await?; enrich_outgoing_albums(app, &mut exported).await; session.store().upload_pending().await?; session.push(pool, exported).await?; @@ -128,7 +158,7 @@ async fn run(app: &AppHandle) -> Result> { let (changed, received) = if pulled.batches.is_empty() { (Vec::new(), BTreeMap::new()) } else { - apply(app, pool, &local.meta.app, &pulled.batches).await? + apply(app, data, &local.meta.app, &pulled.batches).await? }; if !changed.is_empty() { let _ = app.emit(EVENT_REFRESH, &changed); @@ -137,9 +167,9 @@ async fn run(app: &AppHandle) -> Result> { let names = if received.is_empty() { BTreeMap::new() } else { - device_names(&mut session).await.unwrap_or_default() + device_names(session).await.unwrap_or_default() }; - let received: Vec = received + let received = received .into_iter() .map(|(sender, count)| Received { name: names @@ -150,18 +180,10 @@ async fn run(app: &AppHandle) -> Result> { }) .collect(); if snapshot_due(pool, &session.record().link).await? { - let exported = export_collections(pool, &local.meta.app, &Collection::ALL).await?; + let exported = export_collections(data, &local.meta.app, &Collection::ALL).await?; session.write_snapshot(pool, exported).await?; } - session.collect_garbage(pool).await?; - // A server without join requests (older than flow-sync#41) shouldn't fail the sync. - if let Some(server) = session.store().server() - && let Ok(requests) = server.join_requests().await - && !requests.is_empty() - { - let _ = app.emit(EVENT_JOIN_REQUESTS, requests); - } - Ok(received) + Ok(Some(received)) } /// Applies each sender's batches, and returns the collections they touched and how many records diff --git a/src-tauri/tests/profiles/store_inventory.tsv b/src-tauri/tests/profiles/store_inventory.tsv index 7496c0e8..69b913c7 100644 --- a/src-tauri/tests/profiles/store_inventory.tsv +++ b/src-tauri/tests/profiles/store_inventory.tsv @@ -9,21 +9,21 @@ 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 DEFAULT_ONLY sync Only Default syncs; the FLOW-SYNC/2 upload ledger -table sync2_group DEFAULT_ONLY sync Only Default syncs; the FLOW-SYNC/2 group and log positions -table sync2_roster DEFAULT_ONLY sync Only Default syncs; the verified FLOW-SYNC/2 roster -table sync2_streams DEFAULT_ONLY sync Only Default syncs; how far each device's log was read -table sync2_snapshots DEFAULT_ONLY sync Only Default syncs; when each link last got a snapshot -table sync2_acks DEFAULT_ONLY sync Only Default syncs; each device's latest FLOW-SYNC/2 ack -table sync2_manifests DEFAULT_ONLY sync Only Default syncs; snapshot manifests seen, for garbage collection -table sync2_links DEFAULT_ONLY sync Only Default syncs; links that batches name, for garbage collection -table sync2_collected DEFAULT_ONLY sync Only Default syncs; how far each log was deleted -table sync2_held_roster DEFAULT_ONLY sync Only Default syncs; roster entries a LAN group keeps on its devices -table sync2_held_keys DEFAULT_ONLY sync Only Default syncs; wraps and chain boxes a LAN group keeps -table sync2_held_objects DEFAULT_ONLY sync Only Default syncs; log objects and blobs a LAN group keeps -table sync2_live_seen DEFAULT_ONLY sync Only Default syncs; the highest live counter accepted from each device -table sync2_consent DEFAULT_ONLY sync Only Default syncs; which devices may control playback here -table sync2_pending DEFAULT_ONLY sync Only Default syncs; objects written while the server was out of reach +table sync2_sent DEVICE - The group's state, in Default's file whichever profile runs: the FLOW-SYNC/2 upload ledger +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 +table sync2_snapshots DEVICE - The group's state, in Default's file whichever profile runs: when each link last got a snapshot +table sync2_acks DEVICE - The group's state, in Default's file whichever profile runs: each device's latest FLOW-SYNC/2 ack +table sync2_manifests DEVICE - The group's state, in Default's file whichever profile runs: snapshot manifests seen, for garbage collection +table sync2_links DEVICE - The group's state, in Default's file whichever profile runs: links that batches name, for garbage collection +table sync2_collected DEVICE - The group's state, in Default's file whichever profile runs: how far each log was deleted +table sync2_held_roster DEVICE - The group's state, in Default's file whichever profile runs: roster entries a LAN group keeps on its devices +table sync2_held_keys DEVICE - The group's state, in Default's file whichever profile runs: wraps and chain boxes a LAN group keeps +table sync2_held_objects DEVICE - The group's state, in Default's file whichever profile runs: log objects and blobs a LAN group keeps +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 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 new file mode 100644 index 00000000..c0e792e5 --- /dev/null +++ b/src-tauri/tests/sync2_device_state.rs @@ -0,0 +1,58 @@ +//! A profile other than Default applies batches into its own database, while the group's record +//! of what was sent stays in Default's. Runs in its own test binary because the device pool is +//! process-wide. + +mod sync2_common; + +use flow_desktop_lib::profiles::{ActiveProfile, open_database}; +use flow_desktop_lib::sync::canonical::Collection; +use flow_desktop_lib::sync2::batches::{apply_batches, batch_content, parse_batch}; +use sqlx::SqlitePool; +use sync2_common::{empty_pool, exported, watch, watched}; + +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") + .fetch_one(&pool) + .await + .unwrap(); + pool.close().await; + rows +} + +#[tokio::test] +async fn applied_records_count_as_sent_in_defaults_database() { + let dir = std::env::temp_dir().join(format!("flow-sync2-state-{}", uuid::Uuid::new_v4())); + let study = ActiveProfile { + id: "study".into(), + is_default: false, + after_switch: false, + default_name: String::new(), + }; + let pool = open_database(&dir, &study).await.unwrap(); + + let sender = empty_pool().await; + watch(&sender, "aaaaaaaaaaa").await; + let history = exported(&sender, "app-a") + .await + .into_iter() + .find(|c| c.collection == Collection::WatchHistory) + .unwrap(); + let records: Vec<&[u8]> = history.ndjson.split(|&b| b == b'\n').collect(); + let content = batch_content(Collection::WatchHistory, "head", "link", &records); + let batch = parse_batch(&content).unwrap(); + apply_batches(&pool, "app-b", "sender", &[batch], 0) + .await + .unwrap(); + assert_eq!(watched(&pool).await, ["aaaaaaaaaaa"]); + pool.close().await; + + assert_eq!(sent_rows(&dir.join("flow-desktop.sqlite")).await, 1); + assert_eq!( + sent_rows(&dir.join("profiles/study/flow-desktop.sqlite")).await, + 0 + ); + std::fs::remove_dir_all(dir).unwrap(); +} From c21e2cf84f7e0f04789a547e3888f79a2c4300a0 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 31 Oct 2026 18:57:54 +0200 Subject: [PATCH 2/2] feat(sync): store profile links, kept batches and profile names --- src-tauri/migrations/0036_sync2_profiles.sql | 57 +++++++++++++++ src-tauri/src/sync2/batches.rs | 55 +++++++++------ src-tauri/src/sync2/state.rs | 6 ++ src-tauri/tests/profiles/store_inventory.tsv | 8 ++- src-tauri/tests/sync2_device_state.rs | 2 +- src-tauri/tests/sync2_link_storage.rs | 74 ++++++++++++++++++++ 6 files changed, 178 insertions(+), 24 deletions(-) create mode 100644 src-tauri/migrations/0036_sync2_profiles.sql create mode 100644 src-tauri/tests/sync2_link_storage.rs 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); +}