181aad91e6
Sync and the server - crates/sync: revision ledger (stale writers conflict, edit beats delete), client planner, three-way merge, conflict names, path and hash gates - apps/notescraft-sync-server: pairing, device tokens, change log with long poll, blobs, accounts with many vaults, roles, invites, audit log, quota, GC, readiness, request ids - server history (revisions, restore, retention), chunked resumable uploads, live editing endpoints (SSE rooms, presence, flush through the ledger) - engine: HttpRemote, sync_once, SyncWorker, sync.* commands, share.* and server.* commands, live session client, note.capture, vault catalog and vaults.* commands Editing together - crates/collab: operational transformation core (TextOp, transform, client state machine, room, presence) - ui-egui: Settings > Sync, sync pill, conflicts, Share dialog, live pill and remote carets, launcher Connect and Join flows App shell - vault launcher (desktop and phone layouts) wired into the app host (--launcher, File > Vaults...) - compact (phone) layout with a bottom tab bar Verified with cargo xtask ci (fmt, clippy -D warnings, workspace tests, layers, tree, i18n). Android and iOS shells are not built: the build machine has no NDK or Xcode. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
349 lines
18 KiB
Rust
349 lines
18 KiB
Rust
//! Chunked, resumable, deduplicated uploads: `PUT /chunks/{hash}`, `POST /chunks/check`
|
|
//! and `PUT /files/{path}` with `X-Chunks`. Real HTTP with shrunken limits.
|
|
|
|
mod common;
|
|
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use common::{ADMIN, Server, temp_dir};
|
|
use notescraft_sync::hash_hex;
|
|
use notescraft_sync_server::{AppState, Store, TOMBSTONE_GRACE_MS, now_ms};
|
|
use serde_json::{Value, json};
|
|
|
|
/// Tests upload 1 KiB chunks instead of the 4 MiB default.
|
|
const CHUNK_MAX: usize = 1024;
|
|
/// Files may reach 8 KiB in tests instead of the 1 GiB default.
|
|
const FILE_CAP: u64 = 8 * 1024;
|
|
/// A 64 hex digit shape for hostile or placeholder hashes.
|
|
fn hex_zeros(n: usize) -> String {
|
|
"0".repeat(n)
|
|
}
|
|
|
|
fn chunks_server(name: &str, quota: Option<u64>) -> (Server, PathBuf) {
|
|
let data = temp_dir(name);
|
|
let state = AppState::with_quota(Store::open(&data.join("sync.db")).expect("db"), data.join("blobs"), ADMIN, quota);
|
|
state.set_chunk_limits(CHUNK_MAX, FILE_CAP);
|
|
(common::start_with_state(state), data)
|
|
}
|
|
|
|
/// `len` bytes of one value, so chunks of different seeds differ.
|
|
fn chunk_data(seed: u8, len: usize) -> Vec<u8> {
|
|
vec![seed; len]
|
|
}
|
|
|
|
/// The chunk files of a vault, by name.
|
|
fn chunk_names(data: &Path, vault: &str) -> Vec<String> {
|
|
let dir = data.join("blobs").join(vault).join("chunks");
|
|
let mut names: Vec<String> = std::fs::read_dir(&dir)
|
|
.map_or_else(|_| Vec::new(), |entries| entries.filter_map(|entry| entry.ok().and_then(|e| e.file_name().into_string().ok())).collect());
|
|
names.sort();
|
|
names
|
|
}
|
|
|
|
/// `PUT /chunks/{hash}`: the status.
|
|
fn put_chunk(server: &Server, vault: &str, token: &str, hash: &str, body: &[u8]) -> u16 {
|
|
server.send("PUT", &format!("/v1/vaults/{vault}/chunks/{hash}"), Some(token), &[], Some(body)).0
|
|
}
|
|
|
|
/// Uploads the seeds as [`CHUNK_MAX`]-sized chunks; answers their joined bytes and hashes.
|
|
fn upload_chunks(server: &Server, vault: &str, token: &str, seeds: &[u8]) -> (Vec<u8>, Vec<String>) {
|
|
let mut content = Vec::new();
|
|
let mut hashes = Vec::new();
|
|
for seed in seeds {
|
|
let bytes = chunk_data(*seed, CHUNK_MAX);
|
|
assert_eq!(put_chunk(server, vault, token, &hash_hex(&bytes), &bytes), 201);
|
|
hashes.push(hash_hex(&bytes));
|
|
content.extend_from_slice(&bytes);
|
|
}
|
|
(content, hashes)
|
|
}
|
|
|
|
/// `POST /chunks/check`: `(status, missing)`.
|
|
fn check_chunks(server: &Server, vault: &str, token: &str, hashes: &[String]) -> (u16, Vec<String>) {
|
|
let (status, reply) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(token), &[], Some(&json!({"hashes": hashes})));
|
|
let missing = reply["missing"].as_array().map_or_else(Vec::new, |rows| rows.iter().filter_map(|v| v.as_str().map(str::to_owned)).collect());
|
|
(status, missing)
|
|
}
|
|
|
|
/// `PUT /files/{path}` with `X-Chunks`: `(status, reply body)`.
|
|
fn assemble(server: &Server, vault: &str, token: &str, path: &str, hashes: &[String], base_seq: u64, content_hash: Option<&str>) -> (u16, Value) {
|
|
let joined = hashes.join(",");
|
|
let base = base_seq.to_string();
|
|
let mut named: Vec<(&str, &str)> = vec![("X-Chunks", joined.as_str()), ("X-Base-Seq", base.as_str())];
|
|
named.extend(content_hash.map(|hash| ("X-Content-Hash", hash)));
|
|
let (status, reply) = server.send("PUT", &format!("/v1/vaults/{vault}/files/{path}"), Some(token), &named, None);
|
|
(status, serde_json::from_slice(&reply).unwrap_or(Value::Null))
|
|
}
|
|
|
|
#[test]
|
|
fn chunks_upload_idempotently_assemble_and_read_back_exactly() {
|
|
let (server, data) = chunks_server("chunks", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let (content, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3, 4, 5]);
|
|
assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "five chunks stored once each");
|
|
|
|
// Everything is there; the resume endpoint answers complete.
|
|
let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
|
|
assert_eq!((status, missing), (200, Vec::<String>::new()));
|
|
|
|
let expected_hash = hash_hex(&content);
|
|
let (status, record) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, None);
|
|
assert_eq!(status, 201);
|
|
assert_eq!(record["hash"].as_str(), Some(expected_hash.as_str()));
|
|
assert_eq!(record["size"].as_u64(), Some(u64::try_from(content.len()).expect("small")));
|
|
let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
|
|
assert_eq!((status, bytes.as_slice()), (200, content.as_slice()));
|
|
|
|
// A second assembly over a fresh path reuses the chunk files.
|
|
let (status, _) = assemble(&server, &vault, &token, "same.bin", &hashes, 0, None);
|
|
assert_eq!(status, 201, "a fresh path is a new file");
|
|
assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "chunks untouched by assembly");
|
|
}
|
|
|
|
/// The hashes a folder holds, ascending, for comparisons with `chunk_names`.
|
|
fn sorted(hashes: &[String]) -> Vec<String> {
|
|
let mut sorted = hashes.to_vec();
|
|
sorted.sort();
|
|
sorted
|
|
}
|
|
|
|
/// `PUT /files/{path}` with an `X-Chunks` header through a raw HTTP connection, closed
|
|
/// after the answer; `(status, body)`.
|
|
fn raw_put(server: &Server, token: &str, vault: &str, path: &str, chunks: &str) -> (u16, String) {
|
|
use std::io::{Read, Write};
|
|
let port: u16 = server.base.strip_prefix("http://127.0.0.1:").and_then(|raw| raw.parse().ok()).expect("port");
|
|
let request = format!(
|
|
"PUT /v1/vaults/{vault}/files{path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAuthorization: Bearer {token}\r\nX-Chunks: {chunks}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
|
|
);
|
|
let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).expect("connect");
|
|
stream.write_all(request.as_bytes()).expect("write");
|
|
let mut reply = Vec::new();
|
|
stream.read_to_end(&mut reply).expect("read");
|
|
let text = String::from_utf8_lossy(&reply).into_owned();
|
|
let code = text.split_whitespace().nth(1).and_then(|word| word.parse().ok()).unwrap_or(0);
|
|
(code, text)
|
|
}
|
|
|
|
#[test]
|
|
fn resume_after_a_cut_connection_only_sends_missing_chunks() {
|
|
let (server, _) = chunks_server("resume", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let seeds = [1u8, 2, 3, 4, 5];
|
|
let hashes: Vec<String> = seeds.iter().map(|seed| hash_hex(&chunk_data(*seed, CHUNK_MAX))).collect();
|
|
let content: Vec<u8> = seeds.iter().flat_map(|seed| chunk_data(*seed, CHUNK_MAX)).collect();
|
|
|
|
// Three of five arrived before the cut.
|
|
for (index, seed) in seeds.iter().enumerate().take(3) {
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hashes[index], &chunk_data(*seed, CHUNK_MAX)), 201);
|
|
}
|
|
let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
|
|
assert_eq!((status, missing), (200, hashes[3..5].to_vec()));
|
|
|
|
// The resumed connection sends only what is missing, then assembles.
|
|
for (index, seed) in seeds.iter().enumerate().skip(3) {
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hashes[index], &chunk_data(*seed, CHUNK_MAX)), 201);
|
|
}
|
|
assert_eq!(assemble(&server, &vault, &token, "big.bin", &hashes, 0, None).0, 201);
|
|
let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
|
|
assert_eq!((status, bytes.as_slice()), (200, content.as_slice()));
|
|
}
|
|
|
|
#[test]
|
|
fn a_chunk_whose_hash_does_not_match_is_refused_and_stores_nothing() {
|
|
let (server, data) = chunks_server("badhash", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let body = chunk_data(1, 100);
|
|
for wrong in ["a".repeat(64), "A".repeat(64), "0".repeat(63), String::from("x")] {
|
|
let (status, _) = server.send("PUT", &format!("/v1/vaults/{vault}/chunks/{wrong}"), Some(&token), &[], Some(body.as_slice()));
|
|
assert_eq!(status, 400, "hash {wrong}");
|
|
}
|
|
// A well formed hash that does not match the body.
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(b"other"), &body), 400);
|
|
// The empty body is refused as well.
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&body), b""), 400);
|
|
assert_eq!(chunk_names(&data, &vault), Vec::<String>::new(), "nothing was stored");
|
|
}
|
|
|
|
#[test]
|
|
fn an_oversized_chunk_is_refused() {
|
|
let (server, data) = chunks_server("bigchunk", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let body = chunk_data(1, CHUNK_MAX + 1);
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hex_zeros(64), &body), 413);
|
|
assert_eq!(chunk_names(&data, &vault), Vec::<String>::new());
|
|
}
|
|
|
|
#[test]
|
|
fn assembling_with_a_missing_chunk_answers_409_and_makes_no_file() {
|
|
let (server, _) = chunks_server("missing", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let bytes = chunk_data(1, CHUNK_MAX);
|
|
let hash = hash_hex(&bytes);
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hash, &bytes), 201);
|
|
let unknown = "f".repeat(64);
|
|
let (status, reply) = assemble(&server, &vault, &token, "big.bin", &[hash, unknown.clone()], 0, None);
|
|
assert_eq!(status, 409);
|
|
assert_eq!(reply["missing"], json!([unknown]), "the absent hash is named");
|
|
assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None).0, 404);
|
|
}
|
|
|
|
#[test]
|
|
fn assembling_with_a_wrong_content_hash_is_400() {
|
|
let (server, _) = chunks_server("wrongcontent", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2]);
|
|
let (status, _) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, Some(&"f".repeat(64)));
|
|
assert_eq!(status, 400, "the header must match the computed hash");
|
|
let content: Vec<u8> = chunk_data(1, CHUNK_MAX).into_iter().chain(chunk_data(2, CHUNK_MAX)).collect();
|
|
let (status, record) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, Some(&hash_hex(&content)));
|
|
assert_eq!((status, record["size"].as_u64()), (201, Some(u64::try_from(content.len()).expect("small"))));
|
|
}
|
|
|
|
#[test]
|
|
fn an_assembly_over_the_file_cap_is_413() {
|
|
let (server, _) = chunks_server("filecap", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let seeds: Vec<u8> = (1..=10).collect();
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &seeds);
|
|
let (status, reply) = assemble(&server, &vault, &token, "huge.bin", &hashes, 0, None);
|
|
assert_eq!(status, 413, "10 KiB of chunks exceed the 8 KiB file cap");
|
|
assert!(reply["error"].is_string());
|
|
assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/huge.bin"), Some(&token), &[], None).0, 404);
|
|
}
|
|
|
|
#[test]
|
|
fn the_second_file_sharing_chunks_stores_them_once() {
|
|
let (server, data) = chunks_server("dedup", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3]);
|
|
assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()));
|
|
|
|
// A second file with the same chunks in a different order: a different file hash,
|
|
// and the chunk folder does not grow.
|
|
let reversed: Vec<String> = hashes.iter().rev().cloned().collect();
|
|
assert_eq!(assemble(&server, &vault, &token, "a.bin", &hashes, 0, None).0, 201);
|
|
assert_eq!(assemble(&server, &vault, &token, "b.bin", &reversed, 0, None).0, 201);
|
|
assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "three chunks, two files");
|
|
|
|
// Re-uploading a stored chunk is idempotent.
|
|
let bytes = chunk_data(1, CHUNK_MAX);
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&bytes), &bytes), 200, "already there");
|
|
assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()));
|
|
}
|
|
#[test]
|
|
fn a_viewer_cannot_upload_chunks_or_assemble() {
|
|
let (server, _) = chunks_server("viewer", None);
|
|
let (vault, owner) = server.paired("pc");
|
|
let (_, viewer) = server.paired("vw");
|
|
let (status, code) = server.invite(&vault, &owner, "viewer");
|
|
assert_eq!(status, 200);
|
|
assert_eq!(server.accept(&viewer, &code).0, 200);
|
|
|
|
let bytes = chunk_data(1, 100);
|
|
assert_eq!(put_chunk(&server, &vault, &viewer, &hash_hex(&bytes), &bytes), 403);
|
|
let (_, hashes) = upload_chunks(&server, &vault, &owner, &[1, 2]);
|
|
assert_eq!(assemble(&server, &vault, &viewer, "big.bin", &hashes, 0, None).0, 403, "viewer assembly refused");
|
|
assert_eq!(assemble(&server, &vault, &owner, "big.bin", &hashes, 0, None).0, 201);
|
|
}
|
|
|
|
#[test]
|
|
fn the_quota_refuses_a_chunk_and_an_assembly() {
|
|
// A chunk larger than what the quota still allows: refused at upload.
|
|
let (server, _) = chunks_server("quota-chunk", Some(512));
|
|
let (vault, token) = server.paired("pc");
|
|
let bytes = chunk_data(1, CHUNK_MAX);
|
|
assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&bytes), &bytes), 413);
|
|
assert!(server.audit().iter().any(|row| row["kind"] == "quota_rejected"));
|
|
|
|
// Two chunks fit (their upload counts them), the assembly does not.
|
|
let (server, _) = chunks_server("quota-assemble", Some(1536));
|
|
let (vault, token) = server.paired("pc");
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2]);
|
|
let (status, reply) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, None);
|
|
assert_eq!(status, 413);
|
|
assert_eq!(reply["error"], json!("vault storage quota exceeded"));
|
|
assert!(server.audit().iter().any(|row| row["kind"] == "quota_rejected"));
|
|
assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None).0, 404);
|
|
}
|
|
|
|
#[test]
|
|
fn a_chunked_file_has_revisions_and_their_bytes_stay_restorable() {
|
|
let (server, _) = chunks_server("history", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let (v1, hashes1) = upload_chunks(&server, &vault, &token, &[1, 2]);
|
|
let (status, _) = assemble(&server, &vault, &token, "doc.bin", &hashes1, 0, None);
|
|
assert_eq!(status, 201);
|
|
|
|
// The edit arrives as its own chunk set on a fresh base.
|
|
let mut hashes2 = hashes1.clone();
|
|
let (extra_bytes, extra_hash) = (chunk_data(3, CHUNK_MAX), hash_hex(&chunk_data(3, CHUNK_MAX)));
|
|
assert_eq!(put_chunk(&server, &vault, &token, &extra_hash, &extra_bytes), 201);
|
|
hashes2[0] = extra_hash;
|
|
let (status, record) = assemble(&server, &vault, &token, "doc.bin", &hashes2, 1, None);
|
|
assert_eq!(status, 201, "the edited version wins with a fresh base");
|
|
let seq2 = record["seq"].as_u64().expect("seq");
|
|
assert_eq!(seq2, 2);
|
|
|
|
let (status, history) = server.json("GET", &format!("/v1/vaults/{vault}/files-history/doc.bin"), Some(&token), &[], None);
|
|
assert_eq!((status, history.as_array().map(Vec::len)), (200, Some(2)));
|
|
let (status, old) = server.send("GET", &format!("/v1/vaults/{vault}/files/doc.bin?seq=1"), Some(&token), &[], None);
|
|
assert_eq!((status, old.as_slice()), (200, v1.as_slice()));
|
|
}
|
|
|
|
#[test]
|
|
fn gc_removes_unreferenced_chunks_and_keeps_assembled_ones() {
|
|
let (server, data) = chunks_server("gc", None);
|
|
let (vault, token) = server.paired("pc");
|
|
let (_, referenced) = upload_chunks(&server, &vault, &token, &[1, 2]);
|
|
let _spare = upload_chunks(&server, &vault, &token, &[3, 4]);
|
|
assert_eq!(assemble(&server, &vault, &token, "big.bin", &referenced, 0, None).0, 201);
|
|
let content: Vec<u8> = chunk_data(1, CHUNK_MAX).into_iter().chain(chunk_data(2, CHUNK_MAX)).collect();
|
|
|
|
// Unreferenced chunks of a dead upload and an orphan file: collected after the grace.
|
|
let orphan = data.join("blobs").join(&vault).join("chunks").join("garbage");
|
|
std::fs::write(&orphan, b"waste").expect("orphan file");
|
|
let aged = now_ms() + TOMBSTONE_GRACE_MS + 1;
|
|
let report = server.state.gc(&vault, TOMBSTONE_GRACE_MS, aged).expect("gc");
|
|
assert_eq!(report.removed, 3, "two spare chunks plus the orphan file");
|
|
assert_eq!(chunk_names(&data, &vault), sorted(referenced.as_slice()), "the assembled file's chunks stay");
|
|
let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
|
|
assert_eq!((status, bytes.as_slice()), (200, content.as_slice()), "the assembled file survives GC");
|
|
}
|
|
|
|
#[test]
|
|
fn hostile_input_is_refused_without_panic() {
|
|
let (server, data) = chunks_server("hostile", None);
|
|
let (vault, token) = server.paired("pc");
|
|
|
|
// A chunk check of 5001 hashes is refused; 5000 are the cap.
|
|
let many: Vec<String> = (0..5001).map(|i| format!("{i:064x}")).collect();
|
|
let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": many})));
|
|
assert_eq!(status, 400);
|
|
let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": []})));
|
|
assert_eq!(status, 200, "an empty check answers complete");
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3]);
|
|
let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
|
|
assert_eq!((status, missing), (200, Vec::<String>::new()));
|
|
|
|
// An X-Chunks of 5000 entries is refused: beyond the 4096 list limit. The header
|
|
// fits hyper's 408 KiB parse buffer, so it reaches the handler — ureq cannot send
|
|
// headers this large, so the request goes through a raw HTTP connection.
|
|
let beyond: Vec<String> = (0..5000).map(|i| format!("{i:064x}")).collect();
|
|
let (status, _) = raw_put(&server, &token, &vault, "/x.bin", &beyond.join(","));
|
|
assert_eq!(status, 400);
|
|
|
|
// Malformed check entries, a hostile path and a non-empty body.
|
|
|
|
// Malformed check entries, a hostile path and a non-empty body.
|
|
let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": ["zz"]})));
|
|
assert_eq!(status, 400);
|
|
let (_, hashes) = upload_chunks(&server, &vault, &token, &[8]);
|
|
assert_eq!(assemble(&server, &vault, &token, "a/../x.bin", &hashes, 0, None).0, 400);
|
|
let (status, _) =
|
|
server.send("PUT", &format!("/v1/vaults/{vault}/files/x.bin"), Some(&token), &[("X-Chunks", &hash_hex(&chunk_data(8, CHUNK_MAX)))], Some(b"not empty"));
|
|
assert_eq!(status, 400, "a chunked upload carries no body");
|
|
let (status, _) = server.send("PUT", &format!("/v1/vaults/{vault}/files/x.bin"), Some(&token), &[("X-Chunks", "")], None);
|
|
assert_eq!(status, 400, "an empty X-Chunks names no chunks");
|
|
assert!(chunk_names(&data, &vault).len() >= 2, "the honest chunks still exist");
|
|
}
|