From d5d5c92e9aac00b6bd741c52923976f4fff2e0f1 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Thu, 3 Sep 2026 19:15:56 +0000
Subject: [PATCH 01/14] datastore: add generic key-value storage and persisted
 correlation ID

Add a generic keyvalue table to the SQLite datastore so arbitrary
structured data (JSON-serialized) can be stored/retrieved by key, not
just HostStatus. DataStore::get_value/set_value provide the generic
API; keyvalue rows are carried over when a temporary datastore is
persisted, same as HostStatus.

As a first consumer, add DataStore::correlation_id(), which generates
and persists a UUID on first access and returns the same value on
every subsequent call. The correlation ID is retrieved at trident CLI
startup and attached to every trace/metric entry via
TraceStream::set_correlation_id, so all tracing/telemetry for a given
host installation can be correlated.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 crates/trident/src/datastore.rs           | 279 ++++++++++++++++++++++
 crates/trident/src/logging/tracestream.rs |  45 +++-
 crates/trident/src/main.rs                |  22 +-
 crates/trident_api/src/error.rs           |  12 +
 4 files changed, 353 insertions(+), 5 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 9136dba1f..1d8dba182 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -1,7 +1,9 @@
 use std::{fs, path::Path};
 
 use log::{debug, warn};
+use serde::{de::DeserializeOwned, Serialize};
 use sqlite::State;
+use uuid::Uuid;
 
 use trident_api::{
     error::{
@@ -12,6 +14,13 @@ use trident_api::{
 
 use crate::TRIDENT_SEMVER_VERSION;
 
+/// Key under which the datastore's unique correlation ID is stored in the
+/// generic key-value table. This ID is generated once (on first access) and
+/// persisted for the lifetime of the datastore. It is intended to be added to
+/// tracing/telemetry so that all activity for a given host installation can
+/// be correlated.
+const CORRELATION_ID_KEY: &str = "correlation-id";
+
 pub struct DataStore {
     db: Option<sqlite::Connection>,
     host_status: HostStatus,
@@ -128,6 +137,13 @@ impl DataStore {
             )",
         )
         .structured(ServicingError::from(DatastoreError::InitializeDatastore))?;
+        db.execute(
+            "CREATE TABLE IF NOT EXISTS keyvalue (
+                key TEXT PRIMARY KEY,
+                contents TEXT NOT NULL
+            )",
+        )
+        .structured(ServicingError::from(DatastoreError::InitializeDatastore))?;
         Ok(db)
     }
 
@@ -139,6 +155,14 @@ impl DataStore {
                 TridentVersion::SemVer(TRIDENT_SEMVER_VERSION.clone());
             Self::write_host_status(&persistent_db, self.host_status())?;
 
+            // Carry over any generic key-value entries (e.g. the correlation ID)
+            // recorded in the temporary datastore into the persistent one, so
+            // they survive the transition from temporary to persistent
+            // storage.
+            if let Some(temporary_db) = self.db.as_ref() {
+                Self::copy_key_values(temporary_db, &persistent_db)?;
+            }
+
             self.db = Some(persistent_db);
             self.temporary = false;
         }
@@ -146,6 +170,65 @@ impl DataStore {
         Ok(())
     }
 
+    /// Copy all rows of the generic key-value table from `source` into
+    /// `destination`, overwriting any conflicting keys already present in
+    /// `destination`.
+    fn copy_key_values(
+        source: &sqlite::Connection,
+        destination: &sqlite::Connection,
+    ) -> Result<(), TridentError> {
+        let mut query_statement = source
+            .prepare("SELECT key, contents FROM keyvalue")
+            .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+
+        loop {
+            match query_statement.next() {
+                Ok(State::Done) => break,
+                Err(e) => {
+                    warn!(
+                        "Failed to get next keyvalue row while copying datastore: {:?}",
+                        e
+                    );
+                    break;
+                }
+                Ok(State::Row) => {} // continue below
+            }
+
+            let key = query_statement
+                .read::<String, _>(0)
+                .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+            let contents = query_statement
+                .read::<String, _>(1)
+                .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+
+            let mut insert_statement = destination
+                .prepare(
+                    "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
+                     ON CONFLICT(key) DO UPDATE SET contents = excluded.contents",
+                )
+                .structured(ServicingError::Datastore {
+                    inner: DatastoreError::WriteKeyValue { key: key.clone() },
+                })?;
+            insert_statement
+                .bind((1, &*key))
+                .structured(ServicingError::Datastore {
+                    inner: DatastoreError::WriteKeyValue { key: key.clone() },
+                })?;
+            insert_statement
+                .bind((2, &*contents))
+                .structured(ServicingError::Datastore {
+                    inner: DatastoreError::WriteKeyValue { key: key.clone() },
+                })?;
+            insert_statement
+                .next()
+                .structured(ServicingError::Datastore {
+                    inner: DatastoreError::WriteKeyValue { key },
+                })?;
+        }
+
+        Ok(())
+    }
+
     fn write_host_status(
         db: &sqlite::Connection,
         host_status: &HostStatus,
@@ -219,6 +302,128 @@ impl DataStore {
         self.db = None;
     }
 
+    /// Retrieve a structured value stored under `key` in the datastore's
+    /// generic key-value table, if present.
+    ///
+    /// Values are serialized as JSON, so any type implementing
+    /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not
+    /// just `HostStatus`.
+    pub(crate) fn get_value<T: DeserializeOwned>(
+        &self,
+        key: &str,
+    ) -> Result<Option<T>, TridentError> {
+        let db = self
+            .db
+            .as_ref()
+            .structured(ServicingError::from(DatastoreError::OpenDatastore))?;
+
+        let mut statement = db
+            .prepare("SELECT contents FROM keyvalue WHERE key = ?")
+            .structured(ServicingError::Datastore {
+                inner: DatastoreError::ReadKeyValue {
+                    key: key.to_string(),
+                },
+            })?;
+        statement
+            .bind((1, key))
+            .structured(ServicingError::Datastore {
+                inner: DatastoreError::ReadKeyValue {
+                    key: key.to_string(),
+                },
+            })?;
+
+        match statement.next().structured(ServicingError::Datastore {
+            inner: DatastoreError::ReadKeyValue {
+                key: key.to_string(),
+            },
+        })? {
+            State::Row => {
+                let contents =
+                    statement
+                        .read::<String, _>(0)
+                        .structured(ServicingError::Datastore {
+                            inner: DatastoreError::ReadKeyValue {
+                                key: key.to_string(),
+                            },
+                        })?;
+                let value = serde_json::from_str(&contents).structured(
+                    InternalError::DeserializeValue {
+                        key: key.to_string(),
+                    },
+                )?;
+                Ok(Some(value))
+            }
+            State::Done => Ok(None),
+        }
+    }
+
+    /// Store a structured value under `key` in the datastore's generic
+    /// key-value table, overwriting any previous value stored under the
+    /// same key.
+    ///
+    /// Values are serialized as JSON, so any type implementing
+    /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not
+    /// just `HostStatus`.
+    pub(crate) fn set_value<T: Serialize>(&self, key: &str, value: &T) -> Result<(), TridentError> {
+        let db = self
+            .db
+            .as_ref()
+            .structured(ServicingError::from(DatastoreError::WriteToClosedDatastore))?;
+
+        let contents = serde_json::to_string(value).structured(InternalError::SerializeValue {
+            key: key.to_string(),
+        })?;
+
+        let mut statement = db
+            .prepare(
+                "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
+                 ON CONFLICT(key) DO UPDATE SET contents = excluded.contents",
+            )
+            .structured(ServicingError::Datastore {
+                inner: DatastoreError::WriteKeyValue {
+                    key: key.to_string(),
+                },
+            })?;
+        statement
+            .bind((1, key))
+            .structured(ServicingError::Datastore {
+                inner: DatastoreError::WriteKeyValue {
+                    key: key.to_string(),
+                },
+            })?;
+        statement
+            .bind((2, &*contents))
+            .structured(ServicingError::Datastore {
+                inner: DatastoreError::WriteKeyValue {
+                    key: key.to_string(),
+                },
+            })?;
+        statement.next().structured(ServicingError::Datastore {
+            inner: DatastoreError::WriteKeyValue {
+                key: key.to_string(),
+            },
+        })?;
+
+        Ok(())
+    }
+
+    /// Retrieve this datastore's unique correlation ID, generating and
+    /// persisting a new one on first access.
+    ///
+    /// This ID is stable for the lifetime of the datastore (surviving the
+    /// temporary-to-persistent transition performed by `persist`), and is
+    /// intended to be attached to tracing/telemetry so that activity for a
+    /// given host installation can be correlated across logs and traces.
+    pub fn correlation_id(&mut self) -> Result<Uuid, TridentError> {
+        if let Some(id) = self.get_value::<Uuid>(CORRELATION_ID_KEY)? {
+            return Ok(id);
+        }
+
+        let id = Uuid::new_v4();
+        self.set_value(CORRELATION_ID_KEY, &id)?;
+        Ok(id)
+    }
+
     /// Parse a single HostStatus entry from a datastore query result.
     /// 1. Read each row as a string containing YAML-encoded Host Status.
     /// 2. Decode the YAML string into a serde_yaml Value.
@@ -306,6 +511,60 @@ mod tests {
             .parse_host_status(Ok(serde_yaml::to_string(&valid_host_status).unwrap()))
             .is_some());
     }
+
+    #[test]
+    fn test_generic_key_value_store() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let path = temp_dir.path().join("db.sqlite");
+        let db = super::DataStore::make_datastore(&path).unwrap();
+        let datastore = super::DataStore {
+            db: Some(db),
+            host_status: Default::default(),
+            temporary: false,
+        };
+
+        // No value stored yet.
+        assert_eq!(datastore.get_value::<String>("some-key").unwrap(), None);
+
+        // Store and retrieve a value.
+        datastore
+            .set_value("some-key", &"some-value".to_string())
+            .unwrap();
+        assert_eq!(
+            datastore.get_value::<String>("some-key").unwrap(),
+            Some("some-value".to_string())
+        );
+
+        // Overwrite the value.
+        datastore
+            .set_value("some-key", &"other-value".to_string())
+            .unwrap();
+        assert_eq!(
+            datastore.get_value::<String>("some-key").unwrap(),
+            Some("other-value".to_string())
+        );
+
+        temp_dir.close().unwrap();
+    }
+
+    #[test]
+    fn test_correlation_id_is_stable() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let path = temp_dir.path().join("db.sqlite");
+        let db = super::DataStore::make_datastore(&path).unwrap();
+        let mut datastore = super::DataStore {
+            db: Some(db),
+            host_status: Default::default(),
+            temporary: false,
+        };
+
+        let id = datastore.correlation_id().unwrap();
+        // Calling correlation_id again should return the same ID, not generate a
+        // new one.
+        assert_eq!(datastore.correlation_id().unwrap(), id);
+
+        temp_dir.close().unwrap();
+    }
 }
 
 #[cfg(feature = "functional-test")]
@@ -381,4 +640,24 @@ mod functional_test {
             ServicingState::Provisioned
         );
     }
+
+    #[functional_test]
+    fn test_correlation_id_survives_persist() {
+        let temp_dir = TempDir::new().unwrap();
+        let datastore_temp_path = temp_dir.path().join("db-tmp.sqlite");
+        let datastore_path = temp_dir.path().join("db.sqlite");
+
+        // Generate a correlation ID in the temporary datastore, then persist it.
+        let correlation_id = {
+            let mut datastore = DataStore::open_or_create(&datastore_temp_path).unwrap();
+            let correlation_id = datastore.correlation_id().unwrap();
+            datastore.persist(&datastore_path).unwrap();
+            correlation_id
+        };
+
+        // Re-open the persisted datastore and verify the same correlation ID is
+        // returned, rather than a new one being generated.
+        let mut datastore = DataStore::open(&datastore_path).unwrap();
+        assert_eq!(datastore.correlation_id().unwrap(), correlation_id);
+    }
 }
diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index 132696833..20eb511fa 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -84,6 +84,7 @@ pub struct TraceStream {
     // TODO: Consider changing this to a LockOnce when rustc is updated to
     // >=1.70
     target: Arc<RwLock<Option<String>>>,
+    correlation_id: Arc<RwLock<Option<String>>>,
     disabled: bool,
 }
 
@@ -125,14 +126,32 @@ impl TraceStream {
         Ok(())
     }
 
+    /// Set the correlation ID to attach to every trace entry sent from this point
+    /// forward, as an additional field, so that all traces/metrics for a
+    /// given host installation can be correlated. Expected to be called once
+    /// the datastore's persisted correlation ID has been retrieved (see
+    /// `DataStore::correlation_id`).
+    pub fn set_correlation_id(&self, correlation_id: String) {
+        match self.correlation_id.write() {
+            Ok(mut val) => {
+                val.replace(correlation_id);
+            }
+            Err(_) => warn!("Failed to lock tracestream to set correlation ID"),
+        }
+    }
+
     /// Create a Boxed TraceSender
     pub fn make_trace_sender(&self) -> Box<TraceSender> {
-        Box::new(TraceSender::new(self.target.clone()))
+        Box::new(TraceSender::new(
+            self.target.clone(),
+            self.correlation_id.clone(),
+        ))
     }
 }
 
 pub struct TraceSender {
     server: Arc<RwLock<Option<String>>>,
+    correlation_id: Arc<RwLock<Option<String>>>,
     client: reqwest::blocking::Client,
     metrics_file: Option<File>,
 }
@@ -144,9 +163,13 @@ struct ExecutionTime(Instant);
 /// the tracing-subscriber crate to handle the events and send them to the
 /// server.
 impl TraceSender {
-    fn new(server: Arc<RwLock<Option<String>>>) -> Self {
+    fn new(
+        server: Arc<RwLock<Option<String>>>,
+        correlation_id: Arc<RwLock<Option<String>>>,
+    ) -> Self {
         Self {
             server,
+            correlation_id,
             client: reqwest::blocking::Client::new(),
             metrics_file: match files::create_file(TRIDENT_METRICS_FILE_PATH) {
                 Ok(f) => Some(f),
@@ -164,6 +187,20 @@ impl TraceSender {
         self.server.read().map(|s| s.clone()).unwrap_or_default()
     }
 
+    /// Build the `additional_fields` map for a trace entry: the static
+    /// `ADDITIONAL_FIELDS`, plus the correlation ID (if one has been set via
+    /// `TraceStream::set_correlation_id`), so entries can be correlated back to a
+    /// specific host installation.
+    fn additional_fields(&self) -> BTreeMap<String, Value> {
+        let mut fields = ADDITIONAL_FIELDS.clone();
+        if let Ok(correlation_id) = self.correlation_id.read() {
+            if let Some(correlation_id) = correlation_id.as_ref() {
+                fields.insert("correlation_id".to_string(), json!(correlation_id));
+            }
+        }
+        fields
+    }
+
     fn write_metric_to_file(&self, metric: String) {
         if let Some(mut file) = self.metrics_file.as_ref() {
             if let Err(e) = file.write_all(format!("{metric}\n").as_bytes()) {
@@ -229,7 +266,7 @@ where
             timestamp: Utc::now(),
             metric_name,
             value: json!(value),
-            additional_fields: ADDITIONAL_FIELDS.clone(),
+            additional_fields: self.additional_fields(),
             platform_info: PLATFORM_INFO.clone(),
         };
 
@@ -309,7 +346,7 @@ where
             timestamp: Utc::now(),
             metric_name: span.name().to_string(),
             value: json!(visitor.fields),
-            additional_fields: ADDITIONAL_FIELDS.clone(),
+            additional_fields: self.additional_fields(),
             platform_info: PLATFORM_INFO.clone(),
         };
 
diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs
index 9d3f2b7bf..7e121a9ff 100644
--- a/crates/trident/src/main.rs
+++ b/crates/trident/src/main.rs
@@ -2,7 +2,7 @@ use std::{fs, iter, panic, process::ExitCode};
 
 use anyhow::{Context, Error};
 use clap::Parser;
-use log::{error, info, LevelFilter, Log};
+use log::{error, info, warn, LevelFilter, Log};
 
 use osutils::logging::{filter::LogFilter, multilog::MultiLogger};
 use trident::{
@@ -157,6 +157,11 @@ fn run_trident(
                         .message("Datastore file does not exist");
                 }
 
+                // Clone the tracestream handle before it is moved into `Trident::new`,
+                // so the correlation ID can be attached to it below once the
+                // datastore has been opened.
+                let tracestream_for_correlation_id = tracestream.clone();
+
                 let mut trident = Trident::new(
                     config_path.map(HostConfigurationSource::File),
                     agent_config.datastore_path(),
@@ -168,6 +173,21 @@ fn run_trident(
                 let mut datastore = DataStore::open_or_create(agent_config.datastore_path())
                     .message("Failed to open datastore")?;
 
+                // Retrieve (or create, on first run) this host's unique
+                // correlation ID from the datastore, and attach it to all
+                // subsequent traces/metrics so they can be correlated with
+                // this specific installation.
+                match datastore.correlation_id() {
+                    Ok(correlation_id) => {
+                        info!("Correlation ID: {correlation_id}");
+                        tracestream_for_correlation_id
+                            .set_correlation_id(correlation_id.to_string());
+                    }
+                    Err(e) => {
+                        warn!("Failed to get or create correlation ID: {e:?}");
+                    }
+                }
+
                 // Execute the command
                 let res = match args.command {
                     Commands::Install {
diff --git a/crates/trident_api/src/error.rs b/crates/trident_api/src/error.rs
index 064be997d..1a8e4b9f3 100644
--- a/crates/trident_api/src/error.rs
+++ b/crates/trident_api/src/error.rs
@@ -124,6 +124,12 @@ pub enum InternalError {
     #[error("Failed to serialize Host Status")]
     SerializeHostStatus,
 
+    #[error("Failed to serialize value for datastore key '{key}'")]
+    SerializeValue { key: String },
+
+    #[error("Failed to deserialize value for datastore key '{key}'")]
+    DeserializeValue { key: String },
+
     #[error("Failed to set up extension images on the target OS")]
     SetUpExtensionImages,
 
@@ -742,11 +748,17 @@ pub enum DatastoreError {
     #[error("Failed to read from datastore")]
     ReadDatastore,
 
+    #[error("Failed to read key '{key}' from datastore")]
+    ReadKeyValue { key: String },
+
     #[error("Failed to write to datastore as it is closed")]
     WriteToClosedDatastore,
 
     #[error("Failed to write to datastore")]
     WriteToDatastore,
+
+    #[error("Failed to write key '{key}' to datastore")]
+    WriteKeyValue { key: String },
 }
 
 impl ServicingError {

From 80272fbe1a2867ca34d227f3a0cb28f0c9d2ffbe Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Thu, 3 Sep 2026 23:53:46 +0000
Subject: [PATCH 02/14] datastore: fix DEFAULT typo and propagate keyvalue copy
 read errors

- Fix DEFALUT -> DEFAULT typo in hoststatus schema so timestamp
  auto-populates as intended.
- copy_key_values now returns an error instead of warning and
  silently stopping when reading a keyvalue row fails, so persist()
  cannot report success after a partial/failed copy.
---
 crates/trident/src/datastore.rs | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 1d8dba182..99d79c6c7 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -132,7 +132,7 @@ impl DataStore {
         db.execute(
             "CREATE TABLE IF NOT EXISTS hoststatus (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
-                timestamp DATETIME DEFALUT CURRENT_TIMESTAMP,
+                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                 contents TEXT NOT NULL
             )",
         )
@@ -185,11 +185,9 @@ impl DataStore {
             match query_statement.next() {
                 Ok(State::Done) => break,
                 Err(e) => {
-                    warn!(
-                        "Failed to get next keyvalue row while copying datastore: {:?}",
-                        e
-                    );
-                    break;
+                    return Err(e)
+                        .structured(ServicingError::from(DatastoreError::ReadDatastore))
+                        .message("Failed to get next keyvalue row while copying datastore");
                 }
                 Ok(State::Row) => {} // continue below
             }

From ef8b50eb29ce93e84533c09c91d49631159bf5dd Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 00:18:41 +0000
Subject: [PATCH 03/14] datastore: attach correlation ID before trident_start,
 including daemon paths

Trident::new emitted the "trident_start" metric before the CLI path
(main.rs) had retrieved the persisted correlation ID and attached it to
the shared TraceStream, so that very first startup event -- and any
daemon RPC handler that never ran the CLI's correlation-ID block at
all -- went out without it.

Move the correlation ID retrieval into Trident::new itself, using the
datastore_path it already receives, and set it on the TraceStream
before "trident_start" is emitted. Every caller of Trident::new (the
CLI path and each daemon gRPC service handler) now gets the same
treatment for free, since they all supply datastore_path.

main.rs's run_trident no longer needs its own post-hoc correlation-ID
block or the pre-emptive tracestream clone that existed only to work
around the ordering problem.

Note: multiboot installs still open the persisted datastore, then
swap in a fresh temporary one during install() (lib.rs) before the
new installation is later persisted. Whether the same correlation ID
should be carried forward across that swap (vs. each multiboot
install getting its own) is a servicing-flow behavior question left
as a follow-up rather than guessed at here.
---
 crates/trident/src/lib.rs  | 17 +++++++++++++++++
 crates/trident/src/main.rs | 26 +++++---------------------
 2 files changed, 22 insertions(+), 21 deletions(-)

diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs
index 8982b3f58..0dc52b4f7 100644
--- a/crates/trident/src/lib.rs
+++ b/crates/trident/src/lib.rs
@@ -227,6 +227,23 @@ impl Trident {
             ));
         }
 
+        // Retrieve (or create, on first run) this host's unique
+        // correlation ID from the datastore actually used for servicing,
+        // and attach it to the shared TraceStream before any startup
+        // metrics are emitted, so every trace/metric -- including this
+        // very "trident_start" event -- carries it. This runs for every
+        // caller of `Trident::new` (both the CLI path and each daemon
+        // RPC handler), since they all supply `datastore_path`.
+        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.correlation_id()) {
+            Ok(correlation_id) => {
+                info!("Correlation ID: {correlation_id}");
+                tracestream.set_correlation_id(correlation_id.to_string());
+            }
+            Err(e) => {
+                warn!("Failed to get or create correlation ID: {e:?}");
+            }
+        }
+
         tracing::info!(metric_name = "trident_start");
 
         Ok(Self {
diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs
index 7e121a9ff..bcd726845 100644
--- a/crates/trident/src/main.rs
+++ b/crates/trident/src/main.rs
@@ -2,7 +2,7 @@ use std::{fs, iter, panic, process::ExitCode};
 
 use anyhow::{Context, Error};
 use clap::Parser;
-use log::{error, info, warn, LevelFilter, Log};
+use log::{error, info, LevelFilter, Log};
 
 use osutils::logging::{filter::LogFilter, multilog::MultiLogger};
 use trident::{
@@ -157,11 +157,6 @@ fn run_trident(
                         .message("Datastore file does not exist");
                 }
 
-                // Clone the tracestream handle before it is moved into `Trident::new`,
-                // so the correlation ID can be attached to it below once the
-                // datastore has been opened.
-                let tracestream_for_correlation_id = tracestream.clone();
-
                 let mut trident = Trident::new(
                     config_path.map(HostConfigurationSource::File),
                     agent_config.datastore_path(),
@@ -170,24 +165,13 @@ fn run_trident(
                 )
                 .message("Failed to initialize Trident")?;
 
+                // `Trident::new` has already retrieved (or created) this
+                // host's persisted correlation ID and attached it to the
+                // shared TraceStream, so every trace/metric emitted from
+                // here on -- including "trident_start" -- carries it.
                 let mut datastore = DataStore::open_or_create(agent_config.datastore_path())
                     .message("Failed to open datastore")?;
 
-                // Retrieve (or create, on first run) this host's unique
-                // correlation ID from the datastore, and attach it to all
-                // subsequent traces/metrics so they can be correlated with
-                // this specific installation.
-                match datastore.correlation_id() {
-                    Ok(correlation_id) => {
-                        info!("Correlation ID: {correlation_id}");
-                        tracestream_for_correlation_id
-                            .set_correlation_id(correlation_id.to_string());
-                    }
-                    Err(e) => {
-                        warn!("Failed to get or create correlation ID: {e:?}");
-                    }
-                }
-
                 // Execute the command
                 let res = match args.command {
                     Commands::Install {

From fb57fdda37d1b513c013a2deb65948668411759d Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 00:49:00 +0000
Subject: [PATCH 04/14] datastore: apply schema migration on open, not just
 create

Fix two issues flagged by Copilot review:

- DataStore::open() (used for existing datastores) never ran the
  CREATE TABLE statements that make_datastore() runs on create, so an
  existing datastore created before the `keyvalue` table existed would
  fail correlation_id() with "no such table: keyvalue" on upgrade.
  Extract table creation into an idempotent ensure_schema() helper and
  call it from both open() and make_datastore(). Add a regression test
  covering a pre-existing datastore missing the keyvalue table.

- Trident::new() called hc.feature_tracing() (which emits the
  host_config_feature_usage tracing event) before retrieving and
  attaching the correlation ID, so that first metric was missing the
  field despite every other trace/metric carrying it. Move correlation
  ID retrieval earlier, before feature_tracing() and any other tracing
  event.
---
 crates/trident/src/datastore.rs | 46 ++++++++++++++++++++++++++++++++-
 crates/trident/src/lib.rs       | 34 ++++++++++++------------
 2 files changed, 62 insertions(+), 18 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 99d79c6c7..0182972c6 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -51,6 +51,11 @@ impl DataStore {
                 path: path.to_string_lossy().into(),
             },
         })?;
+        // Existing datastores may predate a table added in a later Trident
+        // version (e.g. `keyvalue`). Idempotently ensure the full schema is
+        // present so upgraded hosts don't fail with "no such table" the
+        // first time a new table is accessed.
+        Self::ensure_schema(&db)?;
         let host_status_yaml: Option<serde_yaml::Value> = db
             .prepare("SELECT contents FROM hoststatus ORDER BY id DESC LIMIT 1")
             .structured(ServicingError::Datastore {
@@ -129,6 +134,15 @@ impl DataStore {
 
         let db =
             sqlite::open(path).structured(ServicingError::from(DatastoreError::OpenDatastore))?;
+        Self::ensure_schema(&db)?;
+        Ok(db)
+    }
+
+    /// Idempotently create any tables that don't already exist. Safe to call
+    /// on both newly-created and pre-existing datastores, so that a
+    /// datastore created by an older Trident version picks up tables added
+    /// by a newer version the next time it is opened.
+    fn ensure_schema(db: &sqlite::Connection) -> Result<(), TridentError> {
         db.execute(
             "CREATE TABLE IF NOT EXISTS hoststatus (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -144,7 +158,7 @@ impl DataStore {
             )",
         )
         .structured(ServicingError::from(DatastoreError::InitializeDatastore))?;
-        Ok(db)
+        Ok(())
     }
 
     pub(crate) fn persist(&mut self, path: &Path) -> Result<(), TridentError> {
@@ -545,6 +559,36 @@ mod tests {
         temp_dir.close().unwrap();
     }
 
+    #[test]
+    /// Regression test: a datastore created by an older Trident version that
+    /// predates the `keyvalue` table (only `hoststatus` exists) must still
+    /// be usable after `open()` -- in particular, `correlation_id()` must
+    /// not fail with "no such table: keyvalue".
+    fn test_open_upgrades_pre_existing_datastore_schema() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let path = temp_dir.path().join("db.sqlite");
+
+        // Simulate a datastore created before the `keyvalue` table existed:
+        // create only the `hoststatus` table.
+        {
+            let db = sqlite::open(&path).unwrap();
+            db.execute(
+                "CREATE TABLE IF NOT EXISTS hoststatus (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    contents TEXT NOT NULL
+                )",
+            )
+            .unwrap();
+        }
+
+        let mut datastore = super::DataStore::open(&path).unwrap();
+        // Should not fail with "no such table: keyvalue".
+        datastore.correlation_id().unwrap();
+
+        temp_dir.close().unwrap();
+    }
+
     #[test]
     fn test_correlation_id_is_stable() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs
index 0dc52b4f7..7e7276632 100644
--- a/crates/trident/src/lib.rs
+++ b/crates/trident/src/lib.rs
@@ -207,6 +207,23 @@ impl Trident {
             info!("Running Trident in a container");
         }
 
+        // Retrieve (or create, on first run) this host's unique
+        // correlation ID from the datastore actually used for servicing,
+        // and attach it to the shared TraceStream before any startup
+        // metrics are emitted, so every trace/metric -- including this
+        // very "trident_start" event -- carries it. This runs for every
+        // caller of `Trident::new` (both the CLI path and each daemon
+        // RPC handler), since they all supply `datastore_path`.
+        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.correlation_id()) {
+            Ok(correlation_id) => {
+                info!("Correlation ID: {correlation_id}");
+                tracestream.set_correlation_id(correlation_id.to_string());
+            }
+            Err(e) => {
+                warn!("Failed to get or create correlation ID: {e:?}");
+            }
+        }
+
         // Trace features enabled in the Host Configuration.
         if let Some(hc) = &host_config {
             hc.feature_tracing();
@@ -227,23 +244,6 @@ impl Trident {
             ));
         }
 
-        // Retrieve (or create, on first run) this host's unique
-        // correlation ID from the datastore actually used for servicing,
-        // and attach it to the shared TraceStream before any startup
-        // metrics are emitted, so every trace/metric -- including this
-        // very "trident_start" event -- carries it. This runs for every
-        // caller of `Trident::new` (both the CLI path and each daemon
-        // RPC handler), since they all supply `datastore_path`.
-        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.correlation_id()) {
-            Ok(correlation_id) => {
-                info!("Correlation ID: {correlation_id}");
-                tracestream.set_correlation_id(correlation_id.to_string());
-            }
-            Err(e) => {
-                warn!("Failed to get or create correlation ID: {e:?}");
-            }
-        }
-
         tracing::info!(metric_name = "trident_start");
 
         Ok(Self {

From 45784206ea7e5ed463dcae225c8394f705cd2e46 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 01:14:25 +0000
Subject: [PATCH 05/14] datastore: eliminate correlation-ID race on concurrent
 first access

correlation_id() previously read the keyvalue table, and if absent,
generated a new UUID and wrote it unconditionally. Two connections
racing on first access could each generate a different UUID and the
second write would silently overwrite the first, leaving a caller who
already captured the first UUID tracing with an ID no longer persisted.

Fix: use an atomic INSERT ... ON CONFLICT DO NOTHING (set_value_if_absent)
to claim the row, then re-read it, so all racing callers converge on
whichever UUID actually got persisted. Also set a 5s SQLite busy timeout
on every connection open, since the atomic insert path can hit
SQLITE_BUSY under concurrent writes with the default 0ms timeout.

Adds a concurrency regression test (two threads, separate connections,
barrier-synchronized) asserting both observe the same correlation ID.

logging/tracestream: add a regression test asserting set_correlation_id
is actually copied into additional_fields on emitted trace/metric
entries, closing a gap where existing tests only checked metric_name/
value and would pass even if the correlation ID were dropped.
---
 crates/trident/src/datastore.rs           | 145 +++++++++++++++++++---
 crates/trident/src/logging/tracestream.rs |  42 +++++++
 2 files changed, 168 insertions(+), 19 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 0182972c6..f921b6876 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -46,11 +46,15 @@ impl DataStore {
 
     pub(crate) fn open(path: &Path) -> Result<Self, TridentError> {
         debug!("Loading datastore from {}", path.display());
-        let db = sqlite::open(path).structured(ServicingError::Datastore {
+        let mut db = sqlite::open(path).structured(ServicingError::Datastore {
             inner: DatastoreError::LoadDatastore {
                 path: path.to_string_lossy().into(),
             },
         })?;
+        // Multiple connections to the same datastore file (e.g. concurrent
+        // daemon RPC handlers) can briefly contend for the write lock; wait
+        // for it rather than failing immediately with "database is locked".
+        Self::set_busy_timeout(&mut db)?;
         // Existing datastores may predate a table added in a later Trident
         // version (e.g. `keyvalue`). Idempotently ensure the full schema is
         // present so upgraded hosts don't fail with "no such table" the
@@ -132,12 +136,24 @@ impl DataStore {
             DatastoreError::CreateDatastoreDirectory,
         ))?;
 
-        let db =
+        let mut db =
             sqlite::open(path).structured(ServicingError::from(DatastoreError::OpenDatastore))?;
+        Self::set_busy_timeout(&mut db)?;
         Self::ensure_schema(&db)?;
         Ok(db)
     }
 
+    /// Wait (rather than immediately failing with "database is locked") for
+    /// up to five seconds when another connection holds the write lock on
+    /// this datastore file. Needed because more than one connection to the
+    /// same datastore can legitimately exist at once (e.g. concurrent
+    /// daemon RPC handlers each calling `Trident::new`), and SQLite's
+    /// default busy timeout is zero.
+    fn set_busy_timeout(db: &mut sqlite::Connection) -> Result<(), TridentError> {
+        db.set_busy_timeout(5000)
+            .structured(ServicingError::from(DatastoreError::OpenDatastore))
+    }
+
     /// Idempotently create any tables that don't already exist. Safe to call
     /// on both newly-created and pre-existing datastores, so that a
     /// datastore created by an older Trident version picks up tables added
@@ -376,26 +392,67 @@ impl DataStore {
     /// Values are serialized as JSON, so any type implementing
     /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not
     /// just `HostStatus`.
+    ///
+    /// `correlation_id` is currently the only first-party caller of the
+    /// generic key-value store, and it needs insert-if-absent semantics
+    /// (see `set_value_if_absent`) rather than an unconditional overwrite,
+    /// so this unconditional-overwrite variant is presently exercised only
+    /// by tests. It's kept as part of the generic key-value API (see
+    /// `get_value`) for future callers that do want overwrite semantics.
+    #[allow(dead_code)]
     pub(crate) fn set_value<T: Serialize>(&self, key: &str, value: &T) -> Result<(), TridentError> {
+        let contents = serde_json::to_string(value).structured(InternalError::SerializeValue {
+            key: key.to_string(),
+        })?;
+        self.write_key_value_row(
+            key,
+            &contents,
+            "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
+             ON CONFLICT(key) DO UPDATE SET contents = excluded.contents",
+        )
+    }
+
+    /// Like [`Self::set_value`], but only inserts a row if `key` does not
+    /// already have one; an existing row is left untouched. Used where two
+    /// datastore connections could race to perform "first access"
+    /// initialization of a key (see [`Self::correlation_id`]): whichever
+    /// connection's insert commits first wins, and the other's insert
+    /// becomes a no-op instead of overwriting the winner's value.
+    pub(crate) fn set_value_if_absent<T: Serialize>(
+        &self,
+        key: &str,
+        value: &T,
+    ) -> Result<(), TridentError> {
+        let contents = serde_json::to_string(value).structured(InternalError::SerializeValue {
+            key: key.to_string(),
+        })?;
+        self.write_key_value_row(
+            key,
+            &contents,
+            "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
+             ON CONFLICT(key) DO NOTHING",
+        )
+    }
+
+    /// Execute a parameterized `INSERT ... ON CONFLICT ...` against the
+    /// `keyvalue` table, binding `key` and `contents` as the two `?`
+    /// placeholders in `sql`.
+    fn write_key_value_row(
+        &self,
+        key: &str,
+        contents: &str,
+        sql: &str,
+    ) -> Result<(), TridentError> {
         let db = self
             .db
             .as_ref()
             .structured(ServicingError::from(DatastoreError::WriteToClosedDatastore))?;
 
-        let contents = serde_json::to_string(value).structured(InternalError::SerializeValue {
-            key: key.to_string(),
+        let mut statement = db.prepare(sql).structured(ServicingError::Datastore {
+            inner: DatastoreError::WriteKeyValue {
+                key: key.to_string(),
+            },
         })?;
-
-        let mut statement = db
-            .prepare(
-                "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
-                 ON CONFLICT(key) DO UPDATE SET contents = excluded.contents",
-            )
-            .structured(ServicingError::Datastore {
-                inner: DatastoreError::WriteKeyValue {
-                    key: key.to_string(),
-                },
-            })?;
         statement
             .bind((1, key))
             .structured(ServicingError::Datastore {
@@ -404,7 +461,7 @@ impl DataStore {
                 },
             })?;
         statement
-            .bind((2, &*contents))
+            .bind((2, contents))
             .structured(ServicingError::Datastore {
                 inner: DatastoreError::WriteKeyValue {
                     key: key.to_string(),
@@ -426,14 +483,28 @@ impl DataStore {
     /// temporary-to-persistent transition performed by `persist`), and is
     /// intended to be attached to tracing/telemetry so that activity for a
     /// given host installation can be correlated across logs and traces.
+    ///
+    /// First access is not a simple read-then-write: `Trident::new` may be
+    /// invoked concurrently (e.g. by multiple daemon RPC handlers), each
+    /// opening its own connection to the same datastore file. A naive
+    /// "read, and if absent generate + write" would let two connections
+    /// both observe no row, generate different UUIDs, and each overwrite
+    /// the other -- leaving one caller tracing with an ID that was never
+    /// actually persisted. Instead, unconditionally attempt to insert a
+    /// freshly generated ID with `ON CONFLICT DO NOTHING` (a no-op if
+    /// another connection already inserted one first), then read back
+    /// whichever ID actually won that race.
     pub fn correlation_id(&mut self) -> Result<Uuid, TridentError> {
         if let Some(id) = self.get_value::<Uuid>(CORRELATION_ID_KEY)? {
             return Ok(id);
         }
 
-        let id = Uuid::new_v4();
-        self.set_value(CORRELATION_ID_KEY, &id)?;
-        Ok(id)
+        self.set_value_if_absent(CORRELATION_ID_KEY, &Uuid::new_v4())?;
+
+        self.get_value::<Uuid>(CORRELATION_ID_KEY)?
+            .structured(InternalError::Internal(
+                "Correlation ID missing immediately after being inserted",
+            ))
     }
 
     /// Parse a single HostStatus entry from a datastore query result.
@@ -589,6 +660,42 @@ mod tests {
         temp_dir.close().unwrap();
     }
 
+    #[test]
+    fn test_correlation_id_concurrent_first_access_is_consistent() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let path = temp_dir.path().join("db.sqlite");
+
+        // Create the datastore (and its schema) up front, then open two
+        // separate connections to it, simulating two daemon RPC handlers
+        // concurrently calling `Trident::new` (and therefore
+        // `correlation_id`) against the same datastore path.
+        super::DataStore::make_datastore(&path).unwrap();
+
+        let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
+        let handles: Vec<_> = (0..2)
+            .map(|_| {
+                let path = path.clone();
+                let barrier = barrier.clone();
+                std::thread::spawn(move || {
+                    let mut datastore = super::DataStore::open(&path).unwrap();
+                    // Synchronize so both threads attempt "first access"
+                    // (no correlation ID persisted yet) as close together
+                    // as possible.
+                    barrier.wait();
+                    datastore.correlation_id().unwrap()
+                })
+            })
+            .collect();
+
+        let ids: Vec<uuid::Uuid> = handles.into_iter().map(|h| h.join().unwrap()).collect();
+        assert_eq!(
+            ids[0], ids[1],
+            "concurrent first access returned inconsistent correlation IDs"
+        );
+
+        temp_dir.close().unwrap();
+    }
+
     #[test]
     fn test_correlation_id_is_stable() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index 20eb511fa..bc7919950 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -559,6 +559,48 @@ mod functional_test {
         );
     }
 
+    #[functional_test]
+    /// Regression test: `TraceStream::set_correlation_id` must actually
+    /// reach the serialized trace entry's `additional_fields.correlation_id`
+    /// -- the existing metric/span tests only assert on `metric_name`/
+    /// `value` and would still pass even if the correlation ID were never
+    /// copied into `additional_fields`.
+    fn test_tracestream_correlation_id_written_to_additional_fields() {
+        let tracestream = TraceStream::default();
+        tracestream.set_correlation_id("test-correlation-id".to_string());
+        let trace_sender = tracestream
+            .make_trace_sender()
+            .with_filter(filter::LevelFilter::INFO);
+
+        tracing::subscriber::set_global_default(
+            tracing_subscriber::Registry::default().with(trace_sender),
+        )
+        .context("Failed to set global default subscriber")
+        .unwrap();
+
+        tracing::info!(
+            metric_name = "test_metric_with_correlation_id",
+            value = true
+        );
+
+        // Ensure the trace system has time to write the file.
+        std::thread::sleep(std::time::Duration::from_millis(100));
+
+        let file = File::open(TRIDENT_METRICS_FILE_PATH).unwrap();
+        let reader = BufReader::new(file);
+        let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
+
+        let metric_found = lines.iter().any(|line| {
+            line.contains(r#""metric_name":"test_metric_with_correlation_id""#)
+                && line.contains(r#""correlation_id":"test-correlation-id""#)
+        });
+
+        assert!(
+            metric_found,
+            "Expected metric with correlation_id field not found in the local metrics file"
+        );
+    }
+
     #[functional_test]
     fn test_populate_additional_fields() {
         let additional_fields = populate_additional_fields();

From 864ee32788e4382f4e820787be6f49e2ea985033 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 01:27:33 +0000
Subject: [PATCH 06/14] datastore: fix self-persist deadlock in copy_key_values

copy_key_values kept the source SELECT statement active while writing
to the destination connection. persist() supports a destination path
equal to the currently-open (temporary) datastores own path -- the
offline provisioning flow does this -- so source and destination can
be two live connections to the same underlying file. SQLite locking
is per-connection, so the destination write would wait on the sources
still-active read lock, surfacing as "database is locked" (bounded
only by the busy timeout, not resolved by it).

Fix: fully read and finalize the source query before issuing any
writes to the destination. Adds a regression test that persists a
temporary datastore to its own path.
---
 crates/trident/src/datastore.rs | 81 +++++++++++++++++++++++++--------
 1 file changed, 63 insertions(+), 18 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index f921b6876..a1009ec07 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -203,32 +203,51 @@ impl DataStore {
     /// Copy all rows of the generic key-value table from `source` into
     /// `destination`, overwriting any conflicting keys already present in
     /// `destination`.
+    ///
+    /// `source` and `destination` may be two live connections to the *same*
+    /// underlying SQLite file (e.g. an offline `persist` whose destination
+    /// path is the currently-open datastore's own path). SQLite's locking is
+    /// per-connection, so a `SELECT` left active on `source` holds a read
+    /// lock that a write from `destination` on the same file would have to
+    /// wait on -- and since both connections are driven from this single
+    /// thread, that wait can never be satisfied ("database is locked").
+    /// To avoid this, fully read and finalize the source query *before*
+    /// issuing any writes to `destination`.
     fn copy_key_values(
         source: &sqlite::Connection,
         destination: &sqlite::Connection,
     ) -> Result<(), TridentError> {
-        let mut query_statement = source
-            .prepare("SELECT key, contents FROM keyvalue")
-            .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+        let mut rows: Vec<(String, String)> = Vec::new();
+        {
+            let mut query_statement = source
+                .prepare("SELECT key, contents FROM keyvalue")
+                .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
 
-        loop {
-            match query_statement.next() {
-                Ok(State::Done) => break,
-                Err(e) => {
-                    return Err(e)
-                        .structured(ServicingError::from(DatastoreError::ReadDatastore))
-                        .message("Failed to get next keyvalue row while copying datastore");
+            loop {
+                match query_statement.next() {
+                    Ok(State::Done) => break,
+                    Err(e) => {
+                        return Err(e)
+                            .structured(ServicingError::from(DatastoreError::ReadDatastore))
+                            .message("Failed to get next keyvalue row while copying datastore");
+                    }
+                    Ok(State::Row) => {} // continue below
                 }
-                Ok(State::Row) => {} // continue below
-            }
 
-            let key = query_statement
-                .read::<String, _>(0)
-                .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
-            let contents = query_statement
-                .read::<String, _>(1)
-                .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+                let key = query_statement
+                    .read::<String, _>(0)
+                    .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+                let contents = query_statement
+                    .read::<String, _>(1)
+                    .structured(ServicingError::from(DatastoreError::ReadDatastore))?;
+
+                rows.push((key, contents));
+            }
+            // `query_statement` is dropped here, finalizing the source
+            // SELECT and releasing its read lock before any writes below.
+        }
 
+        for (key, contents) in rows {
             let mut insert_statement = destination
                 .prepare(
                     "INSERT INTO keyvalue (key, contents) VALUES (?, ?) \
@@ -790,6 +809,32 @@ mod functional_test {
         );
     }
 
+    #[functional_test]
+    /// Regression test: `persist` supports a destination path equal to the
+    /// currently-open (temporary) datastore's own path -- the offline
+    /// provisioning flow does this. Before the fix, `copy_key_values` kept
+    /// the source `SELECT` active while writing to the destination
+    /// connection, and since both connections point at the same file, the
+    /// destination write would block on the source's read lock forever
+    /// (mitigated only by the busy timeout, so this would previously fail
+    /// with "database is locked" rather than deadlock outright).
+    fn test_persist_to_same_path_does_not_deadlock() {
+        let temp_dir = TempDir::new().unwrap();
+        let datastore_path = temp_dir.path().join("db.sqlite");
+
+        let mut datastore = DataStore::open_or_create(&datastore_path).unwrap();
+        let correlation_id = datastore.correlation_id().unwrap();
+
+        // Persist to the exact same path the datastore is currently open at.
+        datastore.persist(&datastore_path).unwrap();
+
+        assert_eq!(
+            datastore.correlation_id().unwrap(),
+            correlation_id,
+            "Correlation ID should survive a self-persist"
+        );
+    }
+
     #[functional_test]
     fn test_correlation_id_survives_persist() {
         let temp_dir = TempDir::new().unwrap();

From e0ec5459d2fab1e1ce967385233a940900a0ece2 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 01:49:23 +0000
Subject: [PATCH 07/14] logging/datastore: convert host-independent functional
 tests to plain tests

Several tests marked #[functional_test] (VM-only) did not actually need
a VM: they only touched a TempDir-based SQLite file, or wrote to
/var/log/trident-metrics.jsonl only because that path was hardcoded
into TraceSender::new, not because the test logic itself needed a real
host path. Regular #[test]s are faster and can run directly on the
host, so:

- Add TraceStream::make_trace_sender_with_metrics_path so the local
  metrics file location is injectable. Production (main.rs) still
  goes through TRIDENT_METRICS_FILE_PATH via make_trace_sender();
  tests now point it at a throwaway temp file.
- Move test_tracestream_write_metric_event_to_file,
  test_tracestream_write_span_metric_to_file, and
  test_tracestream_correlation_id_written_to_additional_fields (and
  fix up test_tracestream/test_lock, which already silently touched
  the real path) from #[functional_test] to #[test]. Switch these from
  tracing::subscriber::set_global_default (process-wide, once-only --
  conflicts across tests sharing a process) to set_default (thread-
  local, scoped to a guard), since they can now run alongside other
  tests in the same process. test_populate_additional_fields and
  test_populate_platform_info stay functional tests: they assert
  against the real hosts hardware/platform info, which cannot be
  faked.
- Move test_persist_to_same_path_does_not_deadlock (added this
  session) from #[functional_test] to #[test]: it only used a TempDir
  SQLite file and never needed VM isolation.
---
 crates/trident/src/datastore.rs           |  52 +++----
 crates/trident/src/logging/tracestream.rs | 179 +++++++++++++---------
 2 files changed, 134 insertions(+), 97 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index a1009ec07..f2ebaac1b 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -583,6 +583,32 @@ mod tests {
         temp_dir.close().unwrap();
     }
 
+    #[test]
+    /// Regression test: `persist` supports a destination path equal to the
+    /// currently-open (temporary) datastore's own path -- the offline
+    /// provisioning flow does this. Before the fix, `copy_key_values` kept
+    /// the source `SELECT` active while writing to the destination
+    /// connection, and since both connections point at the same file, the
+    /// destination write would block on the source's read lock forever
+    /// (mitigated only by the busy timeout, so this would previously fail
+    /// with "database is locked" rather than deadlock outright).
+    fn test_persist_to_same_path_does_not_deadlock() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let datastore_path = temp_dir.path().join("db.sqlite");
+
+        let mut datastore = super::DataStore::open_or_create(&datastore_path).unwrap();
+        let correlation_id = datastore.correlation_id().unwrap();
+
+        // Persist to the exact same path the datastore is currently open at.
+        datastore.persist(&datastore_path).unwrap();
+
+        assert_eq!(
+            datastore.correlation_id().unwrap(),
+            correlation_id,
+            "Correlation ID should survive a self-persist"
+        );
+    }
+
     #[test]
     fn test_parse_host_status() {
         let ds = super::DataStore {
@@ -809,32 +835,6 @@ mod functional_test {
         );
     }
 
-    #[functional_test]
-    /// Regression test: `persist` supports a destination path equal to the
-    /// currently-open (temporary) datastore's own path -- the offline
-    /// provisioning flow does this. Before the fix, `copy_key_values` kept
-    /// the source `SELECT` active while writing to the destination
-    /// connection, and since both connections point at the same file, the
-    /// destination write would block on the source's read lock forever
-    /// (mitigated only by the busy timeout, so this would previously fail
-    /// with "database is locked" rather than deadlock outright).
-    fn test_persist_to_same_path_does_not_deadlock() {
-        let temp_dir = TempDir::new().unwrap();
-        let datastore_path = temp_dir.path().join("db.sqlite");
-
-        let mut datastore = DataStore::open_or_create(&datastore_path).unwrap();
-        let correlation_id = datastore.correlation_id().unwrap();
-
-        // Persist to the exact same path the datastore is currently open at.
-        datastore.persist(&datastore_path).unwrap();
-
-        assert_eq!(
-            datastore.correlation_id().unwrap(),
-            correlation_id,
-            "Correlation ID should survive a self-persist"
-        );
-    }
-
     #[functional_test]
     fn test_correlation_id_survives_persist() {
         let temp_dir = TempDir::new().unwrap();
diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index bc7919950..c69bdeb8a 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -142,9 +142,23 @@ impl TraceStream {
 
     /// Create a Boxed TraceSender
     pub fn make_trace_sender(&self) -> Box<TraceSender> {
+        self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH)
+    }
+
+    /// Like `make_trace_sender`, but writes the local metrics file to
+    /// `metrics_file_path` instead of the real host path
+    /// (`TRIDENT_METRICS_FILE_PATH`). This lets tests exercise the full
+    /// metrics-writing pipeline against a throwaway temp file instead of a
+    /// real, shared host path, so they can be plain `#[test]`s instead of
+    /// needing a VM.
+    pub(crate) fn make_trace_sender_with_metrics_path(
+        &self,
+        metrics_file_path: &str,
+    ) -> Box<TraceSender> {
         Box::new(TraceSender::new(
             self.target.clone(),
             self.correlation_id.clone(),
+            metrics_file_path,
         ))
     }
 }
@@ -166,12 +180,13 @@ impl TraceSender {
     fn new(
         server: Arc<RwLock<Option<String>>>,
         correlation_id: Arc<RwLock<Option<String>>>,
+        metrics_file_path: &str,
     ) -> Self {
         Self {
             server,
             correlation_id,
             client: reqwest::blocking::Client::new(),
-            metrics_file: match files::create_file(TRIDENT_METRICS_FILE_PATH) {
+            metrics_file: match files::create_file(metrics_file_path) {
                 Ok(f) => Some(f),
                 Err(err) => {
                     eprintln!(
@@ -454,12 +469,20 @@ fn populate_platform_info() -> BTreeMap<String, Value> {
 mod tests {
     use super::*;
 
-    use std::{fs::File, io::Write};
+    use std::{
+        fs::File,
+        io::{BufRead, BufReader, Write},
+    };
+
+    use tracing_subscriber::{filter, layer::SubscriberExt};
 
     #[test]
     fn test_tracestream() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let metrics_path = temp_dir.path().join("metrics.jsonl");
         let tracestream = TraceStream::default();
-        let trace_sender = tracestream.make_trace_sender();
+        let trace_sender =
+            tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap());
         assert!(
             trace_sender.get_server().is_none(),
             "tracestream should not have a server"
@@ -478,8 +501,11 @@ mod tests {
 
     #[test]
     fn test_lock() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let metrics_path = temp_dir.path().join("metrics.jsonl");
         let mut tracestream = TraceStream::default();
-        let trace_sender = tracestream.make_trace_sender();
+        let trace_sender =
+            tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap());
 
         assert!(
             trace_sender.get_server().is_none(),
@@ -513,31 +539,24 @@ mod tests {
         let uuid = read_product_uuid(filepath.to_str().unwrap().to_string());
         assert_eq!(uuid, "test_uuid");
     }
-}
-
-#[cfg(feature = "functional-test")]
-#[cfg_attr(not(test), allow(unused_imports, dead_code))]
-mod functional_test {
-    use super::*;
 
-    use std::io::{BufRead, BufReader};
-
-    use tracing_subscriber::{filter, layer::SubscriberExt};
-
-    use pytest_gen::functional_test;
-
-    #[functional_test]
+    #[test]
     fn test_tracestream_write_metric_event_to_file() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let metrics_path = temp_dir.path().join("metrics.jsonl");
         let tracestream = TraceStream::default();
         let trace_sender = tracestream
-            .make_trace_sender()
+            .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap())
             .with_filter(filter::LevelFilter::INFO);
 
-        tracing::subscriber::set_global_default(
+        // Use a thread-local scoped default subscriber (rather than
+        // `set_global_default`) since this is a plain `#[test]` that may run
+        // concurrently with other tests in the same process -- the global
+        // default can only be set once per process, but the scoped default
+        // is per-thread and automatically restored when `_guard` drops.
+        let _guard = tracing::subscriber::set_default(
             tracing_subscriber::Registry::default().with(trace_sender),
-        )
-        .context("Failed to set global default subscriber")
-        .unwrap();
+        );
 
         tracing::info!(metric_name = "test_metric", value = true);
 
@@ -545,7 +564,7 @@ mod functional_test {
         std::thread::sleep(std::time::Duration::from_millis(100));
 
         // Check if the specific metric exists in the file.
-        let file = File::open(TRIDENT_METRICS_FILE_PATH).unwrap();
+        let file = File::open(&metrics_path).unwrap();
         let reader = BufReader::new(file);
         let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
 
@@ -559,24 +578,26 @@ mod functional_test {
         );
     }
 
-    #[functional_test]
+    #[test]
     /// Regression test: `TraceStream::set_correlation_id` must actually
     /// reach the serialized trace entry's `additional_fields.correlation_id`
-    /// -- the existing metric/span tests only assert on `metric_name`/
-    /// `value` and would still pass even if the correlation ID were never
-    /// copied into `additional_fields`.
+    /// -- the metric/span tests above only assert on `metric_name`/`value`
+    /// and would still pass even if the correlation ID were never copied
+    /// into `additional_fields`.
     fn test_tracestream_correlation_id_written_to_additional_fields() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let metrics_path = temp_dir.path().join("metrics.jsonl");
         let tracestream = TraceStream::default();
         tracestream.set_correlation_id("test-correlation-id".to_string());
         let trace_sender = tracestream
-            .make_trace_sender()
+            .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap())
             .with_filter(filter::LevelFilter::INFO);
 
-        tracing::subscriber::set_global_default(
+        // See test_tracestream_write_metric_event_to_file for why a scoped
+        // (not global) default subscriber is used here.
+        let _guard = tracing::subscriber::set_default(
             tracing_subscriber::Registry::default().with(trace_sender),
-        )
-        .context("Failed to set global default subscriber")
-        .unwrap();
+        );
 
         tracing::info!(
             metric_name = "test_metric_with_correlation_id",
@@ -586,7 +607,7 @@ mod functional_test {
         // Ensure the trace system has time to write the file.
         std::thread::sleep(std::time::Duration::from_millis(100));
 
-        let file = File::open(TRIDENT_METRICS_FILE_PATH).unwrap();
+        let file = File::open(&metrics_path).unwrap();
         let reader = BufReader::new(file);
         let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
 
@@ -601,6 +622,60 @@ mod functional_test {
         );
     }
 
+    #[test]
+    fn test_tracestream_write_span_metric_to_file() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let metrics_path = temp_dir.path().join("metrics.jsonl");
+        let tracestream = TraceStream::default();
+        let trace_sender = tracestream
+            .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap())
+            .with_filter(filter::LevelFilter::INFO);
+
+        // See test_tracestream_write_metric_event_to_file for why a scoped
+        // (not global) default subscriber is used here.
+        let _guard = tracing::subscriber::set_default(
+            tracing_subscriber::Registry::default().with(trace_sender),
+        );
+
+        // Call test function that will create a span
+        simulate_function_span();
+
+        // Ensure the trace system has time to simulate a span.
+        std::thread::sleep(std::time::Duration::from_millis(100));
+
+        // Check if the specific metric exists in the file.
+        let file = File::open(&metrics_path).unwrap();
+        let reader = BufReader::new(file);
+        let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
+
+        let expected_substring = r#""metric_name":"test_span"#;
+        let span_metric_found = lines.iter().any(|line| line.contains(expected_substring));
+
+        // Assert that the expected metric is present in the file.
+        assert!(
+            span_metric_found,
+            "Expected test metric not found in the local metrics file"
+        );
+    }
+
+    // Helper function to test span metrics
+    #[tracing::instrument(name = "test_span", skip_all)]
+    fn simulate_function_span() {}
+}
+
+#[cfg(feature = "functional-test")]
+#[cfg_attr(not(test), allow(unused_imports, dead_code))]
+mod functional_test {
+    use super::*;
+
+    use pytest_gen::functional_test;
+
+    // These two remain functional tests (VM-only) because they assert
+    // against the actual host's hardware/platform info (CPU count, memory,
+    // product UUID, os-release, kernel version) -- unlike the metrics-file
+    // tests above, there's no way to inject a fake value here, so the
+    // result is inherently host-dependent.
+
     #[functional_test]
     fn test_populate_additional_fields() {
         let additional_fields = populate_additional_fields();
@@ -634,42 +709,4 @@ mod functional_test {
             "Platform info does not match the expected result"
         );
     }
-
-    #[functional_test]
-    fn test_tracestream_write_span_metric_to_file() {
-        let tracestream = TraceStream::default();
-        let trace_sender = tracestream
-            .make_trace_sender()
-            .with_filter(filter::LevelFilter::INFO);
-
-        tracing::subscriber::set_global_default(
-            tracing_subscriber::Registry::default().with(trace_sender),
-        )
-        .context("Failed to set global default subscriber")
-        .unwrap();
-
-        // Call test function that will create a span
-        simulate_function_span();
-
-        // Ensure the trace system has time to simulate a span.
-        std::thread::sleep(std::time::Duration::from_millis(100));
-
-        // Check if the specific metric exists in the file.
-        let file = File::open(TRIDENT_METRICS_FILE_PATH).unwrap();
-        let reader = BufReader::new(file);
-        let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
-
-        let expected_substring = r#""metric_name":"test_span"#;
-        let span_metric_found = lines.iter().any(|line| line.contains(expected_substring));
-
-        // Assert that the expected metric is present in the file.
-        assert!(
-            span_metric_found,
-            "Expected test metric not found in the local metrics file"
-        );
-    }
-
-    // Helper function to test span metrics
-    #[tracing::instrument(name = "test_span", skip_all)]
-    fn simulate_function_span() {}
 }

From 91a54449fbcf74f05f66635f340bb44e7da9a006 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Fri, 4 Sep 2026 02:48:59 +0000
Subject: [PATCH 08/14] management: merge DatastorePath into existing agent
 config missing it

configure_agent_config previously treated an existing agent config file
with no DatastorePath= line as implicitly configured to the default
path, erroring with ImageBadAgentConfiguration if a non-default path
was actually expected. This agent config file can now also carry a
Telemetry= line (see this PR), so a telemetry-only config -- entirely
plausible once operators start pre-populating it to opt in -- would
incorrectly break non-default datastore setups. Missing DatastorePath=
still only means "use the default"; when a non-default path is
expected, merge a DatastorePath= line into the existing file instead
(preserving Telemetry= and any other existing lines), keeping the same
root-verity restriction as the file-does-not-exist case. Add a
regression test.
---
 crates/trident/src/subsystems/management.rs | 110 ++++++++++++++++++--
 1 file changed, 99 insertions(+), 11 deletions(-)

diff --git a/crates/trident/src/subsystems/management.rs b/crates/trident/src/subsystems/management.rs
index dbe408d2f..1a6e3f39f 100644
--- a/crates/trident/src/subsystems/management.rs
+++ b/crates/trident/src/subsystems/management.rs
@@ -96,24 +96,57 @@ fn configure_agent_config(
     if Path::new(agent_config_path).exists() {
         // If the agent config exists, check that the datastore matches the expected path.
         if let Ok(contents) = std::fs::read_to_string(agent_config_path) {
+            let mut datastore_path_line_present = false;
             let mut datastore_path_configured = TRIDENT_DATASTORE_PATH_DEFAULT;
             for line in contents.lines() {
                 if let Some(path) = line.strip_prefix("DatastorePath=") {
+                    datastore_path_line_present = true;
                     datastore_path_configured = path.trim();
                     break;
                 }
             }
-            // If the datastore path in the agent config does not match the expected path,
-            // return an error.
-            if datastore_path != Path::new(datastore_path_configured) {
-                return Err(TridentError::new(
-                    InvalidInputError::ImageBadAgentConfiguration,
-                ))
-                .message(format!(
-                    "Datastore path in agent config ({}) does not match expected path ({})",
-                    datastore_path_configured,
-                    datastore_path.display()
-                ));
+
+            if datastore_path_line_present {
+                // An explicit DatastorePath= line is present: it must match
+                // the expected path exactly.
+                if datastore_path != Path::new(datastore_path_configured) {
+                    return Err(TridentError::new(
+                        InvalidInputError::ImageBadAgentConfiguration,
+                    ))
+                    .message(format!(
+                        "Datastore path in agent config ({}) does not match expected path ({})",
+                        datastore_path_configured,
+                        datastore_path.display()
+                    ));
+                }
+            } else if datastore_path != Path::new(TRIDENT_DATASTORE_PATH_DEFAULT) {
+                // No DatastorePath= line: the file may still carry other
+                // settings (e.g. Telemetry=) that must be preserved. Missing
+                // DatastorePath only implies the default path, so if a
+                // non-default path is expected, merge a DatastorePath= line
+                // into the existing file rather than treating it as a
+                // mismatch.
+                if is_root_verity {
+                    // For root-verity, do not attempt to modify the agent config.
+                    return Err(TridentError::new(
+                        InvalidInputError::ImageBadAgentConfiguration,
+                    ))
+                    .message(
+                        "Agent configuration file does not set a non-default datastore path \
+                         and root filesystem is verity",
+                    );
+                }
+
+                let mut updated_contents = contents;
+                if !updated_contents.is_empty() && !updated_contents.ends_with('\n') {
+                    updated_contents.push('\n');
+                }
+                updated_contents.push_str(&format!("DatastorePath={}\n", datastore_path.display()));
+                fs::write(agent_config_path, updated_contents).structured(
+                    ServicingError::CreateConfigurationFile {
+                        path: agent_config_path.into(),
+                    },
+                )?;
             }
         }
     } else if datastore_path != Path::new(TRIDENT_DATASTORE_PATH_DEFAULT) {
@@ -270,5 +303,60 @@ mod tests {
             )
             .unwrap_err();
         }
+
+        {
+            // Regression test: agent config exists but only carries other
+            // settings (e.g. Telemetry=OptIn), with no DatastorePath= line.
+            // A non-default datastore path must be merged in, preserving the
+            // existing settings, rather than treated as a mismatch.
+            let agent_config_folder = tempfile::tempdir().unwrap();
+            let agent_config_path = agent_config_folder.path().join("trident.conf");
+            fs::write(
+                &agent_config_path,
+                "Telemetry=OptIn
+",
+            )
+            .unwrap();
+
+            configure_agent_config(
+                &agent_config_path.to_string_lossy(),
+                Path::new(nonstandard_datastore_path),
+                false,
+            )
+            .unwrap();
+
+            let contents = std::fs::read_to_string(&agent_config_path).unwrap();
+            assert!(contents.contains("Telemetry=OptIn"));
+            assert!(contents.contains(&format!("DatastorePath={nonstandard_datastore_path}")));
+        }
+
+        {
+            // Same as above, but root-verity: must not attempt to modify
+            // the agent config, and must error like the "file does not
+            // exist" root-verity case.
+            let agent_config_folder = tempfile::tempdir().unwrap();
+            let agent_config_path = agent_config_folder.path().join("trident.conf");
+            fs::write(
+                &agent_config_path,
+                "Telemetry=OptIn
+",
+            )
+            .unwrap();
+
+            configure_agent_config(
+                &agent_config_path.to_string_lossy(),
+                Path::new(nonstandard_datastore_path),
+                true,
+            )
+            .unwrap_err();
+
+            // The file must be left untouched.
+            let contents = std::fs::read_to_string(&agent_config_path).unwrap();
+            assert_eq!(
+                contents,
+                "Telemetry=OptIn
+"
+            );
+        }
     }
 }

From d33e2eb09c9c4e1fb03d3256fc9df4ad19e9bbe2 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Mon, 7 Sep 2026 18:38:18 +0000
Subject: [PATCH 09/14] datastore: describe current copy_key_values behavior in
 test doc comment

The regression test's doc comment referenced "before the fix" framing
describing removed behavior. Rewritten to describe only the current
implementation: copy_key_values fully reads and finalizes the source
SELECT before writing to the destination connection.
---
 crates/trident/src/datastore.rs | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index f2ebaac1b..7f71885fa 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -586,12 +586,11 @@ mod tests {
     #[test]
     /// Regression test: `persist` supports a destination path equal to the
     /// currently-open (temporary) datastore's own path -- the offline
-    /// provisioning flow does this. Before the fix, `copy_key_values` kept
-    /// the source `SELECT` active while writing to the destination
-    /// connection, and since both connections point at the same file, the
-    /// destination write would block on the source's read lock forever
-    /// (mitigated only by the busy timeout, so this would previously fail
-    /// with "database is locked" rather than deadlock outright).
+    /// provisioning flow does this. `copy_key_values` fully reads and
+    /// finalizes the source `SELECT` before writing to the destination
+    /// connection, so a self-persist (both connections pointing at the
+    /// same file) does not block the destination write on the source's
+    /// read lock.
     fn test_persist_to_same_path_does_not_deadlock() {
         let temp_dir = tempfile::tempdir().unwrap();
         let datastore_path = temp_dir.path().join("db.sqlite");

From 7d2a5b1d6eaa67e922c639c7d10d76473f4bdc26 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Mon, 7 Sep 2026 19:04:13 +0000
Subject: [PATCH 10/14] management: unify agent-config datastore-path merge
 logic

configure_agent_config repeated the same "datastore_path is non-default"
check in two branches: agent config exists without a DatastorePath= line,
and agent config doesn't exist at all. Both cases end up doing the same
thing -- merge a DatastorePath= line into whatever contents already exist
(empty, if the file didn't exist) -- so they're now handled by one shared
code path instead of two copies of the check, the root-verity guard, and
the write.
---
 crates/trident/src/subsystems/management.rs | 111 +++++++++-----------
 1 file changed, 51 insertions(+), 60 deletions(-)

diff --git a/crates/trident/src/subsystems/management.rs b/crates/trident/src/subsystems/management.rs
index 1a6e3f39f..ab66e65aa 100644
--- a/crates/trident/src/subsystems/management.rs
+++ b/crates/trident/src/subsystems/management.rs
@@ -92,76 +92,67 @@ fn configure_agent_config(
     datastore_path: &Path,
     is_root_verity: bool,
 ) -> Result<(), TridentError> {
-    // Ensure that Trident agent config exists with correct datastore path
-    if Path::new(agent_config_path).exists() {
-        // If the agent config exists, check that the datastore matches the expected path.
-        if let Ok(contents) = std::fs::read_to_string(agent_config_path) {
-            let mut datastore_path_line_present = false;
-            let mut datastore_path_configured = TRIDENT_DATASTORE_PATH_DEFAULT;
-            for line in contents.lines() {
-                if let Some(path) = line.strip_prefix("DatastorePath=") {
-                    datastore_path_line_present = true;
-                    datastore_path_configured = path.trim();
-                    break;
-                }
-            }
+    // Ensure that Trident agent config exists with correct datastore path.
+    // A missing file and an existing-but-unreadable file are both treated
+    // as "no contents to preserve" -- except an unreadable file is left
+    // untouched entirely, matching the prior behavior of silently skipping
+    // the whole check when the file exists but can't be read.
+    let contents = if Path::new(agent_config_path).exists() {
+        match std::fs::read_to_string(agent_config_path) {
+            Ok(contents) => contents,
+            Err(_) => return Ok(()),
+        }
+    } else {
+        String::new()
+    };
 
-            if datastore_path_line_present {
-                // An explicit DatastorePath= line is present: it must match
-                // the expected path exactly.
-                if datastore_path != Path::new(datastore_path_configured) {
-                    return Err(TridentError::new(
-                        InvalidInputError::ImageBadAgentConfiguration,
-                    ))
-                    .message(format!(
-                        "Datastore path in agent config ({}) does not match expected path ({})",
-                        datastore_path_configured,
-                        datastore_path.display()
-                    ));
-                }
-            } else if datastore_path != Path::new(TRIDENT_DATASTORE_PATH_DEFAULT) {
-                // No DatastorePath= line: the file may still carry other
-                // settings (e.g. Telemetry=) that must be preserved. Missing
-                // DatastorePath only implies the default path, so if a
-                // non-default path is expected, merge a DatastorePath= line
-                // into the existing file rather than treating it as a
-                // mismatch.
-                if is_root_verity {
-                    // For root-verity, do not attempt to modify the agent config.
-                    return Err(TridentError::new(
-                        InvalidInputError::ImageBadAgentConfiguration,
-                    ))
-                    .message(
-                        "Agent configuration file does not set a non-default datastore path \
-                         and root filesystem is verity",
-                    );
-                }
+    let mut datastore_path_line_present = false;
+    let mut datastore_path_configured = TRIDENT_DATASTORE_PATH_DEFAULT;
+    for line in contents.lines() {
+        if let Some(path) = line.strip_prefix("DatastorePath=") {
+            datastore_path_line_present = true;
+            datastore_path_configured = path.trim();
+            break;
+        }
+    }
 
-                let mut updated_contents = contents;
-                if !updated_contents.is_empty() && !updated_contents.ends_with('\n') {
-                    updated_contents.push('\n');
-                }
-                updated_contents.push_str(&format!("DatastorePath={}\n", datastore_path.display()));
-                fs::write(agent_config_path, updated_contents).structured(
-                    ServicingError::CreateConfigurationFile {
-                        path: agent_config_path.into(),
-                    },
-                )?;
-            }
+    if datastore_path_line_present {
+        // An explicit DatastorePath= line is present: it must match
+        // the expected path exactly.
+        if datastore_path != Path::new(datastore_path_configured) {
+            return Err(TridentError::new(
+                InvalidInputError::ImageBadAgentConfiguration,
+            ))
+            .message(format!(
+                "Datastore path in agent config ({}) does not match expected path ({})",
+                datastore_path_configured,
+                datastore_path.display()
+            ));
         }
     } else if datastore_path != Path::new(TRIDENT_DATASTORE_PATH_DEFAULT) {
-        // Only attempt to create the agent config if the datastore path is not the default.
-
+        // No DatastorePath= line -- either the file doesn't exist yet, or
+        // it exists but carries other settings (e.g. Telemetry=) that must
+        // be preserved. Missing DatastorePath only implies the default
+        // path, so if a non-default path is expected, merge a
+        // DatastorePath= line into whatever contents already exist (empty,
+        // if the file didn't exist) rather than treating it as a mismatch.
         if is_root_verity {
-            // For root-verity, do not attempt to create the agent config.
+            // For root-verity, do not attempt to create or modify the agent config.
             return Err(TridentError::new(
                 InvalidInputError::ImageBadAgentConfiguration,
             ))
-            .message("Agent configuration file does not exist and root filesystem is verity");
+            .message(
+                "Agent configuration file does not set a non-default datastore path \
+                 and root filesystem is verity",
+            );
         }
 
-        let datastore_configuration = format!("DatastorePath={}", datastore_path.display());
-        fs::write(agent_config_path, datastore_configuration).structured(
+        let mut updated_contents = contents;
+        if !updated_contents.is_empty() && !updated_contents.ends_with('\n') {
+            updated_contents.push('\n');
+        }
+        updated_contents.push_str(&format!("DatastorePath={}\n", datastore_path.display()));
+        fs::write(agent_config_path, updated_contents).structured(
             ServicingError::CreateConfigurationFile {
                 path: agent_config_path.into(),
             },

From e8ecc9a5514555cc4a5a20316a5b8af3bc3cca53 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Tue, 8 Sep 2026 17:45:33 +0000
Subject: [PATCH 11/14] datastore: rename correlation_id to database_id

Renames the correlation-id concept to database_id to clarify its
actual semantics: it is a stable identifier tied to the datastore
file lifetime (get-or-create on first access), not to a specific
install invocation. This separates it from the forthcoming
installation_id concept (added in a later commit) which is meant
to be tied to individual install() calls.

No behavior change - this is a pure rename across datastore.rs,
lib.rs, and logging/tracestream.rs (including the JSON telemetry
field key, tests, and doc comments).
---
 crates/trident/src/datastore.rs           | 66 +++++++++++------------
 crates/trident/src/lib.rs                 | 12 ++---
 crates/trident/src/logging/tracestream.rs | 52 +++++++++---------
 crates/trident/src/main.rs                |  2 +-
 4 files changed, 66 insertions(+), 66 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 7f71885fa..1459b73a2 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -14,12 +14,12 @@ use trident_api::{
 
 use crate::TRIDENT_SEMVER_VERSION;
 
-/// Key under which the datastore's unique correlation ID is stored in the
+/// Key under which the datastore's unique database ID is stored in the
 /// generic key-value table. This ID is generated once (on first access) and
 /// persisted for the lifetime of the datastore. It is intended to be added to
 /// tracing/telemetry so that all activity for a given host installation can
 /// be correlated.
-const CORRELATION_ID_KEY: &str = "correlation-id";
+const DATABASE_ID_KEY: &str = "database-id";
 
 pub struct DataStore {
     db: Option<sqlite::Connection>,
@@ -185,7 +185,7 @@ impl DataStore {
                 TridentVersion::SemVer(TRIDENT_SEMVER_VERSION.clone());
             Self::write_host_status(&persistent_db, self.host_status())?;
 
-            // Carry over any generic key-value entries (e.g. the correlation ID)
+            // Carry over any generic key-value entries (e.g. the database ID)
             // recorded in the temporary datastore into the persistent one, so
             // they survive the transition from temporary to persistent
             // storage.
@@ -412,7 +412,7 @@ impl DataStore {
     /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not
     /// just `HostStatus`.
     ///
-    /// `correlation_id` is currently the only first-party caller of the
+    /// `database_id` is currently the only first-party caller of the
     /// generic key-value store, and it needs insert-if-absent semantics
     /// (see `set_value_if_absent`) rather than an unconditional overwrite,
     /// so this unconditional-overwrite variant is presently exercised only
@@ -434,7 +434,7 @@ impl DataStore {
     /// Like [`Self::set_value`], but only inserts a row if `key` does not
     /// already have one; an existing row is left untouched. Used where two
     /// datastore connections could race to perform "first access"
-    /// initialization of a key (see [`Self::correlation_id`]): whichever
+    /// initialization of a key (see [`Self::database_id`]): whichever
     /// connection's insert commits first wins, and the other's insert
     /// becomes a no-op instead of overwriting the winner's value.
     pub(crate) fn set_value_if_absent<T: Serialize>(
@@ -495,7 +495,7 @@ impl DataStore {
         Ok(())
     }
 
-    /// Retrieve this datastore's unique correlation ID, generating and
+    /// Retrieve this datastore's unique database ID, generating and
     /// persisting a new one on first access.
     ///
     /// This ID is stable for the lifetime of the datastore (surviving the
@@ -513,16 +513,16 @@ impl DataStore {
     /// freshly generated ID with `ON CONFLICT DO NOTHING` (a no-op if
     /// another connection already inserted one first), then read back
     /// whichever ID actually won that race.
-    pub fn correlation_id(&mut self) -> Result<Uuid, TridentError> {
-        if let Some(id) = self.get_value::<Uuid>(CORRELATION_ID_KEY)? {
+    pub fn database_id(&mut self) -> Result<Uuid, TridentError> {
+        if let Some(id) = self.get_value::<Uuid>(DATABASE_ID_KEY)? {
             return Ok(id);
         }
 
-        self.set_value_if_absent(CORRELATION_ID_KEY, &Uuid::new_v4())?;
+        self.set_value_if_absent(DATABASE_ID_KEY, &Uuid::new_v4())?;
 
-        self.get_value::<Uuid>(CORRELATION_ID_KEY)?
+        self.get_value::<Uuid>(DATABASE_ID_KEY)?
             .structured(InternalError::Internal(
-                "Correlation ID missing immediately after being inserted",
+                "Database ID missing immediately after being inserted",
             ))
     }
 
@@ -596,15 +596,15 @@ mod tests {
         let datastore_path = temp_dir.path().join("db.sqlite");
 
         let mut datastore = super::DataStore::open_or_create(&datastore_path).unwrap();
-        let correlation_id = datastore.correlation_id().unwrap();
+        let database_id = datastore.database_id().unwrap();
 
         // Persist to the exact same path the datastore is currently open at.
         datastore.persist(&datastore_path).unwrap();
 
         assert_eq!(
-            datastore.correlation_id().unwrap(),
-            correlation_id,
-            "Correlation ID should survive a self-persist"
+            datastore.database_id().unwrap(),
+            database_id,
+            "Database ID should survive a self-persist"
         );
     }
 
@@ -677,7 +677,7 @@ mod tests {
     #[test]
     /// Regression test: a datastore created by an older Trident version that
     /// predates the `keyvalue` table (only `hoststatus` exists) must still
-    /// be usable after `open()` -- in particular, `correlation_id()` must
+    /// be usable after `open()` -- in particular, `database_id()` must
     /// not fail with "no such table: keyvalue".
     fn test_open_upgrades_pre_existing_datastore_schema() {
         let temp_dir = tempfile::tempdir().unwrap();
@@ -699,20 +699,20 @@ mod tests {
 
         let mut datastore = super::DataStore::open(&path).unwrap();
         // Should not fail with "no such table: keyvalue".
-        datastore.correlation_id().unwrap();
+        datastore.database_id().unwrap();
 
         temp_dir.close().unwrap();
     }
 
     #[test]
-    fn test_correlation_id_concurrent_first_access_is_consistent() {
+    fn test_database_id_concurrent_first_access_is_consistent() {
         let temp_dir = tempfile::tempdir().unwrap();
         let path = temp_dir.path().join("db.sqlite");
 
         // Create the datastore (and its schema) up front, then open two
         // separate connections to it, simulating two daemon RPC handlers
         // concurrently calling `Trident::new` (and therefore
-        // `correlation_id`) against the same datastore path.
+        // `database_id`) against the same datastore path.
         super::DataStore::make_datastore(&path).unwrap();
 
         let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
@@ -723,10 +723,10 @@ mod tests {
                 std::thread::spawn(move || {
                     let mut datastore = super::DataStore::open(&path).unwrap();
                     // Synchronize so both threads attempt "first access"
-                    // (no correlation ID persisted yet) as close together
+                    // (no database ID persisted yet) as close together
                     // as possible.
                     barrier.wait();
-                    datastore.correlation_id().unwrap()
+                    datastore.database_id().unwrap()
                 })
             })
             .collect();
@@ -734,14 +734,14 @@ mod tests {
         let ids: Vec<uuid::Uuid> = handles.into_iter().map(|h| h.join().unwrap()).collect();
         assert_eq!(
             ids[0], ids[1],
-            "concurrent first access returned inconsistent correlation IDs"
+            "concurrent first access returned inconsistent database IDs"
         );
 
         temp_dir.close().unwrap();
     }
 
     #[test]
-    fn test_correlation_id_is_stable() {
+    fn test_database_id_is_stable() {
         let temp_dir = tempfile::tempdir().unwrap();
         let path = temp_dir.path().join("db.sqlite");
         let db = super::DataStore::make_datastore(&path).unwrap();
@@ -751,10 +751,10 @@ mod tests {
             temporary: false,
         };
 
-        let id = datastore.correlation_id().unwrap();
-        // Calling correlation_id again should return the same ID, not generate a
+        let id = datastore.database_id().unwrap();
+        // Calling database_id again should return the same ID, not generate a
         // new one.
-        assert_eq!(datastore.correlation_id().unwrap(), id);
+        assert_eq!(datastore.database_id().unwrap(), id);
 
         temp_dir.close().unwrap();
     }
@@ -835,22 +835,22 @@ mod functional_test {
     }
 
     #[functional_test]
-    fn test_correlation_id_survives_persist() {
+    fn test_database_id_survives_persist() {
         let temp_dir = TempDir::new().unwrap();
         let datastore_temp_path = temp_dir.path().join("db-tmp.sqlite");
         let datastore_path = temp_dir.path().join("db.sqlite");
 
-        // Generate a correlation ID in the temporary datastore, then persist it.
-        let correlation_id = {
+        // Generate a database ID in the temporary datastore, then persist it.
+        let database_id = {
             let mut datastore = DataStore::open_or_create(&datastore_temp_path).unwrap();
-            let correlation_id = datastore.correlation_id().unwrap();
+            let database_id = datastore.database_id().unwrap();
             datastore.persist(&datastore_path).unwrap();
-            correlation_id
+            database_id
         };
 
-        // Re-open the persisted datastore and verify the same correlation ID is
+        // Re-open the persisted datastore and verify the same database ID is
         // returned, rather than a new one being generated.
         let mut datastore = DataStore::open(&datastore_path).unwrap();
-        assert_eq!(datastore.correlation_id().unwrap(), correlation_id);
+        assert_eq!(datastore.database_id().unwrap(), database_id);
     }
 }
diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs
index 7e7276632..093a93622 100644
--- a/crates/trident/src/lib.rs
+++ b/crates/trident/src/lib.rs
@@ -208,19 +208,19 @@ impl Trident {
         }
 
         // Retrieve (or create, on first run) this host's unique
-        // correlation ID from the datastore actually used for servicing,
+        // database ID from the datastore actually used for servicing,
         // and attach it to the shared TraceStream before any startup
         // metrics are emitted, so every trace/metric -- including this
         // very "trident_start" event -- carries it. This runs for every
         // caller of `Trident::new` (both the CLI path and each daemon
         // RPC handler), since they all supply `datastore_path`.
-        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.correlation_id()) {
-            Ok(correlation_id) => {
-                info!("Correlation ID: {correlation_id}");
-                tracestream.set_correlation_id(correlation_id.to_string());
+        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.database_id()) {
+            Ok(database_id) => {
+                info!("Database ID: {database_id}");
+                tracestream.set_database_id(database_id.to_string());
             }
             Err(e) => {
-                warn!("Failed to get or create correlation ID: {e:?}");
+                warn!("Failed to get or create database ID: {e:?}");
             }
         }
 
diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index c69bdeb8a..d1c03d522 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -84,7 +84,7 @@ pub struct TraceStream {
     // TODO: Consider changing this to a LockOnce when rustc is updated to
     // >=1.70
     target: Arc<RwLock<Option<String>>>,
-    correlation_id: Arc<RwLock<Option<String>>>,
+    database_id: Arc<RwLock<Option<String>>>,
     disabled: bool,
 }
 
@@ -126,17 +126,17 @@ impl TraceStream {
         Ok(())
     }
 
-    /// Set the correlation ID to attach to every trace entry sent from this point
+    /// Set the database ID to attach to every trace entry sent from this point
     /// forward, as an additional field, so that all traces/metrics for a
     /// given host installation can be correlated. Expected to be called once
-    /// the datastore's persisted correlation ID has been retrieved (see
-    /// `DataStore::correlation_id`).
-    pub fn set_correlation_id(&self, correlation_id: String) {
-        match self.correlation_id.write() {
+    /// the datastore's persisted database ID has been retrieved (see
+    /// `DataStore::database_id`).
+    pub fn set_database_id(&self, database_id: String) {
+        match self.database_id.write() {
             Ok(mut val) => {
-                val.replace(correlation_id);
+                val.replace(database_id);
             }
-            Err(_) => warn!("Failed to lock tracestream to set correlation ID"),
+            Err(_) => warn!("Failed to lock tracestream to set database ID"),
         }
     }
 
@@ -157,7 +157,7 @@ impl TraceStream {
     ) -> Box<TraceSender> {
         Box::new(TraceSender::new(
             self.target.clone(),
-            self.correlation_id.clone(),
+            self.database_id.clone(),
             metrics_file_path,
         ))
     }
@@ -165,7 +165,7 @@ impl TraceStream {
 
 pub struct TraceSender {
     server: Arc<RwLock<Option<String>>>,
-    correlation_id: Arc<RwLock<Option<String>>>,
+    database_id: Arc<RwLock<Option<String>>>,
     client: reqwest::blocking::Client,
     metrics_file: Option<File>,
 }
@@ -179,12 +179,12 @@ struct ExecutionTime(Instant);
 impl TraceSender {
     fn new(
         server: Arc<RwLock<Option<String>>>,
-        correlation_id: Arc<RwLock<Option<String>>>,
+        database_id: Arc<RwLock<Option<String>>>,
         metrics_file_path: &str,
     ) -> Self {
         Self {
             server,
-            correlation_id,
+            database_id,
             client: reqwest::blocking::Client::new(),
             metrics_file: match files::create_file(metrics_file_path) {
                 Ok(f) => Some(f),
@@ -203,14 +203,14 @@ impl TraceSender {
     }
 
     /// Build the `additional_fields` map for a trace entry: the static
-    /// `ADDITIONAL_FIELDS`, plus the correlation ID (if one has been set via
-    /// `TraceStream::set_correlation_id`), so entries can be correlated back to a
+    /// `ADDITIONAL_FIELDS`, plus the database ID (if one has been set via
+    /// `TraceStream::set_database_id`), so entries can be correlated back to a
     /// specific host installation.
     fn additional_fields(&self) -> BTreeMap<String, Value> {
         let mut fields = ADDITIONAL_FIELDS.clone();
-        if let Ok(correlation_id) = self.correlation_id.read() {
-            if let Some(correlation_id) = correlation_id.as_ref() {
-                fields.insert("correlation_id".to_string(), json!(correlation_id));
+        if let Ok(database_id) = self.database_id.read() {
+            if let Some(database_id) = database_id.as_ref() {
+                fields.insert("database_id".to_string(), json!(database_id));
             }
         }
         fields
@@ -579,16 +579,16 @@ mod tests {
     }
 
     #[test]
-    /// Regression test: `TraceStream::set_correlation_id` must actually
-    /// reach the serialized trace entry's `additional_fields.correlation_id`
+    /// Regression test: `TraceStream::set_database_id` must actually
+    /// reach the serialized trace entry's `additional_fields.database_id`
     /// -- the metric/span tests above only assert on `metric_name`/`value`
-    /// and would still pass even if the correlation ID were never copied
+    /// and would still pass even if the database ID were never copied
     /// into `additional_fields`.
-    fn test_tracestream_correlation_id_written_to_additional_fields() {
+    fn test_tracestream_database_id_written_to_additional_fields() {
         let temp_dir = tempfile::tempdir().unwrap();
         let metrics_path = temp_dir.path().join("metrics.jsonl");
         let tracestream = TraceStream::default();
-        tracestream.set_correlation_id("test-correlation-id".to_string());
+        tracestream.set_database_id("test-database-id".to_string());
         let trace_sender = tracestream
             .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap())
             .with_filter(filter::LevelFilter::INFO);
@@ -600,7 +600,7 @@ mod tests {
         );
 
         tracing::info!(
-            metric_name = "test_metric_with_correlation_id",
+            metric_name = "test_metric_with_database_id",
             value = true
         );
 
@@ -612,13 +612,13 @@ mod tests {
         let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
 
         let metric_found = lines.iter().any(|line| {
-            line.contains(r#""metric_name":"test_metric_with_correlation_id""#)
-                && line.contains(r#""correlation_id":"test-correlation-id""#)
+            line.contains(r#""metric_name":"test_metric_with_database_id""#)
+                && line.contains(r#""database_id":"test-database-id""#)
         });
 
         assert!(
             metric_found,
-            "Expected metric with correlation_id field not found in the local metrics file"
+            "Expected metric with database_id field not found in the local metrics file"
         );
     }
 
diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs
index bcd726845..c1d996459 100644
--- a/crates/trident/src/main.rs
+++ b/crates/trident/src/main.rs
@@ -166,7 +166,7 @@ fn run_trident(
                 .message("Failed to initialize Trident")?;
 
                 // `Trident::new` has already retrieved (or created) this
-                // host's persisted correlation ID and attached it to the
+                // host's persisted database ID and attached it to the
                 // shared TraceStream, so every trace/metric emitted from
                 // here on -- including "trident_start" -- carries it.
                 let mut datastore = DataStore::open_or_create(agent_config.datastore_path())

From 98a22f7a9e4d25e92bcafcf5dbc43347e35b79ba Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Tue, 8 Sep 2026 18:54:48 +0000
Subject: [PATCH 12/14] fmt: apply rustfmt to database_id test in
 tracestream.rs

---
 crates/trident/src/logging/tracestream.rs | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index d1c03d522..b32734424 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -599,10 +599,7 @@ mod tests {
             tracing_subscriber::Registry::default().with(trace_sender),
         );
 
-        tracing::info!(
-            metric_name = "test_metric_with_database_id",
-            value = true
-        );
+        tracing::info!(metric_name = "test_metric_with_database_id", value = true);
 
         // Ensure the trace system has time to write the file.
         std::thread::sleep(std::time::Duration::from_millis(100));

From 118097909504830cd70789af9e4f6dac66389215 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Tue, 8 Sep 2026 22:57:31 +0000
Subject: [PATCH 13/14] datastore: reject open() of a datastore file missing
 the hoststatus table

open() called ensure_schema() (CREATE TABLE IF NOT EXISTS ...) before
checking whether the datastore already had valid content, so a zero-byte
or otherwise truncated/corrupt existing file was silently accepted as a
fresh, valid, unprovisioned datastore instead of failing loudly. This
could mask real datastore loss.

open() now requires the hoststatus table to already exist before
applying any schema migration; a file missing it is rejected as a load
failure. Existing datastores that only predate a later table (e.g.
keyvalue) are unaffected, since hoststatus already exists for them.

Added test_open_rejects_file_missing_hoststatus_table.
---
 crates/trident/src/datastore.rs | 56 +++++++++++++++++++++++++++++++++
 1 file changed, 56 insertions(+)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 1459b73a2..61476f56a 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -55,6 +55,24 @@ impl DataStore {
         // daemon RPC handlers) can briefly contend for the write lock; wait
         // for it rather than failing immediately with "database is locked".
         Self::set_busy_timeout(&mut db)?;
+        // Require the `hoststatus` table to already be present before
+        // applying any schema migration. This distinguishes a genuinely
+        // pre-existing datastore (which may only be missing a table added
+        // in a later Trident version, e.g. `keyvalue`) from a zero-byte or
+        // otherwise corrupt/truncated file, which must fail loudly here
+        // rather than silently succeeding as an empty, freshly-provisioned
+        // datastore.
+        if !Self::table_exists(&db, "hoststatus")? {
+            return Err(TridentError::new(ServicingError::Datastore {
+                inner: DatastoreError::LoadDatastore {
+                    path: path.to_string_lossy().into(),
+                },
+            }))
+            .message(
+                "Existing datastore file is missing its 'hoststatus' table; \
+                 the file may be corrupt, truncated, or not a Trident datastore",
+            );
+        }
         // Existing datastores may predate a table added in a later Trident
         // version (e.g. `keyvalue`). Idempotently ensure the full schema is
         // present so upgraded hosts don't fail with "no such table" the
@@ -154,6 +172,23 @@ impl DataStore {
             .structured(ServicingError::from(DatastoreError::OpenDatastore))
     }
 
+    /// Returns whether a table with the given name currently exists in the
+    /// datastore's schema (queried via `sqlite_master`).
+    fn table_exists(db: &sqlite::Connection, table: &str) -> Result<bool, TridentError> {
+        let mut statement = db
+            .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?")
+            .structured(ServicingError::from(DatastoreError::InitializeDatastore))?;
+        statement
+            .bind((1, table))
+            .structured(ServicingError::from(DatastoreError::InitializeDatastore))?;
+        Ok(matches!(
+            statement
+                .next()
+                .structured(ServicingError::from(DatastoreError::InitializeDatastore))?,
+            State::Row
+        ))
+    }
+
     /// Idempotently create any tables that don't already exist. Safe to call
     /// on both newly-created and pre-existing datastores, so that a
     /// datastore created by an older Trident version picks up tables added
@@ -704,6 +739,27 @@ mod tests {
         temp_dir.close().unwrap();
     }
 
+    #[test]
+    /// Regression test: a zero-byte or otherwise schema-less file at the
+    /// datastore path must be rejected by `open()` rather than silently
+    /// treated as a valid, freshly-provisioned datastore.
+    fn test_open_rejects_file_missing_hoststatus_table() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let path = temp_dir.path().join("db.sqlite");
+
+        // Create a valid, but empty (no tables), SQLite file at the path --
+        // simulating a truncated/corrupt existing datastore.
+        sqlite::open(&path).unwrap();
+
+        let result = super::DataStore::open(&path);
+        assert!(
+            result.is_err(),
+            "open() must reject a datastore file with no 'hoststatus' table"
+        );
+
+        temp_dir.close().unwrap();
+    }
+
     #[test]
     fn test_database_id_concurrent_first_access_is_consistent() {
         let temp_dir = tempfile::tempdir().unwrap();

From ea89966eef1e9f526f3071b7b4b1afdd540df1d2 Mon Sep 17 00:00:00 2001
From: Brian Fjeldstad <bfjelds@microsoft.com>
Date: Wed, 9 Sep 2026 01:39:08 +0000
Subject: [PATCH 14/14] Rename database_id to datastore_id for terminology
 consistency

The persisted per-datastore identifier was named database_id, but every
other part of the codebase (DataStore, datastore_path,
DataStore::open_or_create, etc.) refers to this abstraction as the
"datastore", not the "database" -- database_id was the one outlier,
presumably because it happens to be backed by SQLite under the hood, an
implementation detail the rest of the code deliberately does not surface.

Renames the constant (DATABASE_ID_KEY -> DATASTORE_ID_KEY), the
persisted key string ("database-id" -> "datastore-id"), the method
(database_id() -> datastore_id()), TraceStream field/setter, the
additional_fields telemetry key, and associated log messages/tests/docs.

Since none of this has been released yet (all PRs in this stack are
still open), renaming the persisted key string now is safe -- no
already-provisioned host exists with data under the old key.
---
 crates/trident/src/datastore.rs           | 52 +++++++++++------------
 crates/trident/src/lib.rs                 |  8 ++--
 crates/trident/src/logging/tracestream.rs | 42 +++++++++---------
 3 files changed, 51 insertions(+), 51 deletions(-)

diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs
index 61476f56a..712a6b9a0 100644
--- a/crates/trident/src/datastore.rs
+++ b/crates/trident/src/datastore.rs
@@ -19,7 +19,7 @@ use crate::TRIDENT_SEMVER_VERSION;
 /// persisted for the lifetime of the datastore. It is intended to be added to
 /// tracing/telemetry so that all activity for a given host installation can
 /// be correlated.
-const DATABASE_ID_KEY: &str = "database-id";
+const DATASTORE_ID_KEY: &str = "datastore-id";
 
 pub struct DataStore {
     db: Option<sqlite::Connection>,
@@ -447,7 +447,7 @@ impl DataStore {
     /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not
     /// just `HostStatus`.
     ///
-    /// `database_id` is currently the only first-party caller of the
+    /// `datastore_id` is currently the only first-party caller of the
     /// generic key-value store, and it needs insert-if-absent semantics
     /// (see `set_value_if_absent`) rather than an unconditional overwrite,
     /// so this unconditional-overwrite variant is presently exercised only
@@ -469,7 +469,7 @@ impl DataStore {
     /// Like [`Self::set_value`], but only inserts a row if `key` does not
     /// already have one; an existing row is left untouched. Used where two
     /// datastore connections could race to perform "first access"
-    /// initialization of a key (see [`Self::database_id`]): whichever
+    /// initialization of a key (see [`Self::datastore_id`]): whichever
     /// connection's insert commits first wins, and the other's insert
     /// becomes a no-op instead of overwriting the winner's value.
     pub(crate) fn set_value_if_absent<T: Serialize>(
@@ -548,16 +548,16 @@ impl DataStore {
     /// freshly generated ID with `ON CONFLICT DO NOTHING` (a no-op if
     /// another connection already inserted one first), then read back
     /// whichever ID actually won that race.
-    pub fn database_id(&mut self) -> Result<Uuid, TridentError> {
-        if let Some(id) = self.get_value::<Uuid>(DATABASE_ID_KEY)? {
+    pub fn datastore_id(&mut self) -> Result<Uuid, TridentError> {
+        if let Some(id) = self.get_value::<Uuid>(DATASTORE_ID_KEY)? {
             return Ok(id);
         }
 
-        self.set_value_if_absent(DATABASE_ID_KEY, &Uuid::new_v4())?;
+        self.set_value_if_absent(DATASTORE_ID_KEY, &Uuid::new_v4())?;
 
-        self.get_value::<Uuid>(DATABASE_ID_KEY)?
+        self.get_value::<Uuid>(DATASTORE_ID_KEY)?
             .structured(InternalError::Internal(
-                "Database ID missing immediately after being inserted",
+                "Datastore ID missing immediately after being inserted",
             ))
     }
 
@@ -631,15 +631,15 @@ mod tests {
         let datastore_path = temp_dir.path().join("db.sqlite");
 
         let mut datastore = super::DataStore::open_or_create(&datastore_path).unwrap();
-        let database_id = datastore.database_id().unwrap();
+        let datastore_id = datastore.datastore_id().unwrap();
 
         // Persist to the exact same path the datastore is currently open at.
         datastore.persist(&datastore_path).unwrap();
 
         assert_eq!(
-            datastore.database_id().unwrap(),
-            database_id,
-            "Database ID should survive a self-persist"
+            datastore.datastore_id().unwrap(),
+            datastore_id,
+            "Datastore ID should survive a self-persist"
         );
     }
 
@@ -712,7 +712,7 @@ mod tests {
     #[test]
     /// Regression test: a datastore created by an older Trident version that
     /// predates the `keyvalue` table (only `hoststatus` exists) must still
-    /// be usable after `open()` -- in particular, `database_id()` must
+    /// be usable after `open()` -- in particular, `datastore_id()` must
     /// not fail with "no such table: keyvalue".
     fn test_open_upgrades_pre_existing_datastore_schema() {
         let temp_dir = tempfile::tempdir().unwrap();
@@ -734,7 +734,7 @@ mod tests {
 
         let mut datastore = super::DataStore::open(&path).unwrap();
         // Should not fail with "no such table: keyvalue".
-        datastore.database_id().unwrap();
+        datastore.datastore_id().unwrap();
 
         temp_dir.close().unwrap();
     }
@@ -761,14 +761,14 @@ mod tests {
     }
 
     #[test]
-    fn test_database_id_concurrent_first_access_is_consistent() {
+    fn test_datastore_id_concurrent_first_access_is_consistent() {
         let temp_dir = tempfile::tempdir().unwrap();
         let path = temp_dir.path().join("db.sqlite");
 
         // Create the datastore (and its schema) up front, then open two
         // separate connections to it, simulating two daemon RPC handlers
         // concurrently calling `Trident::new` (and therefore
-        // `database_id`) against the same datastore path.
+        // `datastore_id`) against the same datastore path.
         super::DataStore::make_datastore(&path).unwrap();
 
         let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
@@ -782,7 +782,7 @@ mod tests {
                     // (no database ID persisted yet) as close together
                     // as possible.
                     barrier.wait();
-                    datastore.database_id().unwrap()
+                    datastore.datastore_id().unwrap()
                 })
             })
             .collect();
@@ -797,7 +797,7 @@ mod tests {
     }
 
     #[test]
-    fn test_database_id_is_stable() {
+    fn test_datastore_id_is_stable() {
         let temp_dir = tempfile::tempdir().unwrap();
         let path = temp_dir.path().join("db.sqlite");
         let db = super::DataStore::make_datastore(&path).unwrap();
@@ -807,10 +807,10 @@ mod tests {
             temporary: false,
         };
 
-        let id = datastore.database_id().unwrap();
-        // Calling database_id again should return the same ID, not generate a
+        let id = datastore.datastore_id().unwrap();
+        // Calling datastore_id again should return the same ID, not generate a
         // new one.
-        assert_eq!(datastore.database_id().unwrap(), id);
+        assert_eq!(datastore.datastore_id().unwrap(), id);
 
         temp_dir.close().unwrap();
     }
@@ -891,22 +891,22 @@ mod functional_test {
     }
 
     #[functional_test]
-    fn test_database_id_survives_persist() {
+    fn test_datastore_id_survives_persist() {
         let temp_dir = TempDir::new().unwrap();
         let datastore_temp_path = temp_dir.path().join("db-tmp.sqlite");
         let datastore_path = temp_dir.path().join("db.sqlite");
 
         // Generate a database ID in the temporary datastore, then persist it.
-        let database_id = {
+        let datastore_id = {
             let mut datastore = DataStore::open_or_create(&datastore_temp_path).unwrap();
-            let database_id = datastore.database_id().unwrap();
+            let datastore_id = datastore.datastore_id().unwrap();
             datastore.persist(&datastore_path).unwrap();
-            database_id
+            datastore_id
         };
 
         // Re-open the persisted datastore and verify the same database ID is
         // returned, rather than a new one being generated.
         let mut datastore = DataStore::open(&datastore_path).unwrap();
-        assert_eq!(datastore.database_id().unwrap(), database_id);
+        assert_eq!(datastore.datastore_id().unwrap(), datastore_id);
     }
 }
diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs
index 093a93622..bb3f62fb4 100644
--- a/crates/trident/src/lib.rs
+++ b/crates/trident/src/lib.rs
@@ -214,10 +214,10 @@ impl Trident {
         // very "trident_start" event -- carries it. This runs for every
         // caller of `Trident::new` (both the CLI path and each daemon
         // RPC handler), since they all supply `datastore_path`.
-        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.database_id()) {
-            Ok(database_id) => {
-                info!("Database ID: {database_id}");
-                tracestream.set_database_id(database_id.to_string());
+        match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.datastore_id()) {
+            Ok(datastore_id) => {
+                info!("Datastore ID: {datastore_id}");
+                tracestream.set_datastore_id(datastore_id.to_string());
             }
             Err(e) => {
                 warn!("Failed to get or create database ID: {e:?}");
diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs
index b32734424..d6f87fdf0 100644
--- a/crates/trident/src/logging/tracestream.rs
+++ b/crates/trident/src/logging/tracestream.rs
@@ -84,7 +84,7 @@ pub struct TraceStream {
     // TODO: Consider changing this to a LockOnce when rustc is updated to
     // >=1.70
     target: Arc<RwLock<Option<String>>>,
-    database_id: Arc<RwLock<Option<String>>>,
+    datastore_id: Arc<RwLock<Option<String>>>,
     disabled: bool,
 }
 
@@ -130,11 +130,11 @@ impl TraceStream {
     /// forward, as an additional field, so that all traces/metrics for a
     /// given host installation can be correlated. Expected to be called once
     /// the datastore's persisted database ID has been retrieved (see
-    /// `DataStore::database_id`).
-    pub fn set_database_id(&self, database_id: String) {
-        match self.database_id.write() {
+    /// `DataStore::datastore_id`).
+    pub fn set_datastore_id(&self, datastore_id: String) {
+        match self.datastore_id.write() {
             Ok(mut val) => {
-                val.replace(database_id);
+                val.replace(datastore_id);
             }
             Err(_) => warn!("Failed to lock tracestream to set database ID"),
         }
@@ -157,7 +157,7 @@ impl TraceStream {
     ) -> Box<TraceSender> {
         Box::new(TraceSender::new(
             self.target.clone(),
-            self.database_id.clone(),
+            self.datastore_id.clone(),
             metrics_file_path,
         ))
     }
@@ -165,7 +165,7 @@ impl TraceStream {
 
 pub struct TraceSender {
     server: Arc<RwLock<Option<String>>>,
-    database_id: Arc<RwLock<Option<String>>>,
+    datastore_id: Arc<RwLock<Option<String>>>,
     client: reqwest::blocking::Client,
     metrics_file: Option<File>,
 }
@@ -179,12 +179,12 @@ struct ExecutionTime(Instant);
 impl TraceSender {
     fn new(
         server: Arc<RwLock<Option<String>>>,
-        database_id: Arc<RwLock<Option<String>>>,
+        datastore_id: Arc<RwLock<Option<String>>>,
         metrics_file_path: &str,
     ) -> Self {
         Self {
             server,
-            database_id,
+            datastore_id,
             client: reqwest::blocking::Client::new(),
             metrics_file: match files::create_file(metrics_file_path) {
                 Ok(f) => Some(f),
@@ -204,13 +204,13 @@ impl TraceSender {
 
     /// Build the `additional_fields` map for a trace entry: the static
     /// `ADDITIONAL_FIELDS`, plus the database ID (if one has been set via
-    /// `TraceStream::set_database_id`), so entries can be correlated back to a
+    /// `TraceStream::set_datastore_id`), so entries can be correlated back to a
     /// specific host installation.
     fn additional_fields(&self) -> BTreeMap<String, Value> {
         let mut fields = ADDITIONAL_FIELDS.clone();
-        if let Ok(database_id) = self.database_id.read() {
-            if let Some(database_id) = database_id.as_ref() {
-                fields.insert("database_id".to_string(), json!(database_id));
+        if let Ok(datastore_id) = self.datastore_id.read() {
+            if let Some(datastore_id) = datastore_id.as_ref() {
+                fields.insert("datastore_id".to_string(), json!(datastore_id));
             }
         }
         fields
@@ -579,16 +579,16 @@ mod tests {
     }
 
     #[test]
-    /// Regression test: `TraceStream::set_database_id` must actually
-    /// reach the serialized trace entry's `additional_fields.database_id`
+    /// Regression test: `TraceStream::set_datastore_id` must actually
+    /// reach the serialized trace entry's `additional_fields.datastore_id`
     /// -- the metric/span tests above only assert on `metric_name`/`value`
     /// and would still pass even if the database ID were never copied
     /// into `additional_fields`.
-    fn test_tracestream_database_id_written_to_additional_fields() {
+    fn test_tracestream_datastore_id_written_to_additional_fields() {
         let temp_dir = tempfile::tempdir().unwrap();
         let metrics_path = temp_dir.path().join("metrics.jsonl");
         let tracestream = TraceStream::default();
-        tracestream.set_database_id("test-database-id".to_string());
+        tracestream.set_datastore_id("test-datastore-id".to_string());
         let trace_sender = tracestream
             .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap())
             .with_filter(filter::LevelFilter::INFO);
@@ -599,7 +599,7 @@ mod tests {
             tracing_subscriber::Registry::default().with(trace_sender),
         );
 
-        tracing::info!(metric_name = "test_metric_with_database_id", value = true);
+        tracing::info!(metric_name = "test_metric_with_datastore_id", value = true);
 
         // Ensure the trace system has time to write the file.
         std::thread::sleep(std::time::Duration::from_millis(100));
@@ -609,13 +609,13 @@ mod tests {
         let lines: Vec<String> = reader.lines().map(|l| l.unwrap()).collect();
 
         let metric_found = lines.iter().any(|line| {
-            line.contains(r#""metric_name":"test_metric_with_database_id""#)
-                && line.contains(r#""database_id":"test-database-id""#)
+            line.contains(r#""metric_name":"test_metric_with_datastore_id""#)
+                && line.contains(r#""datastore_id":"test-datastore-id""#)
         });
 
         assert!(
             metric_found,
-            "Expected metric with database_id field not found in the local metrics file"
+            "Expected metric with datastore_id field not found in the local metrics file"
         );
     }