feat: deliver committed Ticket events to plugin services
This commit is contained in:
@@ -1068,6 +1068,7 @@ where
|
||||
let notify_buffer = worker.notify_buffer_handle();
|
||||
|
||||
worker.finish_startup_feature_cleanup_capture();
|
||||
worker.start_host_ready_feature_tasks();
|
||||
let controller_task = tokio::spawn(controller_loop(
|
||||
worker,
|
||||
method_rx,
|
||||
@@ -4070,6 +4071,147 @@ mod tests {
|
||||
assert!(requests.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PublicationTask {
|
||||
entered: Arc<tokio::sync::Notify>,
|
||||
stopped: Arc<std::sync::atomic::AtomicBool>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::feature::background::FeatureBackgroundTask for PublicationTask {
|
||||
async fn run(
|
||||
&self,
|
||||
_: crate::feature::background::BackgroundTaskContext,
|
||||
_: crate::feature::background::BackgroundTaskCancellation,
|
||||
) -> Result<(), crate::hook::HookError> {
|
||||
self.entered.notify_one();
|
||||
std::future::pending().await
|
||||
}
|
||||
async fn stopped(
|
||||
&self,
|
||||
_: crate::feature::background::BackgroundTaskContext,
|
||||
) -> Result<(), crate::hook::HookError> {
|
||||
self.stopped
|
||||
.store(true, std::sync::atomic::Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl crate::feature::FeatureModule for PublicationTask {
|
||||
fn descriptor(&self) -> crate::feature::FeatureDescriptor {
|
||||
crate::feature::FeatureDescriptor::builtin("publication-task", "Publication task")
|
||||
.with_background_task(
|
||||
crate::feature::BackgroundTaskDeclaration::worker_managed(
|
||||
"observe", "observe",
|
||||
),
|
||||
)
|
||||
}
|
||||
fn install(
|
||||
&self,
|
||||
context: &mut crate::feature::FeatureInstallContext<'_>,
|
||||
) -> Result<(), crate::feature::FeatureInstallError> {
|
||||
let mut spec = crate::feature::background::BackgroundTaskSpec::single_flight(
|
||||
crate::feature::BackgroundTaskDeclaration::worker_managed("observe", "observe"),
|
||||
Duration::from_secs(30),
|
||||
);
|
||||
spec.trigger =
|
||||
crate::feature::background::BackgroundTaskTrigger::HostReady { interval_ms: 1 };
|
||||
context.background_tasks().register(spec, self.clone())
|
||||
}
|
||||
}
|
||||
|
||||
// Real Worker/Store, required Feature installation and Controller
|
||||
// publication. The test task observes the final gate; the model is inert.
|
||||
// The non-canonical branch is shared with restore after head recovery.
|
||||
#[tokio::test]
|
||||
async fn host_ready_tasks_wait_for_start_in_fresh_and_restore_controller_routes() {
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
for (canonical, publish) in [(true, false), (true, true), (false, true)] {
|
||||
let root = TempDir::new().unwrap();
|
||||
let manifest = manifest::WorkerManifest::from_toml(
|
||||
"[worker]\nname = 'publication'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n").unwrap();
|
||||
let engine =
|
||||
Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated(
|
||||
ScriptedClient::new(vec![]),
|
||||
);
|
||||
let store = session_store::CombinedStore::new(
|
||||
session_store::FsStore::new(root.path().join("sessions")).unwrap(),
|
||||
session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(),
|
||||
);
|
||||
let mut worker = Worker::new(
|
||||
manifest,
|
||||
engine,
|
||||
store,
|
||||
crate::worker::WorkerWorkspaceContext::local_filesystem(None),
|
||||
crate::worker::WorkerFilesystemAuthority::None,
|
||||
manifest::Scope::empty(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
worker.set_system_prompt_template(
|
||||
crate::SystemPromptTemplate::parse(
|
||||
"default",
|
||||
crate::PromptCatalogSource::builtins_only(),
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
if !canonical {
|
||||
worker.materialize_durable_session_head().await.unwrap();
|
||||
}
|
||||
let mut initialized =
|
||||
WorkerController::initialize_inner(worker, root.path(), None, canonical)
|
||||
.await
|
||||
.unwrap();
|
||||
let entered = Arc::new(tokio::sync::Notify::new());
|
||||
let stopped = Arc::new(AtomicBool::new(false));
|
||||
let report = initialized.worker.install_features(
|
||||
crate::feature::FeatureRegistryBuilder::new().with_module(PublicationTask {
|
||||
entered: entered.clone(),
|
||||
stopped: stopped.clone(),
|
||||
}),
|
||||
);
|
||||
assert!(!report.has_errors());
|
||||
let publication = initialized.prepare_publication().await.unwrap();
|
||||
assert!(!stopped.load(Ordering::SeqCst));
|
||||
assert!(
|
||||
futures::FutureExt::now_or_never(entered.notified()).is_none(),
|
||||
"preparation cannot start tasks"
|
||||
);
|
||||
if !publish {
|
||||
drop(publication);
|
||||
drop(initialized);
|
||||
assert!(futures::FutureExt::now_or_never(entered.notified()).is_none());
|
||||
assert!(
|
||||
!stopped.load(Ordering::SeqCst),
|
||||
"unstarted task has no on-stop effects"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
let (handle, shutdown, controller) = initialized.start(publication);
|
||||
tokio::time::timeout(Duration::from_secs(2), entered.notified())
|
||||
.await
|
||||
.unwrap();
|
||||
handle
|
||||
.send(Method::Shutdown {
|
||||
command: WorkerCommandEnvelope { command_id: 1 },
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::time::timeout(Duration::from_secs(5), shutdown)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
tokio::time::timeout(Duration::from_secs(5), controller)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
stopped.load(Ordering::SeqCst),
|
||||
"Controller shutdown joins task cleanup"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_initialization_failure_does_not_reclaim_saved_children() {
|
||||
let root = TempDir::new().unwrap();
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -3,9 +3,13 @@
|
||||
//! access: callers bind exact persisted pins and current owner adapters before
|
||||
//! controller publication. Verified bytes and compiled Components are not leases.
|
||||
|
||||
mod service;
|
||||
|
||||
use crate::feature::background::{BackgroundTaskSpec, BackgroundTaskTrigger};
|
||||
use crate::feature::{
|
||||
FeatureDescriptor, FeatureHookPoint, FeatureId, FeatureInstallContext, FeatureInstallError,
|
||||
FeatureModule, FeatureRuntimeKind, HookDeclaration, ToolContribution, ToolDeclaration,
|
||||
BackgroundTaskDeclaration, FeatureDescriptor, FeatureHookPoint, FeatureId,
|
||||
FeatureInstallContext, FeatureInstallError, FeatureModule, FeatureRuntimeKind, HookDeclaration,
|
||||
ToolContribution, ToolDeclaration,
|
||||
};
|
||||
use crate::hook::{
|
||||
Hook, HookError, HookErrorCategory, HookExecutionPolicy, WorkerStopping, WorkerStoppingContext,
|
||||
@@ -39,6 +43,7 @@ pub struct PluginToolPin {
|
||||
pub request: server_api::PluginArtifactDeliveryRequest,
|
||||
pub expectation: ArtifactExpectation,
|
||||
pub tools: Vec<String>,
|
||||
pub services: Vec<String>,
|
||||
pub config: Value,
|
||||
pub grants: Grants,
|
||||
}
|
||||
@@ -96,8 +101,19 @@ impl Drop for CallLease {
|
||||
pub struct PluginToolsFeature {
|
||||
tools: Vec<(ToolMeta, Arc<PluginTool>)>,
|
||||
calls: Arc<Mutex<Calls>>,
|
||||
services: Vec<Arc<service::ServiceDriver>>,
|
||||
}
|
||||
impl PluginToolsFeature {
|
||||
/// Called only after the task registry has joined every driver. Retained
|
||||
/// Service proofs keep shutdown ownership here until release is reconciled;
|
||||
/// this never restarts a task or invokes a lifecycle export again.
|
||||
pub(crate) async fn reconcile_service_shutdown(&self) -> Result<(), HookError> {
|
||||
for service in &self.services {
|
||||
service.reconcile_release().await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn prepare(
|
||||
pins: Vec<PluginToolPin>,
|
||||
source: &dyn ArtifactSource,
|
||||
@@ -116,6 +132,7 @@ impl PluginToolsFeature {
|
||||
let mut names = BTreeSet::new();
|
||||
let mut installations = BTreeSet::new();
|
||||
let mut tools = Vec::new();
|
||||
let mut services = Vec::new();
|
||||
for pin in pins {
|
||||
let descriptor = pin.expectation.descriptor();
|
||||
if pin.workspace_id.is_empty()
|
||||
@@ -138,6 +155,40 @@ impl PluginToolsFeature {
|
||||
.resolve(source, &pin.workspace_id, &pin.request, &pin.expectation)
|
||||
.await
|
||||
.map_err(|_| Delivery)?;
|
||||
if !pin.services.is_empty() {
|
||||
if !pin.tools.is_empty() || !pin.grants.ticket_events || pin.services.len() > 32 {
|
||||
return Err(InvalidSnapshot);
|
||||
}
|
||||
let compiled = Arc::new(
|
||||
executor
|
||||
.compile_service(
|
||||
artifact.bytes(),
|
||||
&descriptor.package_digest,
|
||||
&descriptor.artifact_digest,
|
||||
u64::from(descriptor.artifact_size),
|
||||
)
|
||||
.map_err(|_| Component)?,
|
||||
);
|
||||
for name in &pin.services {
|
||||
if !compiled
|
||||
.services()
|
||||
.iter()
|
||||
.any(|service| &service.name == name)
|
||||
|| !names.insert(format!("{}__{}", pin.installation_id, name))
|
||||
{
|
||||
return Err(InvalidSnapshot);
|
||||
}
|
||||
services.push(Arc::new(service::ServiceDriver::new(
|
||||
executor.clone(),
|
||||
compiled.clone(),
|
||||
owner.clone(),
|
||||
&pin,
|
||||
name.clone(),
|
||||
config_json.clone(),
|
||||
)));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let compiled = Arc::new(
|
||||
executor
|
||||
.compile(
|
||||
@@ -189,7 +240,16 @@ impl PluginToolsFeature {
|
||||
tools.push((meta, tool));
|
||||
}
|
||||
}
|
||||
Ok(Self { tools, calls })
|
||||
// All immutable contributions have been validated before any lifecycle
|
||||
// export. Start is logging-only; failed preparation leaves no consumer.
|
||||
for service in &services {
|
||||
service.prepare_start().await.map_err(|_| Component)?;
|
||||
}
|
||||
Ok(Self {
|
||||
tools,
|
||||
calls,
|
||||
services,
|
||||
})
|
||||
}
|
||||
}
|
||||
impl FeatureModule for PluginToolsFeature {
|
||||
@@ -205,6 +265,14 @@ impl FeatureModule for PluginToolsFeature {
|
||||
.tools
|
||||
.push(ToolDeclaration::new(&meta.name, &meta.description));
|
||||
}
|
||||
for service in &self.services {
|
||||
descriptor
|
||||
.background_tasks
|
||||
.push(BackgroundTaskDeclaration::worker_managed(
|
||||
service.task_name(),
|
||||
"Commit-event Service pump",
|
||||
));
|
||||
}
|
||||
descriptor
|
||||
}
|
||||
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
|
||||
@@ -217,6 +285,17 @@ impl FeatureModule for PluginToolsFeature {
|
||||
.tools()
|
||||
.register(ToolContribution::new(name, definition))?;
|
||||
}
|
||||
for service in &self.services {
|
||||
let mut spec = BackgroundTaskSpec::single_flight(
|
||||
BackgroundTaskDeclaration::worker_managed(
|
||||
service.task_name(),
|
||||
"Commit-event Service pump",
|
||||
),
|
||||
std::time::Duration::from_secs(30),
|
||||
);
|
||||
spec.trigger = BackgroundTaskTrigger::HostReady { interval_ms: 100 };
|
||||
context.background_tasks().register(spec, service.clone())?;
|
||||
}
|
||||
context.hooks().add_worker_stopping(
|
||||
"stop",
|
||||
HookExecutionPolicy::fail_closed(),
|
||||
|
||||
@@ -0,0 +1,629 @@
|
||||
//! Bounded outbox pumps run in the existing Feature task registry. Consumer and
|
||||
//! effect truth stays with the owner; this module retains only an active lease.
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
use super::*;
|
||||
use crate::feature::background::{
|
||||
BackgroundTaskCancellation, BackgroundTaskContext, FeatureBackgroundTask,
|
||||
};
|
||||
use server_api::*;
|
||||
use std::{any::Any, time::Duration};
|
||||
use yoi_plugin_runtime::{
|
||||
CompiledService, ServiceOutput,
|
||||
host::{errors, http, logging},
|
||||
};
|
||||
|
||||
pub(super) struct ServiceDriver {
|
||||
pub executor: Arc<Executor>,
|
||||
pub compiled: Arc<CompiledService>,
|
||||
pub owner: Arc<dyn PluginHostOwner>,
|
||||
pub workspace: String,
|
||||
pub installation: String,
|
||||
pub plan: String,
|
||||
pub subject: String,
|
||||
pub grants: Grants,
|
||||
pub name: String,
|
||||
pub config: String,
|
||||
lease: Mutex<Option<PluginServiceLease>>,
|
||||
stop_started: Mutex<Option<PluginServiceLease>>,
|
||||
// A selector for read-only recovery observation, never a mirrored pause state.
|
||||
parked_consumer: Mutex<Option<String>>,
|
||||
// Unconfirmed control commit, not permission to replay an external effect.
|
||||
unconfirmed_indeterminate: Mutex<Option<String>>,
|
||||
prepared_start: std::sync::atomic::AtomicBool,
|
||||
}
|
||||
impl ServiceDriver {
|
||||
pub fn new(
|
||||
executor: Arc<Executor>,
|
||||
compiled: Arc<CompiledService>,
|
||||
owner: Arc<dyn PluginHostOwner>,
|
||||
pin: &PluginToolPin,
|
||||
name: String,
|
||||
config: String,
|
||||
) -> Self {
|
||||
Self {
|
||||
executor,
|
||||
compiled,
|
||||
owner,
|
||||
workspace: pin.workspace_id.clone(),
|
||||
installation: pin.installation_id.clone(),
|
||||
plan: pin.plan_id.clone(),
|
||||
subject: pin.subject.clone(),
|
||||
grants: pin.grants.clone(),
|
||||
name,
|
||||
config,
|
||||
lease: Mutex::new(None),
|
||||
stop_started: Mutex::new(None),
|
||||
parked_consumer: Mutex::new(None),
|
||||
unconfirmed_indeterminate: Mutex::new(None),
|
||||
prepared_start: std::sync::atomic::AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
fn lifecycle_host(&self) -> Arc<LifecycleHost> {
|
||||
Arc::new(LifecycleHost {
|
||||
owner: self.owner.clone(),
|
||||
plan: self.plan.clone(),
|
||||
installation: self.installation.clone(),
|
||||
name: self.name.clone(),
|
||||
})
|
||||
}
|
||||
pub async fn prepare_start(&self) -> Result<(), HookError> {
|
||||
let output = self
|
||||
.executor
|
||||
.service_start(
|
||||
&self.compiled,
|
||||
self.context(
|
||||
format!("{}:{}:start", self.installation, self.name),
|
||||
CancellationToken::new(),
|
||||
),
|
||||
self.lifecycle_host(),
|
||||
&self.name,
|
||||
&self.config,
|
||||
)
|
||||
.await;
|
||||
if !matches!(output.outcome, Ok(ServiceOutput::Started)) {
|
||||
return Err(dependency());
|
||||
}
|
||||
self.prepared_start
|
||||
.store(true, std::sync::atomic::Ordering::Release);
|
||||
Ok(())
|
||||
}
|
||||
pub fn task_name(&self) -> String {
|
||||
format!("service:{}:{}", self.installation, self.name)
|
||||
}
|
||||
fn context(&self, invocation_id: String, cancellation: CancellationToken) -> InvocationContext {
|
||||
InvocationContext {
|
||||
workspace: self.workspace.clone(),
|
||||
installation: self.installation.clone(),
|
||||
plan: self.plan.clone(),
|
||||
subject: self.subject.clone(),
|
||||
invocation_id,
|
||||
artifact: self.compiled.identity().clone(),
|
||||
tool_name: self.name.clone(),
|
||||
grants: self.grants.clone(),
|
||||
deadline: tokio::time::Instant::now()
|
||||
+ self
|
||||
.executor
|
||||
.limits()
|
||||
.wall_time
|
||||
.min(Duration::from_secs(20)),
|
||||
cancellation,
|
||||
}
|
||||
}
|
||||
async fn request(
|
||||
&self,
|
||||
request: PluginServiceRequest,
|
||||
) -> Result<PluginServiceResponse, HookError> {
|
||||
request_service(self.owner.as_ref(), request)
|
||||
.await
|
||||
.map_err(|_| dependency())
|
||||
}
|
||||
pub(super) async fn reconcile_release(&self) -> Result<(), HookError> {
|
||||
let Some(lease) = self.lease.lock().expect("Service lease lock").clone() else {
|
||||
return Ok(());
|
||||
};
|
||||
let released = match request_service(
|
||||
self.owner.as_ref(),
|
||||
PluginServiceRequest::Release {
|
||||
lease: lease.clone(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(PluginServiceResponse::Released) => true,
|
||||
// Release authenticates this consumer's Worker before its exact-token
|
||||
// CAS. Conflict confirms this proof no longer owns the lease; do not
|
||||
// mutate or wait for the successor as though it were our resource.
|
||||
Err(error) if error.code == PluginHostErrorCode::Conflict => true,
|
||||
_ => false,
|
||||
};
|
||||
if !released {
|
||||
return Err(dependency());
|
||||
}
|
||||
let mut retained = self.lease.lock().expect("Service lease lock");
|
||||
if retained.as_ref() == Some(&lease) {
|
||||
*retained = None;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
async fn acquire(&self) -> Result<PluginServiceLease, HookError> {
|
||||
if let Some(lease) = self.lease.lock().expect("Service lease lock").clone() {
|
||||
if self
|
||||
.stop_started
|
||||
.lock()
|
||||
.expect("Service stop lock")
|
||||
.as_ref()
|
||||
== Some(&lease)
|
||||
{
|
||||
return Err(dependency());
|
||||
}
|
||||
return Ok(lease);
|
||||
}
|
||||
let PluginServiceResponse::Acquired { lease } = self
|
||||
.request(PluginServiceRequest::Acquire {
|
||||
plan_id: self.plan.clone(),
|
||||
installation_id: self.installation.clone(),
|
||||
service_name: self.name.clone(),
|
||||
})
|
||||
.await?
|
||||
else {
|
||||
return Err(dependency());
|
||||
};
|
||||
*self.lease.lock().expect("Service lease lock") = Some(lease.clone());
|
||||
if !self
|
||||
.prepared_start
|
||||
.swap(false, std::sync::atomic::Ordering::AcqRel)
|
||||
{
|
||||
self.prepare_start().await?;
|
||||
self.prepared_start
|
||||
.store(false, std::sync::atomic::Ordering::Release);
|
||||
}
|
||||
Ok(lease)
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl FeatureBackgroundTask for Arc<ServiceDriver> {
|
||||
async fn run(
|
||||
&self,
|
||||
context: BackgroundTaskContext,
|
||||
cancellation: BackgroundTaskCancellation,
|
||||
) -> Result<(), HookError> {
|
||||
context.generation_fence.ensure_current()?;
|
||||
if cancellation.is_cancelled() {
|
||||
return Ok(());
|
||||
}
|
||||
let parked = self
|
||||
.parked_consumer
|
||||
.lock()
|
||||
.expect("Service observer lock")
|
||||
.clone();
|
||||
if let Some(consumer_id) = parked {
|
||||
let diagnostics = match self
|
||||
.request(PluginServiceRequest::Diagnostics { consumer_id })
|
||||
.await
|
||||
{
|
||||
Ok(PluginServiceResponse::Diagnostics { diagnostics }) => diagnostics,
|
||||
// Observation alone neither retries a guest nor overrides live
|
||||
// revocation. Keep the parked pump inert until its owner returns.
|
||||
_ => return Ok(()),
|
||||
};
|
||||
if !diagnostics.lease_active {
|
||||
// A fresh owner observation confirms any retained proof is now
|
||||
// fenced (release may have committed before its response was lost).
|
||||
// Never reuse that proof after expiry or audited recovery.
|
||||
*self.lease.lock().expect("Service lease lock") = None;
|
||||
}
|
||||
if diagnostics.pause_reason.is_some() || diagnostics.lease_active {
|
||||
return Ok(());
|
||||
}
|
||||
let unconfirmed = self
|
||||
.unconfirmed_indeterminate
|
||||
.lock()
|
||||
.expect("Service completion lock")
|
||||
.clone();
|
||||
if let Some(event_id) = unconfirmed {
|
||||
if diagnostics.current_event_id.as_ref() == Some(&event_id) {
|
||||
// No pause is not evidence that an uncertain completion did
|
||||
// not execute. Reconcile the same control intent under a new
|
||||
// owner lease, without polling or invoking the guest again.
|
||||
if let Ok(lease) = self.acquire().await {
|
||||
if matches!(
|
||||
self.request(PluginServiceRequest::Complete {
|
||||
lease,
|
||||
event_id,
|
||||
result: PluginServiceEventResult::Indeterminate,
|
||||
})
|
||||
.await,
|
||||
Ok(PluginServiceResponse::Completed)
|
||||
) {
|
||||
*self
|
||||
.unconfirmed_indeterminate
|
||||
.lock()
|
||||
.expect("Service completion lock") = None;
|
||||
}
|
||||
if self.stopped(context).await.is_err() {
|
||||
tracing::warn!("Plugin Service stop requires owner reconciliation");
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
// The owner has already advanced this head (including an audited
|
||||
// skip). Never recreate it or change its stable event identity.
|
||||
*self
|
||||
.unconfirmed_indeterminate
|
||||
.lock()
|
||||
.expect("Service completion lock") = None;
|
||||
}
|
||||
// Only an audited owner recovery clears a domain pause. Observing
|
||||
// that decision wakes this consumer alone, not the entire registry.
|
||||
*self.parked_consumer.lock().expect("Service observer lock") = None;
|
||||
}
|
||||
if self.lease.lock().expect("Service lease lock").is_none() {
|
||||
self.acquire().await?;
|
||||
return Ok(());
|
||||
}
|
||||
let lease = self.acquire().await?;
|
||||
let response = match self
|
||||
.request(PluginServiceRequest::Poll {
|
||||
lease: lease.clone(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(response) => response,
|
||||
Err(_) => {
|
||||
*self.parked_consumer.lock().expect("Service observer lock") =
|
||||
Some(lease.consumer_id.clone());
|
||||
if self.stopped(context).await.is_err() {
|
||||
tracing::warn!("Plugin Service stop requires owner reconciliation");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
match response {
|
||||
PluginServiceResponse::Idle => Ok(()),
|
||||
PluginServiceResponse::Diagnostics { diagnostics } => {
|
||||
// Retain only the consumer selector while paused. Stop the guest
|
||||
// lifetime and release its lease so management can recover it.
|
||||
// The registry retains a bounded read-only observation pump.
|
||||
*self.parked_consumer.lock().expect("Service observer lock") =
|
||||
Some(diagnostics.consumer_id);
|
||||
// Revoked logging or a failed stop export must not hide the
|
||||
// audited recovery observer. Release is attempted independently.
|
||||
if self.stopped(context).await.is_err() {
|
||||
tracing::warn!("Plugin Service stop requires owner reconciliation");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
PluginServiceResponse::Delivery {
|
||||
envelope,
|
||||
invocation_id,
|
||||
expires_at_ms,
|
||||
} => {
|
||||
context.generation_fence.ensure_current()?;
|
||||
if cancellation.is_cancelled() {
|
||||
return Ok(());
|
||||
}
|
||||
let consumer = PluginConsumer::Consumer {
|
||||
consumer_id: lease.consumer_id.clone(),
|
||||
event_id: envelope.event_id.clone(),
|
||||
export_name: self.name.clone(),
|
||||
};
|
||||
let owner = Arc::new(DeliveryOwner {
|
||||
owner: self.owner.clone(),
|
||||
lease: lease.clone(),
|
||||
consumer: consumer.clone(),
|
||||
invocation_id: invocation_id.clone(),
|
||||
expires_at_ms,
|
||||
});
|
||||
let token = CancellationToken::new();
|
||||
let host = Arc::new(
|
||||
OwnerHost::begin(owner, self.context(invocation_id, token.clone()), consumer)
|
||||
.await
|
||||
.map_err(|_| dependency())?,
|
||||
);
|
||||
let call = self.executor.service_event(
|
||||
&self.compiled,
|
||||
host.context().clone(),
|
||||
host.clone(),
|
||||
&self.name,
|
||||
&envelope,
|
||||
&self.config,
|
||||
);
|
||||
tokio::pin!(call);
|
||||
let output = tokio::select! {
|
||||
result = &mut call => result,
|
||||
_ = cancellation.cancelled() => { token.cancel(); call.await },
|
||||
};
|
||||
let result = if matches!(output.effect, errors::EffectObservation::Indeterminate) {
|
||||
PluginServiceEventResult::Indeterminate
|
||||
} else {
|
||||
match output.outcome {
|
||||
Ok(ServiceOutput::Event(result)) => result,
|
||||
_ => PluginServiceEventResult::RetryAfter { delay_ms: 1_000 },
|
||||
}
|
||||
};
|
||||
context.generation_fence.ensure_current()?;
|
||||
// Guest return is not ack. Only this fenced owner transaction
|
||||
// commits a cursor; transport uncertainty leaves it for recovery.
|
||||
let indeterminate = matches!(result, PluginServiceEventResult::Indeterminate);
|
||||
if !matches!(
|
||||
self.request(PluginServiceRequest::Complete {
|
||||
lease: lease.clone(),
|
||||
event_id: envelope.event_id.clone(),
|
||||
result
|
||||
})
|
||||
.await,
|
||||
Ok(PluginServiceResponse::Completed)
|
||||
) {
|
||||
// The cursor/pause transaction may have committed even if its
|
||||
// reply was lost. Keep the read-only observer alive instead
|
||||
// of permanently deactivating this HostReady registration.
|
||||
if indeterminate {
|
||||
*self
|
||||
.unconfirmed_indeterminate
|
||||
.lock()
|
||||
.expect("Service completion lock") = Some(envelope.event_id.clone());
|
||||
}
|
||||
*self.parked_consumer.lock().expect("Service observer lock") =
|
||||
Some(lease.consumer_id.clone());
|
||||
if self.stopped(context).await.is_err() {
|
||||
tracing::warn!("Plugin Service stop requires owner reconciliation");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
_ => Err(dependency()),
|
||||
}
|
||||
}
|
||||
async fn stopped(&self, _: BackgroundTaskContext) -> Result<(), HookError> {
|
||||
// Inline pause cleanup is cancellable. Retain the exact proof until the
|
||||
// owner confirms release, so protected registry cleanup can finish it if
|
||||
// the run future is dropped during on-stop or transport uncertainty.
|
||||
let lease = self.lease.lock().expect("Service lease lock").clone();
|
||||
let Some(lease) = lease else {
|
||||
return Ok(());
|
||||
};
|
||||
let attempt_stop = {
|
||||
let mut started = self.stop_started.lock().expect("Service stop lock");
|
||||
if started.as_ref() == Some(&lease) {
|
||||
false
|
||||
} else {
|
||||
*started = Some(lease.clone());
|
||||
true
|
||||
}
|
||||
};
|
||||
let stop_failed = if attempt_stop {
|
||||
let output = self
|
||||
.executor
|
||||
.service_stop(
|
||||
&self.compiled,
|
||||
self.context(
|
||||
format!("{}:stop", lease.consumer_id),
|
||||
CancellationToken::new(),
|
||||
),
|
||||
self.lifecycle_host(),
|
||||
&self.name,
|
||||
"Host task stopped",
|
||||
)
|
||||
.await;
|
||||
!matches!(output.outcome, Ok(ServiceOutput::Stopped))
|
||||
} else {
|
||||
false
|
||||
};
|
||||
// Marking the attempt before the await prevents an interrupted inline
|
||||
// cleanup from rerunning the stop export in protected registry cleanup.
|
||||
// A failed/revoked/timed-out export never prevents fencing/release.
|
||||
self.reconcile_release().await?;
|
||||
if stop_failed {
|
||||
return Err(dependency());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
fn dependency() -> HookError {
|
||||
HookError::new(
|
||||
HookErrorCategory::Dependency,
|
||||
"Plugin Service requires owner reconciliation",
|
||||
)
|
||||
}
|
||||
fn unavailable() -> PluginHostError {
|
||||
PluginHostError {
|
||||
code: PluginHostErrorCode::Unavailable,
|
||||
effect: PluginEffectObservation::NotDispatched,
|
||||
diagnostic_id: String::new(),
|
||||
}
|
||||
}
|
||||
async fn request_service(
|
||||
owner: &dyn PluginHostOwner,
|
||||
request: PluginServiceRequest,
|
||||
) -> Result<PluginServiceResponse, PluginHostError> {
|
||||
// Event imports already run under the executor's total invocation deadline.
|
||||
// A shorter transport deadline could abandon a still-running admitted effect
|
||||
// before the owner's HTTP/receipt budgets expire. Control requests remain
|
||||
// separately bounded because they run outside a guest invocation.
|
||||
let budget = if matches!(&request, PluginServiceRequest::Host { .. }) {
|
||||
Duration::from_secs(30)
|
||||
} else {
|
||||
Duration::from_secs(2)
|
||||
};
|
||||
let response = tokio::time::timeout(budget, owner.call(PluginHostRequest::Service { request }))
|
||||
.await
|
||||
.map_err(|_| unavailable())??;
|
||||
if let PluginHostResponse::Service { response } = response {
|
||||
Ok(response)
|
||||
} else {
|
||||
Err(unavailable())
|
||||
}
|
||||
}
|
||||
|
||||
/// Adapts an already owner-issued delivery to the existing receipt-aware Host.
|
||||
/// Begin here only binds that receipt; it cannot create a logical invocation.
|
||||
struct DeliveryOwner {
|
||||
owner: Arc<dyn PluginHostOwner>,
|
||||
lease: PluginServiceLease,
|
||||
consumer: PluginConsumer,
|
||||
invocation_id: String,
|
||||
expires_at_ms: i64,
|
||||
}
|
||||
#[async_trait]
|
||||
impl PluginHostOwner for DeliveryOwner {
|
||||
async fn call(
|
||||
&self,
|
||||
request: PluginHostRequest,
|
||||
) -> Result<PluginHostResponse, PluginHostError> {
|
||||
if let PluginHostRequest::Begin { consumer, .. } = &request {
|
||||
if consumer != &self.consumer {
|
||||
return Err(unavailable());
|
||||
}
|
||||
return Ok(PluginHostResponse::Invocation {
|
||||
invocation_id: self.invocation_id.clone(),
|
||||
expires_at_ms: self.expires_at_ms,
|
||||
});
|
||||
}
|
||||
let may_dispatch = matches!(
|
||||
request,
|
||||
PluginHostRequest::Send { .. } | PluginHostRequest::DataMutate { .. }
|
||||
);
|
||||
let response = request_service(
|
||||
self.owner.as_ref(),
|
||||
PluginServiceRequest::Host {
|
||||
lease: self.lease.clone(),
|
||||
request: Box::new(request.try_into().map_err(|_| unavailable())?),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
match response {
|
||||
Ok(PluginServiceResponse::Host { response }) => Ok((*response).into()),
|
||||
Ok(_) => Err(unavailable()),
|
||||
Err(mut error) => {
|
||||
if may_dispatch && matches!(error.code, PluginHostErrorCode::Unavailable) {
|
||||
error.code = PluginHostErrorCode::Indeterminate;
|
||||
error.effect = PluginEffectObservation::Indeterminate;
|
||||
}
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Lifecycle exports have no object/connection/data resources. Bounded
|
||||
/// logging follows a current owner check, never an event receipt or guest grant.
|
||||
struct LifecycleHost {
|
||||
owner: Arc<dyn PluginHostOwner>,
|
||||
plan: String,
|
||||
installation: String,
|
||||
name: String,
|
||||
}
|
||||
fn denied_host() -> errors::HostError {
|
||||
errors::HostError {
|
||||
code: errors::Code::Denied,
|
||||
message: "lifecycle operation denied".into(),
|
||||
diagnostic_id: String::new(),
|
||||
effect: errors::EffectObservation::NotDispatched,
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl Host for LifecycleHost {
|
||||
async fn authorize(
|
||||
&self,
|
||||
_: &InvocationContext,
|
||||
operation: Operation<'_>,
|
||||
) -> Result<(), errors::HostError> {
|
||||
if !matches!(
|
||||
operation,
|
||||
Operation::Admit | Operation::Publish | Operation::Log
|
||||
) {
|
||||
return Err(denied_host());
|
||||
}
|
||||
if matches!(operation, Operation::Log) {
|
||||
return Ok(());
|
||||
}
|
||||
match request_service(
|
||||
self.owner.as_ref(),
|
||||
PluginServiceRequest::Prepare {
|
||||
plan_id: self.plan.clone(),
|
||||
installation_id: self.installation.clone(),
|
||||
service_name: self.name.clone(),
|
||||
operation: match operation {
|
||||
Operation::Admit => PluginServiceLifecycleOperation::Admit,
|
||||
Operation::Publish => PluginServiceLifecycleOperation::Publish,
|
||||
_ => return Err(denied_host()),
|
||||
},
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(PluginServiceResponse::Prepared) => Ok(()),
|
||||
_ => Err(denied_host()),
|
||||
}
|
||||
}
|
||||
async fn open(
|
||||
&self,
|
||||
_: &InvocationContext,
|
||||
_: &str,
|
||||
) -> Result<Arc<dyn Any + Send + Sync>, errors::HostError> {
|
||||
Err(denied_host())
|
||||
}
|
||||
async fn send(
|
||||
&self,
|
||||
_: &InvocationContext,
|
||||
_: &(dyn Any + Send + Sync),
|
||||
_: http::Request,
|
||||
) -> Result<http::Response, errors::HostError> {
|
||||
Err(denied_host())
|
||||
}
|
||||
async fn log(
|
||||
&self,
|
||||
context: &InvocationContext,
|
||||
entry: logging::Entry,
|
||||
) -> Result<(), errors::HostError> {
|
||||
self.authorize(context, Operation::Log).await?;
|
||||
// This request performs live plugin.log authorization without creating
|
||||
// an event receipt or allowing lifecycle access to effect APIs.
|
||||
let entry = PluginLogEntry {
|
||||
level: match entry.level {
|
||||
logging::Level::Trace => PluginLogLevel::Trace,
|
||||
logging::Level::Debug => PluginLogLevel::Debug,
|
||||
logging::Level::Info => PluginLogLevel::Info,
|
||||
logging::Level::Warn => PluginLogLevel::Warn,
|
||||
logging::Level::Error => PluginLogLevel::Error,
|
||||
},
|
||||
event_code: entry.event_code,
|
||||
message: entry.message,
|
||||
fields: entry
|
||||
.fields
|
||||
.into_iter()
|
||||
.map(|field| {
|
||||
Ok(PluginLogField {
|
||||
name: field.name,
|
||||
value: match field.value {
|
||||
logging::Scalar::Text(v) => PluginLogScalar::Text(v),
|
||||
logging::Scalar::Boolean(v) => PluginLogScalar::Boolean(v),
|
||||
logging::Scalar::Signed(v) => PluginLogScalar::Signed(v),
|
||||
logging::Scalar::Unsigned(v) => PluginLogScalar::Unsigned(v),
|
||||
logging::Scalar::Number(v) if v.is_finite() => {
|
||||
PluginLogScalar::Number(v)
|
||||
}
|
||||
logging::Scalar::Number(_) => return Err(denied_host()),
|
||||
},
|
||||
})
|
||||
})
|
||||
.collect::<Result<_, errors::HostError>>()?,
|
||||
};
|
||||
match request_service(
|
||||
self.owner.as_ref(),
|
||||
PluginServiceRequest::LifecycleLog {
|
||||
plan_id: self.plan.clone(),
|
||||
installation_id: self.installation.clone(),
|
||||
service_name: self.name.clone(),
|
||||
entry,
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(PluginServiceResponse::Logged) => Ok(()),
|
||||
_ => Err(denied_host()),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,672 @@
|
||||
//! Real ServiceDriver/FeatureBackgroundTask + canonical authored Component tests.
|
||||
//! Only the PluginHostOwner is scripted; its pause/recovery state is not evidence
|
||||
//! of durable owner transactions (covered by the owner integration suites).
|
||||
//! Run: cargo test -p worker --lib plugin::service::tests::
|
||||
|
||||
use super::*;
|
||||
use crate::feature::background::FeatureBackgroundTaskRegistryBuilder;
|
||||
use crate::hook::HookInvocationContext;
|
||||
use manifest::plugin::{self, PluginPackageLimits, PluginSourceKind};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tokio::sync::{Semaphore, oneshot};
|
||||
|
||||
const SERVICE: &str = "ticket_events";
|
||||
const MANIFEST: &str = include_str!("../../../../plugin-runtime/tests/service-guest/plugin.toml");
|
||||
const COMPONENT: &[u8] =
|
||||
include_bytes!("../../../../workspace-server/tests/fixtures/ticket-service-v2.component.wasm");
|
||||
|
||||
struct Fixture {
|
||||
executor: Arc<Executor>,
|
||||
compiled: Arc<CompiledService>,
|
||||
}
|
||||
|
||||
impl Fixture {
|
||||
fn new() -> Self {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
std::fs::write(directory.path().join("plugin.toml"), MANIFEST).unwrap();
|
||||
std::fs::write(directory.path().join("plugin.component.wasm"), COMPONENT).unwrap();
|
||||
let limits = PluginPackageLimits::default();
|
||||
let package =
|
||||
plugin::read_plugin_directory(directory.path(), PluginSourceKind::Explicit, &limits)
|
||||
.unwrap();
|
||||
let artifact = plugin::pack_plugin_package(&package, &limits).unwrap();
|
||||
let executor = Arc::new(Executor::new(Limits::default()).unwrap());
|
||||
let compiled = Arc::new(
|
||||
executor
|
||||
.compile_service(
|
||||
&artifact.bytes,
|
||||
&artifact.package_digest,
|
||||
&artifact.artifact_digest,
|
||||
artifact.artifact_size,
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
Self { executor, compiled }
|
||||
}
|
||||
|
||||
fn driver(&self, owner: Arc<Owner>) -> Arc<ServiceDriver> {
|
||||
let namespace = ("notes".into(), "note".into(), "1".into());
|
||||
Arc::new(ServiceDriver {
|
||||
executor: self.executor.clone(),
|
||||
compiled: self.compiled.clone(),
|
||||
workspace: "workspace".into(),
|
||||
installation: owner.installation.clone(),
|
||||
plan: "plan".into(),
|
||||
subject: "worker".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(),
|
||||
},
|
||||
owner,
|
||||
name: SERVICE.into(),
|
||||
config: r#"{"mode":"normal"}"#.into(),
|
||||
lease: Mutex::new(None),
|
||||
stop_started: Mutex::new(None),
|
||||
parked_consumer: Mutex::new(None),
|
||||
unconfirmed_indeterminate: Mutex::new(None),
|
||||
prepared_start: AtomicBool::new(false),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Obtain a real registry-generated generation fence rather than bypassing the
|
||||
// FeatureBackgroundTask boundary or adding a production-only context constructor.
|
||||
struct CaptureContext(Mutex<Option<oneshot::Sender<BackgroundTaskContext>>>);
|
||||
|
||||
#[async_trait]
|
||||
impl FeatureBackgroundTask for CaptureContext {
|
||||
async fn run(
|
||||
&self,
|
||||
context: BackgroundTaskContext,
|
||||
_: BackgroundTaskCancellation,
|
||||
) -> Result<(), HookError> {
|
||||
self.0
|
||||
.lock()
|
||||
.unwrap()
|
||||
.take()
|
||||
.unwrap()
|
||||
.send(context)
|
||||
.unwrap();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn context() -> BackgroundTaskContext {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
let feature = FeatureId::builtin("service-driver-test");
|
||||
let mut builder = FeatureBackgroundTaskRegistryBuilder::default();
|
||||
builder
|
||||
.register(
|
||||
feature.clone(),
|
||||
BackgroundTaskSpec::single_flight(
|
||||
BackgroundTaskDeclaration::worker_managed("capture", "capture test context"),
|
||||
Duration::from_secs(5),
|
||||
),
|
||||
CaptureContext(Mutex::new(Some(sender))),
|
||||
)
|
||||
.unwrap();
|
||||
let registry = builder.build();
|
||||
registry
|
||||
.start(&feature, "capture", HookInvocationContext::default())
|
||||
.unwrap();
|
||||
let context = receiver.await.unwrap();
|
||||
registry.shutdown().await.unwrap();
|
||||
context.generation_fence.ensure_current().unwrap();
|
||||
context
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum CompleteResponse {
|
||||
Lost,
|
||||
Unexpected,
|
||||
ConfirmIndeterminate,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct State {
|
||||
requests: Vec<PluginServiceRequest>,
|
||||
lease: Option<PluginServiceLease>,
|
||||
acquisitions: usize,
|
||||
pause: Option<PluginServicePauseReason>,
|
||||
}
|
||||
|
||||
struct Owner {
|
||||
installation: String,
|
||||
state: Mutex<State>,
|
||||
complete_response: CompleteResponse,
|
||||
gate_log: AtomicBool,
|
||||
deny_log: AtomicBool,
|
||||
deny_release: AtomicBool,
|
||||
deny_publish: AtomicBool,
|
||||
log_entered: Semaphore,
|
||||
}
|
||||
|
||||
impl Owner {
|
||||
fn new(installation: &str, complete_response: CompleteResponse) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
installation: installation.into(),
|
||||
state: Mutex::new(State::default()),
|
||||
complete_response,
|
||||
gate_log: AtomicBool::new(false),
|
||||
deny_log: AtomicBool::new(false),
|
||||
deny_release: AtomicBool::new(false),
|
||||
deny_publish: AtomicBool::new(false),
|
||||
log_entered: Semaphore::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
fn consumer(&self) -> String {
|
||||
format!("consumer:{}", self.installation)
|
||||
}
|
||||
|
||||
fn requests(&self) -> Vec<PluginServiceRequest> {
|
||||
self.state.lock().unwrap().requests.clone()
|
||||
}
|
||||
|
||||
fn diagnostics(&self) -> PluginServiceDiagnostics {
|
||||
let state = self.state.lock().unwrap();
|
||||
PluginServiceDiagnostics {
|
||||
consumer_id: self.consumer(),
|
||||
pause_reason: state.pause,
|
||||
queued: 1,
|
||||
attempt: 1,
|
||||
current_event_id: Some("event".into()),
|
||||
lease_active: state.lease.is_some(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl PluginHostOwner for Owner {
|
||||
async fn call(
|
||||
&self,
|
||||
request: PluginHostRequest,
|
||||
) -> Result<PluginHostResponse, PluginHostError> {
|
||||
let PluginHostRequest::Service { request } = request else {
|
||||
panic!("Service driver must use only the leased/lifecycle channel: {request:?}");
|
||||
};
|
||||
self.state.lock().unwrap().requests.push(request.clone());
|
||||
let response = match request {
|
||||
PluginServiceRequest::Prepare {
|
||||
plan_id,
|
||||
installation_id,
|
||||
service_name,
|
||||
operation,
|
||||
} => {
|
||||
assert_eq!(plan_id, "plan");
|
||||
assert_eq!(installation_id, self.installation);
|
||||
assert_eq!(service_name, SERVICE);
|
||||
if operation == PluginServiceLifecycleOperation::Publish
|
||||
&& self.deny_publish.load(Ordering::SeqCst)
|
||||
{
|
||||
return Err(unavailable());
|
||||
}
|
||||
PluginServiceResponse::Prepared
|
||||
}
|
||||
PluginServiceRequest::LifecycleLog { .. } => {
|
||||
if self.gate_log.load(Ordering::SeqCst) {
|
||||
self.log_entered.add_permits(1);
|
||||
return std::future::pending().await;
|
||||
}
|
||||
if self.deny_log.load(Ordering::SeqCst) {
|
||||
return Err(unavailable());
|
||||
}
|
||||
PluginServiceResponse::Logged
|
||||
}
|
||||
PluginServiceRequest::Acquire {
|
||||
plan_id,
|
||||
installation_id,
|
||||
service_name,
|
||||
} => {
|
||||
assert_eq!(plan_id, "plan");
|
||||
assert_eq!(installation_id, self.installation);
|
||||
assert_eq!(service_name, SERVICE);
|
||||
let mut state = self.state.lock().unwrap();
|
||||
assert!(state.pause.is_none(), "driver cannot clear the owner pause");
|
||||
assert!(state.lease.is_none(), "driver must await inactive lease");
|
||||
state.acquisitions += 1;
|
||||
let lease = PluginServiceLease {
|
||||
consumer_id: self.consumer(),
|
||||
lease_token: format!("lease-{}", state.acquisitions),
|
||||
};
|
||||
state.lease = Some(lease.clone());
|
||||
PluginServiceResponse::Acquired { lease }
|
||||
}
|
||||
PluginServiceRequest::Poll { lease } => {
|
||||
assert_eq!(self.state.lock().unwrap().lease.as_ref(), Some(&lease));
|
||||
if self.state.lock().unwrap().pause.is_some() {
|
||||
PluginServiceResponse::Diagnostics {
|
||||
diagnostics: self.diagnostics(),
|
||||
}
|
||||
} else {
|
||||
PluginServiceResponse::Delivery {
|
||||
envelope: 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,
|
||||
},
|
||||
invocation_id: "invocation".into(),
|
||||
expires_at_ms: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64
|
||||
+ 60_000,
|
||||
}
|
||||
}
|
||||
}
|
||||
PluginServiceRequest::Complete {
|
||||
lease,
|
||||
event_id,
|
||||
result,
|
||||
} => {
|
||||
let mut state = self.state.lock().unwrap();
|
||||
assert_eq!(state.lease.as_ref(), Some(&lease));
|
||||
assert_eq!(event_id, "event");
|
||||
assert_eq!(
|
||||
result,
|
||||
match self.complete_response {
|
||||
CompleteResponse::ConfirmIndeterminate =>
|
||||
PluginServiceEventResult::Indeterminate,
|
||||
_ => PluginServiceEventResult::Ack,
|
||||
}
|
||||
);
|
||||
// Model an owner decision committed before its response is lost.
|
||||
// Audited recovery itself is deliberately outside this unit test.
|
||||
state.pause = Some(PluginServicePauseReason::Indeterminate);
|
||||
match self.complete_response {
|
||||
CompleteResponse::Lost => return Err(unavailable()),
|
||||
CompleteResponse::Unexpected => PluginServiceResponse::Idle,
|
||||
CompleteResponse::ConfirmIndeterminate => PluginServiceResponse::Completed,
|
||||
}
|
||||
}
|
||||
PluginServiceRequest::Release { lease } => {
|
||||
let mut state = self.state.lock().unwrap();
|
||||
assert_eq!(state.lease.as_ref(), Some(&lease));
|
||||
if self.deny_release.load(Ordering::SeqCst) {
|
||||
return Err(unavailable());
|
||||
}
|
||||
state.lease = None;
|
||||
PluginServiceResponse::Released
|
||||
}
|
||||
PluginServiceRequest::Diagnostics { consumer_id } => {
|
||||
assert_eq!(consumer_id, self.consumer());
|
||||
PluginServiceResponse::Diagnostics {
|
||||
diagnostics: self.diagnostics(),
|
||||
}
|
||||
}
|
||||
PluginServiceRequest::Host { lease, request } => {
|
||||
assert_eq!(self.state.lock().unwrap().lease.as_ref(), Some(&lease));
|
||||
let response = match *request {
|
||||
PluginInvocationRequest::Authorize { invocation_id, .. } => {
|
||||
assert_eq!(invocation_id, "invocation");
|
||||
PluginInvocationResponse::Authorized
|
||||
}
|
||||
PluginInvocationRequest::ResolveTicket { ticket_ref, .. } => {
|
||||
assert_eq!(ticket_ref, "T-796");
|
||||
PluginInvocationResponse::TicketResolved {
|
||||
ticket_id: "canonical-ticket".into(),
|
||||
}
|
||||
}
|
||||
PluginInvocationRequest::ReadTicket { .. } => {
|
||||
PluginInvocationResponse::TicketRead {
|
||||
snapshot: PluginTicketSnapshot {
|
||||
title: "note".into(),
|
||||
case_state: PluginTicketCaseState::Open,
|
||||
content_digest: "digest".into(),
|
||||
body: "body".into(),
|
||||
truncated: false,
|
||||
},
|
||||
}
|
||||
}
|
||||
PluginInvocationRequest::DataLookupMutation { .. } => {
|
||||
PluginInvocationResponse::DataMutationLookup {
|
||||
outcome: PluginDataMutationLookup::NotRecorded,
|
||||
}
|
||||
}
|
||||
PluginInvocationRequest::DataGet { .. } => PluginInvocationResponse::DataCell {
|
||||
cell: PluginDataCell::NeverWritten,
|
||||
},
|
||||
PluginInvocationRequest::DataMutate { .. } => {
|
||||
PluginInvocationResponse::DataMutated {
|
||||
receipt: PluginDataReceipt {
|
||||
mutation_id: "mutation".into(),
|
||||
result_digest: "result".into(),
|
||||
},
|
||||
}
|
||||
}
|
||||
PluginInvocationRequest::Open { .. } => PluginInvocationResponse::Opened,
|
||||
PluginInvocationRequest::Send { .. } => PluginInvocationResponse::Response {
|
||||
response: PluginHttpResponse {
|
||||
status: 200,
|
||||
headers: vec![],
|
||||
body: b"ok".to_vec(),
|
||||
},
|
||||
},
|
||||
other => panic!("unexpected guest Host request: {other:?}"),
|
||||
};
|
||||
PluginServiceResponse::Host {
|
||||
response: Box::new(response),
|
||||
}
|
||||
}
|
||||
};
|
||||
Ok(PluginHostResponse::Service { response })
|
||||
}
|
||||
}
|
||||
|
||||
async fn run(driver: &Arc<ServiceDriver>, context: &BackgroundTaskContext) {
|
||||
driver
|
||||
.run(context.clone(), BackgroundTaskCancellation::default())
|
||||
.await
|
||||
.expect("a parked observer must remain a successful HostReady run");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancelled_inline_stop_retains_lease_for_protected_cleanup() {
|
||||
let fixture = Fixture::new();
|
||||
let owner = Owner::new("selected", CompleteResponse::Lost);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
let context = context().await;
|
||||
run(&driver, &context).await;
|
||||
let lease = driver.lease.lock().unwrap().clone().unwrap();
|
||||
owner.state.lock().unwrap().pause = Some(PluginServicePauseReason::Authorization);
|
||||
owner.gate_log.store(true, Ordering::SeqCst);
|
||||
|
||||
// Poll observes a pause and enters inline stopped(). Drop the *run future*
|
||||
// at a known guest logging await, just as HostReady cancellation does.
|
||||
{
|
||||
let pending = driver.run(context.clone(), BackgroundTaskCancellation::default());
|
||||
tokio::pin!(pending);
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
tokio::select! {
|
||||
result = &mut pending => panic!("stop must wait at LifecycleLog: {result:?}"),
|
||||
permit = owner.log_entered.acquire() => permit.unwrap().forget(),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("real guest on_stop must reach gated lifecycle logging");
|
||||
}
|
||||
assert_eq!(driver.lease.lock().unwrap().as_ref(), Some(&lease));
|
||||
assert_eq!(
|
||||
driver.parked_consumer.lock().unwrap().as_deref(),
|
||||
Some(lease.consumer_id.as_str())
|
||||
);
|
||||
assert!(
|
||||
!owner
|
||||
.requests()
|
||||
.iter()
|
||||
.any(|request| matches!(request, PluginServiceRequest::Release { .. }))
|
||||
);
|
||||
|
||||
let logs_before_cleanup = owner
|
||||
.requests()
|
||||
.iter()
|
||||
.filter(|r| matches!(r, PluginServiceRequest::LifecycleLog { .. }))
|
||||
.count();
|
||||
owner.gate_log.store(false, Ordering::SeqCst);
|
||||
driver.stopped(context.clone()).await.unwrap();
|
||||
assert_eq!(
|
||||
owner
|
||||
.requests()
|
||||
.iter()
|
||||
.filter(|r| matches!(r, PluginServiceRequest::LifecycleLog { .. }))
|
||||
.count(),
|
||||
logs_before_cleanup,
|
||||
"interrupted on-stop is not invoked twice"
|
||||
);
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
assert!(owner.state.lock().unwrap().lease.is_none());
|
||||
let releases = owner
|
||||
.requests()
|
||||
.into_iter()
|
||||
.filter_map(|request| match request {
|
||||
PluginServiceRequest::Release { lease } => Some(lease),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(releases, [lease]);
|
||||
let requests = owner.requests();
|
||||
driver.stopped(context).await.unwrap();
|
||||
assert_eq!(
|
||||
owner.requests(),
|
||||
requests,
|
||||
"confirmed release is idempotent"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_stop_export_still_releases_the_owner_lease() {
|
||||
let fixture = Fixture::new();
|
||||
let owner = Owner::new("selected", CompleteResponse::Lost);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
let context = context().await;
|
||||
run(&driver, &context).await;
|
||||
let lease = driver.lease.lock().unwrap().clone().unwrap();
|
||||
owner.deny_log.store(true, Ordering::SeqCst);
|
||||
assert!(driver.stopped(context).await.is_err());
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
assert!(owner.state.lock().unwrap().lease.is_none());
|
||||
assert_eq!(
|
||||
owner.requests().last(),
|
||||
Some(&PluginServiceRequest::Release { lease })
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unconfirmed_release_retains_the_exact_proof_until_retry_is_confirmed() {
|
||||
let fixture = Fixture::new();
|
||||
let owner = Owner::new("selected", CompleteResponse::Lost);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
let context = context().await;
|
||||
run(&driver, &context).await;
|
||||
let lease = driver.lease.lock().unwrap().clone().unwrap();
|
||||
owner.deny_release.store(true, Ordering::SeqCst);
|
||||
assert!(driver.stopped(context.clone()).await.is_err());
|
||||
assert_eq!(driver.lease.lock().unwrap().as_ref(), Some(&lease));
|
||||
assert_eq!(owner.state.lock().unwrap().lease.as_ref(), Some(&lease));
|
||||
assert_eq!(
|
||||
owner.requests().last(),
|
||||
Some(&PluginServiceRequest::Release {
|
||||
lease: lease.clone(),
|
||||
})
|
||||
);
|
||||
owner.deny_release.store(false, Ordering::SeqCst);
|
||||
driver.stopped(context).await.unwrap();
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
assert_eq!(
|
||||
owner.requests().last(),
|
||||
Some(&PluginServiceRequest::Release { lease })
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn uncertain_complete_parks_without_retry_and_reacquires_only_recovered_consumer() {
|
||||
let fixture = Fixture::new();
|
||||
let context = context().await;
|
||||
for response in [CompleteResponse::Lost, CompleteResponse::Unexpected] {
|
||||
let owner = Owner::new("selected", response);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
let other_owner = Owner::new("other", response);
|
||||
let other = fixture.driver(other_owner.clone());
|
||||
for (driver, owner) in [(&driver, &owner), (&other, &other_owner)] {
|
||||
run(driver, &context).await; // Acquire + real guest on_start.
|
||||
run(driver, &context).await; // Real guest effects + uncertain Complete.
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
assert_eq!(
|
||||
driver.parked_consumer.lock().unwrap().as_deref(),
|
||||
Some(owner.consumer().as_str())
|
||||
);
|
||||
let requests = owner.requests();
|
||||
assert_eq!(
|
||||
requests
|
||||
.iter()
|
||||
.filter(|request| matches!(request,
|
||||
PluginServiceRequest::Host { request, .. }
|
||||
if matches!(request.as_ref(), PluginInvocationRequest::DataMutate { .. })
|
||||
))
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
requests
|
||||
.iter()
|
||||
.filter(|request| matches!(request,
|
||||
PluginServiceRequest::Host { request, .. }
|
||||
if matches!(request.as_ref(), PluginInvocationRequest::Send { .. })
|
||||
))
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
requests
|
||||
.iter()
|
||||
.filter(|request| matches!(request, PluginServiceRequest::Complete { .. }))
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
owner.diagnostics().pause_reason,
|
||||
Some(PluginServicePauseReason::Indeterminate)
|
||||
);
|
||||
assert!(!owner.diagnostics().lease_active);
|
||||
let before = requests.len();
|
||||
run(driver, &context).await;
|
||||
assert_eq!(
|
||||
&owner.requests()[before..],
|
||||
&[PluginServiceRequest::Diagnostics {
|
||||
consumer_id: owner.consumer(),
|
||||
}],
|
||||
"paused runs must not poll, re-admit the guest, repeat effects, or complete again"
|
||||
);
|
||||
}
|
||||
|
||||
let other_before = other_owner.requests();
|
||||
let original_lease = owner
|
||||
.requests()
|
||||
.into_iter()
|
||||
.find_map(|request| match request {
|
||||
PluginServiceRequest::Release { lease } => Some(lease),
|
||||
_ => None,
|
||||
})
|
||||
.unwrap();
|
||||
// Script the result of external owner recovery, not an unaudited driver
|
||||
// resume. A cleared pause alone is insufficient while any lease is live.
|
||||
{
|
||||
let mut state = owner.state.lock().unwrap();
|
||||
state.pause = None;
|
||||
state.lease = Some(original_lease.clone());
|
||||
}
|
||||
let before = owner.requests().len();
|
||||
run(&driver, &context).await;
|
||||
assert_eq!(
|
||||
&owner.requests()[before..],
|
||||
&[PluginServiceRequest::Diagnostics {
|
||||
consumer_id: owner.consumer(),
|
||||
}]
|
||||
);
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
owner.state.lock().unwrap().lease = None;
|
||||
run(&driver, &context).await;
|
||||
let reacquired = driver.lease.lock().unwrap().clone().unwrap();
|
||||
assert_eq!(reacquired.consumer_id, original_lease.consumer_id);
|
||||
assert_ne!(reacquired.lease_token, original_lease.lease_token);
|
||||
assert!(driver.parked_consumer.lock().unwrap().is_none());
|
||||
assert_eq!(owner.state.lock().unwrap().acquisitions, 2);
|
||||
assert_eq!(
|
||||
other_owner.requests(),
|
||||
other_before,
|
||||
"no global driver rearm"
|
||||
);
|
||||
let before = other_before.len();
|
||||
run(&other, &context).await;
|
||||
assert_eq!(
|
||||
&other_owner.requests()[before..],
|
||||
&[PluginServiceRequest::Diagnostics {
|
||||
consumer_id: other_owner.consumer(),
|
||||
}]
|
||||
);
|
||||
assert_eq!(other_owner.state.lock().unwrap().acquisitions, 1);
|
||||
assert!(other.lease.lock().unwrap().is_none());
|
||||
driver.stopped(context.clone()).await.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unconfirmed_indeterminate_control_reconciles_pause_without_redelivering_guest() {
|
||||
let fixture = Fixture::new();
|
||||
let owner = Owner::new("selected", CompleteResponse::ConfirmIndeterminate);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
let context = context().await;
|
||||
*driver.parked_consumer.lock().unwrap() = Some(owner.consumer());
|
||||
*driver.unconfirmed_indeterminate.lock().unwrap() = Some("event".into());
|
||||
// A lost Complete(Indeterminate) reply does not prove a durable pause.
|
||||
// Script precisely that observation: same head, no pause, no active lease.
|
||||
run(&driver, &context).await;
|
||||
assert_eq!(
|
||||
owner.diagnostics().pause_reason,
|
||||
Some(PluginServicePauseReason::Indeterminate)
|
||||
);
|
||||
assert!(driver.unconfirmed_indeterminate.lock().unwrap().is_none());
|
||||
assert_eq!(
|
||||
driver.parked_consumer.lock().unwrap().as_ref(),
|
||||
Some(&owner.consumer())
|
||||
);
|
||||
assert!(driver.lease.lock().unwrap().is_none());
|
||||
assert!(!owner.requests().iter().any(|r| matches!(
|
||||
r,
|
||||
PluginServiceRequest::Poll { .. } | PluginServiceRequest::Host { .. }
|
||||
)));
|
||||
assert!(owner.requests().iter().any(|r| matches!(r, PluginServiceRequest::Complete { event_id, result: PluginServiceEventResult::Indeterminate, .. } if event_id == "event")));
|
||||
let before = owner.requests().len();
|
||||
run(&driver, &context).await;
|
||||
assert_eq!(
|
||||
&owner.requests()[before..],
|
||||
&[PluginServiceRequest::Diagnostics {
|
||||
consumer_id: owner.consumer()
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn real_lifecycle_sends_exact_admit_and_publish_operations_and_honors_publish_denial() {
|
||||
let fixture = Fixture::new();
|
||||
let owner = Owner::new("selected", CompleteResponse::Lost);
|
||||
let driver = fixture.driver(owner.clone());
|
||||
driver.prepare_start().await.unwrap();
|
||||
let expected = vec![
|
||||
serde_json::json!({"kind":"prepare", "plan_id":"plan", "installation_id":"selected", "service_name":SERVICE, "operation":"admit"}),
|
||||
serde_json::json!({"kind":"prepare", "plan_id":"plan", "installation_id":"selected", "service_name":SERVICE, "operation":"publish"}),
|
||||
];
|
||||
let prepare_wire = |requests: Vec<PluginServiceRequest>| {
|
||||
requests
|
||||
.into_iter()
|
||||
.filter(|request| matches!(request, PluginServiceRequest::Prepare { .. }))
|
||||
.map(|request| serde_json::to_value(request).unwrap())
|
||||
.collect::<Vec<_>>()
|
||||
};
|
||||
assert_eq!(prepare_wire(owner.requests()), expected);
|
||||
assert_eq!(
|
||||
owner.requests().len(),
|
||||
3,
|
||||
"Admit, guest lifecycle log, Publish"
|
||||
);
|
||||
assert!(owner.state.lock().unwrap().lease.is_none());
|
||||
owner.deny_publish.store(true, Ordering::SeqCst);
|
||||
let before = owner.requests().len();
|
||||
assert!(
|
||||
driver.prepare_start().await.is_err(),
|
||||
"Publish is not another Admit check"
|
||||
);
|
||||
assert_eq!(prepare_wire(owner.requests()[before..].to_vec()), expected);
|
||||
}
|
||||
@@ -100,6 +100,7 @@ impl Fixture {
|
||||
world: m.runtime.world,
|
||||
required_imports: m.runtime.required_imports,
|
||||
tools: serde_json::from_value(serde_json::to_value(m.tools).unwrap()).unwrap(),
|
||||
services: vec![],
|
||||
connections: vec![],
|
||||
requested_grants: vec![],
|
||||
data_collections: vec![],
|
||||
@@ -148,6 +149,7 @@ impl Fixture {
|
||||
},
|
||||
expectation: ArtifactExpectation::new(self.source.artifact.clone()),
|
||||
tools: vec!["example_echo".into()],
|
||||
services: vec![],
|
||||
config: json!({}),
|
||||
grants: Grants::default(),
|
||||
})
|
||||
|
||||
@@ -4674,6 +4674,27 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
||||
}
|
||||
|
||||
impl<C: LlmClient + 'static, St: Store + 'static> Worker<C, St> {
|
||||
/// Shared startup/restore boundary, called only by a published Controller.
|
||||
pub(crate) fn start_host_ready_feature_tasks(&self) {
|
||||
if let Err(error) = self
|
||||
.feature_background_tasks
|
||||
.start_host_ready(self.hook_invocation_context(None))
|
||||
{
|
||||
tracing::warn!(error = %error, "HostReady background task start failed");
|
||||
}
|
||||
}
|
||||
|
||||
/// Host management may wake one task after domain-owned recovery succeeds.
|
||||
/// This does not authorize recovery or reactivate any other paused task.
|
||||
pub fn reactivate_host_ready_feature_task(
|
||||
&self,
|
||||
feature_id: &crate::feature::FeatureId,
|
||||
task_name: &str,
|
||||
) -> Result<crate::feature::background::BackgroundTaskStart, crate::hook::HookError> {
|
||||
self.feature_background_tasks
|
||||
.reactivate_host_ready(feature_id, task_name)
|
||||
}
|
||||
|
||||
pub async fn stop_feature_runtime(&mut self, reason: impl Into<String>) {
|
||||
if let Some(hooks) = self.hook_registry.clone() {
|
||||
let context = WorkerStoppingContext {
|
||||
@@ -4699,6 +4720,15 @@ impl<C: LlmClient + 'static, St: Store + 'static> Worker<C, St> {
|
||||
if let Err(error) = self.feature_background_tasks.shutdown().await {
|
||||
tracing::warn!(error = %error, "feature background task shutdown failed");
|
||||
}
|
||||
if let Some(plugins) = &self.plugin_tools {
|
||||
// The registry has joined the drivers, including bounded on-stop.
|
||||
// Do not free scoped resources or acknowledge shutdown while an
|
||||
// uncertain Service lease release still belongs to this Host.
|
||||
while let Err(error) = plugins.reconcile_service_shutdown().await {
|
||||
tracing::warn!(error = %error, "Plugin Service lease release pending; retaining cleanup ownership");
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user