use std::num::NonZeroUsize;
use std::sync::Arc;
use super::caller::Source;
use super::manager::{Entry, Submission};
use super::wire::EncodedInput;
use super::{EncodedInvocation, OperationManager};
use crate::operations::lifecycle::{Action, Initialize};
use crate::plugin::{PluginKind, PluginLifecycleAuthority, PluginLifecyclePhase, PluginScope};
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
use crate::operations::{OperationDeclaration, OperationPolicy, RestartPolicy, RetryPolicy};
use crate::supervision::executor;
use crate::{Error, Message, Result};
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
pub(super) struct Encoded(pub(super) Vec<u8>);
#[must_use = "retain SDK entries through original invocation settlement"]
pub struct PluginSdkRegistration {
operations: OperationManager,
entries: [Arc<Entry>; 2],
kind: PluginKind,
}
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
impl crate::Message for Encoded {
fn encode_to_vec(&self) -> Vec<u8> {
self.0.clone()
}
fn encoded_len(&self) -> usize {
self.0.len()
}
fn decoding_memory(bytes: &[u8]) -> std::result::Result<usize, crate::MessageError> {
std::mem::size_of::<Self>()
.checked_add(bytes.len())
.ok_or_else(|| crate::MessageError::borrowed("forwarded request size overflows"))
}
fn decode(bytes: &[u8]) -> std::result::Result<Self, crate::MessageError> {
Ok(Self(bytes.to_vec()))
}
}
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
impl crate::schema::Described for Encoded {
const NAME: &'static str = "harmos::sdk_encoded";
const FIELDS: &'static [crate::schema::declaration::Field] = &[];
const MARKER: &'static [u8] = &crate::schema::declaration::encode::<
{ crate::schema::declaration::encoded_len(Self::NAME, Self::FIELDS) },
>(Self::NAME, Self::FIELDS);
}
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
impl crate::MemorySize for Encoded {
fn memory_size(&self) -> usize {
std::mem::size_of::<Self>().saturating_add(self.0.capacity())
}
}
#[cfg(all(feature = "plugin-guest", not(target_arch = "wasm32")))]
impl OperationDeclaration for Encoded {
type Response = Vec<u8>;
type Input = crate::operations::NoStream;
type Output = crate::operations::NoStream;
const VERSION: u32 = 1;
const ID: &'static str = "";
const POLICY: OperationPolicy = OperationPolicy {
timeout: std::time::Duration::ZERO,
attempt_timeout: std::time::Duration::ZERO,
retries: RetryPolicy::Never,
restart: RestartPolicy::Never,
};
}
impl PluginSdkRegistration {
pub(super) fn new(operations: OperationManager, scope: &PluginScope, mut entries: [Arc<Entry>; 2], kind: PluginKind) -> Result<Self> {
scope.issue_generation_context()?;
let pair = scope.lifecycle().ok_or(Error::InvalidInput { field: "SDK lifecycle", reason: "generation has no frozen SDK pair" })?;
for (entry, descriptor) in entries.iter_mut().zip(pair) {
let selected =
entry.plugin().ok_or(Error::InvalidInput { field: "SDK lifecycle", reason: "entry has no original generation" })?;
if !selected.same_activation(scope) || entry.contract.as_ref() != Some(descriptor.contract()) {
return Err(Error::InvalidInput {
field: "SDK lifecycle",
reason: "entry differs from the frozen SDK descriptor or generation",
});
}
let action = descriptor.action();
let mut expected = descriptor.contract().operation().clone();
let suffix = match action {
Action::Initialize => "initialize",
Action::Shutdown => "shutdown",
};
expected.id = crate::Identifier::new(format!("{}.sdk.{suffix}", scope.owner()))
.map_err(|source| Error::OperationDeclaration { source: harmos_domain::Error::from(source) })?;
if entry.declared() != &expected {
return Err(Error::InvalidInput { field: "SDK lifecycle", reason: "entry is not the exact qualified SDK definition" });
}
Arc::get_mut(entry).expect("new SDK entry has one owner").sdk = Some(action);
}
operations.reserve(&entries)?;
Ok(Self { operations, entries, kind })
}
pub fn submit_initialize(
&self, authority: &PluginLifecycleAuthority<'_>, config_json: &[u8], encoding_memory_bytes: NonZeroUsize,
) -> Result<EncodedInvocation> {
self.authorize(authority, 0)?;
let generation_context = self.entries[0].plugin().expect("SDK retains its original scope").issue_generation_context()?;
harmos_domain::validate_config(config_json)
.map_err(|source| Error::OperationFailed { operation: self.entries[0].id().to_owned(), source: Arc::new(source) })?;
self.submit(0, encoding_memory_bytes, |allowance| {
let peak = config_json
.len()
.checked_mul(2)
.and_then(|bytes| bytes.checked_add(std::mem::size_of::<Initialize>() + 64 + 20))
.ok_or(Error::ResourceExhausted { resource: "SDK encoding memory" })?;
if peak > allowance {
return Err(Error::ResourceExhausted { resource: "SDK encoding memory" });
}
let request = Initialize { config_json: config_json.to_vec(), generation_context: generation_context.to_vec() };
Ok(request.encode_to_vec())
})
}
pub fn submit_shutdown(&self, authority: &PluginLifecycleAuthority<'_>) -> Result<EncodedInvocation> {
self.authorize(authority, 1)?;
self.submit(1, NonZeroUsize::new(1).unwrap(), |_| Ok(Vec::new()))
}
fn authorize(&self, authority: &PluginLifecycleAuthority<'_>, index: usize) -> Result<()> {
let entry = &self.entries[index];
let scope = entry.plugin().expect("SDK entries retain their original scope");
let phase = if index == 0 { PluginLifecyclePhase::Initialize } else { PluginLifecyclePhase::Shutdown };
if !authority.matches_scope(scope, self.kind, phase)
|| authority.declaration().map(|descriptor| descriptor.contract()) != entry.contract.as_ref()
{
return Err(Error::InvalidInput {
field: "SDK lifecycle authority",
reason: "current hook loan differs from this original SDK action",
});
}
Ok(())
}
#[tracing::instrument(target = "harmos::operations", name = "operations.submit", level = "debug", skip_all)]
fn submit(&self, index: usize, bytes: NonZeroUsize, encode: impl FnOnce(usize) -> Result<Vec<u8>>) -> Result<EncodedInvocation> {
let submitted = executor::now();
let entry = &self.entries[index];
entry.available()?;
if !entry.settled() {
return Err(Error::ResourceExhausted { resource: "SDK physical settlement" });
}
let preparation = self.operations.claim_preparation(bytes)?;
#[cfg(feature = "plugin-sidecar")]
let native_capacity = entry
.native
.as_ref()
.map(|binding| binding.reserve())
.transpose()
.map_err(|source| Error::OperationFailed { operation: entry.id().to_owned(), source: Arc::new(source) })?;
let payload = encode(bytes.get())?;
if payload.capacity() > bytes.get() {
return Err(Error::ResourceExhausted { resource: "SDK encoding memory" });
}
let dispatch = entry.wire.ok_or(Error::InvalidInput { field: "SDK lifecycle", reason: "entry has no original wire adapter" })?;
let mut invocation = dispatch(
&self.operations,
Submission { submitted, entry: Arc::clone(entry), cause: None, preparation: None },
Source::Host,
None,
EncodedInput {
payload,
preparation: Some(preparation),
#[cfg(feature = "plugin-sidecar")]
native_capacity,
},
)?;
let outcome = invocation.outcome;
let control = Arc::clone(&invocation.control);
invocation.outcome = Box::pin(async move {
let payload = outcome.await?;
if !payload.is_empty() {
return Err(Error::InvalidInput { field: "SDK response", reason: "expected the exact empty SDK acknowledgement" });
}
control.settle().await.map_err(|source| Error::OperationSettlement { source })?;
Ok(payload)
});
Ok(invocation)
}
#[tracing::instrument(target = "harmos::operations", name = "operations.settle", level = "debug", skip_all)]
pub async fn settle(&self) -> Result<()> {
self.operations.settle_owned_entries(&self.entries).await?;
#[cfg(feature = "plugin-sidecar")]
if self.entries[0].plugin().is_some_and(PluginScope::is_closed)
&& let Some(binding) = &self.entries[0].native
{
binding.settle_observers().await?;
}
Ok(())
}
pub fn observer_outcome(&self) -> Result<()> {
#[cfg(feature = "plugin-sidecar")]
if let Some(binding) = &self.entries[0].native {
binding.observer_outcome()?;
}
Ok(())
}
#[tracing::instrument(target = "harmos::operations", name = "operations.close", level = "debug", skip_all)]
pub async fn close(&mut self) -> Result<()> {
self.operations.fence(&self.entries, false);
self.operations.settle_entries(&self.entries).await
}
}
impl Drop for PluginSdkRegistration {
fn drop(&mut self) {
self.operations.fence(&self.entries, true);
self.operations.remove_settled(&self.entries);
}
}