From ed986e7f6a5bc064c5a1e7538e465271aa92265b Mon Sep 17 00:00:00 2001 From: tiammomo Date: Sun, 23 Aug 2026 14:27:28 +0800 Subject: [PATCH 1/3] feat(runtime): persist compute inventory snapshots --- Cargo.lock | 1 + Cargo.toml | 1 + docs/RUNTIME_ADAPTER.md | 12 +- migrations/0014_runtime_compute_snapshots.sql | 25 + src/enterprise_ledger.rs | 4 + src/enterprise_ledger/compute_inventory.rs | 494 ++++++++++++++++++ src/runtime_adapter.rs | 11 + src/runtime_adapter/client.rs | 13 +- 8 files changed, 549 insertions(+), 12 deletions(-) create mode 100644 migrations/0014_runtime_compute_snapshots.sql create mode 100644 src/enterprise_ledger/compute_inventory.rs diff --git a/Cargo.lock b/Cargo.lock index 0d20708..c599134 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1396,6 +1396,7 @@ dependencies = [ "argon2", "async-stream", "axum", + "chrono", "futures-util", "httpdate", "jsonschema", diff --git a/Cargo.toml b/Cargo.toml index 7789956..f8035e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ strip = "symbols" async-stream = "0.3" argon2 = "0.5" axum = { version = "0.8", features = ["macros"] } +chrono = { version = "0.4", default-features = false, features = ["std"] } futures-util = "0.3" httpdate = "1" jsonschema = { version = "0.48", default-features = false } diff --git a/docs/RUNTIME_ADAPTER.md b/docs/RUNTIME_ADAPTER.md index 467b2a8..e87a355 100644 --- a/docs/RUNTIME_ADAPTER.md +++ b/docs/RUNTIME_ADAPTER.md @@ -42,6 +42,14 @@ credential is not serializable and is redacted from debug output and upstream errors. Configuration-file loading, scheduling, and storage are intentionally outside this client boundary. +Accepted Compute observations can be persisted through the Enterprise Ledger. +PostgreSQL keeps immutable, identity-constrained snapshot history and the test +backend provides matching in-memory semantics. Exact retries are idempotent; +conflicting reuse of an adapter snapshot identity or observation timestamp is +rejected. The latest query derives `fresh`, `stale`, or `unavailable` from the +validated observation time, server time, and a bounded server-owned policy. +Derived state never changes the stored Runtime Adapter document. + ## Capability Rules Every document has `apiVersion: runtime.modelport.io/v1alpha1`, @@ -100,8 +108,8 @@ reinterpreted across versions. Additive experimental data belongs in `local-inference-stack` checker remains an explicitly selected compatibility mode, not the source of this contract. -Configuration integration, collection policy, persistence, derived freshness, -admin APIs, and all writes remain deferred to reviewed Issues. Offline +Configuration integration, collection scheduling, retention, admin APIs, and +all writes remain deferred to reviewed Issues. Offline validation cannot start a process, download a model, access a GPU, or call a network endpoint; the collection client performs only the two advertised safe reads requested by its caller. diff --git a/migrations/0014_runtime_compute_snapshots.sql b/migrations/0014_runtime_compute_snapshots.sql new file mode 100644 index 0000000..db2fbc7 --- /dev/null +++ b/migrations/0014_runtime_compute_snapshots.sql @@ -0,0 +1,25 @@ +CREATE TABLE modelport_runtime_compute_snapshots ( + adapter_id TEXT NOT NULL, + snapshot_id TEXT NOT NULL, + observed_at TIMESTAMPTZ NOT NULL, + observed_at_key TEXT NOT NULL, + accepted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + document JSONB NOT NULL, + PRIMARY KEY (adapter_id, snapshot_id), + CONSTRAINT modelport_runtime_compute_snapshots_observation_unique + UNIQUE (adapter_id, observed_at_key), + CONSTRAINT modelport_runtime_compute_snapshots_identity_check CHECK ( + length(adapter_id) BETWEEN 1 AND 63 + AND length(snapshot_id) BETWEEN 1 AND 160 + AND length(observed_at_key) = 30 + AND observed_at_key ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{9}Z$' + AND jsonb_typeof(document) = 'object' + AND document #>> '{metadata,adapterId}' = adapter_id + AND document #>> '{metadata,snapshotId}' = snapshot_id + AND observed_at = (document #>> '{metadata,observedAt}')::timestamptz + AND observed_at = observed_at_key::timestamptz + ) +); + +CREATE INDEX modelport_runtime_compute_snapshots_latest_idx + ON modelport_runtime_compute_snapshots (adapter_id, observed_at_key DESC, accepted_at DESC); diff --git a/src/enterprise_ledger.rs b/src/enterprise_ledger.rs index b085177..3c9e324 100644 --- a/src/enterprise_ledger.rs +++ b/src/enterprise_ledger.rs @@ -42,6 +42,8 @@ use crate::{ usage::{current_period, quota_increment}, }; +pub(crate) mod compute_inventory; + const DEFAULT_LEASE_TTL_SECS: u64 = 300; const DEFAULT_RECONCILE_INTERVAL_SECS: u64 = 60; const MIN_LEASE_TTL_SECS: u64 = 30; @@ -79,6 +81,8 @@ struct MemoryLedger { usage_reservations: HashMap, budget_events: Vec, audit_events: Vec, + #[allow(dead_code)] // Used by the staged collection integration after this storage seam. + runtime_compute_snapshots: HashMap<(String, String), compute_inventory::MemoryComputeSnapshot>, ops_incidents: BTreeMap, ops_event_index: HashMap, ops_heartbeats: BTreeMap, diff --git a/src/enterprise_ledger/compute_inventory.rs b/src/enterprise_ledger/compute_inventory.rs new file mode 100644 index 0000000..07f8e90 --- /dev/null +++ b/src/enterprise_ledger/compute_inventory.rs @@ -0,0 +1,494 @@ +#![allow(dead_code)] // The scheduler/admin integration follows this reviewed storage slice. + +use std::time::Duration; + +use chrono::{DateTime, SecondsFormat, Utc}; +use serde::Serialize; +use serde_json::Value; +use sqlx::Row; + +use super::{EnterpriseLedger, LedgerBackend, now_millis}; +use crate::{ + AppError, + runtime_adapter::{ + RuntimeAdapterComputeInventory, is_valid_runtime_adapter_id, + validate_runtime_adapter_compute_inventory, + }, +}; + +const MAX_FUTURE_SKEW_MS: i64 = 5 * 60 * 1_000; +const MAX_STALE_AFTER_MS: i64 = 7 * 24 * 60 * 60 * 1_000; + +#[derive(Debug, Clone)] +pub(super) struct MemoryComputeSnapshot { + document: RuntimeAdapterComputeInventory, + value: Value, + observed_at_key: String, + observed_at_ms: i64, + accepted_at_ms: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RuntimeComputeSnapshotWrite { + Inserted, + Idempotent, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum RuntimeComputeFreshness { + Fresh, + Stale, + Unavailable, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RuntimeComputeInventoryState { + pub(crate) freshness: RuntimeComputeFreshness, + pub(crate) inventory: Option, + pub(crate) observed_at_ms: Option, + pub(crate) accepted_at_ms: Option, + pub(crate) age_ms: Option, +} + +struct PreparedSnapshot { + document: RuntimeAdapterComputeInventory, + value: Value, + observed_at_key: String, + observed_at_ms: i64, + accepted_at_ms: i64, +} + +impl EnterpriseLedger { + pub(crate) async fn persist_runtime_compute_inventory( + &self, + inventory: &RuntimeAdapterComputeInventory, + ) -> Result { + let snapshot = prepare_snapshot(inventory, now_millis())?; + match self.backend.as_ref() { + LedgerBackend::Memory(ledger) => { + let mut ledger = ledger.lock().expect("enterprise ledger lock poisoned"); + let conflicting = ledger.runtime_compute_snapshots.values().find(|existing| { + existing.document.metadata.adapter_id == snapshot.document.metadata.adapter_id + && (existing.document.metadata.snapshot_id + == snapshot.document.metadata.snapshot_id + || existing.observed_at_key == snapshot.observed_at_key) + }); + if let Some(existing) = conflicting { + if existing.value == snapshot.value + && existing.document.metadata.snapshot_id + == snapshot.document.metadata.snapshot_id + { + return Ok(RuntimeComputeSnapshotWrite::Idempotent); + } + return Err(snapshot_conflict()); + } + let key = ( + snapshot.document.metadata.adapter_id.clone(), + snapshot.document.metadata.snapshot_id.clone(), + ); + ledger.runtime_compute_snapshots.insert( + key, + MemoryComputeSnapshot { + document: snapshot.document, + value: snapshot.value, + observed_at_key: snapshot.observed_at_key, + observed_at_ms: snapshot.observed_at_ms, + accepted_at_ms: snapshot.accepted_at_ms, + }, + ); + Ok(RuntimeComputeSnapshotWrite::Inserted) + } + LedgerBackend::Postgres(pool) => { + let inserted = sqlx::query_scalar::<_, String>( + "INSERT INTO modelport_runtime_compute_snapshots ( + adapter_id, snapshot_id, observed_at, observed_at_key, + accepted_at, document + ) VALUES ($1, $2, $3::timestamptz, $4, + to_timestamp($5::double precision / 1000.0), $6) + ON CONFLICT DO NOTHING + RETURNING snapshot_id", + ) + .bind(&snapshot.document.metadata.adapter_id) + .bind(&snapshot.document.metadata.snapshot_id) + .bind(&snapshot.document.metadata.observed_at) + .bind(&snapshot.observed_at_key) + .bind(snapshot.accepted_at_ms) + .bind(&snapshot.value) + .fetch_optional(pool) + .await?; + if inserted.is_some() { + return Ok(RuntimeComputeSnapshotWrite::Inserted); + } + + let existing = sqlx::query( + "SELECT snapshot_id, document + FROM modelport_runtime_compute_snapshots + WHERE adapter_id = $1 + AND (snapshot_id = $2 OR observed_at_key = $3) + LIMIT 1", + ) + .bind(&snapshot.document.metadata.adapter_id) + .bind(&snapshot.document.metadata.snapshot_id) + .bind(&snapshot.observed_at_key) + .fetch_optional(pool) + .await?; + if existing.is_some_and(|row| { + row.try_get::("snapshot_id").ok() + == Some(snapshot.document.metadata.snapshot_id.clone()) + && row.try_get::("document").ok() == Some(snapshot.value) + }) { + Ok(RuntimeComputeSnapshotWrite::Idempotent) + } else { + Err(snapshot_conflict()) + } + } + } + } + + pub(crate) async fn latest_runtime_compute_inventory( + &self, + adapter_id: &str, + stale_after: Duration, + ) -> Result { + validate_query(adapter_id, stale_after)?; + let latest = match self.backend.as_ref() { + LedgerBackend::Memory(ledger) => ledger + .lock() + .expect("enterprise ledger lock poisoned") + .runtime_compute_snapshots + .values() + .filter(|snapshot| snapshot.document.metadata.adapter_id == adapter_id) + .max_by_key(|snapshot| (&snapshot.observed_at_key, snapshot.accepted_at_ms)) + .cloned(), + LedgerBackend::Postgres(pool) => { + let row = sqlx::query( + "SELECT document, observed_at_key, + (EXTRACT(EPOCH FROM accepted_at) * 1000)::bigint AS accepted_at_ms + FROM modelport_runtime_compute_snapshots + WHERE adapter_id = $1 + ORDER BY observed_at_key DESC, accepted_at DESC + LIMIT 1", + ) + .bind(adapter_id) + .fetch_optional(pool) + .await?; + row.map(|row| -> Result { + let value: Value = row.try_get("document")?; + let document = validate_runtime_adapter_compute_inventory(&value.to_string()) + .map_err(|_| { + AppError::Database( + "stored Runtime Adapter Compute snapshot is invalid".to_owned(), + ) + })?; + let stored_key: String = row.try_get("observed_at_key")?; + let (observed_at_ms, expected_key) = + parse_observed_at(&document.metadata.observed_at)?; + if stored_key != expected_key { + return Err(AppError::Database( + "stored Runtime Adapter observation key is invalid".to_owned(), + )); + } + Ok(MemoryComputeSnapshot { + document, + value, + observed_at_key: stored_key, + observed_at_ms, + accepted_at_ms: row.try_get("accepted_at_ms")?, + }) + }) + .transpose()? + } + }; + Ok(project_freshness(latest, stale_after, now_millis())) + } +} + +fn prepare_snapshot( + inventory: &RuntimeAdapterComputeInventory, + accepted_at_ms: i64, +) -> Result { + let value = serde_json::to_value(inventory)?; + let document = validate_runtime_adapter_compute_inventory(&value.to_string())?; + let (observed_at_ms, observed_at_key) = parse_observed_at(&document.metadata.observed_at)?; + if observed_at_ms > accepted_at_ms.saturating_add(MAX_FUTURE_SKEW_MS) { + return Err(AppError::InvalidRequest( + "Runtime Adapter Compute observedAt exceeds the server clock-skew allowance".to_owned(), + )); + } + Ok(PreparedSnapshot { + document, + value, + observed_at_key, + observed_at_ms, + accepted_at_ms, + }) +} + +fn parse_observed_at(observed_at: &str) -> Result<(i64, String), AppError> { + let observed_at = DateTime::parse_from_rfc3339(observed_at) + .map_err(|_| AppError::InvalidRequest("Compute observedAt is invalid".to_owned()))?; + Ok(( + observed_at.timestamp_millis(), + observed_at + .with_timezone(&Utc) + .to_rfc3339_opts(SecondsFormat::Nanos, true), + )) +} + +fn validate_query(adapter_id: &str, stale_after: Duration) -> Result<(), AppError> { + let stale_after_ms = i64::try_from(stale_after.as_millis()).unwrap_or(i64::MAX); + if !is_valid_runtime_adapter_id(adapter_id) { + return Err(AppError::InvalidRequest( + "Runtime Adapter ID is invalid".to_owned(), + )); + } + if !(1..=MAX_STALE_AFTER_MS).contains(&stale_after_ms) { + return Err(AppError::InvalidRequest( + "Runtime Adapter stale-after policy must be between 1 millisecond and 7 days" + .to_owned(), + )); + } + Ok(()) +} + +fn project_freshness( + latest: Option, + stale_after: Duration, + now_ms: i64, +) -> RuntimeComputeInventoryState { + let Some(snapshot) = latest else { + return RuntimeComputeInventoryState { + freshness: RuntimeComputeFreshness::Unavailable, + inventory: None, + observed_at_ms: None, + accepted_at_ms: None, + age_ms: None, + }; + }; + let age_ms = u64::try_from(now_ms.saturating_sub(snapshot.observed_at_ms)).unwrap_or_default(); + let freshness = if u128::from(age_ms) <= stale_after.as_millis() { + RuntimeComputeFreshness::Fresh + } else { + RuntimeComputeFreshness::Stale + }; + RuntimeComputeInventoryState { + freshness, + inventory: Some(snapshot.document), + observed_at_ms: Some(snapshot.observed_at_ms), + accepted_at_ms: Some(snapshot.accepted_at_ms), + age_ms: Some(age_ms), + } +} + +fn snapshot_conflict() -> AppError { + AppError::StateConflict( + "Runtime Adapter snapshot identity or observedAt was reused with different content" + .to_owned(), + ) +} + +#[cfg(test)] +mod tests { + use std::time::SystemTime; + + use super::*; + + const INVENTORY: &str = include_str!( + "../../tests/fixtures/runtime-adapters/qwen-llama-cpp-compute-inventory-v1alpha1.json" + ); + + #[tokio::test] + async fn memory_snapshots_are_append_only_idempotent_and_latest_by_observation() { + let ledger = EnterpriseLedger::memory(); + let newest = current_fixture("snapshot:qwen-newest"); + assert_eq!( + ledger + .persist_runtime_compute_inventory(&newest) + .await + .unwrap(), + RuntimeComputeSnapshotWrite::Inserted + ); + assert_eq!( + ledger + .persist_runtime_compute_inventory(&newest) + .await + .unwrap(), + RuntimeComputeSnapshotWrite::Idempotent + ); + + let mut older = newest.clone(); + older.metadata.snapshot_id = "snapshot:qwen-older".to_owned(); + older.metadata.observed_at = "2026-08-23T03:59:00Z".to_owned(); + ledger + .persist_runtime_compute_inventory(&older) + .await + .unwrap(); + let latest = ledger + .latest_runtime_compute_inventory( + "qwen-llama-cpp-reference", + Duration::from_secs(7 * 24 * 60 * 60), + ) + .await + .unwrap(); + assert_eq!(latest.freshness, RuntimeComputeFreshness::Fresh); + assert_eq!( + latest.inventory.unwrap().metadata.snapshot_id, + newest.metadata.snapshot_id + ); + assert!(latest.observed_at_ms.is_some()); + assert!(latest.accepted_at_ms.is_some()); + assert!(latest.age_ms.is_some()); + } + + #[tokio::test] + async fn memory_snapshots_reject_conflicts_and_excessive_future_time() { + let ledger = EnterpriseLedger::memory(); + let inventory = fixture(); + ledger + .persist_runtime_compute_inventory(&inventory) + .await + .unwrap(); + let mut conflict = inventory.clone(); + conflict.nodes[0].gpus[0].memory.available_bytes -= 1; + assert!(matches!( + ledger.persist_runtime_compute_inventory(&conflict).await, + Err(AppError::StateConflict(_)) + )); + + let mut future = inventory; + future.metadata.snapshot_id = "snapshot:qwen-future".to_owned(); + future.metadata.observed_at = "2999-01-01T00:00:00Z".to_owned(); + assert!(matches!( + ledger.persist_runtime_compute_inventory(&future).await, + Err(AppError::InvalidRequest(_)) + )); + } + + #[tokio::test] + async fn freshness_is_server_owned_and_unavailable_without_a_snapshot() { + let ledger = EnterpriseLedger::memory(); + let unavailable = ledger + .latest_runtime_compute_inventory("missing-adapter", Duration::from_secs(120)) + .await + .unwrap(); + assert_eq!(unavailable.freshness, RuntimeComputeFreshness::Unavailable); + assert!(unavailable.inventory.is_none()); + + ledger + .persist_runtime_compute_inventory(&fixture()) + .await + .unwrap(); + let stale = ledger + .latest_runtime_compute_inventory("qwen-llama-cpp-reference", Duration::from_millis(1)) + .await + .unwrap(); + assert_eq!(stale.freshness, RuntimeComputeFreshness::Stale); + assert!( + !serde_json::to_value(stale.inventory.unwrap()) + .unwrap() + .to_string() + .contains("freshness") + ); + assert!( + ledger + .latest_runtime_compute_inventory("qwen-llama-cpp-reference", Duration::ZERO,) + .await + .is_err() + ); + } + + #[tokio::test] + async fn postgres_state_runtime_compute_inventory_snapshots_round_trip() { + let Ok(database_url) = std::env::var("MODELPORT_TEST_DATABASE_URL") else { + return; + }; + let ledger = EnterpriseLedger::postgres_for_tests(&database_url) + .await + .unwrap(); + let inventory = + current_fixture(&format!("snapshot:test-{}", uuid::Uuid::new_v4().simple())); + let first = ledger + .persist_runtime_compute_inventory(&inventory) + .await + .unwrap(); + assert!(matches!( + first, + RuntimeComputeSnapshotWrite::Inserted | RuntimeComputeSnapshotWrite::Idempotent + )); + assert_eq!( + ledger + .persist_runtime_compute_inventory(&inventory) + .await + .unwrap(), + RuntimeComputeSnapshotWrite::Idempotent + ); + let LedgerBackend::Postgres(pool) = ledger.backend.as_ref() else { + unreachable!(); + }; + let stored = sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM modelport_runtime_compute_snapshots + WHERE adapter_id = $1 AND snapshot_id = $2", + ) + .bind(&inventory.metadata.adapter_id) + .bind(&inventory.metadata.snapshot_id) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(stored, 1); + let latest = ledger + .latest_runtime_compute_inventory( + "qwen-llama-cpp-reference", + Duration::from_secs(7 * 24 * 60 * 60), + ) + .await + .unwrap(); + assert_eq!(latest.inventory.unwrap(), inventory); + + let mut conflict = inventory.clone(); + conflict.nodes[0].gpus[0].memory.available_bytes -= 1; + assert!(matches!( + ledger.persist_runtime_compute_inventory(&conflict).await, + Err(AppError::StateConflict(_)) + )); + assert!( + sqlx::query( + "INSERT INTO modelport_runtime_compute_snapshots ( + adapter_id, snapshot_id, observed_at, observed_at_key, document + ) VALUES ('wrong-adapter', $1, $2::timestamptz, $3, $4)", + ) + .bind(&inventory.metadata.snapshot_id) + .bind(&inventory.metadata.observed_at) + .bind( + parse_observed_at(&inventory.metadata.observed_at) + .unwrap() + .1 + ) + .bind(serde_json::to_value(&inventory).unwrap()) + .execute(pool) + .await + .is_err() + ); + } + + fn fixture() -> RuntimeAdapterComputeInventory { + validate_runtime_adapter_compute_inventory(INVENTORY).unwrap() + } + + fn current_fixture(snapshot_id: &str) -> RuntimeAdapterComputeInventory { + let mut inventory = fixture(); + inventory.metadata.snapshot_id = snapshot_id.to_owned(); + inventory.metadata.observed_at = + DateTime::::from(SystemTime::now()).to_rfc3339_opts(SecondsFormat::Millis, true); + inventory + } + + #[test] + fn canonical_observation_keys_preserve_instants_and_nanosecond_order() { + let (_, offset) = parse_observed_at("2026-08-23T12:00:00.000000001+08:00").unwrap(); + let (_, utc) = parse_observed_at("2026-08-23T04:00:00.000000001Z").unwrap(); + let (_, later) = parse_observed_at("2026-08-23T04:00:00.000000002Z").unwrap(); + assert_eq!(offset, utc); + assert!(later > utc); + } +} diff --git a/src/runtime_adapter.rs b/src/runtime_adapter.rs index 025c531..1493f27 100644 --- a/src/runtime_adapter.rs +++ b/src/runtime_adapter.rs @@ -19,6 +19,17 @@ pub const RUNTIME_ADAPTER_CAPABILITIES_SCHEMA: &str = pub const RUNTIME_ADAPTER_COMPUTE_INVENTORY_SCHEMA: &str = include_str!("../resources/schemas/runtime-adapter-compute-inventory-v1alpha1.schema.json"); +pub(crate) fn is_valid_runtime_adapter_id(adapter_id: &str) -> bool { + let bytes = adapter_id.as_bytes(); + let edge = |byte: u8| byte.is_ascii_lowercase() || byte.is_ascii_digit(); + (1..=63).contains(&bytes.len()) + && edge(bytes[0]) + && edge(bytes[bytes.len() - 1]) + && bytes + .iter() + .all(|byte| edge(*byte) || matches!(byte, b'.' | b'_' | b'-')) +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(untagged)] pub enum RuntimeAdapterDocument { diff --git a/src/runtime_adapter/client.rs b/src/runtime_adapter/client.rs index 7b41abe..a6fe8e2 100644 --- a/src/runtime_adapter/client.rs +++ b/src/runtime_adapter/client.rs @@ -8,7 +8,8 @@ use crate::{ runtime_adapter::{ RuntimeAdapterAuthenticationScheme, RuntimeAdapterCapabilities, RuntimeAdapterComputeInventory, RuntimeAdapterOperationId, RuntimeAdapterTransport, - validate_runtime_adapter_capabilities, validate_runtime_adapter_compute_inventory, + is_valid_runtime_adapter_id, validate_runtime_adapter_capabilities, + validate_runtime_adapter_compute_inventory, }, }; @@ -186,15 +187,7 @@ impl RuntimeAdapterClient { } fn validate_adapter_id(adapter_id: &str) -> Result<(), AppError> { - let bytes = adapter_id.as_bytes(); - let edge = |byte: u8| byte.is_ascii_lowercase() || byte.is_ascii_digit(); - if !(1..=63).contains(&bytes.len()) - || !edge(bytes[0]) - || !edge(bytes[bytes.len() - 1]) - || !bytes - .iter() - .all(|byte| edge(*byte) || matches!(byte, b'.' | b'_' | b'-')) - { + if !is_valid_runtime_adapter_id(adapter_id) { return Err(AppError::Config( "Runtime Adapter ID must match the v1alpha1 identifier format".to_owned(), )); From 5612fb43b36023607725602fca0a1ce7828c439b Mon Sep 17 00:00:00 2001 From: lovett55 Date: Tue, 8 Sep 2026 19:01:55 +0800 Subject: [PATCH 2/3] feat(runtime): collect adapter compute inventory Signed-off-by: lovett55 --- CHANGELOG.md | 6 + docs/ARCHITECTURE.md | 7 +- docs/OPERATIONS.md | 12 + docs/RUNTIME_ADAPTER.md | 18 +- src/enterprise_ledger.rs | 1 - src/enterprise_ledger/compute_inventory.rs | 5 +- src/metrics.rs | 73 ++- src/runtime_adapter.rs | 1 + src/runtime_adapter/collector.rs | 642 +++++++++++++++++++++ src/server.rs | 24 +- 10 files changed, 776 insertions(+), 13 deletions(-) create mode 100644 src/runtime_adapter/collector.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index ed5736a..97a7aeb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable ModelPort changes are recorded here. The project follows ## [Unreleased] +### Runtime Adapter collection + +- Collect validated Compute inventory immediately and periodically for every + enabled Runtime Adapter, with per-adapter isolation, a shared concurrency + bound, sanitized metrics and logs, bounded retry, and graceful shutdown. + ## [0.1.3] - 2026-09-07 ### Release correction diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 55f0fea..3ada8e4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -101,11 +101,11 @@ they are not eight top-level navigation entries: | --- | --- | --- | --- | | Models | Catalog identities, capabilities, limits, compatibility, and rate cards | Models | Partial: Provider-scoped inventory and logical catalog ship today | | Providers | Hosted/local connectivity, credentials, account pools, health, and trust policy | Providers | Implemented under Settings and model views | -| Compute | Compute Nodes, GPUs, capacity observations, labels, freshness, and provenance | Compute | Target; no first-class inventory yet | +| Compute | Compute Nodes, GPUs, capacity observations, labels, freshness, and provenance | Compute | Partial: validated snapshot persistence and bounded collection ship; admin presentation remains a target | | Deployments | Model/runtime/compute binding, endpoint, desired state, observed state, and reconciliation | Deployments | Target; local endpoints are currently configured as Providers | | Routing | Logical models, aliases, eligibility, fallback, smart decisions, and evidence | Routing | Implemented, with some controls under Settings and Governance | | Governance | Users, teams, keys, policies, quotas, budgets, and approvals | Governance | Implemented | -| Observability | Requests, attempts, usage, cost, latency, GPU/runtime telemetry, and retained evidence | Observability | Partial: request and Provider evidence ship; compute telemetry is target | +| Observability | Requests, attempts, usage, cost, latency, GPU/runtime telemetry, and retained evidence | Observability | Partial: request, Provider, and bounded Runtime Adapter collection metrics ship; compute dashboards remain a target | | Operations | Readiness, incidents, backup, retention, upgrades, diagnostics, and reconciliation | Operations | Implemented for gateway operations; deployment operations are target | This mapping is an information architecture contract, not a requirement to @@ -290,7 +290,8 @@ boundary. Its adapter identities, discovery origins, credentials, and collection/freshness policy do not participate in inference Provider routing or inherit development-harness metadata. Enabled entries are validated and their environment-backed credentials resolved at configuration load; polling -and inventory presentation are separate lifecycle slices. +is a bounded server lifecycle, while inventory presentation remains a separate +admin API slice. Dashboard changes to control-plane records are persisted. They do not rewrite `.env` or `config.toml`. diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 66ae304..04c1859 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -27,6 +27,18 @@ potentially paid request. Health semantics are documented below. For a release or production trial, use [Production](PRODUCTION.md). +## Runtime Adapter Collection + +Configured Runtime Adapters are polled by a provider-neutral, read-only +background collector. Each enabled adapter is attempted immediately at +startup, then on its configured interval. A shared concurrency bound prevents +an adapter fleet from exhausting the process; a failing adapter is isolated +with bounded backoff and sanitized error logging. Accepted Compute inventories +are immutable ledger evidence, while `fresh`, `stale`, and `unavailable` remain +server-owned projections. Shutdown stops new collection attempts and drains +active attempts within a bounded timeout. The collector grants no Runtime +Adapter mutation authority and does not expose an admin API. + ## Health Semantics - `/livez` proves that the HTTP process can answer. It does not inspect storage diff --git a/docs/RUNTIME_ADAPTER.md b/docs/RUNTIME_ADAPTER.md index e87a355..c040271 100644 --- a/docs/RUNTIME_ADAPTER.md +++ b/docs/RUNTIME_ADAPTER.md @@ -3,8 +3,9 @@ ModelPort publishes versioned, read-only discovery and Compute Node/GPU observation contracts for external inference runtimes. The shipped v1alpha1 artifacts include wire contracts, offline validators, and a reusable -authenticated collection client. They do not expose a persisted inventory API, -reconciler, or mutation endpoint. +authenticated collection client. The external contract does not define an +admin API, reconciler, or mutation endpoint; ModelPort separately persists +validated observations and schedules bounded read-only collection. ## Contract Files @@ -50,6 +51,15 @@ rejected. The latest query derives `fresh`, `stale`, or `unavailable` from the validated observation time, server time, and a bounded server-owned policy. Derived state never changes the stored Runtime Adapter document. +When an enabled adapter is configured, ModelPort starts a bounded background +collector. It performs an immediate capabilities-first Compute read, persists +validated observations, then polls at the configured interval. Failed adapters +use bounded exponential backoff and do not stop healthy adapters; collection is +read-only and stops with the server during graceful shutdown. Collector metrics +contain only adapter IDs and bounded error classes. This collector does not add +an admin API, mutation operation, GPU control, or provider-specific runtime +management. + ## Capability Rules Every document has `apiVersion: runtime.modelport.io/v1alpha1`, @@ -108,8 +118,8 @@ reinterpreted across versions. Additive experimental data belongs in `local-inference-stack` checker remains an explicitly selected compatibility mode, not the source of this contract. -Configuration integration, collection scheduling, retention, admin APIs, and -all writes remain deferred to reviewed Issues. Offline +Retention, admin APIs, and all Runtime Adapter mutation operations remain +deferred to reviewed Issues. Offline validation cannot start a process, download a model, access a GPU, or call a network endpoint; the collection client performs only the two advertised safe reads requested by its caller. diff --git a/src/enterprise_ledger.rs b/src/enterprise_ledger.rs index 3c9e324..ff21eb2 100644 --- a/src/enterprise_ledger.rs +++ b/src/enterprise_ledger.rs @@ -81,7 +81,6 @@ struct MemoryLedger { usage_reservations: HashMap, budget_events: Vec, audit_events: Vec, - #[allow(dead_code)] // Used by the staged collection integration after this storage seam. runtime_compute_snapshots: HashMap<(String, String), compute_inventory::MemoryComputeSnapshot>, ops_incidents: BTreeMap, ops_event_index: HashMap, diff --git a/src/enterprise_ledger/compute_inventory.rs b/src/enterprise_ledger/compute_inventory.rs index 07f8e90..a62a707 100644 --- a/src/enterprise_ledger/compute_inventory.rs +++ b/src/enterprise_ledger/compute_inventory.rs @@ -1,5 +1,3 @@ -#![allow(dead_code)] // The scheduler/admin integration follows this reviewed storage slice. - use std::time::Duration; use chrono::{DateTime, SecondsFormat, Utc}; @@ -36,6 +34,7 @@ pub(crate) enum RuntimeComputeSnapshotWrite { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] +#[allow(dead_code)] // Read by the staged admin API after Issue #29. pub(crate) enum RuntimeComputeFreshness { Fresh, Stale, @@ -43,6 +42,7 @@ pub(crate) enum RuntimeComputeFreshness { } #[derive(Debug, Clone, PartialEq, Eq)] +#[allow(dead_code)] // Read by the staged admin API after Issue #29. pub(crate) struct RuntimeComputeInventoryState { pub(crate) freshness: RuntimeComputeFreshness, pub(crate) inventory: Option, @@ -146,6 +146,7 @@ impl EnterpriseLedger { } } + #[allow(dead_code)] // Read by the staged admin API after Issue #29. pub(crate) async fn latest_runtime_compute_inventory( &self, adapter_id: &str, diff --git a/src/metrics.rs b/src/metrics.rs index 3af92ff..210e954 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,7 +1,7 @@ use std::{ collections::BTreeMap, sync::Mutex, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; use crate::control::UsageEstimate; @@ -54,6 +54,7 @@ struct MetricsInner { messages: BTreeMap, rejections: BTreeMap, ledger_operations: BTreeMap, + runtime_adapter_collections: BTreeMap, routing_decisions: BTreeMap, routing_shadow_disagreements_total: u64, reconciled_requests_total: u64, @@ -97,6 +98,14 @@ struct LedgerOperationMetrics { degraded: bool, } +#[derive(Debug, Default)] +struct RuntimeAdapterCollectionMetrics { + successes_total: u64, + failures_total: BTreeMap, + last_attempt_timestamp_seconds: u64, + last_error: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] struct MessageKey { provider: String, @@ -209,6 +218,33 @@ impl Metrics { } } + pub(crate) fn record_runtime_adapter_collection( + &self, + adapter_id: &str, + result: Result<(), &'static str>, + ) { + let mut inner = self.inner.lock().expect("metrics lock poisoned"); + let metrics = inner + .runtime_adapter_collections + .entry(adapter_id.to_owned()) + .or_default(); + metrics.last_attempt_timestamp_seconds = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + match result { + Ok(()) => { + metrics.successes_total = metrics.successes_total.saturating_add(1); + metrics.last_error = None; + } + Err(error) => { + let failures = metrics.failures_total.entry(error.to_owned()).or_default(); + *failures = failures.saturating_add(1); + metrics.last_error = Some(error.to_owned()); + } + } + } + pub fn record_routing_decision( &self, mode: &str, @@ -411,6 +447,33 @@ impl Metrics { "modelport_ledger_reconciled_attempts_total {}\n", inner.reconciled_attempts_total )); + output.push('\n'); + + output.push_str("# HELP modelport_runtime_adapter_collection_successes_total Successful Runtime Adapter Compute collection attempts.\n"); + output.push_str("# TYPE modelport_runtime_adapter_collection_successes_total counter\n"); + output.push_str("# HELP modelport_runtime_adapter_collection_failures_total Failed Runtime Adapter Compute collection attempts by bounded error class.\n"); + output.push_str("# TYPE modelport_runtime_adapter_collection_failures_total counter\n"); + output.push_str("# HELP modelport_runtime_adapter_collection_last_attempt_timestamp_seconds Unix timestamp of the latest Runtime Adapter collection attempt.\n"); + output.push_str( + "# TYPE modelport_runtime_adapter_collection_last_attempt_timestamp_seconds gauge\n", + ); + for (adapter_id, metrics) in &inner.runtime_adapter_collections { + let adapter_id = escape_label_value(adapter_id); + output.push_str(&format!( + "modelport_runtime_adapter_collection_successes_total{{adapter_id=\"{adapter_id}\"}} {}\n", + metrics.successes_total + )); + for (error, count) in &metrics.failures_total { + output.push_str(&format!( + "modelport_runtime_adapter_collection_failures_total{{adapter_id=\"{adapter_id}\",error=\"{}\"}} {count}\n", + escape_label_value(error) + )); + } + output.push_str(&format!( + "modelport_runtime_adapter_collection_last_attempt_timestamp_seconds{{adapter_id=\"{adapter_id}\"}} {}\n", + metrics.last_attempt_timestamp_seconds + )); + } output } @@ -558,6 +621,8 @@ mod tests { metrics.record_rejection("messages", "validation", "invalid_request"); metrics.record_ledger_operation("request_finalization", false); metrics.record_reconciliation(2, 3); + metrics.record_runtime_adapter_collection("edge-1", Err("transport")); + metrics.record_runtime_adapter_collection("edge-1", Ok(())); metrics.record_routing_decision("shadow", "balanced", "mimo", true); metrics.record_message( MessageMetricLabels { @@ -627,6 +692,12 @@ mod tests { )); assert!(rendered.contains("modelport_ledger_reconciled_requests_total 2")); assert!(rendered.contains("modelport_ledger_reconciled_attempts_total 3")); + assert!(rendered.contains( + r#"modelport_runtime_adapter_collection_successes_total{adapter_id="edge-1"} 1"# + )); + assert!(rendered.contains( + r#"modelport_runtime_adapter_collection_failures_total{adapter_id="edge-1",error="transport"} 1"# + )); assert!(rendered.contains( r#"modelport_routing_decisions_total{mode="shadow",profile="balanced",provider="mimo"} 1"# )); diff --git a/src/runtime_adapter.rs b/src/runtime_adapter.rs index 1493f27..08acd69 100644 --- a/src/runtime_adapter.rs +++ b/src/runtime_adapter.rs @@ -6,6 +6,7 @@ use serde_json::Value; use crate::AppError; mod client; +pub(crate) mod collector; pub use client::{ RuntimeAdapterClient, RuntimeAdapterClientConfig, RuntimeAdapterComputeObservation, diff --git a/src/runtime_adapter/collector.rs b/src/runtime_adapter/collector.rs new file mode 100644 index 0000000..7ec6ad5 --- /dev/null +++ b/src/runtime_adapter/collector.rs @@ -0,0 +1,642 @@ +use std::{ + collections::BTreeMap, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; + +use tokio::{ + sync::{Notify, Semaphore}, + task::JoinHandle, +}; +use tracing::{info, warn}; + +use crate::{ + AppError, config::RuntimeAdapterConfig, enterprise_ledger::EnterpriseLedger, metrics::Metrics, + runtime_adapter::RuntimeAdapterClient, +}; + +// One fleet-wide cap keeps the maximum 64 configured adapters from exhausting the process. +const MAX_CONCURRENT_COLLECTIONS: usize = 4; +const MAX_BACKOFF: Duration = Duration::from_secs(30); + +pub(crate) struct RuntimeAdapterCollector { + draining: Arc, + wake: Arc, + tasks: Vec>, +} + +impl RuntimeAdapterCollector { + pub(crate) fn start( + adapters: BTreeMap, + ledger: Arc, + metrics: Arc, + draining: Arc, + ) -> Result { + let clients = adapters + .into_iter() + .map(|(adapter_id, config)| { + RuntimeAdapterClient::new(config.client_config) + .map(|client| (adapter_id, client, config.poll_interval)) + }) + .collect::, AppError>>()?; + let permits = Arc::new(Semaphore::new(MAX_CONCURRENT_COLLECTIONS)); + let wake = Arc::new(Notify::new()); + let mut tasks = Vec::with_capacity(clients.len()); + + for (adapter_id, client, poll_interval) in clients { + tasks.push(tokio::spawn(collection_loop( + adapter_id, + client, + poll_interval, + Arc::clone(&ledger), + Arc::clone(&metrics), + Arc::clone(&permits), + Arc::clone(&draining), + Arc::clone(&wake), + ))); + } + + Ok(Self { + draining, + wake, + tasks, + }) + } + + pub(crate) async fn shutdown(mut self, timeout: Duration) -> bool { + self.draining.store(true, Ordering::Release); + self.wake.notify_waiters(); + let deadline = tokio::time::Instant::now() + timeout; + while let Some(mut task) = self.tasks.pop() { + match tokio::time::timeout_at(deadline, &mut task).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + warn!(error = %error, "Runtime Adapter collector task failed to join"); + } + Err(_) => { + task.abort(); + for pending in &self.tasks { + pending.abort(); + } + return false; + } + } + } + true + } +} + +#[allow(clippy::too_many_arguments)] +async fn collection_loop( + adapter_id: String, + client: RuntimeAdapterClient, + poll_interval: Duration, + ledger: Arc, + metrics: Arc, + permits: Arc, + draining: Arc, + wake: Arc, +) { + let mut consecutive_failures = 0_u32; + loop { + if draining.load(Ordering::Acquire) { + break; + } + + let permit = tokio::select! { + permit = Arc::clone(&permits).acquire_owned() => match permit { + Ok(permit) => permit, + Err(_) => break, + }, + _ = wake.notified() => continue, + }; + if draining.load(Ordering::Acquire) { + drop(permit); + break; + } + + let result = collect_once(&client, &ledger).await; + drop(permit); + let delay = match result { + Ok(()) => { + consecutive_failures = 0; + metrics.record_runtime_adapter_collection(&adapter_id, Ok(())); + poll_interval + } + Err(error) => { + consecutive_failures = consecutive_failures.saturating_add(1); + let error_class = log_collection_failure(&adapter_id, &error); + metrics.record_runtime_adapter_collection(&adapter_id, Err(error_class)); + retry_delay(&adapter_id, consecutive_failures, poll_interval) + } + }; + + // A notification can arrive while collection is in flight, so re-check before sleeping. + if draining.load(Ordering::Acquire) { + break; + } + + tokio::select! { + _ = tokio::time::sleep(delay) => {} + _ = wake.notified() => {} + } + } + info!(adapter_id, "Runtime Adapter Compute collection stopped"); +} + +async fn collect_once( + client: &RuntimeAdapterClient, + ledger: &EnterpriseLedger, +) -> Result<(), AppError> { + let observation = client.collect_compute_inventory().await?; + ledger + .persist_runtime_compute_inventory(&observation.inventory) + .await?; + Ok(()) +} + +fn log_collection_failure(adapter_id: &str, error: &AppError) -> &'static str { + let error_class = error.telemetry_code(); + warn!( + adapter_id, + error_class, "Runtime Adapter Compute collection failed" + ); + error_class +} + +fn retry_delay(adapter_id: &str, consecutive_failures: u32, poll_interval: Duration) -> Duration { + let exponent = consecutive_failures.saturating_sub(1).min(5); + let base = Duration::from_secs(1_u64 << exponent); + let cap = poll_interval.min(MAX_BACKOFF); + let base = base.min(cap); + let jitter_window_ms = (base.as_millis() / 4).max(1); + // Stable per-adapter jitter keeps tests deterministic while de-synchronizing the fleet. + let hash = adapter_id.bytes().fold(0_u64, |value, byte| { + value.wrapping_mul(31).wrapping_add(u64::from(byte)) + }); + let jitter_ms = u64::try_from(u128::from(hash) % jitter_window_ms).unwrap_or(0); + base.saturating_add(Duration::from_millis(jitter_ms)) + .min(cap) +} + +#[cfg(test)] +mod tests { + use std::{ + io::{self, Write}, + sync::{ + Mutex, + atomic::{AtomicUsize, Ordering as AtomicOrdering}, + }, + }; + + use axum::{ + Router, + extract::State, + http::{StatusCode, header::CONTENT_TYPE}, + response::{IntoResponse, Response}, + routing::get, + }; + use tracing_subscriber::fmt::MakeWriter; + + use super::*; + use crate::{config::RuntimeAdapterConfig, runtime_adapter::RuntimeAdapterClientConfig}; + + const CAPABILITIES: &str = include_str!( + "../../tests/fixtures/runtime-adapters/qwen-llama-cpp-capabilities-v1alpha1.json" + ); + const INVENTORY: &str = include_str!( + "../../tests/fixtures/runtime-adapters/qwen-llama-cpp-compute-inventory-v1alpha1.json" + ); + + #[derive(Clone)] + struct FakeAdapter { + calls: Arc>>, + capabilities: String, + inventory: String, + fail: bool, + delay: Duration, + concurrency: Arc, + } + + #[derive(Default)] + struct ConcurrencyProbe { + active: AtomicUsize, + maximum: AtomicUsize, + } + + #[derive(Clone)] + struct CapturedWriter(Arc>>); + + struct CapturedGuard(Arc>>); + + impl Write for CapturedGuard { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for CapturedWriter { + type Writer = CapturedGuard; + + fn make_writer(&'a self) -> Self::Writer { + CapturedGuard(Arc::clone(&self.0)) + } + } + + #[tokio::test] + async fn starts_immediately_persists_and_stops() { + let calls = Arc::new(Mutex::new(Vec::new())); + let base_url = spawn_adapter( + "qwen-llama-cpp-reference", + Arc::clone(&calls), + Arc::default(), + Duration::ZERO, + false, + ) + .await; + let config = adapter_config( + "qwen-llama-cpp-reference", + base_url, + Duration::from_secs(60), + ); + let ledger = Arc::new(EnterpriseLedger::memory()); + let metrics = Arc::new(Metrics::new()); + let draining = Arc::new(AtomicBool::new(false)); + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([("qwen-llama-cpp-reference".to_owned(), config)]), + Arc::clone(&ledger), + Arc::clone(&metrics), + draining, + ) + .unwrap(); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if ledger + .latest_runtime_compute_inventory( + "qwen-llama-cpp-reference", + Duration::from_secs(90), + ) + .await + .unwrap() + .inventory + .is_some() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + + assert_eq!(*calls.lock().unwrap(), vec!["capabilities", "inventory"]); + assert!(collector.shutdown(Duration::from_secs(1)).await); + assert!(metrics.render_prometheus().contains( + r#"modelport_runtime_adapter_collection_successes_total{adapter_id="qwen-llama-cpp-reference"} 1"# + )); + } + + #[test] + fn retry_backoff_is_bounded_and_deterministic() { + let first = retry_delay("edge-1", 1, Duration::from_secs(10)); + assert_eq!(first, retry_delay("edge-1", 1, Duration::from_secs(10))); + assert!(first >= Duration::from_secs(1)); + assert!(retry_delay("edge-1", 20, Duration::from_secs(10)) <= Duration::from_secs(10)); + } + + #[tokio::test] + async fn shutdown_wakes_long_poll_intervals_without_waiting_for_the_timer() { + let collector = RuntimeAdapterCollector { + draining: Arc::new(AtomicBool::new(false)), + wake: Arc::new(Notify::new()), + tasks: Vec::new(), + }; + let started = std::time::Instant::now(); + assert!(collector.shutdown(Duration::from_millis(50)).await); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[tokio::test] + async fn shutdown_waits_for_an_in_flight_collection_and_starts_no_new_attempt() { + let calls = Arc::new(Mutex::new(Vec::new())); + let base_url = spawn_adapter( + "drain-adapter", + Arc::clone(&calls), + Arc::default(), + Duration::from_millis(100), + false, + ) + .await; + let ledger = Arc::new(EnterpriseLedger::memory()); + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([( + "drain-adapter".to_owned(), + adapter_config("drain-adapter", base_url, Duration::from_millis(1)), + )]), + Arc::clone(&ledger), + Arc::new(Metrics::new()), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + wait_until(Duration::from_secs(1), || !calls.lock().unwrap().is_empty()).await; + let started = std::time::Instant::now(); + assert!(collector.shutdown(Duration::from_secs(1)).await); + assert!(started.elapsed() >= Duration::from_millis(50)); + assert_eq!(*calls.lock().unwrap(), vec!["capabilities", "inventory"]); + assert!( + ledger + .latest_runtime_compute_inventory("drain-adapter", Duration::from_secs(90)) + .await + .unwrap() + .inventory + .is_some() + ); + } + + #[tokio::test] + async fn shutdown_aborts_an_in_flight_collection_after_the_deadline() { + let calls = Arc::new(Mutex::new(Vec::new())); + let base_url = spawn_adapter( + "timeout-adapter", + Arc::clone(&calls), + Arc::default(), + Duration::from_secs(1), + false, + ) + .await; + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([( + "timeout-adapter".to_owned(), + adapter_config("timeout-adapter", base_url, Duration::from_secs(60)), + )]), + Arc::new(EnterpriseLedger::memory()), + Arc::new(Metrics::new()), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + wait_until(Duration::from_secs(1), || !calls.lock().unwrap().is_empty()).await; + assert!(!collector.shutdown(Duration::from_millis(20)).await); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(*calls.lock().unwrap(), vec!["capabilities"]); + } + + #[test] + fn failure_logging_exposes_only_adapter_id_and_bounded_error_class() { + let captured = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_writer(CapturedWriter(Arc::clone(&captured))) + .finish(); + let error = AppError::Upstream { + status: 502, + body: "secret upstream body Bearer test-token https://runtime.invalid".to_owned(), + retry_after_secs: None, + }; + + tracing::subscriber::with_default(subscriber, || { + assert_eq!( + log_collection_failure("safe-adapter", &error), + "upstream_http" + ); + }); + let output = String::from_utf8(captured.lock().unwrap().clone()).unwrap(); + assert!(output.contains("safe-adapter")); + assert!(output.contains("upstream_http")); + for sensitive in [ + "secret upstream body", + "Bearer", + "test-token", + "runtime.invalid", + ] { + assert!(!output.contains(sensitive)); + } + } + + #[tokio::test] + async fn repeated_collection_is_idempotent_and_never_overlaps_one_adapter() { + let calls = Arc::new(Mutex::new(Vec::new())); + let concurrency = Arc::new(ConcurrencyProbe::default()); + let base_url = spawn_adapter( + "repeat-adapter", + Arc::clone(&calls), + Arc::clone(&concurrency), + Duration::from_millis(10), + false, + ) + .await; + let ledger = Arc::new(EnterpriseLedger::memory()); + let metrics = Arc::new(Metrics::new()); + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([( + "repeat-adapter".to_owned(), + adapter_config("repeat-adapter", base_url, Duration::from_millis(1)), + )]), + Arc::clone(&ledger), + Arc::clone(&metrics), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + wait_until(Duration::from_secs(2), || calls.lock().unwrap().len() >= 4).await; + assert_eq!(concurrency.maximum.load(AtomicOrdering::Acquire), 1); + assert!(collector.shutdown(Duration::from_secs(1)).await); + assert!(metrics.render_prometheus().contains( + r#"modelport_runtime_adapter_collection_successes_total{adapter_id="repeat-adapter"} 2"# + )); + } + + #[tokio::test] + async fn collection_is_globally_bounded_to_four_adapters() { + let concurrency = Arc::new(ConcurrencyProbe::default()); + let mut adapters = BTreeMap::new(); + for index in 0..5 { + let adapter_id = format!("edge-{index}"); + let base_url = spawn_adapter( + &adapter_id, + Arc::default(), + Arc::clone(&concurrency), + Duration::from_millis(50), + false, + ) + .await; + adapters.insert( + adapter_id.clone(), + adapter_config(&adapter_id, base_url, Duration::from_secs(60)), + ); + } + let collector = RuntimeAdapterCollector::start( + adapters, + Arc::new(EnterpriseLedger::memory()), + Arc::new(Metrics::new()), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + wait_until(Duration::from_secs(2), || { + concurrency.maximum.load(AtomicOrdering::Acquire) >= MAX_CONCURRENT_COLLECTIONS + }) + .await; + assert_eq!( + concurrency.maximum.load(AtomicOrdering::Acquire), + MAX_CONCURRENT_COLLECTIONS + ); + assert!(collector.shutdown(Duration::from_secs(3)).await); + } + + #[tokio::test] + async fn a_failing_adapter_does_not_block_a_healthy_adapter() { + let healthy_url = spawn_adapter( + "healthy-adapter", + Arc::default(), + Arc::default(), + Duration::ZERO, + false, + ) + .await; + let failing_url = spawn_adapter( + "failing-adapter", + Arc::default(), + Arc::default(), + Duration::ZERO, + true, + ) + .await; + let ledger = Arc::new(EnterpriseLedger::memory()); + let metrics = Arc::new(Metrics::new()); + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([ + ( + "failing-adapter".to_owned(), + adapter_config("failing-adapter", failing_url, Duration::from_secs(60)), + ), + ( + "healthy-adapter".to_owned(), + adapter_config("healthy-adapter", healthy_url, Duration::from_secs(60)), + ), + ]), + Arc::clone(&ledger), + Arc::clone(&metrics), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let healthy = ledger + .latest_runtime_compute_inventory( + "healthy-adapter", + Duration::from_secs(90), + ) + .await + .unwrap() + .inventory + .is_some(); + let failure = metrics.render_prometheus().contains( + r#"modelport_runtime_adapter_collection_failures_total{adapter_id="failing-adapter",error="upstream_http"} 1"#, + ); + if healthy && failure { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(collector.shutdown(Duration::from_secs(1)).await); + } + + fn adapter_config( + adapter_id: &str, + base_url: String, + poll_interval: Duration, + ) -> RuntimeAdapterConfig { + RuntimeAdapterConfig { + client_config: RuntimeAdapterClientConfig::new(adapter_id, base_url, "test-token") + .unwrap(), + credential_env: "TEST_TOKEN".to_owned(), + poll_interval, + stale_after: Duration::from_secs(90), + } + } + + async fn spawn_adapter( + adapter_id: &str, + calls: Arc>>, + concurrency: Arc, + delay: Duration, + fail: bool, + ) -> String { + let capabilities = CAPABILITIES.replace("qwen-llama-cpp-reference", adapter_id); + let inventory = INVENTORY.replace("qwen-llama-cpp-reference", adapter_id); + let app = Router::new() + .route( + "/runtime-adapter/v1alpha1/capabilities", + get(|State(state): State| async move { + state.calls.lock().unwrap().push("capabilities"); + adapter_response(&state, state.capabilities.clone()).await + }), + ) + .route( + "/runtime-adapter/v1alpha1/inventory/compute", + get(|State(state): State| async move { + state.calls.lock().unwrap().push("inventory"); + adapter_response(&state, state.inventory.clone()).await + }), + ) + .with_state(FakeAdapter { + calls, + capabilities, + inventory, + fail, + delay, + concurrency, + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{address}") + } + + async fn adapter_response(state: &FakeAdapter, body: String) -> Response { + let active = state + .concurrency + .active + .fetch_add(1, AtomicOrdering::AcqRel) + .saturating_add(1); + state + .concurrency + .maximum + .fetch_max(active, AtomicOrdering::AcqRel); + tokio::time::sleep(state.delay).await; + state + .concurrency + .active + .fetch_sub(1, AtomicOrdering::AcqRel); + if state.fail { + return (StatusCode::BAD_GATEWAY, "secret upstream body").into_response(); + } + ([(CONTENT_TYPE, "application/json")], body).into_response() + } + + async fn wait_until(timeout: Duration, predicate: impl Fn() -> bool) { + tokio::time::timeout(timeout, async { + while !predicate() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } +} diff --git a/src/server.rs b/src/server.rs index 2cec665..b46c5c7 100644 --- a/src/server.rs +++ b/src/server.rs @@ -25,6 +25,7 @@ use crate::{ self, AppState, GatewaySecurityPolicy, RateLimiter, RetentionPreviewStore, TrustedProxyConfig, }, + runtime_adapter::collector::RuntimeAdapterCollector, smart_router::SmartRouter, version, }; @@ -91,6 +92,12 @@ pub(crate) async fn serve() -> Result<(), AppError> { state.oidc.validate_console_access(&state.auth)?; let listener = TcpListener::bind(bind_addr).await?; + let collector = RuntimeAdapterCollector::start( + config.runtime_adapters.clone(), + state.ledger.clone(), + state.metrics.clone(), + draining.clone(), + )?; info!( %bind_addr, version = version::VERSION, @@ -99,12 +106,21 @@ pub(crate) async fn serve() -> Result<(), AppError> { "ModelPort listening" ); - axum::serve( + let serve_result = axum::serve( listener, routes::router(state).into_make_service_with_connect_info::(), ) .with_graceful_shutdown(shutdown_signal(draining)) - .await?; + .await; + + let collector_timeout = runtime_adapter_collector_drain_timeout(); + if !collector.shutdown(collector_timeout).await { + warn!( + timeout_seconds = collector_timeout.as_secs(), + "timed out draining Runtime Adapter collection tasks during shutdown" + ); + } + serve_result?; let drain_timeout = finalization_drain_timeout(); if !finalizers.drain(drain_timeout).await { @@ -136,6 +152,10 @@ fn finalization_drain_timeout() -> Duration { ) } +fn runtime_adapter_collector_drain_timeout() -> Duration { + Duration::from_secs(10) +} + async fn shutdown_signal(draining: Arc) { let ctrl_c = async { if let Err(error) = tokio::signal::ctrl_c().await { From ab4fc62c3f349f5857b5cf082af547618eb27a0e Mon Sep 17 00:00:00 2001 From: lovett55 Date: Wed, 9 Sep 2026 17:12:38 +0800 Subject: [PATCH 3/3] fix(runtime): harden adapter collector lifecycle and telemetry Signed-off-by: lovett55 --- docs/CONFIGURATION.md | 9 +- docs/OPERATIONS.md | 15 +- src/enterprise_ledger.rs | 2 +- src/metrics.rs | 44 ++++-- src/routes.rs | 8 +- src/runtime_adapter/collector.rs | 236 +++++++++++++++++++++++++------ src/server.rs | 2 +- 7 files changed, 253 insertions(+), 63 deletions(-) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 7ba129c..edbd0ab 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -64,9 +64,11 @@ is never part of the TOML document or a serializable configuration type. must cover at least one polling interval. Disabled declarations are inert: their endpoint and credential are neither required nor resolved. Duplicate TOML adapter tables, invalid enabled declarations, missing credentials, and a -registry over 64 entries fail configuration loading closed. This registry does -not start polling; background collection and admin inventory APIs remain -separate reviewed work. +registry over 64 entries fail configuration loading closed. At server startup, +each enabled entry receives a supervised, read-only collection loop with an +immediate capabilities-first attempt and bounded periodic retry. Registry, +credential, polling, and stale-after changes require a process restart; an +admin inventory API remains separate reviewed work. ## Required Minimum: DeepSeek-Only Example @@ -992,6 +994,7 @@ lossless, but its configured output limits are still enforced. | Trusted proxies, health exposure, private/insecure-URL policy, CSRF/origin policy | No | Yes | | Admin bootstrap, session TTL, secure-cookie flag | No | Yes | | Storage backend or state paths | No | Yes | +| Runtime Adapter registry, credentials, polling interval, or stale-after policy | No | Yes | Reload from the dashboard Operations tab or restart the service. A successful reload validates the new base snapshot but does not mutate `.env` or TOML. diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 04c1859..16dc2aa 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -36,8 +36,9 @@ an adapter fleet from exhausting the process; a failing adapter is isolated with bounded backoff and sanitized error logging. Accepted Compute inventories are immutable ledger evidence, while `fresh`, `stale`, and `unavailable` remain server-owned projections. Shutdown stops new collection attempts and drains -active attempts within a bounded timeout. The collector grants no Runtime -Adapter mutation authority and does not expose an admin API. +active attempts for up to 10 seconds before emitting a sanitized warning. The +collector grants no Runtime Adapter mutation authority and does not expose an +admin API. ## Health Semantics @@ -465,10 +466,12 @@ is process-local and requires restart/recreate after a configuration change. ## Configuration Reload The dashboard Operations tab can reload the base configuration. Provider, -model, alias, and route values can update for new requests. Process layers, -security policies, transport settings, storage, sessions, and newly introduced -process environment variables require a restart. Use the full matrix in -[Configuration](CONFIGURATION.md#reload-versus-restart). +model, alias, and route values can update for new requests. Runtime Adapter +registry and collection-policy changes require a restart so the supervised +loops and their credentials stay aligned with the active process. Process +layers, security policies, transport settings, storage, sessions, and newly +introduced process environment variables also require a restart. Use the full +matrix in [Configuration](CONFIGURATION.md#reload-versus-restart). Dashboard Settings exposes effective server/auth/rate values as read-only runtime facts. Default provider and provider order remain runtime control-plane diff --git a/src/enterprise_ledger.rs b/src/enterprise_ledger.rs index ff21eb2..a6da946 100644 --- a/src/enterprise_ledger.rs +++ b/src/enterprise_ledger.rs @@ -881,7 +881,7 @@ impl EnterpriseLedger { } #[cfg(test)] - async fn postgres_for_tests(database_url: &str) -> Result { + pub(crate) async fn postgres_for_tests(database_url: &str) -> Result { let pool = connect_pool(database_url, Some(4)).await?; sqlx::migrate!("./migrations") .run(&pool) diff --git a/src/metrics.rs b/src/metrics.rs index 210e954..c5f5901 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -103,7 +103,6 @@ struct RuntimeAdapterCollectionMetrics { successes_total: u64, failures_total: BTreeMap, last_attempt_timestamp_seconds: u64, - last_error: Option, } #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] @@ -218,7 +217,19 @@ impl Metrics { } } - pub(crate) fn record_runtime_adapter_collection( + pub(crate) fn record_runtime_adapter_collection_attempt(&self, adapter_id: &str) { + let mut inner = self.inner.lock().expect("metrics lock poisoned"); + inner + .runtime_adapter_collections + .entry(adapter_id.to_owned()) + .or_default() + .last_attempt_timestamp_seconds = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + } + + pub(crate) fn record_runtime_adapter_collection_result( &self, adapter_id: &str, result: Result<(), &'static str>, @@ -228,19 +239,13 @@ impl Metrics { .runtime_adapter_collections .entry(adapter_id.to_owned()) .or_default(); - metrics.last_attempt_timestamp_seconds = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); match result { Ok(()) => { metrics.successes_total = metrics.successes_total.saturating_add(1); - metrics.last_error = None; } Err(error) => { let failures = metrics.failures_total.entry(error.to_owned()).or_default(); *failures = failures.saturating_add(1); - metrics.last_error = Some(error.to_owned()); } } } @@ -621,8 +626,10 @@ mod tests { metrics.record_rejection("messages", "validation", "invalid_request"); metrics.record_ledger_operation("request_finalization", false); metrics.record_reconciliation(2, 3); - metrics.record_runtime_adapter_collection("edge-1", Err("transport")); - metrics.record_runtime_adapter_collection("edge-1", Ok(())); + metrics.record_runtime_adapter_collection_attempt("edge-1"); + metrics.record_runtime_adapter_collection_result("edge-1", Err("transport")); + metrics.record_runtime_adapter_collection_attempt("edge-1"); + metrics.record_runtime_adapter_collection_result("edge-1", Ok(())); metrics.record_routing_decision("shadow", "balanced", "mimo", true); metrics.record_message( MessageMetricLabels { @@ -698,6 +705,23 @@ mod tests { assert!(rendered.contains( r#"modelport_runtime_adapter_collection_failures_total{adapter_id="edge-1",error="transport"} 1"# )); + let last_attempt = rendered + .lines() + .find(|line| { + line.starts_with( + "modelport_runtime_adapter_collection_last_attempt_timestamp_seconds{adapter_id=\"edge-1\"}", + ) + }) + .expect("runtime adapter last-attempt metric"); + assert!( + last_attempt + .rsplit_once(' ') + .unwrap() + .1 + .parse::() + .unwrap() + > 0 + ); assert!(rendered.contains( r#"modelport_routing_decisions_total{mode="shadow",profile="balanced",provider="mimo"} 1"# )); diff --git a/src/routes.rs b/src/routes.rs index 242143c..5002c32 100644 --- a/src/routes.rs +++ b/src/routes.rs @@ -1539,7 +1539,7 @@ async fn admin_reload_config( "issues": issues, "reloadScope": { "applied": ["provider catalog", "base provider keys", "base urls", "model lists", "aliases", "legacy client auth token"], - "requiresRestart": ["bind address", "request body limit", "concurrency layer", "rate limits", "HTTP client timeouts", "trusted proxies", "security flags", "admin session and cookie settings", "storage", "new credential-profile environment variables"], + "requiresRestart": ["bind address", "request body limit", "concurrency layer", "rate limits", "HTTP client timeouts", "trusted proxies", "security flags", "admin session and cookie settings", "storage", "Runtime Adapter registry and collection policy", "new credential-profile environment variables"], }, }))) } @@ -6425,6 +6425,12 @@ data: [DONE] assert_eq!(body["providerCount"], json!(2)); assert_eq!(body["settings"]["gateway"]["providerOrder"][1], "custom"); assert_eq!(body["reloadScope"]["requiresRestart"][0], "bind address"); + assert!( + body["reloadScope"]["requiresRestart"] + .as_array() + .unwrap() + .contains(&json!("Runtime Adapter registry and collection policy")) + ); } #[test] diff --git a/src/runtime_adapter/collector.rs b/src/runtime_adapter/collector.rs index 7ec6ad5..1f8a741 100644 --- a/src/runtime_adapter/collector.rs +++ b/src/runtime_adapter/collector.rs @@ -25,7 +25,12 @@ const MAX_BACKOFF: Duration = Duration::from_secs(30); pub(crate) struct RuntimeAdapterCollector { draining: Arc, wake: Arc, - tasks: Vec>, + tasks: Vec, +} + +struct CollectionTask { + adapter_id: String, + handle: JoinHandle<()>, } impl RuntimeAdapterCollector { @@ -47,16 +52,19 @@ impl RuntimeAdapterCollector { let mut tasks = Vec::with_capacity(clients.len()); for (adapter_id, client, poll_interval) in clients { - tasks.push(tokio::spawn(collection_loop( - adapter_id, - client, - poll_interval, - Arc::clone(&ledger), - Arc::clone(&metrics), - Arc::clone(&permits), - Arc::clone(&draining), - Arc::clone(&wake), - ))); + tasks.push(CollectionTask { + adapter_id: adapter_id.clone(), + handle: tokio::spawn(collection_loop( + adapter_id, + client, + poll_interval, + Arc::clone(&ledger), + Arc::clone(&metrics), + Arc::clone(&permits), + Arc::clone(&draining), + Arc::clone(&wake), + )), + }); } Ok(Self { @@ -70,16 +78,24 @@ impl RuntimeAdapterCollector { self.draining.store(true, Ordering::Release); self.wake.notify_waiters(); let deadline = tokio::time::Instant::now() + timeout; - while let Some(mut task) = self.tasks.pop() { - match tokio::time::timeout_at(deadline, &mut task).await { + while let Some(CollectionTask { + adapter_id, + mut handle, + }) = self.tasks.pop() + { + match tokio::time::timeout_at(deadline, &mut handle).await { Ok(Ok(())) => {} - Ok(Err(error)) => { - warn!(error = %error, "Runtime Adapter collector task failed to join"); + Ok(Err(_)) => { + warn!( + adapter_id, + error_class = "task_join", + "Runtime Adapter collector task failed to join" + ); } Err(_) => { - task.abort(); + handle.abort(); for pending in &self.tasks { - pending.abort(); + pending.handle.abort(); } return false; } @@ -118,18 +134,19 @@ async fn collection_loop( break; } + metrics.record_runtime_adapter_collection_attempt(&adapter_id); let result = collect_once(&client, &ledger).await; drop(permit); let delay = match result { - Ok(()) => { + Ok(_) => { consecutive_failures = 0; - metrics.record_runtime_adapter_collection(&adapter_id, Ok(())); + metrics.record_runtime_adapter_collection_result(&adapter_id, Ok(())); poll_interval } Err(error) => { consecutive_failures = consecutive_failures.saturating_add(1); let error_class = log_collection_failure(&adapter_id, &error); - metrics.record_runtime_adapter_collection(&adapter_id, Err(error_class)); + metrics.record_runtime_adapter_collection_result(&adapter_id, Err(error_class)); retry_delay(&adapter_id, consecutive_failures, poll_interval) } }; @@ -150,12 +167,11 @@ async fn collection_loop( async fn collect_once( client: &RuntimeAdapterClient, ledger: &EnterpriseLedger, -) -> Result<(), AppError> { +) -> Result { let observation = client.collect_compute_inventory().await?; ledger .persist_runtime_compute_inventory(&observation.inventory) - .await?; - Ok(()) + .await } fn log_collection_failure(adapter_id: &str, error: &AppError) -> &'static str { @@ -169,17 +185,24 @@ fn log_collection_failure(adapter_id: &str, error: &AppError) -> &'static str { fn retry_delay(adapter_id: &str, consecutive_failures: u32, poll_interval: Duration) -> Duration { let exponent = consecutive_failures.saturating_sub(1).min(5); - let base = Duration::from_secs(1_u64 << exponent); let cap = poll_interval.min(MAX_BACKOFF); - let base = base.min(cap); - let jitter_window_ms = (base.as_millis() / 4).max(1); - // Stable per-adapter jitter keeps tests deterministic while de-synchronizing the fleet. - let hash = adapter_id.bytes().fold(0_u64, |value, byte| { - value.wrapping_mul(31).wrapping_add(u64::from(byte)) - }); - let jitter_ms = u64::try_from(u128::from(hash) % jitter_window_ms).unwrap_or(0); - base.saturating_add(Duration::from_millis(jitter_ms)) - .min(cap) + let nominal = Duration::from_secs(1_u64 << exponent).min(cap); + // An 80-100% factor preserves jitter at the cap instead of truncating it away. + let hash = adapter_id + .bytes() + .fold(u64::from(consecutive_failures), |value, byte| { + value.wrapping_mul(31).wrapping_add(u64::from(byte)) + }); + let factor_basis_points = 8_000_u64.saturating_add(hash % 2_000); + let nominal_ms = u64::try_from(nominal.as_millis()).unwrap_or(u64::MAX); + Duration::from_millis( + nominal_ms + .saturating_mul(factor_basis_points) + .checked_div(10_000) + .unwrap_or(nominal_ms) + .max(1), + ) + .min(cap) } #[cfg(test)] @@ -307,10 +330,22 @@ mod tests { #[test] fn retry_backoff_is_bounded_and_deterministic() { - let first = retry_delay("edge-1", 1, Duration::from_secs(10)); - assert_eq!(first, retry_delay("edge-1", 1, Duration::from_secs(10))); - assert!(first >= Duration::from_secs(1)); - assert!(retry_delay("edge-1", 20, Duration::from_secs(10)) <= Duration::from_secs(10)); + let delays = (1..=5) + .map(|failure| retry_delay("edge-1", failure, Duration::from_secs(20))) + .collect::>(); + assert_eq!( + delays, + (1..=5) + .map(|failure| retry_delay("edge-1", failure, Duration::from_secs(20))) + .collect::>() + ); + assert!(delays[0] >= Duration::from_millis(800)); + assert!(delays[0] < Duration::from_secs(1)); + assert!(delays.windows(2).all(|pair| pair[1] >= pair[0])); + let capped = retry_delay("edge-1", 20, Duration::from_secs(10)); + assert!((Duration::from_secs(8)..Duration::from_secs(10)).contains(&capped)); + assert_ne!(capped, retry_delay("edge-1", 21, Duration::from_secs(10))); + assert_ne!(capped, retry_delay("edge-2", 20, Duration::from_secs(10))); } #[tokio::test] @@ -337,18 +372,24 @@ mod tests { ) .await; let ledger = Arc::new(EnterpriseLedger::memory()); + let metrics = Arc::new(Metrics::new()); let collector = RuntimeAdapterCollector::start( BTreeMap::from([( "drain-adapter".to_owned(), adapter_config("drain-adapter", base_url, Duration::from_millis(1)), )]), Arc::clone(&ledger), - Arc::new(Metrics::new()), + Arc::clone(&metrics), Arc::new(AtomicBool::new(false)), ) .unwrap(); wait_until(Duration::from_secs(1), || !calls.lock().unwrap().is_empty()).await; + assert!(metrics.render_prometheus().lines().any(|line| { + line.starts_with( + "modelport_runtime_adapter_collection_last_attempt_timestamp_seconds{adapter_id=\"drain-adapter\"}", + ) + })); let started = std::time::Instant::now(); assert!(collector.shutdown(Duration::from_secs(1)).await); assert!(started.elapsed() >= Duration::from_millis(50)); @@ -391,6 +432,79 @@ mod tests { assert_eq!(*calls.lock().unwrap(), vec!["capabilities"]); } + #[tokio::test] + async fn collect_once_preserves_snapshot_idempotency() { + let calls = Arc::new(Mutex::new(Vec::new())); + let base_url = spawn_adapter( + "idempotent-adapter", + Arc::clone(&calls), + Arc::default(), + Duration::ZERO, + false, + ) + .await; + let client_config = + adapter_config("idempotent-adapter", base_url, Duration::from_secs(60)).client_config; + let client = RuntimeAdapterClient::new(client_config).unwrap(); + let ledger = EnterpriseLedger::memory(); + + assert_eq!( + collect_once(&client, &ledger).await.unwrap(), + crate::enterprise_ledger::compute_inventory::RuntimeComputeSnapshotWrite::Inserted + ); + assert_eq!( + collect_once(&client, &ledger).await.unwrap(), + crate::enterprise_ledger::compute_inventory::RuntimeComputeSnapshotWrite::Idempotent + ); + } + + #[tokio::test] + async fn postgres_state_collector_persists_validated_inventory() { + let Ok(database_url) = std::env::var("MODELPORT_TEST_DATABASE_URL") else { + return; + }; + let adapter_id = format!("postgres-{}", uuid::Uuid::new_v4().simple()); + let base_url = spawn_adapter( + &adapter_id, + Arc::default(), + Arc::default(), + Duration::ZERO, + false, + ) + .await; + let ledger = Arc::new( + EnterpriseLedger::postgres_for_tests(&database_url) + .await + .unwrap(), + ); + let collector = RuntimeAdapterCollector::start( + BTreeMap::from([( + adapter_id.clone(), + adapter_config(&adapter_id, base_url, Duration::from_secs(60)), + )]), + Arc::clone(&ledger), + Arc::new(Metrics::new()), + Arc::new(AtomicBool::new(false)), + ) + .unwrap(); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let state = ledger + .latest_runtime_compute_inventory(&adapter_id, Duration::from_secs(90)) + .await + .unwrap(); + if state.inventory.is_some() { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(collector.shutdown(Duration::from_secs(1)).await); + } + #[test] fn failure_logging_exposes_only_adapter_id_and_bounded_error_class() { let captured = Arc::new(Mutex::new(Vec::new())); @@ -497,9 +611,11 @@ mod tests { #[tokio::test] async fn a_failing_adapter_does_not_block_a_healthy_adapter() { + let healthy_calls = Arc::new(Mutex::new(Vec::new())); + let failing_calls = Arc::new(Mutex::new(Vec::new())); let healthy_url = spawn_adapter( "healthy-adapter", - Arc::default(), + Arc::clone(&healthy_calls), Arc::default(), Duration::ZERO, false, @@ -507,7 +623,7 @@ mod tests { .await; let failing_url = spawn_adapter( "failing-adapter", - Arc::default(), + Arc::clone(&failing_calls), Arc::default(), Duration::ZERO, true, @@ -523,7 +639,7 @@ mod tests { ), ( "healthy-adapter".to_owned(), - adapter_config("healthy-adapter", healthy_url, Duration::from_secs(60)), + adapter_config("healthy-adapter", healthy_url, Duration::from_millis(20)), ), ]), Arc::clone(&ledger), @@ -554,9 +670,47 @@ mod tests { }) .await .unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(healthy_calls.lock().unwrap().len() >= 4); + assert_eq!(failing_calls.lock().unwrap().len(), 1); assert!(collector.shutdown(Duration::from_secs(1)).await); } + #[test] + fn shutdown_join_failure_logs_only_a_bounded_class() { + let captured = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_writer(CapturedWriter(Arc::clone(&captured))) + .finish(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let task = tokio::spawn(async { + panic!("secret join payload"); + }); + tokio::task::yield_now().await; + let collector = RuntimeAdapterCollector { + draining: Arc::new(AtomicBool::new(false)), + wake: Arc::new(Notify::new()), + tasks: vec![CollectionTask { + adapter_id: "join-adapter".to_owned(), + handle: task, + }], + }; + assert!(collector.shutdown(Duration::from_secs(1)).await); + }); + }); + let output = String::from_utf8(captured.lock().unwrap().clone()).unwrap(); + assert!(output.contains("join-adapter")); + assert!(output.contains("task_join")); + assert!(!output.contains("secret join payload")); + } + fn adapter_config( adapter_id: &str, base_url: String, diff --git a/src/server.rs b/src/server.rs index b46c5c7..50631a3 100644 --- a/src/server.rs +++ b/src/server.rs @@ -93,7 +93,7 @@ pub(crate) async fn serve() -> Result<(), AppError> { state.oidc.validate_console_access(&state.auth)?; let listener = TcpListener::bind(bind_addr).await?; let collector = RuntimeAdapterCollector::start( - config.runtime_adapters.clone(), + state.config.snapshot().runtime_adapters, state.ledger.clone(), state.metrics.clone(), draining.clone(),