feat(clock-drift): Apply clock drift normalization only if sent_at is defined

CHANGELOG.md

@@ -5,8 +5,10 @@
 **Bug fixes:**
 
 - Fix performance regression in disk spooling by using page counts to estimate the spool size. ([#3379](https://github.com/getsentry/relay/pull/3379))
+- Perform clock drift normalization only when `sent_at` is set in the `Envelope` headers. ([#3405](https://github.com/getsentry/relay/pull/3405))
 
 **Features**:
+
 - **Breaking change:** Kafka topic configuration keys now support the default topic name. The previous aliases `metrics` and `metrics_transactions` are no longer supported if configuring topics manually. Use `ingest-metrics` or `metrics_sessions` instead of `metrics`, and `ingest-performance-metrics` or `metrics_generic` instead of `metrics_transactions`. ([#3361](https://github.com/getsentry/relay/pull/3361))
 - Add support for continuous profiling. ([#3270](https://github.com/getsentry/relay/pull/3270))
 - Add support for Reporting API for CSP reports ([#3277](https://github.com/getsentry/relay/pull/3277))

relay-server/src/services/processor/event.rs

@@ -12,7 +12,7 @@ use relay_event_normalization::{nel, ClockDriftProcessor};
 use relay_event_schema::processor::{self, ProcessingState};
 use relay_event_schema::protocol::{
     Breadcrumb, Csp, Event, ExpectCt, ExpectStaple, Hpkp, LenientString, NetworkReportError,
-    OtelContext, RelayInfo, SecurityReportType, Timestamp, Values,
+    OtelContext, RelayInfo, SecurityReportType, Values,
 };
 use relay_pii::PiiProcessor;
 use relay_protocol::{Annotated, Array, Empty, FromValue, Object, Value};
@@ -139,7 +139,6 @@ pub fn finalize<G: EventProcessing>(
     state: &mut ProcessEnvelopeState<G>,
     config: &Config,
 ) -> Result<(), ProcessingError> {
-    let is_transaction = state.event_type() == Some(EventType::Transaction);
     let envelope = state.managed_envelope.envelope_mut();
 
     let event = match state.event.value_mut() {
@@ -236,17 +235,9 @@ pub fn finalize<G: EventProcessing>(
         }
     }
 
-    // TODO: Temporary workaround before processing. Experimental SDKs relied on a buggy
-    // clock drift correction that assumes the event timestamp is the sent_at time. This
-    // should be removed as soon as legacy ingestion has been removed.
-    let sent_at = match envelope.sent_at() {
-        Some(sent_at) => Some(sent_at),
-        None if is_transaction => event.timestamp.value().copied().map(Timestamp::into_inner),
-        None => None,
-    };
-
-    let mut processor = ClockDriftProcessor::new(sent_at, state.managed_envelope.received_at())
-        .at_least(MINIMUM_CLOCK_DRIFT);
+    let mut processor =
+        ClockDriftProcessor::new(envelope.sent_at(), state.managed_envelope.received_at())
+            .at_least(MINIMUM_CLOCK_DRIFT);
     processor::process_value(&mut state.event, &mut processor, ProcessingState::root())
         .map_err(|_| ProcessingError::InvalidTransaction)?;
 

tests/integration/test_clock_drift.py

@@ -0,0 +1,97 @@
+from datetime import timedelta, datetime, timezone
+
+from sentry_sdk.envelope import Envelope
+
+
+def mock_transaction(timestamp):
+    return {
+        "type": "transaction",
+        "timestamp": format_date(timestamp),
+        "start_timestamp": format_date(timestamp - timedelta(seconds=2)),
+        "spans": [],
+        "contexts": {
+            "trace": {
+                "op": "http",
+                "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                "span_id": "968cff94913ebb07",
+            }
+        },
+        "transaction": "/hello",
+    }
+
+
+def format_date(date):
+    return date.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def test_clock_drift_applied_when_timestamp_is_too_old(mini_sentry, relay):
+    relay = relay(mini_sentry)
+    project_id = 42
+    mini_sentry.add_basic_project_config(project_id)
+
+    now = datetime.now(tz=timezone.utc)
+    one_month_ago = now - timedelta(days=30)
+
+    envelope = Envelope(headers={"sent_at": format_date(one_month_ago)})
+    envelope.add_transaction(mock_transaction(one_month_ago))
+
+    relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
+
+    transaction_event = mini_sentry.captured_events.get(
+        timeout=1
+    ).get_transaction_event()
+
+    error_name, error_metadata = transaction_event["_meta"]["timestamp"][""]["err"][0]
+    assert error_name == "clock_drift"
+    assert (
+        one_month_ago.timestamp()
+        < transaction_event["timestamp"]
+        < datetime.now(tz=timezone.utc).timestamp()
+    )
+
+
+def test_clock_drift_not_applied_when_timestamp_is_recent(mini_sentry, relay):
+    relay = relay(mini_sentry)
+    project_id = 42
+    mini_sentry.add_basic_project_config(project_id)
+
+    now = datetime.now(tz=timezone.utc)
+    five_minutes_ago = now - timedelta(minutes=5)
+
+    envelope = Envelope(headers={"sent_at": format_date(five_minutes_ago)})
+    envelope.add_transaction(mock_transaction(five_minutes_ago))
+
+    relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
+
+    transaction_event = mini_sentry.captured_events.get(
+        timeout=1
+    ).get_transaction_event()
+    assert "_meta" not in transaction_event
+    assert transaction_event["timestamp"] == five_minutes_ago.timestamp()
+
+
+def test_clock_drift_not_applied_when_sent_at_is_not_supplied(mini_sentry, relay):
+    relay = relay(mini_sentry)
+    project_id = 42
+    mini_sentry.add_basic_project_config(project_id)
+
+    now = datetime.now(tz=timezone.utc)
+    one_month_ago = now - timedelta(days=30)
+
+    envelope = Envelope()
+    envelope.add_transaction(mock_transaction(one_month_ago))
+
+    relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
+
+    transaction_event = mini_sentry.captured_events.get(
+        timeout=1
+    ).get_transaction_event()
+    error_name, error_metadata = transaction_event["_meta"]["timestamp"][""]["err"][0]
+    # In case clock drift is not run, we expect timestamps normalization to go into effect to mark timestamps as
+    # past or future if they surpass a certain threshold.
+    assert error_name == "past_timestamp"
+    assert (
+        one_month_ago.timestamp()
+        < transaction_event["timestamp"]
+        < datetime.now(tz=timezone.utc).timestamp()
+    )