feat(observability): add structured application logging

Emit correlated JSON logs from the API and FunASR services, propagate request IDs to the client, and configure bounded opt-in Docker logging.

Document the host-wide Grafana/Loki query workflow and keep observability deployment ownership outside the application repository.

server/src/config.rs

@@ -1,5 +1,11 @@
 use std::{env, net::IpAddr, path::PathBuf};
 
+#[derive(Clone, Copy, Eq, PartialEq)]
+pub enum LogFormat {
+    Json,
+    Pretty,
+}
+
 #[derive(Clone)]
 pub enum SpeechProviderConfig {
     Local { api_url: String },
@@ -50,6 +56,10 @@ pub struct Config {
     pub database_url: Option<String>,
     pub database_max_connections: u32,
     pub cors_allowed_origin: Option<String>,
+    pub log_format: LogFormat,
+    pub app_env: String,
+    pub app_version: String,
+    pub git_sha: String,
     pub audio_storage_dir: PathBuf,
     pub speech: Option<SpeechConfig>,
     pub summary: Option<SummaryConfig>,
@@ -70,6 +80,11 @@ impl Config {
             .unwrap_or_else(|_| "5".to_owned())
             .parse()
             .map_err(|_| "DATABASE_MAX_CONNECTIONS must be a valid u32 value".to_owned())?;
+        let log_format = match optional_env("LOG_FORMAT").as_deref().unwrap_or("pretty") {
+            "json" => LogFormat::Json,
+            "pretty" => LogFormat::Pretty,
+            _ => return Err("LOG_FORMAT must be json or pretty".to_owned()),
+        };
 
         let speech = speech_config()?;
         let summary = summary_config()?;
@@ -85,6 +100,11 @@ impl Config {
             cors_allowed_origin: env::var("CORS_ALLOWED_ORIGIN")
                 .ok()
                 .filter(|value| !value.trim().is_empty()),
+            log_format,
+            app_env: optional_env("APP_ENV").unwrap_or_else(|| "development".to_owned()),
+            app_version: optional_env("APP_VERSION")
+                .unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_owned()),
+            git_sha: optional_env("GIT_SHA").unwrap_or_else(|| "unknown".to_owned()),
             audio_storage_dir: env::var("AUDIO_STORAGE_DIR")
                 .map(PathBuf::from)
                 .unwrap_or_else(|_| PathBuf::from("./data/audio")),

server/src/error.rs

@@ -92,8 +92,12 @@ impl IntoResponse for ApiError {
 }
 
 impl From<sqlx::Error> for ApiError {
-    fn from(error: sqlx::Error) -> Self {
-        tracing::error!(?error, "database query failed");
+    fn from(_: sqlx::Error) -> Self {
+        tracing::error!(
+            event = "database_query_failed",
+            error_class = "database_query_failed",
+            "database query failed"
+        );
         Self::Internal
     }
 }

server/src/main.rs

@@ -7,12 +7,25 @@ mod speech;
 mod summary;
 mod transcription_worker;
 
-use std::{path::PathBuf, sync::Arc};
+use std::{path::PathBuf, sync::Arc, time::Duration};
 
-use axum::{Json, Router, extract::DefaultBodyLimit, http::HeaderValue, routing::get};
-use config::Config;
+use axum::{
+    Json, Router,
+    extract::{DefaultBodyLimit, MatchedPath},
+    http::{HeaderName, HeaderValue, Request},
+    response::Response,
+    routing::get,
+};
+use config::{Config, LogFormat};
 use sqlx::{PgPool, postgres::PgPoolOptions};
-use tower_http::{cors::CorsLayer, trace::TraceLayer};
+use tower::ServiceBuilder;
+use tower_http::{
+    classify::ServerErrorsFailureClass,
+    cors::CorsLayer,
+    request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer},
+    trace::TraceLayer,
+};
+use tracing::Span;
 use tracing_subscriber::EnvFilter;
 use utoipa_scalar::{Scalar, Servable};
 
@@ -25,15 +38,30 @@ pub struct AppState {
     pub auth: auth::AuthService,
 }
 
+#[derive(Clone)]
+pub(crate) struct LogContext {
+    pub(crate) environment: String,
+    pub(crate) app_version: String,
+    pub(crate) git_sha: String,
+}
+
+impl From<&Config> for LogContext {
+    fn from(config: &Config) -> Self {
+        Self {
+            environment: config.app_env.clone(),
+            app_version: config.app_version.clone(),
+            git_sha: config.git_sha.clone(),
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     dotenvy::dotenv().ok();
-    tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()))
-        .init();
-
     let config = Config::from_env().map_err(std::io::Error::other)?;
-    let pool = connect_database(&config).await?;
+    init_tracing(&config);
+    let log_context = LogContext::from(&config);
+    let pool = connect_database(&config, &log_context).await?;
     let speech = speech::SpeechService::new(config.speech.clone());
     let auth = auth::AuthService::new(config.auth.clone());
     let state = AppState {
@@ -45,21 +73,57 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     };
     if let Some(pool) = state.pool.clone() {
         if speech.configured() {
-            transcription_worker::spawn(pool, speech);
+            transcription_worker::spawn(pool, speech.clone(), log_context.clone());
         }
     }
-    let app = build_app(state, config.cors_allowed_origin.as_deref())?;
+    let app = build_app(
+        state,
+        config.cors_allowed_origin.as_deref(),
+        log_context.clone(),
+    )?;
     let address = std::net::SocketAddr::from((config.host, config.port));
     let listener = tokio::net::TcpListener::bind(address).await?;
 
-    tracing::info!(%address, scalar = %format!("http://{address}/scalar"), "API server is listening");
+    tracing::info!(
+        service = "api",
+        environment = %log_context.environment,
+        app_version = %log_context.app_version,
+        git_sha = %log_context.git_sha,
+        event = "api_started",
+        %address,
+        scalar = %format!("http://{address}/scalar"),
+        provider = ?speech.provider_name(),
+        "API server is listening"
+    );
     axum::serve(listener, app)
-        .with_graceful_shutdown(shutdown_signal())
+        .with_graceful_shutdown(shutdown_signal(log_context))
         .await?;
     Ok(())
 }
 
-pub fn build_app(state: AppState, cors_allowed_origin: Option<&str>) -> Result<Router, String> {
+fn init_tracing(config: &Config) {
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into());
+    match config.log_format {
+        LogFormat::Json => tracing_subscriber::fmt()
+            .json()
+            .flatten_event(true)
+            .with_ansi(false)
+            .with_current_span(true)
+            .with_span_list(true)
+            .with_env_filter(filter)
+            .init(),
+        LogFormat::Pretty => tracing_subscriber::fmt()
+            .with_env_filter(filter)
+            .pretty()
+            .init(),
+    }
+}
+
+fn build_app(
+    state: AppState,
+    cors_allowed_origin: Option<&str>,
+    log_context: LogContext,
+) -> Result<Router, String> {
     let (api_router, openapi) = routes::router();
     let openapi_json = openapi.clone();
     let mut app = api_router
@@ -72,21 +136,142 @@ pub fn build_app(state: AppState, cors_allowed_origin: Option<&str>) -> Result<R
         )
         .merge(Scalar::with_url("/scalar", openapi))
         .with_state(Arc::new(state))
-        .layer(DefaultBodyLimit::max(20 * 1024 * 1024))
-        .layer(TraceLayer::new_for_http());
+        .layer(DefaultBodyLimit::max(20 * 1024 * 1024));
 
     if let Some(origin) = cors_allowed_origin {
         let allowed_origin = HeaderValue::from_str(origin)
             .map_err(|_| "CORS_ALLOWED_ORIGIN must be a valid HTTP header value".to_owned())?;
-        app = app.layer(CorsLayer::new().allow_origin(allowed_origin));
+        app = app.layer(
+            CorsLayer::new()
+                .allow_origin(allowed_origin)
+                .expose_headers([HeaderName::from_static("x-request-id")]),
+        );
     }
 
-    Ok(app)
+    let request_span_context = log_context.clone();
+    let request_started_context = log_context.clone();
+    let request_finished_context = log_context.clone();
+    let request_failed_context = log_context;
+    Ok(app.layer(
+        ServiceBuilder::new()
+            .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
+            .layer(
+                TraceLayer::new_for_http()
+                    .make_span_with(move |request: &Request<_>| {
+                        make_http_request_span(request, &request_span_context)
+                    })
+                    .on_request(move |request: &Request<_>, span: &Span| {
+                        tracing::info!(
+                            parent: span,
+                            service = "api",
+                            environment = %request_started_context.environment,
+                            app_version = %request_started_context.app_version,
+                            git_sha = %request_started_context.git_sha,
+                            event = "http_request_started",
+                            request_id = request_id_from_headers(request),
+                            "HTTP request started"
+                        );
+                    })
+                    .on_response(move |response: &Response, latency: Duration, span: &Span| {
+                        let status = response.status();
+                        let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
+                        let request_id = response
+                            .headers()
+                            .get("x-request-id")
+                            .and_then(|value| value.to_str().ok())
+                            .unwrap_or("unknown");
+                        if status.is_server_error() {
+                            tracing::error!(
+                                parent: span,
+                                service = "api",
+                                environment = %request_finished_context.environment,
+                                app_version = %request_finished_context.app_version,
+                                git_sha = %request_finished_context.git_sha,
+                                event = "http_request_finished",
+                                status = status.as_u16(),
+                                latency_ms,
+                                request_id,
+                                error_class = "server_error",
+                                "HTTP request finished"
+                            );
+                        } else {
+                            tracing::info!(
+                                parent: span,
+                                service = "api",
+                                environment = %request_finished_context.environment,
+                                app_version = %request_finished_context.app_version,
+                                git_sha = %request_finished_context.git_sha,
+                                event = "http_request_finished",
+                                status = status.as_u16(),
+                                latency_ms,
+                                request_id,
+                                "HTTP request finished"
+                            );
+                        }
+                    })
+                    .on_failure(
+                        move |failure: ServerErrorsFailureClass, latency: Duration, span: &Span| {
+                            let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
+                            tracing::error!(
+                                parent: span,
+                                service = "api",
+                                environment = %request_failed_context.environment,
+                                app_version = %request_failed_context.app_version,
+                                git_sha = %request_failed_context.git_sha,
+                                event = "http_request_failed",
+                                latency_ms,
+                                error_class = ?failure,
+                                "HTTP request failed"
+                            );
+                        },
+                    ),
+            )
+            .layer(PropagateRequestIdLayer::x_request_id()),
+    ))
 }
 
-async fn connect_database(config: &Config) -> Result<Option<PgPool>, sqlx::Error> {
+fn make_http_request_span<B>(request: &Request<B>, context: &LogContext) -> Span {
+    let route = request
+        .extensions()
+        .get::<MatchedPath>()
+        .map(MatchedPath::as_str)
+        .unwrap_or_else(|| request.uri().path());
+    let request_id = request_id_from_headers(request);
+
+    tracing::info_span!(
+        "http_request",
+        service = "api",
+        environment = %context.environment,
+        app_version = %context.app_version,
+        git_sha = %context.git_sha,
+        method = %request.method(),
+        route,
+        path = %request.uri().path(),
+        request_id,
+    )
+}
+
+fn request_id_from_headers<B>(request: &Request<B>) -> &str {
+    request
+        .headers()
+        .get("x-request-id")
+        .and_then(|value| value.to_str().ok())
+        .unwrap_or("invalid")
+}
+
+async fn connect_database(
+    config: &Config,
+    log_context: &LogContext,
+) -> Result<Option<PgPool>, sqlx::Error> {
     let Some(database_url) = &config.database_url else {
-        tracing::warn!("DATABASE_URL is not set; data endpoints will return HTTP 503");
+        tracing::warn!(
+            service = "api",
+            environment = %log_context.environment,
+            app_version = %log_context.app_version,
+            git_sha = %log_context.git_sha,
+            event = "database_not_configured",
+            "DATABASE_URL is not set; data endpoints will return HTTP 503"
+        );
         return Ok(None);
     };
 
@@ -94,11 +279,95 @@ async fn connect_database(config: &Config) -> Result<Option<PgPool>, sqlx::Error
         .max_connections(config.database_max_connections)
         .connect(database_url)
         .await?;
-    tracing::info!("connected to PostgreSQL");
+    tracing::info!(
+        service = "api",
+        environment = %log_context.environment,
+        app_version = %log_context.app_version,
+        git_sha = %log_context.git_sha,
+        event = "database_connected",
+        "connected to PostgreSQL"
+    );
     Ok(Some(pool))
 }
 
-async fn shutdown_signal() {
+async fn shutdown_signal(log_context: LogContext) {
     let _ = tokio::signal::ctrl_c().await;
-    tracing::info!("shutdown signal received");
+    tracing::info!(
+        service = "api",
+        environment = %log_context.environment,
+        app_version = %log_context.app_version,
+        git_sha = %log_context.git_sha,
+        event = "api_shutdown",
+        "shutdown signal received"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use axum::{body::Body, http::Request};
+    use tower::ServiceExt;
+    use uuid::Uuid;
+
+    use super::{AppState, LogContext, build_app};
+    use crate::{
+        auth::AuthService, config::AuthConfig, speech::SpeechService, summary::SummaryService,
+    };
+
+    fn app() -> axum::Router {
+        build_app(
+            AppState {
+                pool: None,
+                audio_storage_dir: std::path::PathBuf::from("./data/audio"),
+                speech: SpeechService::new(None),
+                summary: SummaryService::new(None),
+                auth: AuthService::new(AuthConfig {
+                    wechat: None,
+                    session_ttl_days: 30,
+                    allow_development_user_header: false,
+                }),
+            },
+            None,
+            LogContext {
+                environment: "test".to_owned(),
+                app_version: "test".to_owned(),
+                git_sha: "test".to_owned(),
+            },
+        )
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn health_generates_a_uuid_request_id() {
+        let response = app()
+            .oneshot(
+                Request::builder()
+                    .uri("/health")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        let request_id = response
+            .headers()
+            .get("x-request-id")
+            .unwrap()
+            .to_str()
+            .unwrap();
+        assert!(Uuid::parse_str(request_id).is_ok());
+    }
+
+    #[tokio::test]
+    async fn health_propagates_a_client_request_id() {
+        let response = app()
+            .oneshot(
+                Request::builder()
+                    .uri("/health")
+                    .header("x-request-id", "test-request-id")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.headers()["x-request-id"], "test-request-id");
+    }
 }

server/src/recording_routes.rs

@@ -970,8 +970,12 @@ async fn touch_session(pool: &PgPool, session_id: Uuid) -> Result<(), ApiError>
     Ok(())
 }
 
-fn internal_io_error(error: std::io::Error) -> ApiError {
-    tracing::error!(?error, "audio file operation failed");
+fn internal_io_error(_: std::io::Error) -> ApiError {
+    tracing::error!(
+        event = "audio_storage_io_failed",
+        error_class = "audio_storage_io_failed",
+        "audio file operation failed"
+    );
     ApiError::Internal
 }
 

server/src/speech.rs

@@ -107,6 +107,7 @@ impl SpeechService {
         audio: Vec<u8>,
         audio_format: &str,
         duration_ms: i32,
+        request_id: &str,
     ) -> Result<Transcription, String> {
         let config = self
             .config
@@ -114,7 +115,7 @@ impl SpeechService {
             .ok_or_else(|| "语音转录服务尚未配置".to_owned())?;
         match &config.provider {
             SpeechProviderConfig::Local { api_url } => {
-                self.transcribe_local(api_url, audio, audio_format, duration_ms)
+                self.transcribe_local(api_url, audio, audio_format, duration_ms, request_id)
                     .await
             }
             SpeechProviderConfig::Tencent(config) => {
@@ -129,6 +130,7 @@ impl SpeechService {
         audio: Vec<u8>,
         audio_format: &str,
         duration_ms: i32,
+        request_id: &str,
     ) -> Result<Transcription, String> {
         let part = Part::bytes(audio)
             .file_name(format!("segment.{audio_format}"))
@@ -141,6 +143,7 @@ impl SpeechService {
         let response = self
             .client
             .post(local_transcription_url(api_url))
+            .header("X-Request-ID", request_id)
             .multipart(form)
             .send()
             .await
@@ -151,10 +154,7 @@ impl SpeechService {
             .await
             .map_err(|error| format!("读取本地语音转录响应失败：{error}"))?;
         if !status.is_success() {
-            return Err(format!(
-                "本地语音转录服务返回 HTTP {status}：{}",
-                compact_error(&body)
-            ));
+            return Err(format!("本地语音转录服务返回 HTTP {status}"));
         }
         let result: LocalResponse = serde_json::from_str(&body)
             .map_err(|error| format!("本地语音转录响应格式无效：{error}"))?;
@@ -239,11 +239,6 @@ fn audio_content_type(audio_format: &str) -> &'static str {
     }
 }
 
-fn compact_error(body: &str) -> String {
-    let value = body.split_whitespace().collect::<Vec<_>>().join(" ");
-    value.chars().take(240).collect()
-}
-
 fn signed_request(
     config: &TencentSpeechConfig,
     audio_format: &str,
@@ -279,6 +274,7 @@ fn signed_request(
 mod tests {
     use axum::{
         Json, Router,
+        http::HeaderMap,
         routing::{get, post},
     };
     use serde_json::json;
@@ -318,7 +314,8 @@ mod tests {
             .route("/health", get(|| async { Json(json!({"status": "ok"})) }))
             .route(
                 "/v1/transcriptions",
-                post(|| async {
+                post(|headers: HeaderMap| async move {
+                    assert_eq!(headers["x-request-id"], "segment-request-id");
                     Json(json!({
                         "text": "课堂练习完成。",
                         "duration_ms": 12_345,
@@ -340,7 +337,7 @@ mod tests {
 
         assert!(service.available().await);
         let result = service
-            .transcribe(vec![1, 2, 3], "mp3", 10_000)
+            .transcribe(vec![1, 2, 3], "mp3", 10_000, "segment-request-id")
             .await
             .unwrap();
         assert_eq!(result.text, "课堂练习完成。");

server/src/transcription_worker.rs

@@ -1,9 +1,14 @@
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
+use chrono::{DateTime, Utc};
 use sqlx::{FromRow, PgPool};
+use tracing::Instrument;
 use uuid::Uuid;
 
-use crate::speech::{SpeechService, Transcription};
+use crate::{
+    LogContext,
+    speech::{SpeechService, Transcription},
+};
 
 const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(750);
 const UNAVAILABLE_POLL_INTERVAL: Duration = Duration::from_secs(5);
@@ -12,18 +17,29 @@ const UNAVAILABLE_POLL_INTERVAL: Duration = Duration::from_secs(5);
 struct TranscriptionJob {
     id: Uuid,
     lesson_session_id: Uuid,
+    feedback_session_id: Uuid,
     duration_ms: i32,
     audio_format: String,
     audio_path: String,
+    created_at: DateTime<Utc>,
+    processing_started_at: DateTime<Utc>,
+    transcription_attempts: i32,
 }
 
-pub fn spawn(pool: PgPool, speech: SpeechService) {
+pub fn spawn(pool: PgPool, speech: SpeechService, log_context: LogContext) {
     for worker_id in 1..=speech.worker_concurrency() {
         let pool = pool.clone();
         let speech = speech.clone();
-        tokio::spawn(async move {
-            run(worker_id, pool, speech).await;
-        });
+        let log_context = log_context.clone();
+        let span = tracing::info_span!(
+            "transcription_worker",
+            service = "api",
+            environment = %log_context.environment,
+            app_version = %log_context.app_version,
+            git_sha = %log_context.git_sha,
+            worker_id,
+        );
+        tokio::spawn(async move { run(worker_id, pool, speech).await }.instrument(span));
     }
 }
 
@@ -36,8 +52,13 @@ async fn run(worker_id: usize, pool: PgPool, speech: SpeechService) {
                 continue;
             }
             Ok(true) => {}
-            Err(error) => {
-                tracing::error!(worker_id, error = %error, "failed to inspect transcription queue");
+            Err(_) => {
+                tracing::error!(
+                    event = "transcription_queue_inspection_failed",
+                    worker_id,
+                    error_class = "database_query_failed",
+                    "failed to inspect transcription queue"
+                );
                 tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await;
                 continue;
             }
@@ -45,28 +66,49 @@ async fn run(worker_id: usize, pool: PgPool, speech: SpeechService) {
         let available = speech.available().await;
         if !available {
             if service_available {
-                tracing::warn!(worker_id, provider = ?speech.provider_name(), "ASR service is unavailable; queued audio will wait");
+                tracing::warn!(
+                    event = "asr_service_unavailable",
+                    worker_id,
+                    provider = speech.provider_name().unwrap_or("unknown"),
+                    "ASR service is unavailable; queued audio will wait"
+                );
             }
             service_available = false;
             tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await;
             continue;
         }
         if !service_available {
-            tracing::info!(worker_id, provider = ?speech.provider_name(), "ASR service is available again");
+            tracing::info!(
+                event = "asr_service_recovered",
+                worker_id,
+                provider = speech.provider_name().unwrap_or("unknown"),
+                "ASR service is available again"
+            );
         }
         service_available = true;
 
         match claim_next_job(&pool, speech.job_lease_seconds()).await {
             Ok(Some(job)) => {
-                tracing::info!(worker_id, segment_id = %job.id, "transcribing audio segment");
-                if let Err(error) = process_job(&pool, &speech, &job).await {
-                    tracing::error!(worker_id, segment_id = %job.id, error = %error, "transcription job update failed");
+                if process_job(&pool, &speech, worker_id, &job).await.is_err() {
+                    tracing::error!(
+                        event = "transcription_job_update_failed",
+                        worker_id,
+                        segment_id = %job.id,
+                        feedback_session_id = %job.feedback_session_id,
+                        error_class = "database_update_failed",
+                        "transcription job update failed"
+                    );
                     tokio::time::sleep(IDLE_POLL_INTERVAL).await;
                 }
             }
             Ok(None) => tokio::time::sleep(IDLE_POLL_INTERVAL).await,
-            Err(error) => {
-                tracing::error!(worker_id, error = %error, "failed to claim transcription job");
+            Err(_) => {
+                tracing::error!(
+                    event = "transcription_job_claim_failed",
+                    worker_id,
+                    error_class = "database_query_failed",
+                    "failed to claim transcription job"
+                );
                 tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await;
             }
         }
@@ -104,10 +146,13 @@ async fn claim_next_job(
          SET processing_started_at = now(), \
              transcription_attempts = transcription_attempts + 1, \
              updated_at = now() \
-         FROM next_job \
+         FROM next_job, lesson_sessions AS lesson \
          WHERE segment.id = next_job.id \
-         RETURNING segment.id, segment.lesson_session_id, segment.duration_ms, \
-                   segment.audio_format, segment.audio_path",
+           AND lesson.id = segment.lesson_session_id \
+         RETURNING segment.id, segment.lesson_session_id, lesson.feedback_session_id, \
+                   segment.duration_ms, segment.audio_format, segment.audio_path, \
+                   segment.created_at, segment.processing_started_at, \
+                   segment.transcription_attempts",
     )
     .bind(lease_seconds)
     .fetch_optional(pool)
@@ -117,29 +162,96 @@ async fn claim_next_job(
 async fn process_job(
     pool: &PgPool,
     speech: &SpeechService,
+    worker_id: usize,
     job: &TranscriptionJob,
 ) -> Result<(), sqlx::Error> {
-    let result = match tokio::fs::read(&job.audio_path).await {
-        Ok(audio) => {
-            speech
-                .transcribe(audio, &job.audio_format, job.duration_ms)
-                .await
+    let queue_wait_ms = elapsed_since(job.created_at);
+    tracing::info!(
+        event = "transcription_job_started",
+        worker_id,
+        segment_id = %job.id,
+        feedback_session_id = %job.feedback_session_id,
+        lesson_id = %job.lesson_session_id,
+        attempt = job.transcription_attempts,
+        queue_wait_ms,
+        audio_duration_ms = job.duration_ms,
+        audio_format = %job.audio_format,
+        status = "processing",
+        "transcribing audio segment"
+    );
+
+    let audio_read_started = Instant::now();
+    let audio = match tokio::fs::read(&job.audio_path).await {
+        Ok(audio) => audio,
+        Err(_) => {
+            fail_job(
+                pool,
+                job,
+                worker_id,
+                queue_wait_ms,
+                elapsed_ms(audio_read_started),
+                0,
+                0,
+                "audio_read_failed",
+                "unable to read audio for transcription",
+            )
+            .await?;
+            refresh_lesson(pool, job.lesson_session_id).await?;
+            return Ok(());
         }
-        Err(error) => Err(format!("读取录音文件失败：{error}")),
     };
+    let audio_read_latency_ms = elapsed_ms(audio_read_started);
+    let audio_size_bytes = u64::try_from(audio.len()).unwrap_or(u64::MAX);
+    let asr_request_id = job.id.to_string();
+    let inference_started = Instant::now();
+    let result = speech
+        .transcribe(audio, &job.audio_format, job.duration_ms, &asr_request_id)
+        .await;
+    let inference_latency_ms = elapsed_ms(inference_started);
 
     match result {
-        Ok(transcription) => complete_job(pool, job, transcription).await?,
-        Err(message) => fail_job(pool, job, &message).await?,
+        Ok(transcription) => {
+            complete_job(
+                pool,
+                job,
+                worker_id,
+                transcription,
+                queue_wait_ms,
+                audio_read_latency_ms,
+                inference_latency_ms,
+                audio_size_bytes,
+            )
+            .await?
+        }
+        Err(message) => {
+            fail_job(
+                pool,
+                job,
+                worker_id,
+                queue_wait_ms,
+                audio_read_latency_ms,
+                inference_latency_ms,
+                audio_size_bytes,
+                transcription_error_class(&message),
+                &message,
+            )
+            .await?
+        }
     }
     refresh_lesson(pool, job.lesson_session_id).await?;
     Ok(())
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn complete_job(
     pool: &PgPool,
     job: &TranscriptionJob,
+    worker_id: usize,
     transcription: Transcription,
+    queue_wait_ms: i64,
+    audio_read_latency_ms: u64,
+    inference_latency_ms: u64,
+    audio_size_bytes: u64,
 ) -> Result<(), sqlx::Error> {
     let duration_ms = if transcription.duration_ms > 0 {
         transcription.duration_ms
@@ -158,12 +270,40 @@ async fn complete_job(
     .bind(&transcription.request_id)
     .execute(pool)
     .await?;
-    tracing::info!(segment_id = %job.id, "audio segment transcription is ready");
+    tracing::info!(
+        event = "transcription_job_finished",
+        worker_id,
+        segment_id = %job.id,
+        feedback_session_id = %job.feedback_session_id,
+        lesson_id = %job.lesson_session_id,
+        asr_request_id = %transcription.request_id,
+        attempt = job.transcription_attempts,
+        queue_wait_ms,
+        audio_read_latency_ms,
+        inference_latency_ms,
+        processing_elapsed_ms = elapsed_since(job.processing_started_at),
+        audio_size_bytes,
+        audio_duration_ms = duration_ms,
+        audio_format = %job.audio_format,
+        provider = "local-funasr",
+        status = "ready",
+        "audio segment transcription is ready"
+    );
     Ok(())
 }
 
-async fn fail_job(pool: &PgPool, job: &TranscriptionJob, message: &str) -> Result<(), sqlx::Error> {
-    tracing::warn!(segment_id = %job.id, error = %message, "audio transcription failed");
+#[allow(clippy::too_many_arguments)]
+async fn fail_job(
+    pool: &PgPool,
+    job: &TranscriptionJob,
+    worker_id: usize,
+    queue_wait_ms: i64,
+    audio_read_latency_ms: u64,
+    inference_latency_ms: u64,
+    audio_size_bytes: u64,
+    error_class: &str,
+    message: &str,
+) -> Result<(), sqlx::Error> {
     sqlx::query(
         "UPDATE audio_segments \
          SET status = 'failed', error_message = $2, processing_started_at = NULL, updated_at = now() \
@@ -173,9 +313,48 @@ async fn fail_job(pool: &PgPool, job: &TranscriptionJob, message: &str) -> Resul
     .bind(message)
     .execute(pool)
     .await?;
+    tracing::warn!(
+        event = "transcription_job_failed",
+        worker_id,
+        segment_id = %job.id,
+        feedback_session_id = %job.feedback_session_id,
+        lesson_id = %job.lesson_session_id,
+        asr_request_id = %job.id,
+        attempt = job.transcription_attempts,
+        queue_wait_ms,
+        audio_read_latency_ms,
+        inference_latency_ms,
+        processing_elapsed_ms = elapsed_since(job.processing_started_at),
+        audio_size_bytes,
+        audio_duration_ms = job.duration_ms,
+        audio_format = %job.audio_format,
+        status = "failed",
+        error_class,
+        "audio transcription failed"
+    );
     Ok(())
 }
 
+fn transcription_error_class(message: &str) -> &'static str {
+    if message.contains("连接失败") || message.contains("unavailable") {
+        "asr_unavailable"
+    } else if message.contains("HTTP") {
+        "asr_http_error"
+    } else if message.contains("读取录音") {
+        "audio_read_failed"
+    } else {
+        "transcription_failed"
+    }
+}
+
+fn elapsed_ms(start: Instant) -> u64 {
+    u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX)
+}
+
+fn elapsed_since(start: DateTime<Utc>) -> i64 {
+    (Utc::now() - start).num_milliseconds().max(0)
+}
+
 async fn refresh_lesson(pool: &PgPool, lesson_id: Uuid) -> Result<(), sqlx::Error> {
     sqlx::query(
         "UPDATE lesson_sessions AS lesson \