878 lines
28 KiB
Rust
878 lines
28 KiB
Rust
//! Authored Rust -> PDK -> validated Component -> fresh Wasmtime Store -> OwnerHost.
|
|
//! The deterministic owner replaces storage/network, not the execution adapter;
|
|
//! transaction/lease/live policy guarantees are tested by the domain owner suites.
|
|
use async_trait::async_trait;
|
|
use manifest::plugin::{self, PluginPackageLimits, PluginSourceKind};
|
|
use server_api::*;
|
|
use std::{
|
|
collections::BTreeMap,
|
|
fs,
|
|
path::Path,
|
|
process::Command,
|
|
sync::{Arc, Mutex, OnceLock},
|
|
time::{Duration, SystemTime, UNIX_EPOCH},
|
|
};
|
|
use tokio::time::Instant;
|
|
use yoi_plugin_runtime::{
|
|
host::{CancellationToken, Grants, InvocationContext},
|
|
owner::{OwnerHost, PluginHostOwner},
|
|
*,
|
|
};
|
|
|
|
fn component() -> &'static [u8] {
|
|
static COMPONENT: OnceLock<Vec<u8>> = OnceLock::new();
|
|
COMPONENT.get_or_init(|| {
|
|
let target =
|
|
Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/plugin-service-fixture/guest");
|
|
let output = Command::new(env!("CARGO"))
|
|
.args([
|
|
"build",
|
|
"--locked",
|
|
"--offline",
|
|
"--release",
|
|
"--target",
|
|
"wasm32-unknown-unknown",
|
|
"--manifest-path",
|
|
])
|
|
.arg(Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/service-guest/Cargo.toml"))
|
|
.arg("--target-dir")
|
|
.arg(&target)
|
|
.output()
|
|
.unwrap();
|
|
fs::create_dir_all(target.parent().unwrap()).unwrap();
|
|
fs::write(
|
|
target.parent().unwrap().join("guest-build.log"),
|
|
&output.stderr,
|
|
)
|
|
.unwrap();
|
|
assert!(
|
|
output.status.success(),
|
|
"Service fixture requires wasm32 std + lld on PATH: {}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
let core = fs::read(target.join("wasm32-unknown-unknown/release/service_test_guest.wasm"))
|
|
.unwrap();
|
|
let bytes = wit_component::ComponentEncoder::default()
|
|
.module(&core)
|
|
.unwrap()
|
|
.validate(true)
|
|
.encode()
|
|
.unwrap();
|
|
fs::write(
|
|
target.parent().unwrap().join("service.component.wasm"),
|
|
&bytes,
|
|
)
|
|
.unwrap();
|
|
bytes
|
|
})
|
|
}
|
|
const MANIFEST: &str = include_str!("service-guest/plugin.toml");
|
|
fn package(manifest: &str) -> Result<plugin::CanonicalPluginArtifact, plugin::PluginDiagnostic> {
|
|
package_bytes(manifest, component())
|
|
}
|
|
fn package_bytes(
|
|
manifest: &str,
|
|
bytes: &[u8],
|
|
) -> Result<plugin::CanonicalPluginArtifact, plugin::PluginDiagnostic> {
|
|
let target = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/plugin-service-fixture");
|
|
fs::create_dir_all(&target).unwrap();
|
|
let dir = tempfile::tempdir_in(target).unwrap();
|
|
fs::write(dir.path().join("plugin.toml"), manifest).unwrap();
|
|
fs::write(dir.path().join("plugin.component.wasm"), bytes).unwrap();
|
|
let materialized = plugin::read_plugin_directory(
|
|
dir.path(),
|
|
PluginSourceKind::Explicit,
|
|
&PluginPackageLimits::default(),
|
|
)?;
|
|
plugin::pack_plugin_package(&materialized, &PluginPackageLimits::default())
|
|
}
|
|
fn compile(ex: &Executor) -> CompiledService {
|
|
let p = package(MANIFEST).unwrap();
|
|
ex.compile_service(
|
|
&p.bytes,
|
|
&p.package_digest,
|
|
&p.artifact_digest,
|
|
p.artifact_size,
|
|
)
|
|
.unwrap()
|
|
}
|
|
#[derive(Clone, Copy, Default)]
|
|
enum Fault {
|
|
#[default]
|
|
None,
|
|
HttpUnknown,
|
|
DataUnknown,
|
|
HttpWait,
|
|
DataWait,
|
|
LookupUnavailable,
|
|
LookupExpired,
|
|
ReadDenied,
|
|
}
|
|
#[derive(Default)]
|
|
struct State {
|
|
calls: Vec<PluginHostRequest>,
|
|
writes: usize,
|
|
sends: usize,
|
|
receipts: BTreeMap<String, PluginDataReceipt>,
|
|
http: BTreeMap<String, PluginHttpResponse>,
|
|
}
|
|
struct Owner {
|
|
state: Mutex<State>,
|
|
fault: Fault,
|
|
waiting: tokio::sync::Notify,
|
|
}
|
|
impl Owner {
|
|
fn new(fault: Fault) -> Arc<Self> {
|
|
Arc::new(Self {
|
|
state: Mutex::new(State::default()),
|
|
fault,
|
|
waiting: tokio::sync::Notify::new(),
|
|
})
|
|
}
|
|
}
|
|
fn rejected(code: PluginHostErrorCode, effect: PluginEffectObservation) -> PluginHostError {
|
|
PluginHostError {
|
|
code,
|
|
effect,
|
|
diagnostic_id: "test".into(),
|
|
}
|
|
}
|
|
#[async_trait]
|
|
impl PluginHostOwner for Owner {
|
|
async fn call(
|
|
&self,
|
|
request: PluginHostRequest,
|
|
) -> Result<PluginHostResponse, PluginHostError> {
|
|
if (matches!(self.fault, Fault::HttpWait)
|
|
&& matches!(request, PluginHostRequest::Send { .. }))
|
|
|| (matches!(self.fault, Fault::DataWait)
|
|
&& matches!(request, PluginHostRequest::DataMutate { .. }))
|
|
{
|
|
self.waiting.notify_one();
|
|
return std::future::pending().await;
|
|
}
|
|
let mut s = self.state.lock().unwrap();
|
|
s.calls.push(request.clone());
|
|
Ok(match request {
|
|
PluginHostRequest::Begin { .. } => PluginHostResponse::Invocation {
|
|
invocation_id: "stable-service-invocation".into(),
|
|
expires_at_ms: SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_millis() as i64
|
|
+ 60_000,
|
|
},
|
|
PluginHostRequest::Authorize { .. } => PluginHostResponse::Authorized,
|
|
PluginHostRequest::Log { .. } => PluginHostResponse::Logged,
|
|
PluginHostRequest::ResolveTicket { ticket_ref, .. } => {
|
|
assert_eq!(ticket_ref, "T-796");
|
|
PluginHostResponse::TicketResolved {
|
|
ticket_id: "canonical-ticket".into(),
|
|
}
|
|
}
|
|
PluginHostRequest::ReadTicket { .. } => {
|
|
if matches!(self.fault, Fault::ReadDenied) {
|
|
return Err(rejected(
|
|
PluginHostErrorCode::Denied,
|
|
PluginEffectObservation::NotDispatched,
|
|
));
|
|
}
|
|
PluginHostResponse::TicketRead {
|
|
snapshot: PluginTicketSnapshot {
|
|
title: "note".into(),
|
|
case_state: PluginTicketCaseState::Open,
|
|
content_digest: "digest".into(),
|
|
body: "body".into(),
|
|
truncated: false,
|
|
},
|
|
}
|
|
}
|
|
PluginHostRequest::DataLookupMutation { operation_key, .. } => {
|
|
if matches!(self.fault, Fault::LookupUnavailable) {
|
|
return Err(rejected(
|
|
PluginHostErrorCode::Unavailable,
|
|
PluginEffectObservation::NotDispatched,
|
|
));
|
|
}
|
|
PluginHostResponse::DataMutationLookup {
|
|
outcome: if matches!(self.fault, Fault::LookupExpired) {
|
|
PluginDataMutationLookup::Expired
|
|
} else {
|
|
s.receipts
|
|
.get(&operation_key)
|
|
.cloned()
|
|
.map(PluginDataMutationLookup::Committed)
|
|
.unwrap_or(PluginDataMutationLookup::NotRecorded)
|
|
},
|
|
}
|
|
}
|
|
PluginHostRequest::DataGet { .. } => PluginHostResponse::DataCell {
|
|
cell: PluginDataCell::NeverWritten,
|
|
},
|
|
PluginHostRequest::DataMutate { operation_key, .. } => {
|
|
if matches!(self.fault, Fault::DataUnknown) {
|
|
return Err(rejected(
|
|
PluginHostErrorCode::Indeterminate,
|
|
PluginEffectObservation::Indeterminate,
|
|
));
|
|
}
|
|
s.writes += 1;
|
|
let receipt = PluginDataReceipt {
|
|
mutation_id: "write-1".into(),
|
|
result_digest: "result".into(),
|
|
};
|
|
assert!(s.receipts.insert(operation_key, receipt.clone()).is_none());
|
|
PluginHostResponse::DataMutated { receipt }
|
|
}
|
|
PluginHostRequest::Open { .. } => PluginHostResponse::Opened,
|
|
PluginHostRequest::Send { request, .. } => {
|
|
if matches!(self.fault, Fault::HttpUnknown) {
|
|
return Err(rejected(
|
|
PluginHostErrorCode::Indeterminate,
|
|
PluginEffectObservation::Indeterminate,
|
|
));
|
|
}
|
|
let response = if let Some(response) = s.http.get(&request.effect_key) {
|
|
response.clone()
|
|
} else {
|
|
s.sends += 1;
|
|
let response = PluginHttpResponse {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: b"ok".to_vec(),
|
|
};
|
|
s.http.insert(request.effect_key, response.clone());
|
|
response
|
|
};
|
|
PluginHostResponse::Response { response }
|
|
}
|
|
_ => {
|
|
return Err(rejected(
|
|
PluginHostErrorCode::Unsupported,
|
|
PluginEffectObservation::NotDispatched,
|
|
));
|
|
}
|
|
})
|
|
}
|
|
}
|
|
fn context(c: &CompiledService) -> InvocationContext {
|
|
let namespace = ("notes".into(), "note".into(), "1".into());
|
|
InvocationContext {
|
|
workspace: "workspace".into(),
|
|
installation: "installation".into(),
|
|
plan: "plan".into(),
|
|
subject: "worker".into(),
|
|
invocation_id: "ignored-before-begin".into(),
|
|
artifact: c.identity().clone(),
|
|
tool_name: "ticket_events".into(),
|
|
grants: Grants {
|
|
logging: true,
|
|
connection_slots: ["api".into()].into(),
|
|
ticket_events: true,
|
|
ticket_read: true,
|
|
ticket_data_read: [namespace.clone()].into(),
|
|
ticket_data_write: [namespace].into(),
|
|
},
|
|
deadline: Instant::now() + Duration::from_secs(10),
|
|
cancellation: CancellationToken::new(),
|
|
}
|
|
}
|
|
async fn host(c: &CompiledService, owner: Arc<Owner>) -> Arc<OwnerHost> {
|
|
Arc::new(
|
|
OwnerHost::begin(
|
|
owner,
|
|
context(c),
|
|
PluginConsumer::Consumer {
|
|
consumer_id: "consumer".into(),
|
|
event_id: "event".into(),
|
|
export_name: "ticket_events".into(),
|
|
},
|
|
)
|
|
.await
|
|
.unwrap(),
|
|
)
|
|
}
|
|
fn envelope() -> PluginTicketEventEnvelope {
|
|
PluginTicketEventEnvelope {
|
|
event_id: "event".into(),
|
|
event_type: PluginTicketEventType::Changed,
|
|
version: 1,
|
|
occurred_at: "2026-10-30T00:00:00Z".into(),
|
|
ticket_id: "T-796".into(),
|
|
domain_event_ref: "domain-event".into(),
|
|
content_digest: "digest".into(),
|
|
correlation_id: "correlation".into(),
|
|
causation_id: None,
|
|
attempt: 1,
|
|
}
|
|
}
|
|
#[tokio::test]
|
|
async fn authored_service_recovers_with_receipt_lookup_before_rebuilding_a_mutation() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
let owner = Owner::new(Fault::None);
|
|
let host = host(&c, owner.clone()).await;
|
|
assert!(matches!(
|
|
ex.service_start(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
r#"{"mode":"normal"}"#
|
|
)
|
|
.await
|
|
.outcome,
|
|
Ok(ServiceOutput::Started)
|
|
));
|
|
for attempt in 1..=2 {
|
|
let mut e = envelope();
|
|
e.attempt = attempt;
|
|
let result = ex
|
|
.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&e,
|
|
r#"{"mode":"normal"}"#,
|
|
)
|
|
.await;
|
|
assert!(
|
|
matches!(
|
|
result.outcome,
|
|
Ok(ServiceOutput::Event(PluginServiceEventResult::Ack))
|
|
),
|
|
"{result:?}"
|
|
);
|
|
assert!(matches!(
|
|
result.effect,
|
|
host::errors::EffectObservation::ResponseObserved
|
|
));
|
|
}
|
|
assert!(matches!(
|
|
ex.service_stop(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
"normal"
|
|
)
|
|
.await
|
|
.outcome,
|
|
Ok(ServiceOutput::Stopped)
|
|
));
|
|
let state = owner.state.lock().unwrap();
|
|
assert_eq!(state.writes, 1);
|
|
assert_eq!(state.sends, 1);
|
|
assert_eq!(
|
|
state
|
|
.calls
|
|
.iter()
|
|
.filter(|r| matches!(r, PluginHostRequest::DataGet { .. }))
|
|
.count(),
|
|
1
|
|
);
|
|
assert_eq!(
|
|
state
|
|
.calls
|
|
.iter()
|
|
.filter(|r| matches!(r, PluginHostRequest::DataLookupMutation { .. }))
|
|
.count(),
|
|
2
|
|
);
|
|
}
|
|
#[tokio::test]
|
|
async fn lifecycle_hooks_allow_logging_but_never_open_resolve_read_data_or_http() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
let owner = Owner::new(Fault::None);
|
|
let host = host(&c, owner.clone()).await;
|
|
for mode in ["open", "resolve", "read", "data", "http"] {
|
|
let config = format!(r#"{{"mode":"{mode}"}}"#);
|
|
let start = ex
|
|
.service_start(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&config,
|
|
)
|
|
.await;
|
|
let stop = ex
|
|
.service_stop(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
mode,
|
|
)
|
|
.await;
|
|
assert!(
|
|
matches!(start.outcome, Err(ExecutionError::Trap(_))),
|
|
"{mode}: {start:?}"
|
|
);
|
|
assert!(
|
|
matches!(stop.outcome, Err(ExecutionError::Trap(_))),
|
|
"{mode}: {stop:?}"
|
|
);
|
|
}
|
|
let s = owner.state.lock().unwrap();
|
|
assert!(s.calls.iter().all(|r| matches!(
|
|
r,
|
|
PluginHostRequest::Begin { .. } | PluginHostRequest::Authorize { .. }
|
|
)));
|
|
}
|
|
#[tokio::test]
|
|
async fn guest_ack_cannot_hide_unknown_http_data_or_failed_receipt_lookup() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
for fault in [
|
|
Fault::HttpUnknown,
|
|
Fault::DataUnknown,
|
|
Fault::LookupUnavailable,
|
|
Fault::LookupExpired,
|
|
] {
|
|
let host = host(&c, Owner::new(fault)).await;
|
|
let result = ex
|
|
.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&envelope(),
|
|
r#"{"mode":"swallow-unknown"}"#,
|
|
)
|
|
.await;
|
|
assert!(
|
|
matches!(
|
|
result.outcome,
|
|
Ok(ServiceOutput::Event(PluginServiceEventResult::Ack))
|
|
),
|
|
"{result:?}"
|
|
);
|
|
assert!(
|
|
matches!(
|
|
result.effect,
|
|
host::errors::EffectObservation::Indeterminate
|
|
),
|
|
"{result:?}"
|
|
);
|
|
}
|
|
}
|
|
#[tokio::test]
|
|
async fn cancellation_during_effect_wait_preserves_unknown_observation() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
for fault in [Fault::HttpWait, Fault::DataWait] {
|
|
let owner = Owner::new(fault);
|
|
let host = host(&c, owner.clone()).await;
|
|
let cancellation = host.context().cancellation.clone();
|
|
let e = envelope();
|
|
let (result, ()) = tokio::time::timeout(Duration::from_secs(5), async {
|
|
tokio::join!(
|
|
ex.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&e,
|
|
r#"{"mode":"normal"}"#
|
|
),
|
|
async {
|
|
owner.waiting.notified().await;
|
|
cancellation.cancel();
|
|
}
|
|
)
|
|
})
|
|
.await
|
|
.expect("effect was not dispatched");
|
|
assert!(
|
|
matches!(result.outcome, Err(ExecutionError::Cancelled)),
|
|
"{result:?}"
|
|
);
|
|
assert!(
|
|
matches!(
|
|
result.effect,
|
|
host::errors::EffectObservation::Indeterminate
|
|
),
|
|
"{result:?}"
|
|
);
|
|
}
|
|
}
|
|
#[tokio::test]
|
|
async fn deleted_metadata_never_resolves_a_ticket_and_live_read_denial_remains_a_failure() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
let owner = Owner::new(Fault::ReadDenied);
|
|
let host = host(&c, owner.clone()).await;
|
|
let result = ex
|
|
.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&envelope(),
|
|
r#"{"mode":"normal"}"#,
|
|
)
|
|
.await;
|
|
assert!(matches!(
|
|
result.outcome,
|
|
Ok(ServiceOutput::Event(
|
|
PluginServiceEventResult::RetryAfter { .. }
|
|
))
|
|
));
|
|
let before = owner.state.lock().unwrap().calls.len();
|
|
let mut e = envelope();
|
|
e.event_type = PluginTicketEventType::Deleted;
|
|
e.content_digest.clear();
|
|
let result = ex
|
|
.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host,
|
|
"ticket_events",
|
|
&e,
|
|
r#"{"mode":"normal"}"#,
|
|
)
|
|
.await;
|
|
assert!(matches!(
|
|
result.outcome,
|
|
Ok(ServiceOutput::Event(PluginServiceEventResult::Ack))
|
|
));
|
|
assert!(
|
|
owner.state.lock().unwrap().calls[before..]
|
|
.iter()
|
|
.all(|r| matches!(r, PluginHostRequest::Authorize { .. }))
|
|
);
|
|
}
|
|
#[test]
|
|
fn service_manifest_rejects_old_world_subscription_and_grant_mismatches() {
|
|
package(MANIFEST).unwrap();
|
|
for bad in [
|
|
MANIFEST.replace("@2.2.0", "@2.1.0"),
|
|
MANIFEST.replace("ticket_changed", "arbitrary_event"),
|
|
MANIFEST.replace("kind = \"ticket_events\"", "kind = \"ticket_read\""),
|
|
MANIFEST.replace(
|
|
"ticket_changed\", \"ticket_deleted",
|
|
"ticket_changed\", \"ticket_changed",
|
|
),
|
|
MANIFEST.replace("surfaces = [\"service\"]", "surfaces = [\"tool\"]"),
|
|
] {
|
|
assert!(package(&bad).is_err());
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn service_host_and_guest_limits_fail_before_publication() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
let c = compile(&ex);
|
|
let host = host(&c, Owner::new(Fault::None)).await;
|
|
let invalid = ex
|
|
.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&envelope(),
|
|
r#"{"mode":"oversize-retry"}"#,
|
|
)
|
|
.await;
|
|
assert!(matches!(
|
|
invalid.outcome,
|
|
Err(ExecutionError::InvalidOutput(_))
|
|
));
|
|
let mut ctx = host.context().clone();
|
|
ctx.grants.ticket_events = false;
|
|
let denied = ex
|
|
.service_event(
|
|
&c,
|
|
ctx,
|
|
host.clone(),
|
|
"ticket_events",
|
|
&envelope(),
|
|
r#"{"mode":"normal"}"#,
|
|
)
|
|
.await;
|
|
assert!(matches!(
|
|
denied.outcome,
|
|
Err(ExecutionError::InvalidInput(_))
|
|
));
|
|
let mut e = envelope();
|
|
e.version = 2;
|
|
assert!(matches!(
|
|
ex.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&e,
|
|
"{}"
|
|
)
|
|
.await
|
|
.outcome,
|
|
Err(ExecutionError::InvalidInput(_))
|
|
));
|
|
e = envelope();
|
|
e.correlation_id = "x".repeat(65536);
|
|
assert!(matches!(
|
|
ex.service_event(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
&e,
|
|
"{}"
|
|
)
|
|
.await
|
|
.outcome,
|
|
Err(ExecutionError::InvalidInput(_))
|
|
));
|
|
let ctx = host.context().clone();
|
|
ctx.cancellation.cancel();
|
|
assert!(matches!(
|
|
ex.service_start(&c, ctx, host.clone(), "ticket_events", "{}")
|
|
.await
|
|
.outcome,
|
|
Err(ExecutionError::Cancelled)
|
|
));
|
|
}
|
|
|
|
fn startup_component(logging: bool) -> Vec<u8> {
|
|
let mut resolve = wit_parser::Resolve::default();
|
|
let (package, _) = resolve
|
|
.push_dir(Path::new(env!("CARGO_MANIFEST_DIR")).join("../../resources/plugin/wit-service"))
|
|
.unwrap();
|
|
let world = resolve
|
|
.select_world(&[package], Some("ticket-service-plugin"))
|
|
.unwrap();
|
|
let core = wit_component::dummy_module(&resolve, world, wit_parser::ManglingAndAbi::Standard32);
|
|
let wat = wasmprinter::print_bytes(core).unwrap();
|
|
let init = wat
|
|
.lines()
|
|
.find(|l| l.contains("(export \"cm32p2_initialize\""))
|
|
.unwrap();
|
|
let index = init
|
|
.split("(func ")
|
|
.nth(1)
|
|
.unwrap()
|
|
.split(')')
|
|
.next()
|
|
.unwrap();
|
|
let needle = wat
|
|
.lines()
|
|
.find(|l| l.contains(&format!("(func (;{index};)")))
|
|
.unwrap();
|
|
assert!(needle.trim_end().ends_with("))"), "{needle}");
|
|
let import = wat
|
|
.lines()
|
|
.find(|l| {
|
|
l.contains(if logging {
|
|
"yoi:host/logging@2"
|
|
} else {
|
|
"yoi:host/connections@2"
|
|
}) && l.contains(if logging { "\"log\"" } else { "\"open\"" })
|
|
})
|
|
.unwrap();
|
|
let imported_index = import
|
|
.split("(func (;")
|
|
.nth(1)
|
|
.unwrap()
|
|
.split(';')
|
|
.next()
|
|
.unwrap();
|
|
let body = if logging {
|
|
"i32.const 0 ".repeat(7) + &format!("i32.const 128 call {imported_index}")
|
|
} else {
|
|
format!("i32.const 0 i32.const 0 i32.const 128 call {imported_index}")
|
|
};
|
|
let replacement = format!("{} {body})", needle.trim_end().strip_suffix(')').unwrap());
|
|
let wat = wat
|
|
.replace(needle, &replacement)
|
|
.replace("(memory (;0;) 0)", "(memory (;0;) 1)");
|
|
let mut core = wat::parse_str(wat).unwrap();
|
|
wit_component::embed_component_metadata(
|
|
&mut core,
|
|
&resolve,
|
|
world,
|
|
wit_component::StringEncoding::UTF8,
|
|
)
|
|
.unwrap();
|
|
wit_component::ComponentEncoder::default()
|
|
.module(&core)
|
|
.unwrap()
|
|
.validate(true)
|
|
.encode()
|
|
.unwrap()
|
|
}
|
|
#[tokio::test]
|
|
async fn component_start_sections_never_receive_a_service_export_context() {
|
|
let ex = Executor::new(Limits::default()).unwrap();
|
|
for logging in [false, true] {
|
|
let p = package_bytes(MANIFEST, &startup_component(logging)).unwrap();
|
|
let c = ex
|
|
.compile_service(
|
|
&p.bytes,
|
|
&p.package_digest,
|
|
&p.artifact_digest,
|
|
p.artifact_size,
|
|
)
|
|
.unwrap();
|
|
let owner = Owner::new(Fault::None);
|
|
let host = host(&c, owner.clone()).await;
|
|
let result = ex
|
|
.service_start(
|
|
&c,
|
|
host.context().clone(),
|
|
host.clone(),
|
|
"ticket_events",
|
|
r#"{"mode":"normal"}"#,
|
|
)
|
|
.await;
|
|
assert!(
|
|
matches!(result.outcome, Err(ExecutionError::Trap(_))),
|
|
"{result:?}"
|
|
);
|
|
assert!(owner.state.lock().unwrap().calls.iter().all(|r| matches!(
|
|
r,
|
|
PluginHostRequest::Begin { .. } | PluginHostRequest::Authorize { .. }
|
|
)));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn artifact_delivery_binds_service_names_and_subscriptions_to_component_package() {
|
|
use delivery::{
|
|
ArtifactCache, ArtifactExpectation, ArtifactSource, ArtifactStream, CacheLimits,
|
|
DeliveryError,
|
|
};
|
|
|
|
struct Source {
|
|
bytes: Vec<u8>,
|
|
metadata: PluginArtifactDeliveryMetadata,
|
|
}
|
|
#[async_trait]
|
|
impl ArtifactSource for Source {
|
|
fn server_authority_identity(&self) -> &str {
|
|
"service-test-authority"
|
|
}
|
|
async fn authorize(
|
|
&self,
|
|
_: &PluginArtifactDeliveryRequest,
|
|
) -> Result<PluginArtifactDeliveryMetadata, DeliveryError> {
|
|
Ok(self.metadata.clone())
|
|
}
|
|
async fn fetch(
|
|
&self,
|
|
_: &PluginArtifactDeliveryRequest,
|
|
) -> Result<ArtifactStream, DeliveryError> {
|
|
Ok(Box::pin(std::io::Cursor::new(self.bytes.clone())))
|
|
}
|
|
}
|
|
fn metadata<T: serde::Serialize, U: serde::de::DeserializeOwned>(value: &T) -> U {
|
|
serde_json::from_value(serde_json::to_value(value).unwrap()).unwrap()
|
|
}
|
|
|
|
let artifact = package(MANIFEST).unwrap();
|
|
let verified = plugin::verify_plugin_artifact(
|
|
&artifact.bytes,
|
|
&artifact.package_digest,
|
|
&artifact.artifact_digest,
|
|
artifact.artifact_size,
|
|
&PluginPackageLimits::default(),
|
|
)
|
|
.unwrap();
|
|
let m = &verified.package.manifest;
|
|
let schema = plugin::plugin_config_schema_identity(m);
|
|
let descriptor = PluginArtifact {
|
|
package_id: m.id.clone(),
|
|
name: m.name.clone(),
|
|
version: m.version.clone(),
|
|
description: m.description.clone(),
|
|
package_digest: artifact.package_digest.clone(),
|
|
artifact_digest: artifact.artifact_digest.clone(),
|
|
artifact_size: artifact.artifact_size.try_into().unwrap(),
|
|
schema_version: m.schema_version,
|
|
world: m.runtime.world.clone(),
|
|
required_imports: m.runtime.required_imports.clone(),
|
|
tools: metadata(&m.tools),
|
|
services: metadata(&m.services),
|
|
connections: metadata(&m.connections),
|
|
requested_grants: metadata(&m.requested_grants),
|
|
data_collections: metadata(&m.data_collections),
|
|
config_schema: PluginSchemaContribution {
|
|
identity: schema.identity,
|
|
version: schema.version,
|
|
digest: schema.digest,
|
|
namespace: schema.namespace,
|
|
dcdl: m.config_schema.clone(),
|
|
},
|
|
revoked: false,
|
|
revocation_reason: None,
|
|
};
|
|
assert_eq!(
|
|
descriptor.services[0].subscriptions,
|
|
["ticket_changed", "ticket_deleted"]
|
|
);
|
|
let request = PluginArtifactDeliveryRequest {
|
|
plan_id: "service-plan".into(),
|
|
plan_fingerprint: "service-plan-fingerprint".into(),
|
|
installation_id: "installation".into(),
|
|
artifact_digest: artifact.artifact_digest.clone(),
|
|
};
|
|
let directory = tempfile::tempdir().unwrap();
|
|
let cache = ArtifactCache::new(directory.path().join("cache"), CacheLimits::default()).unwrap();
|
|
let mut source = Source {
|
|
bytes: artifact.bytes.clone(),
|
|
metadata: PluginArtifactDeliveryMetadata {
|
|
workspace_id: "workspace".into(),
|
|
request: request.clone(),
|
|
artifact: descriptor.clone(),
|
|
},
|
|
};
|
|
let resolved = cache
|
|
.resolve(
|
|
&source,
|
|
"workspace",
|
|
&request,
|
|
&ArtifactExpectation::new(descriptor.clone()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resolved.bytes(), artifact.bytes);
|
|
|
|
let mut renamed = descriptor.clone();
|
|
renamed.services[0].name = "different_service".into();
|
|
let mut narrowed = descriptor.clone();
|
|
narrowed.services[0].subscriptions = vec!["ticket_changed".into()];
|
|
let mut unknown = descriptor.clone();
|
|
unknown.services[0].subscriptions = vec!["arbitrary_event".into()];
|
|
let mut missing = descriptor.clone();
|
|
missing.services.clear();
|
|
for tampered in [renamed, narrowed, unknown, missing] {
|
|
// Source metadata and the trusted descriptor agree; only verification
|
|
// against the actual package can detect this mismatch, including hits.
|
|
source.metadata.artifact = tampered.clone();
|
|
let expected = ArtifactExpectation::new(tampered);
|
|
assert!(matches!(
|
|
cache
|
|
.resolve(&source, "workspace", &request, &expected)
|
|
.await,
|
|
Err(DeliveryError::MetadataMismatch)
|
|
));
|
|
}
|
|
source.metadata.artifact = descriptor.clone();
|
|
assert_eq!(
|
|
cache
|
|
.resolve(
|
|
&source,
|
|
"workspace",
|
|
&request,
|
|
&ArtifactExpectation::new(descriptor)
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.bytes(),
|
|
artifact.bytes
|
|
);
|
|
}
|