Add server-side auction telemetry (edge emission for Tinybird/Grafana)

crates/trusted-server-adapter-fastly/src/auction_sink.rs

@@ -0,0 +1,59 @@
+//! Fastly implementation of the auction telemetry sink.
+//!
+//! Writes one NDJSON line per telemetry row to the configured Fastly
+//! real-time log endpoint, stamping a shared `event_ts` per batch. The write is
+//! buffered by the host and flushed asynchronously, so it never blocks the
+//! response.
+
+use std::io::Write as _;
+
+use chrono::{SecondsFormat, Utc};
+use fastly::log::Endpoint;
+use trusted_server_core::auction::telemetry::{
+    to_json_line_with_event_ts, AuctionEventRow, AuctionEventSink,
+};
+use trusted_server_core::auction_config_types::DEFAULT_AUCTION_TELEMETRY_LOG_ENDPOINT;
+
+/// Sink that serializes telemetry rows to NDJSON and writes them to the Fastly
+/// auction-events log endpoint.
+pub struct FastlyAuctionEventSink {
+    endpoint_name: String,
+}
+
+impl FastlyAuctionEventSink {
+    /// Create a sink that writes to `endpoint_name`.
+    #[must_use]
+    pub fn new(endpoint_name: impl Into<String>) -> Self {
+        let endpoint_name = endpoint_name.into();
+        let endpoint_name = endpoint_name.trim();
+        let endpoint_name = if endpoint_name.is_empty() {
+            DEFAULT_AUCTION_TELEMETRY_LOG_ENDPOINT.to_string()
+        } else {
+            endpoint_name.to_string()
+        };
+        Self { endpoint_name }
+    }
+}
+
+impl AuctionEventSink for FastlyAuctionEventSink {
+    fn emit(&self, rows: &[AuctionEventRow]) {
+        if rows.is_empty() {
+            return;
+        }
+        let event_ts = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
+        let mut endpoint = Endpoint::from_name(&self.endpoint_name);
+        for row in rows {
+            match to_json_line_with_event_ts(row, &event_ts) {
+                Ok(line) => {
+                    if let Err(error) = writeln!(endpoint, "{line}") {
+                        log::warn!("auction telemetry log write failed: {error}");
+                        break;
+                    }
+                }
+                Err(error) => {
+                    log::warn!("auction telemetry serialization failed: {error}");
+                }
+            }
+        }
+    }
+}

crates/trusted-server-adapter-fastly/src/main.rs

@@ -50,6 +50,7 @@ use trusted_server_core::request_signing::{
 use trusted_server_core::settings::Settings;
 use trusted_server_core::settings_data::get_settings;
 
+mod auction_sink;
 mod error;
 mod logging;
 mod management_api;
@@ -186,7 +187,8 @@ fn main() {
     // any request-derived context or converting to the core HTTP types.
     compat::sanitize_fastly_forwarded_headers(&mut req);
 
-    let runtime_services = build_runtime_services(&req, kv_store);
+    let runtime_services =
+        build_runtime_services(&req, kv_store, settings.auction.telemetry_log_endpoint());
     let http_req = compat::from_fastly_request(req);
 
     let route_result = futures::executor::block_on(route_request(

crates/trusted-server-adapter-fastly/src/platform.rs

@@ -555,10 +555,14 @@ impl PlatformGeo for FastlyPlatformGeo {
 ///
 /// `kv_store` is an [`Arc<dyn PlatformKvStore>`] opened by the caller for
 /// the primary KV store. Use [`open_kv_store`] to construct it.
+///
+/// `auction_event_log_endpoint` names the Fastly real-time log endpoint used
+/// for auction telemetry.
 #[must_use]
 pub fn build_runtime_services(
     req: &Request,
     kv_store: Arc<dyn PlatformKvStore>,
+    auction_event_log_endpoint: &str,
 ) -> RuntimeServices {
     RuntimeServices::builder()
         .config_store(Arc::new(FastlyPlatformConfigStore))
@@ -576,6 +580,9 @@ pub fn build_runtime_services(
             server_hostname: std::env::var("FASTLY_HOSTNAME").ok(),
             server_region: std::env::var("FASTLY_REGION").ok(),
         })
+        .auction_event_sink(std::sync::Arc::new(
+            crate::auction_sink::FastlyAuctionEventSink::new(auction_event_log_endpoint),
+        ))
         .build()
 }
 
@@ -696,7 +703,7 @@ mod tests {
     #[test]
     fn build_runtime_services_client_info_is_none_without_tls() {
         let req = Request::get("https://example.com/");
-        let services = build_runtime_services(&req, noop_kv_store());
+        let services = build_runtime_services(&req, noop_kv_store(), "ts_auction_events");
 
         assert!(
             services.client_info().tls_protocol.is_none(),
@@ -711,7 +718,7 @@ mod tests {
     #[test]
     fn build_runtime_services_returns_cloneable_services() {
         let req = Request::get("https://example.com/");
-        let services = build_runtime_services(&req, noop_kv_store());
+        let services = build_runtime_services(&req, noop_kv_store(), "ts_auction_events");
         let cloned = services.clone();
 
         assert_eq!(

crates/trusted-server-core/src/auction/README.md

@@ -445,6 +445,7 @@ enabled = true                      # Enable/disable auction orchestration
 providers = ["prebid", "aps"]        # List of bidder providers
 mediator = "adserver_mock"          # Optional: if set, uses mediation; if omitted, highest bid wins
 timeout_ms = 2000                   # Overall auction timeout
+telemetry_log_endpoint = "ts_auction_events"  # Fastly auction telemetry log endpoint
 ```
 
 **Strategy Auto-Detection:**

crates/trusted-server-core/src/auction/endpoints.rs

@@ -9,6 +9,7 @@ use serde_json::Value as JsonValue;
 
 use crate::auction::formats::AdRequest;
 use crate::auction::orchestrator::OrchestrationResult;
+use crate::auction::telemetry::{emit_completed_auction_telemetry, AuctionSource};
 use crate::consent::{consent_allows_server_side_auction, gate_eids_by_consent};
 use crate::constants::COOKIE_TS_EIDS;
 use crate::ec::eids::{resolve_partner_ids, to_eids};
@@ -270,6 +271,15 @@ pub async fn handle_auction(
         result.total_time_ms
     );
 
+    // Emit completed-auction telemetry off the response path via the shared
+    // helper. Buffered/non-blocking in production, no-op by default in tests.
+    emit_completed_auction_telemetry(
+        services,
+        AuctionSource::AuctionApi,
+        &auction_request,
+        &result,
+    );
+
     // Convert to OpenRTB response format with inline creative HTML
     convert_to_openrtb_response(&result, settings, &auction_request, ec_context.ec_allowed())
 }
@@ -490,8 +500,10 @@ mod tests {
     use crate::consent::jurisdiction::Jurisdiction;
     use crate::consent::types::ConsentContext;
     use crate::openrtb::Uid;
-    use crate::platform::test_support::noop_services;
-    use crate::platform::{PlatformPendingRequest, PlatformResponse};
+    use crate::platform::test_support::{
+        build_services_with_http_client, noop_services, StubHttpClient,
+    };
+    use crate::platform::{PlatformHttpRequest, PlatformPendingRequest, PlatformResponse};
     use crate::test_support::tests::create_test_settings;
     use base64::engine::general_purpose::STANDARD as BASE64;
     use base64::Engine as _;
@@ -543,6 +555,61 @@ mod tests {
         }
     }
 
+    /// Provider that launches through the stub HTTP client and parses a no-bid
+    /// success, so `run_auction` returns a completed `OrchestrationResult`. This
+    /// is the path that must emit telemetry.
+    struct StubLaunchProvider;
+
+    #[async_trait::async_trait(?Send)]
+    impl AuctionProvider for StubLaunchProvider {
+        fn provider_name(&self) -> &'static str {
+            "stub_provider"
+        }
+
+        async fn request_bids(
+            &self,
+            _request: &AuctionRequest,
+            context: &AuctionContext<'_>,
+        ) -> Result<PlatformPendingRequest, Report<TrustedServerError>> {
+            let req = PlatformHttpRequest::new(
+                Request::builder()
+                    .method("POST")
+                    .uri("https://example.com/bid")
+                    .body(EdgeBody::empty())
+                    .expect("should build stub bid request"),
+                "stub-backend",
+            );
+            context
+                .services
+                .http_client()
+                .send_async(req)
+                .await
+                .change_context(TrustedServerError::Auction {
+                    message: "stub launch failed".to_string(),
+                })
+        }
+
+        async fn parse_response(
+            &self,
+            _response: PlatformResponse,
+            response_time_ms: u64,
+        ) -> Result<AuctionResponse, Report<TrustedServerError>> {
+            Ok(AuctionResponse::success(
+                "stub_provider",
+                vec![],
+                response_time_ms,
+            ))
+        }
+
+        fn timeout_ms(&self) -> u32 {
+            2000
+        }
+
+        fn backend_name(&self, _timeout_ms: u32) -> Option<String> {
+            Some("stub-backend".to_string())
+        }
+    }
+
     #[tokio::test]
     async fn auction_endpoint_consent_gate_returns_no_bid_without_contacting_providers() {
         // GDPR/unknown jurisdiction lacking effective TCF Purpose 1 must not run
@@ -962,6 +1029,69 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn auction_endpoint_emits_completed_telemetry() {
+        // A non-regulated, ungated auction completes (one provider launches via
+        // the stub HTTP client and parses a no-bid success), so it must emit one
+        // summary row tagged auction_api to the injected sink.
+        let settings = create_test_settings();
+        let config = AuctionConfig {
+            enabled: true,
+            providers: vec!["stub_provider".to_string()],
+            timeout_ms: 2000,
+            mediator: None,
+            ..Default::default()
+        };
+        let mut orchestrator = AuctionOrchestrator::new(config);
+        orchestrator.register_provider(Arc::new(StubLaunchProvider));
+        let http_client = Arc::new(StubHttpClient::new());
+        http_client.push_response(200, b"{}".to_vec());
+        let sink = Arc::new(crate::auction::telemetry::InMemorySink::default());
+        let services =
+            build_services_with_http_client(http_client).with_auction_event_sink(sink.clone());
+        let ec_id = format!("{}.ABC123", "a".repeat(64));
+        let ec_context = make_ec_context(Jurisdiction::NonRegulated, Some(&ec_id));
+
+        let body = json!({
+            "adUnits": [
+                {
+                    "code": "div-gpt-ad-1",
+                    "mediaTypes": { "banner": { "sizes": [[300, 250]] } }
+                }
+            ]
+        });
+        let req = Request::builder()
+            .method("POST")
+            .uri("https://test-publisher.com/auction")
+            .body(EdgeBody::from(
+                serde_json::to_vec(&body).expect("should serialize body"),
+            ))
+            .expect("should build auction request");
+
+        let response = handle_auction(
+            &settings,
+            &orchestrator,
+            None,
+            None,
+            &ec_context,
+            &services,
+            req,
+        )
+        .await
+        .expect("auction should return a valid response");
+
+        assert_eq!(response.status(), StatusCode::OK, "should return 200");
+        let rows = sink.rows();
+        assert!(
+            rows.iter().any(
+                |r| r.event_kind == crate::auction::telemetry::EventKind::Summary
+                    && r.auction_source == crate::auction::telemetry::AuctionSource::AuctionApi
+            ),
+            "should emit a summary row tagged auction_api, got {} rows",
+            rows.len()
+        );
+    }
+
     #[tokio::test]
     async fn auction_rejects_oversized_body() {
         use edgezero_core::body::Body as EdgeBody;

crates/trusted-server-core/src/auction/mod.rs

@@ -19,6 +19,7 @@ pub mod endpoints;
 pub mod formats;
 pub mod orchestrator;
 pub mod provider;
+pub mod telemetry;
 #[cfg(test)]
 pub(crate) mod test_support;
 pub mod types;