Menu
Public repository / master
AkurAI-Framework / crates/telemetry/src/lib.rs
Latest change 91bd2a37654909c5a3aef6ddd614991bd378ccb8 - feat: add shared lifecycle telemetry contract by Ólafur Búi Ólafsson
//! Shared lifecycle telemetry contract for AkurAI applications.
//!
//! The crate intentionally contains only a versioned, transport-neutral envelope
//! and a small `std`-only delivery adapter. Applications choose their receiver;
//! the request path only validates and attempts a non-blocking enqueue.
//!
//! ```
//! use akurai_telemetry::{fixtures, PrivacyClassification, TelemetryContext};
//!
//! let context = TelemetryContext::new("org-1", "workspace-1", "install-1", "drive");
//! let event = fixtures::first_value("evt-1", &context, 1_700_000_000).unwrap();
//! assert_eq!(event.privacy, PrivacyClassification::Operational);
//! ```
#![forbid(unsafe_code)]
use std::collections::{HashMap, VecDeque};
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{self, RecvTimeoutError, SyncSender, TrySendError};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
pub const SCHEMA_VERSION: u16 = 1;
/// Maximum amount by which a producer timestamp may be ahead of this host.
pub const MAX_FUTURE_SECONDS: i64 = 7 * 24 * 60 * 60;
/// Default bound used by [`TelemetryDelivery::shutdown`].
pub const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
const MAX_IDENTIFIER_LEN: usize = 128;
const MAX_EVENT_NAME_LEN: usize = 128;
const MAX_ATTRIBUTE_COUNT: usize = 32;
const MAX_ATTRIBUTE_KEY_LEN: usize = 64;
const MAX_ATTRIBUTE_VALUE_LEN: usize = 256;
/// Privacy classifications accepted by the shared boundary.
///
/// `ApprovedPii` is intentionally explicit. Secrets and content are never
/// accepted, while unapproved PII is rejected unless a caller opts into the
/// approved classification for that event.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum PrivacyClassification {
/// Operational metadata with no personal data.
Operational,
/// Personal data that has been approved for this event contract.
ApprovedPii,
/// Personal data without an explicit approval.
UnapprovedPii,
/// User/application content, which is outside lifecycle telemetry.
Content,
/// Secrets or credentials, which must never enter telemetry.
Secret,
}
/// An identifier shared by all consumers of the contract.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TelemetryContext {
pub organization_id: String,
pub workspace_id: String,
pub installation_id: String,
pub app_id: String,
}
impl TelemetryContext {
/// Build the customer linkage portion of an envelope.
pub fn new(
organization_id: impl Into<String>,
workspace_id: impl Into<String>,
installation_id: impl Into<String>,
app_id: impl Into<String>,
) -> Self {
Self {
organization_id: organization_id.into(),
workspace_id: workspace_id.into(),
installation_id: installation_id.into(),
app_id: app_id.into(),
}
}
}
/// A schema-versioned, deduplicable lifecycle event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TelemetryEnvelope {
/// A stable producer-generated idempotency key.
pub event_id: String,
pub schema_version: u16,
pub organization_id: String,
pub workspace_id: String,
pub installation_id: String,
pub app_id: String,
pub event_name: String,
/// Unix timestamp in seconds supplied by the producer.
pub occurred_at: i64,
pub privacy: PrivacyClassification,
/// Bounded, non-content dimensions. Do not place request bodies or secrets here.
pub attributes: Vec<(String, String)>,
}
impl TelemetryEnvelope {
/// Construct and validate an envelope at the producer boundary.
pub fn new(
event_id: impl Into<String>,
context: &TelemetryContext,
event_name: impl Into<String>,
occurred_at: i64,
privacy: PrivacyClassification,
attributes: Vec<(String, String)>,
) -> Result<Self, TelemetryError> {
let envelope = Self {
event_id: event_id.into(),
schema_version: SCHEMA_VERSION,
organization_id: context.organization_id.clone(),
workspace_id: context.workspace_id.clone(),
installation_id: context.installation_id.clone(),
app_id: context.app_id.clone(),
event_name: event_name.into(),
occurred_at,
privacy,
attributes,
};
envelope.validate()?;
Ok(envelope)
}
/// Validate a decoded or externally assembled envelope before delivery.
pub fn validate(&self) -> Result<(), TelemetryError> {
if self.schema_version != SCHEMA_VERSION {
return Err(TelemetryError::UnsupportedSchemaVersion {
received: self.schema_version,
current: SCHEMA_VERSION,
});
}
validate_identifier("event_id", &self.event_id)?;
validate_identifier("organization_id", &self.organization_id)?;
validate_identifier("workspace_id", &self.workspace_id)?;
validate_identifier("installation_id", &self.installation_id)?;
validate_identifier("app_id", &self.app_id)?;
validate_event_name(&self.event_name)?;
if self.occurred_at < 0 {
return Err(TelemetryError::InvalidField {
field: "occurred_at",
reason: "must be a non-negative Unix timestamp",
});
}
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_secs().min(i64::MAX as u64) as i64);
if self.occurred_at > now.saturating_add(MAX_FUTURE_SECONDS) {
return Err(TelemetryError::InvalidField {
field: "occurred_at",
reason: "must not be more than seven days in the future",
});
}
if matches!(
self.privacy,
PrivacyClassification::UnapprovedPii
| PrivacyClassification::Content
| PrivacyClassification::Secret
) {
return Err(TelemetryError::PrivacyRejected {
classification: self.privacy,
field: "privacy",
});
}
if self.attributes.len() > MAX_ATTRIBUTE_COUNT {
return Err(TelemetryError::InvalidField {
field: "attributes",
reason: "too many attributes",
});
}
for (key, value) in &self.attributes {
validate_attribute(key, value, self.privacy)?;
}
Ok(())
}
}
fn validate_identifier(field: &'static str, value: &str) -> Result<(), TelemetryError> {
if value.is_empty() || value.len() > MAX_IDENTIFIER_LEN {
return Err(TelemetryError::InvalidField {
field,
reason: "must be non-empty and at most 128 bytes",
});
}
if !value.is_ascii()
|| value.chars().any(char::is_whitespace)
|| value.chars().any(char::is_control)
{
return Err(TelemetryError::InvalidField {
field,
reason: "must contain only printable ASCII without whitespace",
});
}
Ok(())
}
fn validate_event_name(value: &str) -> Result<(), TelemetryError> {
if value.is_empty() || value.len() > MAX_EVENT_NAME_LEN {
return Err(TelemetryError::InvalidField {
field: "event_name",
reason: "must be non-empty and at most 128 bytes",
});
}
if !value.is_ascii()
|| value
.bytes()
.any(|byte| !(byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')))
{
return Err(TelemetryError::InvalidField {
field: "event_name",
reason: "must contain only ASCII letters, numbers, '.', '_' or '-'",
});
}
Ok(())
}
fn validate_attribute(
key: &str,
value: &str,
privacy: PrivacyClassification,
) -> Result<(), TelemetryError> {
if key.is_empty() || key.len() > MAX_ATTRIBUTE_KEY_LEN {
return Err(TelemetryError::InvalidField {
field: "attributes",
reason: "attribute keys must be non-empty and at most 64 bytes",
});
}
if !key.is_ascii()
|| key
.bytes()
.any(|byte| !(byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')))
{
return Err(TelemetryError::InvalidField {
field: "attributes",
reason: "attribute keys must contain only ASCII letters, numbers, '.', '_' or '-'",
});
}
if value.len() > MAX_ATTRIBUTE_VALUE_LEN || value.chars().any(char::is_control) {
return Err(TelemetryError::InvalidField {
field: "attributes",
reason: "attribute values must be at most 256 bytes without control characters",
});
}
let key_lower = key.to_ascii_lowercase();
let value_lower = value.to_ascii_lowercase();
if contains_marker(&key_lower, &SECRET_MARKERS)
|| contains_marker(&value_lower, &SECRET_MARKERS)
|| looks_like_secret_value(&value_lower)
{
return Err(TelemetryError::PrivacyRejected {
classification: PrivacyClassification::Secret,
field: "attributes",
});
}
if contains_marker(&key_lower, &CONTENT_MARKERS)
|| contains_marker(&value_lower, &CONTENT_MARKERS)
{
return Err(TelemetryError::PrivacyRejected {
classification: PrivacyClassification::Content,
field: "attributes",
});
}
let looks_like_pii = key_lower == "ip"
|| contains_marker(&key_lower, &PII_MARKERS)
|| contains_marker(&value_lower, &PII_MARKERS)
|| looks_like_email(&value_lower)
|| looks_like_ipv4(&value_lower);
if looks_like_pii && privacy != PrivacyClassification::ApprovedPii {
return Err(TelemetryError::PrivacyRejected {
classification: PrivacyClassification::UnapprovedPii,
field: "attributes",
});
}
if value.chars().any(char::is_whitespace) {
return Err(TelemetryError::InvalidField {
field: "attributes",
reason: "attribute values must be bounded dimension values without whitespace",
});
}
Ok(())
}
fn contains_marker(value: &str, markers: &[&str]) -> bool {
markers.iter().any(|marker| value.contains(marker))
}
const SECRET_MARKERS: [&str; 14] = [
"secret",
"password",
"credential",
"token",
"api_key",
"authorization",
"cookie",
"bearer",
"private_key",
"access_key",
"client_secret",
"jwt",
"sk-",
"pk_",
];
const CONTENT_MARKERS: [&str; 8] = [
"content", "body", "prompt", "message", "document", "text", "payload", "markdown",
];
const PII_MARKERS: [&str; 9] = [
"email",
"phone",
"address",
"user_name",
"username",
"person",
"user_id",
"full_name",
"ip_address",
];
fn looks_like_secret_value(value: &str) -> bool {
value.contains("-----begin") || value.contains("token=") || value.contains("password=")
}
fn looks_like_email(value: &str) -> bool {
let Some((local, domain)) = value.split_once('@') else {
return false;
};
!local.is_empty() && domain.contains('.') && !domain.starts_with('.') && !domain.ends_with('.')
}
fn looks_like_ipv4(value: &str) -> bool {
let parts: Vec<_> = value.split('.').collect();
parts.len() == 4
&& parts.iter().all(|part| {
!part.is_empty() && part.len() <= 3 && part.bytes().all(|byte| byte.is_ascii_digit())
})
}
/// Errors produced before an event enters the asynchronous queue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TelemetryError {
/// The event cannot be represented by the current contract.
InvalidField {
field: &'static str,
reason: &'static str,
},
/// A producer attempted to use an incompatible schema version.
UnsupportedSchemaVersion { received: u16, current: u16 },
/// The privacy boundary rejected the classification or an attribute key.
PrivacyRejected {
classification: PrivacyClassification,
field: &'static str,
},
/// A zero-capacity queue cannot provide asynchronous buffering.
InvalidCapacity,
}
impl fmt::Display for TelemetryError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidField { field, reason } => write!(formatter, "invalid {field}: {reason}"),
Self::UnsupportedSchemaVersion { received, current } => {
write!(
formatter,
"unsupported schema version {received}; current is {current}"
)
}
Self::PrivacyRejected {
classification,
field,
} => {
write!(
formatter,
"privacy classification {classification:?} rejected for {field}"
)
}
Self::InvalidCapacity => {
formatter.write_str("telemetry capacity must be greater than zero")
}
}
}
}
impl std::error::Error for TelemetryError {}
/// A receiver implemented by Drive, Notes, Tasks, Mail, or another service.
///
/// The worker invokes this method off the request path. Receiver errors are
/// intentionally not returned to producers: an unavailable telemetry target
/// must not fail a customer request.
pub trait TelemetryReceiver: Send + Sync + 'static {
fn receive(&self, event: TelemetryEnvelope) -> Result<(), ReceiverError>;
}
/// A receiver-side failure, retained only for receiver logging/metrics.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReceiverError {
message: String,
}
impl ReceiverError {
pub fn new(message: impl Into<String>) -> Self {
Self {
message: message.into(),
}
}
/// Return the receiver-safe error text for metrics or diagnostics.
pub fn message(&self) -> &str {
&self.message
}
}
impl fmt::Display for ReceiverError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ReceiverError {}
struct SeenIds {
ids: HashMap<String, u64>,
order: VecDeque<(String, u64)>,
capacity: usize,
sequence: u64,
}
impl SeenIds {
fn new(capacity: usize) -> Self {
Self {
ids: HashMap::new(),
order: VecDeque::new(),
capacity,
sequence: 0,
}
}
fn reserve(&mut self, event_id: &str) -> bool {
if self.ids.contains_key(event_id) {
return false;
}
self.sequence = self.sequence.wrapping_add(1);
let owned = event_id.to_owned();
let sequence = self.sequence;
self.ids.insert(owned.clone(), sequence);
self.order.push_back((owned, sequence));
while self.order.len() > self.capacity {
if let Some((evicted, sequence)) = self.order.pop_front() {
if self.ids.get(&evicted) == Some(&sequence) {
self.ids.remove(&evicted);
}
}
}
true
}
fn release(&mut self, event_id: &str) {
self.ids.remove(event_id);
self.order.retain(|(id, _)| id != event_id);
}
}
/// Bounded operational counters for a delivery worker. No event payloads or
/// receiver error text are retained in this snapshot.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TelemetryMetrics {
/// Events accepted by the queue.
pub queued: u64,
/// Events rejected because their id is in the recent dedupe window.
pub duplicates: u64,
/// Events rejected by envelope validation.
pub rejected: u64,
/// Non-blocking sends rejected because the queue was full.
pub buffer_full: u64,
/// Sends rejected because the delivery was closed.
pub closed: u64,
/// Events successfully handed to the receiver.
pub processed: u64,
/// Receiver calls returning a `ReceiverError`.
pub receiver_failures: u64,
/// Events dropped because the receiver returned an error.
pub receiver_drops: u64,
/// Shutdowns that detached a worker after the configured timeout.
pub shutdown_timeouts: u64,
}
struct Metrics {
queued: AtomicU64,
duplicates: AtomicU64,
rejected: AtomicU64,
buffer_full: AtomicU64,
closed: AtomicU64,
processed: AtomicU64,
receiver_failures: AtomicU64,
receiver_drops: AtomicU64,
shutdown_timeouts: AtomicU64,
}
impl Metrics {
fn new() -> Self {
Self {
queued: AtomicU64::new(0),
duplicates: AtomicU64::new(0),
rejected: AtomicU64::new(0),
buffer_full: AtomicU64::new(0),
closed: AtomicU64::new(0),
processed: AtomicU64::new(0),
receiver_failures: AtomicU64::new(0),
receiver_drops: AtomicU64::new(0),
shutdown_timeouts: AtomicU64::new(0),
}
}
fn snapshot(&self) -> TelemetryMetrics {
TelemetryMetrics {
queued: self.queued.load(Ordering::Relaxed),
duplicates: self.duplicates.load(Ordering::Relaxed),
rejected: self.rejected.load(Ordering::Relaxed),
buffer_full: self.buffer_full.load(Ordering::Relaxed),
closed: self.closed.load(Ordering::Relaxed),
processed: self.processed.load(Ordering::Relaxed),
receiver_failures: self.receiver_failures.load(Ordering::Relaxed),
receiver_drops: self.receiver_drops.load(Ordering::Relaxed),
shutdown_timeouts: self.shutdown_timeouts.load(Ordering::Relaxed),
}
}
}
struct Inner {
sender: Mutex<Option<SyncSender<TelemetryEnvelope>>>,
seen: Mutex<SeenIds>,
metrics: Metrics,
}
/// The cheap, clonable producer handle used by request handlers.
#[derive(Clone)]
pub struct TelemetrySender {
inner: Arc<Inner>,
}
/// Bounded queue, dedupe, and shutdown settings for a delivery worker.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TelemetryDeliveryConfig {
pub queue_capacity: usize,
pub dedupe_capacity: usize,
pub shutdown_timeout: Duration,
}
impl Default for TelemetryDeliveryConfig {
fn default() -> Self {
Self {
queue_capacity: 128,
dedupe_capacity: 1024,
shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
}
}
}
/// The asynchronous delivery worker and its bounded queue.
pub struct TelemetryDelivery {
inner: Arc<Inner>,
worker: Option<JoinHandle<()>>,
done: mpsc::Receiver<()>,
shutdown_timeout: Duration,
}
/// Result of a non-blocking enqueue attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmitOutcome {
Queued,
Duplicate,
BufferFull,
Closed,
Rejected,
}
impl TelemetryDelivery {
/// Start a worker with queue and recent-dedupe capacities coupled for compatibility.
pub fn new<R>(receiver: R, capacity: usize) -> Result<Self, TelemetryError>
where
R: TelemetryReceiver,
{
Self::new_with_capacities(receiver, capacity, capacity)
}
/// Start a worker with independently bounded queue and recent-dedupe windows.
pub fn new_with_capacities<R>(
receiver: R,
queue_capacity: usize,
dedupe_capacity: usize,
) -> Result<Self, TelemetryError>
where
R: TelemetryReceiver,
{
Self::new_with_config(
receiver,
TelemetryDeliveryConfig {
queue_capacity,
dedupe_capacity,
..TelemetryDeliveryConfig::default()
},
)
}
/// Start a worker from explicit bounded delivery settings.
pub fn new_with_config<R>(
receiver: R,
config: TelemetryDeliveryConfig,
) -> Result<Self, TelemetryError>
where
R: TelemetryReceiver,
{
if config.queue_capacity == 0 || config.dedupe_capacity == 0 {
return Err(TelemetryError::InvalidCapacity);
}
let (sender, receiver_queue) = mpsc::sync_channel(config.queue_capacity);
let inner = Arc::new(Inner {
sender: Mutex::new(Some(sender)),
seen: Mutex::new(SeenIds::new(config.dedupe_capacity)),
metrics: Metrics::new(),
});
let worker_inner = Arc::clone(&inner);
let (done_sender, done) = mpsc::channel();
let worker = thread::spawn(move || {
while let Ok(event) = receiver_queue.recv() {
match receiver.receive(event) {
Ok(()) => {
worker_inner
.metrics
.processed
.fetch_add(1, Ordering::Relaxed);
}
Err(_error) => {
worker_inner
.metrics
.receiver_failures
.fetch_add(1, Ordering::Relaxed);
worker_inner
.metrics
.receiver_drops
.fetch_add(1, Ordering::Relaxed);
}
}
}
let _ = done_sender.send(());
});
Ok(Self {
inner,
worker: Some(worker),
done,
shutdown_timeout: config.shutdown_timeout,
})
}
/// Get a producer handle. Sending only uses `try_send` and never waits.
pub fn sender(&self) -> TelemetrySender {
TelemetrySender {
inner: Arc::clone(&self.inner),
}
}
/// Read bounded operational counters without exposing event payloads.
pub fn metrics(&self) -> TelemetryMetrics {
self.inner.metrics.snapshot()
}
/// Close the queue and wait up to the configured bound for the worker.
///
/// A timeout detaches the worker so a hung receiver cannot block shutdown.
/// Receivers should implement their own durable idempotency: this in-memory
/// dedupe window is only a recent best-effort producer guard.
pub fn shutdown(self) {
let timeout = self.shutdown_timeout;
self.shutdown_with_timeout(timeout);
}
/// Close the queue and wait up to `timeout` for the worker.
pub fn shutdown_with_timeout(mut self, timeout: Duration) {
self.close();
if let Some(worker) = self.worker.take() {
match self.done.recv_timeout(timeout) {
Ok(()) | Err(RecvTimeoutError::Disconnected) => {
let _ = worker.join();
}
Err(RecvTimeoutError::Timeout) => {
self.inner
.metrics
.shutdown_timeouts
.fetch_add(1, Ordering::Relaxed);
drop(worker);
}
}
}
}
fn close(&self) {
let _ = self.inner.sender.lock().map(|mut sender| sender.take());
}
}
impl Drop for TelemetryDelivery {
fn drop(&mut self) {
self.close();
// Dropping the handle detaches the worker. It is deliberately not
// joined here because shutdown must never block a customer request.
let _ = self.worker.take();
}
}
impl TelemetrySender {
/// Validate and enqueue without waiting. Full/closed targets are reported
/// to the caller but never surfaced as a receiver error.
pub fn emit(&self, event: TelemetryEnvelope) -> EmitOutcome {
if event.validate().is_err() {
self.inner.metrics.rejected.fetch_add(1, Ordering::Relaxed);
return EmitOutcome::Rejected;
}
let event_id = event.event_id.clone();
let mut seen = match self.inner.seen.lock() {
Ok(guard) => guard,
Err(_) => {
self.inner.metrics.closed.fetch_add(1, Ordering::Relaxed);
return EmitOutcome::Closed;
}
};
if !seen.reserve(&event_id) {
self.inner
.metrics
.duplicates
.fetch_add(1, Ordering::Relaxed);
return EmitOutcome::Duplicate;
}
let result = match self.inner.sender.lock() {
Ok(sender) => match sender.as_ref() {
Some(sender) => sender.try_send(event),
None => {
seen.release(&event_id);
self.inner.metrics.closed.fetch_add(1, Ordering::Relaxed);
return EmitOutcome::Closed;
}
},
Err(_) => {
seen.release(&event_id);
self.inner.metrics.closed.fetch_add(1, Ordering::Relaxed);
return EmitOutcome::Closed;
}
};
match result {
Ok(()) => {
self.inner.metrics.queued.fetch_add(1, Ordering::Relaxed);
EmitOutcome::Queued
}
Err(TrySendError::Full(_)) => {
seen.release(&event_id);
self.inner
.metrics
.buffer_full
.fetch_add(1, Ordering::Relaxed);
EmitOutcome::BufferFull
}
Err(TrySendError::Disconnected(_)) => {
seen.release(&event_id);
self.inner.metrics.closed.fetch_add(1, Ordering::Relaxed);
EmitOutcome::Closed
}
}
}
/// Read the same bounded operational counters as the delivery owner.
pub fn metrics(&self) -> TelemetryMetrics {
self.inner.metrics.snapshot()
}
}
/// Stable event names and constructors shared by lifecycle consumers.
pub mod fixtures {
use super::{PrivacyClassification, TelemetryContext, TelemetryEnvelope, TelemetryError};
pub const FIRST_VALUE_EVENT: &str = "lifecycle.first_value";
pub const ACTIVE_USE_EVENT: &str = "lifecycle.active_use";
/// Construct the shared first-value fixture without product-specific fields.
pub fn first_value(
event_id: impl Into<String>,
context: &TelemetryContext,
occurred_at: i64,
) -> Result<TelemetryEnvelope, TelemetryError> {
TelemetryEnvelope::new(
event_id,
context,
FIRST_VALUE_EVENT,
occurred_at,
PrivacyClassification::Operational,
Vec::new(),
)
}
/// Construct the shared active-use fixture without product-specific fields.
pub fn active_use(
event_id: impl Into<String>,
context: &TelemetryContext,
occurred_at: i64,
) -> Result<TelemetryEnvelope, TelemetryError> {
TelemetryEnvelope::new(
event_id,
context,
ACTIVE_USE_EVENT,
occurred_at,
PrivacyClassification::Operational,
Vec::new(),
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Instant;
fn context() -> TelemetryContext {
TelemetryContext::new("org", "workspace", "installation", "app")
}
fn event(id: &str) -> TelemetryEnvelope {
TelemetryEnvelope::new(
id,
&context(),
"lifecycle.first_value",
10,
PrivacyClassification::Operational,
Vec::new(),
)
.unwrap()
}
#[test]
fn envelope_carries_shared_linkage_and_schema() {
let envelope = event("event-1");
assert_eq!(envelope.schema_version, SCHEMA_VERSION);
assert_eq!(envelope.organization_id, "org");
assert_eq!(envelope.workspace_id, "workspace");
assert_eq!(envelope.installation_id, "installation");
assert_eq!(envelope.app_id, "app");
}
#[test]
fn lifecycle_fixtures_are_product_neutral() {
let first = fixtures::first_value("first-1", &context(), 1).unwrap();
let active = fixtures::active_use("active-1", &context(), 2).unwrap();
assert_eq!(first.event_name, fixtures::FIRST_VALUE_EVENT);
assert_eq!(active.event_name, fixtures::ACTIVE_USE_EVENT);
assert!(first.attributes.is_empty());
assert!(active.attributes.is_empty());
}
#[test]
fn forbidden_privacy_classes_are_rejected() {
for privacy in [
PrivacyClassification::UnapprovedPii,
PrivacyClassification::Content,
PrivacyClassification::Secret,
] {
let result = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
privacy,
Vec::new(),
);
assert!(matches!(
result,
Err(TelemetryError::PrivacyRejected { .. })
));
}
}
#[test]
fn forbidden_attribute_keys_are_rejected_even_when_operational() {
for key in ["password", "content", "request_body", "user_id"] {
let result = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::Operational,
vec![(key.to_owned(), "value".to_owned())],
);
assert!(matches!(
result,
Err(TelemetryError::PrivacyRejected { .. })
));
}
}
#[test]
fn attribute_values_cannot_carry_secret_or_content() {
for value in ["Bearer sk-abc123", "freeform message", "api_secret_key"] {
let result = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::Operational,
vec![("dimension".into(), value.into())],
);
assert!(result.is_err(), "value unexpectedly accepted: {value}");
}
}
#[test]
fn unicode_and_middle_marker_attribute_keys_are_rejected() {
for key in ["раssword", "ѕecret", "tоkеn", "api_secret_key"] {
let result = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::Operational,
vec![(key.into(), "value".into())],
);
assert!(result.is_err(), "key unexpectedly accepted: {key}");
}
}
#[test]
fn ordinary_dimensions_are_not_rejected_by_substring_matches() {
let result = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::Operational,
vec![("ship_mode".into(), "air".into())],
);
assert!(result.is_ok());
}
#[test]
fn approved_pii_is_explicit_but_secrets_remain_forbidden() {
let approved = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::ApprovedPii,
vec![("user_id".into(), "approved-reference".into())],
);
assert!(approved.is_ok());
let secret = TelemetryEnvelope::new(
"event",
&context(),
"lifecycle.active_use",
1,
PrivacyClassification::ApprovedPii,
vec![("api_token".into(), "never".into())],
);
assert!(matches!(
secret,
Err(TelemetryError::PrivacyRejected { .. })
));
}
#[test]
fn invalid_schema_and_timestamp_are_rejected() {
let mut envelope = event("event");
envelope.schema_version = SCHEMA_VERSION + 1;
assert!(matches!(
envelope.validate(),
Err(TelemetryError::UnsupportedSchemaVersion { .. })
));
envelope.schema_version = SCHEMA_VERSION;
envelope.occurred_at = -1;
assert!(matches!(
envelope.validate(),
Err(TelemetryError::InvalidField {
field: "occurred_at",
..
})
));
let future = event("future");
let mut too_future = future;
too_future.occurred_at = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs() as i64
+ MAX_FUTURE_SECONDS
+ 1;
assert!(matches!(
too_future.validate(),
Err(TelemetryError::InvalidField {
field: "occurred_at",
..
})
));
}
#[test]
fn repeated_release_does_not_retain_stale_order_entries() {
let mut seen = SeenIds::new(2);
for index in 0..100 {
let event_id = format!("event-{index}");
assert!(seen.reserve(&event_id));
seen.release(&event_id);
}
assert!(seen.ids.is_empty());
assert!(seen.order.is_empty());
}
struct CountingReceiver {
calls: Arc<AtomicUsize>,
}
impl TelemetryReceiver for CountingReceiver {
fn receive(&self, _event: TelemetryEnvelope) -> Result<(), ReceiverError> {
self.calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
#[test]
fn delivery_is_non_blocking_and_deduplicates_ids() {
let calls = Arc::new(AtomicUsize::new(0));
let delivery = TelemetryDelivery::new(
CountingReceiver {
calls: Arc::clone(&calls),
},
2,
)
.unwrap();
let sender = delivery.sender();
assert_eq!(sender.emit(event("same")), EmitOutcome::Queued);
assert_eq!(sender.emit(event("same")), EmitOutcome::Duplicate);
delivery.shutdown();
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
struct BlockingReceiver {
started: Arc<AtomicUsize>,
release: Arc<AtomicUsize>,
}
impl TelemetryReceiver for BlockingReceiver {
fn receive(&self, _event: TelemetryEnvelope) -> Result<(), ReceiverError> {
self.started.fetch_add(1, Ordering::SeqCst);
while self.release.load(Ordering::SeqCst) == 0 {
thread::yield_now();
}
Ok(())
}
}
#[test]
fn delivery_reports_full_without_waiting() {
let started = Arc::new(AtomicUsize::new(0));
let release = Arc::new(AtomicUsize::new(0));
let delivery = TelemetryDelivery::new(
BlockingReceiver {
started: Arc::clone(&started),
release: Arc::clone(&release),
},
1,
)
.unwrap();
let sender = delivery.sender();
let _ = sender.emit(event("first"));
while started.load(Ordering::SeqCst) == 0 {
thread::yield_now();
}
let _ = sender.emit(event("second"));
let result = sender.emit(event("third"));
assert_eq!(result, EmitOutcome::BufferFull);
release.store(1, Ordering::SeqCst);
delivery.shutdown();
}
struct FailingReceiver;
impl TelemetryReceiver for FailingReceiver {
fn receive(&self, _event: TelemetryEnvelope) -> Result<(), ReceiverError> {
Err(ReceiverError::new("target unavailable"))
}
}
#[test]
fn receiver_failure_does_not_fail_enqueue_and_is_observable() {
let delivery = TelemetryDelivery::new(FailingReceiver, 1).unwrap();
let sender = delivery.sender();
assert_eq!(sender.emit(event("outage-safe")), EmitOutcome::Queued);
while sender.metrics().receiver_failures == 0 {
thread::yield_now();
}
let metrics = sender.metrics();
assert_eq!(metrics.receiver_failures, 1);
assert_eq!(metrics.receiver_drops, 1);
assert_eq!(
ReceiverError::new("target unavailable").message(),
"target unavailable"
);
delivery.shutdown();
}
struct PanickingReceiver {
started: Arc<AtomicUsize>,
}
impl TelemetryReceiver for PanickingReceiver {
fn receive(&self, _event: TelemetryEnvelope) -> Result<(), ReceiverError> {
self.started.fetch_add(1, Ordering::SeqCst);
panic!("worker exited before signalling completion");
}
}
#[test]
fn disconnected_worker_shutdown_is_not_counted_as_timeout() {
let started = Arc::new(AtomicUsize::new(0));
let delivery = TelemetryDelivery::new(
PanickingReceiver {
started: Arc::clone(&started),
},
1,
)
.unwrap();
let sender = delivery.sender();
assert_eq!(sender.emit(event("panic")), EmitOutcome::Queued);
while started.load(Ordering::SeqCst) == 0 {
thread::yield_now();
}
delivery.shutdown_with_timeout(Duration::from_secs(1));
assert_eq!(sender.metrics().shutdown_timeouts, 0);
}
#[test]
fn queue_and_dedupe_capacities_are_independent() {
let delivery = TelemetryDelivery::new_with_capacities(
CountingReceiver {
calls: Arc::new(AtomicUsize::new(0)),
},
1,
3,
)
.unwrap();
assert_eq!(delivery.sender().metrics(), TelemetryMetrics::default());
delivery.shutdown();
}
#[test]
fn shutdown_is_bounded_when_receiver_hangs() {
let started = Arc::new(AtomicUsize::new(0));
let release = Arc::new(AtomicUsize::new(0));
let delivery = TelemetryDelivery::new(
BlockingReceiver {
started: Arc::clone(&started),
release: Arc::clone(&release),
},
1,
)
.unwrap();
let sender = delivery.sender();
assert_eq!(sender.emit(event("hang")), EmitOutcome::Queued);
while started.load(Ordering::SeqCst) == 0 {
thread::yield_now();
}
let started_at = Instant::now();
delivery.shutdown_with_timeout(Duration::from_millis(20));
assert_eq!(sender.metrics().shutdown_timeouts, 1);
assert!(started_at.elapsed() < Duration::from_millis(500));
release.store(1, Ordering::SeqCst);
}
#[test]
fn zero_capacity_is_rejected() {
let result = TelemetryDelivery::new(
CountingReceiver {
calls: Arc::new(AtomicUsize::new(0)),
},
0,
);
assert!(matches!(result, Err(TelemetryError::InvalidCapacity)));
}
}