Merge pull request 'Tracers whose settings change start over on reload' (#50) from fix/tracer-live-reload into main

crates/common/src/config/telemetry.rs

@@ -31,6 +31,10 @@ pub struct TelemetrySubscriber {
     pub interests: Interests,
     pub typ: TelemetrySubscriberType,
     pub lossy: bool,
+    /// inbuxa: a hash of the settings the running tracer is built from
+    /// (everything but its events, level and lossiness, which change in
+    /// place), so a reload can tell which tracers to start over.
+    pub settings: u64,
 }
 
 #[allow(clippy::large_enum_variant)]
@@ -167,6 +171,7 @@ impl Tracers {
             for tracer in bp.list_infallible::<Tracer>().await {
                 let id = tracer.id;
                 let tracer = tracer.object;
+                let settings = tracer_settings(&tracer);
                 let level;
                 let lossy;
                 let events;
@@ -379,6 +384,7 @@ impl Tracers {
                     interests: Default::default(),
                     lossy,
                     typ,
+                    settings,
                 };
 
                 // Parse disabled events
@@ -426,6 +432,7 @@ impl Tracers {
             for hook in bp.list_infallible::<WebHook>().await {
                 let id = hook.id;
                 let hook = hook.object;
+                let settings = webhook_settings(&hook);
 
                 if !hook.enable {
                     continue;
@@ -448,6 +455,7 @@ impl Tracers {
                     id: format!("w_{}", id.id()),
                     interests: Default::default(),
                     lossy: hook.lossy,
+                    settings,
                     typ: TelemetrySubscriberType::Webhook(WebhookTracer {
                         url: hook.url,
                         timeout: hook.timeout.into_inner(),
@@ -516,6 +524,8 @@ impl Tracers {
                         data: storage.data.clone(),
                     }),
                     lossy: true,
+                    // Stores take a restart
+                    settings: 0,
                 });
             }
 
@@ -541,6 +551,7 @@ impl Tracers {
                         buffered: true,
                     }),
                     lossy: false,
+                    settings: 0,
                 });
             }
         } else {
@@ -568,6 +579,7 @@ impl Tracers {
                     buffered: true,
                 }),
                 lossy: false,
+                settings: 0,
             });
         }
 
@@ -701,6 +713,42 @@ impl Metrics {
     }
 }
 
+// inbuxa: what a tracer is built from, less what changes in place
+macro_rules! in_place_reset {
+    ($tracer:expr) => {{
+        $tracer.enable = true;
+        $tracer.level = Default::default();
+        $tracer.lossy = false;
+        $tracer.events = Default::default();
+        $tracer.events_policy = Default::default();
+    }};
+}
+
+fn settings_hash(settings: &impl std::fmt::Debug) -> u64 {
+    use std::hash::{Hash, Hasher};
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    format!("{settings:?}").hash(&mut hasher);
+    hasher.finish()
+}
+
+fn tracer_settings(tracer: &Tracer) -> u64 {
+    let mut tracer = tracer.clone();
+    match &mut tracer {
+        Tracer::Log(tracer) => in_place_reset!(tracer),
+        Tracer::Stdout(tracer) => in_place_reset!(tracer),
+        Tracer::Journal(tracer) => in_place_reset!(tracer),
+        Tracer::OtelHttp(tracer) => in_place_reset!(tracer),
+        Tracer::OtelGrpc(tracer) => in_place_reset!(tracer),
+    }
+    settings_hash(&tracer)
+}
+
+fn webhook_settings(hook: &WebHook) -> u64 {
+    let mut hook = hook.clone();
+    in_place_reset!(hook);
+    settings_hash(&hook)
+}
+
 fn apply_events(
     event_types: impl IntoIterator<Item = EventType>,
     policy: EventPolicy,

crates/common/src/telemetry/mod.rs

@@ -14,15 +14,26 @@ pub mod webhooks;
 use tracers::log::spawn_log_tracer;
 use tracers::otel::spawn_otel_tracer;
 use tracers::stdout::spawn_console_tracer;
+use ahash::AHashMap;
+use parking_lot::Mutex;
 use trc::{Collector, ipc::subscriber::SubscriberBuilder};
 use webhooks::spawn_webhook_tracer;
 
 use crate::config::telemetry::{Telemetry, TelemetrySubscriberType};
 
+/// inbuxa: the tracers this server started, by subscriber id, with the
+/// settings each was built from. Live-tracing streams and other subscribers
+/// registered elsewhere aren't listed, so a reload leaves them running.
+static RUNNING_TRACERS: Mutex<Option<AHashMap<String, u64>>> = Mutex::new(None);
+
 impl Telemetry {
     pub fn enable(self) {
+        let mut running = RUNNING_TRACERS.lock();
+        let running = running.get_or_insert_with(AHashMap::new);
+
         // Spawn tracers
         for tracer in self.tracers.subscribers {
+            running.insert(tracer.id.clone(), tracer.settings);
             tracer.typ.spawn(
                 SubscriberBuilder::new(tracer.id)
                     .with_interests(tracer.interests)
@@ -37,25 +48,39 @@ impl Telemetry {
         Collector::reload();
     }
 
+    // inbuxa: upstream only refreshed the events, level and lossiness of a
+    // tracer that was already running, so a Log tracer moved to another
+    // path (or any tracer whose own settings changed) kept going as it was
+    // built until a restart, while the reload reported the change applied.
+    // A tracer whose settings changed is now started over: the new one is
+    // registered under the same id and the collector swaps it in at an
+    // event boundary, so no event is lost or written twice (see
+    // Update::RegisterSubscriber); the old one writes what it has queued
+    // and stops.
     pub fn update(self) {
+        let mut running = RUNNING_TRACERS.lock();
+        let running = running.get_or_insert_with(AHashMap::new);
+
         // Remove tracers that are no longer active
-        let active_subscribers = Collector::get_subscribers();
-        for subscribed_id in &active_subscribers {
-            if !self
+        running.retain(|id, _| {
+            let keep = self
                 .tracers
                 .subscribers
                 .iter()
-                .any(|tracer| tracer.id == *subscribed_id)
-            {
-                Collector::remove_subscriber(subscribed_id.clone());
+                .any(|tracer| tracer.id == *id);
+            if !keep {
+                Collector::remove_subscriber(id.clone());
             }
-        }
+            keep
+        });
 
-        // Activate new tracers or update existing ones
+        // Start new tracers, start over those whose settings changed and
+        // update the rest in place
         for tracer in self.tracers.subscribers {
-            if active_subscribers.contains(&tracer.id) {
+            if running.get(&tracer.id) == Some(&tracer.settings) {
                 Collector::update_subscriber(tracer.id, tracer.interests, tracer.lossy);
             } else {
+                running.insert(tracer.id.clone(), tracer.settings);
                 tracer.typ.spawn(
                     SubscriberBuilder::new(tracer.id)
                         .with_interests(tracer.interests)

crates/common/src/telemetry/tracers/log.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use std::{path::PathBuf, time::SystemTime};
@@ -15,9 +17,27 @@ use tokio::{
 };
 use trc::{TelemetryEvent, ipc::subscriber::SubscriberBuilder, serializers::text::FmtWriter};
 
+// inbuxa: when a Log tracer is started over on the same files (its rotation
+// or format changed), the new one waits for the old one to write what it
+// has queued, so their lines don't interleave. Keyed by path and prefix;
+// each entry is the last tracer's "done" signal, sent when it ends.
+type LogFileOwners = ahash::AHashMap<(String, String), tokio::sync::oneshot::Receiver<()>>;
+static LOG_FILE_OWNERS: parking_lot::Mutex<Option<LogFileOwners>> = parking_lot::Mutex::new(None);
+
 pub(crate) fn spawn_log_tracer(builder: SubscriberBuilder, settings: LogTracer) {
+    let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
+    let previous = LOG_FILE_OWNERS
+        .lock()
+        .get_or_insert_with(Default::default)
+        .insert((settings.path.clone(), settings.prefix.clone()), done_rx);
     let (_, mut rx) = builder.register();
     tokio::spawn(async move {
+        // Dropped when this tracer ends, however it ends
+        let _done = done_tx;
+        if let Some(previous) = previous {
+            let _ = previous.await;
+        }
+
         if let Some(writer) = settings.build_writer().await {
             let mut buf = FmtWriter::new(writer)
                 .with_ansi(settings.ansi)

crates/common/src/telemetry/tracers/otel.rs

@@ -47,6 +47,10 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer
         let mut pending_spans = Vec::new();
 
         let mut active_spans = AHashMap::new();
+        let mut closing = false;
+        let started = std::time::SystemTime::now()
+            .duration_since(std::time::SystemTime::UNIX_EPOCH)
+            .map_or(0, |d| d.as_secs());
 
         loop {
             // Wait for the next event or timeout
@@ -75,12 +79,26 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer
                                     events.iter().chain(std::iter::once(&event)),
                                     &instrumentation,
                                 ));
+                            } else if span.inner.timestamp < started {
+                                // inbuxa: a span that was open when this
+                                // tracer replaced another one (its settings
+                                // changed) is exported with its end event
+                                // rather than dropped
+                                pending_spans.push(build_span_data(
+                                    span,
+                                    &event,
+                                    std::iter::once(&event),
+                                    &instrumentation,
+                                ));
                             }
                         }
                     }
                 }
                 Ok(None) => {
-                    break;
+                    // inbuxa: the tracer was removed or replaced; export
+                    // what is pending now rather than drop it
+                    closing = true;
+                    next_delivery = Instant::now();
                 }
                 Err(_) => (),
             }
@@ -131,6 +149,9 @@ pub(crate) fn spawn_otel_tracer(builder: SubscriberBuilder, mut otel: OtelTracer
                     }
                 }
             }
+            if closing {
+                break;
+            }
             wakeup_time = next_retry.unwrap_or(LONG_1Y_SLUMBER);
         }
     });

crates/common/src/telemetry/webhooks/mod.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use crate::{LONG_1Y_SLUMBER, config::telemetry::WebhookTracer};
@@ -25,6 +27,11 @@ use trc::{
 
 pub(crate) fn spawn_webhook_tracer(builder: SubscriberBuilder, settings: WebhookTracer) {
     let (tx, mut rx) = builder.register();
+    // inbuxa: failed deliveries come back through a weak sender, so the
+    // channel closes when the collector drops this webhook (removed, or
+    // replaced after a settings change) and the task ends; upstream held a
+    // sender here and the task outlived its subscription
+    let tx = tx.downgrade();
     tokio::spawn(async move {
         let settings = Arc::new(settings);
         let mut wakeup_time = LONG_1Y_SLUMBER;
@@ -58,6 +65,15 @@ pub(crate) fn spawn_webhook_tracer(builder: SubscriberBuilder, settings: Webhook
                     }
                 }
                 Ok(None) => {
+                    // inbuxa: deliver what is pending rather than drop it
+                    if !pending_events.is_empty() {
+                        spawn_webhook_handler(
+                            settings.clone(),
+                            in_flight.clone(),
+                            std::mem::take(&mut pending_events),
+                            tx.clone(),
+                        );
+                    }
                     break;
                 }
                 Err(_) => (),
@@ -102,7 +118,7 @@ fn spawn_webhook_handler(
     settings: Arc<WebhookTracer>,
     in_flight: Arc<AtomicBool>,
     events: EventBatch,
-    webhook_tx: mpsc::Sender<EventBatch>,
+    webhook_tx: mpsc::WeakSender<EventBatch>,
 ) {
     tokio::spawn(async move {
         in_flight.store(true, Ordering::Relaxed);
@@ -113,7 +129,11 @@ fn spawn_webhook_handler(
         if let Err(err) = post_webhook_events(&settings, &wrapper).await {
             trc::event!(Telemetry(TelemetryEvent::WebhookError), Details = err);
 
-            if webhook_tx.send(wrapper.events.into_inner()).await.is_err() {
+            let sent = match webhook_tx.upgrade() {
+                Some(webhook_tx) => webhook_tx.send(wrapper.events.into_inner()).await.is_ok(),
+                None => false,
+            };
+            if !sent {
                 trc::event!(
                     Server(ServerEvent::ThreadError),
                     Details = "Failed to send failed webhook events back to main thread",

crates/trc/src/ipc/collector.rs

@@ -245,9 +245,27 @@ impl Collector {
                 Update::RegisterReceiver { receiver } => {
                     self.receivers.push(receiver);
                 }
-                Update::RegisterSubscriber { subscriber } => {
-                    ACTIVE_SUBSCRIBERS.lock().push(subscriber.id.clone());
-                    self.subscribers.push(subscriber);
+                Update::RegisterSubscriber { mut subscriber } => {
+                    // inbuxa: a subscriber registered under the id of a
+                    // running one replaces it (a tracer whose settings
+                    // changed). Every event collected so far went to the old
+                    // one, every later event goes to the new one: the old
+                    // one's batch is sent first (anything its full channel
+                    // can't take moves over, rather than being dropped), and
+                    // dropping it closes its channel, so its task writes
+                    // what is queued and ends.
+                    if let Some(old) = self.subscribers.iter_mut().find(|s| s.id == subscriber.id) {
+                        let _ = old.send_batch();
+                        if !old.batch.is_empty() {
+                            let mut batch = std::mem::take(&mut old.batch);
+                            batch.append(&mut subscriber.batch);
+                            subscriber.batch = batch;
+                        }
+                        *old = subscriber;
+                    } else {
+                        ACTIVE_SUBSCRIBERS.lock().push(subscriber.id.clone());
+                        self.subscribers.push(subscriber);
+                    }
                 }
                 Update::UnregisterSubscriber { id } => {
                     ACTIVE_SUBSCRIBERS.lock().retain(|s| s != &id);

crates/trc/src/ipc/subscriber.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ *
+ * Modified by Coffey Labs in 2026 for INBUXA.
  */
 
 use std::sync::Arc;
@@ -105,6 +107,9 @@ impl SubscriberBuilder {
         self
     }
 
+    /// Registers the subscriber with the collector. inbuxa: one registered
+    /// under the id of a running subscriber replaces it, handing over at an
+    /// event boundary; the old one's channel then closes.
     pub fn register(self) -> (mpsc::Sender<EventBatch>, mpsc::Receiver<EventBatch>) {
         let (tx, rx) = mpsc::channel(8192);
 

tests/src/system/mod.rs

@@ -24,6 +24,7 @@ pub mod quota;
 pub mod reload; // inbuxa: reloads and build errors
 pub mod security;
 pub mod task;
+pub mod tracer_reload; // inbuxa: tracers start over when their settings change
 pub mod tenant;
 pub mod undelete;
 

tests/src/system/tracer_reload.rs

@@ -0,0 +1,204 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+// inbuxa: a tracer whose own settings change is started over by the reload
+// that follows the write: a Log tracer moved to another directory writes
+// there from then on, and no event is lost or written twice on the way.
+
+use crate::utils::{
+    jmap::JmapResponse,
+    server::{TestServer, TestServerBuilder},
+};
+use registry::{
+    schema::{
+        enums::{EventPolicy, LogRotateFrequency, TracingLevel},
+        prelude::ObjectType,
+        structs::{Expression, MtaStageAuth, Tracer, TracerLog},
+    },
+    types::map::Map,
+};
+use serde_json::json;
+use std::{
+    path::{Path, PathBuf},
+    time::{Duration, Instant},
+};
+use trc::{EventType, ServerEvent};
+
+const PREFIX: &str = "tracer-reload";
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn tracer_reload_tests() {
+    let mut test = TestServerBuilder::new("tracer_reload_tests")
+        .await
+        .with_default_listeners()
+        .await
+        .with_object(MtaStageAuth {
+            require: Expression {
+                else_: "false".to_string(),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await
+        .build()
+        .await;
+
+    let admin = test
+        .create_user_account(
+            "admin",
+            "[email protected]",
+            "these_pretzels_are_making_me_thirsty",
+            &[],
+            "Admin",
+        )
+        .await;
+    test.account("admin")
+        .assign_roles_to_account(admin.id(), &["user", "system"])
+        .await;
+    test.insert_account(admin);
+
+    test_log_tracer_moves(&test).await;
+
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+async fn test_log_tracer_moves(test: &TestServer) {
+    println!("Running Log tracer path change...");
+    let admin = test.account("[email protected]");
+    let old_dir = test.temp_dir.path.join("tracer-old");
+    let new_dir = test.temp_dir.path.join("tracer-new");
+    for dir in [&old_dir, &new_dir] {
+        let _ = std::fs::remove_dir_all(dir);
+        std::fs::create_dir_all(dir).unwrap();
+    }
+    let old_file = old_dir.join(PREFIX);
+    let new_file = new_dir.join(PREFIX);
+
+    // A Log tracer for one event type, written to the old directory
+    let response = admin
+        .registry_create([Tracer::Log(TracerLog {
+            path: old_dir.to_string_lossy().into_owned(),
+            prefix: PREFIX.into(),
+            rotate: LogRotateFrequency::Never,
+            ansi: false,
+            multiline: false,
+            enable: true,
+            level: TracingLevel::Trace,
+            lossy: false,
+            events: Map::new(vec![EventType::Server(ServerEvent::Licensing)]),
+            events_policy: EventPolicy::Include,
+        })])
+        .await;
+    assert_applied(&response);
+    let tracer_id = response.created_id(0);
+    emit("marker-before");
+    wait_for(&old_file, "marker-before").await;
+
+    // Events keep coming while the path changes
+    let stream = tokio::spawn(async {
+        for i in 0..2000u32 {
+            emit(&format!("seq-{i:05}-end"));
+            if i % 50 == 0 {
+                tokio::time::sleep(Duration::from_millis(1)).await;
+            }
+        }
+    });
+    tokio::time::sleep(Duration::from_millis(5)).await;
+    let response = admin
+        .registry_update(
+            ObjectType::Tracer,
+            [(tracer_id, json!({"path": new_dir.to_string_lossy()}))],
+        )
+        .await;
+    assert_applied(&response);
+    stream.await.unwrap();
+
+    // Once the reload has run, events go to the new file only
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    emit("marker-after");
+    wait_for(&new_file, "marker-after").await;
+    wait_for(&new_file, "seq-01999-end").await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    let old = read(&old_file);
+    let new = read(&new_file);
+    assert!(!old.contains("marker-after"), "old file still written to");
+    assert!(!new.contains("marker-before"));
+
+    // Every event written once, in one file or the other
+    let old_seq = count_seq(&old);
+    let new_seq = count_seq(&new);
+    println!(
+        "{} events in the old file, {} in the new one",
+        old_seq.iter().filter(|c| **c > 0).count(),
+        new_seq.iter().filter(|c| **c > 0).count()
+    );
+    for i in 0..2000 {
+        assert_eq!(
+            old_seq[i] + new_seq[i],
+            1,
+            "seq-{i:05} written {} + {} times",
+            old_seq[i],
+            new_seq[i]
+        );
+    }
+    assert!(
+        new_seq.iter().any(|c| *c > 0),
+        "no event of the stream reached the new file"
+    );
+
+    // Removing the tracer stops it
+    let response = admin
+        .registry_destroy(ObjectType::Tracer, [tracer_id])
+        .await;
+    assert_applied(&response);
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    emit("marker-removed");
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert!(!read(&new_file).contains("marker-removed"));
+    assert!(!read(&old_file).contains("marker-removed"));
+}
+
+fn emit(marker: &str) {
+    trc::event!(Server(ServerEvent::Licensing), Details = marker.to_string());
+}
+
+fn read(path: &Path) -> String {
+    std::fs::read_to_string(path).unwrap_or_default()
+}
+
+fn count_seq(text: &str) -> Vec<u32> {
+    let mut counts = vec![0u32; 2000];
+    for part in text.split("seq-").skip(1) {
+        if let Some(n) = part.get(..5).and_then(|n| n.parse::<usize>().ok())
+            && part[5..].starts_with("-end")
+        {
+            counts[n] += 1;
+        }
+    }
+    counts
+}
+
+async fn wait_for(path: &PathBuf, marker: &str) {
+    let started = Instant::now();
+    while !read(path).contains(marker) {
+        assert!(
+            started.elapsed() < Duration::from_secs(10),
+            "{marker} not in {}",
+            path.display()
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
+
+fn assert_applied(response: &JmapResponse) {
+    assert_eq!(
+        response.pointer("/methodResponses/0/1/x:settingsReload"),
+        Some(&json!({"applied": true})),
+        "{response:?}"
+    );
+}