GameCult/Ghostlight
Watch
9
0
Fork
You've already forked Ghostlight
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Expose managed Ghostlight route presence

Browse source
This commit is contained in:
Meta Vens 2026-09-03 09:25:32 +02:00
commit 92566480bb
2 changed files with 479 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

187
crates/ghostlight-dungeon/src/idunn_health.rs
View file

@ -41,7 +41,7 @@ pub const IDUNN_RUNTIME_BUNDLE_ENVIRONMENT: &str = "GAMECULT_IDUNN_RUNTIME_BUNDL
pub const IDUNN_RUNTIME_CANDIDATE_BIND_ENVIRONMENT: &str = "GAMECULT_IDUNN_CANDIDATE_BIND";
pub const IDUNN_PROCESS_WRITE_LEASE_ENVIRONMENT: &str = "GAMECULT_IDUNN_PROCESS_WRITE_LEASE";
const TARGET: &str = "ghostlight";
pub(crate) const TARGET: &str = "ghostlight";
const STATE_SCHEMA_GENERATION: &str = "world-v2";
const STATE_CONTRACT_SHA256: &str =
"sha256-bf6ec06d885a59ddb237c6224d0abb4ccceac8c7ba23761d1326d7f562a4c21e";
@ -305,6 +305,32 @@ impl RuntimePresencePublisher {
self.publish("active", "world-owner-serving", Some(lease_sha256))
}
pub(crate) fn route_observation(
&mut self,
message_id: &str,
ready: bool,
lease: &ProcessWriteLeaseGuard,
) -> Result<CultNetMessage> {
ensure!(
!message_id.is_empty(),
"route-observation message id is empty"
);
lease.require_current()?;
let lease_sha256 = lease.canonical_sha256();
ensure!(
self.active_write_lease_sha256.as_deref() == Some(lease_sha256),
"route observation does not carry the active process write lease"
);
let state = if ready { "active" } else { "degraded" };
let detail = format!("route-observation:{message_id}");
let record = self.signed_record(state, &detail, Some(lease_sha256.into()))?;
let payload = canonical_presence_payload(&record)?;
Ok(CultNetMessage::SnapshotResponseRaw {
message_id: message_id.into(),
documents: vec![runtime_presence_document(&record, payload, None)?],
})
}
fn publish(
&mut self,
state: &str,
@ -953,36 +979,21 @@ fn publish_runtime_presence(
record: &GameCultRuntimePresenceHealthRecord,
) -> Result<String> {
record.validate()?;
let payload = rmp_serde::to_vec(record).context("encoding canonical runtime presence")?;
let decoded: GameCultRuntimePresenceHealthRecord = rmp_serde::from_slice(&payload)?;
ensure!(
decoded == *record && rmp_serde::to_vec(&decoded)? == payload,
"runtime presence encoding is noncanonical"
);
let payload = canonical_presence_payload(record)?;
let canonical_sha256 = record.canonical_sha256()?;
let message = CultNetMessage::DocumentPutRaw {
message_id: format!(
"runtime-presence:{}:{}:{}",
record.target, record.runtime_instance_id, record.publisher_sequence
),
document: CultNetRawDocumentRecord {
schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(),
record_key: record.target.clone(),
stored_at: chrono::DateTime::from_timestamp_millis(
record.observed_at_unix_millis.try_into()?,
)
.context("runtime presence timestamp is invalid")?
.to_rfc3339(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
document: runtime_presence_document(
record,
payload,
source_runtime_id: Some(record.runtime_id.clone()),
source_agent_id: Some(record.signer_identity_id.clone()),
source_role: Some("runtime-presence-publisher".into()),
tags: Some(vec![
Some(vec![
CULTNET_RUDP_PROTOCOL_ID.into(),
"runtime-presence".into(),
]),
},
)?,
};
let bind = if endpoint.is_ipv4() {
"0.0.0.0:0"
@ -1032,6 +1043,38 @@ fn publish_runtime_presence(
Ok(canonical_sha256)
}
fn canonical_presence_payload(record: &GameCultRuntimePresenceHealthRecord) -> Result<Vec<u8>> {
let payload = rmp_serde::to_vec(record).context("encoding canonical runtime presence")?;
let decoded: GameCultRuntimePresenceHealthRecord = rmp_serde::from_slice(&payload)?;
ensure!(
decoded == *record && rmp_serde::to_vec(&decoded)? == payload,
"runtime presence encoding is noncanonical"
);
Ok(payload)
}
fn runtime_presence_document(
record: &GameCultRuntimePresenceHealthRecord,
payload: Vec<u8>,
tags: Option<Vec<String>>,
) -> Result<CultNetRawDocumentRecord> {
Ok(CultNetRawDocumentRecord {
schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(),
record_key: record.target.clone(),
stored_at: chrono::DateTime::from_timestamp_millis(
record.observed_at_unix_millis.try_into()?,
)
.context("runtime presence timestamp is invalid")?
.to_rfc3339(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
payload,
source_runtime_id: Some(record.runtime_id.clone()),
source_agent_id: Some(record.signer_identity_id.clone()),
source_role: Some("runtime-presence-publisher".into()),
tags,
})
}
fn sibling_lock_path(path: &Path) -> PathBuf {
let mut name = path
.file_name()
@ -1042,7 +1085,7 @@ fn sibling_lock_path(path: &Path) -> PathBuf {
}
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
use cultnet_rs::{
GameCultRuntimePresenceHealthPurpose, IdunnExpectedCapability, IdunnExpectedRoute,
@ -1061,6 +1104,12 @@ mod tests {
activation: IdunnRuntimeActivationRecord,
}
pub(crate) struct RouteObservationFixture {
pub(crate) publisher: RuntimePresencePublisher,
pub(crate) write_lease: ProcessWriteLeaseGuard,
pub(crate) authority: cultnet_rs::VerifiedRuntimeAuthority,
}
fn managed_dependency(
kind: &str,
capability: &str,
@ -1189,6 +1238,51 @@ mod tests {
fixture_with_state_contract(root, STATE_CONTRACT_SHA256)
}
pub(crate) fn route_observation_fixture(root: &Path) -> Result<RouteObservationFixture> {
let mut fixture = fixture(root)?;
let warming =
fixture
.publisher
.signed_record("warming", "process-write-lease-pending", None)?;
let warming_sha256 = warming.canonical_sha256()?;
let lease = IdunnProcessWriteLeaseRecord {
schema_version: IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into(),
target: fixture.expected.target.clone(),
expected_projection_sha256: fixture.expected.canonical_sha256()?,
plan_id: fixture.expected.plan_id.clone(),
incarnation_id: fixture.expected.incarnation_id.clone(),
sealed_release_id: fixture.expected.sealed_release_id.clone(),
activation_witness_sha256: fixture.activation.canonical_sha256()?,
state_schema_generation: STATE_SCHEMA_GENERATION.into(),
state_contract_sha256: fixture.expected.state_contract_sha256.clone().unwrap(),
runtime_id: fixture.expected.runtime_id.clone(),
runtime_instance_id: fixture.activation.runtime_instance_id.clone(),
warming_presence_sha256: warming_sha256.clone(),
lease_epoch: 1,
issued_at_unix_millis: 200,
};
let path = root.join("write-lease.cc");
SingleFileMessagePackBackingStore::new(&path).push(&CultCacheEnvelope {
key: lease.target.clone(),
r#type: IdunnProcessWriteLeaseRecord::TYPE.into(),
payload: lease.canonical_bytes()?,
stored_at: "2026-09-03T00:00:00Z".into(),
schema_id: Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into()),
})?;
let write_lease = ProcessWriteLeaseGuard::acquire_fixture(
path,
fixture.expected,
fixture.activation,
warming_sha256,
)?;
fixture.publisher.active_write_lease_sha256 = Some(write_lease.canonical_sha256().into());
Ok(RouteObservationFixture {
publisher: fixture.publisher,
write_lease,
authority: fixture.authority,
})
}
#[cfg(target_os = "linux")]
#[test]
fn inherited_authority_fd_layout_rejects_substitution_and_extras() -> Result<()> {
@ -1361,6 +1455,55 @@ mod tests {
)?;
guard.require_current()?;
assert_eq!(guard.canonical_sha256(), lease.canonical_sha256()?);
let lease_sha256 = lease.canonical_sha256()?;
fixture.publisher.active_write_lease_sha256 = Some(lease_sha256.clone());
let response = fixture
.publisher
.route_observation("challenge-31", true, &guard)?;
let CultNetMessage::SnapshotResponseRaw {
message_id,
documents,
} = response
else {
panic!("route observation was not a raw snapshot response");
};
assert_eq!(message_id, "challenge-31");
let [document] = documents.as_slice() else {
panic!("route observation did not contain exactly one document");
};
assert_eq!(document.schema_id, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA);
assert_eq!(document.record_key, TARGET);
assert_eq!(document.tags, None);
let presence: GameCultRuntimePresenceHealthRecord =
rmp_serde::from_slice(&document.payload)?;
assert_eq!(presence.state, "active");
assert_eq!(presence.detail, "route-observation:challenge-31");
assert_eq!(
presence.write_lease_sha256.as_deref(),
Some(lease_sha256.as_str())
);
let claim = authenticate_runtime_presence_claim(
&document.payload,
&fixture.authority,
RuntimePresenceAuthenticationContext {
trusted_received_at_unix_millis: presence.observed_at_unix_millis,
maximum_age_millis: 1_000,
maximum_future_skew_millis: 10,
},
)?;
assert_eq!(claim.record(), &presence);
let degraded = fixture
.publisher
.route_observation("challenge-32", false, &guard)?;
let CultNetMessage::SnapshotResponseRaw { documents, .. } = degraded else {
panic!("degraded route observation was not a raw snapshot response");
};
let degraded: GameCultRuntimePresenceHealthRecord =
rmp_serde::from_slice(&documents[0].payload)?;
assert_eq!(degraded.state, "degraded");
assert_eq!(degraded.detail, "route-observation:challenge-32");
drop(guard);
let mut wrong = lease;

345
crates/ghostlight-dungeon/src/runtime.rs
View file

@ -6,6 +6,7 @@ use crate::{
heimdall::{self, HeimdallClient},
idunn_health::{
IDUNN_RUNTIME_CANDIDATE_BIND_ENVIRONMENT, ProcessWriteLeaseGuard, RuntimePresencePublisher,
TARGET as GHOSTLIGHT_TARGET,
},
mesh::{self, MeshPublisher, MeshRuntimeIdentity},
world::{
@ -19,7 +20,8 @@ use crate::{
use anyhow::{Context, bail, ensure};
use axum::{
Json, Router,
extract::{Path, State},
body::Bytes,
extract::{ConnectInfo, DefaultBodyLimit, Path, State},
http::{HeaderMap, HeaderValue, StatusCode, header},
response::{
IntoResponse, Response,
@ -28,6 +30,10 @@ use axum::{
routing::{get, post},
};
use chrono::{DateTime, Utc};
use cultnet_rs::{
CultNetMessage, CultNetWireContract, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA,
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
};
use serde::Deserialize;
use serde_json::{Value, json};
use std::{convert::Infallible, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
@ -36,6 +42,13 @@ use tower_http::services::ServeDir;
use tracing_subscriber::EnvFilter;
const COOKIE_NAME: &str = "ghostlight_session";
const CULTNET_SNAPSHOT_BODY_LIMIT: usize = 16 * 1024;
#[derive(Clone)]
struct RuntimeHealthOwner {
publisher: Arc<Mutex<RuntimePresencePublisher>>,
write_lease: Arc<ProcessWriteLeaseGuard>,
}
#[derive(Clone)]
struct AppState {
@ -45,6 +58,7 @@ struct AppState {
heimdall: Arc<HeimdallClient>,
mesh: Option<MeshPublisher>,
mesh_identity: MeshRuntimeIdentity,
runtime_health: Option<RuntimeHealthOwner>,
revisions: broadcast::Sender<u64>,
fatal: mpsc::UnboundedSender<String>,
}
@ -191,7 +205,7 @@ pub(crate) async fn run(state_root_binding: Option<PathBuf>) -> anyhow::Result<(
};
require_no_runtime_custody_failure(&mut fatal_events)?;
let (revisions, _) = broadcast::channel(32);
let state = AppState {
let mut state = AppState {
world,
controllers: Arc::new(Mutex::new(controllers)),
sessions: Arc::new(Mutex::new(sessions)),
@ -200,10 +214,10 @@ pub(crate) async fn run(state_root_binding: Option<PathBuf>) -> anyhow::Result<(
mesh_identity: mesh_identity.clone(),
revisions,
fatal,
runtime_health: None,
};
require_no_runtime_custody_failure(&mut fatal_events)?;
publish_projection(&state).await?;
tokio::spawn(maintain_mesh_projection(state.clone()));
let release_web_root = std::env::current_exe()?
.parent()
@ -211,7 +225,6 @@ pub(crate) async fn run(state_root_binding: Option<PathBuf>) -> anyhow::Result<(
let web_root = release_web_root
.filter(|path| path.join("index.html").is_file())
.unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../web/dist"));
let app = app_router(state.clone(), web_root);
require_no_runtime_custody_failure(&mut fatal_events)?;
require_current_write_lease(write_lease_guard.as_ref())?;
canonical_readiness(&state)
@ -226,6 +239,16 @@ pub(crate) async fn run(state_root_binding: Option<PathBuf>) -> anyhow::Result<(
.context("publishing initial active runtime presence")?;
}
require_current_write_lease(write_lease_guard.as_ref())?;
state.runtime_health = match (idunn_health, write_lease_guard) {
(Some(publisher), Some(write_lease)) => Some(RuntimeHealthOwner {
publisher: Arc::new(Mutex::new(publisher)),
write_lease: Arc::new(write_lease),
}),
(None, None) => None,
_ => bail!("managed runtime health authority is partial"),
};
tokio::spawn(maintain_mesh_projection(state.clone()));
let app = app_router(state.clone(), web_root);
tracing::info!(address = %bound_address, "Ghostlight Dungeon world owner serving");
let server = axum::serve(
listener,
@ -237,11 +260,7 @@ pub(crate) async fn run(state_root_binding: Option<PathBuf>) -> anyhow::Result<(
result.context("world-owner task panicked")?;
bail!("world-owner task stopped while the daemon was serving")
}
result = maintain_runtime_health(
idunn_health,
state.clone(),
write_lease_guard.as_ref(),
) => {
result = maintain_runtime_health(state.clone()) => {
result?;
bail!("runtime readiness owner stopped while the daemon was serving")
}
@ -303,6 +322,10 @@ fn app_router(state: AppState, web_root: PathBuf) -> Router {
fn api_router(state: AppState) -> Router {
Router::new()
.route("/health", get(health))
.route(
"/cultnet/snapshot",
post(cultnet_snapshot).layer(DefaultBodyLimit::max(CULTNET_SNAPSHOT_BODY_LIMIT)),
)
.route("/api/eve/provider", get(eve_provider))
.route("/api/eve/surfaces/{surface_id}", get(eve_surface))
.route("/api/eve/commands", post(eve_command))
@ -310,6 +333,85 @@ fn api_router(state: AppState) -> Router {
.with_state(state)
}
async fn cultnet_snapshot(
ConnectInfo(peer): ConnectInfo<SocketAddr>,
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
) -> Response {
if !peer.ip().is_loopback() {
return StatusCode::FORBIDDEN.into_response();
}
if headers.get(header::CONTENT_TYPE) != Some(&HeaderValue::from_static("application/msgpack")) {
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
}
let message_id = match exact_route_observation_message_id(&body) {
Ok(message_id) => message_id,
Err(_) => return StatusCode::BAD_REQUEST.into_response(),
};
let Some(owner) = state.runtime_health.clone() else {
return StatusCode::SERVICE_UNAVAILABLE.into_response();
};
// Hold the one publisher across readiness observation and signing so a
// periodic publication cannot age the checked state before this challenge
// is bound into the signed record.
let write_lease = owner.write_lease.clone();
let mut publisher = owner.publisher.clone().lock_owned().await;
let ready = canonical_readiness(&state).await.is_ok();
let response = tokio::task::spawn_blocking(move || {
let response = publisher.route_observation(&message_id, ready, &write_lease)?;
encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0)
})
.await;
let Ok(Ok(response)) = response else {
return StatusCode::SERVICE_UNAVAILABLE.into_response();
};
(
StatusCode::OK,
[(header::CONTENT_TYPE, "application/msgpack")],
response,
)
.into_response()
}
fn exact_route_observation_message_id(body: &[u8]) -> anyhow::Result<String> {
let request = decode_cultnet_message_from_slice(body, CultNetWireContract::CultNetSchemaV0)
.context("decoding route-observation request")?;
ensure!(
encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)? == body,
"route-observation request is not canonical MessagePack"
);
let CultNetMessage::SnapshotRequest {
message_id,
schema_ids,
record_keys,
} = request
else {
bail!("route-observation request is not a snapshot request");
};
ensure!(
!message_id.is_empty(),
"route-observation message id is empty"
);
ensure!(
matches!(
schema_ids.as_deref(),
Some([schema_id]) if schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA
) && matches!(
record_keys.as_deref(),
Some([record_key]) if record_key == GHOSTLIGHT_TARGET
),
"route-observation request is not the exact Ghostlight presence record"
);
let detail = format!("route-observation:{message_id}");
ensure!(
detail.len() <= 512 && !detail.chars().any(char::is_control),
"route-observation message id cannot be signed canonically"
);
Ok(message_id)
}
async fn health(State(state): State<AppState>) -> Response {
match runtime_readiness(&state).await {
Ok(health) => Json(health).into_response(),
@ -1464,12 +1566,7 @@ async fn initialize_production_admission(
}))
}
async fn maintain_runtime_health(
publisher: Option<RuntimePresencePublisher>,
state: AppState,
write_lease: Option<&ProcessWriteLeaseGuard>,
) -> anyhow::Result<()> {
let mut publisher = publisher;
async fn maintain_runtime_health(state: AppState) -> anyhow::Result<()> {
let mut interval = tokio::time::interval(Duration::from_secs(10));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
@ -1477,22 +1574,25 @@ async fn maintain_runtime_health(
canonical_readiness(&state)
.await
.context("runtime is not ready for signed health")?;
require_current_write_lease(write_lease)
let Some(owner) = state.runtime_health.clone() else {
continue;
};
owner
.write_lease
.require_current()
.context("runtime process write lease is not current")?;
if let Some(mut owned) = publisher.take() {
let published = tokio::task::spawn_blocking(move || {
let result = owned.republish_active();
(owned, result)
})
.await;
let Ok((returned, result)) = published else {
tracing::warn!("runtime-presence worker panicked; publication is disabled");
continue;
};
publisher = Some(returned);
if let Err(error) = result {
tracing::warn!(%error, "signed runtime-presence publication failed");
}
let mut publisher = owner.publisher.clone().lock_owned().await;
let published = tokio::task::spawn_blocking(move || {
publisher.republish_active()?;
Ok::<(), anyhow::Error>(())
})
.await;
let Ok(result) = published else {
tracing::warn!("runtime-presence worker panicked; publication will retry");
continue;
};
if let Err(error) = result {
tracing::warn!(%error, "signed runtime-presence publication failed");
}
}
}
@ -1658,8 +1758,16 @@ fn sibling_state_lock_path(path: &std::path::Path) -> PathBuf {
#[cfg(test)]
mod tests {
use super::*;
use crate::idunn_health::tests::route_observation_fixture;
use crate::world::AffordanceKind;
use axum::{body::Body, http::Request};
use axum::{
body::{Body, to_bytes},
http::Request,
};
use cultnet_rs::{
GameCultRuntimePresenceHealthRecord, RuntimePresenceAuthenticationContext,
authenticate_runtime_presence_claim,
};
use tower::ServiceExt;
#[cfg(target_os = "linux")]
@ -1789,6 +1897,7 @@ mod tests {
heimdall: Arc::new(HeimdallClient::fixture()),
mesh: Some(mesh),
mesh_identity,
runtime_health: None,
revisions,
fatal,
};
@ -1800,6 +1909,180 @@ mod tests {
}
}
fn route_snapshot_request(
message_id: &str,
schema_ids: Option<Vec<String>>,
record_keys: Option<Vec<String>>,
) -> Vec<u8> {
encode_cultnet_message_to_vec(
&CultNetMessage::SnapshotRequest {
message_id: message_id.into(),
schema_ids,
record_keys,
},
CultNetWireContract::CultNetSchemaV0,
)
.unwrap()
}
#[test]
fn route_observation_request_is_one_exact_canonical_record() {
let exact = route_snapshot_request(
"route-challenge-41",
Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
Some(vec![GHOSTLIGHT_TARGET.into()]),
);
assert_eq!(
exact_route_observation_message_id(&exact).unwrap(),
"route-challenge-41"
);
for refused in [
route_snapshot_request("broad", None, None),
route_snapshot_request(
"foreign-schema",
Some(vec!["gamecult.other.v1".into()]),
Some(vec![GHOSTLIGHT_TARGET.into()]),
),
route_snapshot_request(
"foreign-record",
Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
Some(vec!["other-target".into()]),
),
] {
assert!(exact_route_observation_message_id(&refused).is_err());
}
}
#[tokio::test]
async fn http_route_probe_bypasses_provider_and_enforces_exact_admission() {
let mut fixture = fixture().await;
let body = route_snapshot_request(
"route-challenge-42",
Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
Some(vec![GHOSTLIGHT_TARGET.into()]),
);
let request = |peer: &str, content_type: Option<&str>, body: Vec<u8>| {
let mut builder = Request::builder()
.method("POST")
.uri("/cultnet/snapshot")
.extension(ConnectInfo(peer.parse::<SocketAddr>().unwrap()));
if let Some(content_type) = content_type {
builder = builder.header(header::CONTENT_TYPE, content_type);
}
builder.body(Body::from(body)).unwrap()
};
let unmanaged = api_router(fixture.state.clone())
.oneshot(request(
"127.0.0.1:39001",
Some("application/msgpack"),
body.clone(),
))
.await
.unwrap();
assert_eq!(unmanaged.status(), StatusCode::SERVICE_UNAVAILABLE);
let remote = api_router(fixture.state.clone())
.oneshot(request(
"192.0.2.9:39001",
Some("application/msgpack"),
body.clone(),
))
.await
.unwrap();
assert_eq!(remote.status(), StatusCode::FORBIDDEN);
let wrong_media = api_router(fixture.state.clone())
.oneshot(request(
"127.0.0.1:39001",
Some("application/json"),
body.clone(),
))
.await
.unwrap();
assert_eq!(wrong_media.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
let broad = api_router(fixture.state.clone())
.oneshot(request(
"127.0.0.1:39001",
Some("application/msgpack"),
route_snapshot_request("broad", None, None),
))
.await
.unwrap();
assert_eq!(broad.status(), StatusCode::BAD_REQUEST);
let health = route_observation_fixture(fixture._directory.path()).unwrap();
let authority = health.authority;
let write_lease_sha256 = health.write_lease.canonical_sha256().to_owned();
fixture.state.runtime_health = Some(RuntimeHealthOwner {
publisher: Arc::new(Mutex::new(health.publisher)),
write_lease: Arc::new(health.write_lease),
});
// If this route enters the controller/provider boundary, the held lock
// makes the request time out. Route admission and signing need only the
// canonical health owner and managed runtime authority.
let controllers = fixture.state.controllers.clone();
let _provider_execution_barrier = controllers.lock().await;
let response = tokio::time::timeout(
Duration::from_secs(2),
api_router(fixture.state.clone()).oneshot(request(
"127.0.0.1:39001",
Some("application/msgpack"),
body,
)),
)
.await
.expect("route observation entered provider execution")
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static("application/msgpack"))
);
let bytes = to_bytes(response.into_body(), 64 * 1024).await.unwrap();
let message =
decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0)
.unwrap();
assert_eq!(
encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(),
bytes
);
let CultNetMessage::SnapshotResponseRaw {
message_id,
documents,
} = message
else {
panic!("route observation was not a raw snapshot response");
};
assert_eq!(message_id, "route-challenge-42");
let [document] = documents.as_slice() else {
panic!("route observation did not contain exactly one document");
};
assert_eq!(document.schema_id, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA);
assert_eq!(document.record_key, GHOSTLIGHT_TARGET);
let presence: GameCultRuntimePresenceHealthRecord =
rmp_serde::from_slice(&document.payload).unwrap();
assert_eq!(presence.state, "active");
assert_eq!(presence.detail, "route-observation:route-challenge-42");
assert_eq!(
presence.write_lease_sha256.as_deref(),
Some(write_lease_sha256.as_str())
);
authenticate_runtime_presence_claim(
&document.payload,
&authority,
RuntimePresenceAuthenticationContext {
trusted_received_at_unix_millis: presence.observed_at_unix_millis,
maximum_age_millis: 1_000,
maximum_future_skew_millis: 10,
},
)
.unwrap();
}
fn invocation(operation: &str, schema: &str, version: u64, payload: Value, id: &str) -> Value {
json!({
"schema":"gamecult.eve.command_invocation.v1",