Migrate handler layer to HTTP types (PR 12) #624
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,17 +1,21 @@ | ||
| use edgezero_core::body::Body as EdgeBody; | ||
| use edgezero_core::http::{ | ||
| header, HeaderName, HeaderValue, Method, Request as HttpRequest, Response as HttpResponse, | ||
| }; | ||
| use error_stack::Report; | ||
| use fastly::http::Method as FastlyMethod; | ||
| use fastly::{Request as FastlyRequest, Response as FastlyResponse}; | ||
|
|
||
| use trusted_server_core::auction::endpoints::handle_auction; | ||
| use trusted_server_core::auction::{build_orchestrator, AuctionOrchestrator}; | ||
| use trusted_server_core::auth::enforce_basic_auth; | ||
| use trusted_server_core::compat; | ||
| use trusted_server_core::constants::{ | ||
| ENV_FASTLY_IS_STAGING, ENV_FASTLY_SERVICE_VERSION, HEADER_X_GEO_INFO_AVAILABLE, | ||
| HEADER_X_TS_ENV, HEADER_X_TS_VERSION, | ||
| }; | ||
| use trusted_server_core::error::{IntoHttpResponse, TrustedServerError}; | ||
| use trusted_server_core::geo::GeoInfo; | ||
| use trusted_server_core::integrations::IntegrationRegistry; | ||
| use trusted_server_core::platform::RuntimeServices; | ||
| use trusted_server_core::proxy::{ | ||
|
|
@@ -41,20 +45,17 @@ use crate::platform::{build_runtime_services, open_kv_store, UnavailableKvStore} | |
|
|
||
| /// Entry point for the Fastly Compute program. | ||
| /// | ||
| /// Uses an undecorated `main()` with `FastlyRequest::from_client()` instead of | ||
| /// `#[fastly::main]` so we can call `send_to_client()` explicitly when needed. | ||
| fn main() { | ||
| init_logger(); | ||
|
|
||
| let mut req = FastlyRequest::from_client(); | ||
|
|
||
| // Keep the health probe independent from settings loading and routing so | ||
| // readiness checks still get a cheap liveness response during startup. | ||
| if req.get_method() == FastlyMethod::GET && req.get_path() == "/health" { | ||
| FastlyResponse::from_status(200) | ||
| .with_body_text_plain("ok") | ||
| .send_to_client(); | ||
| return; | ||
|
|
@@ -89,71 +90,63 @@ fn main() { | |
| } | ||
| }; | ||
|
|
||
| let kv_store = std::sync::Arc::new(UnavailableKvStore) | ||
| as std::sync::Arc<dyn trusted_server_core::platform::PlatformKvStore>; | ||
| // Strip client-spoofable forwarded headers at the edge before building | ||
| // 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 http_req = compat::from_fastly_request(req); | ||
|
|
||
| let mut response = futures::executor::block_on(route_request( | ||
| &settings, | ||
| &orchestrator, | ||
| &integration_registry, | ||
| &runtime_services, | ||
| http_req, | ||
| )) | ||
| .unwrap_or_else(|e| http_error_response(&e)); | ||
|
|
||
| let geo_info = if response.status() == edgezero_core::http::StatusCode::UNAUTHORIZED { | ||
| None | ||
| } else { | ||
| runtime_services | ||
| .geo() | ||
| .lookup(runtime_services.client_info().client_ip) | ||
| .unwrap_or_else(|e| { | ||
| log::warn!("geo lookup failed: {e}"); | ||
| None | ||
| }) | ||
| }; | ||
|
Collaborator
There was a problem hiding this comment. 🤔 thinking — 401 responses no longer carry geo headers — silent behavior change. Previously This may be intentional (avoid leaking geo to unauthenticated callers) but it's a silent change. Add a one-line comment to make the rationale explicit: // Skip geo lookup for 401s so geo headers are not exposed to unauthenticated
// callers. Authenticated routes do their own lookups for consent context.
let geo_info = if response.status() == StatusCode::UNAUTHORIZED { … }; |
||
|
|
||
| finalize_response(&settings, geo_info.as_ref(), &mut response); | ||
|
|
||
| compat::to_fastly_response(response).send_to_client(); | ||
| } | ||
|
|
||
| async fn route_request( | ||
| settings: &Settings, | ||
| orchestrator: &AuctionOrchestrator, | ||
| integration_registry: &IntegrationRegistry, | ||
| runtime_services: &RuntimeServices, | ||
| req: HttpRequest, | ||
| ) -> Result<HttpResponse, Report<TrustedServerError>> { | ||
| // `get_settings()` should already have rejected invalid handler regexes. | ||
| // Keep this fallback so manually-constructed or otherwise unprepared | ||
| // settings still become an error response instead of panicking. | ||
| match enforce_basic_auth(settings, &req) { | ||
| Ok(Some(response)) => return Ok(response), | ||
| Ok(None) => {} | ||
| Err(e) => return Err(e), | ||
| } | ||
|
|
||
| // Get path and method for routing | ||
| let path = req.uri().path().to_string(); | ||
| let method = req.method().clone(); | ||
|
|
||
| // Match known routes and handle them | ||
| match (method, path.as_str()) { | ||
| // Serve the tsjs library | ||
| (Method::GET, path) if path.starts_with("/static/tsjs=") => { | ||
| handle_tsjs_dynamic(&req, integration_registry) | ||
|
|
@@ -199,14 +192,24 @@ async fn route_request( | |
| (Method::POST, "/first-party/proxy-rebuild") => { | ||
| handle_first_party_proxy_rebuild(settings, runtime_services, req).await | ||
| } | ||
| (m, path) if integration_registry.has_route(&m, path) => { | ||
| // TODO(PR13): migrate integration trait to http types here | ||
|
Collaborator
There was a problem hiding this comment. 📝 note — |
||
| integration_registry | ||
| .handle_proxy( | ||
| &m, | ||
| path, | ||
| settings, | ||
| runtime_services, | ||
| compat::to_fastly_request(req), | ||
| ) | ||
| .await | ||
| .unwrap_or_else(|| { | ||
| Err(Report::new(TrustedServerError::BadRequest { | ||
| message: format!("Unknown integration route: {path}"), | ||
| })) | ||
| }) | ||
| .map(compat::from_fastly_response) | ||
| } | ||
|
|
||
| // No known route matched, proxy to publisher origin as fallback | ||
| _ => { | ||
|
|
@@ -216,64 +219,48 @@ async fn route_request( | |
| ); | ||
|
|
||
| match runtime_services_for_consent_route(settings, runtime_services) { | ||
| Ok(publisher_services) => handle_publisher_request( | ||
| settings, | ||
| integration_registry, | ||
| &publisher_services, | ||
| req, | ||
| ) | ||
| .await | ||
| .and_then(|pub_response| { | ||
| resolve_publisher_response(pub_response, settings, integration_registry) | ||
| }), | ||
| Err(e) => Err(e), | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| fn resolve_publisher_response( | ||
| publisher_response: PublisherResponse, | ||
| settings: &Settings, | ||
| integration_registry: &IntegrationRegistry, | ||
| ) -> Result<HttpResponse, Report<TrustedServerError>> { | ||
| match publisher_response { | ||
| PublisherResponse::Buffered(response) => Ok(response), | ||
| PublisherResponse::Stream { | ||
| mut response, | ||
| body, | ||
| params, | ||
| } => { | ||
| let mut output = Vec::new(); | ||
| stream_publisher_body(body, &mut output, ¶ms, settings, integration_registry)?; | ||
|
Collaborator
There was a problem hiding this comment. 🔧 wrench — Streaming regression: this arm now buffers the full pipeline output instead of streaming chunks to the client. Before this PR, the adapter handled After this PR,
Structural cause: Fix (preferred) — keep the streaming dispatch in the adapter: // Have route_request return Result<HandlerOutcome, …> where HandlerOutcome
// distinguishes streamed vs buffered responses, then in main():
match outcome {
HandlerOutcome::Streaming { mut response, body, params } => {
finalize_response(&settings, geo_info.as_ref(), &mut response);
let mut fastly_resp = compat::to_fastly_response_skeleton(response);
let mut sb = fastly_resp.stream_to_client();
if let Err(e) = stream_publisher_body(body, &mut sb, ¶ms, &settings, &integration_registry) {
log::error!("Streaming processing failed: {e:?}");
drop(sb);
} else if let Err(e) = sb.finish() {
log::error!("Failed to finish streaming body: {e}");
}
}
HandlerOutcome::Buffered(response) => {
compat::to_fastly_response(response).send_to_client();
}
}Alternative — fold this arm into No test today asserts streaming behavior — the publisher tests ( |
||
| response.headers_mut().insert( | ||
| header::CONTENT_LENGTH, | ||
| HeaderValue::from(output.len() as u64), | ||
| ); | ||
| *response.body_mut() = EdgeBody::from(output); | ||
| Ok(response) | ||
| } | ||
| PublisherResponse::PassThrough { mut response, body } => { | ||
| *response.body_mut() = body; | ||
| Ok(response) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| fn runtime_services_for_consent_route( | ||
|
|
@@ -302,21 +289,48 @@ fn runtime_services_for_consent_route( | |
| /// Header precedence (last write wins): geo headers are set first, then | ||
| /// version/staging, then operator-configured `settings.response_headers`. | ||
| /// This means operators can intentionally override any managed header. | ||
| fn finalize_response(settings: &Settings, geo_info: Option<&GeoInfo>, response: &mut HttpResponse) { | ||
| if let Some(geo) = geo_info { | ||
| geo.set_response_headers(response); | ||
| } else { | ||
| response.headers_mut().insert( | ||
| HEADER_X_GEO_INFO_AVAILABLE, | ||
| HeaderValue::from_static("false"), | ||
| ); | ||
| } | ||
|
|
||
| if let Ok(v) = ::std::env::var(ENV_FASTLY_SERVICE_VERSION) { | ||
| if let Ok(value) = HeaderValue::from_str(&v) { | ||
| response.headers_mut().insert(HEADER_X_TS_VERSION, value); | ||
| } else { | ||
| log::warn!("Skipping invalid FASTLY_SERVICE_VERSION response header value"); | ||
| } | ||
| } | ||
| if ::std::env::var(ENV_FASTLY_IS_STAGING).as_deref() == Ok("1") { | ||
| response | ||
| .headers_mut() | ||
| .insert(HEADER_X_TS_ENV, HeaderValue::from_static("staging")); | ||
| } | ||
|
|
||
| for (key, value) in &settings.response_headers { | ||
| let header_name = HeaderName::from_bytes(key.as_bytes()) | ||
| .expect("settings.response_headers validated at load time"); | ||
| let header_value = | ||
| HeaderValue::from_str(value).expect("settings.response_headers validated at load time"); | ||
| response.headers_mut().insert(header_name, header_value); | ||
| } | ||
| } | ||
|
|
||
| fn http_error_response(report: &Report<TrustedServerError>) -> HttpResponse { | ||
| let root_error = report.current_context(); | ||
| log::error!("Error occurred: {:?}", report); | ||
|
|
||
|
prk-Jr marked this conversation as resolved.
|
||
| let mut response = | ||
| HttpResponse::new(EdgeBody::from(format!("{}\n", root_error.user_message()))); | ||
| *response.status_mut() = root_error.status_code(); | ||
| response.headers_mut().insert( | ||
| header::CONTENT_TYPE, | ||
| HeaderValue::from_static("text/plain; charset=utf-8"), | ||
| ); | ||
| response | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.