feat: collect bounded optional client streams

Cargo.toml

@@ -4,7 +4,7 @@ resolver = "3"
 
 [package]
 name = "nddev-device-sync-observability"
-version = "0.0.1-alpha.2"
+version = "0.0.1-alpha.3"
 edition = "2024"
 rust-version = "1.99"
 license = "AGPL-3.0-only"

crates/sdk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nddev-device-sync-telemetry"
-version = "0.0.1-alpha.3"
+version = "0.0.1-alpha.4"
 edition = "2024"
 rust-version = "1.99"
 license = "AGPL-3.0-only"

crates/sdk/src/feed.rs

@@ -0,0 +1,110 @@
+//! One caller-owned export feed of already validated/redacted events. This queue
+//! is not durable; callers report delivery separately from local OS output.
+use crate::Error;
+use std::{
+    collections::VecDeque,
+    sync::{Arc, Mutex, Weak},
+};
+const CAPACITY: usize = 128;
+const MAX_COUNTER: u64 = 9_007_199_254_740_991;
+#[derive(Clone, Copy, Default)]
+pub struct FeedHealth {
+    pub pending: u32,
+    pub dropped: u64,
+    pub delivered: u64,
+    pub failures: u64,
+}
+#[derive(Default)]
+struct State {
+    items: VecDeque<Vec<u8>>,
+    dropped: u64,
+    delivered: u64,
+    failures: u64,
+    instance: String,
+    sequence: u64,
+}
+pub struct EventFeed(Arc<Mutex<State>>);
+#[derive(Default)]
+pub(crate) struct Router(Mutex<Weak<Mutex<State>>>);
+impl Router {
+    pub fn subscribe(&self) -> Result<EventFeed, Error> {
+        let mut slot = self.0.lock().map_err(|_| Error)?;
+        if slot.upgrade().is_some() {
+            return Err(Error);
+        }
+        let state = Arc::new(Mutex::new(State {
+            instance: uuid::Uuid::new_v4().to_string(),
+            ..State::default()
+        }));
+        *slot = Arc::downgrade(&state);
+        Ok(EventFeed(state))
+    }
+    pub fn enqueue(&self, fields: &serde_json::Map<String, serde_json::Value>) {
+        let queue = self.0.lock().ok().and_then(|slot| slot.upgrade());
+        if let Some(queue) = queue {
+            let mut queue = queue.lock().unwrap_or_else(|error| error.into_inner());
+            if queue.sequence == MAX_COUNTER {
+                queue.dropped = queue.dropped.saturating_add(1).min(MAX_COUNTER);
+                return;
+            }
+            queue.sequence += 1;
+            if queue.items.len() == CAPACITY {
+                queue.dropped = queue.dropped.saturating_add(1).min(MAX_COUNTER);
+            } else {
+                let mut fields = fields.clone();
+                fields.insert("producer.instance_id".into(), queue.instance.clone().into());
+                fields.insert("producer.sequence".into(), queue.sequence.into());
+                if let Ok(mut bytes) = serde_json::to_vec(&fields) {
+                    bytes.push(b'\n');
+                    queue.items.push_back(bytes);
+                }
+            }
+        }
+    }
+}
+impl EventFeed {
+    /// Remove at most 32 complete NDJSON events / 64 KiB. The native sender owns
+    /// this one batch until acknowledgement, expiry or cancellation.
+    pub fn take_batch(&self) -> Option<(Vec<u8>, u32)> {
+        let mut state = self.0.lock().unwrap_or_else(|error| error.into_inner());
+        let mut bytes = Vec::new();
+        let mut count = 0;
+        while count < 32
+            && state
+                .items
+                .front()
+                .is_some_and(|item| bytes.len() + item.len() <= 65_536)
+        {
+            bytes.extend(state.items.pop_front()?);
+            count += 1;
+        }
+        (count > 0).then_some((bytes, count))
+    }
+    pub fn delivered(&self, count: u32) {
+        let mut state = self.0.lock().unwrap_or_else(|error| error.into_inner());
+        state.delivered = state
+            .delivered
+            .saturating_add(u64::from(count))
+            .min(MAX_COUNTER);
+    }
+    pub fn failed(&self) {
+        let mut state = self.0.lock().unwrap_or_else(|error| error.into_inner());
+        state.failures = state.failures.saturating_add(1).min(MAX_COUNTER);
+    }
+    pub fn discard(&self, count: u32) {
+        let mut state = self.0.lock().unwrap_or_else(|error| error.into_inner());
+        state.dropped = state
+            .dropped
+            .saturating_add(u64::from(count))
+            .min(MAX_COUNTER);
+    }
+    pub fn health(&self) -> FeedHealth {
+        let state = self.0.lock().unwrap_or_else(|error| error.into_inner());
+        FeedHealth {
+            pending: state.items.len() as u32,
+            dropped: state.dropped,
+            delivered: state.delivered,
+            failures: state.failures,
+        }
+    }
+}

crates/sdk/src/lib.rs

@@ -29,6 +29,8 @@ use tracing_subscriber::{
     registry::LookupSpan,
 };
 
+mod feed;
+pub use feed::{EventFeed, FeedHealth};
 mod output;
 pub use output::{LocalWriter, OutputHealth};
 mod http;
@@ -326,6 +328,7 @@ struct ActiveDebug {
     remaining: u64,
 }
 struct Shared<W> {
+    feed: feed::Router,
     policy: Policy,
     sink: Mutex<Sink<W>>,
     rejected: AtomicU64,
@@ -382,6 +385,15 @@ impl<W: Write> Shared<W> {
             return;
         }
         bytes.push(b'\n');
+        // Delivery operations keep local diagnostics but never feed themselves.
+        if !fields
+            .get("route")
+            .and_then(Value::as_str)
+            .is_some_and(|route| route.starts_with("/v2/observability/"))
+            && fields.get("module").and_then(Value::as_str) != Some("telemetry")
+        {
+            self.feed.enqueue(&fields);
+        }
         if sink.writer.write_all(&bytes).is_err() {
             self.write_failures.fetch_add(1, Ordering::Relaxed);
         }
@@ -592,6 +604,9 @@ impl<W: Write> Clone for Control<W> {
     }
 }
 impl<W: Write> Control<W> {
+    pub fn subscribe(&self) -> Result<EventFeed, Error> {
+        self.shared.feed.subscribe()
+    }
     pub fn report_client_lifecycle(&self, state: view::ClientLifecycle) {
         self.shared.emit(
             Map::from_iter([
@@ -712,6 +727,7 @@ pub fn subscriber<W: Write + Send + Sync + 'static>(
 ) -> Result<(impl Subscriber + Send + Sync, Control<W>), Error> {
     let policy = Policy::new(&config)?;
     let shared = Arc::new(Shared {
+        feed: feed::Router::default(),
         policy,
         sink: Mutex::new(Sink {
             writer,
@@ -916,6 +932,48 @@ mod tests {
             Ok(())
         }
     }
+    #[test]
+    fn private_fields_and_delivery_feedback_stay_out_of_bounded_export_feed() {
+        let output = Arc::new(Mutex::new(Vec::new()));
+        let provider = SdkTracerProvider::builder().build();
+        let (subscriber, control) =
+            subscriber(config(), Capture(output), provider.tracer("export-feed")).unwrap();
+        let feed = control.subscribe().unwrap();
+        assert!(control.subscribe().is_err());
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::info!(
+                event.name = "client.acceptance",
+                authorization = "PRIVATE_AUTHORIZATION_MUST_NEVER_EXPORT",
+                outcome = "ok"
+            );
+            tracing::info!(
+                event.name = "identity.request_completed",
+                route = "/v2/observability/events",
+                outcome = "ok"
+            );
+        });
+        for _ in 0..300 {
+            control.report_client_lifecycle(view::ClientLifecycle::Ready);
+        }
+        assert!(feed.health().pending <= 128 && feed.health().dropped > 0);
+        let mut events = 0;
+        while let Some((bytes, count)) = feed.take_batch() {
+            assert!(bytes.len() <= 65_536 && count <= 32);
+            let text = String::from_utf8(bytes).unwrap();
+            assert!(
+                !text.contains("PRIVATE_AUTHORIZATION") && !text.contains("/v2/observability/")
+            );
+            for line in text.lines() {
+                decode_event(serde_json::from_str(line).unwrap()).unwrap();
+            }
+            feed.delivered(count);
+            events += count;
+        }
+        assert_eq!(feed.health().delivered, u64::from(events));
+        drop(feed);
+        assert!(control.subscribe().is_ok());
+    }
+
     fn config() -> Config {
         Config {
             service: "generated-sdk-acceptance".into(),

migrations/0002_client_sources.sql

@@ -0,0 +1,16 @@
+DO $$ BEGIN
+    IF NOT EXISTS(SELECT 1 FROM nds_observability_meta WHERE key='schema' AND value='1') THEN
+        RAISE EXCEPTION 'client streams require observability schema 1';
+    END IF;
+END $$;
+ALTER TABLE sources ADD COLUMN required boolean NOT NULL DEFAULT true;
+CREATE TABLE client_producers (
+    tenant_id text NOT NULL REFERENCES delivery_state(tenant_id),
+    instance_id uuid NOT NULL,
+    service_name text NOT NULL CHECK(length(service_name) BETWEEN 1 AND 128),
+    last_sequence bigint NOT NULL CHECK(last_sequence BETWEEN 1 AND 9007199254740991),
+    last_received_at timestamptz NOT NULL,
+    PRIMARY KEY(tenant_id,instance_id),
+    FOREIGN KEY(tenant_id,service_name) REFERENCES sources(tenant_id,service_name)
+);
+UPDATE nds_observability_meta SET value='2' WHERE key='schema';

module.yaml

@@ -1,6 +1,6 @@
 id: observability
 name: nddev-device-sync-observability
-version: 0.0.1-alpha.2
+version: 0.0.1-alpha.3
 channel: alpha
 license: AGPL-3.0-only
 repository: NDDev-OpenNetwork/nddev-device-sync-observability

scripts/check-compose.py

@@ -194,6 +194,21 @@ def ingest(events):
             observed=eventually(lambda:search('nds','logs',"trace_id='"+failure['trace_id']+"'"))
             assert any(row.get('error_type',row.get('error.type'))=='flutter_framework_error' for row in observed)
             print('Actual shared SDK client failure searchable: one incident, duplicate replay retained, readiness resolved it')
+            # The real SDK's frames pass the same context enrichment boundary
+            # supplied by an authenticated gateway, not an always-on source.
+            import hashlib
+            device=str(uuid.uuid4())
+            device_source='nddev-device-sync-client.'+hashlib.sha256(device.encode()).hexdigest()[:24]
+            device_cycle=[event | {'service.name':'nddev-device-sync-client','device.id':device,'producer.instance_id':str(uuid.uuid5(uuid.NAMESPACE_OID,event['producer.instance_id']+device))} for event in cycle]
+            assert ingest(device_cycle[:ready_index])['accepted']==ready_index
+            device_incident=eventually(lambda:next((row for row in incidents() if row['service_name']==device_source and row['kind']=='process' and row['state']=='open'),None))
+            assert ingest(device_cycle[:ready_index])['duplicates']==ready_index
+            assert ingest(device_cycle[ready_index:])['accepted']==len(device_cycle)-ready_index
+            eventually(lambda:any(row['incident_id']==device_incident['incident_id'] and row['state']=='resolved' for row in incidents()))
+            assert device_source not in {source['service_name'] for source in status()['sources']}
+            assert sql('postgres',"SELECT count(*) FROM client_producers WHERE service_name='"+device_source+"'")== '1'
+            print('Actual SDK client stream: device-scoped incident, producer replay/sequence and optional liveness passed')
+
             if client_events:
                 for start in range(0,len(client_events),100):ingest(client_events[start:start+100])
                 for kind in ['flutter_framework_error','dart_unhandled_error']:
@@ -286,14 +301,14 @@ def recovered_source():
             registration_name=project+'-registration'
             compose('run','--detach','--rm','--no-deps','--name',registration_name,'--env','NDS_EXPECTED_SOURCES='+retained,'--env','NDS_TELEMETRY_ENABLED=false','observability')
             try:
-                eventually(lambda:sql('postgres','SELECT count(*) FROM sources')=='16')
+                eventually(lambda:sql('postgres','SELECT count(*) FROM sources WHERE required')=='16')
             finally:command('docker','stop','--time','15',registration_name,check=False,timeout=20)
             compose('restart','observability',timeout=30)
             eventually(lambda:request(obs+'/v1/ready')[0]==200)
             sources=status()['sources'];assert len(sources)==16 and sum(row['state']=='inactive' for row in sources)==16-len(source_names)
             rejected=compose('run','--rm','--no-deps','--env','NDS_EXPECTED_SOURCES=unregistered-source','--env','NDS_TELEMETRY_ENABLED=false','observability',check=False,timeout=20)
             assert rejected.returncode!=0
-            assert sql('postgres','SELECT count(*) FROM sources')=='16'
+            assert sql('postgres','SELECT count(*) FROM sources WHERE required')=='16'
             print('Real PostgreSQL source registration: canonical set capped at16, retained inactive rows visible after restart,17th source refused transactionally')
             sql('postgres',"UPDATE delivery_state SET time_floor=clock_timestamp()+interval '2 hours'")
             eventually(lambda:sql('postgres','SELECT count(*) FROM incident_actions')=='0')

src/application.rs

@@ -28,6 +28,7 @@ pub struct AcceptedEvent {
     pub route: Option<String>,
     pub status: Option<i32>,
     pub timestamp_ms: i64,
+    pub device: Option<String>,
     pub instance: Option<String>,
     pub sequence: Option<i64>,
 }

src/http.rs

@@ -423,7 +423,7 @@ async fn snapshot(app: &App) -> Result<wire::ObservabilityStatus, Error> {
         .await
         .map_err(|_| Error::Unavailable)?;
     let rows = sqlx::query(
-        "SELECT *,service_name=ANY($2) AS configured FROM sources WHERE tenant_id=$1 ORDER BY service_name LIMIT 16",
+        "SELECT *,service_name=ANY($2) AS configured FROM sources WHERE tenant_id=$1 AND required ORDER BY service_name LIMIT 16",
     )
     .bind(&*app.repository.tenant)
     .bind(app.repository.expected.as_ref())

src/normalize.rs

@@ -33,6 +33,7 @@ pub fn accept(
         route: event.route.clone(),
         status: event.status,
         timestamp_ms: time.timestamp_millis(),
+        device: event.device_id.clone(),
         instance: event.producer_instance_id.clone(),
         sequence: event.producer_sequence,
     };