feat: chunked resumable uploads in the sync driver and the phone sync sheet
ci / ci (macos-latest) (push) Has been cancelled
ci / ci (ubuntu-latest) (push) Has been cancelled
ci / ci (windows-latest) (push) Has been cancelled

- engine: Remote gains missing_chunks, put_chunk and put_chunked; upload_chunked streams a big file in 4 MiB chunks, sends only the chunks the server lacks, retries with backoff and reports progress (SyncStatus.transfer); the scan hashes by streaming and skips only files over 1 GiB
- engine: server.devices command and a sync.attachments option (non-Markdown files are skipped when it is off)
- ui-egui: the sync pill opens a bottom sheet on a phone-width window (status, devices, conflicts row, options, Sync now); wider windows keep Settings > Sync
- tests: chunk resume against the real server, 6 driven sheet tests, devices stub tests

Verified with cargo xtask ci (fmt, clippy -D warnings, workspace tests, layers, tree, i18n).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Marcos Paulo
2026-10-20 08:57:07 -03:00
parent a71e294566
commit 9bd7dc931e
37 changed files with 2771 additions and 77 deletions
@@ -0,0 +1,239 @@
//! The real engine sync client against the real server: a big file travels in
//! chunks, a cut connection resumes with the chunks that never arrived.
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use notescraft_engine::{Changes, HttpRemote, Limits, Progress, PutReply, Remote, SyncError, SyncOptions, SyncReport, SyncResult, SyncState, sync_once_with};
use notescraft_sync_server::{AppState, Store, router};
use notescraft_vault::Vault;
const ADMIN: &str = "test-admin-token-0123456789abcdef";
/// The size of one chunk the client sends.
const CHUNK: usize = 32 * 1024;
/// The size above which the client sends a file in chunks.
const THRESHOLD: u64 = 16 * 1024;
/// The biggest chunk the server takes.
const SERVER_CHUNK_MAX: usize = 64 * 1024;
/// The biggest file the server assembles.
const SERVER_FILE_CAP: u64 = 1024 * 1024;
/// The size of the file that travels.
const FILE_BYTES: usize = 300 * 1024;
/// A temp folder that is empty when the test starts.
fn temp(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("nc-chunk-e2e-{name}-{}", std::process::id()));
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).expect("dir");
dir
}
/// Starts the server on an ephemeral port and returns its base url.
fn start(data: &Path) -> String {
std::fs::create_dir_all(data.join("blobs")).expect("blobs");
let state = AppState::new(Store::open(&data.join("sync.db")).expect("db"), data.join("blobs"), ADMIN);
// The chunk limits are set before the router is built: the route takes its
// body cap from the state.
state.set_chunk_limits(SERVER_CHUNK_MAX, SERVER_FILE_CAP);
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
listener.set_nonblocking(true).expect("nonblocking");
let port = listener.local_addr().expect("addr").port();
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("runtime");
runtime.block_on(async {
let listener = tokio::net::TcpListener::from_std(listener).expect("listener");
axum::serve(listener, router(state)).await.ok();
});
});
format!("http://127.0.0.1:{port}")
}
fn admin(base: &str, path: &str) -> serde_json::Value {
let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
let mut response = agent.post(format!("{base}{path}")).header("Authorization", format!("Bearer {ADMIN}")).send_empty().expect("admin call");
response.body_mut().read_json().expect("json")
}
/// Issues a pairing code for the vault, bound to `user` when given.
fn issue(base: &str, vault_id: &str, user: Option<&str>) -> String {
let path = user.map_or_else(|| format!("/v1/vaults/{vault_id}/pairing"), |user| format!("/v1/vaults/{vault_id}/pairing?user={user}"));
admin(base, &path)["code"].as_str().expect("code").to_owned()
}
/// Redeems a code with raw HTTP to also learn the account the device joined.
fn pair_raw(base: &str, code: &str, name: &str) -> (String, String) {
let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
let mut response = agent.post(format!("{base}/v1/devices")).send_json(serde_json::json!({"code": code, "name": name})).expect("pair");
let device: serde_json::Value = response.body_mut().read_json().expect("json");
(device["token"].as_str().expect("token").to_owned(), device["user_id"].as_str().expect("user id").to_owned())
}
/// A remote that can cut the connection in the middle of a chunked upload.
struct CutRemote {
inner: HttpRemote,
/// How many chunks the client sent.
sent: Mutex<usize>,
/// From this send on, the connection is cut.
cut_from: Mutex<Option<usize>>,
}
impl CutRemote {
fn new(inner: HttpRemote) -> Self {
Self { inner, sent: Mutex::new(0), cut_from: Mutex::new(None) }
}
/// Cuts the connection from the given chunk send on.
fn cut_from(&self, call: usize) {
*self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(call);
}
/// Lets the connection through again.
fn mend(&self) {
*self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = None;
}
/// How many chunks were sent since the last count.
fn take_sent(&self) -> usize {
let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
std::mem::take(&mut *sent)
}
}
impl Remote for CutRemote {
fn changes(&self, since: u64, limit: usize, wait: u64) -> SyncResult<Changes> {
self.inner.changes(since, limit, wait)
}
fn get(&self, path: &str) -> SyncResult<Vec<u8>> {
self.inner.get(path)
}
fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
self.inner.put(path, bytes, base_seq, mtime_ms)
}
fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply> {
self.inner.delete(path, base_seq)
}
fn missing_chunks(&self, hashes: &[String]) -> SyncResult<Vec<String>> {
self.inner.missing_chunks(hashes)
}
fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> {
let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
*sent = sent.saturating_add(1);
let cut = *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
if cut.is_some_and(|from| *sent >= from) {
return Err(SyncError::Http("the connection was cut".to_owned()));
}
drop(sent);
self.inner.put_chunk(hash, bytes)
}
fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
self.inner.put_chunked(path, chunk_hashes, content_hash, base_seq, mtime_ms)
}
}
/// A device with its own vault, its own state and a remote that can cut.
struct Device {
vault: Vault,
remote: CutRemote,
state: SyncState,
name: &'static str,
}
impl Device {
fn new(base: &str, vault_id: &str, token: &str, name: &'static str) -> Self {
let root = temp(name);
let vault = Vault::open(&root).expect("vault");
Self { vault, remote: CutRemote::new(HttpRemote::new(base, vault_id, token)), state: SyncState::default(), name }
}
fn write(&self, path: &str, bytes: &[u8]) {
self.vault.write(path, bytes).expect("write");
}
fn read(&self, path: &str) -> Vec<u8> {
self.vault.read(path).expect("read")
}
/// Syncs once and returns the report.
fn sync(&mut self) -> SyncReport {
let options = SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) };
sync_once_with(&self.vault, &self.remote, &mut self.state, self.name, "2026-10-20", options).expect("sync")
}
}
/// Bytes that look random but come from a small generator, so the test is stable.
fn random_bytes(len: usize, seed: u64) -> Vec<u8> {
let mut state = seed;
(0..len)
.map(|_| {
state = state.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407);
u8::try_from((state >> 33) & 0xff).expect("byte")
})
.collect()
}
/// How many chunks the server holds for the vault.
fn stored_chunks(blobs: &Path, vault_id: &str) -> usize {
let dir = blobs.join(vault_id).join("chunks");
let Ok(entries) = std::fs::read_dir(dir) else { return 0 };
entries.filter_map(std::result::Result::ok).filter(|entry| entry.path().is_file()).count()
}
#[test]
fn a_big_file_travels_in_chunks_and_a_cut_resumes_with_the_rest() {
let data = temp("server-data");
let base = start(&data);
let blobs = data.join("blobs");
let vault_id = format!("v-chunk-{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.subsec_nanos()));
let (phone_token, user_id) = pair_raw(&base, &issue(&base, &vault_id, None), "phone");
let laptop = pair_raw(&base, &issue(&base, &vault_id, Some(&user_id)), "laptop");
let mut phone = Device::new(&base, &vault_id, &phone_token, "phone");
let mut laptop = Device::new(&base, &vault_id, &laptop.0, "laptop");
// A file that is too big for one request travels in chunks.
let bytes = random_bytes(FILE_BYTES, 42);
phone.write("big.bin", &bytes);
let report = phone.sync();
assert_eq!(report.uploaded, vec!["big.bin".to_owned()]);
let chunks = FILE_BYTES.div_ceil(CHUNK);
assert_eq!(stored_chunks(&blobs, &vault_id), chunks, "the server holds every chunk");
// The other device gets the very same bytes.
let report = laptop.sync();
assert_eq!(report.downloaded, vec!["big.bin".to_owned()]);
assert_eq!(laptop.read("big.bin"), bytes);
// The connection dies on the fourth chunk of the next file.
let other = random_bytes(FILE_BYTES, 7);
phone.write("other.bin", &other);
phone.remote.take_sent();
phone.remote.cut_from(4);
let error = sync_once_with(
&phone.vault,
&phone.remote,
&mut phone.state,
"phone",
"2026-10-20",
SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) },
)
.expect_err("cut");
assert!(matches!(error, SyncError::Http(_)), "{error:?}");
phone.remote.take_sent();
assert_eq!(stored_chunks(&blobs, &vault_id), chunks + 3, "the three chunks that arrived are kept");
// The next pass sends the chunks that never arrived and nothing else.
phone.remote.mend();
phone.sync();
assert_eq!(phone.remote.take_sent(), chunks - 3, "only the missing chunks travel again");
assert_eq!(stored_chunks(&blobs, &vault_id), chunks * 2, "the server holds every chunk of both files");
// The other device converges on the second file too.
laptop.sync();
assert_eq!(laptop.read("other.bin"), other);
assert_eq!(laptop.read("big.bin"), bytes);
}
+1
View File
@@ -199,6 +199,7 @@ pub const READ_COMMANDS: &[&str] = &[
"sync.status",
"sync.conflicts",
"server.vaults",
"server.devices",
"share.members",
"share.invites",
"settings.get",
+7 -5
View File
@@ -68,7 +68,7 @@ pub(crate) use vault::{attachments, bookmarks, buffers, content_search, daily, e
pub use crate::app_settings::{
APP_ID, AppSettings, AssistantSettings, EditorSettings, GitSettings, GrammarInfo, MAX_FONT_SIZE, MAX_INDENT_WIDTH, MAX_RECENT_VAULTS, MAX_SHORTCUT_LEN,
MAX_UI_BYTES, MIN_FONT_SIZE, MIN_INDENT_WIDTH, PluginSettings, PreviewSettings, SCHEMA_VERSION, ViewDefault,
MAX_UI_BYTES, MIN_FONT_SIZE, MIN_INDENT_WIDTH, PluginSettings, PreviewSettings, SCHEMA_VERSION, SyncSettings, ViewDefault,
};
pub use crate::assistant::{is_gated_source, run_as_assistant};
pub use crate::assistant_chat::{
@@ -113,13 +113,15 @@ pub use crate::registry::{MAX_COMMANDS, Registry, current_command_id};
pub use crate::reminders_config::RemindersConfig;
pub use crate::session::Session;
pub use crate::settings_store::{load, load_at, load_checked, load_checked_at, save, save_at};
pub use crate::share::{Accepted, Invite, InviteInfo, Member, ServerClient, ServerVault, ShareError, ShareResult, bind_folder, connect_with_code};
pub use crate::share::{
Accepted, Invite, InviteInfo, Member, ServerClient, ServerDevice, ServerVault, ShareError, ShareResult, bind_folder, connect_with_code,
};
pub use crate::slash::{SlashEntry, filter_entries, trigger_ok};
pub use crate::spec::{CommandResult, CommandSpec};
pub use crate::sync::{
Binding, Changes, ConflictPair, FileRecord, HttpRemote, Keep, Pairing, PutReply, Remote, ReportCounts, SyncError, SyncPhase, SyncReport, SyncResult,
SyncState, SyncStatus, SyncWorker, Timings, clear_binding, list_conflicts, load_binding, load_token, pair, resolve_conflict, save_binding, save_token,
sync_once,
Binding, CHUNK_SIZE, CHUNK_THRESHOLD, Changes, ConflictPair, FileRecord, HttpRemote, Keep, Limits, Pairing, Progress, PutReply, Remote, ReportCounts,
SyncError, SyncOptions, SyncPhase, SyncReport, SyncResult, SyncState, SyncStatus, SyncWorker, Timings, Transfer, clear_binding, list_conflicts,
load_binding, load_token, pair, resolve_conflict, save_binding, save_token, sync_once, sync_once_with, upload_chunked,
};
pub use crate::vault::snapshots::Snapshot;
pub use notescraft_vault::{FileCount, LARGE_FOLDER_FILES, count_files};
+33
View File
@@ -73,6 +73,8 @@ pub struct AppSettings {
/// launcher; a list, for the same reason `grammars` is: a forgotten vault
/// must not come back.
pub vault_catalog: Vec<crate::prefs::catalog::VaultMeta>,
/// The three sync options the sync sheet and the Sync settings page share.
pub sync: SyncSettings,
}
/// One installed syntax grammar.
@@ -108,6 +110,7 @@ impl Default for AppSettings {
hotkeys: BTreeMap::new(),
grammars: Vec::new(),
vault_catalog: Vec::new(),
sync: SyncSettings::default(),
}
}
}
@@ -164,6 +167,28 @@ pub fn is_grammar_id(id: &str) -> bool {
(1..=MAX_GRAMMAR_ID).contains(&id.chars().count()) && id.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
}
/// The three options a sync vault may offer (`design/mobile/03-sync-sheet.png`).
///
/// Booleans cannot hold a bad value, so there is nothing to repair: the
/// `#[serde(default)]` fields keep a missing key at its default when an older
/// `config.toml` is read.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct SyncSettings {
/// Sync only when the machine is on Wi-Fi.
pub wifi_only: bool,
/// Sync files that are not Markdown.
pub attachments: bool,
/// Sync the settings and the themes of the app between the devices.
pub settings_and_themes: bool,
}
impl Default for SyncSettings {
fn default() -> Self {
Self { wifi_only: true, attachments: true, settings_and_themes: false }
}
}
/// Drops grammar entries whose id is not a grammar id, is repeated, or whose file is not
/// `<id>.sublime-syntax`, and keeps the list sorted by id.
fn repair_grammars(grammars: &mut Vec<GrammarInfo>) -> Vec<String> {
@@ -438,6 +463,14 @@ mod tests {
assert_eq!(settings.editor.font_size, MIN_FONT_SIZE);
}
#[test]
fn sync_options_carry_their_defaults_and_an_old_file_reads() {
assert_eq!(SyncSettings::default(), SyncSettings { wifi_only: true, attachments: true, settings_and_themes: false });
let empty: AppSettings = serde_json::from_str("{}").unwrap_or_default();
assert_eq!(empty.sync, SyncSettings::default(), "a file without the section keeps the defaults");
assert!(empty.sync.wifi_only);
}
#[test]
fn session_repairs_are_prefixed_and_applied() {
let mut settings = AppSettings::default();
+1 -1
View File
@@ -187,7 +187,7 @@ fn write_key(doc: &mut Value, segments: &[String], value: Value) -> std::result:
fn write_section(section: &mut Value, name: &str, rest: &[String], value: Value) -> std::result::Result<(), CommandError> {
match name {
"session" => session_path(section, rest, value),
"git" | "editor" => match rest {
"git" | "editor" | "sync" => match rest {
[] => {
whole_section(section, value);
Ok(())
+15
View File
@@ -310,6 +310,21 @@ impl ServerClient {
self.post_json("/v1/invites/accept", &json!({ "code": code }))
}
/// The devices paired with a vault (`GET /v1/vaults/{vault}/devices`).
///
/// The rows carry no token and no token hash: the server never keeps one
/// for a listing, and this request sends none either.
///
/// # Errors
/// [`ShareError::Protocol`] when the vault id cannot travel in a URL,
/// [`ShareError::Http`] when the server cannot be reached and
/// [`ShareError::Status`] when it refuses the request.
pub fn devices(&self, vault: &str) -> ShareResult<Vec<crate::share::devices::ServerDevice>> {
refuse_id(vault)?;
let path = format!("/v1/vaults/{}/devices", encode_segment(vault));
self.get_json(&path)
}
/// The URL of a path on the server.
fn url(&self, path: &str) -> String {
format!("{}{path}", self.base)
+8
View File
@@ -37,6 +37,7 @@ const DEVICE_NAME: &str = "NotesCraft";
pub(crate) fn register_share(reg: &mut Registry<Session>) -> crate::error::Result<()> {
let specs = vec![
CommandSpec::new("server.vaults", "Server vaults", "Lists the vaults of the server account this vault is bound to", |_| true, server_vaults),
CommandSpec::new("server.devices", "Vault devices", "Lists the devices paired with the bound vault; never a token", |_| true, server_devices),
CommandSpec::new(
"server.create_vault",
"Create server vault",
@@ -119,6 +120,13 @@ fn server_vaults(session: &mut Session, _params: &Value) -> CommandResult {
crate::commands::json(json!({ "vaults": vaults }))
}
/// `server.devices`: the devices of the bound vault; never a token.
fn server_devices(session: &mut Session, _params: &Value) -> CommandResult {
let (client, binding) = client_of(session)?;
let devices = client.devices(&binding.vault_id).map_err(share_error)?;
crate::commands::json(json!({ "vault_id": binding.vault_id, "devices": devices }))
}
/// `server.create_vault`: a new vault, owned by this account.
fn server_create_vault(session: &mut Session, params: &Value) -> CommandResult {
let name = name_param(params, "name")?;
+42
View File
@@ -0,0 +1,42 @@
//! The devices of a bound vault, as the server lists them.
//!
//! The server names `id`, `name`, `created_ms`, `last_seen_ms` and `current`
//! (design `03-sync-sheet`); a token or its hash is never part of the
//! request, the answer or a row.
use serde::{Deserialize, Serialize};
/// One device paired with the bound vault, as `GET /v1/vaults/{vault}/devices`
/// answers it.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServerDevice {
/// The device's id.
#[serde(default)]
pub id: String,
/// The name the device paired under.
#[serde(default)]
pub name: String,
/// When the device was paired, in Unix milliseconds.
#[serde(default)]
pub created_ms: u64,
/// When the server saw the device last, if ever.
#[serde(default)]
pub last_seen_ms: Option<u64>,
/// True for the device that asked; the server decides.
#[serde(default)]
pub current: bool,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_row_of_the_server_reads_whole_or_defaults() {
let row = serde_json::from_str::<ServerDevice>(r#"{"id":"d-1","name":"Pixel 9","current":true}"#).unwrap_or_default();
assert_eq!(row, ServerDevice { id: "d-1".to_owned(), name: "Pixel 9".to_owned(), created_ms: 0, last_seen_ms: None, current: true });
let full =
serde_json::from_str::<ServerDevice>(r#"{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}"#).unwrap_or_default();
assert_eq!(full.last_seen_ms, Some(60));
}
}
+2
View File
@@ -15,9 +15,11 @@
pub(crate) mod bind;
pub(crate) mod client;
pub(crate) mod commands;
mod devices;
pub use bind::{bind_folder, connect_with_code};
pub use client::{Accepted, Invite, InviteInfo, Member, ServerClient, ServerVault};
pub use devices::ServerDevice;
use serde::Deserialize;
use thiserror::Error as ThisError;
+223
View File
@@ -0,0 +1,223 @@
//! Chunked uploads: a big file travels to the server in content-addressed chunks.
//!
//! The server keeps every chunk by its own hash, so a file that was cut off half
//! way resumes with the chunks that never arrived: the client asks which ones
//! the server is missing and sends only those. One chunk at a time is read from
//! disk, so the memory a big file needs does not grow with the file.
//!
//! The client hashes the file once to learn the hash of every chunk and of the
//! whole content, asks the server which chunks it is missing, and reads the file
//! a second time to send the missing ones. The second pass checks every chunk
//! against the hash of the first pass, so a file that changed in between is
//! refused instead of being uploaded in pieces.
use std::collections::HashSet;
use std::fs::File;
use std::io::Read;
use std::path::Path;
use std::time::Duration;
use notescraft_sync::{hash_hex, hasher};
use notescraft_vault::Vault;
use crate::sync::{PutReply, Remote, SyncError, SyncResult};
/// Files above this size upload in chunks.
pub const CHUNK_THRESHOLD: u64 = 8 * 1024 * 1024;
/// How many bytes one chunk holds.
pub const CHUNK_SIZE: usize = 4 * 1024 * 1024;
/// The most hashes one "which chunks are missing" question holds.
const MAX_CHECK_BATCH: usize = 5000;
/// How many times a chunk is sent again before the upload is given up.
const CHUNK_RETRIES: usize = 3;
/// The waits between the tries of a chunk, in milliseconds.
const BACKOFF_MS: [u64; CHUNK_RETRIES] = [200, 400, 800];
/// How far a big file has moved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Progress {
/// The vault path of the file.
pub path: String,
/// The bytes of the file the server holds now.
pub done_bytes: u64,
/// The bytes the file has.
pub total_bytes: u64,
}
/// When files upload in chunks and how big a chunk is.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Limits {
/// Files above this size upload in chunks.
pub threshold: u64,
/// How many bytes one chunk holds.
pub chunk: usize,
}
impl Default for Limits {
fn default() -> Self {
Self { threshold: CHUNK_THRESHOLD, chunk: CHUNK_SIZE }
}
}
/// Uploads `path` to the server in chunks.
///
/// The file is read from `vault` twice: once to learn the hash of every chunk
/// and of the whole content, once to send the chunks the server is missing.
/// `progress` is called after every chunk, with the bytes the server holds.
///
/// The answer is the one of a plain write: [`PutReply::Applied`] when the
/// server took the file, [`PutReply::Unchanged`] when it already held it and
/// [`PutReply::Conflict`] when somebody else wrote first.
///
/// # Errors
/// [`SyncError::Vault`] when the file cannot be read, [`SyncError::BadRemote`]
/// when the file changed while it was uploaded, and whatever the transport
/// reports for the chunks and the write itself.
pub fn upload_chunked(
remote: &dyn Remote,
vault: &Vault,
path: &str,
base_seq: u64,
mtime_ms: u64,
limits: Limits,
progress: &mut dyn FnMut(&Progress),
) -> SyncResult<PutReply> {
if limits.chunk == 0 || limits.chunk > CHUNK_SIZE {
return Err(SyncError::Io(format!("a chunk must hold between one byte and {CHUNK_SIZE} bytes")));
}
let file = vault.resolve(path).map_err(|error| SyncError::Vault(error.to_string()))?;
let plan = plan_chunks(&file, limits.chunk)?;
if plan.chunks.is_empty() {
// An empty file has no chunk: it goes as a plain write.
return remote.put(path, &[], base_seq, mtime_ms);
}
let missing = missing_set(remote, &plan.chunks)?;
send_chunks(remote, &file, &plan, &missing, path, limits, progress)?;
match remote.put_chunked(path, &plan.chunks, &plan.content_hash, base_seq, mtime_ms) {
Ok(reply) => Ok(reply),
Err(SyncError::MissingChunks(list)) => {
// The server lost a chunk on its way: send those again, try once more.
let gone: HashSet<String> = list.into_iter().collect();
send_chunks(remote, &file, &plan, &gone, path, limits, progress)?;
remote.put_chunked(path, &plan.chunks, &plan.content_hash, base_seq, mtime_ms)
}
Err(error) => Err(error),
}
}
/// What the hash pass learned about a file.
struct Plan {
/// The hash of every chunk, in the order they make up the file.
chunks: Vec<String>,
/// The hash of the whole content.
content_hash: String,
/// How many bytes the file has.
total: u64,
}
/// Reads the file once and answers the hash of every chunk and of the whole
/// content.
fn plan_chunks(file: &Path, chunk: usize) -> SyncResult<Plan> {
let mut handle = open(file)?;
let mut content = hasher();
let mut chunks = Vec::new();
let mut total = 0u64;
let mut buffer = vec![0u8; chunk];
loop {
let read = read_upto(&mut handle, &mut buffer)?;
if read == 0 {
break;
}
let bytes = buffer.get(..read).ok_or_else(|| SyncError::Io("the chunk buffer is shorter than the bytes read".to_owned()))?;
content.update(bytes);
chunks.push(hash_hex(bytes));
total = total
.checked_add(u64::try_from(read).map_err(|_| SyncError::Io("a chunk holds too many bytes".to_owned()))?)
.ok_or_else(|| SyncError::Io("the file is too long to count".to_owned()))?;
}
Ok(Plan { chunks, content_hash: content.finish_hex(), total })
}
/// The chunks of the list the server does not hold yet.
fn missing_set(remote: &dyn Remote, chunks: &[String]) -> SyncResult<HashSet<String>> {
let mut missing = HashSet::new();
for batch in chunks.chunks(MAX_CHECK_BATCH) {
for hash in remote.missing_chunks(batch)? {
missing.insert(hash);
}
}
Ok(missing)
}
/// Sends the chunks of `wanted` the server is missing and reports the progress.
///
/// The file is read again: every chunk is checked against the hash the first
/// pass gave it, so a file that changed in between is refused before any of it
/// is sent.
fn send_chunks(
remote: &dyn Remote,
file: &Path,
plan: &Plan,
wanted: &HashSet<String>,
path: &str,
limits: Limits,
progress: &mut dyn FnMut(&Progress),
) -> SyncResult<()> {
let mut handle = open(file)?;
let mut buffer = vec![0u8; limits.chunk];
let mut sent: HashSet<String> = HashSet::new();
let mut done = 0u64;
for expected in &plan.chunks {
let read = read_upto(&mut handle, &mut buffer)?;
if read == 0 {
return Err(SyncError::BadRemote(format!("the file `{path}` shrank while it was uploaded")));
}
let bytes = buffer.get(..read).ok_or_else(|| SyncError::Io("the chunk buffer is shorter than the bytes read".to_owned()))?;
let hash = hash_hex(bytes);
if hash != *expected {
return Err(SyncError::BadRemote(format!("the file `{path}` changed while it was uploaded")));
}
if wanted.contains(&hash) && !sent.contains(&hash) {
put_with_retry(remote, &hash, bytes)?;
sent.insert(hash.clone());
}
done = done.saturating_add(u64::try_from(read).map_err(|_| SyncError::Io("a chunk holds too many bytes".to_owned()))?);
progress(&Progress { path: path.to_owned(), done_bytes: done, total_bytes: plan.total });
}
Ok(())
}
/// Sends one chunk, trying again with a growing wait before giving up.
fn put_with_retry(remote: &dyn Remote, hash: &str, bytes: &[u8]) -> SyncResult<()> {
let mut attempt = 0usize;
loop {
match remote.put_chunk(hash, bytes) {
Ok(()) => return Ok(()),
Err(error) => {
let Some(wait) = BACKOFF_MS.get(attempt).copied() else { return Err(error) };
std::thread::sleep(Duration::from_millis(wait));
attempt = attempt.saturating_add(1);
}
}
}
}
/// Opens a file of the vault.
fn open(file: &Path) -> SyncResult<File> {
File::open(file).map_err(|error| SyncError::Io(format!("{}: {error}", file.display())))
}
/// Reads into `buffer` until it is full or the file ends; at most a full buffer.
fn read_upto(file: &mut File, buffer: &mut [u8]) -> SyncResult<usize> {
let mut read = 0usize;
while read < buffer.len() {
let Some(rest) = buffer.get_mut(read..) else { break };
match file.read(rest) {
Ok(0) => break,
Ok(count) => read = read.saturating_add(count),
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
Err(error) => return Err(SyncError::Io(error.to_string())),
}
}
Ok(read)
}
+100 -15
View File
@@ -5,6 +5,7 @@ use std::collections::BTreeMap;
use notescraft_sync::{Action, Base, FileRecord, check_hash, check_syncable, hash_hex, plan};
use notescraft_vault::Vault;
use crate::sync::chunks::{self, Limits, Progress};
use crate::sync::{PutReply, Remote, SyncError, SyncResult, SyncState, reconcile, scan, state};
/// How many records one `/changes` page asks for.
@@ -13,6 +14,9 @@ const PAGE_LIMIT: usize = 1000;
const MAX_PAGES: usize = 10_000;
/// Most names a conflict copy gets before the first one is used again.
const MAX_COPY_TRIES: usize = 10;
/// The largest file that is fetched into memory; bigger ones are left for a
/// later day instead of being pulled into the memory of the app.
pub(crate) const MAX_DOWNLOAD_IN_MEMORY: u64 = 64 * 1024 * 1024;
/// What one sync pass did.
#[derive(Debug, Default, PartialEq, Eq)]
@@ -29,10 +33,26 @@ pub struct SyncReport {
pub merged: Vec<String>,
/// Paths of the conflict copies written.
pub conflicts: Vec<String>,
/// Paths left alone because the client cannot fetch them yet.
pub skipped: Vec<String>,
/// How many paths were already in step.
pub adopted: usize,
}
/// How one sync pass behaves.
pub struct SyncOptions {
/// When files upload in chunks and how big a chunk is.
pub limits: Limits,
/// Called after every chunk of a big file, with the bytes that moved.
pub progress: Box<dyn FnMut(&Progress)>,
}
impl Default for SyncOptions {
fn default() -> Self {
Self { limits: Limits::default(), progress: Box::new(|_: &Progress| {}) }
}
}
/// Syncs `vault` with `remote`, once.
///
/// The pass reads the vault, reads the change log of the server, asks
@@ -51,12 +71,40 @@ pub struct SyncReport {
/// [`SyncError`] when the server cannot be reached, when it sends something
/// unusable, or when the vault refuses a read or a write.
pub fn sync_once(vault: &Vault, remote: &dyn Remote, state: &mut SyncState, device: &str, stamp: &str) -> SyncResult<SyncReport> {
sync_once_with(vault, remote, state, device, stamp, SyncOptions::default())
}
/// [`sync_once`] with the options of the pass: the chunk limits and the
/// progress callback that watches a big file travel.
///
/// # Errors
/// [`SyncError`] when the server cannot be reached, when it sends something
/// unusable, or when the vault refuses a read or a write.
pub fn sync_once_with(vault: &Vault, remote: &dyn Remote, state: &mut SyncState, device: &str, stamp: &str, options: SyncOptions) -> SyncResult<SyncReport> {
sync_once_scanned(vault, remote, state, device, stamp, options, scan::ScanOptions::default())
}
/// [`sync_once_with`] also naming what the scan of the vault leaves out: with
/// `attachments: false` the pass sees only the Markdown notes.
///
/// # Errors
/// [`SyncError`] when the server cannot be reached, when it sends something
/// unusable, or when the vault refuses a read or a write.
pub(crate) fn sync_once_scanned(
vault: &Vault,
remote: &dyn Remote,
state: &mut SyncState,
device: &str,
stamp: &str,
options: SyncOptions,
scan_options: scan::ScanOptions,
) -> SyncResult<SyncReport> {
let mut report = SyncReport::default();
let local = scan::scan(vault)?;
let local = scan::scan(vault, scan_options)?;
let hashes: BTreeMap<String, String> = local.iter().map(|(path, file)| (path.clone(), file.hash.clone())).collect();
let (records, cursor) = fetch(remote, state.last_seq)?;
let actions = plan(&hashes, &state.bases, &records);
let mut ctx = Ctx { vault, remote, state, report: &mut report, local: &local, device, stamp };
let mut ctx = Ctx { vault, remote, state, report: &mut report, local: &local, device, stamp, options };
for action in actions {
perform(&mut ctx, action)?;
ctx.state.save(ctx.vault)?;
@@ -76,12 +124,19 @@ struct Ctx<'a> {
local: &'a BTreeMap<String, scan::LocalFile>,
device: &'a str,
stamp: &'a str,
options: SyncOptions,
}
/// Performs one planned action.
fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> {
match action {
Action::Download(record) => {
if record.size > MAX_DOWNLOAD_IN_MEMORY {
// Too big to fetch here yet: the file stays as it is, without a
// conflict and without a half-written copy.
ctx.report.skipped.push(record.path.clone());
return Ok(());
}
let bytes = fetch_blob(ctx.remote, &record)?;
ctx.vault.write(&record.path, &bytes).map_err(|e| vault_error(&e))?;
ctx.state.bases.insert(record.path.clone(), Base { hash: record.hash.clone(), seq: record.seq });
@@ -94,19 +149,11 @@ fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> {
ctx.report.deleted_local.push(record.path.clone());
}
Action::Upload { path, base_seq } => {
let bytes = ctx.vault.read(&path).map_err(|e| vault_error(&e))?;
let hash = hash_hex(&bytes);
let mtime_ms = ctx.local.get(&path).map_or(0, |file| file.mtime_ms);
match ctx.remote.put(&path, &bytes, base_seq, mtime_ms)? {
PutReply::Applied(reply) | PutReply::Unchanged(reply) => {
ctx.state.bases.insert(path.clone(), Base { hash, seq: reply.seq });
ctx.report.uploaded.push(path.clone());
state::store_base_copy(ctx.vault, &path, &bytes)?;
}
PutReply::Conflict(current) => {
let base_hash = ctx.state.bases.get(&path).map(|base| base.hash.clone());
reconcile_path(ctx, &current, &bytes, base_hash.as_deref())?;
}
let (size, mtime_ms) = ctx.local.get(&path).map_or((0, 0), |file| (file.size, file.mtime_ms));
if size > ctx.options.limits.threshold {
upload_big(ctx, &path, base_seq, mtime_ms)?;
} else {
upload_small(ctx, &path, base_seq, mtime_ms)?;
}
}
Action::DeleteRemote { path, base_seq } => match ctx.remote.delete(&path, base_seq)? {
@@ -143,6 +190,44 @@ fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> {
Ok(())
}
/// Uploads a small file in one piece.
fn upload_small(ctx: &mut Ctx<'_>, path: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<()> {
let bytes = ctx.vault.read(path).map_err(|e| vault_error(&e))?;
let hash = hash_hex(&bytes);
match ctx.remote.put(path, &bytes, base_seq, mtime_ms)? {
PutReply::Applied(reply) | PutReply::Unchanged(reply) => {
ctx.state.bases.insert(path.to_owned(), Base { hash, seq: reply.seq });
ctx.report.uploaded.push(path.to_owned());
state::store_base_copy(ctx.vault, path, &bytes)?;
Ok(())
}
PutReply::Conflict(current) => {
let base_hash = ctx.state.bases.get(path).map(|base| base.hash.clone());
reconcile_path(ctx, &current, &bytes, base_hash.as_deref())
}
}
}
/// Uploads a big file in chunks, so a cut connection only loses one chunk.
fn upload_big(ctx: &mut Ctx<'_>, path: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<()> {
let limits = ctx.options.limits;
let progress: &mut dyn FnMut(&Progress) = &mut ctx.options.progress;
let reply = chunks::upload_chunked(ctx.remote, ctx.vault, path, base_seq, mtime_ms, limits, progress)?;
match reply {
PutReply::Applied(record) | PutReply::Unchanged(record) => {
ctx.state.bases.insert(path.to_owned(), Base { hash: record.hash.clone(), seq: record.seq });
ctx.report.uploaded.push(path.to_owned());
Ok(())
}
PutReply::Conflict(current) => {
// The other side wrote first: the content comes back for the merge.
let bytes = ctx.vault.read(path).map_err(|e| vault_error(&e))?;
let base_hash = ctx.state.bases.get(path).map(|base| base.hash.clone());
reconcile_path(ctx, &current, &bytes, base_hash.as_deref())
}
}
}
/// Settles a file both sides changed.
fn reconcile_path(ctx: &mut Ctx<'_>, record: &FileRecord, ours: &[u8], base_hash: Option<&str>) -> SyncResult<()> {
let theirs = fetch_blob(ctx.remote, record)?;
+63 -2
View File
@@ -8,14 +8,15 @@ use std::time::Duration;
use notescraft_sync::{Changes, FileRecord};
use serde::{Deserialize, Serialize};
use crate::sync::driver::MAX_DOWNLOAD_IN_MEMORY;
use crate::sync::{PutReply, Remote, SyncError, SyncResult};
/// Seconds a request may take when the caller does not ask for a long poll.
const DEFAULT_TIMEOUT_SECS: u64 = 60;
/// Seconds added to the long poll, to cover the round trip.
const POLL_MARGIN_SECS: u64 = 15;
/// Largest body the client reads.
const MAX_BODY_BYTES: u64 = 25 * 1024 * 1024;
/// Largest body the client reads: the biggest file it fetches.
const MAX_BODY_BYTES: u64 = MAX_DOWNLOAD_IN_MEMORY;
/// Longest body kept in an error message.
const MAX_ERROR_BODY: usize = 200;
/// The hexadecimal digits of a percent escape.
@@ -121,6 +122,53 @@ impl Remote for HttpRemote {
other => Err(status_error(other, &body)),
}
}
fn missing_chunks(&self, hashes: &[String]) -> SyncResult<Vec<String>> {
let url = format!("{}/chunks/check", self.vault_url());
let body = serde_json::json!({ "hashes": hashes });
let response = self.agent.post(url.as_str()).header("Authorization", self.auth()).send_json(body).map_err(|e| SyncError::Http(e.to_string()))?;
let (status, body) = read(response)?;
if status != 200 {
return Err(status_error(status, &body));
}
let missing: MissingChunks = serde_json::from_slice(&body).map_err(|e| SyncError::BadRemote(format!("the chunk list is not readable: {e}")))?;
Ok(missing.missing)
}
fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> {
let url = format!("{}/chunks/{}", self.vault_url(), encode_segment(hash));
let response = self.agent.put(url.as_str()).header("Authorization", self.auth()).send(bytes).map_err(|e| SyncError::Http(e.to_string()))?;
let (status, body) = read(response)?;
// 201 when the chunk was stored, 200 when the server already had it.
if status == 200 || status == 201 {
return Ok(());
}
Err(status_error(status, &body))
}
fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
let url = self.file_url(path);
let response = self
.agent
.put(url.as_str())
.header("Authorization", self.auth())
.header("X-Chunks", chunk_hashes.join(","))
.header("X-Content-Hash", content_hash)
.header("X-Base-Seq", base_seq.to_string())
.header("X-Mtime-Ms", mtime_ms.to_string())
.send_empty()
.map_err(|e| SyncError::Http(e.to_string()))?;
let (status, body) = read(response)?;
match status {
201 => Ok(PutReply::Applied(parse_record(&body)?)),
200 => Ok(PutReply::Unchanged(parse_record(&body)?)),
409 => match parse_missing(&body) {
Some(missing) => Err(SyncError::MissingChunks(missing)),
None => Ok(PutReply::Conflict(parse_current(&body)?)),
},
other => Err(status_error(other, &body)),
}
}
}
/// The device a pairing code buys.
@@ -181,6 +229,19 @@ fn parse_current(body: &[u8]) -> SyncResult<FileRecord> {
Ok(current.current)
}
/// Reads the `{"missing": [...]}` body of a chunked write whose chunks are not
/// all there; `None` when the body is a conflict about the revision instead.
fn parse_missing(body: &[u8]) -> Option<Vec<String>> {
serde_json::from_slice::<MissingChunks>(body).ok().map(|missing| missing.missing)
}
/// The `{"missing": [...]}` answer of the chunk check.
#[derive(Deserialize)]
struct MissingChunks {
/// The hashes the server does not hold.
missing: Vec<String>,
}
/// Turns an unexpected status into an error.
fn status_error(status: u16, body: &[u8]) -> SyncError {
let text: String = String::from_utf8_lossy(body).chars().take(MAX_ERROR_BODY).collect();
+11 -2
View File
@@ -12,6 +12,7 @@
//! file cannot be merged, both are kept.
mod binding;
mod chunks;
pub(crate) mod commands;
mod conflicts;
mod driver;
@@ -23,18 +24,22 @@ mod state;
mod worker;
pub use binding::{Binding, clear_binding, load_binding, load_token, save_binding, save_token};
pub use chunks::{CHUNK_SIZE, CHUNK_THRESHOLD, Limits, Progress, upload_chunked};
pub use conflicts::{ConflictPair, Keep, list_conflicts, resolve as resolve_conflict};
pub use driver::{SyncReport, sync_once};
pub use driver::{SyncOptions, SyncReport, sync_once, sync_once_with};
pub use http::{HttpRemote, Pairing, pair};
pub use remote::{PutReply, Remote};
pub use state::SyncState;
pub use worker::{ReportCounts, SyncPhase, SyncStatus, SyncWorker, Timings};
pub use worker::{ReportCounts, SyncPhase, SyncStatus, SyncWorker, Timings, Transfer};
/// The HTTP helpers the share client shares with the sync transport.
pub(crate) use http::{agent, encode_segment};
pub use notescraft_sync::{Changes, FileRecord};
/// The scan options of the worker's pass; the session sets them from the user's options.
pub(crate) use scan::ScanOptions;
use thiserror::Error as ThisError;
/// Everything that can go wrong while syncing.
@@ -59,6 +64,10 @@ pub enum SyncError {
/// vault, a broken hash or bytes that do not match the hash they came with.
#[error("the sync server sent something unusable: {0}")]
BadRemote(String),
/// The server does not hold every chunk of a file: the hashes it is
/// missing come with the answer, so the client can send them again.
#[error("the sync server is missing {} chunk(s) of the file", .0.len())]
MissingChunks(Vec<String>),
/// The sync state could not be read or written.
#[error("sync state: {0}")]
State(String),
+38 -1
View File
@@ -2,7 +2,7 @@
use notescraft_sync::{Changes, FileRecord};
use crate::sync::SyncResult;
use crate::sync::{SyncError, SyncResult};
/// What the server did with a write.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -51,4 +51,41 @@ pub trait Remote {
/// [`SyncError::Http`] when the server cannot be reached and
/// [`SyncError::Status`] when it refuses the delete.
fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply>;
/// The chunks of `hashes` the server does not hold yet.
///
/// The answer holds only the hashes that are missing, so a client that was
/// cut off sends the rest and nothing else.
///
/// # Errors
/// [`SyncError::Http`] when the server cannot be reached and
/// [`SyncError::BadRemote`] when the server does not know chunks.
fn missing_chunks(&self, _hashes: &[String]) -> SyncResult<Vec<String>> {
Err(SyncError::BadRemote("chunks unsupported".to_owned()))
}
/// Stores one chunk under its own hash.
///
/// The server keeps the chunk by content, so sending it twice costs
/// nothing.
///
/// # Errors
/// [`SyncError::Http`] when the server cannot be reached and
/// [`SyncError::BadRemote`] when the server does not know chunks.
fn put_chunk(&self, _hash: &str, _bytes: &[u8]) -> SyncResult<()> {
Err(SyncError::BadRemote("chunks unsupported".to_owned()))
}
/// Writes a file whose content the server already holds as chunks.
///
/// `chunk_hashes` are the hashes of the chunks in the order they make up
/// the file, `content_hash` the hash of the whole content.
///
/// # Errors
/// [`SyncError::Http`] when the server cannot be reached,
/// [`SyncError::MissingChunks`] when the server is missing a chunk of the
/// list, and [`SyncError::BadRemote`] when the server does not know chunks.
fn put_chunked(&self, _path: &str, _chunk_hashes: &[String], _content_hash: &str, _base_seq: u64, _mtime_ms: u64) -> SyncResult<PutReply> {
Err(SyncError::BadRemote("chunks unsupported".to_owned()))
}
}
+87 -9
View File
@@ -1,14 +1,18 @@
//! The disk side of a sync: the files the vault holds right now.
use std::collections::BTreeMap;
use std::fs::File;
use std::io::Read;
use notescraft_sync::{check_syncable, hash_hex};
use notescraft_sync::{check_syncable, hasher};
use notescraft_vault::Vault;
use crate::sync::{SyncError, SyncResult};
/// Files above this size are left alone; the server refuses them too.
const MAX_FILE_BYTES: u64 = 25 * 1024 * 1024;
const MAX_FILE_BYTES: u64 = 1024 * 1024 * 1024;
/// How many bytes are read at a time while a file is hashed.
const READ_BUFFER: usize = 64 * 1024;
/// One file of the vault.
pub(crate) struct LocalFile {
@@ -16,6 +20,22 @@ pub(crate) struct LocalFile {
pub hash: String,
/// Last modification in Unix milliseconds.
pub mtime_ms: u64,
/// How many bytes the file has.
pub size: u64,
}
/// What a scan leaves out of the vault picture.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ScanOptions {
/// Whether files that are not Markdown travel; the sync sheet calls them
/// the attachments (`SyncSettings.attachments`).
pub attachments: bool,
}
impl Default for ScanOptions {
fn default() -> Self {
Self { attachments: true }
}
}
/// Reads every syncable file of `vault` and hashes it.
@@ -24,21 +44,79 @@ pub(crate) struct LocalFile {
/// `.git`) are left out, and so are the ones above [`MAX_FILE_BYTES`]. A file
/// that vanishes while the vault is being read is left out too.
///
/// The content of a file is never held in memory: it is hashed as it is read,
/// so a file of any size costs the same memory.
///
/// An `attachments` of `false` leaves out every file that is not Markdown.
///
/// # Errors
/// [`SyncError::Vault`] when a file cannot be read.
pub(crate) fn scan(vault: &Vault) -> SyncResult<BTreeMap<String, LocalFile>> {
pub(crate) fn scan(vault: &Vault, options: ScanOptions) -> SyncResult<BTreeMap<String, LocalFile>> {
let entries = vault.list().map_err(|e| SyncError::Vault(e.to_string()))?;
let mut files = BTreeMap::new();
for entry in entries {
if entry.is_dir || entry.size > MAX_FILE_BYTES || check_syncable(&entry.path).is_err() {
if entry.is_dir || entry.size > MAX_FILE_BYTES || check_syncable(&entry.path).is_err() || (!is_markdown(&entry.path) && !options.attachments) {
continue;
}
let bytes = match vault.read(&entry.path) {
Ok(bytes) => bytes,
Err(notescraft_vault::Error::Io { source, .. }) if source.kind() == std::io::ErrorKind::NotFound => continue,
Err(error) => return Err(SyncError::Vault(error.to_string())),
let full = vault.resolve(&entry.path).map_err(|e| SyncError::Vault(e.to_string()))?;
let hash = match hash_file(&full) {
Ok(hash) => hash,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
Err(error) => return Err(SyncError::Io(format!("{}: {error}", entry.path))),
};
files.insert(entry.path.clone(), LocalFile { hash: hash_hex(&bytes), mtime_ms: entry.modified.unwrap_or(0) });
files.insert(entry.path.clone(), LocalFile { hash, mtime_ms: entry.modified.unwrap_or(0), size: entry.size });
}
Ok(files)
}
/// Whether the path ends in `.md`, read case-insensitively; a file without an
/// extension is not Markdown.
fn is_markdown(path: &str) -> bool {
std::path::Path::new(path).extension().is_some_and(|extension| extension.eq_ignore_ascii_case("md"))
}
/// The SHA-256 hex of the content of a file, read in pieces.
fn hash_file(file: &std::path::Path) -> std::result::Result<String, std::io::Error> {
let mut handle = File::open(file)?;
let mut hash = hasher();
let mut buffer = vec![0u8; READ_BUFFER];
loop {
let read = handle.read(&mut buffer)?;
if read == 0 {
break;
}
let bytes = buffer.get(..read).ok_or_else(|| std::io::Error::other("the read buffer is shorter than the bytes read"))?;
hash.update(bytes);
}
Ok(hash.finish_hex())
}
#[cfg(test)]
mod tests {
use super::*;
use notescraft_vault::Vault;
/// A vault with one note and one attachment, under a fresh temp folder.
fn seeded(name: &str) -> Vault {
let root = std::env::temp_dir().join(format!("notescraft-scan-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).expect("create the temp vault");
let vault = Vault::open(&root).expect("open the temp vault");
vault.write("notes/idea.md", b"text").expect("write the note");
vault.write("notes/diagram.png", b"bytes").expect("write the attachment");
vault
}
#[test]
fn attachments_on_or_off_changes_what_the_scan_sees() {
let vault = seeded("on-off");
let note_only = scan(&vault, ScanOptions { attachments: false }).expect("scan without attachments");
assert!(!note_only.contains_key("notes/diagram.png"), "the png is left out");
assert_eq!(note_only.len(), 1, "the note stays");
let with_all = scan(&vault, ScanOptions { attachments: true }).expect("scan with attachments");
assert_eq!(with_all.len(), 2, "the note and the png both travel");
let defaulted = scan(&vault, ScanOptions::default()).expect("scan with the defaults");
assert_eq!(defaulted.len(), 2, "the default scan keeps the attachments");
}
}
+85 -27
View File
@@ -21,7 +21,8 @@ use notescraft_templates::Date;
use notescraft_vault::Vault;
use crate::sync::binding::{load_binding, load_token};
use crate::sync::driver::{SyncReport, sync_once};
use crate::sync::chunks::{Limits, Progress};
use crate::sync::driver::{SyncOptions, SyncReport, sync_once_scanned};
use crate::sync::http::HttpRemote;
use crate::sync::remote::Remote;
use crate::sync::scan;
@@ -106,6 +107,17 @@ pub struct ReportCounts {
pub conflicts: usize,
}
/// The file that is moving right now.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct Transfer {
/// The vault path of the file.
pub path: String,
/// The bytes of the file the server holds now.
pub done_bytes: u64,
/// The bytes the file has.
pub total_bytes: u64,
}
/// What a worker knows: a cheap clone of its shared status.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SyncStatus {
@@ -130,13 +142,26 @@ pub struct SyncStatus {
pub last_report: Option<ReportCounts>,
/// The conflict copies the last cycle wrote.
pub conflicts: Vec<String>,
/// The big file that is moving right now, while one does.
pub transfer: Option<Transfer>,
}
/// What the worker thread and the readers of its status share.
#[derive(Debug)]
struct Shared {
/// The status every reader sees.
status: Mutex<SyncStatus>,
/// The waits the loop uses.
timings: RwLock<Timings>,
/// What the scan of every cycle leaves out (the sync options); the app pushes
/// the user's toggles here.
scan_options: RwLock<scan::ScanOptions>,
}
/// A background thread syncing one vault.
#[derive(Debug)]
pub struct SyncWorker {
status: Arc<Mutex<SyncStatus>>,
timings: Arc<RwLock<Timings>>,
shared: Arc<Shared>,
thread: Option<std::thread::JoinHandle<()>>,
tx: Option<Sender<()>>,
}
@@ -153,7 +178,7 @@ impl SyncWorker {
let (server, vault_id) = (binding.server, binding.vault_id);
let worker = Self::start_with(vault, device, Box::new(HttpRemote::new(&server, &vault_id, &token)), clock_millis);
{
let mut status = lock_status(&worker.status);
let mut status = lock_status(&worker.shared.status);
status.server = Some(server);
status.vault_id = Some(vault_id);
}
@@ -164,18 +189,21 @@ impl SyncWorker {
#[must_use]
pub fn start_with(vault: &Vault, device: &str, remote: Box<dyn Remote + Send>, clock: fn() -> u64) -> Self {
let (tx, rx) = channel();
let status = Arc::new(Mutex::new(SyncStatus { bound: true, ..SyncStatus::default() }));
let timings = Arc::new(RwLock::new(Timings::default()));
let shared = Arc::new(Shared {
status: Mutex::new(SyncStatus { bound: true, ..SyncStatus::default() }),
timings: RwLock::new(Timings::default()),
scan_options: RwLock::new(scan::ScanOptions::default()),
});
let vault = vault.sibling();
let device = device.to_owned();
let (thread_status, thread_timings) = (Arc::clone(&status), Arc::clone(&timings));
let thread_shared = Arc::clone(&shared);
let thread = std::thread::Builder::new()
.name("notescraft-sync".to_owned())
.spawn(move || run(&vault, &device, remote.as_ref(), clock, &thread_status, &thread_timings, &rx))
.spawn(move || run(&vault, &device, remote.as_ref(), clock, &thread_shared, &rx))
.ok();
let worker = Self { status, timings, thread, tx: Some(tx) };
let worker = Self { shared, thread, tx: Some(tx) };
if worker.thread.is_none() {
set_status(&worker.status, |status| {
set_status(&worker.shared.status, |status| {
status.phase = SyncPhase::Error;
status.last_error = Some("the sync thread could not start".to_owned());
});
@@ -185,11 +213,19 @@ impl SyncWorker {
/// Replaces the waits; they take effect on the next loop wake-up. Tests shorten them.
pub fn set_timings(&self, timings: Timings) {
if let Ok(mut guard) = self.timings.write() {
if let Ok(mut guard) = self.shared.timings.write() {
*guard = timings;
}
}
/// Replaces what the scan of the cycles leaves out, the way the engine
/// applies the user's sync options; it takes effect on the next cycle.
pub(crate) fn set_scan_options(&self, options: scan::ScanOptions) {
if let Ok(mut guard) = self.shared.scan_options.write() {
*guard = options;
}
}
/// Wakes the worker up. Requests of the same run coalesce: the thread waits for the
/// debounce after the last one.
pub fn request_sync(&self) {
@@ -201,7 +237,7 @@ impl SyncWorker {
/// The status as it stands.
#[must_use]
pub fn status(&self) -> SyncStatus {
lock_status(&self.status).clone()
lock_status(&self.shared.status).clone()
}
/// Signals and joins the thread, giving it [`STOP_CAP`]; the same work [`Drop`] does.
@@ -241,7 +277,7 @@ enum Fate {
/// The loop of the worker thread. It owns a vault sibling and a receiver; everything else
/// it borrows.
fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex<SyncStatus>, timings: &RwLock<Timings>, rx: &Receiver<()>) {
fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc<Shared>, rx: &Receiver<()>) {
// The start counts as one request, so the first cycle runs right after the debounce.
let mut last_request = Some(Instant::now());
let mut last_cycle: Option<Instant> = None;
@@ -250,7 +286,7 @@ fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, sta
let mut wait_for_request = false;
let mut cycles: u64 = 0;
loop {
let Some(due) = next_due(timings, last_request, last_cycle, failed_at, retries, wait_for_request) else {
let Some(due) = next_due(&shared.timings, last_request, last_cycle, failed_at, retries, wait_for_request) else {
// Sign-in-again: nothing times out, only a request or the end of the worker.
match rx.recv() {
Ok(()) => {
@@ -278,7 +314,7 @@ fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, sta
last_request = None;
last_cycle = Some(now);
cycles = cycles.wrapping_add(1);
let fate = run_cycle(vault, device, remote, clock, status, cycles);
let fate = run_cycle(vault, device, remote, clock, shared, cycles);
match fate {
Fate::Success | Fate::Fail => {
failed_at = None;
@@ -325,18 +361,18 @@ fn next_due(
/// Runs one cycle and reports what the loop should do next. `cycle` is the number of
/// cycles the thread has run, carried in the status file.
fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex<SyncStatus>, cycle: u64) -> Fate {
set_status(status, |status| status.phase = SyncPhase::Syncing);
fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc<Shared>, cycle: u64) -> Fate {
set_status(&shared.status, |status| status.phase = SyncPhase::Syncing);
match SyncState::load(vault) {
Ok(mut state) => one_pass(vault, device, remote, clock, status, &mut state, cycle),
Ok(mut state) => one_pass(vault, device, remote, clock, shared, &mut state, cycle),
Err(error) => {
set_status(status, |status| {
set_status(&shared.status, |status| {
status.phase = SyncPhase::Error;
status.last_error = Some(short(&error.to_string()));
status.last_report = None;
status.pending = 0;
});
write_status_file(vault, &fresh_status(status), cycle);
write_status_file(vault, &fresh_status(&shared.status), cycle);
Fate::Fail
}
}
@@ -344,9 +380,10 @@ fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u6
/// The sync run of one cycle. The one status update of the cycle carries everything the
/// run decided, so a reader never sees a half-updated status.
fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex<SyncStatus>, state: &mut SyncState, cycle: u64) -> Fate {
let run = sync_once(vault, remote, state, device, &stamp_from_millis(clock()));
let pending = pending_count(vault, state);
fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc<Shared>, state: &mut SyncState, cycle: u64) -> Fate {
let scan = *shared.scan_options.read().unwrap_or_else(PoisonError::into_inner);
let run = sync_once_scanned(vault, remote, state, device, &stamp_from_millis(clock()), options_for(shared, vault, cycle), scan);
let pending = pending_count(vault, state, scan);
let seq = state.last_seq;
let (fate, phase, error, counts, conflicts, last_sync): (Fate, SyncPhase, Option<String>, Option<ReportCounts>, Vec<String>, Option<u64>) = match run {
Ok(report) => (Fate::Success, SyncPhase::Idle, None, Some(counts_of(&report)), report.conflicts.clone(), Some(clock())),
@@ -356,7 +393,7 @@ fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64
}
Err(other) => (Fate::Fail, SyncPhase::Error, Some(short(&other.to_string())), None, Vec::new(), None),
};
set_status(status, |status| {
set_status(&shared.status, |status| {
status.last_seq = seq;
status.pending = pending;
status.phase = phase;
@@ -364,15 +401,33 @@ fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64
status.last_report = counts;
status.conflicts = conflicts;
status.last_sync_ms = last_sync;
status.transfer = None;
});
write_status_file(vault, &fresh_status(status), cycle);
write_status_file(vault, &fresh_status(&shared.status), cycle);
fate
}
/// The options of one pass: the default chunk limits and a callback that shows
/// the big file that is moving.
fn options_for(shared: &Arc<Shared>, vault: &Vault, cycle: u64) -> SyncOptions {
let watching = Arc::clone(shared);
let file_vault = vault.sibling();
SyncOptions {
limits: Limits::default(),
progress: Box::new(move |progress: &Progress| {
set_status(&watching.status, |status| {
status.transfer = Some(Transfer { path: progress.path.clone(), done_bytes: progress.done_bytes, total_bytes: progress.total_bytes });
});
write_status_file(&file_vault, &fresh_status(&watching.status), cycle);
}),
}
}
/// The local files whose content differs from the synced base of their path; a scan that
/// fails leaves the count unknown, which reads as nothing waiting.
fn pending_count(vault: &Vault, state: &SyncState) -> usize {
scan::scan(vault).map_or(0, |local| local.iter().filter(|(path, file)| state.bases.get(*path).is_none_or(|base| base.hash != file.hash)).count())
fn pending_count(vault: &Vault, state: &SyncState, scan_options: scan::ScanOptions) -> usize {
scan::scan(vault, scan_options)
.map_or(0, |local| local.iter().filter(|(path, file)| state.bases.get(*path).is_none_or(|base| base.hash != file.hash)).count())
}
/// The counts of a report.
@@ -446,6 +501,8 @@ struct StatusFile {
last_error: Option<String>,
pending: usize,
conflicts: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
transfer: Option<Transfer>,
}
/// Writes the status file of the last cycle so other processes can show it. A vault that
@@ -458,6 +515,7 @@ fn write_status_file(vault: &Vault, status: &SyncStatus, cycle: u64) {
last_error: status.last_error.clone(),
pending: status.pending,
conflicts: status.conflicts.clone(),
transfer: status.transfer.clone(),
};
if let Ok(bytes) = serde_json::to_vec(&file) {
let _ = vault.write_private(STATUS_FILE, &bytes);
+18
View File
@@ -63,6 +63,9 @@ pub struct Session {
pub(crate) pending: Vec<PendingChange>,
/// The highest pending-change id handed out so far.
pub(crate) next_pending_id: u64,
/// Whether the sync scan keeps the non-Markdown files (the applied user
/// option; the next started worker reads it too).
pub(crate) sync_attachments: bool,
/// The notes still to read after a lazy open.
pub(crate) scan: crate::indexing::ScanState,
/// The format/lint batch: its queue and the proposals read so far.
@@ -140,10 +143,18 @@ impl Session {
pub fn start_sync(&mut self, device_name: &str) -> crate::sync::SyncResult<()> {
self.stop_sync();
let worker = crate::sync::SyncWorker::start(&self.vault, device_name)?;
worker.set_scan_options(self.sync_scan_options());
self.sync = Some(worker);
Ok(())
}
/// The options the scan of the sync worker uses (the user's sync options;
/// the session remembers the last applied value for the next start).
#[must_use]
fn sync_scan_options(&self) -> crate::sync::ScanOptions {
crate::sync::ScanOptions { attachments: self.sync_attachments }
}
/// Stops the sync worker, if one runs; the notes are the worker's last known state.
pub fn stop_sync(&mut self) {
self.sync = None;
@@ -209,6 +220,12 @@ impl Session {
self.set_config(app.session.clone())?;
self.plugin_settings = app.plugins.clone();
self.assistant = app.assistant.clone();
// The sync worker scans with the user's options; a worker started later
// (the next open, the next pairing) reads the remembered value.
self.sync_attachments = app.sync.attachments;
if let Some(worker) = self.sync.as_ref() {
worker.set_scan_options(self.sync_scan_options());
}
Ok(())
}
@@ -373,6 +390,7 @@ impl Session {
batch: crate::batch::BatchState::default(),
grammar_transport: Box::new(notescraft_syntax::fetch::Curl),
sync: None,
sync_attachments: true,
};
session.begin_scan();
session
+1
View File
@@ -207,6 +207,7 @@ const CORE_IDS: &[&str] = &[
"sync.resolve",
// The server account, its vaults and the sharing surface (SPEC 19, spec 5-7).
"server.vaults",
"server.devices",
"server.create_vault",
"server.rename_vault",
"server.delete_vault",
+217
View File
@@ -0,0 +1,217 @@
//! The `server.devices` command against a stub server: the request it makes,
//! the rows it lists and the secrets it never carries.
//!
//! The stub is a plain TCP listener that answers one canned JSON and remembers
//! the request line, like the stub in `share_commands.rs`, kept smaller here
//! because the command needs one route.
mod common;
use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use common::temp_vault;
use notescraft_engine::{Error, Registry, Session, bind_folder, register_core};
use serde_json::{Value, json};
/// The token of the bound vault; it must never reach an answer or an error.
const TOKEN: &str = "secret-token";
/// One request the stub received.
#[derive(Debug, Clone)]
struct Request {
/// The HTTP method.
method: String,
/// The path of the request line.
path: String,
/// The `Authorization` header, when there was one.
auth: Option<String>,
}
/// A stub server: one canned answer, the requests it saw.
struct Stub {
/// The base URL of the server.
base: String,
/// Every request the stub received.
seen: Arc<Mutex<Vec<Request>>>,
/// Whether the listener thread should stop.
stop: Arc<AtomicBool>,
}
impl Stub {
/// Starts a stub on an ephemeral port that answers one route.
fn spawn(status: u16, body: &str) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind the stub");
let port = listener.local_addr().expect("addr").port();
let seen: Arc<Mutex<Vec<Request>>> = Arc::new(Mutex::new(Vec::new()));
let stop = Arc::new(AtomicBool::new(false));
let head = format!("HTTP/1.1 {status} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len());
let thread_seen = Arc::clone(&seen);
let thread_stop = Arc::clone(&stop);
std::thread::spawn(move || {
while !thread_stop.load(Ordering::Relaxed) {
match listener.accept() {
Ok((stream, _)) => {
let Ok(mut writer) = stream.try_clone() else { continue };
let mut reader = BufReader::new(stream);
if let Some(request) = read_request(&mut reader) {
thread_seen.lock().expect("seen").push(request);
let _ = writer.write_all(head.as_bytes());
let _ = writer.flush();
}
}
Err(_) => std::thread::sleep(Duration::from_millis(2)),
}
}
});
Self { base: format!("http://127.0.0.1:{port}"), seen, stop }
}
/// Every request the stub received.
fn requests(&self) -> Vec<Request> {
self.seen.lock().expect("seen").clone()
}
}
impl Drop for Stub {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
}
}
/// Reads one request: method, path, the authorization header; no body is expected.
fn read_request(reader: &mut BufReader<TcpStream>) -> Option<Request> {
let mut line = String::new();
reader.read_line(&mut line).ok()?;
let mut parts = line.split_whitespace();
let method = parts.next()?.to_owned();
let path = parts.next()?.to_owned();
let mut auth = None;
loop {
let mut header = String::new();
if reader.read_line(&mut header).ok()? == 0 {
break;
}
let trimmed = header.trim_end();
if trimmed.is_empty() {
break;
}
if let Some((name, value)) = trimmed.split_once(':')
&& name.trim().eq_ignore_ascii_case("authorization")
{
auth = Some(value.trim().to_owned());
}
}
Some(Request { method, path, auth })
}
/// A vault bound to `stub`, with the registry, the session and the vault guard.
fn bound(stub: &Stub, name: &str) -> (Registry<Session>, Session, common::TempVault) {
let vault = temp_vault(name);
bind_folder(vault.path(), &stub.base, "v-1", TOKEN, "d-1", Some("u-1")).expect("bind the vault");
let mut reg = Registry::new();
register_core(&mut reg).expect("register the core commands");
let session = Session::open(vault.path()).expect("open the vault");
(reg, session, vault)
}
/// The code of a command error.
fn code(result: &Result<Value, Error>) -> String {
match result {
Err(Error::Command { source, .. }) => source.code.clone(),
other => panic!("expected a command error, got {other:?}"),
}
}
/// The message of a command error.
fn message(result: &Result<Value, Error>) -> String {
match result {
Err(Error::Command { source, .. }) => source.message.clone(),
other => panic!("expected a command error, got {other:?}"),
}
}
/// `server.devices` asks the server with the token and lists its rows.
#[test]
fn the_device_list_goes_to_the_server_with_the_token() {
let stub = Stub::spawn(
200,
r#"[{"id":"d-1","name":"Pixel 9","created_ms":100,"last_seen_ms":null,"current":true},{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}]"#,
);
let (reg, mut session, _vault) = bound(&stub, "devices");
let answer = reg.run("server.devices", &mut session, &Value::Null).expect("server.devices");
let devices = answer.get("devices").and_then(Value::as_array).expect("the devices");
assert_eq!(devices.len(), 2);
assert_eq!(devices.first().and_then(|device| device.get("id")), Some(&json!("d-1")));
assert_eq!(devices.first().and_then(|device| device.get("name")), Some(&json!("Pixel 9")));
assert_eq!(devices.first().and_then(|device| device.get("current")), Some(&json!(true)));
assert_eq!(devices.first().and_then(|device| device.get("last_seen_ms")), Some(&Value::Null));
assert_eq!(devices.get(1).and_then(|device| device.get("last_seen_ms")), Some(&json!(60)));
let requests = stub.requests();
assert_eq!(requests.len(), 1);
assert_eq!(requests.first().map(|request| request.method.as_str()), Some("GET"));
assert_eq!(requests.first().map(|request| request.path.as_str()), Some("/v1/vaults/v-1/devices"));
assert_eq!(requests.first().and_then(|request| request.auth.clone()), Some(format!("Bearer {TOKEN}")));
assert!(!answer.to_string().contains(TOKEN), "the token never reaches the answer: {answer}");
}
#[test]
fn no_token_reaches_the_answer_or_the_rows() {
let stub = Stub::spawn(200, r#"[{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}]"#);
let (reg, mut session, _vault) = bound(&stub, "no-token");
let answer = reg.run("server.devices", &mut session, &Value::Null).expect("server.devices");
assert!(!answer.to_string().contains(TOKEN), "{}", answer);
assert!(!answer.to_string().contains("token"), "{}", answer);
assert!(!answer.to_string().contains("secret"), "{}", answer);
}
/// A vault that is not bound says so instead of asking the server.
#[test]
fn an_unbound_vault_reports_not_bound() {
let stub = Stub::spawn(200, "[]");
let mut reg = Registry::new();
register_core(&mut reg).expect("register the core commands");
let vault = temp_vault("devices-unbound");
let mut session = Session::open(vault.path()).expect("open the vault");
let result = reg.run("server.devices", &mut session, &Value::Null);
assert_eq!(code(&result), "share");
assert!(message(&result).contains("not bound"), "{}", message(&result));
assert!(stub.requests().is_empty(), "no request was made");
}
/// The message of the server reaches the user, the token never does.
#[test]
fn the_server_answer_refuses_like_the_others() {
let stub = Stub::spawn(401, r#"{"error":"token is not valid for this vault"}"#);
let (reg, mut session, _vault) = bound(&stub, "devices-401");
let result = reg.run("server.devices", &mut session, &Value::Null);
assert_eq!(code(&result), "share");
let text = message(&result);
assert!(text.contains("token is not valid for this vault"), "{text}");
assert!(!text.contains(TOKEN), "{text}");
}
/// `server.devices` never panics on hostile params, and asks nothing.
#[test]
fn hostile_params_never_panic_the_devices_command() {
let stub = Stub::spawn(200, "[]");
let (reg, mut session, _vault) = bound(&stub, "devices-hostile");
let previous = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let mut panicked = Vec::new();
for params in notescraft_engine::adversarial_params() {
if let Err(Error::Panicked(other)) = reg.run("server.devices", &mut session, &params) {
panicked.push(other);
}
}
std::panic::set_hook(previous);
assert!(panicked.is_empty(), "commands panicked: {panicked:?}");
}
+360
View File
@@ -0,0 +1,360 @@
//! Chunked uploads: the client against a server that keeps chunks in memory.
//!
//! The fake server holds the chunks it was sent in a map and can be told to
//! cut the connection, to lose a chunk or to change the file under the
//! client's feet.
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard};
use notescraft_engine::{Changes, Limits, Progress, PutReply, Remote, SyncError, SyncOptions, SyncResult, SyncState, sync_once_with, upload_chunked};
use notescraft_sync::{Ledger, Outcome, Write, hash_hex};
use notescraft_vault::Vault;
/// The size of one chunk in the tests.
const CHUNK: usize = 1024;
/// The size above which a file uploads in chunks.
const THRESHOLD: u64 = 2 * 1024;
/// The server the client talks to.
struct Server {
inner: Mutex<Inner>,
}
/// What the server keeps.
struct Inner {
ledger: Ledger,
/// The chunks the server holds, by their hash.
chunks: BTreeMap<String, Vec<u8>>,
/// The files the server holds, by their path.
files: BTreeMap<String, Vec<u8>>,
/// The hashes of every chunk that was sent.
chunk_calls: Vec<String>,
/// The chunk list of every chunked write.
chunked_calls: Vec<Vec<String>>,
/// How many chunk sends there were.
calls: usize,
/// From this send on, the connection is cut.
cut_from: Option<usize>,
/// The file that changes while the client is uploading it.
corrupt: Option<(PathBuf, Vec<u8>)>,
}
impl Inner {
fn new() -> Self {
Self {
ledger: Ledger::new(),
chunks: BTreeMap::new(),
files: BTreeMap::new(),
chunk_calls: Vec::new(),
chunked_calls: Vec::new(),
calls: 0,
cut_from: None,
corrupt: None,
}
}
}
impl Server {
fn new() -> Arc<Server> {
Arc::new(Server { inner: Mutex::new(Inner::new()) })
}
fn lock(&self) -> MutexGuard<'_, Inner> {
self.inner.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// The hashes of the chunks that were sent.
fn chunk_calls(&self) -> Vec<String> {
self.lock().chunk_calls.clone()
}
/// How many chunked writes the server saw.
fn chunked_calls(&self) -> usize {
self.lock().chunked_calls.len()
}
/// The file the server holds.
fn file(&self, path: &str) -> Option<Vec<u8>> {
self.lock().files.get(path).cloned()
}
/// Cuts the connection from the given chunk send on.
fn cut_from(&self, call: usize) {
self.lock().cut_from = Some(call);
}
/// Lets the connection through again.
fn mend(&self) {
self.lock().cut_from = None;
}
/// Forgets the chunks that were sent, so a test can look at one pass only.
fn forget_calls(&self) {
self.lock().chunk_calls.clear();
}
/// Changes the file while the client uploads it.
fn corrupt(&self, path: PathBuf, bytes: Vec<u8>) {
self.lock().corrupt = Some((path, bytes));
}
}
/// The client the driver talks to.
struct FakeRemote {
server: Arc<Server>,
}
impl Remote for FakeRemote {
fn changes(&self, since: u64, limit: usize, _wait: u64) -> SyncResult<Changes> {
let inner = self.server.lock();
Ok(inner.ledger.changes_since(since, limit))
}
fn get(&self, path: &str) -> SyncResult<Vec<u8>> {
let inner = self.server.lock();
inner.files.get(path).cloned().ok_or_else(|| SyncError::Status { status: 404, body: path.to_owned() })
}
fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
let mut inner = self.server.lock();
let outcome = inner
.ledger
.apply(Write { path: path.to_owned(), hash: Some(hash_hex(bytes)), size: u64::try_from(bytes.len()).expect("size"), mtime_ms, base_seq })
.map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?;
match outcome {
Outcome::Applied(record) => {
inner.files.insert(path.to_owned(), bytes.to_vec());
Ok(PutReply::Applied(record))
}
Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
}
}
fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply> {
let mut inner = self.server.lock();
let outcome = inner
.ledger
.apply(Write { path: path.to_owned(), hash: None, size: 0, mtime_ms: 0, base_seq })
.map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?;
match outcome {
Outcome::Applied(record) => {
inner.files.remove(path);
Ok(PutReply::Applied(record))
}
Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
}
}
fn missing_chunks(&self, hashes: &[String]) -> SyncResult<Vec<String>> {
let mut inner = self.server.lock();
if let Some((path, bytes)) = inner.corrupt.take() {
std::fs::write(&path, &bytes).map_err(|error| SyncError::Io(error.to_string()))?;
}
Ok(hashes.iter().filter(|hash| !inner.chunks.contains_key(*hash)).cloned().collect())
}
fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> {
let mut inner = self.server.lock();
inner.calls = inner.calls.saturating_add(1);
inner.chunk_calls.push(hash.to_owned());
if inner.cut_from.is_some_and(|from| inner.calls >= from) {
return Err(SyncError::Http("the connection was cut".to_owned()));
}
inner.chunks.insert(hash.to_owned(), bytes.to_vec());
Ok(())
}
fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
let mut inner = self.server.lock();
inner.chunked_calls.push(chunk_hashes.to_vec());
let mut bytes = Vec::new();
for hash in chunk_hashes {
let Some(chunk) = inner.chunks.get(hash) else { return Err(SyncError::MissingChunks(vec![hash.clone()])) };
bytes.extend_from_slice(chunk);
}
let outcome = inner
.ledger
.apply(Write { path: path.to_owned(), hash: Some(content_hash.to_owned()), size: u64::try_from(bytes.len()).expect("size"), mtime_ms, base_seq })
.map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?;
match outcome {
Outcome::Applied(record) => {
inner.files.insert(path.to_owned(), bytes);
Ok(PutReply::Applied(record))
}
Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
}
}
}
/// A vault in a fresh folder.
fn temp_vault(name: &str) -> Vault {
let root = std::env::temp_dir().join(format!("nc-engine-chunks-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).expect("temp folder");
Vault::open(&root).expect("vault")
}
/// The limits the tests use: small chunks, so a small file has many of them.
fn limits() -> Limits {
Limits { threshold: THRESHOLD, chunk: CHUNK }
}
/// Ten kilobytes whose chunks differ from one another.
fn ten_chunks() -> Vec<u8> {
(0..10u8).flat_map(|index| vec![index; CHUNK]).collect()
}
/// The hash of the chunk number `index` of `bytes`.
fn chunk_hash(bytes: &[u8], index: usize) -> String {
let start = index * CHUNK;
let end = start.saturating_add(CHUNK);
hash_hex(bytes.get(start..end).expect("chunk"))
}
#[test]
fn a_big_file_travels_in_chunks_and_is_written_once() {
let server = Server::new();
let vault = temp_vault("big");
let bytes = ten_chunks();
vault.write("big.bin", &bytes).expect("write");
let remote = FakeRemote { server: server.clone() };
let reply = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload");
assert!(matches!(reply, PutReply::Applied(_)), "{reply:?}");
assert_eq!(server.chunk_calls().len(), 10, "one send per chunk");
assert_eq!(server.chunked_calls(), 1, "the file is written once");
assert_eq!(server.file("big.bin"), Some(bytes), "the server holds the whole file");
}
#[test]
fn a_cut_connection_resumes_with_the_chunks_that_never_arrived() {
let server = Server::new();
let vault = temp_vault("resume");
let bytes = ten_chunks();
vault.write("big.bin", &bytes).expect("write");
let remote = FakeRemote { server: server.clone() };
// The connection dies on the sixth chunk; the chunk is tried again and fails.
server.cut_from(6);
let error = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect_err("cut");
assert!(matches!(error, SyncError::Http(_)), "{error:?}");
assert_eq!(server.chunked_calls(), 0, "the file was never written");
// The next pass sends only the chunks that never arrived.
server.mend();
server.forget_calls();
upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("resumed");
let sent: Vec<String> = (5..10).map(|index| chunk_hash(&bytes, index)).collect();
assert_eq!(server.chunk_calls(), sent, "only the missing chunks are sent again");
assert_eq!(server.file("big.bin"), Some(bytes), "the server holds the whole file");
}
#[test]
fn the_same_chunk_of_a_file_is_sent_once() {
let server = Server::new();
let vault = temp_vault("same");
let bytes = vec![7u8; CHUNK * 4];
vault.write("same.bin", &bytes).expect("write");
let remote = FakeRemote { server: server.clone() };
upload_chunked(&remote, &vault, "same.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload");
assert_eq!(server.chunk_calls().len(), 1, "the four equal chunks travel once");
assert_eq!(server.file("same.bin"), Some(bytes), "the file still holds the four blocks");
}
#[test]
fn the_progress_grows_and_ends_at_the_size_of_the_file() {
let server = Server::new();
let vault = temp_vault("progress");
let bytes = ten_chunks();
vault.write("big.bin", &bytes).expect("write");
let remote = FakeRemote { server: server.clone() };
let mut seen: Vec<Progress> = Vec::new();
upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |progress: &Progress| seen.push(progress.clone())).expect("upload");
assert_eq!(seen.len(), 10, "one report per chunk");
assert!(seen.windows(2).all(|pair| pair[0].done_bytes <= pair[1].done_bytes), "the progress never goes back");
let last = seen.last().expect("a report");
assert_eq!(last.done_bytes, last.total_bytes);
assert_eq!(last.total_bytes, u64::try_from(bytes.len()).expect("size"));
assert_eq!(last.path, "big.bin");
}
#[test]
fn a_small_file_never_touches_the_chunk_methods() {
let server = Server::new();
let vault = temp_vault("small");
vault.write("small.md", b"# one\n").expect("write");
let remote = FakeRemote { server: server.clone() };
let mut state = SyncState::default();
let report = sync_once_with(&vault, &remote, &mut state, "phone", "2026-10-20", SyncOptions { limits: limits(), progress: Box::new(|_: &Progress| {}) })
.expect("sync");
assert_eq!(report.uploaded, vec!["small.md".to_owned()]);
assert_eq!(server.chunk_calls(), Vec::<String>::new(), "a small file is not chunked");
assert_eq!(server.chunked_calls(), 0);
assert_eq!(server.file("small.md"), Some(b"# one\n".to_vec()));
}
#[test]
fn a_file_that_changes_while_it_is_uploaded_is_refused() {
let server = Server::new();
let vault = temp_vault("changed");
let bytes = ten_chunks();
vault.write("big.bin", &bytes).expect("write");
let remote = FakeRemote { server: server.clone() };
// The file changes between the hash pass and the upload pass.
server.corrupt(vault.root().join("big.bin"), vec![1u8; CHUNK * 10]);
let error = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect_err("changed");
assert!(matches!(error, SyncError::BadRemote(_)), "{error:?}");
assert_eq!(server.chunked_calls(), 0, "no mixed file reaches the server");
assert_eq!(server.file("big.bin"), None);
}
#[test]
fn an_empty_file_goes_as_a_plain_write() {
let server = Server::new();
let vault = temp_vault("empty");
vault.write("empty.bin", b"").expect("write");
let remote = FakeRemote { server: server.clone() };
let reply = upload_chunked(&remote, &vault, "empty.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload");
assert!(matches!(reply, PutReply::Applied(_)), "{reply:?}");
assert_eq!(server.chunk_calls(), Vec::<String>::new(), "an empty file has no chunk");
assert_eq!(server.chunked_calls(), 0);
assert_eq!(server.file("empty.bin"), Some(Vec::new()));
}
#[test]
fn the_threshold_decides_what_is_chunked() {
let server = Server::new();
let vault = temp_vault("threshold");
let at_threshold: Vec<u8> = (0..THRESHOLD).map(|index| u8::try_from(index % 251).expect("byte")).collect();
let over_threshold: Vec<u8> = (0..=THRESHOLD).map(|index| u8::try_from(index % 251).expect("byte")).collect();
vault.write("at.bin", &at_threshold).expect("write");
vault.write("over.bin", &over_threshold).expect("write");
let remote = FakeRemote { server: server.clone() };
let mut state = SyncState::default();
let report = sync_once_with(&vault, &remote, &mut state, "phone", "2026-10-20", SyncOptions { limits: limits(), progress: Box::new(|_: &Progress| {}) })
.expect("sync");
assert_eq!(report.uploaded.len(), 2);
assert_eq!(server.chunked_calls(), 1, "only the file over the threshold is chunked");
assert_eq!(server.chunk_calls().len(), 3, "the file over the threshold has three chunks");
assert_eq!(server.file("at.bin"), Some(at_threshold));
assert_eq!(server.file("over.bin"), Some(over_threshold));
}
+25
View File
@@ -1328,6 +1328,31 @@ ago-minutes = "{n} min ago"
ago-hours = "{n} h ago"
ago-days = "{n} d ago"
[ui.sync-sheet]
title = "Sync"
status-up-to-date = "Everything is up to date"
status-syncing = "Syncing..."
status-offline = "Offline, will retry"
status-sign-in = "Sign in again"
status-error = "Sync stopped"
last-sync = "Last sync {when} · {host}"
never-synced = "Not yet synced"
devices = "Devices"
this-phone = "This phone"
this-computer = "This computer"
last-seen = "last seen {when}"
never-seen = "last seen never"
chip-online = "online"
chip-synced = "synced"
chip-stale = "stale"
conflicts = "{n} conflicts to review"
wifi-only = "Sync on Wi-Fi only"
attachments = "Sync attachments"
settings-and-themes = "Sync settings and themes"
sync-now = "Sync now"
loading = "Loading devices..."
pairing-failed = "The pairing failed. Check the server address and the code."
[ui.live]
pill-live = "Live"
go-live = "Go live"
+61 -5
View File
@@ -9,12 +9,33 @@ const HEX: &[u8; 16] = b"0123456789abcdef";
/// SHA-256 of `bytes` as 64 lowercase hex digits.
#[must_use]
pub fn hash_hex(bytes: &[u8]) -> String {
let mut out = String::with_capacity(64);
for byte in &Sha256::digest(bytes) {
out.push(char::from(HEX.get(usize::from(byte >> 4)).copied().unwrap_or(b'0')));
out.push(char::from(HEX.get(usize::from(byte & 0x0f)).copied().unwrap_or(b'0')));
hex_digest(&Sha256::digest(bytes))
}
/// An incremental SHA-256 hash: the pieces are added one at a time, so content
/// too large to hold in memory can still be hashed.
#[derive(Debug, Clone, Default)]
pub struct Hasher {
inner: Sha256,
}
impl Hasher {
/// Adds `bytes` to the hash.
pub fn update(&mut self, bytes: &[u8]) {
self.inner.update(bytes);
}
out
/// The hash of everything added so far, as 64 lowercase hex digits.
#[must_use]
pub fn finish_hex(self) -> String {
hex_digest(&self.inner.finalize())
}
}
/// A fresh incremental hasher, for content that never sits in memory whole.
#[must_use]
pub fn hasher() -> Hasher {
Hasher::default()
}
/// Checks the shape of a hash received from outside.
@@ -28,3 +49,38 @@ pub fn check_hash(hash: &str) -> Result<()> {
Err(Error::InvalidHash(hash.chars().take(80).collect()))
}
}
/// A digest as 64 lowercase hex digits.
fn hex_digest(digest: &[u8]) -> String {
let mut out = String::with_capacity(digest.len() * 2);
for byte in digest {
out.push(char::from(HEX.get(usize::from(byte >> 4)).copied().unwrap_or(b'0')));
out.push(char::from(HEX.get(usize::from(byte & 0x0f)).copied().unwrap_or(b'0')));
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_incremental_hash_is_the_hash_of_the_joined_bytes() {
let mut hasher = hasher();
hasher.update(b"hello ");
hasher.update(b"world");
assert_eq!(hasher.finish_hex(), hash_hex(b"hello world"));
}
#[test]
fn an_empty_hash_is_the_hash_of_no_bytes() {
assert_eq!(hasher().finish_hex(), hash_hex(b""));
}
#[test]
fn a_broken_hash_is_refused() {
assert!(check_hash(&hash_hex(b"x")).is_ok());
assert!(check_hash("not a hash").is_err());
assert!(check_hash(&hash_hex(b"x").to_uppercase()).is_err());
}
}
+2 -2
View File
@@ -9,7 +9,7 @@
//! of each path and a `/changes` page to a list of [`Action`]s.
//! * [`merge3`] — three-way line merge; anything doubtful is a conflict, never a guess.
//! * [`conflict_path`] — the name of the copy a conflict leaves behind.
//! * [`check_path`], [`check_syncable`], [`hash_hex`] — the gate every path and hash passes.
//! * [`check_path`], [`check_syncable`], [`hash_hex`], [`hasher`] — the gate every path and hash passes.
//!
//! ```
//! use notescraft_sync::{Ledger, Outcome, Write, hash_hex};
@@ -33,7 +33,7 @@ mod record;
pub use conflict::conflict_path;
pub use error::{Error, Result};
pub use hash::{check_hash, hash_hex};
pub use hash::{Hasher, check_hash, hash_hex, hasher};
pub use ledger::{Ledger, Outcome, Write};
pub use merge::{Merge, merge3};
pub use path::{MAX_PATH_BYTES, check_path, check_syncable, is_excluded};
+1
View File
@@ -13,6 +13,7 @@ pub mod sidebar;
pub mod sidebar_tabs;
pub mod source_panel;
pub mod sync_pill;
pub mod sync_sheet;
pub mod tags_panel;
pub mod tree;
#[cfg(test)]
+7 -1
View File
@@ -106,7 +106,13 @@ pub fn show(ui: &mut Ui, tokens: &Tokens, status: &SyncStatus, right: f32, middl
ui.painter().galley(text, galley, tint);
let response = ui.interact(rect, ui.id().with("sync-pill"), Sense::click());
let _ = crate::motion::hover(ui.ctx(), "sync-pill", &response);
response.clicked()
let clicked = response.clicked();
// Phone width: the click asks the shell for the sheet, not for Settings >
// Sync; the flag is a same-frame egui temp the shell reads and clears.
if clicked && crate::style::compact::is_compact(ui.ctx()) {
ui.ctx().data_mut(|data| data.insert_temp(egui::Id::new("sync-sheet-request"), true));
}
clicked
}
#[cfg(test)]
@@ -0,0 +1,284 @@
//! The state half of the sync sheet: the status card, the device rows and the
//! conflict row — everything the vault's standing on the server decides.
use egui::{Align2, CornerRadius, FontId, Rect, Sense, Ui, Vec2, vec2};
use notescraft_engine::{ServerDevice, SyncPhase, SyncStatus};
use super::{Facts, SheetAction, SyncSheet};
use crate::text::sync_sheet as strings;
use crate::theme::{Tokens, color};
/// Radius of one card, points.
const CARD_RADIUS: u8 = 12;
/// Height of one device row, points.
const DEVICE_ROW: f32 = 72.0;
/// Height of the conflict row, points.
const CONFLICT_ROW: f32 = 56.0;
/// Height of the status card, points.
const STATUS_ROW: f32 = 96.0;
/// Side of the status icon, points.
const STATUS_ICON: f32 = 24.0;
/// Left pad inside a card, points.
const CARD_PAD: f32 = 16.0;
/// Gap between the icon and the text block, points.
const TEXT_GAP: f32 = 16.0;
/// Side of the chip's state dot, points.
const CHIP_DOT: f32 = 8.0;
/// Space between the chip's dot and its word, points.
const CHIP_GAP: f32 = 6.0;
/// Text size of a name, points.
const NAME: f32 = 15.0;
/// Text size of muted detail lines, points.
const DETAIL: f32 = 13.0;
/// Gap between the headline and the detail of the status card.
const LINE_GAP: f32 = 16.0;
/// Icons of the status card, by phase; a phase without one shows its error.
const UP_TO_DATE_ICON: &str = "circle-check";
const SYNCING_ICON: &str = "refresh-cw";
const ATTENTION_ICON: &str = "circle-alert";
const ERROR_ICON: &str = "triangle-alert";
/// Milliseconds a device may be out of sight to still count as online.
const ONLINE_WINDOW_MS: u64 = 2 * 60 * 1000;
/// Milliseconds in one day.
const MS_PER_DAY: u64 = 24 * 60 * 60 * 1000;
/// The icon of the status card, from the phase.
fn status_icon(status: Option<&SyncStatus>) -> &'static str {
match status.map(|status| status.phase) {
Some(SyncPhase::Idle) => UP_TO_DATE_ICON,
Some(SyncPhase::Syncing) => SYNCING_ICON,
Some(SyncPhase::Offline | SyncPhase::SignInAgain) => ATTENTION_ICON,
Some(SyncPhase::Error) | None => ERROR_ICON,
}
}
/// The tint of the status card's icon.
fn status_tint(tokens: &Tokens, status: Option<&SyncStatus>) -> egui::Color32 {
let colors = &tokens.theme.colors;
match status.map(|status| status.phase) {
Some(SyncPhase::Idle) => color(colors.success),
Some(SyncPhase::Syncing | SyncPhase::SignInAgain) => color(colors.warning),
Some(SyncPhase::Offline) => color(colors.text_muted),
Some(SyncPhase::Error) | None => color(colors.danger),
}
}
/// The headline of the status card, from the phase; the error speaks for itself.
#[must_use]
pub fn headline(status: Option<&SyncStatus>) -> String {
match status.map(|status| status.phase) {
Some(SyncPhase::Idle) => strings::status_up_to_date().to_owned(),
Some(SyncPhase::Syncing) => strings::status_syncing().to_owned(),
Some(SyncPhase::Offline) => strings::status_offline().to_owned(),
Some(SyncPhase::SignInAgain) => strings::status_sign_in().to_owned(),
Some(SyncPhase::Error) | None => status.and_then(|status| status.last_error.clone()).unwrap_or_else(|| strings::status_error().to_owned()),
}
}
/// The second line of the status card: "Last sync 2 min ago · host".
fn status_detail(status: Option<&SyncStatus>, now_ms: u64) -> String {
let host = status.and_then(|status| status.server.as_deref()).map(crate::views::settings_view::sync::server_host).unwrap_or_default();
let Some(last) = status.and_then(|status| status.last_sync_ms) else {
// Never synced: no "Last sync" prefix, and the host only when there is one.
let never = strings::never_synced();
return if host.is_empty() { never.to_owned() } else { format!("{never} · {host}") };
};
let text = strings::last_sync().replace("{when}", &crate::views::settings_view::sync::ago(last, now_ms)).replace("{host}", host);
text.trim_end_matches([' ', '·']).to_owned()
}
/// The status card (design: the check and the two lines).
pub fn status_card(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>) {
let colors = &tokens.theme.colors;
let (rect, _) = ui.allocate_exact_size(Vec2::new(ui.available_width(), STATUS_ROW), Sense::hover());
ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2));
let icon = Rect::from_center_size(rect.left_top() + vec2(CARD_PAD + STATUS_ICON / 2.0, rect.height() / 2.0), Vec2::splat(STATUS_ICON));
crate::icons::paint(ui, status_icon(facts.status.as_ref()), icon, status_tint(tokens, facts.status.as_ref()));
let text_left = rect.left() + CARD_PAD + STATUS_ICON + TEXT_GAP;
let width = rect.right() - text_left - CARD_PAD;
ui.painter().text(
egui::pos2(text_left, rect.center().y - LINE_GAP),
Align2::LEFT_CENTER,
headline(facts.status.as_ref()),
FontId::new(NAME, crate::fonts::semibold()),
color(colors.text),
);
let font = FontId::proportional(DETAIL);
let galley = ui.painter().layout(status_detail(facts.status.as_ref(), facts.now_ms), font, color(colors.text_muted), width);
ui.painter().galley(egui::pos2(text_left, rect.center().y + LINE_GAP - galley.size().y / 2.0), galley, color(colors.text_muted));
}
/// The devices: the heading, the rows (current first) and the line of a load.
pub fn section(sheet: &mut SyncSheet, ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option<SheetAction>) {
let colors = &tokens.theme.colors;
heading(ui, tokens, strings::devices());
if sheet.loading() {
ui.horizontal(|ui| {
ui.add(egui::Spinner::new().size(14.0).color(color(colors.accent)));
ui.add_space(8.0);
ui.label(egui::RichText::new(strings::loading()).size(DETAIL).color(color(colors.text_muted)));
});
} else if let Some(error) = sheet.error.as_deref() {
ui.label(egui::RichText::new(error).size(DETAIL).color(color(colors.danger)));
} else {
for device in ordered(&sheet.devices) {
device_row(ui, tokens, &device, facts);
}
}
conflict_row(ui, tokens, facts, action);
}
/// The devices with the current one first, the others in the order of the server.
fn ordered(devices: &[ServerDevice]) -> Vec<ServerDevice> {
let mut sorted = devices.to_vec();
sorted.sort_by_key(|device| !device.current);
sorted
}
/// The card of one device (design: name, "last seen ago", state chip).
fn device_row(ui: &mut Ui, tokens: &Tokens, device: &ServerDevice, facts: &Facts<'_>) {
let colors = &tokens.theme.colors;
let (rect, _) = ui.allocate_exact_size(Vec2::new(ui.available_width(), DEVICE_ROW), Sense::hover());
ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2));
let name = match device.current {
true if facts.compact => strings::this_phone(),
true => strings::this_computer(),
false => device.name.as_str(),
};
ui.painter().text(
egui::pos2(rect.left() + CARD_PAD, rect.center().y - 12.0),
Align2::LEFT_CENTER,
name,
FontId::new(NAME, crate::fonts::semibold()),
color(colors.text),
);
let seen = match device.last_seen_ms {
Some(seen) => strings::last_seen().replace("{when}", &crate::views::settings_view::sync::ago(seen, facts.now_ms)),
None => strings::never_seen().to_owned(),
};
ui.painter().text(
egui::pos2(rect.left() + CARD_PAD, rect.center().y + 12.0),
Align2::LEFT_CENTER,
seen,
FontId::proportional(DETAIL),
color(colors.text_muted),
);
paint_chip(ui, tokens, rect, device, facts.now_ms);
}
/// The state chip at the right end of a device row.
fn paint_chip(ui: &Ui, tokens: &Tokens, row: Rect, device: &ServerDevice, now_ms: u64) {
let (label, tint) = chip_state(device.last_seen_ms, now_ms, tokens);
let galley = ui.painter().layout_no_wrap(label.to_owned(), FontId::new(DETAIL, crate::fonts::semibold()), tint);
let chip_width = CHIP_DOT + CHIP_GAP + galley.size().x + CARD_PAD;
let left = row.right() - chip_width;
let centre_y = row.center().y;
ui.painter().circle_filled(egui::pos2(left + CHIP_DOT / 2.0, centre_y), CHIP_DOT / 2.0, tint);
ui.painter().galley(egui::pos2(left + CHIP_DOT + CHIP_GAP, centre_y - galley.size().y / 2.0), galley, tint);
}
/// The word and colour of the state chip: online, synced, stale.
fn chip_state(last_seen_ms: Option<u64>, now_ms: u64, tokens: &Tokens) -> (&'static str, egui::Color32) {
let colors = &tokens.theme.colors;
let seen_ago = last_seen_ms.map(|seen| now_ms.saturating_sub(seen));
let label = match seen_ago {
Some(ago) if ago <= ONLINE_WINDOW_MS => strings::chip_online(),
Some(ago) if ago <= MS_PER_DAY => strings::chip_synced(),
_ => strings::chip_stale(),
};
(
label,
match seen_ago {
Some(ago) if ago <= MS_PER_DAY => color(colors.success),
Some(_) => color(colors.warning),
None => color(colors.text_muted),
},
)
}
/// The row "N conflicts to review"; hidden unless there is something to review.
fn conflict_row(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option<SheetAction>) {
if facts.conflicts == 0 {
return;
}
let colors = &tokens.theme.colors;
let label = strings::conflicts().replace("{n}", &facts.conflicts.to_string());
let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), CONFLICT_ROW), Sense::click());
hit.widget_info(|| egui::WidgetInfo::selected(egui::WidgetType::Button, true, false, "conflicts"));
ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2));
ui.painter().text(
egui::pos2(rect.left() + CARD_PAD, rect.center().y),
Align2::LEFT_CENTER,
label,
FontId::new(NAME, crate::fonts::semibold()),
color(colors.text),
);
crate::icons::paint(
ui,
"chevron-right",
Rect::from_center_size(egui::pos2(rect.right() - CARD_PAD - 8.0, rect.center().y), vec2(16.0, 16.0)),
color(colors.text_muted),
);
if hit.clicked() {
*action = Some(SheetAction::ReviewConflicts);
}
}
/// The small caps heading over a group of cards.
fn heading(ui: &mut Ui, tokens: &Tokens, text: &str) {
let colors = &tokens.theme.colors;
ui.add_space(4.0);
ui.label(egui::RichText::new(text).font(FontId::new(12.0, crate::fonts::semibold())).color(color(colors.text_muted)));
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_headlines_follow_the_phases() {
let status = SyncStatus { phase: SyncPhase::Idle, ..SyncStatus::default() };
assert_eq!(headline(Some(&status)), "Everything is up to date");
let syncing = SyncStatus { phase: SyncPhase::Syncing, ..SyncStatus::default() };
assert_eq!(headline(Some(&syncing)), "Syncing...");
assert_eq!(headline(None), "Sync stopped", "a sheet without a status speaks its own state");
}
#[test]
fn the_error_speaks_as_the_headline() {
let status = SyncStatus { phase: SyncPhase::Error, last_error: Some("quota is full".to_owned()), ..SyncStatus::default() };
assert_eq!(headline(Some(&status)), "quota is full");
}
#[test]
fn the_detail_carries_the_last_sync_and_the_host() {
let status = SyncStatus { server: Some("https://sync.example.com".to_owned()), last_sync_ms: Some(998_000), ..SyncStatus::default() };
assert_eq!(status_detail(Some(&status), 1_000_000), "Last sync 2 s ago · sync.example.com");
assert_eq!(status_detail(None, 1_000_000), strings::never_synced());
}
#[test]
fn the_chips_age() {
let tokens = Tokens::new(notescraft_themes::builtin("catppuccin-mocha").expect("theme"));
let now = 1_000_000_000_u64;
let (online, _) = chip_state(Some(now - 1000), now, &tokens);
assert_eq!(online, strings::chip_online());
let (synced, _) = chip_state(Some(now - 3_600_000), now, &tokens);
assert_eq!(synced, strings::chip_synced());
let (stale, _) = chip_state(Some(now - MS_PER_DAY - 1), now, &tokens);
assert_eq!(stale, strings::chip_stale());
let (never, _) = chip_state(None, now, &tokens);
assert_eq!(never, strings::chip_stale());
}
#[test]
fn the_current_device_comes_first() {
let devices = [
ServerDevice { id: "d-2".to_owned(), name: "Laptop".to_owned(), ..ServerDevice::default() },
ServerDevice { id: "d-1".to_owned(), name: "Phone".to_owned(), current: true, ..ServerDevice::default() },
];
let listed = ordered(&devices);
assert_eq!(listed.first().map(|device| device.id.as_str()), Some("d-1"), "the current device first");
assert_eq!(listed.get(1).map(|device| device.id.as_str()), Some("d-2"));
}
}
+224
View File
@@ -0,0 +1,224 @@
//! The sync sheet (design `03-sync-sheet`): everything about sync of the
//! bound vault in one bottom sheet of the compact layout.
//!
//! The compact-layout sync pill opens it; on a wider window the pill keeps
//! opening Settings > Sync (see [`crate::panels::sync_pill`]). The sheet holds
//! the state (devices, the running load job) and gets the live facts — the
//! status, the conflicts and the user's options — handed in every frame.
//!
//! No ask waits on the UI thread: the devices load on a worker thread that
//! holds the one [`ServerClient`] (the token never crosses back) and answers
//! through a channel the sheet polls, exactly like the Share dialog.
mod devices;
mod options;
use std::sync::mpsc::{Receiver, TryRecvError};
use std::sync::{Arc, Mutex};
use egui::{Context, ScrollArea};
use notescraft_engine::{ServerClient, ServerDevice, SyncSettings, SyncStatus, load_binding, load_token};
use crate::text::sync_sheet as strings;
use crate::theme::Tokens;
/// The id of the sheet, also the salt of its widgets.
const SHEET_ID: &str = "notescraft-sync-sheet";
/// Most of the screen the open sheet may take, as a fraction, so the content
/// scrolls instead of reaching the top edge of the window; the sheet of the
/// design (status, three devices, the conflict row, three options and the
/// button) is about 700 pt tall on a 390 x 844 phone and shows whole.
const MAX_SHEET_HEIGHT: f32 = 0.9;
/// How long a sheet without a job waits before it repaints for the spinner.
const LOAD_MILLIS: u64 = 40;
/// What the shell hands to the sheet every frame.
pub struct Facts<'a> {
/// The status of the sync worker, `None` when the vault has no worker.
pub status: Option<SyncStatus>,
/// The conflict copies waiting for a decision.
pub conflicts: usize,
/// The user's three sync options.
pub options: &'a SyncSettings,
/// Now, in Unix milliseconds, for the ages.
pub now_ms: u64,
/// Whether the window is in the compact (phone) layout.
pub compact: bool,
}
/// What the sheet asks the app to do.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SheetAction {
/// Close the sheet (the ×, a tap on the dimmed area or Escape).
Close,
/// Run `sync.now`.
SyncNow,
/// Open Settings > Sync's conflict list.
ReviewConflicts,
/// One of the three options flipped; the shell stores the new value.
Toggle(SheetOption, bool),
}
/// Which of the three options a [`SheetAction::Toggle`] names.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SheetOption {
/// `sync.wifi_only`, stored but not yet honoured by the worker.
WifiOnly,
/// `sync.attachments`: `false` leaves the non-Markdown files out of the scan.
Attachments,
/// `sync.settings_and_themes`, stored but not yet honoured by the worker.
SettingsAndThemes,
}
impl SheetOption {
/// The settings key of the option.
#[must_use]
pub fn key(self) -> &'static str {
match self {
SheetOption::WifiOnly => "sync.wifi_only",
SheetOption::Attachments => "sync.attachments",
SheetOption::SettingsAndThemes => "sync.settings_and_themes",
}
}
}
/// The state of the open sheet between frames.
pub struct SyncSheet {
/// The client of the account, `None` when the vault is not bound.
client: Option<Arc<Mutex<ServerClient>>>,
/// The vault id on the server.
vault_id: String,
/// The devices of the last load, current first.
devices: Vec<ServerDevice>,
/// What the last failed load said.
error: Option<String>,
/// The running load job, when there is one.
job: Option<Arc<Mutex<Receiver<DevicesOutcome>>>>,
}
/// The answer of a load job.
enum DevicesOutcome {
/// The devices the server named.
Listed(Vec<ServerDevice>),
/// What the server or the connection said.
Failed(String),
}
impl SyncSheet {
/// Builds the sheet over the vault of `session` and asks for its devices.
#[must_use]
pub fn open(session: &notescraft_engine::Session) -> Self {
let mut sheet = Self { client: None, vault_id: String::new(), devices: Vec::new(), error: None, job: None };
let binding = load_binding(session.vault()).ok().flatten();
let token = load_token(session.vault()).ok().flatten();
if let Some(binding) = binding
&& let Some(token) = token
{
sheet.client = Some(Arc::new(Mutex::new(ServerClient::new(&binding.server, &token))));
sheet.vault_id.clone_from(&binding.vault_id);
sheet.start_load();
}
sheet
}
/// Draws the sheet; returns what it asked the app to do.
pub fn show(&mut self, ctx: &Context, tokens: &Tokens, facts: &Facts<'_>) -> Option<SheetAction> {
self.poll();
let max_height = ctx.content_rect().height() * MAX_SHEET_HEIGHT;
let mut action = None;
let closed = crate::widgets::bottom_sheet::show(
ctx,
tokens,
&crate::widgets::bottom_sheet::Sheet {
id: egui::Id::new(SHEET_ID),
icon: crate::icons::image("refresh-cw", 16.0, egui::Color32::TRANSPARENT),
title: strings::title(),
chip: None,
},
|ui| {
ScrollArea::vertical()
.id_salt("notescraft-sync-sheet-scroll")
.auto_shrink([false, true])
.min_scrolled_height(max_height)
.max_height(max_height)
.show(ui, |ui| {
devices::status_card(ui, tokens, facts);
let mut sheet_action = None;
devices::section(self, ui, tokens, facts, &mut sheet_action);
options::section(ui, tokens, facts, &mut sheet_action);
if sheet_action.is_some() {
action = sheet_action;
}
});
},
);
if self.job.is_some() {
ctx.request_repaint_after(std::time::Duration::from_millis(LOAD_MILLIS));
}
if closed {
return Some(action.unwrap_or(SheetAction::Close));
}
action
}
/// Whether a load job is running.
#[must_use]
pub fn loading(&self) -> bool {
self.job.is_some()
}
/// Starts a load of the devices on a fresh worker thread; the client moves
/// in, so the token never reaches the screen.
fn start_load(&mut self) {
if self.job.is_some() {
return;
}
let Some(client) = self.client.clone() else { return };
let vault_id = self.vault_id.clone();
let (sender, receiver) = std::sync::mpsc::channel();
match std::thread::Builder::new().name("notescraft-sync-sheet".to_owned()).spawn(move || {
let outcome = match client.lock().unwrap_or_else(std::sync::PoisonError::into_inner).devices(&vault_id) {
Ok(devices) => DevicesOutcome::Listed(devices),
Err(error) => DevicesOutcome::Failed(error.to_string()),
};
let _ = sender.send(outcome);
}) {
Ok(_) => {
self.job = Some(Arc::new(Mutex::new(receiver)));
self.error = None;
}
Err(error) => self.error = Some(error.to_string()),
}
}
/// Takes the answer of the load job, when it is there.
fn poll(&mut self) {
let Some(job) = self.job.clone() else { return };
let outcome = {
let guard = job.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
match guard.try_recv() {
Ok(outcome) => Some(outcome),
Err(TryRecvError::Empty) => None,
Err(TryRecvError::Disconnected) => Some(DevicesOutcome::Failed(strings::pairing_failed().to_owned())),
}
};
match outcome {
None => {}
Some(DevicesOutcome::Listed(devices)) => {
self.job = None;
self.devices = devices;
self.error = None;
}
Some(DevicesOutcome::Failed(error)) => {
self.job = None;
self.error = Some(error);
}
}
}
}
/// The current Unix time in milliseconds; the one clock of the sheet.
#[must_use]
pub fn now_millis() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(u64::MAX))
}
@@ -0,0 +1,96 @@
//! The options half of the sync sheet: the three toggles and Sync now.
//!
//! The toggles report a flip; the shell stores it through `settings.set` and
//! the next frame's facts show it. The Sync now button is the primary action
//! of the sheet; it asks for `sync.now` and stays open.
use egui::{Align2, CornerRadius, FontId, Rect, Sense, Ui, Vec2, vec2};
use super::{Facts, SHEET_ID, SheetAction, SheetOption};
use crate::text::sync_sheet as strings;
use crate::theme::{Tokens, color};
/// Height of one toggle row, points.
const TOGGLE_ROW: f32 = 48.0;
/// Height of the Sync now button, points.
const BUTTON_ROW: f32 = 52.0;
/// Left pad inside a row, points.
const ROW_PAD: f32 = 16.0;
/// Radius of the Sync now card, points.
const BUTTON_RADIUS: u8 = 12;
/// Track of the switch: how wide and how high, points.
const TRACK: Vec2 = vec2(40.0, 22.0);
/// Radius of the switch's knob, points.
const KNOB: f32 = 9.0;
/// Space between a row and the button, points.
const BUTTON_GAP: f32 = 8.0;
/// The three rows and the primary button.
pub fn section(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option<SheetAction>) {
toggle(ui, tokens, SheetOption::WifiOnly, facts.options.wifi_only, action);
toggle(ui, tokens, SheetOption::Attachments, facts.options.attachments, action);
toggle(ui, tokens, SheetOption::SettingsAndThemes, facts.options.settings_and_themes, action);
ui.add_space(BUTTON_GAP);
button(ui, tokens, action);
}
/// The card of one option; `true` when this frame pressed it.
fn toggle(ui: &mut Ui, tokens: &Tokens, option: SheetOption, on: bool, action: &mut Option<SheetAction>) {
let colors = &tokens.theme.colors;
let label = match option {
SheetOption::WifiOnly => strings::wifi_only(),
SheetOption::Attachments => strings::attachments(),
SheetOption::SettingsAndThemes => strings::settings_and_themes(),
};
let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), TOGGLE_ROW), Sense::click());
hit.widget_info(|| egui::WidgetInfo::selected(egui::WidgetType::Checkbox, true, on, label));
let glow = crate::motion::hover(ui.ctx(), ui.id().with((SHEET_ID, "toggle", label)), &hit);
ui.painter().rect_filled(rect, CornerRadius::same(u8::MAX / 2), color(colors.selection).gamma_multiply(glow * 0.6));
ui.painter().text(egui::pos2(rect.left() + ROW_PAD, rect.center().y), Align2::LEFT_CENTER, label, FontId::proportional(15.0), color(colors.text));
paint_switch(ui, tokens, rect, on);
if hit.clicked() {
*action = Some(SheetAction::Toggle(option, !on));
}
}
/// The switch at the right end of a row (design: mauve track, dark knob).
fn paint_switch(ui: &Ui, tokens: &Tokens, row: Rect, on: bool) {
let colors = &tokens.theme.colors;
let track = Rect::from_center_size(egui::pos2(row.right() - ROW_PAD - TRACK.x / 2.0, row.center().y), TRACK);
ui.painter().rect_filled(track, CornerRadius::same(u8::MAX / 2), if on { color(colors.accent) } else { color(colors.surface_2) });
let knob_x = if on { track.right() - KNOB } else { track.left() + KNOB };
ui.painter().circle_filled(egui::pos2(knob_x, track.center().y), KNOB, color(colors.text_muted));
}
/// The Sync now button, full width, the accent behind the primary word.
fn button(ui: &mut Ui, tokens: &Tokens, action: &mut Option<SheetAction>) {
let colors = &tokens.theme.colors;
let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), BUTTON_ROW), Sense::click());
hit.widget_info(|| egui::WidgetInfo::labeled(egui::WidgetType::Button, true, strings::sync_now()));
let glow = crate::motion::hover(ui.ctx(), ui.id().with((SHEET_ID, "sync-now")), &hit);
let base = crate::motion::mix(color(colors.accent), color(colors.text), glow * 0.12);
ui.painter().rect_filled(rect, CornerRadius::same(BUTTON_RADIUS), base);
ui.painter().text(rect.center(), Align2::CENTER_CENTER, strings::sync_now(), FontId::new(16.0, crate::fonts::semibold()), color(colors.text));
if hit.clicked() {
*action = Some(SheetAction::SyncNow);
}
}
#[cfg(test)]
mod tests {
use super::*;
use notescraft_engine::SyncSettings;
#[test]
fn the_toggles_target_the_settings_keys() {
assert_eq!(SheetOption::WifiOnly.key(), "sync.wifi_only");
assert_eq!(SheetOption::Attachments.key(), "sync.attachments");
assert_eq!(SheetOption::SettingsAndThemes.key(), "sync.settings_and_themes");
}
#[test]
fn the_sync_settings_carry_their_defaults() {
assert_eq!(SyncSettings::default(), SyncSettings { wifi_only: true, attachments: true, settings_and_themes: false });
}
}
+3
View File
@@ -106,6 +106,8 @@ pub struct NotesApp {
pub quit_prompt: Option<SavePrompt>,
/// The Share dialog and its worker result, while it is open.
pub share_dialog: Option<crate::views::share_dialog::ShareDialog>,
/// The sync bottom sheet of the compact layout, while it is open.
pub sync_sheet: Option<crate::panels::sync_sheet::SyncSheet>,
/// PDF export dialog and its worker result receiver.
pub export_dialog: Option<crate::export_dialog::ExportDialog>,
pub(crate) export_result: Option<Receiver<ExportResult>>,
@@ -263,6 +265,7 @@ impl NotesApp {
quit_prompt: None,
export_dialog: None,
share_dialog: None,
sync_sheet: None,
export_result: None,
export_open_path: None,
right_tab: RightTab::default(),
+1 -1
View File
@@ -24,7 +24,7 @@ impl NotesApp {
if self.sync_next.is_none_or(|next| now >= next) {
self.sync_next = Some(now + STATUS_PERIOD);
self.ensure_sync_worker();
if self.tabs.active_path() == Some(crate::tabs::SETTINGS_PATH) {
if self.tabs.active_path() == Some(crate::tabs::SETTINGS_PATH) || self.sync_sheet.is_some() {
self.refresh_sync_status();
if self.sync_conflicts_next.is_none_or(|next| now >= next) {
self.sync_conflicts_next = Some(now + CONFLICTS_PERIOD);
+52 -1
View File
@@ -7,6 +7,7 @@
use egui::Key;
use serde_json::{Map, Value};
use crate::panels::sync_sheet::Facts;
use notescraft_engine::CommandInfo;
use crate::app::NotesApp;
@@ -236,9 +237,17 @@ impl NotesApp {
let edited = response.edited;
let now = ctx.input(|i| i.time);
self.handle_tree(tree_response, ctx);
// The sync pill asks for the sheet at phone width and for Settings > Sync
// otherwise; the request is a same-frame egui temp the pill wrote on click.
let sheet_requested = Self::take_sync_sheet_request(ctx);
if let Some(category) = response.settings_navigated {
self.cue(crate::sounds::Cue::Tick);
self.open_sync_page_if(category);
if !(sheet_requested && category == crate::views::settings_view::SettingsCategory::Sync) {
self.open_sync_page_if(category);
}
}
if sheet_requested {
self.sync_sheet = Some(crate::panels::sync_sheet::SyncSheet::open(&self.session));
}
self.apply_settings_changes(response.settings_changes.clone(), ctx);
self.apply_settings_actions(std::mem::take(&mut response.settings_actions));
@@ -359,6 +368,37 @@ impl NotesApp {
self.open_sync_page_if(crate::views::settings_view::SettingsCategory::Sync);
}
}
// The sync sheet over everything; the shell stores a toggle and answers the rows.
let sheet_action = if self.sync_sheet.is_some() {
let facts = Facts {
status: self.session.sync_worker().map(notescraft_engine::SyncWorker::status),
conflicts: self.preferences.view.sync.data.conflicts.len(),
options: &self.preferences.settings.sync,
now_ms: crate::panels::sync_sheet::now_millis(),
compact: crate::style::compact::is_compact(ctx),
};
let action = self.sync_sheet.as_mut().and_then(|sheet| sheet.show(ctx, &self.tokens, &facts));
let click = ctx.input(|input| input.pointer.interact_pos());
let rect = ctx.read_response(egui::Id::new("notescraft-sync-sheet")).map(|response| response.rect);
println!("DIAG sheet action {action:?} pointer {click:?} rect {rect:?}");
action
} else {
None
};
if let Some(action) = sheet_action {
match action {
crate::panels::sync_sheet::SheetAction::Close => self.sync_sheet = None,
crate::panels::sync_sheet::SheetAction::SyncNow => self.run_command("sync.now", &Value::Null),
crate::panels::sync_sheet::SheetAction::ReviewConflicts => {
self.sync_sheet = None;
self.open_sync_page_if(crate::views::settings_view::SettingsCategory::Sync);
}
crate::panels::sync_sheet::SheetAction::Toggle(option, on) => {
let params = serde_json::json!({ "key": option.key(), "value": on });
self.run_command("settings.set", &params);
}
}
}
crate::dialog::paint_closing(ctx, &self.tokens);
if self.export_open_path.is_some() {
egui::Area::new(egui::Id::new("export-open-action")).anchor(egui::Align2::RIGHT_BOTTOM, egui::vec2(-24.0, -24.0)).show(ctx, |ui| {
@@ -373,6 +413,17 @@ impl NotesApp {
self.flush_cues(now);
}
/// Whether the compact sync pill asked for the sheet this frame; the request
/// is read and cleared, so one click opens one sheet.
fn take_sync_sheet_request(ctx: &egui::Context) -> bool {
let id = egui::Id::new("sync-sheet-request");
ctx.data_mut(|data| {
let asked = data.get_temp::<bool>(id).unwrap_or(false);
data.remove::<bool>(id);
asked
})
}
/// Runs a chosen engine/UI command; commands with required parameters toast instead.
/// `ui.save` is intercepted: saving needs the session and the tab state.
pub(in crate::shell::app) fn run_palette_choice(&mut self, id: &str, ctx: &egui::Context) {
+1 -1
View File
@@ -32,7 +32,7 @@ fn lookup(key: &'static str) -> &'static str {
pub mod app;
mod features;
pub use features::{compact, launcher, live};
pub use features::{compact, launcher, live, sync_sheet};
mod panels;
mod settings;
mod views;
+2 -1
View File
@@ -1,5 +1,6 @@
//! Copy for the newer feature screens: compact layout, vault launcher, live editing.
//! Copy for the newer feature screens: compact layout, vault launcher, live editing and the sync sheet.
pub mod compact;
pub mod launcher;
pub mod live;
pub mod sync_sheet;
@@ -0,0 +1,28 @@
//! Copy for the sync sheet (design `03-sync-sheet`): status, devices, the
//! conflict row, the three options and Sync now.
table! {
title = "ui.sync-sheet.title";
status_up_to_date = "ui.sync-sheet.status-up-to-date";
status_syncing = "ui.sync-sheet.status-syncing";
status_offline = "ui.sync-sheet.status-offline";
status_sign_in = "ui.sync-sheet.status-sign-in";
status_error = "ui.sync-sheet.status-error";
last_sync = "ui.sync-sheet.last-sync";
never_synced = "ui.sync-sheet.never-synced";
devices = "ui.sync-sheet.devices";
this_phone = "ui.sync-sheet.this-phone";
this_computer = "ui.sync-sheet.this-computer";
last_seen = "ui.sync-sheet.last-seen";
never_seen = "ui.sync-sheet.never-seen";
chip_online = "ui.sync-sheet.chip-online";
chip_synced = "ui.sync-sheet.chip-synced";
chip_stale = "ui.sync-sheet.chip-stale";
conflicts = "ui.sync-sheet.conflicts";
wifi_only = "ui.sync-sheet.wifi-only";
attachments = "ui.sync-sheet.attachments";
settings_and_themes = "ui.sync-sheet.settings-and-themes";
sync_now = "ui.sync-sheet.sync-now";
loading = "ui.sync-sheet.loading";
pairing_failed = "ui.sync-sheet.pairing-failed";
}
+430
View File
@@ -0,0 +1,430 @@
//! The sync sheet driven the way a user does: real frames at 390 x 844 over a
//! vault bound to a stub server (design `03-sync-sheet`).
use egui::{Context, Event, Key, PointerButton, Pos2, RawInput};
use notescraft_engine::{Session, Timings, bind_folder};
use notescraft_ui_egui::app::NotesApp;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
static NEXT: AtomicU64 = AtomicU64::new(0);
/// The window of a phone.
const PHONE: egui::Vec2 = egui::vec2(390.0, 844.0);
/// The window of a desktop.
const DESKTOP: egui::Vec2 = egui::vec2(1280.0, 800.0);
/// The token of the bound vault; it must never reach the screen.
const TOKEN: &str = "secret-token";
/// Longest the tests wait for a stub request or an idle worker.
const WAIT: Duration = Duration::from_secs(5);
/// One request the stub received.
#[derive(Debug, Clone)]
struct Seen {
/// The HTTP method.
method: String,
/// The path of the request line, without the query.
path: String,
}
/// A stub server: canned answers per path and method, and every request it saw.
struct Stub {
base: String,
seen: Arc<Mutex<Vec<Seen>>>,
stop: Arc<AtomicBool>,
}
impl Stub {
/// Starts a stub on an ephemeral port. `devices` is the body of the device
/// list; `/changes` answers an empty log and `/files/a.md` answers applied,
/// so the worker's cycle lands on "Synced".
fn spawn(devices: &str) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind the stub");
let port = listener.local_addr().expect("addr").port();
let record = r#"{"path":"a.md","hash":"x","size":2,"mtime_ms":0,"seq":1,"deleted":false}"#;
let empty = r#"{"head":0,"records":[],"more":false}"#;
let answer = |status: &str, body: &str| format!("HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len());
let answers: Vec<(String, String, String)> = vec![
("GET".to_owned(), "/changes".to_owned(), answer("200 OK", empty)),
("PUT".to_owned(), "/files/a.md".to_owned(), answer("201 Created", record)),
("GET".to_owned(), "/devices".to_owned(), answer("200 OK", devices)),
];
let seen: Arc<Mutex<Vec<Seen>>> = Arc::new(Mutex::new(Vec::new()));
let stop = Arc::new(AtomicBool::new(false));
let (thread_seen, thread_stop) = (Arc::clone(&seen), Arc::clone(&stop));
std::thread::spawn(move || {
while !thread_stop.load(Ordering::Relaxed) {
match listener.accept() {
Ok((stream, _)) => {
// One thread per connection: the client may keep the connection
// open, and the listener must go on accepting.
let Ok(mut writer) = stream.try_clone() else { continue };
let mut reader = BufReader::new(stream);
let answers = Arc::new(answers.clone());
let seen = Arc::clone(&thread_seen);
std::thread::spawn(move || {
while let Some(request) = read_request(&mut reader) {
seen.lock().expect("seen").push(request.clone());
let hit = answers.iter().find(|(method, path, _)| *method == request.method && request.path.contains(path.as_str()));
let canned = hit.map_or("HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n", |(_, _, canned)| canned.as_str());
let _ = writer.write_all(canned.as_bytes());
let _ = writer.flush();
}
});
}
Err(_) => std::thread::sleep(Duration::from_millis(2)),
}
}
});
Self { base: format!("http://127.0.0.1:{port}"), seen, stop }
}
/// How many times the change log was polled.
fn change_polls(&self) -> u32 {
self.seen.lock().expect("seen").iter().filter(|seen| seen.method == "GET" && seen.path.ends_with("/changes")).count().try_into().unwrap_or(u32::MAX)
}
/// The requests the stub saw.
fn requests(&self) -> Vec<Seen> {
self.seen.lock().expect("seen").clone()
}
/// Whether the stub saw a write to `path`.
fn uploaded(&self, path: &str) -> bool {
self.requests().iter().any(|seen| seen.method == "PUT" && seen.path.ends_with(path))
}
/// Waits, bounded, until `wanted` holds.
fn wait_for(&self, wanted: impl Fn(&[Seen]) -> bool) {
let deadline = Instant::now() + WAIT;
while Instant::now() < deadline && !wanted(&self.requests()) {
std::thread::sleep(Duration::from_millis(5));
}
}
}
impl Drop for Stub {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
}
}
/// Reads one request: method, path, and its body skipped; the request ends when
/// the client closes the connection.
fn read_request(reader: &mut BufReader<TcpStream>) -> Option<Seen> {
let mut line = String::new();
reader.read_line(&mut line).ok()?;
let mut parts = line.split_whitespace();
let method = parts.next()?.to_owned();
let path = parts.next()?.to_owned();
let mut length = 0_usize;
loop {
let mut header = String::new();
if reader.read_line(&mut header).ok()? == 0 {
break;
}
if header.trim_end().is_empty() {
break;
}
if let Some((name, value)) = header.trim_end().split_once(':')
&& name.trim().eq_ignore_ascii_case("content-length")
{
length = value.trim().parse().unwrap_or(0);
}
}
let mut body = vec![0_u8; length];
if length > 0 {
reader.read_exact(&mut body).ok()?;
}
Some(Seen { method, path: path.split('?').next().unwrap_or_default().to_owned() })
}
struct Driver {
app: NotesApp,
ctx: Context,
time: f64,
size: egui::Vec2,
}
/// A painted text: what it says, where it starts and its galley.
#[derive(Debug)]
struct Painted {
text: String,
pos: Pos2,
galley: Arc<egui::Galley>,
}
/// An app over a fresh vault bound to `stub`, with a note open; the settings
/// dir lives beside the vault, never inside it (the vault is the sync scope).
fn driven(stub: &Stub, files: &[(&str, &str)], size: egui::Vec2) -> Driver {
let root = std::env::temp_dir().join(format!("notescraft-sheet-{}-{}", std::process::id(), NEXT.fetch_add(1, Ordering::Relaxed)));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).expect("root");
seed(&root, files);
bind_folder(&root, &stub.base, "v-1", TOKEN, "d-1", Some("u-1")).expect("bind the vault");
app_for(&root, &settings_dir(&root), size)
}
/// Writes `(path, content)` into `root`, creating the folders.
fn seed(root: &Path, files: &[(&str, &str)]) {
for (path, text) in files {
if let Some((folder, _)) = path.rsplit_once('/') {
std::fs::create_dir_all(root.join(folder)).expect("folder");
}
std::fs::write(root.join(path), text).expect("seed");
}
}
/// The settings dir beside `root`; the vault never holds the settings (SPEC 4.1).
fn settings_dir(root: &Path) -> std::path::PathBuf {
root.parent().unwrap_or(root).join("notescraft-sheet-config")
}
/// An app over an already seeded and bound vault at `root`.
fn app_for(root: &Path, config: &Path, size: egui::Vec2) -> Driver {
let mut session = Session::open(root).expect("session");
session.use_settings_dir(config);
let mut app = NotesApp::new(session, notescraft_themes::builtin("catppuccin-mocha").expect("theme"));
app.state.animations = false;
app.open_note("a.md");
Driver { app, ctx: Context::default(), time: 1.0, size }
}
impl Driver {
/// One frame with `events`; the texts it painted.
fn frame(&mut self, events: Vec<Event>) -> Vec<Painted> {
self.time += 0.05;
let input = RawInput { screen_rect: Some(egui::Rect::from_min_size(Pos2::ZERO, self.size)), time: Some(self.time), events, ..Default::default() };
let app = &mut self.app;
let mut out = self.ctx.run_ui(input, |ui| app.ui(ui.ctx()));
out.textures_delta.clear();
out.shapes
.iter()
.filter_map(|shape| match &shape.shape {
egui::Shape::Text(text) => Some(Painted { text: text.galley.text().to_owned(), pos: text.pos, galley: text.galley.clone() }),
_ => None,
})
.collect()
}
fn settle(&mut self) -> Vec<Painted> {
let mut last = Vec::new();
for _ in 0..24 {
last = self.frame(Vec::new());
}
last
}
fn click(&mut self, at: Pos2) -> Vec<Painted> {
self.frame(vec![Event::PointerMoved(at)]);
self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: true, modifiers: egui::Modifiers::NONE }]);
self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: false, modifiers: egui::Modifiers::NONE }]);
self.settle()
}
/// The centre of the pill's painted `Synced` label; the hit area holds it.
fn pill_centre(&mut self) -> Pos2 {
let texts = self.settle();
let status = self.app.session.sync_worker().map(|worker| format!("{:?}", worker.status()));
let pill = texts
.iter()
.find(|painted| painted.text == "Synced")
.unwrap_or_else(|| panic!("no pill among {:?} -- worker {status:?}", texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()));
pill.pos + pill.galley.size() / 2.0
}
}
fn painted<'a>(texts: &'a [Painted], wanted: &str) -> Option<&'a Painted> {
texts.iter().find(|painted| painted.text == wanted)
}
fn expect<'a>(texts: &'a [Painted], wanted: &str) -> &'a Painted {
painted(texts, wanted).unwrap_or_else(|| panic!("no {wanted:?} among {:?}", texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()))
}
/// The device list the stub answers: another device last seen an hour ago,
/// this device now.
fn devices_json(now_ms: u64) -> String {
format!(
r#"[{{"id":"d-2","name":"Laptop","created_ms":0,"last_seen_ms":{},"current":false}},{{"id":"d-1","name":"Pixel 9","created_ms":0,"last_seen_ms":{},"current":true}}]"#,
now_ms - 3_600_000,
now_ms
)
}
/// The Unix time now, in milliseconds.
fn now_ms() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(u64::MAX))
}
/// The worker runs fast: the tests never wait for the real debounce.
fn fast_worker(d: &mut Driver) {
if let Some(worker) = d.app.session.sync_worker() {
worker.set_timings(Timings {
debounce: Duration::from_millis(10),
poll: Duration::from_mins(15),
backoff: Duration::from_millis(10),
backoff_max: Duration::from_secs(1),
});
}
}
/// Waits, bounded, until the worker is idle; the cycles land before the frames read it.
fn worker_idle(d: &mut Driver) {
let deadline = Instant::now() + WAIT;
while Instant::now() < deadline && !d.app.session.sync_worker().is_some_and(|worker| worker.status().phase == notescraft_engine::SyncPhase::Idle) {
std::thread::sleep(Duration::from_millis(5));
}
}
#[test]
fn the_pill_opens_the_sheet_and_lists_the_current_device_first() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut d = driven(&stub, &[("a.md", "A")], PHONE);
d.settle();
fast_worker(&mut d);
stub.wait_for(|requests| requests.iter().any(|seen| seen.method == "PUT" && seen.path.ends_with("a.md")));
worker_idle(&mut d);
let from = d.pill_centre();
let texts = d.click(from);
assert!(d.app.sync_sheet.is_some(), "the pill opened the sheet");
let current = expect(&texts, "This phone");
let laptop = expect(&texts, "Laptop");
assert!(current.pos.y < laptop.pos.y, "the current device first: {current:?} vs {laptop:?}");
assert!(
texts.iter().any(|painted| painted.text.contains("last seen")),
"the ages show: {:?}",
texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()
);
assert!(!texts.iter().any(|painted| painted.text.contains(TOKEN)), "the token never reaches the screen");
assert!(painted(&texts, "Everything is up to date").is_some(), "the status headline shows");
assert!(painted(&texts, "Sync now").is_some(), "the primary button shows");
}
#[test]
fn the_conflicts_row_appears_only_over_a_conflict_file_on_disk() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut with = driven(&stub, &[("a.md", "A"), ("a (conflict Laptop 2026-10-20).md", "A")], PHONE);
with.settle();
let from = with.pill_centre();
with.click(from);
let over = with.settle();
assert!(painted(&over, "1 conflicts to review").is_some(), "the row shows");
drop(with);
let mut none_driver = driven(&stub, &[("a.md", "A")], PHONE);
none_driver.settle();
let from = none_driver.pill_centre();
none_driver.click(from);
let none = none_driver.settle();
assert!(painted(&none, "conflicts to review").is_none(), "nothing to review: {:?}", none.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>());
let _ = none_driver.app.sync_sheet.take();
}
#[test]
fn toggling_attachments_persists_and_the_first_cycle_skips_the_png() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut d = driven(&stub, &[("a.md", "A"), ("img/scan.png", "bytes")], PHONE);
d.settle();
worker_idle(&mut d);
let from = d.pill_centre();
d.click(from);
let texts = d.settle();
let label = expect(&texts, "Sync attachments");
let hit = label.pos + label.galley.size() / 2.0;
d.click(hit);
// The stored option goes into the settings file the next load reads; the
// vault root and its settings dir outlive the app they were opened over.
let root = d.app.session.vault().root().to_path_buf();
let config = settings_dir(&root);
let read = |text: &str| text.lines().filter(|line| line.contains("attachments") || line.contains("[sync]")).collect::<Vec<_>>().join(", ");
let stored = std::fs::read_to_string(config.join("config.toml")).unwrap_or_default();
println!("DIAG toml: {:?}", read(&stored));
assert!(stored.contains("attachments = false"), "the toggle stored: {:?}", read(&stored));
drop(d);
let mut again = app_for(&root, &config, PHONE);
again.settle();
fast_worker(&mut again);
worker_idle(&mut again);
assert!(!stub.uploaded("scan.png"), "the png never left: {}", stub_uploaded_list(&stub));
assert!(again.app.session.sync_worker().is_some_and(|worker| worker.status().phase == notescraft_engine::SyncPhase::Idle), "the cycle ran");
drop(again);
let _ = std::fs::remove_dir_all(&root);
let _ = std::fs::remove_dir_all(&config);
}
/// The paths the stub saw written to, for the failure messages.
fn stub_uploaded_list(stub: &Stub) -> String {
stub.requests().iter().map(|seen| format!("{}/{}", seen.method, seen.path)).collect::<Vec<_>>().join(", ")
}
#[test]
fn sync_now_wakes_the_worker() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut d = driven(&stub, &[("a.md", "A")], PHONE);
d.settle();
fast_worker(&mut d);
stub.wait_for(|requests| requests.iter().any(|seen| seen.method == "PUT" && seen.path.ends_with("a.md")));
worker_idle(&mut d);
let before = stub.change_polls();
let from = d.pill_centre();
d.click(from);
let texts = d.settle();
let label = expect(&texts, "Sync now");
let hit = label.pos + label.galley.size() / 2.0;
d.click(hit);
stub.wait_for(|requests| requests.iter().filter(|seen| seen.method == "GET" && seen.path.ends_with("/changes")).count() > before as usize);
assert!(stub.change_polls() > before, "the change log was polled again");
assert!(d.app.sync_sheet.is_some(), "Sync now keeps the sheet open");
}
#[test]
fn the_sheet_closes_with_x_and_escape_and_a_tap_outside() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut d = driven(&stub, &[("a.md", "A")], PHONE);
d.settle();
let from = d.pill_centre();
d.click(from);
assert!(d.app.sync_sheet.is_some(), "open");
d.settle();
// The × sits in the title row, in the top-right of the sheet.
let sheet = d.ctx.read_response(egui::Id::new("notescraft-sync-sheet")).expect("the sheet area");
let rect = sheet.rect;
d.click(egui::pos2(rect.right() - 16.0 - 22.0, rect.top() + 16.0 + 4.0 + 12.0 + 22.0));
assert!(d.app.sync_sheet.is_none(), "the × closed the sheet");
let again_from = d.pill_centre();
d.click(again_from);
assert!(d.app.sync_sheet.is_some(), "open again");
let escape = Event::Key { key: Key::Escape, physical_key: None, pressed: true, repeat: false, modifiers: egui::Modifiers::NONE };
d.frame(vec![escape]);
d.settle();
assert!(d.app.sync_sheet.is_none(), "Escape closed the sheet");
let third_from = d.pill_centre();
d.click(third_from);
assert!(d.app.sync_sheet.is_some(), "open again for the tap outside");
d.click(egui::pos2(20.0, 20.0));
assert!(d.app.sync_sheet.is_none(), "a tap on the dimmed area closed the sheet");
}
#[test]
fn at_desktop_width_the_pill_opens_the_settings_page() {
let stub = Stub::spawn(&devices_json(now_ms()));
let mut d = driven(&stub, &[("a.md", "A")], DESKTOP);
d.settle();
let from = d.pill_centre();
let texts = d.click(from);
assert!(d.app.sync_sheet.is_none(), "no sheet at desktop width");
assert_eq!(d.app.tabs.active_path(), Some(notescraft_ui_egui::tabs::SETTINGS_PATH), "the settings page opened");
assert!(
texts.iter().any(|painted| painted.text == "Server") && texts.iter().any(|painted| painted.text == "Sync now"),
"the Sync page shows: {:?}",
texts.iter().map(|painted| painted.text.as_str()).collect::<Vec<_>>()
);
}
+2 -2
View File
@@ -30,7 +30,7 @@ Real-time delta sync of Markdown (files are small), cross-vault dedup in encrypt
2. Content-defined chunking (rolling hash) instead of fixed 4 MiB? Recommended: fixed first; revisit if edits to big files are common.
## Tasks
1. Server: `chunks` table and endpoints, assembled-file reads, quota accounting.
1. [x] Server: `chunks` table and endpoints, assembled-file reads, quota accounting. Done 2026-10-20 by GLM 5.3 Flash via Pi.
2. [x] Server: `file_revisions`, history, restore, retention GC. Done 2026-10-20 by GLM 5.3 Flash via Pi (MiniMax was out of quota): `src/history/`, `store/revisions.rs`, endpoints `files-history`, `files/{path}?seq=`, `files-restore`, per-vault `retention_days` (default 30, newest 20 always kept), quota counts retained revisions; verified by the orchestrator (all server suites green, clippy clean). Engine and UI still open (task 4).
3. Engine: chunked upload/download in the driver with resume and progress events.
3. [x] Engine: chunked upload in the driver with resume and progress events (`sync/chunks.rs`, `Remote::{missing_chunks, put_chunk, put_chunked}`, `SyncStatus.transfer`). Done 2026-10-20 by MiniMax M3.1 Flash through the Claude Code harness; verified against the real server. Big-file download still goes through the existing `get` (capped by what fits in memory).
4. UI: transfer progress in the status pill and Settings > Sync; Versions tab shows server revisions.
+1 -1
View File
@@ -53,7 +53,7 @@ Rules: each task is independently testable; order is execution order. `[x]` done
- [ ] 25. Quick capture sheet and share-target receive (`note.new_from_share` command).
- [ ] 26. Quick switcher and command palette on phone, over the engine registry.
- [ ] 27. Background sync via WorkManager and `BGTaskScheduler` (the worker from task 5 runs headless).
- [ ] 28. Conflict resolution UI plus an offline queue view on phone (builds on task 6).
- [~] 28. Conflict resolution UI plus an offline queue view on phone (builds on task 6). The phone sync sheet with devices, conflicts row, options and Sync now is done (2026-10-20, GLM 5.3 Flash, 6 driven tests); the offline queue view is open.
- [ ] 29. Mobile toolbar above the keyboard and long-press context menus (design 02).
## Related spec sets