crates/telemetry/src/actor.rs
Rama main · réplica de lectura del repositorio original.
//! The background writer.
//!
//! One dedicated OS thread behind a bounded `mpsc`. `record()` is a
//! non-blocking `try_send` and a hard no-op when the process is not armed;
//! everything the thread does is wrapped in `catch_unwind`, so a panic inside
//! telemetry costs telemetry and nothing else.
//!
//! A plain thread rather than a `tokio` task, deliberately: `init` is called
//! from six subcommand dispatch points, several of which have no runtime yet,
//! and a telemetry subsystem that only works when someone remembered to be
//! inside an executor is a subsystem that silently collects nothing on half its
//! surfaces.
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender, sync_channel};
use std::time::{Duration, Instant};
use crate::buffer;
use crate::client::{self, SendOutcome};
use crate::decision::{self, TelemetryDecision};
use crate::envelope;
use crate::event::{Batch, Event, SCHEMA_VERSION, Surface};
/// Events per batch.
pub const BATCH_MAX_EVENTS: usize = 200;
/// Byte ceiling per batch body.
pub const BATCH_MAX_BYTES: usize = 64 * 1024;
pub(crate) enum Message {
Event(Box<Event>),
PersistLocal(SyncSender<FlushOutcome>),
Shutdown(SyncSender<FlushOutcome>),
}
/// What a flush attempt did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FlushOutcome {
/// Nothing was buffered.
Empty,
/// Events remain in the local buffer for a later network flush.
Buffered,
/// A batch was written to the dry-run sink.
DryRun,
/// A batch was accepted by the endpoint.
Sent,
/// A batch was assembled and dropped — offline, refused, or contended.
Dropped,
/// Telemetry was off by the time the flush ran; nothing was sent.
Suppressed,
/// The actor did not answer inside the caller's deadline.
TimedOut,
}
/// Facts the writer thread needs, fixed at arming time.
#[derive(Debug, Clone)]
pub(crate) struct Context {
pub root: PathBuf,
pub endpoint: Option<String>,
pub surface: Surface,
pub config_path: Option<PathBuf>,
pub app_version: String,
pub git_sha: Option<String>,
pub tty: bool,
}
/// A handle to the writer thread.
#[derive(Debug)]
pub(crate) struct Handle {
tx: SyncSender<Message>,
pending: Arc<AtomicUsize>,
closing: AtomicBool,
}
impl Handle {
/// Start the writer thread.
pub(crate) fn spawn(context: Context) -> Self {
let (tx, rx) = sync_channel::<Message>(buffer::MAX_EVENTS + 1);
let pending = Arc::new(AtomicUsize::new(0));
let worker_pending = pending.clone();
// A detached thread: nothing joins it, and the process exiting while it
// is mid-write is exactly the case the torn-line tolerance covers.
let _ = std::thread::Builder::new()
.name("codewhale-telemetry".to_string())
.spawn(move || run(&context, &rx, &worker_pending));
Self {
tx,
pending,
closing: AtomicBool::new(false),
}
}
/// Queue an event without blocking; drop it when full, closing or closed.
pub(crate) fn record(&self, event: Event) {
if self.closing.load(Ordering::Acquire) {
return;
}
let mut pending = self.pending.load(Ordering::Acquire);
loop {
if pending >= buffer::MAX_EVENTS {
return;
}
match self.pending.compare_exchange_weak(
pending,
pending + 1,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => break,
Err(current) => pending = current,
}
}
if self.closing.load(Ordering::Acquire)
|| self.tx.try_send(Message::Event(Box::new(event))).is_err()
{
self.pending.fetch_sub(1, Ordering::AcqRel);
}
}
/// Ask for a final flush and stop the thread.
pub(crate) fn shutdown(&self, deadline: Duration) -> FlushOutcome {
self.round_trip(deadline, Message::Shutdown)
}
/// Persist queued events locally and stop without making a network request.
///
/// Unlike [`Handle::shutdown`], this waits without a deadline: the local
/// path performs only bounded disk work, and racing it against a clock is
/// what lost dry-run receipts on slow hosts (#6269). The writer always
/// acknowledges — even on panic — so the only way out without an outcome
/// is a writer that is already gone, which fails open as `TimedOut`.
pub(crate) fn persist_local(&self) -> FlushOutcome {
if self.closing.swap(true, Ordering::AcqRel) {
return FlushOutcome::TimedOut;
}
let (ack_tx, ack_rx) = sync_channel::<FlushOutcome>(1);
if self.tx.try_send(Message::PersistLocal(ack_tx)).is_err() {
return FlushOutcome::TimedOut;
}
ack_rx.recv().unwrap_or(FlushOutcome::TimedOut)
}
fn round_trip(
&self,
deadline: Duration,
build: impl FnOnce(SyncSender<FlushOutcome>) -> Message,
) -> FlushOutcome {
let started = Instant::now();
if self.closing.swap(true, Ordering::AcqRel) {
return FlushOutcome::TimedOut;
}
let (ack_tx, ack_rx) = sync_channel::<FlushOutcome>(1);
if self.tx.try_send(build(ack_tx)).is_err() {
return FlushOutcome::TimedOut;
}
match ack_rx.recv_timeout(deadline.saturating_sub(started.elapsed())) {
Ok(outcome) => outcome,
Err(RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected) => {
FlushOutcome::TimedOut
}
}
}
}
fn run(context: &Context, rx: &Receiver<Message>, pending: &AtomicUsize) {
while let Ok(message) = rx.recv() {
// Telemetry never takes the process with it. The hook has already been
// installed by the time this thread exists, so a panic here is caught,
// dropped, and the loop continues.
let result = catch_unwind(AssertUnwindSafe(|| match message {
Message::Event(event) => {
pending.fetch_sub(1, Ordering::AcqRel);
append(context, &event);
None
}
Message::PersistLocal(ack) => {
// The caller joins this without a deadline (#6269), so the
// acknowledgement must be unconditional: a panic here would
// otherwise hang a short CLI command at exit. The outer
// `catch_unwind` still guards the other arms.
let outcome = match catch_unwind(AssertUnwindSafe(|| persist_local(context))) {
Ok(outcome) => outcome,
Err(_) => {
tracing::debug!("telemetry local persist recovered from a panic");
FlushOutcome::Dropped
}
};
let _ = ack.send(outcome);
Some(())
}
Message::Shutdown(ack) => {
let _ = ack.send(flush(context));
Some(())
}
}));
match result {
Ok(Some(())) => return,
Ok(None) => {}
Err(_) => {
tracing::debug!("telemetry writer recovered from a panic");
}
}
}
}
fn append(context: &Context, event: &Event) {
let Ok(line) = serde_json::to_string(event) else {
return;
};
let path = buffer::buffer_path(&context.root);
let _ = buffer::append(&context.root, &path, &line);
}
/// Seal every event queued before this message without reaching the network.
///
/// The channel is FIFO, so all prior event appends have settled before this
/// runs. Re-deciding here preserves the same mid-session opt-out boundary as a
/// full flush. An explicitly empty endpoint is the local dry-run sink and can
/// therefore be finalized immediately; a configured endpoint leaves the
/// events in `buffer.jsonl` for the next interactive flush.
fn persist_local(context: &Context) -> FlushOutcome {
match decision::re_decide(context.config_path.as_deref(), context.surface) {
TelemetryDecision::Enabled(_) => {}
TelemetryDecision::OptedOut | TelemetryDecision::ForcedOff => {
return FlushOutcome::Suppressed;
}
}
if buffer::tombstone_present(&context.root) {
return FlushOutcome::Suppressed;
}
if context.endpoint.is_none() {
return flush(context);
}
if buffer::read_lines(&buffer::buffer_path(&context.root)).is_empty() {
FlushOutcome::Empty
} else {
FlushOutcome::Buffered
}
}
/// Drain, re-check consent, and deliver.
///
/// The re-check is the point: telemetry is resolved once at init, but the
/// documented mid-session opt-out is an external file write this process would
/// otherwise never observe. If the answer is now `OptedOut`, `decide` has
/// already wiped and left the tombstone, and the drained events go nowhere.
fn flush(context: &Context) -> FlushOutcome {
match decision::re_decide(context.config_path.as_deref(), context.surface) {
TelemetryDecision::Enabled(_) => {}
TelemetryDecision::OptedOut | TelemetryDecision::ForcedOff => {
return FlushOutcome::Suppressed;
}
}
if buffer::tombstone_present(&context.root) {
return FlushOutcome::Suppressed;
}
let lines = buffer::drain(&context.root);
if lines.is_empty() {
return FlushOutcome::Empty;
}
let events = parse_events(&lines);
if events.is_empty() {
return FlushOutcome::Empty;
}
let Ok(install) = envelope::read_or_create_install_id(&context.root) else {
return FlushOutcome::Dropped;
};
let mut state = envelope::read_state(&context.root);
state.schema_version = SCHEMA_VERSION;
state.last_flush = Some(envelope::now_rfc3339());
// Written on attempt, not on success, so a permanently offline machine
// attempts at most once per interval rather than on every launch.
let _ = envelope::write_state(&context.root, &state);
let batch = Batch {
schema_version: SCHEMA_VERSION,
notice_version: crate::event::NOTICE_VERSION,
sent_at: envelope::now_rfc3339(),
install_id: install.install_id,
app_version: context.app_version.clone(),
git_sha: context.git_sha.clone(),
surface: context.surface,
os: envelope::current_os(),
arch: envelope::current_arch(),
libc: envelope::current_libc(),
tty: context.tty,
events,
};
match client::send(&context.root, context.endpoint.as_deref(), &batch) {
SendOutcome::DryRun => FlushOutcome::DryRun,
SendOutcome::Accepted => FlushOutcome::Sent,
SendOutcome::Dropped => FlushOutcome::Dropped,
}
}
/// Parse drained lines into events, capped at [`BATCH_MAX_EVENTS`] and
/// [`BATCH_MAX_BYTES`], skipping anything that does not parse **or does not
/// satisfy its declared string bounds**.
///
/// The bound re-check is the point. Everything upstream of here builds events
/// from closed enums, `u32`s, and two reducers — but this function is a
/// deserializer, and its input is a file on disk that any process running as
/// the user can append to. `Event::is_bounded` is what stops
/// `{"event":"panic","site":"<anything at all>"}` from becoming a first-party
/// POST under the user's install id.
pub(crate) fn parse_events(lines: &[String]) -> Vec<Event> {
let mut events = Vec::new();
let mut bytes = 0usize;
for line in lines {
if events.len() >= BATCH_MAX_EVENTS || bytes + line.len() > BATCH_MAX_BYTES {
break;
}
if let Ok(event) = serde_json::from_str::<Event>(line) {
if !event.is_bounded() {
tracing::debug!("telemetry dropped an out-of-bounds buffered event");
continue;
}
bytes += line.len();
events.push(event);
}
}
events
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn local_persistence_joins_the_writer_without_a_deadline() {
// The caller blocks on a mock writer that answers on its own
// schedule: no sleep, no timing assertion, no race with the clock.
let (tx, rx) = sync_channel(buffer::MAX_EVENTS + 1);
let handle = Handle {
tx,
pending: Arc::new(AtomicUsize::new(0)),
closing: AtomicBool::new(false),
};
handle.record(Event::SessionStart {
source: crate::SessionSource::Unknown,
});
let waiter = std::thread::spawn(move || handle.persist_local());
assert!(matches!(rx.try_recv(), Ok(Message::Event(_))));
let Message::PersistLocal(ack) = rx.recv().expect("local persistence request") else {
panic!("short CLI persistence must not request a network shutdown flush");
};
ack.send(FlushOutcome::Buffered)
.expect("the caller is still joined");
assert_eq!(waiter.join().expect("waiter"), FlushOutcome::Buffered);
assert!(
rx.try_recv().is_err(),
"one persist request enqueues exactly one message"
);
}
#[test]
fn local_persistence_fails_open_when_the_writer_is_gone() {
// The writer thread exits after a persist or shutdown request; a
// second caller must fail open, not hang on a dead receiver. Send
// fails deterministically here, so no timing assertion is needed.
let (tx, rx) = sync_channel::<Message>(buffer::MAX_EVENTS + 1);
drop(rx);
let handle = Handle {
tx,
pending: Arc::new(AtomicUsize::new(0)),
closing: AtomicBool::new(false),
};
assert_eq!(handle.persist_local(), FlushOutcome::TimedOut);
}
}