diff --git a/rust_snuba/src/processors/eap_items.rs b/rust_snuba/src/processors/eap_items.rs index 684af637ca..953487b5b1 100644 --- a/rust_snuba/src/processors/eap_items.rs +++ b/rust_snuba/src/processors/eap_items.rs @@ -15,7 +15,10 @@ use sentry_protos::snuba::v1::any_value::Value; use sentry_protos::snuba::v1::{ArrayValue, TraceItem, TraceItemType}; use crate::config::ProcessorConfig; -use crate::processors::utils::{enforce_retentions, SilencedDLQMessage}; +use crate::processors::utils::{ + enforce_retentions, get_drop_invalid_timestamps_enabled, out_of_valid_interval_secs, + SilencedDLQMessage, +}; use crate::strategies::clickhouse::rowbinary; use crate::types::CogsData; use crate::types::{item_type_name, InsertBatch, ItemTypeMetrics, KafkaMessageMetadata}; @@ -63,11 +66,13 @@ struct ProcessedItem { cogs_data: CogsData, } +/// `Ok(None)` means the item was dropped for being out of range; the message +/// is committed without inserting anything. fn process_eap_item( msg: KafkaPayload, metadata: &KafkaMessageMetadata, config: &ProcessorConfig, -) -> anyhow::Result { +) -> anyhow::Result> { let payload: &[u8] = msg.payload().context("Expected payload")?; let trace_item = TraceItem::decode(payload)?; let origin_timestamp = trace_item @@ -82,8 +87,25 @@ fn process_eap_item( let item_type = TraceItemType::try_from(trace_item.item_type).unwrap_or(TraceItemType::Unspecified); + let (retention_days, downsampled_retention_days) = enforce_retentions( + Some(trace_item.retention_days as u16), + Some(trace_item.downsampled_retention_days as u16), + ); + + let now = Utc::now(); + if let Some(event_ts) = event_timestamp { - let now = Utc::now(); + if get_drop_invalid_timestamps_enabled() + && out_of_valid_interval_secs(event_ts, now, retention_days) + { + counter!( + "eap_items.messages.dropped_out_of_range_timestamp", + 1, + "reason" => if event_ts > now { "future_ts" } else { "past_ts" }, + "item_type" => item_type_name(item_type) + ); + return Ok(None); + } if let Some(grace_min) = get_dlq_grace_period_min(&config.storage_name) { if let Some(reason) = should_dlq(event_ts, now, grace_min) { @@ -98,11 +120,6 @@ fn process_eap_item( } } - let (retention_days, downsampled_retention_days) = enforce_retentions( - Some(trace_item.retention_days as u16), - Some(trace_item.downsampled_retention_days as u16), - ); - let mut eap_item = EAPItem::try_from(trace_item)?; eap_item.retention_days = Some(retention_days); @@ -118,7 +135,7 @@ fn process_eap_item( } eap_item.attributes.insert_int( "sentry._internal.ingested_at".into(), - Utc::now().timestamp_millis(), + now.timestamp_millis(), ); // we are using this to compare to outcomes which stores the timestamp as seconds if let Some(received_at) = origin_timestamp { @@ -156,12 +173,12 @@ fn process_eap_item( data: BTreeMap::from([(app_feature, payload.len() as u64)]), }; - Ok(ProcessedItem { + Ok(Some(ProcessedItem { eap_item, origin_timestamp, item_type_metrics, cogs_data, - }) + })) } fn get_dlq_grace_period_min(storage_name: &str) -> Option { @@ -270,7 +287,9 @@ pub fn process_message( metadata: KafkaMessageMetadata, config: &ProcessorConfig, ) -> anyhow::Result { - let processed = process_eap_item(msg, &metadata, config)?; + let Some(processed) = process_eap_item(msg, &metadata, config)? else { + return Ok(InsertBatch::skip()); + }; let mut batch = InsertBatch::from_rows([processed.eap_item], processed.origin_timestamp)?; batch.item_type_metrics = Some(processed.item_type_metrics); batch.cogs_data = Some(processed.cogs_data); @@ -282,7 +301,9 @@ pub fn process_message_row_binary( metadata: KafkaMessageMetadata, config: &ProcessorConfig, ) -> anyhow::Result { - let processed = process_eap_item(msg, &metadata, config)?; + let Some(processed) = process_eap_item(msg, &metadata, config)? else { + return Ok(InsertBatch::skip()); + }; let row = EAPItemRow::try_from(processed.eap_item)?; // Encode the row to RowBinary bytes inline so the wide typed struct (~80 @@ -315,7 +336,13 @@ pub(crate) fn process_message_row_binary_typed( metadata: KafkaMessageMetadata, config: &ProcessorConfig, ) -> anyhow::Result { - let processed = process_eap_item(msg, &metadata, config)?; + let Some(processed) = process_eap_item(msg, &metadata, config)? else { + return Ok(EAPItemRowBatch { + rows: vec![], + cogs_data: None, + item_type_metrics: None, + }); + }; Ok(EAPItemRowBatch { rows: vec![EAPItemRow::try_from(processed.eap_item)?], cogs_data: Some(processed.cogs_data), @@ -800,6 +827,7 @@ mod tests { use std::sync::Once; use super::*; + use crate::processors::utils::DROP_INVALID_TIMESTAMPS_KEY; static INIT: Once = Once::new(); fn init_options() { @@ -1380,6 +1408,107 @@ mod tests { assert_eq!(get_dlq_grace_period_min(""), None); } + fn encode_with_timestamp(event_ts: i64, retention_days: u32, downsampled: u32) -> Vec { + let mut trace_item = generate_trace_item(Uuid::new_v4()); + trace_item.timestamp = Some(Timestamp { + seconds: event_ts, + nanos: 0, + }); + trace_item.retention_days = retention_days; + trace_item.downsampled_retention_days = downsampled; + let mut payload = Vec::new(); + trace_item.encode(&mut payload).unwrap(); + payload + } + + fn rows_inserted(payload: &[u8]) -> (usize, usize, usize) { + let metadata = KafkaMessageMetadata { + partition: 0, + offset: 1, + timestamp: Utc::now(), + }; + let config = ProcessorConfig::default(); + let json = process_message( + KafkaPayload::new(None, None, Some(payload.to_vec())), + metadata.clone(), + &config, + ) + .unwrap(); + let row_binary = process_message_row_binary( + KafkaPayload::new(None, None, Some(payload.to_vec())), + metadata.clone(), + &config, + ) + .unwrap(); + let typed = process_message_row_binary_typed( + KafkaPayload::new(None, None, Some(payload.to_vec())), + metadata, + &config, + ) + .unwrap(); + ( + json.rows.num_rows, + row_binary.rows.num_rows, + typed.rows.len(), + ) + } + + #[test] + fn test_drop_invalid_timestamps_skips_items_past_retention() { + init_options(); + let now = Utc::now().timestamp(); + let past_90_days = encode_with_timestamp(now - 120 * 86_400, 90, 90); + let within_90_days = encode_with_timestamp(now - 60 * 86_400, 90, 90); + // NOTE: 60 days old on 30 day standard retention: dropped even though the + // downsample tiers would keep it for 396 days. + let past_standard_retention = encode_with_timestamp(now - 60 * 86_400, 30, 396); + let far_future = encode_with_timestamp(now + 8 * 86_400, 90, 90); + + // Option off: everything is inserted. + assert_eq!(rows_inserted(&past_90_days), (1, 1, 1)); + assert_eq!(rows_inserted(&far_future), (1, 1, 1)); + + let _guard = + override_options(&[("snuba", DROP_INVALID_TIMESTAMPS_KEY, json!(true))]).unwrap(); + assert_eq!(rows_inserted(&past_90_days), (0, 0, 0)); + assert_eq!(rows_inserted(&far_future), (0, 0, 0)); + assert_eq!(rows_inserted(&within_90_days), (1, 1, 1)); + assert_eq!(rows_inserted(&past_standard_retention), (0, 0, 0)); + } + + #[test] + fn test_drop_invalid_timestamps_runs_before_dlq() { + // An item past retention is dropped (committed, no rows), not DLQ'd, + // even when the prior-partition DLQ would otherwise catch it. + init_options(); + let _guard = override_options(&[ + ("snuba", DROP_INVALID_TIMESTAMPS_KEY, json!(true)), + ( + "snuba", + "eap_items_dlq_grace_period_min", + json!({ "eap_items_dlq_test": 0 }), + ), + ]) + .unwrap(); + let config = ProcessorConfig { + storage_name: "eap_items_dlq_test".to_string(), + ..Default::default() + }; + let metadata = KafkaMessageMetadata { + partition: 0, + offset: 1, + timestamp: Utc::now(), + }; + let payload = encode_with_timestamp(Utc::now().timestamp() - 120 * 86_400, 90, 90); + let batch = process_message( + KafkaPayload::new(None, None, Some(payload)), + metadata, + &config, + ) + .unwrap(); + assert_eq!(batch.rows.num_rows, 0); + } + #[test] fn test_get_dlq_grace_period_min_reads_dict_option() { init_options(); diff --git a/rust_snuba/src/processors/utils.rs b/rust_snuba/src/processors/utils.rs index 394255ad0e..e351290c37 100644 --- a/rust_snuba/src/processors/utils.rs +++ b/rust_snuba/src/processors/utils.rs @@ -3,10 +3,42 @@ use schemars::JsonSchema; use sentry_options::options; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +/// One week in seconds. The eap_items table is partitioned by +/// `(retention_days, toMonday(timestamp))`, so any event more than a week +/// in the future will add more parts. +pub const INVALID_TIMESTAMP_FUTURE_INTERVAL_SECONDS: i64 = 7 * 24 * 60 * 60; + +/// sentry-options key. When `true`, the eap-items consumer skips messages +/// whose event `timestamp` is more than one week in the future or older than +/// the item's retention (see `out_of_valid_interval_secs`). +pub const DROP_INVALID_TIMESTAMPS_KEY: &str = "eap_items_drop_invalid_timestamps"; + // Equivalent to "%Y-%m-%dT%H:%M:%S.%fZ" in python // Notice the differennce of .%fZ vs %.fZ, this comes from a difference in how rust's chrono handles the format const PAYLOAD_DATETIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.fZ"; +/// Returns true if `ts` is more than one week after `now` or more than +/// `retention_days` before `now`. Items past retention would be TTL-deleted on +/// arrival, so there is nothing to keep. +pub fn out_of_valid_interval_secs( + ts: DateTime, + now: DateTime, + retention_days: u16, +) -> bool { + if ts < now - chrono::Duration::days(retention_days as i64) { + return true; + } + ts > now + chrono::Duration::seconds(INVALID_TIMESTAMP_FUTURE_INTERVAL_SECONDS) +} + +pub fn get_drop_invalid_timestamps_enabled() -> bool { + options("snuba") + .ok() + .and_then(|o| o.get(DROP_INVALID_TIMESTAMPS_KEY).ok()) + .and_then(|v| v.as_bool()) + .unwrap_or(false) +} + #[derive(Clone, Copy)] pub enum RetentionKind { Standard, @@ -262,4 +294,66 @@ mod tests { assert_eq!(enforce_downsampled_retention(None), 396); assert_eq!(enforce_downsampled_retention(Some(420)), 396); } + + const NOW_SECS: i64 = 2_000_000_000; + const DAY_SECS: i64 = 24 * 60 * 60; + + #[test] + fn test_out_of_valid_interval_secs_drops_messages_older_than_retention() { + let now = DateTime::from_timestamp(NOW_SECS, 0).unwrap(); + let ts = DateTime::from_timestamp(NOW_SECS - 30 * DAY_SECS - 1, 0).unwrap(); + assert!(out_of_valid_interval_secs(ts, now, 30)); + } + + #[test] + fn test_out_of_valid_interval_secs_drops_messages_more_than_a_week_in_the_future() { + let now = DateTime::from_timestamp(NOW_SECS, 0).unwrap(); + let ts = + DateTime::from_timestamp(NOW_SECS + INVALID_TIMESTAMP_FUTURE_INTERVAL_SECONDS + 1, 0) + .unwrap(); + assert!(out_of_valid_interval_secs(ts, now, 30)); + } + + #[test] + fn test_out_of_valid_interval_secs_keeps_messages_at_exactly_retention_old() { + let now = DateTime::from_timestamp(NOW_SECS, 0).unwrap(); + let ts = DateTime::from_timestamp(NOW_SECS - 30 * DAY_SECS, 0).unwrap(); + assert!(!out_of_valid_interval_secs(ts, now, 30)); + } + + #[test] + fn test_out_of_valid_interval_secs_keeps_messages_within_a_week() { + let now = DateTime::from_timestamp(NOW_SECS, 0).unwrap(); + let past = DateTime::from_timestamp(NOW_SECS - DAY_SECS, 0).unwrap(); + assert!(!out_of_valid_interval_secs(past, now, 30)); + let future = DateTime::from_timestamp(NOW_SECS + DAY_SECS, 0).unwrap(); + assert!(!out_of_valid_interval_secs(future, now, 30)); + } + + #[test] + fn test_out_of_valid_interval_secs_uses_item_retention() { + // INC-2585: an item dated 2026-04-27 arriving on 2026-10-19 is past + // 90 day retention but still inside 396 day retention. + let now = DateTime::parse_from_rfc3339("2026-10-19T04:08:00Z") + .unwrap() + .to_utc(); + let ts = DateTime::parse_from_rfc3339("2026-04-27T12:00:00Z") + .unwrap() + .to_utc(); + assert!(out_of_valid_interval_secs(ts, now, 90)); + assert!(!out_of_valid_interval_secs(ts, now, 396)); + // 31 days old: kept on 90 day retention, dropped on 30. + let ts = DateTime::from_timestamp(now.timestamp() - 31 * DAY_SECS, 0).unwrap(); + assert!(!out_of_valid_interval_secs(ts, now, 90)); + assert!(out_of_valid_interval_secs(ts, now, 30)); + } + + #[test] + fn test_drop_invalid_timestamps_option() { + init_options(); + assert!(!get_drop_invalid_timestamps_enabled()); + let _guard = + override_options(&[("snuba", DROP_INVALID_TIMESTAMPS_KEY, json!(true))]).unwrap(); + assert!(get_drop_invalid_timestamps_enabled()); + } } diff --git a/rust_snuba/src/strategies/accepted_outcomes/aggregator.rs b/rust_snuba/src/strategies/accepted_outcomes/aggregator.rs index 62f254c292..c1b983a485 100644 --- a/rust_snuba/src/strategies/accepted_outcomes/aggregator.rs +++ b/rust_snuba/src/strategies/accepted_outcomes/aggregator.rs @@ -1,6 +1,7 @@ use std::collections::{BTreeMap, HashMap}; use std::time::{Duration, Instant}; +use chrono::{DateTime, Utc}; use prost::Message as ProstMessage; use sentry_arroyo::backends::kafka::types::KafkaPayload; use sentry_arroyo::counter; @@ -12,6 +13,9 @@ use sentry_arroyo::types::{InnerMessage, Message, Partition}; use sentry_arroyo::utils::timing::Deadline; use sentry_protos::snuba::v1::{TraceItem, TraceItemType}; +use crate::processors::utils::{ + enforce_standard_retention, get_drop_invalid_timestamps_enabled, out_of_valid_interval_secs, +}; use crate::types::{item_type_name, AggregatedOutcomesBatch, BucketKey, ItemDedupKey}; #[derive(Debug, Default)] @@ -248,6 +252,32 @@ impl> ProcessingStrategy if event_ts > now { "future_ts" } else { "past_ts" }, + "item_type" => item_type_name(item_type) + ); + return Ok(()); + } + } + let ts_secs = trace_item .received .as_ref() @@ -767,6 +797,95 @@ mod tests { ); } + fn make_payload_with_timestamp( + event_ts: i64, + retention_days: u32, + quantity: u64, + ) -> KafkaPayload { + let trace_item = TraceItem { + organization_id: 1, + project_id: 2, + item_id: uuid::Uuid::new_v4().as_bytes().to_vec(), + item_type: TraceItemType::Span.into(), + retention_days, + downsampled_retention_days: retention_days, + timestamp: Some(Timestamp { + seconds: event_ts, + nanos: 0, + }), + received: Some(Timestamp { + seconds: 1_700_000_000, + nanos: 0, + }), + outcomes: Some(Outcomes { + key_id: 3, + category_count: vec![CategoryCount { + data_category: 4, + quantity, + }], + }), + ..Default::default() + }; + let mut buf = Vec::new(); + trace_item.encode(&mut buf).unwrap(); + KafkaPayload::new(None, None, Some(buf)) + } + + #[test] + fn submit_drops_outcomes_for_items_past_retention() { + static INIT: std::sync::Once = std::sync::Once::new(); + INIT.call_once(|| crate::init_sentry_options().unwrap()); + let _guard = sentry_options::testing::override_options(&[( + "snuba", + crate::processors::utils::DROP_INVALID_TIMESTAMPS_KEY, + serde_json::json!(true), + )]) + .unwrap(); + + let mut aggregator = OutcomesAggregator::new( + Noop { last_message: None }, + 500, + Duration::from_millis(5_000), + 60, + ); + let partition = Partition::new(Topic::new("snuba-items"), 0); + let now = Utc::now().timestamp(); + + // Past 90 day retention: not billed. + aggregator + .submit(Message::new_broker_message( + make_payload_with_timestamp(now - 120 * 86_400, 90, 7), + partition, + 0, + Utc::now(), + )) + .unwrap(); + // More than a week in the future: not billed. + aggregator + .submit(Message::new_broker_message( + make_payload_with_timestamp(now + 8 * 86_400, 90, 11), + partition, + 1, + Utc::now(), + )) + .unwrap(); + // Within retention: billed. + aggregator + .submit(Message::new_broker_message( + make_payload_with_timestamp(now - 60 * 86_400, 90, 5), + partition, + 2, + Utc::now(), + )) + .unwrap(); + + assert_eq!(aggregator.batch.num_buckets(), 1); + let total: u64 = aggregator.batch.buckets.values().map(|s| s.quantity).sum(); + assert_eq!(total, 5); + // Dropped messages still advance the committed offset. + assert_eq!(aggregator.latest_offsets.get(&partition), Some(&2)); + } + #[test] fn is_duplicate_per_item_type() { let mut aggregator = OutcomesAggregator::new( diff --git a/sentry-options/schemas/snuba/schema.json b/sentry-options/schemas/snuba/schema.json index 9f00bb38f3..815e5d4633 100644 --- a/sentry-options/schemas/snuba/schema.json +++ b/sentry-options/schemas/snuba/schema.json @@ -30,6 +30,11 @@ "default": 300, "description": "Timeout in seconds for --health-check partition-stall. A partition with in-flight work and no commit progress for this long is unhealthy (the health file is not touched). 0 disables stall detection (file is touched on every poll). Ignored for other --health-check values." }, + "eap_items_drop_invalid_timestamps": { + "type": "boolean", + "default": false, + "description": "When true, the eap-items consumer and the accepted outcomes consumer skip messages whose event timestamp is more than one week in the future or older than the item's retention_days (at most 90)." + }, "clickhouse_load_balancing": { "type": "object", "additionalProperties": {