Add structured tracing fields for debuggability

crates/node/src/database/maintenance.rs

@@ -44,8 +44,12 @@ impl DatabaseMaintenance {
     async fn step(&mut self) -> Result<()> {
         let timer = self.metrics.db_maintenance_cleanup_notes();
 
-        self.database.cleanup_old_notes(self.config.retention_days).await?;
-        info!("Cleaned up old notes");
+        let deleted = self.database.cleanup_old_notes(self.config.retention_days).await?;
+        info!(
+            notes_deleted = deleted,
+            retention_days = self.config.retention_days,
+            "Maintenance cleanup completed"
+        );
 
         timer.finish("ok");
 

crates/node/src/database/sqlite/mod.rs

@@ -115,8 +115,10 @@ impl DatabaseBackend for SqliteDatabase {
         Ok(Self { pool, metrics })
     }
 
-    #[tracing::instrument(skip(self), fields(operation = "db.store_note"))]
+    #[tracing::instrument(skip(self, note), fields(operation = "db.store_note"))]
     async fn store_note(&self, note: &StoredNote) -> Result<(), DatabaseError> {
+        tracing::debug!(note_id = %note.header.id(), tag = note.header.metadata().tag().as_u32(), "db store_note");
+
         let timer = self.metrics.db_store_note();
 
         let new_note = NewNote::from(note);
@@ -130,7 +132,6 @@ impl DatabaseBackend for SqliteDatabase {
         Ok(())
     }
 
-    #[tracing::instrument(skip(self), fields(operation = "db.fetch_notes"))]
     async fn fetch_notes(
         &self,
         tag: NoteTag,
@@ -139,7 +140,12 @@ impl DatabaseBackend for SqliteDatabase {
         self.fetch_notes_by_tags(&[tag], cursor).await
     }
 
-    #[tracing::instrument(skip(self, tags), fields(operation = "db.fetch_notes_by_tags"))]
+    #[tracing::instrument(skip(self, tags), fields(
+        operation = "db.fetch_notes_by_tags",
+        tag_count = tags.len(),
+        cursor = cursor,
+        notes_returned = tracing::field::Empty,
+    ))]
     async fn fetch_notes_by_tags(
         &self,
         tags: &[NoteTag],
@@ -153,6 +159,7 @@ impl DatabaseBackend for SqliteDatabase {
         // so operators can see when pre-migration clients are being reset.
         let effective_cursor = if cursor > LEGACY_CURSOR_THRESHOLD {
             self.metrics.db_fetch_notes_legacy_cursor_reset();
+            tracing::info!(original_cursor = cursor, "Legacy cursor reset to 0");
             0
         } else {
             cursor
@@ -200,6 +207,7 @@ impl DatabaseBackend for SqliteDatabase {
             stored_notes.push(stored_note);
         }
 
+        tracing::Span::current().record("notes_returned", stored_notes.len());
         timer.finish("ok");
 
         Ok(stored_notes)

crates/node/src/node/grpc/mod.rs

@@ -134,27 +134,45 @@ impl StreamerCtx {
 impl miden_note_transport_proto::miden_note_transport::miden_note_transport_server::MidenNoteTransport
     for GrpcServer
 {
-    #[tracing::instrument(skip(self), fields(operation = "grpc.send_note.request"))]
+    #[tracing::instrument(skip(self, request), fields(
+        operation = "grpc.send_note.request",
+        note_size = tracing::field::Empty,
+    ))]
     async fn send_note(
         &self,
         request: tonic::Request<SendNoteRequest>,
     ) -> Result<tonic::Response<SendNoteResponse>, tonic::Status> {
         let request_data = request.into_inner();
         let pnote = request_data.note.ok_or_else(|| Status::invalid_argument("Missing note"))?;
 
-        // `header` + `details` are the stored payload; cap and metric use the
-        // same number so the recorded size matches what's actually limited.
-        let payload_size = pnote.header.len() + pnote.details.len();
-        let timer = self.metrics.grpc_send_note_request(payload_size as u64);
+        // `header` + `details` are the stored payload; the cap, the metric, and
+        // the span field all use the same number so accept and reject report
+        // the same size.
+        let note_size = pnote.header.len() + pnote.details.len();
+        let span = tracing::Span::current();
+        span.record("note_size", note_size);
+
+        let timer = self.metrics.grpc_send_note_request(note_size as u64);
 
         // Validate note size
-        if payload_size > self.config.max_note_size {
-            return Err(Status::resource_exhausted(format!("Note too large ({payload_size})")));
+        if note_size > self.config.max_note_size {
+            tracing::warn!(reason = "note_too_large", size = note_size, max = self.config.max_note_size, "send_note rejected");
+            return Err(Status::resource_exhausted(format!("Note too large ({note_size})")));
         }
 
         // Convert protobuf request to internal types
         let header = miden_protocol::note::NoteHeader::read_from_bytes(&pnote.header)
-            .map_err(|e| Status::invalid_argument(format!("Invalid header: {e:?}")))?;
+            .map_err(|e| {
+                tracing::warn!(reason = "invalid_header", "send_note rejected");
+                Status::invalid_argument(format!("Invalid header: {e:?}"))
+            })?;
+
+        tracing::debug!(
+            note_id = %header.id(),
+            tag = header.metadata().tag().as_u32(),
+            has_after_block_num = pnote.after_block_num.is_some(),
+            "send_note accepted"
+        );
 
         // Create note for database
         let note_for_db = crate::types::StoredNote {
@@ -175,7 +193,13 @@ impl miden_note_transport_proto::miden_note_transport::miden_note_transport_serv
         Ok(tonic::Response::new(SendNoteResponse {}))
     }
 
-    #[tracing::instrument(skip(self), fields(operation = "grpc.fetch_notes.request"))]
+    #[tracing::instrument(skip(self, request), fields(
+        operation = "grpc.fetch_notes.request",
+        tag_count = tracing::field::Empty,
+        cursor = tracing::field::Empty,
+        notes_returned = tracing::field::Empty,
+        response_cursor = tracing::field::Empty,
+    ))]
     async fn fetch_notes(
         &self,
         request: tonic::Request<FetchNotesRequest>,
@@ -190,6 +214,12 @@ impl miden_note_transport_proto::miden_note_transport::miden_note_transport_serv
         // through `SQLITE_MAX_VARIABLE_NUMBER` or return a pathologically
         // expensive query plan.
         if request_data.tags.len() > MAX_TAGS_PER_FETCH_REQUEST {
+            tracing::warn!(
+                reason = "too_many_tags",
+                tag_count = request_data.tags.len(),
+                max = MAX_TAGS_PER_FETCH_REQUEST,
+                "fetch_notes rejected"
+            );
             return Err(Status::invalid_argument(format!(
                 "Too many tags in fetch_notes request: {} (max {})",
                 request_data.tags.len(),
@@ -203,6 +233,10 @@ impl miden_note_transport_proto::miden_note_transport::miden_note_transport_serv
         let tags: Vec<crate::types::NoteTag> = tag_set.into_iter().map(Into::into).collect();
         let cursor = request_data.cursor;
 
+        let span = tracing::Span::current();
+        span.record("tag_count", tags.len());
+        span.record("cursor", cursor);
+
         // Single-snapshot fetch across ALL tags. Running per-tag queries back
         // to back exposed a race where a concurrent INSERT could land between
         // two per-tag queries and get leapfrogged when rcursor advanced past
@@ -225,6 +259,9 @@ impl miden_note_transport_proto::miden_note_transport::miden_note_transport_serv
 
         let proto_notes: Vec<_> = stored_notes.into_iter().map(TransportNote::from).collect();
 
+        span.record("notes_returned", proto_notes.len());
+        span.record("response_cursor", rcursor);
+
         timer.finish("ok");
 
         let proto_notes_size = proto_notes.iter().map(|pnote| (pnote.header.len() + pnote.details.len()) as u64).sum();
@@ -237,14 +274,23 @@ impl miden_note_transport_proto::miden_note_transport::miden_note_transport_serv
     }
 
     type StreamNotesStream = Sub;
-    #[tracing::instrument(skip(self), fields(operation = "grpc.stream_notes.request"))]
+    #[tracing::instrument(skip(self, request), fields(
+        operation = "grpc.stream_notes.request",
+        subscription_id = tracing::field::Empty,
+    ))]
     async fn stream_notes(
         &self,
         request: tonic::Request<StreamNotesRequest>,
     ) -> Result<tonic::Response<Self::StreamNotesStream>, tonic::Status> {
         let request_data = request.into_inner();
         let tag = request_data.tag.into();
-        let id = rand::rng().random();
+        let id: u64 = rand::rng().random();
+
+        let span = tracing::Span::current();
+        span.record("subscription_id", id);
+
+        tracing::debug!(tag = crate::types::NoteTag::as_u32(&tag), cursor = request_data.cursor, "stream_notes subscribe");
+
         let (sub_tx, sub_rx) = mpsc::channel(32);
         let sub = Sub::new(id, tag, sub_rx, self.streamer.tx.clone());
         let subf = Subface::new(id, tag, sub_tx);

crates/node/src/node/grpc/streaming.rs

@@ -1,6 +1,7 @@
 use core::task::{Poll, Waker};
 use std::collections::BTreeMap;
 use std::sync::Arc;
+use std::time::Instant;
 
 use miden_note_transport_proto::miden_note_transport::{StreamNotesUpdate, TransportNote};
 use tokio::sync::mpsc;
@@ -35,8 +36,12 @@ struct NoteStreamerManager {
 pub(crate) enum StreamerMessage {
     /// New sub
     AddSub(Subface),
-    /// Remove sub
-    RemoveSub((u64, NoteTag)),
+    /// Remove sub, tagged with the reason for the lifecycle log
+    RemoveSub {
+        id: u64,
+        tag: NoteTag,
+        reason: &'static str,
+    },
     /// Update waker for sub
     Waker((u64, Waker)),
     /// Shutdown the streamer
@@ -49,7 +54,14 @@ pub struct TagData {
     /// forwarded to subscribers. Next fetch uses this to query
     /// `seq > cursor` and pick up only new arrivals.
     cursor: u64,
-    subs: BTreeMap<u64, mpsc::Sender<TransportNotesPg>>,
+    subs: BTreeMap<u64, SubEntry>,
+}
+
+/// A subscriber's send channel plus when it was registered, so the removal
+/// event can report how long the subscription lived.
+struct SubEntry {
+    tx: mpsc::Sender<TransportNotesPg>,
+    created_at: Instant,
 }
 
 /// Subscription
@@ -113,9 +125,9 @@ impl NoteStreamerManager {
         for (tag, notes) in tag_notes {
             if let Some(tag_data) = self.tags.get(&tag) {
                 // Wake-up subs with `tag`
-                for (sub_id, sub_tx) in &tag_data.subs {
+                for (sub_id, sub_entry) in &tag_data.subs {
                     if let Some(waker) = self.wakers.remove(sub_id) {
-                        if let Ok(()) = sub_tx.try_send(notes.clone()) {
+                        if let Ok(()) = sub_entry.tx.try_send(notes.clone()) {
                             waker.wake();
                         } else {
                             remove_subs.push((*sub_id, tag));
@@ -124,9 +136,9 @@ impl NoteStreamerManager {
                 }
             }
         }
-        // Remove non-responding subs
+        // Remove non-responding subs (backpressure)
         for (sub_id, tag) in remove_subs {
-            self.remove_sub(sub_id, tag);
+            self.remove_sub(sub_id, tag, "backpressure");
         }
     }
 
@@ -145,21 +157,36 @@ impl NoteStreamerManager {
 
     pub fn add_sub(&mut self, sub: Subface) {
         let entry = self.tags.entry(sub.tag).or_insert_with(TagData::new);
-        entry.subs.insert(sub.id, sub.tx);
+        entry.subs.insert(sub.id, SubEntry { tx: sub.tx, created_at: Instant::now() });
+        let active = self.tags.values().map(|td| td.subs.len()).sum::<usize>();
+        tracing::info!(subscription_id = %sub.id, active_subscriptions = active, "Subscription added");
     }
 
-    pub fn remove_sub(&mut self, sub_id: u64, tag: NoteTag) {
+    /// Removes a subscription and emits the single canonical removal event.
+    ///
+    /// Only logs when a sub was actually present: a sub evicted for
+    /// backpressure is later dropped client-side too, and that second call
+    /// must not emit a duplicate event or mislabel the reason.
+    pub fn remove_sub(&mut self, sub_id: u64, tag: NoteTag, reason: &'static str) {
+        let mut removed_at = None;
         let mut remove_tag = false;
         if let Some(tag_data) = self.tags.get_mut(&tag) {
-            tag_data.subs.remove(&sub_id);
-            if tag_data.subs.is_empty() {
-                // No more subscribers for this tag
-                remove_tag = true;
-            }
+            removed_at = tag_data.subs.remove(&sub_id).map(|entry| entry.created_at);
+            remove_tag = tag_data.subs.is_empty();
         }
         if remove_tag {
             self.tags.remove(&tag);
         }
+        if let Some(created_at) = removed_at {
+            let active = self.tags.values().map(|td| td.subs.len()).sum::<usize>();
+            tracing::info!(
+                subscription_id = %sub_id,
+                active_subscriptions = active,
+                duration_secs = created_at.elapsed().as_secs(),
+                reason,
+                "Subscription removed"
+            );
+        }
     }
 }
 
@@ -201,7 +228,9 @@ impl NoteStreamer {
             Some(msg) = rx.recv() => {
                 match msg {
                     StreamerMessage::AddSub(sub) => manager.add_sub(sub),
-                    StreamerMessage::RemoveSub((id, tag)) => manager.remove_sub(id, tag),
+                    StreamerMessage::RemoveSub { id, tag, reason } => {
+                        manager.remove_sub(id, tag, reason);
+                    },
                     StreamerMessage::Waker((id, waker)) => manager.update_waker(id, waker),
                     StreamerMessage::Shutdown => return Ok(false),
                 }
@@ -267,8 +296,15 @@ impl tonic::codegen::tokio_stream::Stream for Sub {
 
 impl Drop for Sub {
     fn drop(&mut self) {
-        if let Err(e) = self.streamer_tx.try_send(StreamerMessage::RemoveSub((self.id, self.tag))) {
-            tracing::error!("Streamer remove sub control message sending error: {e}");
+        // Hand removal to the manager, which emits the single lifecycle event.
+        // If the sub was already evicted (e.g. backpressure) this is a no-op
+        // and logs nothing.
+        if let Err(e) = self.streamer_tx.try_send(StreamerMessage::RemoveSub {
+            id: self.id,
+            tag: self.tag,
+            reason: "client_disconnect",
+        }) {
+            tracing::error!(subscription_id = %self.id, error = %e, "Streamer remove sub control message sending error");
         }
     }
 }