Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/v0.1.0-beta4.txt
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ SYNC
- Play on: picking a device now works as on Android. This computer takes playback back from the device you control, picking another device moves that device's session there, and a device that stops answering stops showing as playing.
- Linked devices: choose which profiles sync. A profile starts its own synced profile or joins one from your other devices, and Profiles under Linked devices adds one that only your other devices have.
- A synced profile that isn't open keeps what arrives and catches up when you switch to it. Stopping a profile on every device keeps its data on each one.
- Linked devices has a Dashboard: where the group lives, how much of a Flow server's storage it uses, your devices, each synced profile's state and the latest sync errors.

FIXES
- Restarting Flow after an update no longer opens the link Flow was first started with a second time.
Expand Down
6 changes: 6 additions & 0 deletions src-tauri/src/commands/linked.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ use tauri::AppHandle;

use crate::errors::ErrorResponse;
use crate::sync2::actions::{self, LinkedDevice, LinkedStatus};
use crate::sync2::dashboard::{self, Dashboard};
use crate::sync2::engine::EngineError;
use crate::sync2::join_requests::JoinRequest;
use crate::sync2::link_task::{self, LinkPhase};
Expand Down Expand Up @@ -242,3 +243,8 @@ pub async fn linked_choose_profile(
.await
.map_err(|e| err(&e))
}

#[tauri::command]
pub async fn linked_dashboard(app: AppHandle) -> Result<Dashboard, ErrorResponse> {
dashboard::dashboard(&app).await.map_err(|e| err(&e))
}
11 changes: 6 additions & 5 deletions src-tauri/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,11 +63,11 @@ use commands::files::{write_backup_file, write_lyrics_file};
use commands::linked::{
linked_accept_join, linked_add_device, linked_answer, linked_ask_to_join, linked_cancel,
linked_catch_up, linked_choose_profile, linked_create, linked_create_lan, linked_create_webdav,
linked_decline_join, linked_delete_profile, linked_devices, linked_enter_code, linked_forget,
linked_join_requests, linked_leave, linked_link_phase, linked_profiles, linked_recovery_saved,
linked_remove, linked_rename, linked_restore, linked_retry_keychain, linked_show_code,
linked_status, linked_stop_profile, linked_sync_added_profile, linked_sync_now,
linked_sync_profile, linked_update_login,
linked_dashboard, linked_decline_join, linked_delete_profile, linked_devices,
linked_enter_code, linked_forget, linked_join_requests, linked_leave, linked_link_phase,
linked_profiles, linked_recovery_saved, linked_remove, linked_rename, linked_restore,
linked_retry_keychain, linked_show_code, linked_status, linked_stop_profile,
linked_sync_added_profile, linked_sync_now, linked_sync_profile, linked_update_login,
};
use commands::live::{live_consents, live_now, live_send, live_set_consent, live_status};
use commands::lyrics::{
Expand Down Expand Up @@ -826,6 +826,7 @@ pub fn run() {
linked_stop_profile,
linked_delete_profile,
linked_choose_profile,
linked_dashboard,
linked_link_phase,
linked_add_device,
linked_show_code,
Expand Down
1 change: 1 addition & 0 deletions src-tauri/src/sync2/actions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -500,6 +500,7 @@ async fn forget_with_keys(pool: &SqlitePool) -> Result<()> {
/// Forgets the group on a database, keeping this device's keys.
async fn forget_here(pool: &SqlitePool) -> Result<()> {
forget_group(pool).await?;
super::error_log::clear(pool).await?;
set_setting(pool, STOPPED, "").await?;
set_login_refused(pool, false).await?;
set_recovery_saved(pool, true).await
Expand Down
104 changes: 104 additions & 0 deletions src-tauri/src/sync2/dashboard.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
//! What the Sync page's dashboard shows about the group: where it lives, the store's usage,
//! recent failures, and each synced profile's state on this device. Devices come from
//! [`super::actions::devices`].

use serde::Serialize;
use sqlx::SqlitePool;
use tauri::AppHandle;

use super::actions::local;
use super::engine::{EngineError, Result};
use super::error_log::{LoggedError, recent};
use super::kept::kept_count;
use super::profile_links::linked;
use super::server::{ServerInfo, ServerStore};
use super::snapshot::own_snapshot_ms;
use super::state::{GroupRecord, StoreKind, load_group};

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Storage {
pub used: u64,
pub quota: u64,
pub objects: u64,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ProfileState {
pub profile: String,
pub link: String,
/// Batches waiting for the profile to run.
pub kept: i64,
/// When this device last wrote a snapshot of the link.
pub own_snapshot_ms: Option<i64>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Dashboard {
pub kind: StoreKind,
/// The server or folder; `None` for a group on the LAN.
pub store_url: Option<String>,
/// What a Flow server speaks, when it answered.
pub server: Option<ServerInfo>,
/// A Flow server's usage, when it answered. Other stores aren't listed to measure it.
pub storage: Option<Storage>,
pub errors: Vec<LoggedError>,
pub profiles: Vec<ProfileState>,
}

pub async fn dashboard(app: &AppHandle) -> Result<Dashboard> {
let local = local(app).await?;
let record = load_group(&local.pool)
.await?
.ok_or(EngineError::NotLinked)?;
let (server, storage) = match record.kind {
StoreKind::Server => server_state(&record, &local.device).await,
StoreKind::WebDav | StoreKind::Lan => (None, None),
};
Ok(Dashboard {
kind: record.kind,
store_url: (record.kind != StoreKind::Lan).then(|| record.store_url.clone()),
server,
storage,
errors: recent(&local.pool).await?,
profiles: profile_states(&local.pool, &hex::encode(local.device.did)).await?,
})
}

/// Best effort: the rest of the dashboard shows when the server can't be asked.
async fn server_state(
record: &GroupRecord,
device: &super::keys::MemberSecrets,
) -> (Option<ServerInfo>, Option<Storage>) {
let Ok(store) = ServerStore::connect(
&record.store_url,
device.clone(),
record.instance.clone(),
Some(record.group_id.clone()),
)
.await
else {
return (None, None);
};
let storage = store.status().await.ok().map(|status| Storage {
used: status.used,
quota: status.quota,
objects: status.objects,
});
(store.info().await.ok(), storage)
}

async fn profile_states(pool: &SqlitePool, me: &str) -> Result<Vec<ProfileState>> {
let mut states = Vec::new();
for (profile, link) in linked(pool).await? {
states.push(ProfileState {
kept: kept_count(pool, &link).await?,
own_snapshot_ms: own_snapshot_ms(pool, me, &link).await?,
profile,
link,
});
}
Ok(states)
}
44 changes: 44 additions & 0 deletions src-tauri/src/sync2/error_log.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
//! Recent sync failures for the dashboard: the error kind and when, nothing else, newest first.
//! Kept in Default's settings with the rest of the group's state.

use serde::{Deserialize, Serialize};
use sqlx::SqlitePool;

use super::engine::Result;
use crate::db::settings::{get_setting, set_setting};

const ERRORS_KEY: &str = "sync2_errors";
const KEPT: usize = 20;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LoggedError {
pub kind: String,
pub at_ms: i64,
}

pub async fn recent(pool: &SqlitePool) -> Result<Vec<LoggedError>> {
Ok(get_setting(pool, ERRORS_KEY)
.await?
.and_then(|json| serde_json::from_str(&json).ok())
.unwrap_or_default())
}

/// Keeps `kind` as the newest failure, dropping the oldest beyond a few.
pub async fn note(pool: &SqlitePool, kind: &str, at_ms: i64) -> Result<()> {
let mut errors = recent(pool).await?;
errors.insert(
0,
LoggedError {
kind: kind.to_owned(),
at_ms,
},
);
errors.truncate(KEPT);
let json = serde_json::to_string(&errors).map_err(crate::sync::error::SyncError::from)?;
Ok(set_setting(pool, ERRORS_KEY, &json).await?)
}

pub async fn clear(pool: &SqlitePool) -> Result<()> {
Ok(set_setting(pool, ERRORS_KEY, "[]").await?)
}
11 changes: 5 additions & 6 deletions src-tauri/src/sync2/kept.rs
Original file line number Diff line number Diff line change
Expand Up @@ -78,13 +78,12 @@ pub async fn applied(pool: &SqlitePool, ids: &[i64]) -> Result<(), SyncError> {
Ok(())
}

/// Whether batches of `link` wait to be applied.
pub async fn has_kept(pool: &SqlitePool, link: &str) -> Result<bool, SyncError> {
/// How many batches of `link` wait to be applied.
pub async fn kept_count(pool: &SqlitePool, link: &str) -> Result<i64, SyncError> {
Ok(
sqlx::query_scalar::<_, i64>("SELECT 1 FROM sync2_kept WHERE link = ? LIMIT 1")
sqlx::query_scalar("SELECT COUNT(*) FROM sync2_kept WHERE link = ?")
.bind(link)
.fetch_optional(pool)
.await?
.is_some(),
.fetch_one(pool)
.await?,
)
}
2 changes: 2 additions & 0 deletions src-tauri/src/sync2/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,14 @@ pub mod blob;
pub mod bootstrap;
pub mod codes;
pub mod curve;
pub mod dashboard;
pub mod device;
pub mod engine;
pub mod entries;
pub mod envelope;
pub mod epoch;
pub mod error;
pub mod error_log;
pub mod gc;
pub mod group_store;
pub mod held;
Expand Down
4 changes: 2 additions & 2 deletions src-tauri/src/sync2/profile_links.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ use std::collections::BTreeSet;

use sqlx::SqlitePool;

use super::kept::has_kept;
use super::kept::kept_count;
use super::state::links_to_read;
use crate::sync::error::SyncError;

Expand Down Expand Up @@ -41,7 +41,7 @@ impl Reading {

/// Whether `link` still waits here on data other devices sent: kept batches, or its snapshot.
pub async fn catching_up(pool: &SqlitePool, link: &str) -> Result<bool, SyncError> {
Ok(has_kept(pool, link).await? || links_to_read(pool).await?.iter().any(|l| l == link))
Ok(kept_count(pool, link).await? > 0 || links_to_read(pool).await?.iter().any(|l| l == link))
}

/// Links `profile` to `link`.
Expand Down
58 changes: 52 additions & 6 deletions src-tauri/src/sync2/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -233,14 +233,44 @@ impl ServerStore {
unreachable!("the loop returns on its second pass")
}

/// When the server last heard from each device, by `did` (`GET /v1/status`).
pub async fn last_seen(&self) -> Result<BTreeMap<String, i64>> {
/// The group's storage and when the server last heard from each device (`GET /v1/status`).
pub async fn status(&self) -> Result<ServerStatus> {
let status = decode(self.authorized(Method::GET, "v1/status", None).await?).await?;
let devices = status["devices"].as_object().ok_or(StoreError::Malformed)?;
Ok(devices
.iter()
.filter_map(|(did, device)| Some((did.clone(), device["seen"].as_i64()?)))
.collect())
let number = |name: &str| status[name].as_u64().ok_or(StoreError::Malformed);
Ok(ServerStatus {
used: number("used")?,
quota: number("quota")?,
objects: number("objects")?,
seen: devices
.iter()
.filter_map(|(did, device)| Some((did.clone(), device["seen"].as_i64()?)))
.collect(),
})
}

/// When the server last heard from each device, by `did`.
pub async fn last_seen(&self) -> Result<BTreeMap<String, i64>> {
Ok(self.status().await?.seen)
}

/// What the server speaks (`GET /v1/info`, no token needed).
pub async fn info(&self) -> Result<ServerInfo> {
let response = self
.client
.get(self.url("v1/info")?)
.send()
.await
.map_err(StoreError::Network)?;
let info = decode(response).await?;
let protocols = info["protocol"].as_array().ok_or(StoreError::Malformed)?;
Ok(ServerInfo {
protocols: protocols
.iter()
.filter_map(|p| p.as_str().map(str::to_owned))
.collect(),
live: info["live"].as_bool().unwrap_or(false),
})
}

/// A request with the operator's admin token instead of a device token.
Expand Down Expand Up @@ -465,6 +495,22 @@ async fn check(response: reqwest::Response) -> Result<reqwest::Response> {
})
}

/// The group's storage on a Flow server, and when it last heard from each device.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServerStatus {
pub used: u64,
pub quota: u64,
pub objects: u64,
pub seen: BTreeMap<String, i64>,
}

/// The protocols a Flow server speaks, and whether it relays live messages.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct ServerInfo {
pub protocols: Vec<String>,
pub live: bool,
}

pub(super) async fn decode(response: reqwest::Response) -> Result<Value> {
check(response)
.await?
Expand Down
10 changes: 8 additions & 2 deletions src-tauri/src/sync2/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@ use super::actions::{
};
use super::batches::{IncomingBatch, apply_batches};
use super::engine::{EngineError, Result, Session, now_ms};
use super::error_log;
use super::group_store::GroupStore;
use super::kept::{applied as applied_kept, has_kept, keep, kept};
use super::kept::{applied as applied_kept, keep, kept, kept_count};
use super::live_hub::forget_keys;
use super::names::device_names;
use super::profile_actions::name_linked;
Expand Down Expand Up @@ -104,6 +105,11 @@ pub async fn sync_now(app: &AppHandle) -> Result<Vec<Received>> {
at_ms: now_ms(),
error: result.as_ref().err().map(|e| e.kind().to_owned()),
};
if let Some(kind) = &last.error
&& let Err(failed) = error_log::note(&pool, kind, last.at_ms).await
{
tracing::warn!(target: "flow::sync2", kind = failed.kind(), "sync error not logged");
}
if let Ok(mut slot) = state.last.lock() {
*slot = Some(last.clone());
}
Expand Down Expand Up @@ -183,7 +189,7 @@ async fn exchange(
let received = applied.received(session).await;
if let Some(link) = &reading.active
&& snapshot_due(pool, link).await?
&& !has_kept(pool, link).await?
&& kept_count(pool, link).await? == 0
&& !links_to_read(pool).await?.contains(link)
{
let exported = export_collections(data, &local.meta.app, &Collection::ALL).await?;
Expand Down
Loading
Loading