Add User-Agent, fetch context headers, and Web API tests

- Set default User-Agent (openworkers-runner/{version}) on reqwest client
- Inject x-fetch-worker and x-fetch-tenant-id on outgoing requests
- Disable redirect auto-follow for worker fetch
- Add URL, URLSearchParams, btoa/atob integration tests
- Polish adaptive span processor and telemetry
- Use runtime-v8 v0.13.1 from remote
- Bump to v0.13.2

Author: max-lt

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "openworkers-runner"
-version = "0.13.1"
+version = "0.13.2"
 edition = "2024"
 license = "MIT"
 default-run = "openworkers-runner"
@@ -58,7 +58,7 @@ reqwest = { version = "0.13.1", default-features = false, features = ["rustls",
 openworkers-core = { git = "https://github.com/openworkers/openworkers-core", tag = "v0.13.0", features = ["hyper"] }
 
 # Runtime backend (v8 only for now, others require older version of core)
-openworkers-runtime-v8 = { git = "https://github.com/openworkers/openworkers-runtime-v8", tag = "v0.13.0", optional = true, features = ["ptrcomp"] }
+openworkers-runtime-v8 = { git = "https://github.com/openworkers/openworkers-runtime-v8", tag = "v0.13.1", optional = true, features = ["ptrcomp"] }
 
 # WASM runtime (optional)
 # openworkers-runtime-wasm = { path = "../openworkers-runtime-wasm", optional = true }

Cargo.toml

@@ -26,12 +26,12 @@
 //! Traffic is measured with exponential decay (60s half-life) to create
 //! a sliding window effect.
 
-use opentelemetry::trace::TraceId;
-use opentelemetry_sdk::export::trace::SpanData;
-use opentelemetry_sdk::trace::{Context, SpanProcessor};
+use opentelemetry::Context;
+use opentelemetry_sdk::error::OTelSdkResult;
+use opentelemetry_sdk::trace::{SpanData, SpanProcessor};
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 /// Decay half-life for traffic measurement (seconds)
 const DECAY_HALF_LIFE: f64 = 60.0;
@@ -40,17 +40,20 @@ const DECAY_HALF_LIFE: f64 = 60.0;
 const TRAFFIC_THRESHOLD: f64 = 10.0;
 
 /// Span processor that applies adaptive sampling at export time
+#[derive(Debug)]
 pub struct AdaptiveSpanProcessor<T: SpanProcessor> {
     inner: T,
     state: Arc<RwLock<ProcessorState>>,
     min_rate: f64,
     max_rate: f64,
 }
 
+#[derive(Debug)]
 struct ProcessorState {
     worker_counts: HashMap<String, WorkerStats>,
 }
 
+#[derive(Debug)]
 struct WorkerStats {
     count: f64,
     last_update: Instant,
@@ -75,7 +78,7 @@ impl<T: SpanProcessor> AdaptiveSpanProcessor<T> {
             .iter()
             .find(|kv| kv.key.as_str() == "worker_id")
             .map(|kv| kv.value.to_string())
-            .filter(|id| !id.is_empty() && id != "Empty");
+            .filter(|id: &String| !id.is_empty() && id != "Empty");
 
         // Always export spans without worker_id (system spans, etc.)
         let worker_id = match worker_id {
@@ -122,8 +125,7 @@ impl<T: SpanProcessor> AdaptiveSpanProcessor<T> {
         stats.count += 1.0;
 
         // Asymptotic sampling rate
-        self.min_rate
-            + (self.max_rate - self.min_rate) / (1.0 + stats.count / TRAFFIC_THRESHOLD)
+        self.min_rate + (self.max_rate - self.min_rate) / (1.0 + stats.count / TRAFFIC_THRESHOLD)
     }
 }
 
@@ -138,11 +140,11 @@ impl<T: SpanProcessor> SpanProcessor for AdaptiveSpanProcessor<T> {
         }
     }
 
-    fn force_flush(&self) -> opentelemetry::trace::TraceResult<()> {
+    fn force_flush(&self) -> OTelSdkResult {
         self.inner.force_flush()
     }
 
-    fn shutdown(&self) -> opentelemetry::trace::TraceResult<()> {
-        self.inner.shutdown()
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
+        self.inner.shutdown_with_timeout(timeout)
     }
 }

src/adaptive_span_processor.rs

@@ -42,7 +42,9 @@ use tracing::Instrument;
 use crate::limiter::BindingLimiters;
 #[cfg(feature = "database")]
 use crate::services::database::{self as db_service, QueryMode};
-use crate::services::fetch::{do_fetch, generate_request_id, try_internal_worker_route};
+use crate::services::fetch::{
+    FetchContext, do_fetch, generate_request_id, try_internal_worker_route,
+};
 use crate::services::kv as kv_service;
 use crate::services::storage::{build_s3_url, execute_s3_operation, sign_s3_request};
 use crate::store::{Binding, DatabaseConfig, KvConfig, StorageConfig, WorkerBindingConfig};
@@ -241,7 +243,11 @@ impl RunnerOperations {
                     &config.bucket,
                 )?;
 
-                do_fetch(auth_request, &self.stats, Some(&signed_headers)).await
+                let ctx = FetchContext {
+                    worker_id: self.worker_id.as_deref(),
+                    user_id: self.user_id.as_deref(),
+                };
+                do_fetch(auth_request, &self.stats, Some(&signed_headers), Some(&ctx)).await
             }
             .instrument(span),
         )
@@ -284,16 +290,21 @@ impl OperationsHandler for RunnerOperations {
                     self.worker_id
                 );
 
+                let ctx = FetchContext {
+                    worker_id: self.worker_id.as_deref(),
+                    user_id: self.user_id.as_deref(),
+                };
+
                 // Check if this is an internal worker URL that should be routed directly
                 if let Some(internal_request) = try_internal_worker_route(&request) {
                     tracing::debug!(
                         "[ops] fetch shortcut: {} -> internal routing via x-worker-name",
                         request.url
                     );
-                    return do_fetch(internal_request, &self.stats, None).await;
+                    return do_fetch(internal_request, &self.stats, None, Some(&ctx)).await;
                 }
 
-                do_fetch(request, &self.stats, None).await
+                do_fetch(request, &self.stats, None, Some(&ctx)).await
             }
             .instrument(span),
         )
@@ -617,7 +628,7 @@ impl OperationsHandler for RunnerOperations {
                 };
 
                 // Execute the request through the runner
-                do_fetch(internal_request, &OperationsStats::new(), None).await
+                do_fetch(internal_request, &OperationsStats::new(), None, None).await
             }
             .instrument(span),
         )

src/ops.rs

@@ -57,10 +57,12 @@ thread_local! {
             .unwrap_or(100);
 
         reqwest::Client::builder()
+            .user_agent(format!("openworkers-runner/{}", env!("CARGO_PKG_VERSION")))
             .pool_max_idle_per_host(pool_size)
             .pool_idle_timeout(Duration::from_secs(90))
             .connect_timeout(Duration::from_secs(5))
             .timeout(Duration::from_secs(30))
+            .redirect(reqwest::redirect::Policy::none())
             .build()
             .expect("Failed to create HTTP client")
     });
@@ -109,6 +111,12 @@ pub fn try_internal_worker_route(request: &HttpRequest) -> Option<HttpRequest> {
     })
 }
 
+/// Optional worker context injected as headers on outgoing fetch requests.
+pub struct FetchContext<'a> {
+    pub worker_id: Option<&'a str>,
+    pub user_id: Option<&'a str>,
+}
+
 /// Execute an HTTP request with streaming response.
 ///
 /// This is the core HTTP function used by both direct fetch and binding fetches.
@@ -117,6 +125,7 @@ pub async fn do_fetch(
     request: HttpRequest,
     stats: &OperationsStats,
     extra_headers: Option<&HashMap<String, String>>,
+    ctx: Option<&FetchContext<'_>>,
 ) -> Result<HttpResponse, String> {
     use std::sync::atomic::Ordering;
 
@@ -142,13 +151,24 @@ pub async fn do_fetch(
         req_builder = req_builder.header(key, value);
     }
 
-    // Add extra headers if provided
+    // Add extra headers if provided (e.g., AWS signatures)
     if let Some(headers) = extra_headers {
         for (key, value) in headers {
             req_builder = req_builder.header(key, value);
         }
     }
 
+    // Add worker context headers for observability
+    if let Some(ctx) = ctx {
+        if let Some(worker_id) = ctx.worker_id {
+            req_builder = req_builder.header("x-fetch-worker", worker_id);
+        }
+
+        if let Some(user_id) = ctx.user_id {
+            req_builder = req_builder.header("x-fetch-tenant-id", user_id);
+        }
+    }
+
     // Add body if present
     if let RequestBody::Bytes(body) = &request.body {
         stats

src/services/fetch.rs

@@ -126,14 +126,15 @@ fn init_otel() -> Result<(), Box<dyn std::error::Error>> {
     let span_exporter = span_builder.build()?;
 
     // Create batch processor
-    let batch_processor = opentelemetry_sdk::trace::BatchSpanProcessor::builder(span_exporter, opentelemetry_sdk::runtime::Tokio).build();
+    let batch_processor =
+        opentelemetry_sdk::trace::BatchSpanProcessor::builder(span_exporter).build();
 
     // Wrap with adaptive processor for tail-based sampling
     // This filters spans at export time based on worker traffic
     let adaptive_processor = crate::adaptive_span_processor::AdaptiveSpanProcessor::new(
         batch_processor,
-        0.01,  // min_rate: 1% for high-traffic workers
-        1.0,   // max_rate: 100% for low-traffic workers
+        0.01, // min_rate: 1% for high-traffic workers
+        1.0,  // max_rate: 100% for low-traffic workers
     );
 
     // Create tracer provider