feat: surface the swallowed errors across the fleet

crates/api-core/src/handlers/credential.rs

@@ -526,7 +526,8 @@ pub(crate) async fn get_switch_nvos_credentials(
             db::ObjectColumnFilter::One(db::switch::IdColumn, &switch_id),
         )
         .await?;
-        let _ = txn.rollback().await;
+        txn.rollback_or_log("read-only load of switch for credential lookup")
+            .await;
 
         let switch = switches
             .first()

crates/api-core/src/handlers/expected_machine.rs

@@ -590,7 +590,8 @@ async fn process_batch_operations(
                     }
                 },
                 Err(e) => {
-                    let _ = txn.rollback().await;
+                    txn.rollback_or_log("expected-machine write after operation failure")
+                        .await;
                     results.push(build_failure_result(id, format!("Operation failed: {}", e)));
                 }
             }
@@ -624,7 +625,8 @@ async fn process_batch_operations(
         )
         .await
         {
-            let _ = txn.rollback().await;
+            txn.rollback_or_log("expected-machine write after operation failure")
+                .await;
             return Err(e);
         }
         results.push(build_success_result(machine_for_result));

crates/api-core/src/handlers/instance.rs

@@ -278,7 +278,8 @@ pub(crate) async fn find_by_ids(
     for snapshot in snapshots.into_iter() {
         instances.push(snapshot_to_instance(snapshot)?);
     }
-    let _ = txn.rollback().await;
+    txn.rollback_or_log("read-only load of instances by id")
+        .await;
 
     Ok(Response::new(rpc::InstanceList { instances }))
 }

crates/api-core/src/handlers/power_shelf.rs

@@ -138,7 +138,8 @@ pub async fn find_by_ids(
     )
     .await?;
 
-    let _ = txn.rollback().await;
+    txn.rollback_or_log("read-only load of power shelves by id")
+        .await;
 
     let power_shelves = convert_power_shelves(power_shelf_list)?;
 

crates/api-core/src/handlers/rack.rs

@@ -125,7 +125,7 @@ pub async fn find_by_ids(
         result.push(rack.into());
     }
 
-    let _ = txn.rollback().await;
+    txn.rollback_or_log("read-only load of racks by id").await;
 
     Ok(Response::new(rpc::RackList { racks: result }))
 }

crates/api-core/src/handlers/switch.rs

@@ -176,7 +176,8 @@ pub async fn find_by_ids(
             .map(|row| (row.switch_id, row))
             .collect();
 
-    let _ = txn.rollback().await;
+    txn.rollback_or_log("read-only load of switches by id")
+        .await;
 
     let switches: Vec<rpc::Switch> = switch_list
         .into_iter()

crates/api-db/src/lib.rs

@@ -686,6 +686,31 @@ impl<'a> Transaction<'a> {
         })
     }
 
+    // This function can just async when
+    // https://github.com/rust-lang/rust/issues/110011 will be
+    // implemented
+    /// Roll back the transaction, logging a warning tagged with `context` if
+    /// the rollback itself fails, rather than propagating -- for read-only or
+    /// cleanup paths where the caller can't act on a rollback error but must
+    /// not swallow it.
+    #[track_caller]
+    pub fn rollback_or_log(
+        self,
+        context: &'static str,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        let loc = Location::caller();
+        Box::pin(async move {
+            if let Err(e) = self
+                .inner
+                .rollback()
+                .await
+                .map_err(|e| DatabaseError::txn_rollback(e, loc))
+            {
+                tracing::warn!(error = %e, context, "Failed to roll back transaction");
+            }
+        })
+    }
+
     pub fn as_pgconn(&mut self) -> &mut sqlx::PgConnection {
         &mut self.inner
     }

crates/dhcp/src/lib.rs

@@ -75,12 +75,27 @@ impl Default for CarbideDhcpContext {
         Self {
             api_endpoint: "https://[::1]:1079".to_string(),
             nameservers: vec![Ipv4Addr::new(1, 1, 1, 1)],
-            forge_root_ca_path: std::env::var("FORGE_ROOT_CAFILE_PATH")
-                .unwrap_or_else(|_| tls_default::ROOT_CA.to_string()),
-            forge_client_cert_path: std::env::var("FORGE_CLIENT_CERT_PATH")
-                .unwrap_or_else(|_| tls_default::CLIENT_CERT.to_string()),
-            forge_client_key_path: std::env::var("FORGE_CLIENT_KEY_PATH")
-                .unwrap_or_else(|_| tls_default::CLIENT_KEY.to_string()),
+            forge_root_ca_path: std::env::var("FORGE_ROOT_CAFILE_PATH").unwrap_or_else(|e| {
+                log::warn!(
+                    "FORGE_ROOT_CAFILE_PATH unset or unreadable ({e}); falling back to the built-in default root CA at {}",
+                    tls_default::ROOT_CA
+                );
+                tls_default::ROOT_CA.to_string()
+            }),
+            forge_client_cert_path: std::env::var("FORGE_CLIENT_CERT_PATH").unwrap_or_else(|e| {
+                log::warn!(
+                    "FORGE_CLIENT_CERT_PATH unset or unreadable ({e}); falling back to the built-in default client certificate at {}",
+                    tls_default::CLIENT_CERT
+                );
+                tls_default::CLIENT_CERT.to_string()
+            }),
+            forge_client_key_path: std::env::var("FORGE_CLIENT_KEY_PATH").unwrap_or_else(|e| {
+                log::warn!(
+                    "FORGE_CLIENT_KEY_PATH unset or unreadable ({e}); falling back to the built-in default client key at {}",
+                    tls_default::CLIENT_KEY
+                );
+                tls_default::CLIENT_KEY.to_string()
+            }),
             ntpservers: vec![
                 Ipv4Addr::new(172, 20, 0, 24),
                 Ipv4Addr::new(172, 20, 0, 26),

crates/dsx-exchange-consumer/Cargo.toml

@@ -62,6 +62,7 @@ carbide-version = { path = "../version" }
 [dev-dependencies]
 carbide-instrument = { path = "../instrument", features = ["test-support"] }
 carbide-test-support = { path = "../test-support" }
+tracing = { workspace = true }
 
 [lints]
 workspace = true

crates/dsx-exchange-consumer/src/health_updater.rs

@@ -30,7 +30,7 @@ use crate::ConsumerMetrics;
 use crate::api_client::{HEALTH_REPORT_SOURCE, RackHealthReportSink};
 use crate::config::CacheConfig;
 use crate::messages::{FaultValue, LeakMetadata, LeakPointType, ValueMessage};
-use crate::metrics::{MessageDeduplicated, MessageProcessed};
+use crate::metrics::{HealthReportPersistFailed, MessageDeduplicated, MessageProcessed};
 use crate::mqtt_consumer::MqttMessage;
 
 /// Health status updater that processes MQTT messages and updates the API.
@@ -212,8 +212,14 @@ impl<S: RackHealthReportSink> HealthUpdater<S> {
             Ok(_) => {
                 emit(MessageProcessed);
             }
-            Err(_) => {
-                // API call failed - will retry on next message
+            Err(e) => {
+                // Surface the persist failure and count it. The value re-arrives
+                // on the next message, so processing still retries -- but the
+                // drop is now visible instead of silently discarded.
+                emit(HealthReportPersistFailed {
+                    rack_id: metadata.rack_id.clone(),
+                    error: e.to_string(),
+                });
             }
         }
     }

crates/dsx-exchange-consumer/src/metrics.rs

@@ -119,6 +119,33 @@ pub struct MessageDropped;
 )]
 pub struct MessageDeduplicated;
 
+/// A rack health report could not be persisted to the Carbide API -- either a
+/// coolant-leak override insert or its clearing removal failed. The value
+/// re-arrives on the next message, so processing still retries, but a rising
+/// rate is safety-relevant: leak state is not reaching the API. Logs the
+/// failure and moves the counter from one `emit`.
+///
+/// Unlike the grandfathered counters above, this is a new metric, so it uses
+/// the standard checked name: the exporter appends the single `_total` that
+/// `/metrics` shows.
+#[derive(Event)]
+#[event(
+    name = "carbide_dsx_exchange_consumer_health_report_persist_failures_total",
+    component = "nico-dsx-exchange-consumer",
+    log = warn,
+    metric = counter,
+    message = "Failed to persist rack health report",
+    describe = "Number of rack health report persist failures against the Carbide API"
+)]
+pub struct HealthReportPersistFailed {
+    /// The rack whose health report could not be persisted.
+    #[context]
+    pub rack_id: String,
+    /// The API error that prevented the persist.
+    #[context]
+    pub error: String,
+}
+
 /// Consumer metrics that remain hand-rolled OpenTelemetry counters.
 ///
 /// Only `alerts_detected` stays here: its `point_type` label is a

crates/dsx-exchange-consumer/tests/metric_exposition.rs

@@ -20,15 +20,18 @@
 //! registering and the OTel Prometheus exporter appends exactly one, so
 //! `/metrics` shows one suffix, not the historical doubled `_total_total`.
 //!
-//! One test in its own binary (its own process-global registry) keeps the
-//! `counter_delta` measurements deterministic: the crate's other unit tests
-//! emit these same events, but from a different test process.
+//! These tests live in their own binary (its own process-global registry) to
+//! keep the `counter_delta` measurements deterministic: the crate's other unit
+//! tests emit these same events -- the message counters here, and the
+//! health-report persist failure below -- but from a different test process, so
+//! they cannot advance a shared counter between a test's baseline and delta.
 
 use carbide_dsx_exchange_consumer::metrics::{
-    MessageDeduplicated, MessageDropped, MessageProcessed, MessageReceived,
+    HealthReportPersistFailed, MessageDeduplicated, MessageDropped, MessageProcessed,
+    MessageReceived,
 };
 use carbide_instrument::emit;
-use carbide_instrument::testing::{MetricsCapture, capture_logs};
+use carbide_instrument::testing::{CapturedLog, MetricsCapture, capture_logs};
 
 /// Emitting each event once moves exactly its counter, under the single
 /// `_total` name the OTel Prometheus exporter produces (the framework strips
@@ -81,3 +84,43 @@ fn message_events_expose_single_total_names_and_are_metric_only() {
     // remain.
     assert!(logs.is_empty(), "events must be metric-only: {logs:?}");
 }
+
+/// A persist failure writes the WARN line carrying the rack id and error, and
+/// moves `carbide_dsx_exchange_consumer_health_report_persist_failures_total` once --
+/// the "log it AND count it" the safety-relevant drop needs. Isolated here
+/// because `health_updater`'s failure-path unit tests emit this same event
+/// without a `MetricsCapture`; from a shared process they could advance the
+/// zero-label counter between this test's baseline and delta.
+#[test]
+fn health_report_persist_failed_logs_warn_and_counts() {
+    let metrics = MetricsCapture::start();
+    let logs = capture_logs(|| {
+        emit(HealthReportPersistFailed {
+            rack_id: "rack-42".to_string(),
+            error: "API call failed: deadline exceeded".to_string(),
+        });
+    });
+
+    assert_eq!(logs.len(), 1);
+    assert_eq!(logs[0].level, tracing::Level::WARN);
+    assert_eq!(logs[0].message, "Failed to persist rack health report");
+    fn field<'a>(log: &'a CapturedLog, name: &str) -> Option<&'a str> {
+        log.fields
+            .iter()
+            .find(|(key, _)| key == name)
+            .map(|(_, value)| value.as_str())
+    }
+    assert_eq!(field(&logs[0], "rack_id"), Some("rack-42"));
+    assert_eq!(
+        field(&logs[0], "error"),
+        Some("API call failed: deadline exceeded")
+    );
+
+    assert_eq!(
+        metrics.counter_delta(
+            "carbide_dsx_exchange_consumer_health_report_persist_failures_total",
+            &[],
+        ),
+        1.0
+    );
+}

crates/health/src/sink/log_file.rs

@@ -208,14 +208,18 @@ impl SyncLogFileWriter {
         tracing::info!(size = self.current_size, "rotating log file");
 
         // flush and drop the current file handle before renaming
-        if let Some(mut file) = self.current_file.take() {
-            let _ = file.flush();
+        if let Some(mut file) = self.current_file.take()
+            && let Err(e) = file.flush()
+        {
+            tracing::warn!(error = %e, "failed to flush log file before rotation");
         }
 
         let current_path = self.log_path();
 
         if self.max_backups == 0 {
-            let _ = fs::remove_file(&current_path);
+            if let Err(e) = fs::remove_file(&current_path) {
+                tracing::warn!(error = %e, path = %current_path.display(), "failed to remove current log file during rotation");
+            }
         } else {
             self.shift_backups(&current_path);
         }
@@ -229,19 +233,25 @@ impl SyncLogFileWriter {
         for i in (1..self.max_backups).rev() {
             let from = self.rotated_path(i);
             let to = self.rotated_path(i + 1);
-            if from.exists() {
-                let _ = fs::rename(&from, &to);
+            if from.exists()
+                && let Err(e) = fs::rename(&from, &to)
+            {
+                tracing::warn!(error = %e, from = %from.display(), to = %to.display(), "failed to shift rotated log file");
             }
         }
 
         // current -> .1
         let backup = self.rotated_path(1);
-        let _ = fs::rename(current_path, &backup);
+        if let Err(e) = fs::rename(current_path, &backup) {
+            tracing::warn!(error = %e, from = %current_path.display(), to = %backup.display(), "failed to rotate current log file to backup");
+        }
 
         // prune the oldest backup beyond the limit
         let oldest = self.rotated_path(self.max_backups + 1);
-        if oldest.exists() {
-            let _ = fs::remove_file(&oldest);
+        if oldest.exists()
+            && let Err(e) = fs::remove_file(&oldest)
+        {
+            tracing::warn!(error = %e, path = %oldest.display(), "failed to prune oldest rotated log file");
         }
     }
 }

crates/machine-controller/src/handler/dpf.rs

@@ -195,6 +195,13 @@ async fn create_and_register_dpudevices_and_dpunode(
             .and_then(|x| x.dmi_data.as_ref())
             .map(|x| x.product_serial.as_str())
             .unwrap_or_default();
+        if serial_number.is_empty() {
+            tracing::warn!(
+                dpu_machine_id = %dpu.id,
+                host_machine_id = %state.host_snapshot.id,
+                "DPU product serial is missing; registering DPU device with DPF using an empty serial"
+            );
+        }
         let device_info = carbide_dpf::DpuDeviceInfo {
             device_id: dpf_id(dpu)?,
             dpu_bmc_ip: bmc_ip(dpu)?,