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/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/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/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 66ae304..16dc2aa 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -27,6 +27,19 @@ 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 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 - `/livez` proves that the HTTP process can answer. It does not inspect storage @@ -453,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/docs/RUNTIME_ADAPTER.md b/docs/RUNTIME_ADAPTER.md index 467b2a8..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 @@ -42,6 +43,23 @@ 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. + +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`, @@ -100,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 policy, persistence, derived freshness, -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/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..a6da946 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,7 @@ struct MemoryLedger { usage_reservations: HashMap, budget_events: Vec, audit_events: Vec, + runtime_compute_snapshots: HashMap<(String, String), compute_inventory::MemoryComputeSnapshot>, ops_incidents: BTreeMap, ops_event_index: HashMap, ops_heartbeats: BTreeMap, @@ -878,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/enterprise_ledger/compute_inventory.rs b/src/enterprise_ledger/compute_inventory.rs new file mode 100644 index 0000000..a62a707 --- /dev/null +++ b/src/enterprise_ledger/compute_inventory.rs @@ -0,0 +1,495 @@ +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")] +#[allow(dead_code)] // Read by the staged admin API after Issue #29. +pub(crate) enum RuntimeComputeFreshness { + Fresh, + Stale, + Unavailable, +} + +#[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, + 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()) + } + } + } + } + + #[allow(dead_code)] // Read by the staged admin API after Issue #29. + 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/metrics.rs b/src/metrics.rs index 3af92ff..c5f5901 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,13 @@ struct LedgerOperationMetrics { degraded: bool, } +#[derive(Debug, Default)] +struct RuntimeAdapterCollectionMetrics { + successes_total: u64, + failures_total: BTreeMap, + last_attempt_timestamp_seconds: u64, +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] struct MessageKey { provider: String, @@ -209,6 +217,39 @@ impl Metrics { } } + 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>, + ) { + let mut inner = self.inner.lock().expect("metrics lock poisoned"); + let metrics = inner + .runtime_adapter_collections + .entry(adapter_id.to_owned()) + .or_default(); + match result { + Ok(()) => { + metrics.successes_total = metrics.successes_total.saturating_add(1); + } + Err(error) => { + let failures = metrics.failures_total.entry(error.to_owned()).or_default(); + *failures = failures.saturating_add(1); + } + } + } + pub fn record_routing_decision( &self, mode: &str, @@ -411,6 +452,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 +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_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 { @@ -627,6 +699,29 @@ 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"# + )); + 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.rs b/src/runtime_adapter.rs index 025c531..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, @@ -19,6 +20,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(), )); diff --git a/src/runtime_adapter/collector.rs b/src/runtime_adapter/collector.rs new file mode 100644 index 0000000..1f8a741 --- /dev/null +++ b/src/runtime_adapter/collector.rs @@ -0,0 +1,796 @@ +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, +} + +struct CollectionTask { + adapter_id: String, + handle: JoinHandle<()>, +} + +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(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 { + 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(CollectionTask { + adapter_id, + mut handle, + }) = self.tasks.pop() + { + match tokio::time::timeout_at(deadline, &mut handle).await { + Ok(Ok(())) => {} + Ok(Err(_)) => { + warn!( + adapter_id, + error_class = "task_join", + "Runtime Adapter collector task failed to join" + ); + } + Err(_) => { + handle.abort(); + for pending in &self.tasks { + pending.handle.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; + } + + metrics.record_runtime_adapter_collection_attempt(&adapter_id); + let result = collect_once(&client, &ledger).await; + drop(permit); + let delay = match result { + Ok(_) => { + consecutive_failures = 0; + 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_result(&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 { + let observation = client.collect_compute_inventory().await?; + ledger + .persist_runtime_compute_inventory(&observation.inventory) + .await +} + +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 cap = poll_interval.min(MAX_BACKOFF); + 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)] +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 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] + 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 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::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)); + 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"]); + } + + #[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())); + 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_calls = Arc::new(Mutex::new(Vec::new())); + let failing_calls = Arc::new(Mutex::new(Vec::new())); + let healthy_url = spawn_adapter( + "healthy-adapter", + Arc::clone(&healthy_calls), + Arc::default(), + Duration::ZERO, + false, + ) + .await; + let failing_url = spawn_adapter( + "failing-adapter", + Arc::clone(&failing_calls), + 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_millis(20)), + ), + ]), + 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(); + 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, + 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..50631a3 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( + state.config.snapshot().runtime_adapters, + 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 {