исправить: закрыть ревью сквозной корреляции

apps/admin-api/tests/dc08.rs

@@ -0,0 +1,208 @@
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
+use std::{io, sync::Mutex};
+
+use axum::{Json, Router, extract::State, routing::post};
+use crank_core::{OperationId, WorkspaceId};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::{net::TcpListener, sync::Notify};
+use tracing_subscriber::fmt::MakeWriter;
+
+#[path = "integration/common.rs"]
+mod common;
+
+use common::*;
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn preserves_external_success_when_invocation_history_is_lost() {
+    let registry = test_registry().await;
+    let registry_for_failure = registry.clone();
+    let storage_root = test_storage_root("observability_history_loss");
+    let upstream = spawn_blocking_upstream_server().await;
+    let log_writer = SharedLogWriter::default();
+    let subscriber = crank_observability::build_subscriber(
+        crank_observability::ObservabilityConfig::new(
+            crank_observability::ServiceIdentity::try_new("admin-api", "test", "test").unwrap(),
+            "info",
+            crank_observability::RedactionLimits::default(),
+        ),
+        log_writer.clone(),
+    )
+    .unwrap();
+    tracing::subscriber::set_global_default(subscriber).unwrap();
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream.base_url,
+            "crm_history_loss",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+    let request_client = client.clone();
+    let request_url = format!("{base_url}/operations/{operation_id}/test-runs");
+    let before = crank_observability::operational_incident_total(
+        crank_observability::OperationalIncident::InvocationHistoryLost,
+    );
+
+    let request = tokio::spawn(async move {
+        request_client
+            .post(request_url)
+            .header("x-request-id", "req_dc08_admin")
+            .json(&json!({
+                "version": 1,
+                "input": { "email": "dc08-canary-secret@example.com" }
+            }))
+            .send()
+            .await
+            .unwrap()
+    });
+
+    upstream.started.notified().await;
+    registry_for_failure
+        .delete_operation(
+            &WorkspaceId::new(DEFAULT_WORKSPACE_ID),
+            &OperationId::new(operation_id.clone()),
+        )
+        .await
+        .unwrap();
+    upstream.release.notify_one();
+
+    let response = request.await.unwrap();
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+    assert_eq!(
+        response.headers()["x-request-id"].to_str().unwrap(),
+        "req_dc08_admin"
+    );
+    let body = response.json::<Value>().await.unwrap();
+    assert_eq!(body["ok"], true);
+    assert_eq!(body["response_preview"]["id"], "lead_123");
+    assert_eq!(upstream.calls.load(Ordering::SeqCst), 1);
+    assert!(
+        crank_observability::operational_incident_total(
+            crank_observability::OperationalIncident::InvocationHistoryLost
+        ) > before
+    );
+
+    let logs = client
+        .get(format!("{base_url}/logs?period=7d"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert!(logs["items"].as_array().unwrap().is_empty());
+
+    let output = log_writer.output();
+    assert!(!output.contains("dc08-canary-secret"));
+    let incident = output
+        .lines()
+        .filter_map(|line| serde_json::from_str::<Value>(line).ok())
+        .find(|event| event["event"] == "admin.invocation_history.lost")
+        .expect("DC-08 incident");
+    assert_eq!(incident["request_id"], "req_dc08_admin");
+    assert_eq!(incident["fields"]["source"], "admin_test_run");
+}
+
+struct BlockingUpstream {
+    base_url: String,
+    started: Arc<Notify>,
+    release: Arc<Notify>,
+    calls: Arc<AtomicUsize>,
+}
+
+#[derive(Clone)]
+struct BlockingUpstreamState {
+    started: Arc<Notify>,
+    release: Arc<Notify>,
+    calls: Arc<AtomicUsize>,
+}
+
+async fn spawn_blocking_upstream_server() -> BlockingUpstream {
+    let state = BlockingUpstreamState {
+        started: Arc::new(Notify::new()),
+        release: Arc::new(Notify::new()),
+        calls: Arc::new(AtomicUsize::new(0)),
+    };
+    let app = Router::new()
+        .route("/crm/leads", post(blocking_create_lead))
+        .with_state(state.clone());
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+
+    BlockingUpstream {
+        base_url: format!("http://{address}"),
+        started: state.started,
+        release: state.release,
+        calls: state.calls,
+    }
+}
+
+async fn blocking_create_lead(
+    State(state): State<BlockingUpstreamState>,
+    Json(payload): Json<Value>,
+) -> Json<Value> {
+    state.calls.fetch_add(1, Ordering::SeqCst);
+    state.started.notify_one();
+    state.release.notified().await;
+
+    Json(json!({
+        "id": "lead_123",
+        "status": "created",
+        "email": payload["email"]
+    }))
+}
+
+#[derive(Clone, Default)]
+struct SharedLogWriter {
+    buffer: Arc<Mutex<Vec<u8>>>,
+}
+
+impl SharedLogWriter {
+    fn output(&self) -> String {
+        String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap()
+    }
+}
+
+impl<'a> MakeWriter<'a> for SharedLogWriter {
+    type Writer = SharedLogGuard;
+
+    fn make_writer(&'a self) -> Self::Writer {
+        SharedLogGuard {
+            buffer: Arc::clone(&self.buffer),
+        }
+    }
+}
+
+struct SharedLogGuard {
+    buffer: Arc<Mutex<Vec<u8>>>,
+}
+
+impl io::Write for SharedLogGuard {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.buffer.lock().unwrap().extend_from_slice(bytes);
+        Ok(bytes.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}