fix(server): log engine failures by their class outside the Blob routes

changelog.d/server-logs-error-class.fixed.md

@@ -0,0 +1,12 @@
+- Server logs no longer carry engine error text outside the Blob routes. An
+  internal failure on a change route logs `error_kind="change_route_internal"`,
+  and an export or change-baseline response that fails after its 200 headers
+  logs `error_kind="served_stream_failed"`, each with the error's class
+  (`error_variant`, `storage_kind`, `manifest_kind`) instead of its message,
+  which can hold object URIs or credentials. A stored-query invocation logs
+  the graph's ID instead of its storage root, a graph that fails to open at
+  startup logs its root without userinfo, query or fragment, and a refused
+  inline embedding `api_key` is no longer echoed. Responses are unchanged. See
+  [HTTP errors][server-logs-error-class].
+
+[server-logs-error-class]: ../docs/user/operations/troubleshooting.md#http-errors

changelog.d/server-restricts-lance-log-targets.changed.md

@@ -0,0 +1,8 @@
+- The server logs Lance's `lance::dataset_events` and `lance::file_audit`
+  targets only at warning and error, whatever `RUST_LOG` says, as it already
+  does for `rmcp`. At `info` they logged every table's storage URI, every file
+  Lance created or deleted, and each delete predicate with its entity IDs.
+  Embedders using their own subscriber should apply the same filter. See
+  [server logging][server-restricted-log-targets].
+
+[server-restricted-log-targets]: ../docs/user/operations/server.md#logs

crates/omnigraph-cluster/src/types.rs

@@ -526,11 +526,9 @@ fn secret_ref_name(value: &str) -> Result<&str, String> {
         .strip_prefix("${")
         .and_then(|s| s.strip_suffix('}'))
         .filter(|name| !name.trim().is_empty())
+        // Never echo the value: an inline one is the secret itself.
         .ok_or_else(|| {
-            format!(
-                "embedding api_key must be a ${{NAME}} env reference, got '{}'",
-                value.trim()
-            )
+            "embedding api_key must be a ${NAME} env reference, not an inline secret".to_string()
         })
 }
 
@@ -697,6 +695,10 @@ mod embedding_provider_config_tests {
         };
         let err = profile.resolve().unwrap_err();
         assert!(err.contains("${NAME}"), "got: {err}");
+        assert!(
+            !err.contains("sk-inline"),
+            "the refusal echoed the key: {err}"
+        );
     }
 
     #[test]

crates/omnigraph-server/src/blob_transport.rs

@@ -22,11 +22,12 @@ use axum::response::{IntoResponse, Response};
 use futures::future::BoxFuture;
 use futures::{Stream, StreamExt, stream};
 use headers::{ETag as TypedEtag, HeaderMapExt, IfMatch, IfNoneMatch, IfRange};
-use omnigraph::error::{ManifestErrorKind, OmniError, StorageFailureKind};
+use omnigraph::error::OmniError;
 use omnigraph::{BLOB_READ_RANGE_MAX_BYTES, BlobContent, BlobRead, BlobReader, ExternalBlobRef};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
 use crate::ApiError;
+use crate::redacted_cause::RedactedCause;
 
 const SNAPSHOT_ID_HEADER: HeaderName = HeaderName::from_static("omnigraph-snapshot-id");
 
@@ -36,68 +37,6 @@ pub(crate) const BLOB_BODY_MAX_RETAINED_CHUNKS: usize = 2;
 pub(crate) const BLOB_BODY_MAX_RETAINED_BYTES: u64 =
     BLOB_BODY_MAX_RETAINED_CHUNKS as u64 * BLOB_READ_RANGE_MAX_BYTES;
 
-/// Log-safe class of a redacted engine failure. Never carries message text:
-/// storage and integrity messages can hold object URIs, presigned query
-/// strings or credentials, and `OmniError`'s `Display` and `Debug` both
-/// include them.
-#[derive(Clone, Copy, Debug)]
-pub(crate) struct RedactedCause {
-    pub variant: &'static str,
-    pub storage_kind: Option<StorageFailureKind>,
-    pub manifest_kind: Option<ManifestErrorKind>,
-}
-
-impl RedactedCause {
-    pub(crate) fn of(error: &OmniError) -> Self {
-        // Completion evidence wraps the typed cause; the log names the cause.
-        let mut error = error;
-        while let OmniError::Completion { source, .. } = error {
-            error = source;
-        }
-        // Exhaustive on purpose: a new variant must choose its log name here.
-        let variant = match error {
-            OmniError::Completion { .. } => "Completion",
-            OmniError::Compiler(_) => "Compiler",
-            OmniError::Storage(_) => "Storage",
-            OmniError::HistoricalVersionReclaimed { .. } => "HistoricalVersionReclaimed",
-            OmniError::FullTextIndexRebuildRequired { .. } => "FullTextIndexRebuildRequired",
-            OmniError::RetryableCommitConflict(_) => "RetryableCommitConflict",
-            OmniError::DataFusion(_) => "DataFusion",
-            OmniError::Io(_) => "Io",
-            OmniError::Manifest(_) => "Manifest",
-            OmniError::MergeConflicts(_) => "MergeConflicts",
-            OmniError::KeyConflict { .. } => "KeyConflict",
-            OmniError::ResourceLimitExceeded { .. } => "ResourceLimitExceeded",
-            OmniError::ChangeCursorRejected { .. } => "ChangeCursorRejected",
-            OmniError::BranchNotFound { .. } => "BranchNotFound",
-            OmniError::ChangeFeedGap { .. } => "ChangeFeedGap",
-            OmniError::CommitHasNoParent { .. } => "CommitHasNoParent",
-            OmniError::ChangeSchemaBoundary { .. } => "ChangeSchemaBoundary",
-            OmniError::ExternalBlobPolicy { .. } => "ExternalBlobPolicy",
-            OmniError::ExternalBlobSource { .. } => "ExternalBlobSource",
-            OmniError::StoredExternalBlobDenied { .. } => "StoredExternalBlobDenied",
-            OmniError::BlobIntegrity { .. } => "BlobIntegrity",
-            OmniError::BlobRangeNotSatisfiable { .. } => "BlobRangeNotSatisfiable",
-            OmniError::RecoveryRequired { .. } => "RecoveryRequired",
-            OmniError::PreconditionFailed { .. } => "PreconditionFailed",
-            OmniError::Policy(_) => "Policy",
-            OmniError::AlreadyInitialized { .. } => "AlreadyInitialized",
-            OmniError::InitializationCommitted { .. } => "InitializationCommitted",
-            OmniError::InitializationIndeterminate { .. } => "InitializationIndeterminate",
-            OmniError::InitializationClaimed { .. } => "InitializationClaimed",
-        };
-        let manifest_kind = match error {
-            OmniError::Manifest(manifest) => Some(manifest.kind),
-            _ => None,
-        };
-        Self {
-            variant,
-            storage_kind: error.storage_failure().map(|failure| failure.kind),
-            manifest_kind,
-        }
-    }
-}
-
 trait RangeReader: Send + Sync + 'static {
     fn read_range(&self, range: Range<u64>) -> BoxFuture<'static, Result<Bytes, OmniError>>;
 }

crates/omnigraph-server/src/export_transport.rs

@@ -10,6 +10,8 @@ use std::time::Duration;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc};
 use tokio::time::timeout;
 
+use crate::redacted_cause::RedactedCause;
+
 /// At most two produced chunks may wait behind the response consumer.
 pub(crate) const EXPORT_QUEUE_CHUNKS: usize = 2;
 /// Queued and yielded chunks share these credits; clones and slices keep a
@@ -23,6 +25,9 @@ pub(crate) const EXPORT_QUEUE_RESERVED_BYTES: usize =
 pub(crate) const EXPORT_PROCESS_QUEUE_RESERVED_BYTES: usize = 8 * EXPORT_QUEUE_RESERVED_BYTES;
 /// A saturated request waits only briefly for a queue reservation.
 pub(crate) const EXPORT_RESERVATION_TIMEOUT: Duration = Duration::from_millis(250);
+/// The body error of a stream that failed after its 200 headers. The client
+/// reads no text from it; the server's HTTP trace layer logs it.
+const SERVED_STREAM_FAILED: &str = "served stream failed after response headers";
 
 static PROCESS_EXPORT_QUEUE_BYTES: OnceLock<Arc<Semaphore>> = OnceLock::new();
 
@@ -105,6 +110,8 @@ pub(crate) struct ExportQueueLease {
 pub(crate) struct ExportSender {
     sender: mpsc::Sender<ExportFrame>,
     lease: Arc<ExportQueueLease>,
+    /// Names the stream in the failure log.
+    stream: &'static str,
 }
 
 impl ExportSender {
@@ -159,7 +166,25 @@ impl ExportSender {
         self.send_chunk(output.0).await
     }
 
-    pub(crate) async fn finish(&self, cut: ExportCut, error: Option<io::Error>) {
+    /// End the response, with the failure that ended it. An engine message
+    /// can hold object URIs or credentials, so the body error is a constant
+    /// and the log names the failure by its class. A failure after the client
+    /// has gone reaches nobody and is not logged.
+    pub(crate) async fn finish(&self, cut: ExportCut, failure: Option<OmniError>) {
+        let error = failure.map(|failure| {
+            if !self.sender.is_closed() {
+                let cause = RedactedCause::of(&failure);
+                tracing::error!(
+                    error_kind = "served_stream_failed",
+                    stream = self.stream,
+                    error_variant = cause.variant,
+                    storage_kind = ?cause.storage_kind,
+                    manifest_kind = ?cause.manifest_kind,
+                    "served stream failed after response headers"
+                );
+            }
+            io::Error::other(SERVED_STREAM_FAILED)
+        });
         let _ = self
             .sender
             .send(ExportFrame::Terminal {
@@ -221,12 +246,16 @@ pub(crate) struct ExportBodyStream {
     done: bool,
 }
 
-pub(crate) fn channel(lease: Arc<ExportQueueLease>) -> (ExportSender, ExportBodyStream) {
+pub(crate) fn channel(
+    lease: Arc<ExportQueueLease>,
+    stream: &'static str,
+) -> (ExportSender, ExportBodyStream) {
     let (sender, receiver) = mpsc::channel(EXPORT_QUEUE_CHUNKS);
     (
         ExportSender {
             sender,
             lease: Arc::clone(&lease),
+            stream,
         },
         ExportBodyStream {
             receiver,
@@ -313,7 +342,7 @@ mod tests {
         );
         let lease = transport.reserve().await.unwrap();
         let producer_lease = Arc::clone(&lease);
-        let (sender, mut body) = channel(lease);
+        let (sender, mut body) = channel(lease, "export");
         sender
             .send_chunk(vec![b'x'; EXPORT_CHUNK_MAX_BYTES])
             .await
@@ -357,7 +386,7 @@ mod tests {
             Duration::from_millis(10),
         );
         let lease = transport.reserve().await.unwrap();
-        let (sender, mut body) = channel(lease);
+        let (sender, mut body) = channel(lease, "export");
 
         sender
             .send_chunk(vec![b'a'; EXPORT_CHUNK_MAX_BYTES])
@@ -454,7 +483,7 @@ mod tests {
     async fn missing_terminal_frame_is_a_body_error_not_clean_eof() {
         let transport = ExportTransport::new(4, 4, Duration::from_millis(10));
         let lease = transport.reserve().await.unwrap();
-        let (sender, mut body) = channel(lease);
+        let (sender, mut body) = channel(lease, "export");
         drop(sender);
 
         let error = body.next().await.unwrap().unwrap_err();
@@ -484,7 +513,7 @@ mod tests {
         );
         let lease = transport.reserve().await.unwrap();
         let producer_lease = Arc::clone(&lease);
-        let (sender, mut body) = channel(lease);
+        let (sender, mut body) = channel(lease, "export");
 
         let data_sender = sender.clone();
         let (cut, result) = cut
@@ -527,11 +556,45 @@ mod tests {
             EXPORT_QUEUE_RESERVED_BYTES
         );
 
+        // An engine failure after the headers reaches the body as a constant
+        // and the log as its class: the trace layer logs a body error's text,
+        // and an engine message can hold object URIs or credentials.
+        let capture = crate::test_log_capture::Capture::default();
+        let _logs = tracing::subscriber::set_default(capture.subscriber("info"));
+        let cut = db.capture_served_export_cut("main", &[]).await.unwrap();
+        let (sender, mut body) = channel(transport.reserve().await.unwrap(), "export");
+        sender
+            .finish(
+                cut,
+                Some(OmniError::Storage(omnigraph::error::StorageFailure::new(
+                    omnigraph::error::StorageFailureKind::Unknown,
+                    "storage: GET s3://private-bucket/graph/nodes/0a.lance?token=secret",
+                ))),
+            )
+            .await;
+        drop(sender);
+        let error = body.next().await.unwrap().unwrap_err();
+        assert_eq!(error.to_string(), SERVED_STREAM_FAILED);
+        assert!(body.next().await.is_none());
+        drop(body);
+        let logs = capture.output();
+        for expected in [
+            r#"error_kind="served_stream_failed""#,
+            r#"stream="export""#,
+            r#"error_variant="Storage""#,
+            "storage_kind=Some(Unknown)",
+        ] {
+            assert!(logs.contains(expected), "missing {expected}: {logs}");
+        }
+        for leaked in ["private-bucket", "0a.lance", "token", "secret"] {
+            assert!(!logs.contains(leaked), "log leaked {leaked}: {logs}");
+        }
+
         // Baseline sends its cursor after the outer export/closed select has
         // completed. Receiver closure must release this producer's cut even
         // when all frame credits survive in transport-owned clones or slices.
         let cut = db.capture_served_export_cut("main", &[]).await.unwrap();
-        let (sender, mut body) = channel(transport.reserve().await.unwrap());
+        let (sender, mut body) = channel(transport.reserve().await.unwrap(), "change_baseline");
         let mut retained = Vec::new();
         for _ in 0..EXPORT_OUTSTANDING_FRAMES {
             sender
@@ -547,16 +610,19 @@ mod tests {
                     .await
                     .unwrap_err();
                 assert!(matches!(error, OmniError::Io(_)));
-                sender
-                    .finish(cut, Some(io::Error::other(error.to_string())))
-                    .await;
+                sender.finish(cut, Some(error)).await;
             };
             tokio::pin!(terminal);
             assert!(futures::poll!(&mut terminal).is_pending());
             drop(body);
             assert!(futures::poll!(&mut terminal).is_ready());
         }
         drop(sender);
+        assert_eq!(
+            capture.output().matches("served_stream_failed").count(),
+            1,
+            "a failure after the client has gone is not logged"
+        );
         assert_eq!(transport.available_bytes.available_permits(), 0);
         let retry = db.capture_served_export_cut("main", &[]).await.unwrap();
         drop(retry);