diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 02d69c43..c2fc41ec 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -35,6 +35,8 @@ jobs: run: cargo build --workspace --verbose - name: Run tests run: cargo test --workspace --verbose + - name: Run aw-sync v2 tests (sync-v2 feature) + run: cargo test -p aw-sync --features sync-v2 v2 --verbose - uses: actions/upload-artifact@v7 with: # TODO: These binaries are debug builds diff --git a/Cargo.lock b/Cargo.lock index 39e210f1..f200bd93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -321,8 +321,11 @@ dependencies = [ "rusqlite", "serde", "serde_json", + "sha2", "tempfile", "toml", + "windows-sys 0.59.0", + "zstd", ] [[package]] @@ -442,6 +445,8 @@ version = "1.2.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "deec109607ca693028562ed836a5f1c4b8bd77755c4e132fc5ce11b0b6211ae7" dependencies = [ + "jobserver", + "libc", "shlex", ] @@ -1063,11 +1068,22 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi", + "r-efi 5.3.0", "wasi 0.14.2+wasi-0.2.4", "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi 6.0.0", +] + [[package]] name = "gimli" version = "0.31.1" @@ -1599,6 +1615,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130" +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.77" @@ -2235,6 +2261,12 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "rand" version = "0.8.6" @@ -4130,3 +4162,31 @@ dependencies = [ "quote", "syn", ] + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64d80649ab6db9d9f6f9c80a40becd948eda4714a0a5ac8c4d157a32231c7882" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.1.0+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ef0a8027ec3ee71300ab3bcbcd0393f434aa72b91ca6d635a39941deae8eea0" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/aw-sync/Cargo.toml b/aw-sync/Cargo.toml index 85e0af66..3653e56b 100644 --- a/aw-sync/Cargo.toml +++ b/aw-sync/Cargo.toml @@ -28,6 +28,10 @@ clap = { version = "4.1", features = ["derive"], optional = true } ctrlc = { version = "3.4.5", optional = true } rusqlite = { version = "0.30", features = ["bundled"], optional = true } +# sync-v2 dependencies (optional) +zstd = { version = "0.13", optional = true } +sha2 = { version = "0.10", optional = true } + aw-server = { path = "../aw-server" } aw-models = { path = "../aw-models" } aw-datastore = { path = "../aw-datastore" } @@ -40,9 +44,13 @@ openssl = { version = "0.10.80", features = ["vendored"] } # https://github.com jni = { version = "0.21", default-features = false } android_logger = "0.13" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = ["Win32_Storage_FileSystem", "Win32_Foundation"] } + [dev-dependencies] tempfile = "3" [features] default = ["cli"] cli = ["clap", "ctrlc", "dep:rusqlite"] +sync-v2 = ["dep:zstd", "dep:sha2"] diff --git a/aw-sync/README.md b/aw-sync/README.md index fd1d3985..2591e535 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -57,6 +57,16 @@ Working paths (bare `aw-sync sync`, Android) write: The default daemon still writes `~/ActivityWatchSync/{device_id}/test.db` (two levels). `aw-sync sync` and the Android app cannot see that file. +#### Experimental v2 folder format (`sync-v2` feature) + +Building with `--features sync-v2` compiles a writer for the immutable-segment format (`devices/{device_id}/manifest.json` plus `{slug}.{generation}.jsonl.zst` segments). It is not wired into any command yet. + +The writer stages every file as `*.tmp` next to its final name and renames it into place, so a file syncer can ship a half-written temp file. Readers ignore unknown names, but tell your syncer to skip them too. For Syncthing, add this to the sync folder's `.stignore`: + +```txt +(?d)*.tmp +``` + ### Running from source If you want to run it from source, in the root of the repository run: diff --git a/aw-sync/src/lib.rs b/aw-sync/src/lib.rs index 3a85988a..30d55d24 100644 --- a/aw-sync/src/lib.rs +++ b/aw-sync/src/lib.rs @@ -31,5 +31,8 @@ mod status; pub use status::run_status; mod util; +#[cfg(feature = "sync-v2")] +pub mod v2; + #[cfg(target_os = "android")] pub mod android; diff --git a/aw-sync/src/v2/manifest.rs b/aw-sync/src/v2/manifest.rs new file mode 100644 index 00000000..a67f9033 --- /dev/null +++ b/aw-sync/src/v2/manifest.rs @@ -0,0 +1,317 @@ +//! Manifest format for aw-sync v2. +//! +//! The manifest is the entry point for importers. It maps bucket ids to their +//! segment list and metadata without requiring importers to open any segment file. + +use std::collections::HashMap; +use std::fs; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +use super::MAX_V; + +/// Top-level manifest file (`manifest.json`). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Manifest { + /// Format version. Importers must refuse if v > MAX_V. + pub v: u32, + pub device_id: String, + pub hostname: String, + /// ISO-8601 timestamp of the last manifest write. + pub written_at: DateTime, + /// Keyed by real bucket_id. + pub buckets: HashMap, +} + +/// Per-bucket metadata and segment list inside the manifest. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BucketEntry { + /// First 16 hex chars of sha256(bucket_id). Used in filenames. + pub slug: String, + #[serde(rename = "type")] + pub _type: String, + pub client: String, + pub hostname: String, + pub created: Option>, + pub latest_generation: u64, + pub total_events: u64, + pub segments: Vec, +} + +/// One segment's metadata inside a BucketEntry. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SegmentEntry { + pub generation: u64, + pub file: String, + pub n_events: u64, + pub start_ts: Option>, + pub end_ts: Option>, + /// SHA-256 hex of the compressed segment bytes. + /// Only authoritative when `sealed == true`. + pub sha256: String, + /// True once the segment will not be rewritten. An unsealed segment is the + /// current open tail; its sha256 changes on every writer pass. + pub sealed: bool, + /// Wall-clock time this generation was first written. Unlike `start_ts` + /// (the oldest *event* timestamp), this never changes across rewrites of + /// the same open tail — it drives the age-based seal policy so importing + /// old historical events doesn't look "already expired" on the first pass. + #[serde(default)] + pub first_written_at: Option>, +} + +impl Manifest { + /// Load an existing manifest or return a fresh default. + pub fn load_or_default(dir: &Path, device_id: &str, hostname: &str) -> Result { + validate_device_id(device_id)?; + let path = manifest_path(dir, device_id); + if path.exists() { + let data = fs::read_to_string(&path).map_err(|e| format!("read manifest: {e}"))?; + let m: Manifest = + serde_json::from_str(&data).map_err(|e| format!("parse manifest: {e}"))?; + if m.v > MAX_V { + return Err(format!( + "manifest v{} > MAX_V {MAX_V} — refusing to open", + m.v + )); + } + Ok(m) + } else { + Ok(Manifest { + v: 1, + device_id: device_id.to_string(), + hostname: hostname.to_string(), + written_at: Utc::now(), + buckets: HashMap::new(), + }) + } + } + + /// Upsert or replace the entry for a bucket. + pub fn upsert_bucket(&mut self, bucket_id: &str, entry: BucketEntry) { + self.buckets.insert(bucket_id.to_string(), entry); + } + + /// Write atomically: temp file → fsync → rename. + pub fn save(&self, dir: &Path, device_id: &str) -> Result<(), String> { + let target = manifest_path(dir, device_id); + fs::create_dir_all(target.parent().unwrap()) + .map_err(|e| format!("create manifest dir: {e}"))?; + + let tmp = target.with_extension("json.tmp"); + { + let f = fs::File::create(&tmp).map_err(|e| format!("create manifest tmp: {e}"))?; + let mut w = BufWriter::new(&f); + let json = serde_json::to_string_pretty(self) + .map_err(|e| format!("serialize manifest: {e}"))?; + w.write_all(json.as_bytes()) + .map_err(|e| format!("write manifest: {e}"))?; + w.flush().map_err(|e| format!("flush manifest: {e}"))?; + f.sync_all().map_err(|e| format!("fsync manifest: {e}"))?; + } + durable_rename(&tmp, &target)?; + fsync_dir(target.parent().unwrap())?; + Ok(()) + } +} + +/// fsync a directory so a preceding `rename` into it survives a crash. +/// +/// No-op on Windows: directory handles opened via `File::open` cannot be +/// fsynced on that platform. Durability for the rename itself is instead +/// provided by `durable_rename`, which uses `MOVEFILE_WRITE_THROUGH` there. +pub(crate) fn fsync_dir(dir: &Path) -> Result<(), String> { + #[cfg(unix)] + { + fs::File::open(dir) + .and_then(|f| f.sync_all()) + .map_err(|e| format!("fsync dir {}: {e}", dir.display()))?; + } + #[cfg(not(unix))] + let _ = dir; + Ok(()) +} + +/// Rename `from` to `to`, replacing any existing file at `to`, with the same +/// crash-durability guarantee on every supported platform. +/// +/// On Unix, a plain `rename` is already durable once the directory entry is +/// fsynced (`fsync_dir`, called by the caller). Windows has no directory-fsync +/// primitive, so the equivalent guarantee there is `MOVEFILE_WRITE_THROUGH`: +/// `MoveFileExW` does not return until the rename is flushed to disk. +pub(crate) fn durable_rename(from: &Path, to: &Path) -> Result<(), String> { + #[cfg(windows)] + { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, + }; + + let from_wide: Vec = from.as_os_str().encode_wide().chain(Some(0)).collect(); + let to_wide: Vec = to.as_os_str().encode_wide().chain(Some(0)).collect(); + // SAFETY: both buffers are valid, NUL-terminated UTF-16 strings that + // outlive the call. + let ok = unsafe { + MoveFileExW( + from_wide.as_ptr(), + to_wide.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if ok == 0 { + return Err(format!( + "rename {} -> {}: {}", + from.display(), + to.display(), + std::io::Error::last_os_error() + )); + } + Ok(()) + } + #[cfg(not(windows))] + { + fs::rename(from, to) + .map_err(|e| format!("rename {} -> {}: {e}", from.display(), to.display())) + } +} + +/// Path for manifest inside the device directory. +pub fn manifest_path(sync_root: &Path, device_id: &str) -> PathBuf { + device_dir(sync_root, device_id).join("manifest.json") +} + +/// `/devices/` +pub fn device_dir(sync_root: &Path, device_id: &str) -> PathBuf { + sync_root.join("devices").join(device_id) +} + +/// Reject a device_id that could escape the sync root via path traversal. +pub(super) fn validate_device_id(device_id: &str) -> Result<(), String> { + if device_id.is_empty() { + return Err("device_id must not be empty".to_string()); + } + if device_id.contains('/') || device_id.contains('\\') || device_id.contains("..") { + return Err(format!( + "invalid device_id {device_id:?}: must not contain path separators or '..'" + )); + } + Ok(()) +} + +/// Compute the 16-char slug used in segment filenames. +pub fn bucket_slug(bucket_id: &str) -> String { + use sha2::{Digest, Sha256}; + let mut h = Sha256::new(); + h.update(bucket_id.as_bytes()); + let result = h.finalize(); + // First 8 bytes = 16 hex chars + format!( + "{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}", + result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7] + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_bucket_slug_length() { + let slug = bucket_slug("aw-watcher-window_my-desktop"); + assert_eq!(slug.len(), 16, "slug must be exactly 16 hex chars"); + } + + #[test] + fn test_bucket_slug_stable() { + // Same input must always produce the same slug. + let s1 = bucket_slug("aw-watcher-window_test"); + let s2 = bucket_slug("aw-watcher-window_test"); + assert_eq!(s1, s2); + } + + #[test] + fn test_bucket_slug_distinct() { + let s1 = bucket_slug("aw-watcher-window_host-a"); + let s2 = bucket_slug("aw-watcher-afk_host-a"); + assert_ne!(s1, s2); + } + + #[test] + fn test_manifest_round_trip() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_abc123"; + let hostname = "test-host"; + + let mut m = Manifest::load_or_default(dir.path(), device_id, hostname).unwrap(); + assert_eq!(m.v, 1); + assert!(m.buckets.is_empty()); + + // Upsert a bucket entry + m.upsert_bucket( + "aw-watcher-window_test-host", + BucketEntry { + slug: bucket_slug("aw-watcher-window_test-host"), + _type: "currentwindow".to_string(), + client: "aw-watcher-window".to_string(), + hostname: hostname.to_string(), + created: None, + latest_generation: 1, + total_events: 42, + segments: vec![SegmentEntry { + generation: 1, + file: "a1b2c3d4e5f60718.00000001.jsonl.zst".to_string(), + n_events: 42, + start_ts: None, + end_ts: None, + sha256: "abc".to_string(), + sealed: false, + first_written_at: Some(Utc::now()), + }], + }, + ); + m.save(dir.path(), device_id).unwrap(); + + // Reload and verify + let m2 = Manifest::load_or_default(dir.path(), device_id, hostname).unwrap(); + assert_eq!(m2.buckets.len(), 1); + let entry = m2.buckets.get("aw-watcher-window_test-host").unwrap(); + assert_eq!(entry.latest_generation, 1); + assert_eq!(entry.total_events, 42); + assert_eq!(entry.segments.len(), 1); + assert!(!entry.segments[0].sealed); + } + + #[test] + fn test_manifest_version_guard() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_abc123"; + let hostname = "test-host"; + + // Write a manifest with v=99 + let device_path = device_dir(dir.path(), device_id); + fs::create_dir_all(&device_path).unwrap(); + let manifest_json = r#"{"v":99,"device_id":"test-host_abc123","hostname":"test-host","written_at":"2026-01-01T00:00:00Z","buckets":{}}"#; + fs::write(manifest_path(dir.path(), device_id), manifest_json).unwrap(); + + let result = Manifest::load_or_default(dir.path(), device_id, hostname); + assert!(result.is_err(), "should refuse v > MAX_V"); + assert!(result.unwrap_err().contains("refusing to open")); + } + + #[test] + fn test_device_id_path_traversal_rejected() { + let dir = tempfile::tempdir().unwrap(); + let hostname = "test-host"; + for bad_id in &["../evil", "../../etc/passwd", "foo/bar", "foo\\bar"] { + let result = Manifest::load_or_default(dir.path(), bad_id, hostname); + assert!( + result.is_err(), + "should reject path-unsafe device_id: {bad_id}" + ); + assert!(result.unwrap_err().contains("invalid device_id")); + } + } +} diff --git a/aw-sync/src/v2/mod.rs b/aw-sync/src/v2/mod.rs new file mode 100644 index 00000000..a76aaad2 --- /dev/null +++ b/aw-sync/src/v2/mod.rs @@ -0,0 +1,17 @@ +//! aw-sync v2: immutable JSONL+zstd segment format +//! +//! Layout under the sync root: +//! devices/{device_id}/manifest.json +//! devices/{device_id}/{bucket_slug}.{gen:08d}.jsonl.zst +//! +//! Gated by the `sync-v2` feature flag. No existing sync code paths are +//! modified; this module is entirely additive. + +pub mod manifest; +pub mod segment; + +pub use manifest::Manifest; +pub use segment::SegmentWriter; + +/// Maximum format version this build understands. +pub const MAX_V: u32 = 1; diff --git a/aw-sync/src/v2/segment.rs b/aw-sync/src/v2/segment.rs new file mode 100644 index 00000000..c88b09b9 --- /dev/null +++ b/aw-sync/src/v2/segment.rs @@ -0,0 +1,798 @@ +//! Segment writer for aw-sync v2. +//! +//! Produces immutable JSONL+zstd segment files. Write pattern: +//! 1. Open a temp file (.{name}.tmp) +//! 2. Write zstd-compressed JSONL (header + events) +//! 3. fsync + atomic rename to the final path +//! 4. Update the manifest +//! +//! An open-tail segment (below the size threshold) is rewritten in-place each +//! daemon pass using the same tmp+rename sequence, so a reader always sees a +//! complete file. + +use std::collections::BTreeMap; +use std::fs; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use aw_models::{Bucket, Event}; + +use super::manifest::{bucket_slug, device_dir, BucketEntry, Manifest, SegmentEntry}; + +/// Minimum compressed segment size before sealing (1 MiB). +const SEAL_SIZE_BYTES: u64 = 1024 * 1024; + +/// Maximum age of an open-tail generation before it is sealed regardless of +/// size, measured from when the generation was first written. Ensures a +/// low-volume bucket doesn't rewrite the same tail forever. +fn seal_max_age() -> chrono::Duration { + chrono::Duration::days(1) +} + +/// (n_events, start_ts, end_ts, sha256_hex, compressed_size) +type SegmentWriteResult = ( + u64, + Option>, + Option>, + String, + u64, +); + +/// Segment header (line 1 of each segment JSONL). +#[derive(Debug, Serialize, Deserialize)] +struct SegmentHeader { + v: u32, + device_id: String, + bucket_id: String, + bucket_type: String, + bucket_client: String, + bucket_hostname: String, + bucket_created: Option>, + generation: u64, + written_at: DateTime, + /// `null` for append-only; `{from, to}` for a range-replace segment. + replaces: Option, +} + +/// Range-replace marker — reserved for future use; writer always emits null. +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct ReplaceRange { + pub from: DateTime, + pub to: DateTime, +} + +/// Wire event format matching the aw-core REST API response. +#[derive(Debug, Serialize)] +struct WireEvent<'a> { + id: Option, + timestamp: String, + duration: f64, + data: &'a serde_json::Map, +} + +/// Writes v2 segment files for a single bucket. +pub struct SegmentWriter { + device_id: String, + bucket_id: String, + sync_dir: PathBuf, + generation: u64, + slug: String, +} + +impl SegmentWriter { + /// Load or initialize the writer for a bucket, restoring the generation + /// counter from the existing manifest if present. + pub fn new(sync_dir: &Path, device_id: &str, bucket_id: &str) -> Result { + let hostname = gethostname::gethostname() + .into_string() + .unwrap_or_else(|_| "unknown".to_string()); + let manifest = Manifest::load_or_default(sync_dir, device_id, &hostname)?; + let generation = manifest + .buckets + .get(bucket_id) + .map(|e| e.latest_generation) + .unwrap_or(0); + Ok(SegmentWriter { + device_id: device_id.to_string(), + bucket_id: bucket_id.to_string(), + sync_dir: sync_dir.to_path_buf(), + generation, + slug: bucket_slug(bucket_id), + }) + } + + /// Write events to a segment and update the manifest. + /// + /// If the previous generation's segment is unsealed (< SEAL_SIZE_BYTES + /// and younger than `SEAL_MAX_AGE`), it is rewritten under the same + /// generation number — merged with whatever events that segment already + /// held, so a caller passing only newly-available events never drops the + /// existing tail. Otherwise a new generation is started. + /// + /// Returns the generation number written. + pub fn write_events(&mut self, bucket: &Bucket, events: &[Event]) -> Result { + if events.is_empty() { + return Ok(self.generation); + } + + let device_dir = device_dir(&self.sync_dir, &self.device_id); + fs::create_dir_all(&device_dir).map_err(|e| format!("create device dir: {e}"))?; + + // Load once; reused both for the seal decision and the final update. + let hostname = gethostname::gethostname() + .into_string() + .unwrap_or_else(|_| "unknown".to_string()); + let mut manifest = Manifest::load_or_default(&self.sync_dir, &self.device_id, &hostname)?; + + // Determine whether to seal the previous segment and start a new one. + let should_start_new = if self.generation == 0 { + true // First ever segment + } else { + let prev_path = self.segment_path(self.generation); + if prev_path.exists() { + let size = fs::metadata(&prev_path).map(|m| m.len()).unwrap_or(0); + // Age is measured from when this generation was first written + // to disk, not from its oldest event timestamp — otherwise + // importing events that are already older than SEAL_MAX_AGE + // (e.g. a historical backfill) would seal on every single + // pass instead of accumulating in the open tail. + let prev_created = manifest + .buckets + .get(&self.bucket_id) + .and_then(|e| e.segments.iter().find(|s| s.generation == self.generation)) + .and_then(|s| s.first_written_at); + let too_old = prev_created + .map(|created| Utc::now() - created >= seal_max_age()) + .unwrap_or(false); + size >= SEAL_SIZE_BYTES || too_old + } else { + true + } + }; + + let write_gen = if should_start_new && self.generation > 0 { + self.generation + 1 + } else if should_start_new { + 1 + } else { + self.generation + }; + + // Seal previous generation if we're moving forward + let prev_sealed = should_start_new && self.generation > 0; + + // When rewriting the open tail, merge with whatever it already holds + // so a caller passing only the events new since the last pass can't + // make earlier events vanish from the sync folder. + let merged_events; + let events_to_write: &[Event] = if !should_start_new { + let prev_path = self.segment_path(write_gen); + let existing = Self::read_segment_events(&prev_path)?; + merged_events = merge_events(existing, events); + &merged_events + } else { + events + }; + + // Write the segment + let (n_events, start_ts, end_ts, sha256_hex, compressed_size) = + self.write_segment_file(bucket, events_to_write, write_gen)?; + + let sealed = compressed_size >= SEAL_SIZE_BYTES; + + let slug = self.slug.clone(); + let existing = manifest.buckets.remove(&self.bucket_id); + let mut segments: Vec = existing + .as_ref() + .map(|e| e.segments.clone()) + .unwrap_or_default(); + + // If sealing the previous generation, mark it sealed in the manifest. + // Guard: only seal if the file is actually present — if it was + // externally deleted (NotFound), drop the dangling entry instead of + // advertising a sealed segment with a stale sha256 that points + // nowhere. Any other I/O error (permissions, hardware, sharing + // violation on Windows) must propagate: silently dropping the entry + // on a transient error would silently lose history. + if prev_sealed { + let prev_path = self.segment_path(self.generation); + match fs::read(&prev_path) { + Ok(bytes) => { + if let Some(s) = segments + .iter_mut() + .find(|s| s.generation == self.generation) + { + let mut h = Sha256::new(); + h.update(&bytes); + s.sha256 = format!("{:x}", h.finalize()); + s.sealed = true; + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + segments.retain(|s| s.generation != self.generation); + } + Err(e) => { + return Err(format!("read previous segment for sealing: {e}")); + } + } + } + + // Preserve the generation's original creation time across open-tail + // rewrites; only a brand-new generation gets a fresh timestamp. + let first_written_at = segments + .iter() + .find(|s| s.generation == write_gen) + .and_then(|s| s.first_written_at) + .unwrap_or_else(Utc::now); + + // Remove any existing entry for this generation (open-tail rewrite) + segments.retain(|s| s.generation != write_gen); + segments.push(SegmentEntry { + generation: write_gen, + file: self.segment_filename(write_gen), + n_events, + start_ts, + end_ts, + sha256: sha256_hex, + sealed, + first_written_at: Some(first_written_at), + }); + segments.sort_by_key(|s| s.generation); + + let total_events = segments.iter().map(|s| s.n_events).sum(); + let bucket_entry = BucketEntry { + slug, + _type: bucket._type.clone(), + client: bucket.client.clone(), + hostname: bucket.hostname.clone(), + created: bucket.created, + latest_generation: write_gen, + total_events, + segments, + }; + manifest.upsert_bucket(&self.bucket_id, bucket_entry); + manifest.written_at = Utc::now(); + manifest.save(&self.sync_dir, &self.device_id)?; + + self.generation = write_gen; + Ok(write_gen) + } + + /// Write the compressed JSONL segment file atomically. + /// Returns (n_events, start_ts, end_ts, sha256_hex, compressed_size). + fn write_segment_file( + &self, + bucket: &Bucket, + events: &[Event], + generation: u64, + ) -> Result { + let segment_path = self.segment_path(generation); + let tmp_path = segment_path + .parent() + .unwrap() + .join(format!(".{}.tmp", self.segment_filename(generation))); + + // Build JSONL in memory, then compress + let mut jsonl = Vec::new(); + + // Line 1: header + let header = SegmentHeader { + v: 1, + device_id: self.device_id.clone(), + bucket_id: self.bucket_id.clone(), + bucket_type: bucket._type.clone(), + bucket_client: bucket.client.clone(), + bucket_hostname: bucket.hostname.clone(), + bucket_created: bucket.created, + generation, + written_at: Utc::now(), + replaces: None, + }; + serde_json::to_writer(&mut jsonl, &header).map_err(|e| format!("serialize header: {e}"))?; + jsonl.push(b'\n'); + + // Lines 2+: events in chronological order. Sort here rather than + // trusting the caller — the datastore's own retrieval order is + // newest-first, so passing that straight through would silently + // violate the segment format's ordering contract. + let mut sorted_events: Vec<&Event> = events.iter().collect(); + sorted_events.sort_by_key(|e| e.timestamp); + + let mut start_ts: Option> = None; + let mut end_ts: Option> = None; + let mut n_events: u64 = 0; + + for event in sorted_events { + let ts_str = event + .timestamp + .to_rfc3339_opts(chrono::SecondsFormat::Nanos, true); + let duration_secs = + event.duration.num_nanoseconds().unwrap_or(0) as f64 / 1_000_000_000.0; + let wire = WireEvent { + id: event.id, + timestamp: ts_str, + duration: duration_secs, + data: &event.data, + }; + serde_json::to_writer(&mut jsonl, &wire) + .map_err(|e| format!("serialize event: {e}"))?; + jsonl.push(b'\n'); + + if start_ts.is_none() || event.timestamp < start_ts.unwrap() { + start_ts = Some(event.timestamp); + } + let end = event.timestamp + event.duration; + if end_ts.is_none() || end > end_ts.unwrap() { + end_ts = Some(end); + } + n_events += 1; + } + + // Compress with zstd + let compressed = + zstd::encode_all(jsonl.as_slice(), 0).map_err(|e| format!("zstd compress: {e}"))?; + + let compressed_size = compressed.len() as u64; + + // SHA-256 of compressed bytes + let mut hasher = Sha256::new(); + hasher.update(&compressed); + let sha256_hex = format!("{:x}", hasher.finalize()); + + // Atomic write + { + let f = fs::File::create(&tmp_path).map_err(|e| format!("create segment tmp: {e}"))?; + let mut w = BufWriter::new(&f); + w.write_all(&compressed) + .map_err(|e| format!("write segment: {e}"))?; + w.flush().map_err(|e| format!("flush segment: {e}"))?; + f.sync_all().map_err(|e| format!("fsync segment: {e}"))?; + } + super::manifest::durable_rename(&tmp_path, &segment_path)?; + super::manifest::fsync_dir(segment_path.parent().unwrap())?; + + Ok((n_events, start_ts, end_ts, sha256_hex, compressed_size)) + } + + /// Read and decode an existing segment's events (skipping the header + /// line). Returns an empty vec if the segment doesn't exist yet. + fn read_segment_events(path: &Path) -> Result, String> { + if !path.exists() { + return Ok(Vec::new()); + } + let compressed = fs::read(path).map_err(|e| format!("read segment: {e}"))?; + let decompressed = zstd::decode_all(compressed.as_slice()) + .map_err(|e| format!("zstd decompress segment: {e}"))?; + let text = String::from_utf8(decompressed).map_err(|e| format!("segment utf8: {e}"))?; + text.lines() + .skip(1) // header + .filter(|line| !line.is_empty()) + .map(|line| { + serde_json::from_str::(line).map_err(|e| format!("parse event line: {e}")) + }) + .collect() + } + + fn segment_filename(&self, generation: u64) -> String { + format!("{}.{:08}.jsonl.zst", self.slug, generation) + } + + fn segment_path(&self, generation: u64) -> PathBuf { + device_dir(&self.sync_dir, &self.device_id).join(self.segment_filename(generation)) + } +} + +/// Union of `existing` and `new`, deduped by event id (later entries win), +/// events without an id kept as-is, sorted chronologically. +fn merge_events(existing: Vec, new: &[Event]) -> Vec { + let mut by_id: BTreeMap = BTreeMap::new(); + let mut unkeyed: Vec = Vec::new(); + + for event in existing.into_iter().chain(new.iter().cloned()) { + match event.id { + Some(id) => { + by_id.insert(id, event); + } + None => unkeyed.push(event), + } + } + + let mut merged: Vec = by_id.into_values().collect(); + merged.extend(unkeyed); + merged.sort_by_key(|e| e.timestamp); + merged +} + +#[cfg(test)] +mod tests { + use super::*; + use aw_models::Event; + use chrono::Duration; + use serde_json::json; + + fn make_event(ts_offset_s: i64, id: i64) -> Event { + let ts = chrono::Utc::now() + Duration::seconds(ts_offset_s); + let mut data = serde_json::Map::new(); + data.insert("app".to_string(), json!("TestApp")); + Event { + id: Some(id), + timestamp: ts, + duration: Duration::seconds(30), + data, + } + } + + fn make_bucket() -> Bucket { + Bucket { + bid: None, + id: "aw-watcher-window_test-host".to_string(), + _type: "currentwindow".to_string(), + client: "aw-watcher-window".to_string(), + hostname: "test-host".to_string(), + created: Some(Utc::now()), + data: serde_json::Map::new(), + metadata: Default::default(), + events: None, + last_updated: None, + } + } + + #[test] + fn test_segment_write_produces_valid_zstd() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_abc123"; + let bucket = make_bucket(); + let events = vec![make_event(0, 1), make_event(30, 2)]; + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + let gen = writer.write_events(&bucket, &events).unwrap(); + assert_eq!(gen, 1); + + // File must exist and be decompressible + let seg_path = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000001.jsonl.zst", bucket_slug(&bucket.id))); + assert!(seg_path.exists(), "segment file not found: {seg_path:?}"); + + let compressed = fs::read(&seg_path).unwrap(); + let decompressed = zstd::decode_all(compressed.as_slice()).unwrap(); + let text = String::from_utf8(decompressed).unwrap(); + let lines: Vec<&str> = text.lines().collect(); + assert_eq!(lines.len(), 3, "header + 2 events"); + + // Line 1 is the header + let header: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(header["v"], 1); + assert_eq!(header["bucket_id"], "aw-watcher-window_test-host"); + assert_eq!(header["generation"], 1); + assert!(header["replaces"].is_null()); + } + + #[test] + fn test_segment_immutability_after_seal() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_seal"; + let bucket = make_bucket(); + + // Write enough data to cross the seal threshold + // We use a compressed-size trick: write a first small segment, manually + // mark it as >= SEAL_SIZE_BYTES by creating a large fake file. + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + let events_a = vec![make_event(0, 1)]; + let gen_a = writer.write_events(&bucket, &events_a).unwrap(); + assert_eq!(gen_a, 1); + + // Read first segment bytes before the second write + let seg_path_1 = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000001.jsonl.zst", bucket_slug(&bucket.id))); + let bytes_before = fs::read(&seg_path_1).unwrap(); + + // Force the first segment to be "large" so the writer seals it + // by replacing it with a file of >= SEAL_SIZE_BYTES + let fake_large = vec![0u8; SEAL_SIZE_BYTES as usize]; + fs::write(&seg_path_1, &fake_large).unwrap(); + + // Second write — should create generation 2, not overwrite generation 1 + let events_b = vec![make_event(60, 2)]; + let gen_b = writer.write_events(&bucket, &events_b).unwrap(); + assert_eq!(gen_b, 2, "second write should start a new generation"); + + // Generation 1 must NOT have been touched (it was large/sealed) + let bytes_after = fs::read(&seg_path_1).unwrap(); + assert_eq!( + bytes_after.len(), + fake_large.len(), + "sealed segment must not be overwritten" + ); + // The original small content must not have come back + assert_ne!( + bytes_after, bytes_before, + "sanity: the large file we wrote is still there" + ); + + // Generation 2 must exist + let seg_path_2 = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000002.jsonl.zst", bucket_slug(&bucket.id))); + assert!(seg_path_2.exists(), "generation 2 segment not found"); + } + + #[test] + fn test_manifest_reflects_two_segments() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_manifest"; + let bucket = make_bucket(); + let hostname = gethostname::gethostname() + .into_string() + .unwrap_or_else(|_| "unknown".to_string()); + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + + // Write first segment (small, will stay unsealed) + writer.write_events(&bucket, &[make_event(0, 1)]).unwrap(); + + // Force segment 1 to be large so segment 2 starts + let seg1_path = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000001.jsonl.zst", bucket_slug(&bucket.id))); + fs::write(&seg1_path, vec![0u8; SEAL_SIZE_BYTES as usize]).unwrap(); + + // Write second segment + writer.write_events(&bucket, &[make_event(60, 2)]).unwrap(); + + // Manifest must reflect both segments + let manifest = Manifest::load_or_default(dir.path(), device_id, &hostname).unwrap(); + let entry = manifest + .buckets + .get(&bucket.id) + .expect("bucket in manifest"); + assert_eq!(entry.latest_generation, 2); + assert_eq!(entry.segments.len(), 2, "both segments must be listed"); + assert_eq!(entry.segments[0].generation, 1); + assert_eq!(entry.segments[1].generation, 2); + // Generation 1 is now sealed (was made large) + assert!(entry.segments[0].sealed, "gen 1 should be sealed"); + // Generation 2 is small, unsealed + assert!( + !entry.segments[1].sealed, + "gen 2 should be unsealed (small)" + ); + } + + #[test] + fn test_no_events_is_noop() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_noop"; + let bucket = make_bucket(); + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + let gen = writer.write_events(&bucket, &[]).unwrap(); + // Should not advance generation or create any files + assert_eq!(gen, 0); + let device_path = dir.path().join("devices").join(device_id); + assert!(!device_path.exists() || fs::read_dir(&device_path).unwrap().count() == 0); + } + + #[test] + fn test_open_tail_rewrite_preserves_earlier_events() { + // Regression test for the open-tail data loss found in review: a + // daemon pass that supplies only the events new since the last pass + // must not make the earlier events of that generation disappear. + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_tail_merge"; + let bucket = make_bucket(); + let hostname = gethostname::gethostname() + .into_string() + .unwrap_or_else(|_| "unknown".to_string()); + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + + // Pass 1: write event A. + let events_a = vec![make_event(0, 1)]; + let gen_a = writer.write_events(&bucket, &events_a).unwrap(); + assert_eq!(gen_a, 1); + + // Pass 2: caller only passes event B (new since last pass) — the + // segment stays well under SEAL_SIZE_BYTES, so this rewrites gen 1. + let events_b = vec![make_event(60, 2)]; + let gen_b = writer.write_events(&bucket, &events_b).unwrap(); + assert_eq!(gen_b, 1, "still the open tail, not a new generation"); + + // The decoded tail must contain A ∪ B, not just B. + let seg_path = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000001.jsonl.zst", bucket_slug(&bucket.id))); + let compressed = fs::read(&seg_path).unwrap(); + let decompressed = zstd::decode_all(compressed.as_slice()).unwrap(); + let text = String::from_utf8(decompressed).unwrap(); + let ids: Vec = text + .lines() + .skip(1) + .map(|line| serde_json::from_str::(line).unwrap().id.unwrap()) + .collect(); + assert_eq!(ids, vec![1, 2], "decoded tail must contain A union B"); + + // The manifest's n_events must reflect the union too. + let manifest = Manifest::load_or_default(dir.path(), device_id, &hostname).unwrap(); + let entry = manifest.buckets.get(&bucket.id).unwrap(); + assert_eq!( + entry.total_events, 2, + "manifest n_events must be |A union B|" + ); + assert_eq!(entry.segments.len(), 1); + assert_eq!(entry.segments[0].n_events, 2); + } + + #[test] + fn test_segments_written_chronologically_regardless_of_input_order() { + // Regression test: the datastore's natural retrieval order is + // newest-first. The writer must not trust caller order. + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_chrono"; + let bucket = make_bucket(); + + // Pass events in reverse-chronological order (newest first). + let events = vec![make_event(60, 2), make_event(0, 1)]; + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + writer.write_events(&bucket, &events).unwrap(); + + let seg_path = dir + .path() + .join("devices") + .join(device_id) + .join(format!("{}.00000001.jsonl.zst", bucket_slug(&bucket.id))); + let compressed = fs::read(&seg_path).unwrap(); + let decompressed = zstd::decode_all(compressed.as_slice()).unwrap(); + let text = String::from_utf8(decompressed).unwrap(); + let ids: Vec = text + .lines() + .skip(1) + .map(|line| serde_json::from_str::(line).unwrap().id.unwrap()) + .collect(); + assert_eq!( + ids, + vec![1, 2], + "segment lines must be chronological, oldest first" + ); + } + + #[test] + fn test_historical_import_does_not_seal_every_pass() { + // Regression test for review finding: seal-by-age must be measured + // from the generation's own creation time, not from the oldest event + // timestamp. A historical backfill (events already older than + // SEAL_MAX_AGE) must still accumulate in one open tail across passes + // instead of sealing a tiny segment on every single write. + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_historical"; + let bucket = make_bucket(); + let hostname = gethostname::gethostname() + .into_string() + .unwrap_or_else(|_| "unknown".to_string()); + + // Events dated well before SEAL_MAX_AGE (a "historical" backfill). + let old_offset = -(seal_max_age().num_seconds() * 3); + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + + let events_a = vec![make_event(old_offset, 1)]; + let gen_a = writer.write_events(&bucket, &events_a).unwrap(); + assert_eq!(gen_a, 1); + + let events_b = vec![make_event(old_offset + 60, 2)]; + let gen_b = writer.write_events(&bucket, &events_b).unwrap(); + assert_eq!( + gen_b, 1, + "old event timestamps must not trigger seal-by-age; only wall-clock \ + time since the generation's own creation should" + ); + + let manifest = Manifest::load_or_default(dir.path(), device_id, &hostname).unwrap(); + let entry = manifest.buckets.get(&bucket.id).unwrap(); + assert_eq!(entry.segments.len(), 1, "must stay a single open tail"); + assert!(!entry.segments[0].sealed); + assert_eq!(entry.segments[0].n_events, 2); + } + + #[test] + fn test_externally_deleted_segment_drops_manifest_entry() { + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_abc123"; + let hostname = "test-host"; + let bucket = make_bucket(); + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + let events_a = vec![make_event(0, 1)]; + writer.write_events(&bucket, &events_a).unwrap(); + + // Confirm segment exists in manifest + { + let m = Manifest::load_or_default(dir.path(), device_id, &hostname).unwrap(); + assert_eq!(m.buckets[&bucket.id].segments.len(), 1); + } + + // Externally delete the segment file + let seg_path = writer.segment_path(1); + fs::remove_file(&seg_path).unwrap(); + + // A second write with enough size/events to force a new generation + // would normally try to seal gen 1 — but the file is gone. + // We simulate by calling write_events again; since the file is absent, + // should_start_new triggers and the old entry must be removed, not + // marked sealed with a stale sha256. + let events_b = vec![make_event(120, 2)]; + writer.write_events(&bucket, &events_b).unwrap(); + + let m = Manifest::load_or_default(dir.path(), device_id, &hostname).unwrap(); + let entry = &m.buckets[&bucket.id]; + // Gen 1 entry must not be present — it would have a stale/empty sha256 + for seg in &entry.segments { + assert_ne!(seg.generation, 1, "dangling gen=1 entry must be removed"); + assert!( + !seg.sealed || !seg.sha256.is_empty(), + "sealed segment must have a valid sha256" + ); + } + } + + // On Unix: replace a "large" gen-1 segment with a self-referential symlink. + // Path::exists() follows the link and returns false (so should_start_new + // fires and prev_sealed = true), but fs::read fails with ELOOP (not + // ENOENT) — which must propagate rather than silently drop the history. + #[cfg(unix)] + #[test] + fn test_io_error_during_seal_propagates() { + use std::os::unix::fs as unix_fs; + + let dir = tempfile::tempdir().unwrap(); + let device_id = "test-host_io_err"; + let bucket = make_bucket(); + + let mut writer = SegmentWriter::new(dir.path(), device_id, &bucket.id).unwrap(); + writer.write_events(&bucket, &[make_event(0, 1)]).unwrap(); + + // Inflate gen 1 past SEAL_SIZE_BYTES so the writer knows to advance. + let seg_path = writer.segment_path(1); + fs::write(&seg_path, vec![0u8; SEAL_SIZE_BYTES as usize]).unwrap(); + + // Replace with a self-referential symlink: seg_path -> seg_path. + // exists() resolves the chain, hits ELOOP, and returns false, so the + // writer treats it as missing (starts gen 2) and sets prev_sealed=true. + // fs::read on the loop symlink returns ELOOP, not ENOENT. + fs::remove_file(&seg_path).unwrap(); + unix_fs::symlink(&seg_path, &seg_path).unwrap(); + + let result = writer.write_events(&bucket, &[make_event(60, 2)]); + + // Remove the loop symlink so tempdir cleanup can proceed. + fs::remove_file(&seg_path).ok(); + + assert!( + result.is_err(), + "non-NotFound I/O error must propagate, not silently drop history" + ); + assert!( + result + .as_ref() + .unwrap_err() + .contains("read previous segment"), + "error must name the seal-read failure, got: {:?}", + result + ); + } +}