GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 24 Pull Requests 7 Actions 1 Packages Projects 1 Releases Wiki Activity

Merge remote-tracking branch 'refs/remotes/origin/master' into feat/gptimage2to2.5
Project CI / AI game creator shell Rust shard 1/4 (pull_request) Successful in 7m17s
Details
Project CI / AI game creator shell Rust shard 3/4 (pull_request) Successful in 7m25s
Details
Project CI / AI game creator shell Rust shard 2/4 (pull_request) Successful in 7m34s
Details
Project CI / AI game creator shell Rust shard 4/4 (pull_request) Successful in 7m37s
Details
Project CI / AI game creator shell Rust smoke (pull_request) Successful in 2m3s
Details
Project CI / AI game creator shell Rust crates (pull_request) Successful in 3m23s
Details
Project CI / Frontend tests (pull_request) Failing after 14m1s
Details
Project CI / Repository checks (pull_request) Successful in 12m35s
Details
Project CI / Native shell tests (pull_request) Successful in 17m2s
Details
Project CI / Backend tests (pull_request) Successful in 17m18s
Details
Project CI / AI game creator shell web tests (pull_request) Failing after 14m48s
Details

Browse Source
This commit is contained in:
王德宇
2026-09-19 15:39:41 +08:00
parent c8311a47b0 33f5ad68bf
commit 3da0936c1f
228 changed files with 15831 additions and 3162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server-rs/crates/api-server/src/app.rs
+263 -121
View File
@@ -1,3 +1,5 @@
use std::time::Duration;
use axum::{
Router,
body::Body,
@@ -9,8 +11,8 @@ use axum::{
};
use serde_json::json;
use tower_http::{
classify::ServerErrorsFailureClass,
trace::{DefaultOnRequest, TraceLayer},
classify::{ServerErrorsAsFailures, ServerErrorsFailureClass, SharedClassifier},
trace::{DefaultOnBodyChunk, DefaultOnEos, DefaultOnRequest, TraceLayer},
};
use tracing::{Level, Span, error, info_span};
@@ -85,66 +87,7 @@ pub fn build_router(state: AppState) -> Router {
state.clone(),
record_http_observability,
))
// 当前阶段先统一挂接 HTTP tracing，后续 request_id、响应头与错误中间件继续在这里扩展。
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &Request<Body>| {
let request_id =
resolve_request_id(request).unwrap_or_else(|| "unknown".to_string());
let route = crate::telemetry::observability_route(request.uri().path());
let scheme = crate::telemetry::resolve_request_scheme(request.headers());
let span_name = format!("{} {}", request.method(), route);
info_span!(
"http.request",
otel.kind = "server",
otel.name = %span_name,
otel.status_code = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
method = %request.method(),
http.request.method = %request.method(),
http.route = %route,
url.scheme = %scheme,
url.path = %request.uri().path(),
request_id = %request_id,
status = tracing::field::Empty,
latency_ms = tracing::field::Empty,
)
})
.on_request(DefaultOnRequest::new().level(Level::INFO))
.on_response(
|response: &axum::response::Response,
latency: std::time::Duration,
span: &Span| {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
let status = response.status().as_u16();
span.record("status", status);
span.record("http.response.status_code", status);
span.record(
"otel.status_code",
if response.status().is_server_error() {
"ERROR"
} else {
"OK"
},
);
span.record("latency_ms", latency_ms);
},
)
.on_failure(
|failure: ServerErrorsFailureClass,
latency: std::time::Duration,
span: &Span| {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
error!(
parent: span,
latency_ms,
failure = %failure,
"http request failed"
);
},
),
)
.layer(http_trace_layer())
// request_id 中间件先进入请求链，确保后续 tracing、错误处理和响应头层都能复用同一份请求标识。
.layer(middleware::from_fn(attach_request_context))
.with_state(state)
@@ -159,68 +102,72 @@ pub fn build_spacetime_unavailable_router(message: String) -> Router {
// 依赖不可用模式不挂业务 state，统一返回 503，并继续保留 request_id / API 版本 / 耗时响应头。
.layer(middleware::from_fn(normalize_error_response))
.layer(middleware::from_fn(propagate_request_id_header))
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &Request<Body>| {
let request_id =
resolve_request_id(request).unwrap_or_else(|| "unknown".to_string());
let route = crate::telemetry::observability_route(request.uri().path());
let scheme = crate::telemetry::resolve_request_scheme(request.headers());
let span_name = format!("{} {}", request.method(), route);
info_span!(
"http.request",
otel.kind = "server",
otel.name = %span_name,
otel.status_code = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
method = %request.method(),
http.request.method = %request.method(),
http.route = %route,
url.scheme = %scheme,
url.path = %request.uri().path(),
request_id = %request_id,
status = tracing::field::Empty,
latency_ms = tracing::field::Empty,
)
})
.on_request(DefaultOnRequest::new().level(Level::INFO))
.on_response(
|response: &axum::response::Response,
latency: std::time::Duration,
span: &Span| {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
let status = response.status().as_u16();
span.record("status", status);
span.record("http.response.status_code", status);
span.record(
"otel.status_code",
if response.status().is_server_error() {
"ERROR"
} else {
"OK"
},
);
span.record("latency_ms", latency_ms);
},
)
.on_failure(
|failure: ServerErrorsFailureClass,
latency: std::time::Duration,
span: &Span| {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
error!(
parent: span,
latency_ms,
failure = %failure,
"http request failed"
);
},
),
)
.layer(http_trace_layer())
.layer(middleware::from_fn(attach_request_context))
}
type HttpTraceLayer = TraceLayer<
SharedClassifier<ServerErrorsAsFailures>,
fn(&Request<Body>) -> Span,
DefaultOnRequest,
fn(&Response, Duration, &Span),
DefaultOnBodyChunk,
DefaultOnEos,
fn(ServerErrorsFailureClass, Duration, &Span),
>;
fn http_trace_layer() -> HttpTraceLayer {
TraceLayer::new_for_http()
.make_span_with(make_http_request_span as fn(&Request<Body>) -> Span)
.on_request(DefaultOnRequest::new().level(Level::INFO))
.on_response(record_http_response_span as fn(&Response, Duration, &Span))
.on_failure(record_http_failure as fn(ServerErrorsFailureClass, Duration, &Span))
}
fn make_http_request_span(request: &Request<Body>) -> Span {
let request_id = resolve_request_id(request).unwrap_or_else(|| "unknown".to_string());
let route = crate::telemetry::observability_route(request);
let scheme = crate::telemetry::resolve_request_scheme(request.headers());
let span_name = format!("{} {}", request.method(), route);
info_span!(
"http.request",
otel.kind = "server",
otel.name = %span_name,
otel.status_code = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
method = %request.method(),
http.request.method = %request.method(),
http.route = %route,
url.scheme = %scheme,
url.path = %request.uri().path(),
request_id = %request_id,
status = tracing::field::Empty,
latency_ms = tracing::field::Empty,
)
}
fn record_http_response_span(response: &Response, latency: Duration, span: &Span) {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
let status = response.status().as_u16();
span.record("status", status);
span.record("http.response.status_code", status);
span.record(
"otel.status_code",
if response.status().is_server_error() {
"ERROR"
} else {
"OK"
},
);
span.record("latency_ms", latency_ms);
}
fn record_http_failure(failure: ServerErrorsFailureClass, latency: Duration, span: &Span) {
let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64;
error!(parent: span, latency_ms, failure = %failure, "http request failed");
}
#[derive(Clone, Debug)]
struct SpacetimeUnavailableState {
message: std::sync::Arc<str>,
@@ -308,6 +255,201 @@ mod tests {
const TEST_PASSWORD: &str = "secret123";
const INTERNAL_TEST_SECRET: &str = "test-internal-secret";
mod http_tracing {
use std::{
collections::BTreeMap,
sync::{Arc, Mutex},
};
use axum::extract::FromRef;
use tracing::{
Metadata, Subscriber,
field::{Field, Visit},
instrument::WithSubscriber,
span::{Attributes, Id, Record},
};
use super::*;
use crate::state::{BackpressureState, HttpRequestPermitPoolKind};
#[derive(Clone, Default)]
struct HttpSpanCapture(Arc<Mutex<Vec<BTreeMap<String, String>>>>);
struct SpanFields<'a>(&'a mut BTreeMap<String, String>);
impl Visit for SpanFields<'_> {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
self.0
.insert(field.name().to_string(), format!("{value:?}"));
}
fn record_str(&mut self, field: &Field, value: &str) {
self.0.insert(field.name().to_string(), value.to_string());
}
}
impl Subscriber for HttpSpanCapture {
fn enabled(&self, metadata: &Metadata<'_>) -> bool {
metadata.is_span() && metadata.name() == "http.request"
}
fn new_span(&self, attributes: &Attributes<'_>) -> Id {
let mut spans = self.0.lock().expect("span capture should lock");
let mut fields = BTreeMap::new();
attributes.record(&mut SpanFields(&mut fields));
spans.push(fields);
Id::from_u64(spans.len() as u64)
}
fn record(&self, span: &Id, values: &Record<'_>) {
let mut spans = self.0.lock().expect("span capture should lock");
values.record(&mut SpanFields(&mut spans[span.into_u64() as usize - 1]));
}
fn record_follows_from(&self, _: &Id, _: &Id) {}
fn event(&self, _: &tracing::Event<'_>) {}
fn enter(&self, _: &Id) {}
fn exit(&self, _: &Id) {}
}
async fn assert_rejection_observed(
app: Router,
request: Request<Body>,
expected_status: StatusCode,
expected_route: &str,
expected_code: &str,
) -> (Value, String) {
let expected_request_id = request.headers().get("x-request-id").cloned();
let path = request.uri().path().to_string();
let capture = HttpSpanCapture::default();
let response = app
.oneshot(request)
.with_subscriber(capture.clone())
.await
.expect("rejected request should complete");
assert_eq!(response.status(), expected_status);
let request_id = response.headers()["x-request-id"]
.to_str()
.expect("request id should be a header string")
.to_string();
assert!(!request_id.is_empty());
assert_ne!(request_id, "unknown");
if let Some(expected_request_id) = expected_request_id {
assert_eq!(response.headers()["x-request-id"], expected_request_id);
}
for name in ["x-api-version", "x-route-version"] {
assert_eq!(response.headers()[name], shared_contracts::api::API_VERSION);
}
assert!(
response.headers()["x-response-time-ms"]
.to_str()
.unwrap()
.parse::<u64>()
.is_ok()
);
if expected_status == StatusCode::TOO_MANY_REQUESTS {
assert_eq!(response.headers()["retry-after"], "1");
}
let payload = read_json_response(response).await;
assert_eq!(payload["error"]["code"], expected_code);
let spans = capture.0.lock().expect("span capture should lock");
assert_eq!(
spans.len(),
1,
"each rejected request should have one HTTP span"
);
let fields = &spans[0];
assert_eq!(fields["request_id"], request_id);
assert_eq!(fields["http.route"], expected_route);
assert_eq!(fields["otel.name"], format!("GET {expected_route}"));
assert_eq!(fields["url.path"], path);
assert_eq!(fields["http.request.method"], "GET");
assert_eq!(fields["status"], expected_status.as_u16().to_string());
assert_eq!(
fields["http.response.status_code"],
expected_status.as_u16().to_string()
);
assert_eq!(
fields["otel.status_code"],
if expected_status.is_server_error() {
"ERROR"
} else {
"OK"
}
);
assert!(fields["latency_ms"].parse::<u64>().is_ok());
(payload, request_id)
}
#[tokio::test]
async fn auth_rejections_share_the_matched_route_template() {
let app =
build_router(AppState::new(AppConfig::default()).expect("state should build"));
for project_id in ["project-one", "project-two"] {
assert_rejection_observed(
app.clone(),
Request::builder()
.uri(format!(
"/api/editor/projects/{project_id}/agent-conversations"
))
.header("x-request-id", format!("req-trace-{project_id}"))
.body(Body::empty())
.expect("request should build"),
StatusCode::UNAUTHORIZED,
"/api/editor/projects/{project_id}/agent-conversations",
"UNAUTHORIZED",
)
.await;
}
}
#[tokio::test]
async fn backpressure_rejection_keeps_generated_context_and_headers() {
let config = AppConfig {
max_concurrent_requests: Some(1),
..AppConfig::default()
};
let state = AppState::new(config).expect("state should build");
let (_, pool) = BackpressureState::from_ref(&state)
.request_permit_pool(HttpRequestPermitPoolKind::Default)
.expect("default request pool should exist");
let _held_permit = pool
.try_acquire_owned()
.expect("pool should have one permit");
let (payload, request_id) = assert_rejection_observed(
build_router(state),
Request::builder()
.uri("/api/editor/projects/project-one/agent-conversations")
.body(Body::empty())
.expect("request should build"),
StatusCode::TOO_MANY_REQUESTS,
"/api/editor/projects/{project_id}/agent-conversations",
"TOO_MANY_REQUESTS",
)
.await;
assert_eq!(payload["meta"]["requestId"], request_id);
}
#[tokio::test]
async fn unavailable_router_rejection_keeps_generated_context_and_headers() {
let (payload, request_id) = assert_rejection_observed(
build_spacetime_unavailable_router("test unavailable".to_string()),
Request::builder()
.uri("/api/auth/login-options")
.body(Body::empty())
.expect("request should build"),
StatusCode::SERVICE_UNAVAILABLE,
"/api/*",
"SERVICE_UNAVAILABLE",
)
.await;
assert_eq!(payload["meta"]["requestId"], request_id);
}
}
async fn seed_phone_user_with_password(
state: &AppState,
phone_number: &str,
server-rs/crates/api-server/src/editor_project.rs
+67 -225
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/editor_project/metadata_tests.rs
+364
View File
@@ -0,0 +1,364 @@
use super::*;
use crate::{
request_context::attach_request_context,
state::project_metadata::{EditorProjectMediaRepair, EditorProjectRepository},
};
use axum::{
Router,
body::{Body, to_bytes},
http::Request,
middleware,
routing::{get, patch},
};
use futures_util::future::BoxFuture;
use platform_auth::{AccessTokenClaims, AuthProvider, BindingStatus};
use spacetime_client::EditorProjectLayoutSaveV2AckRecord;
use std::sync::Mutex;
use tower::ServiceExt;
#[derive(Default)]
struct RecordingProjects {
calls: Mutex<Vec<(&'static str, Value)>>,
error: Option<&'static str>,
}
impl RecordingProjects {
fn finish<T: Send + 'static>(
&self,
operation: &'static str,
input: Value,
result: T,
) -> BoxFuture<'_, Result<T, SpacetimeClientError>> {
self.calls.lock().unwrap().push((operation, input));
Box::pin(async move {
match self.error {
Some(message) => Err(SpacetimeClientError::Procedure(message.to_string())),
None => Ok(result),
}
})
}
}
fn project_record(project_id: &str, owner: &str, title: &str) -> EditorProjectRecord {
let viewport = EditorCanvasViewportRecord {
x: 1.0,
y: 2.0,
scale: 1.0,
};
EditorProjectRecord {
project_id: project_id.to_string(),
owner_user_id: owner.to_string(),
title: title.to_string(),
canvas: EditorCanvasRecord {
canvas_id: "canvas-fixture".to_string(),
project_id: project_id.to_string(),
title: title.to_string(),
viewport: viewport.clone(),
layers: json!([]),
revision: 7,
layout_storage_version: 2,
background_color: None,
created_at: "0.000000Z".to_string(),
updated_at: "0.000000Z".to_string(),
},
viewport,
layers: json!([]),
resources: vec![],
created_at: "0.000000Z".to_string(),
updated_at: "0.000000Z".to_string(),
}
}
impl EditorProjectRepository for RecordingProjects {
fn get_recent_editor_project(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Option<EditorProjectRecord>, SpacetimeClientError>> {
let record = project_record("project-fixture", &owner_user_id, "原名称");
self.finish("recent", json!({"owner": owner_user_id}), Some(record))
}
fn list_editor_projects(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Vec<EditorProjectRecord>, SpacetimeClientError>> {
let record = project_record("project-fixture", &owner_user_id, "原名称");
self.finish("list", json!({"owner": owner_user_id}), vec![record])
}
fn get_editor_project(
&self,
input: EditorProjectGetRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
let record = project_record(&input.project_id, &input.owner_user_id, "原名称");
self.finish(
"get",
json!({"owner": input.owner_user_id, "project": input.project_id}),
record,
)
}
fn create_editor_project(
&self,
input: EditorProjectCreateRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
let record = project_record(&input.project_id, &input.owner_user_id, &input.title);
self.finish("create", json!({"owner": input.owner_user_id, "project": input.project_id, "title": input.title, "at": input.now_micros}), record)
}
fn rename_editor_project(
&self,
input: EditorProjectRenameRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
let record = project_record(&input.project_id, &input.owner_user_id, &input.title);
self.finish("rename", json!({"owner": input.owner_user_id, "project": input.project_id, "title": input.title, "at": input.updated_at_micros}), record)
}
fn delete_editor_project(
&self,
input: EditorProjectDeleteRecordInput,
) -> BoxFuture<'_, Result<String, SpacetimeClientError>> {
self.finish(
"delete",
json!({"owner": input.owner_user_id, "project": input.project_id}),
input.project_id,
)
}
fn save_editor_project_layout_v2_ack(
&self,
input: EditorProjectLayoutSaveV2RecordInput,
) -> BoxFuture<'_, Result<EditorProjectLayoutSaveV2AckRecord, SpacetimeClientError>> {
let ack = EditorProjectLayoutSaveV2AckRecord {
project_id: input.project_id.clone(),
canvas_id: "canvas-fixture".to_string(),
revision: input.expected_revision + 1,
updated_at: "1.000000Z".to_string(),
updated_at_micros: input.updated_at_micros,
};
self.finish("layout", json!({"owner": input.owner_user_id, "project": input.project_id, "revision": input.expected_revision, "layers": input.layers_json, "viewport": input.viewport}), ack)
}
}
#[derive(Default)]
struct RecordingMediaRepair {
calls: Mutex<Vec<(String, String)>>,
}
impl EditorProjectMediaRepair for RecordingMediaRepair {
fn repair_project(
&self,
mut project: EditorProjectRecord,
) -> BoxFuture<'_, EditorProjectRecord> {
self.calls
.lock()
.unwrap()
.push((project.project_id.clone(), project.owner_user_id.clone()));
Box::pin(async move {
project.updated_at = "media-repaired".to_string();
project
})
}
}
fn authenticated_owner() -> AuthenticatedAccessToken {
AuthenticatedAccessToken::new(AccessTokenClaims {
iss: "fixture".to_string(),
sub: "owner-from-token".to_string(),
sid: "fixture-session".to_string(),
provider: AuthProvider::Password,
roles: vec!["user".to_string()],
ver: 1,
phone_verified: true,
binding_status: BindingStatus::Active,
display_name: None,
device: None,
iat: 1,
exp: u64::MAX,
})
}
fn metadata_router(projects: Arc<RecordingProjects>, media: Arc<RecordingMediaRepair>) -> Router {
// 直接以窄状态组装实际 handler，不构造 AppConfig、连接数据库或初始化其它平台。
Router::new()
.route("/projects/recent", get(load_recent_editor_project))
.route(
"/projects",
get(list_editor_projects).post(create_editor_project),
)
.route(
"/projects/{project_id}",
get(get_editor_project)
.patch(save_editor_project_layout)
.delete(delete_editor_project),
)
.route(
"/projects/{project_id}/metadata",
patch(rename_editor_project),
)
.layer(Extension(authenticated_owner()))
.layer(middleware::from_fn(attach_request_context))
.with_state(EditorProjectState::new(projects, media))
}
async fn send(router: &Router, method: &str, path: &str, body: Value) -> (StatusCode, Value) {
let response = router
.clone()
.oneshot(
Request::builder()
.method(method)
.uri(path)
.header("content-type", "application/json")
.header("idempotency-key", "metadata-fixture")
.body(Body::from(body.to_string()))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let body = to_bytes(response.into_body(), 65_536).await.unwrap();
(status, serde_json::from_slice(&body).unwrap())
}
#[tokio::test]
async fn narrow_project_reads_preserve_owner_and_use_media_repair_results() {
let projects = Arc::new(RecordingProjects::default());
let media = Arc::new(RecordingMediaRepair::default());
let router = metadata_router(projects.clone(), media.clone());
for (path, result_path) in [
("/projects/recent", "/project"),
("/projects", "/projects/0"),
("/projects/project-fixture", "/project"),
] {
let (status, body) = send(&router, "GET", path, Value::Null).await;
assert_eq!(status, StatusCode::OK);
let result = body.pointer(result_path).unwrap();
assert_eq!(result["projectId"], "project-fixture");
assert_eq!(result["updatedAt"], "media-repaired");
}
let calls = projects.calls.lock().unwrap();
assert_eq!(
calls.iter().map(|call| call.0).collect::<Vec<_>>(),
["recent", "list", "get"]
);
assert!(
calls
.iter()
.all(|call| call.1["owner"] == "owner-from-token")
);
assert_eq!(
*media.calls.lock().unwrap(),
vec![
(
"project-fixture".to_string(),
"owner-from-token".to_string()
);
3
]
);
}
#[tokio::test]
async fn narrow_project_writes_preserve_owner_idempotency_and_revision() {
let projects = Arc::new(RecordingProjects::default());
let media = Arc::new(RecordingMediaRepair::default());
let router = metadata_router(projects.clone(), media.clone());
let (status, created) = send(
&router,
"POST",
"/projects",
json!({"title": " 测试项目 ", "ownerUserId": "untrusted-owner"}),
)
.await;
assert_eq!(status, StatusCode::OK);
let project_id = editor_idempotent_create_id(
EDITOR_PROJECT_ID_PREFIX,
"owner-from-token",
"internal-editor-project",
"metadata-fixture",
);
assert_eq!(created["project"]["projectId"], project_id);
assert_eq!(created["project"]["title"], "测试项目");
let (status, renamed) = send(
&router,
"PATCH",
"/projects/project-fixture/metadata",
json!({"title": "新名称"}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(renamed["project"]["title"], "新名称");
let (status, saved) = send(
&router,
"PATCH",
"/projects/project-fixture",
json!({"viewport": {"x": 10, "y": 20, "scale": 2}, "layers": [], "expectedRevision": 7}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(saved["revision"], 8);
assert_eq!(saved["projectId"], "project-fixture");
let (status, deleted) = send(&router, "DELETE", "/projects/project-fixture", Value::Null).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(deleted["deletedProjectId"], "project-fixture");
let calls = projects.calls.lock().unwrap();
assert_eq!(
calls.iter().map(|call| call.0).collect::<Vec<_>>(),
["create", "rename", "layout", "delete"]
);
assert!(
calls
.iter()
.all(|call| call.1["owner"] == "owner-from-token")
);
assert_eq!(calls[0].1["project"], project_id);
assert_eq!(calls[2].1["revision"], 7);
assert_eq!(calls[2].1["layers"], "[]");
assert_eq!(
calls[2].1["viewport"],
json!({"x": 10.0, "y": 20.0, "scale": 2.0})
);
assert!(media.calls.lock().unwrap().is_empty());
}
#[tokio::test]
async fn narrow_project_handlers_propagate_facade_errors_without_media_side_effects() {
for (message, expected_status) in [
("项目不存在", StatusCode::NOT_FOUND),
("无权访问该项目", StatusCode::FORBIDDEN),
("项目版本冲突", StatusCode::CONFLICT),
] {
let projects = Arc::new(RecordingProjects {
error: Some(message),
..Default::default()
});
let media = Arc::new(RecordingMediaRepair::default());
let router = metadata_router(projects.clone(), media.clone());
for (method, path, body) in [
("GET", "/projects/recent", Value::Null),
("GET", "/projects", Value::Null),
("GET", "/projects/project-fixture", Value::Null),
("POST", "/projects", json!({"title": "测试"})),
(
"PATCH",
"/projects/project-fixture/metadata",
json!({"title": "测试"}),
),
(
"PATCH",
"/projects/project-fixture",
json!({"viewport": {"x": 0, "y": 0, "scale": 1}, "layers": [], "expectedRevision": 7}),
),
("DELETE", "/projects/project-fixture", Value::Null),
] {
let (status, body) = send(&router, method, path, body).await;
assert_eq!(status, expected_status, "{method} {path}");
assert!(body.to_string().contains(message));
}
assert_eq!(projects.calls.lock().unwrap().len(), 7);
assert!(media.calls.lock().unwrap().is_empty());
}
}
server-rs/crates/api-server/src/external_api_auth.rs
+182 -4
View File
@@ -16,7 +16,7 @@ use crate::{
external_api_keys::{hash_external_api_key, map_external_api_key_error},
http_error::AppError,
request_context::RequestContext,
state::AppState,
state::ExternalApiAuthState,
};
#[derive(Clone, Debug)]
@@ -50,7 +50,7 @@ impl ExternalApiPrincipal {
}
pub async fn require_external_api_key(
State(state): State<AppState>,
State(state): State<ExternalApiAuthState>,
mut request: Request,
next: Next,
) -> Result<Response, AppError> {
@@ -61,7 +61,7 @@ pub async fn require_external_api_key(
.unwrap_or_else(|| "unknown".to_string());
let raw_key = extract_external_api_bearer(request.headers())?;
let key = state
.spacetime_client()
.authenticator()
.authenticate_external_api_key(ExternalApiKeyAuthenticateRecordInput {
key_hash: hash_external_api_key(raw_key.as_str()),
used_at_micros: current_utc_micros(),
@@ -88,7 +88,7 @@ pub async fn require_external_api_key(
}
pub async fn require_external_mcp_api_key(
State(state): State<AppState>,
State(state): State<ExternalApiAuthState>,
request: Request,
next: Next,
) -> Result<Response, AppError> {
@@ -171,6 +171,184 @@ fn extract_external_api_bearer(headers: &HeaderMap) -> Result<String, AppError>
#[cfg(test)]
mod tests {
use super::*;
use crate::state::external_api_auth::ExternalApiKeyAuthenticator;
use axum::{
Router,
body::{Body, to_bytes},
extract::Extension,
middleware,
routing::get,
};
use futures_util::future::BoxFuture;
use spacetime_client::{ExternalApiKeyRecord, SpacetimeClientError};
use std::sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
};
use tower::ServiceExt;
struct RecordingAuthenticator {
requests: Mutex<Vec<ExternalApiKeyAuthenticateRecordInput>>,
error: Option<&'static str>,
}
impl ExternalApiKeyAuthenticator for RecordingAuthenticator {
fn authenticate_external_api_key(
&self,
input: ExternalApiKeyAuthenticateRecordInput,
) -> BoxFuture<'_, Result<ExternalApiKeyRecord, SpacetimeClientError>> {
self.requests.lock().unwrap().push(input);
Box::pin(async move {
if let Some(error) = self.error {
return Err(SpacetimeClientError::Procedure(error.to_string()));
}
Ok(ExternalApiKeyRecord {
key_id: "key-from-store".to_string(),
owner_user_id: "owner-from-store".to_string(),
name: "测试密钥".to_string(),
key_prefix: "tnr_sk_fixture".to_string(),
scopes: vec!["editor:project".to_string()],
created_at: "0.000000Z".to_string(),
last_used_at: None,
revoked_at: None,
updated_at: "0.000000Z".to_string(),
})
})
}
}
fn auth_test_router(
dependency: Arc<RecordingAuthenticator>,
entered: Arc<AtomicUsize>,
mcp: bool,
) -> Router {
let state = ExternalApiAuthState::new(dependency);
let router = Router::new().route("/protected", get(move |Extension(principal): Extension<ExternalApiPrincipal>| {
let entered = entered.clone();
async move {
entered.fetch_add(1, Ordering::Relaxed);
axum::Json(json!({"owner": principal.owner_user_id(), "projectScope": principal.has_scope("editor:project")}))
}
}));
if mcp {
router.layer(middleware::from_fn_with_state(
state,
require_external_mcp_api_key,
))
} else {
router.layer(middleware::from_fn_with_state(
state,
require_external_api_key,
))
}
}
#[tokio::test]
async fn narrow_external_auth_forwards_store_identity_and_only_hashes_credentials() {
let dependency = Arc::new(RecordingAuthenticator {
requests: Mutex::default(),
error: None,
});
let entered = Arc::new(AtomicUsize::new(0));
let response = auth_test_router(dependency.clone(), entered.clone(), false)
.oneshot(
Request::builder()
.uri("/protected")
.header(AUTHORIZATION, "Bearer tnr_sk_fixture_secret")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let principal = response.extensions().get::<ExternalApiPrincipal>().unwrap();
assert_eq!(principal.owner_user_id(), "owner-from-store");
assert_eq!(principal.key_id(), "key-from-store");
let body: serde_json::Value =
serde_json::from_slice(&to_bytes(response.into_body(), 1024).await.unwrap()).unwrap();
assert_eq!(
body,
json!({"owner": "owner-from-store", "projectScope": true})
);
assert_eq!(entered.load(Ordering::Relaxed), 1);
let requests = dependency.requests.lock().unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(
requests[0].key_hash,
hash_external_api_key("tnr_sk_fixture_secret")
);
assert!(requests[0].used_at_micros > 0);
}
#[tokio::test]
async fn narrow_external_auth_preserves_failure_mapping_and_mcp_guide() {
for (message, expected_status) in [
("API Key 不存在", StatusCode::UNAUTHORIZED),
("无权使用此密钥", StatusCode::FORBIDDEN),
("校验请求失败", StatusCode::BAD_REQUEST),
] {
for mcp in [false, true] {
let dependency = Arc::new(RecordingAuthenticator {
requests: Mutex::default(),
error: Some(message),
});
let entered = Arc::new(AtomicUsize::new(0));
let response = auth_test_router(dependency.clone(), entered.clone(), mcp)
.oneshot(
Request::builder()
.uri("/protected")
.header(AUTHORIZATION, "Bearer invalid-fixture-key")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), expected_status);
assert!(
response
.extensions()
.get::<ExternalApiPrincipal>()
.is_none()
);
assert_eq!(entered.load(Ordering::Relaxed), 0);
assert_eq!(dependency.requests.lock().unwrap().len(), 1);
let body = String::from_utf8(
to_bytes(response.into_body(), 16_384)
.await
.unwrap()
.to_vec(),
)
.unwrap();
if mcp && expected_status == StatusCode::UNAUTHORIZED {
assert!(body.contains("MCP_AUTHENTICATION_REQUIRED"));
assert!(!body.contains(message));
} else {
assert!(body.contains(message));
}
}
}
}
#[tokio::test]
async fn narrow_external_auth_rejects_missing_credentials_before_store_access() {
let dependency = Arc::new(RecordingAuthenticator {
requests: Mutex::default(),
error: None,
});
let entered = Arc::new(AtomicUsize::new(0));
let response = auth_test_router(dependency.clone(), entered.clone(), false)
.oneshot(
Request::builder()
.uri("/protected")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
assert!(dependency.requests.lock().unwrap().is_empty());
assert_eq!(entered.load(Ordering::Relaxed), 0);
}
#[test]
fn mcp_authentication_guide_replaces_sensitive_key_diagnostics() {
server-rs/crates/api-server/src/modules/admin.rs
+311 -245
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/modules/editor_project.rs
+557 -222
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/modules/external_api.rs
+423 -172
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/state.rs
+26
View File
@@ -55,6 +55,12 @@ use crate::work_author::{
ORPHAN_WORK_AUTHOR_DISPLAY_NAME, ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ORPHAN_WORK_OWNER_USER_ID,
};
pub(crate) mod external_api_auth;
pub(crate) mod project_metadata;
pub use external_api_auth::ExternalApiAuthState;
pub(crate) use project_metadata::EditorMediaStorageState;
pub use project_metadata::EditorProjectState;
const ADMIN_ROLE: &str = "admin";
const EDITOR_AGENT_LLM_MAX_RETRIES: u32 = 1;
const EDITOR_AGENT_LLM_MAX_RETRY_BACKOFF_MS: u64 = 60_000;
@@ -300,6 +306,8 @@ pub struct AppStateInner {
#[cfg_attr(not(test), allow(dead_code))]
ai_task_service: AiTaskService,
spacetime_client: SpacetimeClient,
external_api_auth_state: ExternalApiAuthState,
editor_project_state: EditorProjectState,
#[cfg(any())]
puzzle_gallery_cache: PuzzleGalleryCache,
tracking_outbox: Option<Arc<TrackingOutbox>>,
@@ -662,6 +670,14 @@ impl AppState {
let character_animation_oss_io_limiter =
Arc::new(Semaphore::new(CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY));
let editor_oss_http_client = build_editor_oss_http_client()?;
let editor_project_state = EditorProjectState::new(
Arc::new(spacetime_client.clone()),
Arc::new(EditorMediaStorageState::new(
spacetime_client.clone(),
oss_client.clone(),
editor_oss_http_client.clone(),
)),
);
let http_request_permit_pools = HttpRequestPermitPools::from_config(&config);
let (profile_recharge_order_updates, _) = broadcast::channel(128);
// `ensure_orphan_work_owner_user` 只为公开作品作者回退提供进程内占位账号，
@@ -709,6 +725,8 @@ impl AppState {
wechat_client,
wechat_pay_client,
ai_task_service,
external_api_auth_state: ExternalApiAuthState::new(Arc::new(spacetime_client.clone())),
editor_project_state,
spacetime_client,
#[cfg(any())]
puzzle_gallery_cache: PuzzleGalleryCache::new(),
@@ -1603,6 +1621,14 @@ impl AppState {
&self.spacetime_client
}
#[cfg(test)]
pub(crate) fn with_external_api_auth_state(mut self, state: ExternalApiAuthState) -> Self {
Arc::get_mut(&mut self.0)
.expect("configure external authentication before sharing AppState")
.external_api_auth_state = state;
self
}
#[cfg(any())]
pub fn puzzle_gallery_cache(&self) -> &PuzzleGalleryCache {
&self.puzzle_gallery_cache
server-rs/crates/api-server/src/state/external_api_auth.rs
+47
View File
@@ -0,0 +1,47 @@
use std::sync::Arc;
use axum::extract::FromRef;
use futures_util::future::BoxFuture;
use spacetime_client::{
ExternalApiKeyAuthenticateRecordInput, ExternalApiKeyRecord, SpacetimeClient,
SpacetimeClientError,
};
use super::AppState;
pub(crate) trait ExternalApiKeyAuthenticator: Send + Sync {
fn authenticate_external_api_key(
&self,
input: ExternalApiKeyAuthenticateRecordInput,
) -> BoxFuture<'_, Result<ExternalApiKeyRecord, SpacetimeClientError>>;
}
impl ExternalApiKeyAuthenticator for SpacetimeClient {
fn authenticate_external_api_key(
&self,
input: ExternalApiKeyAuthenticateRecordInput,
) -> BoxFuture<'_, Result<ExternalApiKeyRecord, SpacetimeClientError>> {
Box::pin(SpacetimeClient::authenticate_external_api_key(self, input))
}
}
#[derive(Clone)]
pub struct ExternalApiAuthState {
authenticator: Arc<dyn ExternalApiKeyAuthenticator>,
}
impl ExternalApiAuthState {
pub(crate) fn new(authenticator: Arc<dyn ExternalApiKeyAuthenticator>) -> Self {
Self { authenticator }
}
pub(crate) fn authenticator(&self) -> &dyn ExternalApiKeyAuthenticator {
self.authenticator.as_ref()
}
}
impl FromRef<AppState> for ExternalApiAuthState {
fn from_ref(state: &AppState) -> Self {
state.external_api_auth_state.clone()
}
}
server-rs/crates/api-server/src/state/project_metadata.rs
+195
View File
@@ -0,0 +1,195 @@
use std::sync::Arc;
use axum::extract::FromRef;
use futures_util::future::BoxFuture;
use platform_oss::OssClient;
use spacetime_client::{
EditorProjectCreateRecordInput, EditorProjectDeleteRecordInput, EditorProjectGetRecordInput,
EditorProjectLayoutSaveV2AckRecord, EditorProjectLayoutSaveV2RecordInput, EditorProjectRecord,
EditorProjectRenameRecordInput, SpacetimeClient, SpacetimeClientError,
};
use super::AppState;
// 元数据入口只依赖现役 typed facade 的项目能力，不暴露其它领域的数据库操作。
pub(crate) trait EditorProjectRepository: Send + Sync {
fn get_recent_editor_project(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Option<EditorProjectRecord>, SpacetimeClientError>>;
fn list_editor_projects(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Vec<EditorProjectRecord>, SpacetimeClientError>>;
fn get_editor_project(
&self,
input: EditorProjectGetRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>>;
fn create_editor_project(
&self,
input: EditorProjectCreateRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>>;
fn rename_editor_project(
&self,
input: EditorProjectRenameRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>>;
fn delete_editor_project(
&self,
input: EditorProjectDeleteRecordInput,
) -> BoxFuture<'_, Result<String, SpacetimeClientError>>;
fn save_editor_project_layout_v2_ack(
&self,
input: EditorProjectLayoutSaveV2RecordInput,
) -> BoxFuture<'_, Result<EditorProjectLayoutSaveV2AckRecord, SpacetimeClientError>>;
}
impl EditorProjectRepository for SpacetimeClient {
fn get_recent_editor_project(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Option<EditorProjectRecord>, SpacetimeClientError>> {
Box::pin(SpacetimeClient::get_recent_editor_project(
self,
owner_user_id,
))
}
fn list_editor_projects(
&self,
owner_user_id: String,
) -> BoxFuture<'_, Result<Vec<EditorProjectRecord>, SpacetimeClientError>> {
Box::pin(SpacetimeClient::list_editor_projects(self, owner_user_id))
}
fn get_editor_project(
&self,
input: EditorProjectGetRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
Box::pin(SpacetimeClient::get_editor_project(self, input))
}
fn create_editor_project(
&self,
input: EditorProjectCreateRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
Box::pin(SpacetimeClient::create_editor_project(self, input))
}
fn rename_editor_project(
&self,
input: EditorProjectRenameRecordInput,
) -> BoxFuture<'_, Result<EditorProjectRecord, SpacetimeClientError>> {
Box::pin(SpacetimeClient::rename_editor_project(self, input))
}
fn delete_editor_project(
&self,
input: EditorProjectDeleteRecordInput,
) -> BoxFuture<'_, Result<String, SpacetimeClientError>> {
Box::pin(SpacetimeClient::delete_editor_project(self, input))
}
fn save_editor_project_layout_v2_ack(
&self,
input: EditorProjectLayoutSaveV2RecordInput,
) -> BoxFuture<'_, Result<EditorProjectLayoutSaveV2AckRecord, SpacetimeClientError>> {
Box::pin(SpacetimeClient::save_editor_project_layout_v2_ack(
self, input,
))
}
}
pub(crate) trait EditorProjectMediaRepair: Send + Sync {
fn repair_project(&self, project: EditorProjectRecord) -> BoxFuture<'_, EditorProjectRecord>;
}
#[derive(Clone)]
pub struct EditorProjectState {
projects: Arc<dyn EditorProjectRepository>,
media_repair: Arc<dyn EditorProjectMediaRepair>,
}
impl EditorProjectState {
pub(crate) fn new(
projects: Arc<dyn EditorProjectRepository>,
media_repair: Arc<dyn EditorProjectMediaRepair>,
) -> Self {
Self {
projects,
media_repair,
}
}
pub(crate) fn projects(&self) -> &dyn EditorProjectRepository {
self.projects.as_ref()
}
pub(crate) async fn repair_project_media(
&self,
project: EditorProjectRecord,
) -> EditorProjectRecord {
self.media_repair.repair_project(project).await
}
}
impl FromRef<AppState> for EditorProjectState {
fn from_ref(state: &AppState) -> Self {
state.editor_project_state.clone()
}
}
// 读项目时的内联媒体修复与生成写入共用相同存储依赖，保持原有上传和登记顺序。
#[derive(Clone)]
pub(crate) struct EditorMediaStorageState {
spacetime_client: SpacetimeClient,
oss_client: Option<OssClient>,
http_client: reqwest::Client,
}
impl EditorMediaStorageState {
pub(super) fn new(
spacetime_client: SpacetimeClient,
oss_client: Option<OssClient>,
http_client: reqwest::Client,
) -> Self {
Self {
spacetime_client,
oss_client,
http_client,
}
}
pub(crate) fn spacetime_client(&self) -> &SpacetimeClient {
&self.spacetime_client
}
pub(crate) fn oss_client(&self) -> Option<&OssClient> {
self.oss_client.as_ref()
}
pub(crate) fn editor_oss_http_client(&self) -> &reqwest::Client {
&self.http_client
}
}
impl FromRef<AppState> for EditorMediaStorageState {
fn from_ref(state: &AppState) -> Self {
Self::new(
state.spacetime_client().clone(),
state.oss_client().cloned(),
state.editor_oss_http_client().clone(),
)
}
}
impl EditorProjectMediaRepair for EditorMediaStorageState {
fn repair_project(&self, project: EditorProjectRecord) -> BoxFuture<'_, EditorProjectRecord> {
Box::pin(crate::editor_project::repair_editor_project_record_inline_media(self, project))
}
}
server-rs/crates/api-server/src/telemetry.rs
+261 -13
View File
@@ -1,6 +1,6 @@
use axum::{
body::Body,
extract::State,
extract::{MatchedPath, State},
http::{HeaderMap, Request, Response},
middleware::Next,
};
@@ -28,28 +28,42 @@ pub async fn record_http_observability(
State(state): State<AppState>,
request: Request<Body>,
next: Next,
) -> Response<Body> {
observe_http_request(
http_metrics(),
state.config.slow_request_threshold_ms,
request,
next,
)
.await
}
async fn observe_http_request(
metrics: &HttpMetrics,
slow_request_threshold_ms: u64,
request: Request<Body>,
next: Next,
) -> Response<Body> {
let method = request.method().as_str().to_string();
let route = observability_route(request.uri().path());
let route = observability_route(&request);
let scheme = resolve_request_scheme(request.headers());
let path = request.uri().path().to_string();
let request_id = resolve_request_id(&request).unwrap_or_else(|| "unknown".to_string());
let base_labels = http_base_labels(method.clone(), route.clone());
let metrics = http_metrics();
metrics.in_flight.add(1, &base_labels);
let in_flight = RequestInFlightGuard::new(&metrics.in_flight, base_labels.clone());
let started_at = std::time::Instant::now();
let response = next.run(request).await;
let status = response.status().as_u16();
let status_class = status_class(status);
let latency_ms = started_at.elapsed().as_millis().min(u64::MAX as u128) as u64;
let slow_request = latency_ms >= state.config.slow_request_threshold_ms;
let slow_request = latency_ms >= slow_request_threshold_ms;
let labels = http_response_labels(base_labels, status);
metrics.requests.add(1, &labels);
metrics
.duration
.record(started_at.elapsed().as_secs_f64(), &labels);
metrics.in_flight.add(-1, &labels[..2]);
drop(in_flight);
if slow_request {
warn!(
@@ -201,12 +215,32 @@ fn track_response_body_in_flight(response: Response<Body>) -> Response<Body> {
})
}
#[derive(Clone)]
struct HttpMetrics {
requests: Counter<u64>,
in_flight: opentelemetry::metrics::UpDownCounter<i64>,
duration: opentelemetry::metrics::Histogram<f64>,
}
// 请求 Future 被取消或 panic unwind 时也必须释放计数；响应体存活由另一 guard 统计。
struct RequestInFlightGuard<'a> {
counter: &'a opentelemetry::metrics::UpDownCounter<i64>,
labels: Vec<KeyValue>,
}
impl<'a> RequestInFlightGuard<'a> {
fn new(counter: &'a opentelemetry::metrics::UpDownCounter<i64>, labels: Vec<KeyValue>) -> Self {
counter.add(1, &labels);
Self { counter, labels }
}
}
impl Drop for RequestInFlightGuard<'_> {
fn drop(&mut self) {
self.counter.add(-1, &self.labels);
}
}
#[cfg(any())]
struct PuzzleGalleryCacheMetrics {
hits: Counter<u64>,
@@ -467,7 +501,11 @@ fn status_class(status: u16) -> &'static str {
}
}
pub(crate) fn observability_route(path: &str) -> String {
pub(crate) fn observability_route<B>(request: &Request<B>) -> String {
if let Some(path) = request.extensions().get::<MatchedPath>() {
return path.as_str().to_string();
}
let path = request.uri().path();
if path.starts_with("/admin/api/") {
"/admin/api/*".to_string()
} else if path.starts_with("/api/") {
@@ -490,22 +528,232 @@ pub(crate) fn resolve_request_scheme(headers: &HeaderMap) -> String {
#[cfg(test)]
mod tests {
use axum::http::{HeaderMap, HeaderValue};
use axum::{
Router,
body::{Body, Bytes},
http::{HeaderMap, HeaderValue, Request, StatusCode},
middleware,
routing::get,
};
use http_body_util::BodyExt;
use opentelemetry::{
KeyValue,
metrics::{Counter, Histogram, SyncInstrument, UpDownCounter},
};
use std::{
convert::Infallible,
sync::{Arc, Mutex},
time::Duration,
};
use tokio::sync::Notify;
use tower::ServiceExt;
use super::{observability_route, resolve_request_scheme};
use super::{HttpMetrics, observability_route, observe_http_request, resolve_request_scheme};
#[derive(Default)]
struct Measurements<T> {
values: Mutex<Vec<(T, Vec<KeyValue>)>>,
}
impl<T: Send> SyncInstrument<T> for Measurements<T> {
fn measure(&self, value: T, attributes: &[KeyValue]) {
self.values
.lock()
.expect("measurement lock")
.push((value, attributes.to_vec()));
}
}
fn observed_router(router: Router) -> (Router, Arc<Measurements<i64>>) {
let in_flight = Arc::new(Measurements::default());
let metrics = HttpMetrics {
requests: Counter::new(Arc::new(Measurements::<u64>::default())),
in_flight: UpDownCounter::new(in_flight.clone()),
duration: Histogram::new(Arc::new(Measurements::<f64>::default())),
};
let router = router.layer(middleware::from_fn(move |request, next| {
let metrics = metrics.clone();
async move { observe_http_request(&metrics, u64::MAX, request, next).await }
}));
(router, in_flight)
}
fn request(uri: &str) -> Request<Body> {
Request::builder()
.uri(uri)
.body(Body::empty())
.expect("request")
}
fn assert_request_finished(measurements: &Measurements<i64>) {
let values = measurements.values.lock().expect("measurement lock");
assert_eq!(
values.iter().map(|(value, _)| *value).collect::<Vec<_>>(),
vec![1, -1]
);
assert_eq!(
values[0].1, values[1].1,
"decrement must use the original labels"
);
}
#[tokio::test]
async fn cancelled_request_releases_in_flight_measurement() {
let entered = Arc::new(Notify::new());
let handler_entered = entered.clone();
let (router, measurements) = observed_router(Router::new().route(
"/api/pending/{id}",
get(move || {
let entered = handler_entered.clone();
async move {
entered.notify_one();
std::future::pending::<StatusCode>().await
}
}),
));
let task = tokio::spawn(router.oneshot(request("/api/pending/123")));
tokio::time::timeout(Duration::from_secs(5), entered.notified())
.await
.expect("handler entered");
assert_eq!(
measurements.values.lock().expect("measurement lock")[0].0,
1
);
task.abort();
assert!(
task.await
.expect_err("request should be cancelled")
.is_cancelled()
);
assert_request_finished(&measurements);
}
#[tokio::test]
async fn panicking_request_releases_in_flight_measurement() {
async fn panic_handler() -> StatusCode {
panic!("handler panic for cancellation cleanup test");
}
let (router, measurements) =
observed_router(Router::new().route("/panic", get(panic_handler)));
let task = tokio::spawn(router.oneshot(request("/panic")));
assert!(task.await.expect_err("handler should panic").is_panic());
assert_request_finished(&measurements);
}
#[tokio::test]
async fn responses_release_in_flight_once_for_success_and_errors() {
for status in [
StatusCode::OK,
StatusCode::UNAUTHORIZED,
StatusCode::INTERNAL_SERVER_ERROR,
] {
let (router, measurements) = observed_router(
Router::new().route("/response", get(move || async move { status })),
);
let response = router
.oneshot(request("/response"))
.await
.expect("response");
assert_eq!(response.status(), status);
assert_request_finished(&measurements);
drop(response);
assert_request_finished(&measurements);
}
}
#[tokio::test]
async fn streaming_response_releases_request_before_body_completion() {
use futures_util::StreamExt;
let (router, measurements) = observed_router(Router::new().route(
"/events",
get(|| async {
let stream = futures_util::stream::iter([Ok::<_, Infallible>(Bytes::from_static(
b"data: ready\n\n",
))])
.chain(futures_util::stream::pending());
(
[("content-type", "text/event-stream")],
Body::from_stream(stream),
)
}),
));
let mut response = router
.oneshot(request("/events"))
.await
.expect("streaming response");
assert_request_finished(&measurements);
let frame = response
.body_mut()
.frame()
.await
.expect("first frame")
.expect("body frame");
assert_eq!(frame.into_data().expect("data frame"), "data: ready\n\n");
drop(response);
assert_request_finished(&measurements);
}
#[tokio::test]
async fn matched_routes_share_templates_and_preserve_distinct_endpoints() {
let routes = Router::new().nest(
"/api",
Router::new()
.route("/projects/{project_id}", get(|| async {}))
.route("/assets/{asset_id}", get(|| async {})),
);
let (router, measurements) = observed_router(routes);
for (uri, template) in [
(
"/api/projects/project-1?cursor=private",
"/api/projects/{project_id}",
),
("/api/projects/project-2", "/api/projects/{project_id}"),
("/api/assets/asset-1", "/api/assets/{asset_id}"),
("/api/missing/private-id?token=private", "/api/*"),
] {
let response = router
.clone()
.oneshot(request(uri))
.await
.expect("response");
drop(response);
assert_request_finished(&measurements);
let mut values = measurements.values.lock().expect("measurement lock");
assert_eq!(
values[0].1,
vec![
KeyValue::new("http.request.method", "GET"),
KeyValue::new("http.route", template)
]
);
values.clear();
}
}
#[test]
fn observability_route_keeps_metrics_labels_low_cardinality() {
assert_eq!(
observability_route("/api/editor/showcase/resources?cursor=abc"),
observability_route(&request("/api/editor/showcase/resources?cursor=abc")),
"/api/*"
);
assert_eq!(
observability_route("/api/editor/projects/project-1"),
observability_route(&request("/api/editor/projects/project-1")),
"/api/*"
);
assert_eq!(observability_route("/api/runtime/settings"), "/api/*");
assert_eq!(observability_route("/admin/api/debug/http"), "/admin/api/*");
assert_eq!(
observability_route(&request("/api/runtime/settings")),
"/api/*"
);
assert_eq!(
observability_route(&request("/admin/api/debug/http")),
"/admin/api/*"
);
assert_eq!(
observability_route(&request("/missing/private-id")),
"other"
);
}
#[test]