| 1 | //! The on-disk ring buffer, the tombstone, and the wipe. |
| 2 | //! |
| 3 | //! Everything lives under `$CODEWHALE_HOME/telemetry/`, created `0700`, with |
| 4 | //! every file `0600`: |
| 5 | //! |
| 6 | //! | file | role | |
| 7 | //! |---|---| |
| 8 | //! | `buffer.jsonl` | one JSON event per line, awaiting a flush | |
| 9 | //! | `buffer.jsonl.lock` | a **sibling** lock file; never the data file | |
| 10 | //! | `dryrun.jsonl` | the sink when the endpoint resolves empty, same ring policy | |
| 11 | //! | `state.json` | last version seen and last flush attempt | |
| 12 | //! | `install_id.json` | the random install id | |
| 13 | //! | `disabled` | the tombstone: present ⇒ nothing is appended, drained, or sent | |
| 14 | //! |
| 15 | //! Appends, compaction, delivery, identity/state writes, startup arming, and |
| 16 | //! wipe share one sibling lock. Runtime and exit paths take it with |
| 17 | //! `try_write()`: on contention the event or batch is dropped. Only startup |
| 18 | //! arming and the user-requested wipe may wait. That keeps the panic hook and |
| 19 | //! SIGINT path non-blocking while also making opt-out an ordering boundary — |
| 20 | //! after wipe returns, no pre-wipe writer or sender can still publish data. |
| 21 | //! |
| 22 | //! Appenders re-open per append, so a compaction rewrite cannot leave anyone |
| 23 | //! writing to a stale inode. |
| 24 | |
| 25 | use std::fs::{self, DirBuilder, File, OpenOptions}; |
| 26 | use std::io::{Read as _, Write as _}; |
| 27 | use std::path::{Path, PathBuf}; |
| 28 | |
| 29 | use anyhow::{Context, Result}; |
| 30 | |
| 31 | /// Newest events retained in either sink. |
| 32 | pub const MAX_EVENTS: usize = 512; |
| 33 | /// Byte ceiling for either sink. |
| 34 | pub const MAX_BYTES: u64 = 256 * 1024; |
| 35 | /// A single append must fit in one atomic `write(2)`. |
| 36 | pub const MAX_LINE_BYTES: usize = 4096; |
| 37 | |
| 38 | /// Exact contents of the opt-out tombstone observed by a consent decision. |
| 39 | /// |
| 40 | /// A fresh nonce is written when a machine transitions into an opted-out |
| 41 | /// period. Arming may clear only the exact generation its decision observed, |
| 42 | /// so an older consent token cannot erase a newer opt-out. |
| 43 | #[derive(Debug, Clone, PartialEq, Eq)] |
| 44 | pub(crate) struct TombstoneGeneration(Vec<u8>); |
| 45 | |
| 46 | const MAX_TOMBSTONE_BYTES: u64 = 128; |
| 47 | |
| 48 | /// Below this size a sink cannot possibly hold [`MAX_EVENTS`] lines, so an |
| 49 | /// append skips the count probe entirely. The shortest serializable event line |
| 50 | /// is well over 8 bytes, and `512 * 9 > 4096`, so this bound is safe by |
| 51 | /// construction — `probe_threshold_cannot_hide_an_over_cap_buffer` pins it. |
| 52 | const PROBE_BYTES: u64 = 4096; |
| 53 | |
| 54 | /// `buffer.jsonl` — the pending-event sink. |
| 55 | #[must_use] |
| 56 | pub fn buffer_path(root: &Path) -> PathBuf { |
| 57 | root.join("buffer.jsonl") |
| 58 | } |
| 59 | |
| 60 | /// `dryrun.jsonl` — where batches go when the endpoint resolves to `None`. |
| 61 | /// |
| 62 | /// Reached by configuring `telemetry_endpoint` empty; an unconfigured endpoint |
| 63 | /// resolves to `codewhale_config::DEFAULT_TELEMETRY_ENDPOINT` instead. |
| 64 | #[must_use] |
| 65 | pub fn dryrun_path(root: &Path) -> PathBuf { |
| 66 | root.join("dryrun.jsonl") |
| 67 | } |
| 68 | |
| 69 | /// `buffer.jsonl.lock` — the sibling lock file. Never the data file, and never |
| 70 | /// unlinked: replacing it would leave appenders and compactors holding |
| 71 | /// different inodes and serialising against nothing. |
| 72 | #[must_use] |
| 73 | pub fn lock_path(root: &Path) -> PathBuf { |
| 74 | root.join("buffer.jsonl.lock") |
| 75 | } |
| 76 | |
| 77 | /// `disabled` — the tombstone. |
| 78 | #[must_use] |
| 79 | pub fn tombstone_path(root: &Path) -> PathBuf { |
| 80 | root.join("disabled") |
| 81 | } |
| 82 | |
| 83 | /// `install_id.json`. |
| 84 | #[must_use] |
| 85 | pub fn install_id_path(root: &Path) -> PathBuf { |
| 86 | root.join("install_id.json") |
| 87 | } |
| 88 | |
| 89 | /// `state.json`. |
| 90 | #[must_use] |
| 91 | pub fn state_path(root: &Path) -> PathBuf { |
| 92 | root.join("state.json") |
| 93 | } |
| 94 | |
| 95 | /// Whether the tombstone is present. |
| 96 | /// |
| 97 | /// Re-checked on **every** append and immediately before **every** send. This is |
| 98 | /// what makes `codewhale config set telemetry false` — an external write by |
| 99 | /// another process — observable to a session that is already running. |
| 100 | #[must_use] |
| 101 | pub fn tombstone_present(root: &Path) -> bool { |
| 102 | tombstone_path(root).exists() |
| 103 | } |
| 104 | |
| 105 | /// Read the exact tombstone generation, or `None` when collection has never |
| 106 | /// been disabled in this home. |
| 107 | pub(crate) fn tombstone_generation(root: &Path) -> Result<Option<TombstoneGeneration>> { |
| 108 | let path = tombstone_path(root); |
| 109 | let file = match File::open(&path) { |
| 110 | Ok(file) => file, |
| 111 | Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), |
| 112 | Err(error) => { |
| 113 | return Err( |
| 114 | anyhow::Error::new(error).context(format!("failed to open {}", path.display())) |
| 115 | ); |
| 116 | } |
| 117 | }; |
| 118 | let mut bytes = Vec::new(); |
| 119 | file.take(MAX_TOMBSTONE_BYTES + 1) |
| 120 | .read_to_end(&mut bytes) |
| 121 | .with_context(|| format!("failed to read {}", path.display()))?; |
| 122 | if bytes.len() as u64 > MAX_TOMBSTONE_BYTES { |
| 123 | anyhow::bail!("{} exceeds the tombstone size limit", path.display()); |
| 124 | } |
| 125 | Ok(Some(TombstoneGeneration(bytes))) |
| 126 | } |
| 127 | |
| 128 | /// Create the telemetry directory `0700`, if it is missing. |
| 129 | pub fn ensure_dir(root: &Path) -> Result<()> { |
| 130 | if root.is_dir() { |
| 131 | return Ok(()); |
| 132 | } |
| 133 | let mut builder = DirBuilder::new(); |
| 134 | builder.recursive(true); |
| 135 | #[cfg(unix)] |
| 136 | { |
| 137 | use std::os::unix::fs::DirBuilderExt as _; |
| 138 | builder.mode(0o700); |
| 139 | } |
| 140 | builder |
| 141 | .create(root) |
| 142 | .with_context(|| format!("failed to create {}", root.display())) |
| 143 | } |
| 144 | |
| 145 | #[cfg(unix)] |
| 146 | fn secure(file: &File) -> Result<()> { |
| 147 | use std::os::unix::fs::PermissionsExt as _; |
| 148 | file.set_permissions(fs::Permissions::from_mode(0o600)) |
| 149 | .context("failed to restrict telemetry file permissions") |
| 150 | } |
| 151 | |
| 152 | #[cfg(not(unix))] |
| 153 | fn secure(_file: &File) -> Result<()> { |
| 154 | Ok(()) |
| 155 | } |
| 156 | |
| 157 | /// Append one serialized event or batch to `path`. |
| 158 | /// |
| 159 | /// Returns `None` — never an error — when the tombstone is present, the privacy |
| 160 | /// lock is held, the line would not fit in one atomic write, or any filesystem |
| 161 | /// step fails. The lock acquisition is non-blocking, including on the panic and |
| 162 | /// signal paths. |
| 163 | pub fn append(root: &Path, path: &Path, line: &str) -> Option<()> { |
| 164 | append_with_limit(root, path, line, MAX_LINE_BYTES) |
| 165 | } |
| 166 | |
| 167 | /// Append a line that is too large for one atomic `write(2)`, serialising |
| 168 | /// against other writers with the compaction lock instead. |
| 169 | /// |
| 170 | /// Only the dry-run sink uses this: a whole batch does not fit under |
| 171 | /// `PIPE_BUF`, and the flush path is neither the panic hook nor the signal |
| 172 | /// handler, so a **non-blocking** `try_write` is safe there. On contention the |
| 173 | /// batch is dropped, which is the same fail-open behavior as a failed POST. |
| 174 | pub fn append_locked(root: &Path, path: &Path, line: &str) -> Option<()> { |
| 175 | append_with_limit(root, path, line, MAX_BYTES as usize) |
| 176 | } |
| 177 | |
| 178 | fn append_with_limit(root: &Path, path: &Path, line: &str, limit: usize) -> Option<()> { |
| 179 | let bytes = line.as_bytes(); |
| 180 | if bytes.is_empty() || bytes.len() + 1 > limit { |
| 181 | return None; |
| 182 | } |
| 183 | |
| 184 | let mut buf = Vec::with_capacity(bytes.len() + 1); |
| 185 | buf.extend_from_slice(bytes); |
| 186 | buf.push(b'\n'); |
| 187 | |
| 188 | let wrote = try_with_lock(root, || append_under_lock(root, path, &buf)) |
| 189 | .ok() |
| 190 | .flatten() |
| 191 | .unwrap_or(false); |
| 192 | if !wrote { |
| 193 | return None; |
| 194 | } |
| 195 | |
| 196 | enforce_ring(root, path); |
| 197 | Some(()) |
| 198 | } |
| 199 | |
| 200 | /// Append bytes while the caller holds the sibling privacy lock. |
| 201 | fn append_under_lock(root: &Path, path: &Path, buf: &[u8]) -> Result<bool> { |
| 202 | if tombstone_present(root) { |
| 203 | return Ok(false); |
| 204 | } |
| 205 | let file = OpenOptions::new() |
| 206 | .create(true) |
| 207 | .append(true) |
| 208 | .open(path) |
| 209 | .with_context(|| format!("failed to open {}", path.display()))?; |
| 210 | secure(&file)?; |
| 211 | // One `write(2)`, not `write_fmt` and not two calls: a split write is what |
| 212 | // a concurrent appender would interleave with. |
| 213 | (&file) |
| 214 | .write_all(buf) |
| 215 | .with_context(|| format!("failed to append to {}", path.display()))?; |
| 216 | file.sync_data() |
| 217 | .with_context(|| format!("failed to sync {}", path.display()))?; |
| 218 | Ok(true) |
| 219 | } |
| 220 | |
| 221 | /// Keep the newest [`MAX_EVENTS`] lines and at most [`MAX_BYTES`], under the |
| 222 | /// compaction lock. On lock contention this cycle is skipped: the next append |
| 223 | /// tries again, and the cap is a ceiling on disk footprint, not an invariant |
| 224 | /// that must hold at every instant. |
| 225 | fn enforce_ring(root: &Path, path: &Path) { |
| 226 | let Ok(meta) = fs::metadata(path) else { |
| 227 | return; |
| 228 | }; |
| 229 | let len = meta.len(); |
| 230 | if len < PROBE_BYTES { |
| 231 | return; |
| 232 | } |
| 233 | let _ = try_with_lock(root, || { |
| 234 | // Re-read under the same lock as wipe. Reusing a snapshot captured |
| 235 | // before a concurrent wipe would resurrect the records it truncated. |
| 236 | if tombstone_present(root) { |
| 237 | return Ok(()); |
| 238 | } |
| 239 | let Ok(meta) = fs::metadata(path) else { |
| 240 | return Ok(()); |
| 241 | }; |
| 242 | let len = meta.len(); |
| 243 | if len < PROBE_BYTES { |
| 244 | return Ok(()); |
| 245 | } |
| 246 | let contents = fs::read_to_string(path) |
| 247 | .with_context(|| format!("failed to read {}", path.display()))?; |
| 248 | let lines: Vec<&str> = contents.lines().filter(|l| !l.trim().is_empty()).collect(); |
| 249 | if lines.len() <= MAX_EVENTS && len <= MAX_BYTES { |
| 250 | return Ok(()); |
| 251 | } |
| 252 | let mut kept: Vec<&str> = lines |
| 253 | .iter() |
| 254 | .rev() |
| 255 | .take(MAX_EVENTS) |
| 256 | .rev() |
| 257 | .copied() |
| 258 | .collect::<Vec<_>>(); |
| 259 | // Byte ceiling second: drop from the oldest end until the survivors fit. |
| 260 | while kept.len() > 1 && byte_len(&kept) > MAX_BYTES { |
| 261 | kept.remove(0); |
| 262 | } |
| 263 | let mut body = kept.join("\n"); |
| 264 | if !body.is_empty() { |
| 265 | body.push('\n'); |
| 266 | } |
| 267 | rewrite(path, body.as_bytes()) |
| 268 | }); |
| 269 | } |
| 270 | |
| 271 | fn byte_len(lines: &[&str]) -> u64 { |
| 272 | lines.iter().map(|l| l.len() as u64 + 1).sum() |
| 273 | } |
| 274 | |
| 275 | /// Replace `path` atomically through a sibling temp file in the same directory. |
| 276 | fn rewrite(path: &Path, bytes: &[u8]) -> Result<()> { |
| 277 | let dir = path.parent().unwrap_or_else(|| Path::new(".")); |
| 278 | let mut tmp = tempfile::NamedTempFile::new_in(dir) |
| 279 | .with_context(|| format!("failed to stage a rewrite of {}", path.display()))?; |
| 280 | tmp.write_all(bytes) |
| 281 | .with_context(|| format!("failed to write a rewrite of {}", path.display()))?; |
| 282 | tmp.flush() |
| 283 | .with_context(|| format!("failed to flush a rewrite of {}", path.display()))?; |
| 284 | secure(tmp.as_file())?; |
| 285 | tmp.persist(path) |
| 286 | .map_err(|error| error.error) |
| 287 | .with_context(|| format!("failed to persist {}", path.display()))?; |
| 288 | Ok(()) |
| 289 | } |
| 290 | |
| 291 | /// Open (creating if needed) the sibling lock file. |
| 292 | fn open_lock(root: &Path) -> Result<File> { |
| 293 | ensure_dir(root)?; |
| 294 | let path = lock_path(root); |
| 295 | let file = OpenOptions::new() |
| 296 | .create(true) |
| 297 | .read(true) |
| 298 | .write(true) |
| 299 | // The file is only a lock handle; its contents are never read and |
| 300 | // truncating it would race other holders for no benefit. |
| 301 | .truncate(false) |
| 302 | .open(&path) |
| 303 | .with_context(|| format!("failed to open {}", path.display()))?; |
| 304 | secure(&file)?; |
| 305 | Ok(file) |
| 306 | } |
| 307 | |
| 308 | /// Run `operation` holding the exclusive compaction lock, **blocking**. |
| 309 | /// |
| 310 | /// Only the opt-out wipe uses this. It is not an exit path, so blocking is |
| 311 | /// fine there and nowhere else. |
| 312 | pub fn with_lock<T>(root: &Path, operation: impl FnOnce() -> Result<T>) -> Result<T> { |
| 313 | let file = open_lock(root)?; |
| 314 | let mut lock = fd_lock::RwLock::new(file); |
| 315 | let _guard = lock.write().context("failed to take the telemetry lock")?; |
| 316 | operation() |
| 317 | } |
| 318 | |
| 319 | /// Run `operation` holding the exclusive compaction lock if it is free. |
| 320 | /// |
| 321 | /// Returns `Ok(None)` when the lock is held elsewhere. Never blocks. |
| 322 | pub fn try_with_lock<T>(root: &Path, operation: impl FnOnce() -> Result<T>) -> Result<Option<T>> { |
| 323 | let file = open_lock(root)?; |
| 324 | let mut lock = fd_lock::RwLock::new(file); |
| 325 | match lock.try_write() { |
| 326 | Ok(_guard) => operation().map(Some), |
| 327 | Err(_) => Ok(None), |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | /// Read every intact line from `path`, dropping a torn trailing line. |
| 332 | /// |
| 333 | /// `std::process::exit` on the signal path can truncate a concurrent write, so |
| 334 | /// the last line may be a partial JSON document. Skipping unparseable lines is |
| 335 | /// the whole tolerance: a drain must never fail because one record was cut. |
| 336 | #[must_use] |
| 337 | pub fn read_lines(path: &Path) -> Vec<String> { |
| 338 | let Ok(contents) = fs::read_to_string(path) else { |
| 339 | return Vec::new(); |
| 340 | }; |
| 341 | contents |
| 342 | .lines() |
| 343 | .filter(|line| !line.trim().is_empty()) |
| 344 | .map(str::to_string) |
| 345 | .collect() |
| 346 | } |
| 347 | |
| 348 | /// Take every buffered line and truncate the buffer, under the compaction lock. |
| 349 | /// |
| 350 | /// Returns an empty vector when the tombstone is present or the lock is held |
| 351 | /// elsewhere. Truncates rather than unlinks — `crates/tui/src/fleet/ledger.rs` |
| 352 | /// documents the rule: replacing the file leaves appenders holding the old |
| 353 | /// inode. |
| 354 | #[must_use] |
| 355 | pub fn drain(root: &Path) -> Vec<String> { |
| 356 | if tombstone_present(root) { |
| 357 | return Vec::new(); |
| 358 | } |
| 359 | let path = buffer_path(root); |
| 360 | let drained = try_with_lock(root, || { |
| 361 | // Re-check under the lock: a wipe may have landed between the check |
| 362 | // above and the acquisition. |
| 363 | if tombstone_present(root) { |
| 364 | return Ok(Vec::new()); |
| 365 | } |
| 366 | let lines = read_lines(&path); |
| 367 | if !lines.is_empty() { |
| 368 | truncate(&path)?; |
| 369 | } |
| 370 | Ok(lines) |
| 371 | }); |
| 372 | drained.ok().flatten().unwrap_or_default() |
| 373 | } |
| 374 | |
| 375 | /// Truncate a file to zero length, leaving the inode in place. A missing file |
| 376 | /// is not an error. |
| 377 | pub fn truncate(path: &Path) -> Result<()> { |
| 378 | if !path.exists() { |
| 379 | return Ok(()); |
| 380 | } |
| 381 | let file = OpenOptions::new() |
| 382 | .write(true) |
| 383 | .truncate(true) |
| 384 | .open(path) |
| 385 | .with_context(|| format!("failed to truncate {}", path.display()))?; |
| 386 | secure(&file)?; |
| 387 | Ok(()) |
| 388 | } |
| 389 | |
| 390 | /// Wipe every trace of collection, leaving a permanent tombstone. |
| 391 | /// |
| 392 | /// Order matters and is the whole of the guarantee: |
| 393 | /// |
| 394 | /// 1. take the blocking lock — this is not an exit path; |
| 395 | /// 2. write the tombstone **first**, and never remove it here; |
| 396 | /// 3. truncate `buffer.jsonl` and `dryrun.jsonl` — do **not** unlink them, and |
| 397 | /// never unlink the lock file; |
| 398 | /// 4. remove `install_id.json` and `state.json`. |
| 399 | /// |
| 400 | /// If any step after the tombstone fails, the error is returned and the caller |
| 401 | /// logs it — but the tombstone alone already makes the buffer permanently |
| 402 | /// undrainable, so a failed wipe fails **closed**. |
| 403 | pub fn wipe(root: &Path) -> Result<()> { |
| 404 | with_lock(root, || { |
| 405 | let tombstone = tombstone_path(root); |
| 406 | // A tombstone generation identifies one durable opted-out period. Keep |
| 407 | // it stable across later launches that re-observe the same persistent |
| 408 | // choice; re-enable removes the file, so the next real opt-out creates |
| 409 | // a naturally distinct generation. An unreadable or oversized file is |
| 410 | // repaired in the fail-closed direction by replacing it here; legacy |
| 411 | // empty tombstones remain valid stable generations. |
| 412 | if tombstone_generation(root).ok().flatten().is_none() { |
| 413 | let mut file = OpenOptions::new() |
| 414 | .create(true) |
| 415 | .write(true) |
| 416 | .truncate(true) |
| 417 | .open(&tombstone) |
| 418 | .with_context(|| format!("failed to write {}", tombstone.display()))?; |
| 419 | secure(&file)?; |
| 420 | file.write_all(uuid::Uuid::new_v4().to_string().as_bytes()) |
| 421 | .with_context(|| format!("failed to write {}", tombstone.display()))?; |
| 422 | file.sync_data() |
| 423 | .with_context(|| format!("failed to sync {}", tombstone.display()))?; |
| 424 | drop(file); |
| 425 | } |
| 426 | |
| 427 | let mut failure: Option<anyhow::Error> = None; |
| 428 | for path in [buffer_path(root), dryrun_path(root)] { |
| 429 | if let Err(error) = truncate(&path) { |
| 430 | failure.get_or_insert(error); |
| 431 | } |
| 432 | } |
| 433 | for path in [install_id_path(root), state_path(root)] { |
| 434 | if path.exists() |
| 435 | && let Err(error) = fs::remove_file(&path) |
| 436 | { |
| 437 | failure.get_or_insert( |
| 438 | anyhow::Error::new(error) |
| 439 | .context(format!("failed to remove {}", path.display())), |
| 440 | ); |
| 441 | } |
| 442 | } |
| 443 | match failure { |
| 444 | Some(error) => Err(error), |
| 445 | None => Ok(()), |
| 446 | } |
| 447 | }) |
| 448 | } |
| 449 | |
| 450 | /// Clear the tombstone and drop anything buffered before this process was |
| 451 | /// permitted. |
| 452 | /// |
| 453 | /// Called by `init` on every arming. Both the exact tombstone generation and a |
| 454 | /// fresh durable permission check must still match while the wipe lock is held. |
| 455 | /// A stale buffer left by an earlier run or by a bug cannot enter the new |
| 456 | /// process's batch. |
| 457 | pub(crate) fn arm( |
| 458 | root: &Path, |
| 459 | observed_generation: Option<&TombstoneGeneration>, |
| 460 | permission_still_enabled: impl FnOnce() -> bool, |
| 461 | ) -> Result<()> { |
| 462 | ensure_dir(root)?; |
| 463 | with_lock(root, || { |
| 464 | let current_generation = tombstone_generation(root)?; |
| 465 | if current_generation.as_ref() != observed_generation { |
| 466 | anyhow::bail!("telemetry permission changed before arming"); |
| 467 | } |
| 468 | if !permission_still_enabled() { |
| 469 | anyhow::bail!("telemetry permission is no longer enabled"); |
| 470 | } |
| 471 | let tombstone = tombstone_path(root); |
| 472 | if tombstone.exists() { |
| 473 | fs::remove_file(&tombstone) |
| 474 | .with_context(|| format!("failed to remove {}", tombstone.display()))?; |
| 475 | } |
| 476 | truncate(&buffer_path(root)) |
| 477 | }) |
| 478 | } |
| 479 |