diff --git a/Cargo.lock b/Cargo.lock index 6336d25..06295b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1195,7 +1195,7 @@ dependencies = [ [[package]] name = "nddev-device-sync-observability" -version = "0.0.1-alpha.2" +version = "0.0.1-alpha.3" dependencies = [ "axum", "chrono", @@ -1223,7 +1223,7 @@ dependencies = [ [[package]] name = "nddev-device-sync-telemetry" -version = "0.0.1-alpha.3" +version = "0.0.1-alpha.4" dependencies = [ "android_log-sys", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 2b13a55..6a61342 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/sdk/Cargo.toml b/crates/sdk/Cargo.toml index fd2735c..0da2f70 100644 --- a/crates/sdk/Cargo.toml +++ b/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" diff --git a/crates/sdk/src/feed.rs b/crates/sdk/src/feed.rs new file mode 100644 index 0000000..29dc249 --- /dev/null +++ b/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>, + dropped: u64, + delivered: u64, + failures: u64, + instance: String, + sequence: u64, +} +pub struct EventFeed(Arc>); +#[derive(Default)] +pub(crate) struct Router(Mutex>>); +impl Router { + pub fn subscribe(&self) -> Result { + 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) { + 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, 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, + } + } +} diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 42a07e3..173cba0 100644 --- a/crates/sdk/src/lib.rs +++ b/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 { + feed: feed::Router, policy: Policy, sink: Mutex>, rejected: AtomicU64, @@ -382,6 +385,15 @@ impl Shared { 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 Clone for Control { } } impl Control { + pub fn subscribe(&self) -> Result { + 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( ) -> Result<(impl Subscriber + Send + Sync, Control), 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(), diff --git a/migrations/0002_client_sources.sql b/migrations/0002_client_sources.sql new file mode 100644 index 0000000..27effe2 --- /dev/null +++ b/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'; diff --git a/module.yaml b/module.yaml index f09ef37..66567fa 100644 --- a/module.yaml +++ b/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 diff --git a/scripts/check-compose.py b/scripts/check-compose.py index f381fe6..d9f5456 100644 --- a/scripts/check-compose.py +++ b/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') diff --git a/src/application.rs b/src/application.rs index 0d49d58..8c186b2 100644 --- a/src/application.rs +++ b/src/application.rs @@ -28,6 +28,7 @@ pub struct AcceptedEvent { pub route: Option, pub status: Option, pub timestamp_ms: i64, + pub device: Option, pub instance: Option, pub sequence: Option, } diff --git a/src/http.rs b/src/http.rs index 60d12ce..9381594 100644 --- a/src/http.rs +++ b/src/http.rs @@ -423,7 +423,7 @@ async fn snapshot(app: &App) -> Result { .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()) diff --git a/src/normalize.rs b/src/normalize.rs index f4ff89f..715e489 100644 --- a/src/normalize.rs +++ b/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, }; diff --git a/src/store.rs b/src/store.rs index 13a45c1..29095d7 100644 --- a/src/store.rs +++ b/src/store.rs @@ -83,12 +83,13 @@ impl PgRepository { .await .map_err(unavailable)?; repo.budget(&mut tx).await?; - let retained: Vec = - sqlx::query_scalar("SELECT service_name FROM sources WHERE tenant_id=$1 LIMIT 17") - .bind(&*repo.tenant) - .fetch_all(&mut *tx) - .await - .map_err(unavailable)?; + let retained: Vec = sqlx::query_scalar( + "SELECT service_name FROM sources WHERE tenant_id=$1 AND required LIMIT 17", + ) + .bind(&*repo.tenant) + .fetch_all(&mut *tx) + .await + .map_err(unavailable)?; let total: std::collections::BTreeSet<_> = retained.iter().chain(repo.expected.iter()).collect(); if total.len() > 16 { @@ -110,7 +111,7 @@ impl PgRepository { pub async fn ready(&self) -> Result<(), Error> { let current: bool = - sqlx::query_scalar("SELECT value='1' FROM nds_observability_meta WHERE key='schema'") + sqlx::query_scalar("SELECT value='2' FROM nds_observability_meta WHERE key='schema'") .fetch_one(&self.pool) .await .map_err(unavailable)?; @@ -305,7 +306,14 @@ impl Repository for PgRepository { let mut accepted = 0; let mut duplicates = 0; for event in events { - if event.service != OWN_SERVICE && !self.expected.iter().any(|s| s == &event.service) { + let client = matches!( + event.service.as_str(), + "nddev-device-sync-client" | "nddev-device-sync-agent" + ); + if !client + && event.service != OWN_SERVICE + && !self.expected.iter().any(|s| s == &event.service) + { return Err(Error::Invalid); } let inserted=sqlx::query("INSERT INTO event_receipts(tenant_id,event_id,expires_at,payload,received_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING") @@ -324,36 +332,96 @@ impl Repository for PgRepository { return Err(Error::Capacity); } } - let source=sqlx::query("SELECT instance_id::text,last_sequence FROM sources WHERE tenant_id=$1 AND service_name=$2 FOR UPDATE") - .bind(&*self.tenant).bind(&event.service).fetch_one(&mut *tx).await.map_err(unavailable)?; - let old_instance: Option = - source.try_get("instance_id").map_err(unavailable)?; - let old_sequence: Option = source.try_get("last_sequence").map_err(unavailable)?; - let gap = event.instance.is_some() - && old_instance == event.instance - && event + let (service, gap) = if client { + let device = event.device.as_deref().ok_or(Error::Invalid)?; + let instance = event.instance.as_deref().ok_or(Error::Invalid)?; + let sequence = event .sequence - .zip(old_sequence) - .is_some_and(|(a, b)| a != b.saturating_add(1)); - sqlx::query("UPDATE sources SET last_received_at=$3,last_event_at=$4,instance_id=$5::uuid,last_sequence=$6,gap_observations=LEAST(gap_observations+$7,9007199254740991),gap_active=$8 WHERE tenant_id=$1 AND service_name=$2") - .bind(&*self.tenant).bind(&event.service).bind(now).bind(instant(event.timestamp_ms)?).bind(&event.instance).bind(event.sequence).bind(if gap {1_i64}else{0_i64}).bind(gap).execute(&mut *tx).await.map_err(unavailable)?; - if gap { - self.condition( - &mut tx, - &event.service, - "source_gap", - Condition::Failed, - now, + .filter(|sequence| *sequence > 0) + .ok_or(Error::Invalid)?; + // Device-bound client sources have no always-on liveness promise. + // Producer rows separate concurrent app/CLI streams on one device. + use sha2::{Digest, Sha256}; + let digest = format!("{:x}", Sha256::digest(device.as_bytes())); + let service = format!("{}.{}", event.service, &digest[..24]); + let known: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM sources WHERE tenant_id=$1 AND service_name=$2)", ) - .await?; + .bind(&*self.tenant) + .bind(&service) + .fetch_one(&mut *tx) + .await + .map_err(unavailable)?; + if !known { + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM sources WHERE tenant_id=$1 AND NOT required", + ) + .bind(&*self.tenant) + .fetch_one(&mut *tx) + .await + .map_err(unavailable)?; + if count >= 64 { + return Err(Error::Capacity); + } + sqlx::query( + "INSERT INTO sources(tenant_id,service_name,required) VALUES($1,$2,false)", + ) + .bind(&*self.tenant) + .bind(&service) + .execute(&mut *tx) + .await + .map_err(unavailable)?; + } + sqlx::query("DELETE FROM client_producers WHERE tenant_id=$1 AND last_received_at<$2-interval '1 hour'").bind(&*self.tenant).bind(now).execute(&mut *tx).await.map_err(unavailable)?; + let previous:Option<(String,i64)>=sqlx::query_as("SELECT service_name,last_sequence FROM client_producers WHERE tenant_id=$1 AND instance_id=$2::uuid FOR UPDATE").bind(&*self.tenant).bind(instance).fetch_optional(&mut *tx).await.map_err(unavailable)?; + if previous + .as_ref() + .is_some_and(|(stored, last)| stored != &service || sequence <= *last) + { + return Err(Error::Invalid); + } + if previous.is_none() { + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM client_producers WHERE tenant_id=$1", + ) + .bind(&*self.tenant) + .fetch_one(&mut *tx) + .await + .map_err(unavailable)?; + if count >= 4096 { + return Err(Error::Capacity); + } + } + let gap = previous.is_some_and(|(_, last)| sequence != last.saturating_add(1)); + sqlx::query("INSERT INTO client_producers(tenant_id,instance_id,service_name,last_sequence,last_received_at) VALUES($1,$2::uuid,$3,$4,$5) ON CONFLICT(tenant_id,instance_id) DO UPDATE SET last_sequence=EXCLUDED.last_sequence,last_received_at=EXCLUDED.last_received_at").bind(&*self.tenant).bind(instance).bind(&service).bind(sequence).bind(now).execute(&mut *tx).await.map_err(unavailable)?; + sqlx::query("UPDATE sources SET last_received_at=$3,last_event_at=$4,gap_observations=LEAST(gap_observations+$5,9007199254740991),gap_active=$6 WHERE tenant_id=$1 AND service_name=$2").bind(&*self.tenant).bind(&service).bind(now).bind(instant(event.timestamp_ms)?).bind(if gap{1_i64}else{0_i64}).bind(gap).execute(&mut *tx).await.map_err(unavailable)?; + (service, gap) + } else { + let source=sqlx::query("SELECT instance_id::text,last_sequence FROM sources WHERE tenant_id=$1 AND service_name=$2 FOR UPDATE").bind(&*self.tenant).bind(&event.service).fetch_one(&mut *tx).await.map_err(unavailable)?; + let old_instance: Option = + source.try_get("instance_id").map_err(unavailable)?; + let old_sequence: Option = + source.try_get("last_sequence").map_err(unavailable)?; + let gap = event.instance.is_some() + && old_instance == event.instance + && event + .sequence + .zip(old_sequence) + .is_some_and(|(a, b)| a != b.saturating_add(1)); + sqlx::query("UPDATE sources SET last_received_at=$3,last_event_at=$4,instance_id=$5::uuid,last_sequence=$6,gap_observations=LEAST(gap_observations+$7,9007199254740991),gap_active=$8 WHERE tenant_id=$1 AND service_name=$2").bind(&*self.tenant).bind(&event.service).bind(now).bind(instant(event.timestamp_ms)?).bind(&event.instance).bind(event.sequence).bind(if gap{1_i64}else{0_i64}).bind(gap).execute(&mut *tx).await.map_err(unavailable)?; + (event.service.clone(), gap) + }; + if gap { + self.condition(&mut tx, &service, "source_gap", Condition::Failed, now) + .await?; } - if let Some((kind, state)) = domain::condition( + if let Some((kind, condition)) = domain::condition( &event.event_name, event.resource.as_deref(), event.route.as_deref(), event.status, ) { - self.condition(&mut tx, &event.service, kind, state, now) + self.condition(&mut tx, &service, kind, condition, now) .await?; } accepted += 1; diff --git a/standarts.lock b/standarts.lock index 05e5cff..ab62f6e 100644 --- a/standarts.lock +++ b/standarts.lock @@ -1,4 +1,4 @@ source_repo = "NDDev-OpenNetwork/nddev-device-sync" -source_release = "v0.0.1-alpha.9" -source_commit = "16a477beac6127adf73ef102df74a48451d607f1" +source_release = "unreleased" +source_commit = "c0efed7969e5b9961de57df62e08b41f09464463" license = "AGPL-3.0-only"