harmos-runtime 0.1.4

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

#[path = "../support/process_graph_contract.rs"]
mod contract;

#[path = "../fixtures/process_provider.rs"]
mod provider;

#[path = "../fixtures/process_client.rs"]
mod client;

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::collections::BTreeMap;
use std::future::Future;
use std::num::NonZeroUsize;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::task::{Context, Poll};
// Third Party Imports
use contract::{POLICY, Probe, Step};
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::Notify;
// Local Imports
use super::{HostLanes, assembly, operations, service_status};
use crate::sidecar::support::Bank;
use crate::sidecar::witness::Witness;
use harmos_runtime::operations::OperationDeclaration;
use harmos_runtime::operations::OperationPolicy;
use harmos_runtime::sidecar::{Attach, Attached, Notice, Refusal, Supervised, Supervisor};
use harmos_runtime::{Message, Runtime, services::Service, services::ServiceStatus};
use harmos_sidecar::attach::{End, Ending};

/* === Constants === */

/* === Macros === */

macro_rules! forwarded {
    ($self:ident, $method:ident) => {
        match &$self.source {
            Source::Provider(source) => source.$method(),
            Source::Client(source) => source.$method(),
            Source::Independent(source) => source.$method(),
        }
    };
}

/* === Type Aliases === */

type Wait = Pin<Box<dyn Future<Output = ()> + Send>>;

/* === Structs === */

#[derive(Default)]
struct Gate {
    attached: AtomicUsize,
    starts: AtomicUsize,
    failures: AtomicUsize,
    joined: AtomicBool,
    finished: Notify,
    hold: AtomicBool,
    fail: AtomicBool,
    entered: Notify,
    release: Arc<Notify>,
    cut: Arc<Notify>,
}

#[derive(Clone)]
struct Joined {
    source: Source,
    gate: Arc<Gate>,
}

struct GatedEnding {
    original: Ending,
    gate: Arc<Gate>,
    started: bool,
    waiting: Option<Wait>,
}

struct CutRead {
    original: Box<dyn AsyncRead + Send + Unpin>,
    cut: Wait,
    closed: bool,
}

struct Independent {
    resources: Arc<Witness>,
}

struct Fixture {
    runtime: Runtime<Bank>,
    _lease: harmos_runtime::operations::OperationInstance<harmos_runtime::operations::resident::Activation>,
    marks: BTreeMap<&'static str, Arc<Witness>>,
    client: Arc<Gate>,
    provider: Arc<Gate>,
    independent: Arc<Gate>,
}

/* === Enums === */

#[derive(Clone)]
enum Source {
    Provider(harmos_sidecar::PluginNative<provider::Provider>),
    Client(harmos_sidecar::PluginNative<client::Client>),
    Independent(harmos_sidecar::PluginNative<Independent>),
}

/* === Traits === */

/* === Implementations === */

impl harmos_runtime::operations::lifecycle::Policy for Independent {
    const INITIALIZATION_POLICY: OperationPolicy = POLICY;
    const SHUTDOWN_POLICY: OperationPolicy = POLICY;
}

#[harmos_runtime::plugin(id = "process-independent", provides = (), requires = ())]
impl harmos_runtime::Plugin for Independent {
    type Context = Supervisor;
    type Config = String;
    type Error = Refusal;
    type Resources = Witness;
    fn prepare(key: String) -> Result<Self, Refusal> {
        Ok(Self { resources: Witness::find(&key).ok_or_else(|| Refusal::of("the story witness is gone"))? })
    }
    async fn initialize(&mut self, _: &Supervisor) -> Result<(), Refusal> {
        self.resources.mark("initialized");
        Ok(())
    }
    fn resources(&self) -> Arc<Witness> {
        self.resources.clone()
    }
    fn bind_operations(
        &self, _: &mut harmos_runtime::operations::WireTable<Supervisor>,
    ) -> Result<(), harmos_runtime::operations::ServingError> {
        Ok(())
    }
    async fn stop(&mut self, _: &Supervisor) -> Result<(), Refusal> {
        self.resources.mark("shutdown");
        Ok(())
    }
    async fn cleanup(&mut self) -> Result<(), Refusal> {
        self.resources.mark("cleaned");
        Ok(())
    }
}

impl AsyncRead for CutRead {
    fn poll_read(self: Pin<&mut Self>, context: &mut Context<'_>, buffer: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
        let reader = self.get_mut();
        if reader.closed || reader.cut.as_mut().poll(context).is_ready() {
            reader.closed = true;
            return Poll::Ready(Ok(()));
        }
        Pin::new(&mut reader.original).poll_read(context, buffer)
    }
}

impl End for GatedEnding {
    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>> {
        Box::pin(async move {
            if !self.started {
                self.started = true;
                self.gate.starts.fetch_add(1, Ordering::SeqCst);
                self.gate.entered.notify_one();
            }
            if self.gate.hold.load(Ordering::SeqCst) {
                if self.waiting.is_none() {
                    let release = self.gate.release.clone();
                    self.waiting = Some(Box::pin(async move { release.notified().await }));
                }
                self.waiting.as_mut().unwrap().await;
                self.waiting.take();
                self.gate.hold.store(false, Ordering::SeqCst);
            }
            if self.gate.fail.load(Ordering::SeqCst) {
                self.gate.failures.fetch_add(1, Ordering::SeqCst);
                return Err(std::io::Error::other("original fixture ending remains failed"));
            }
            self.original.settle().await?;
            self.gate.joined.store(true, Ordering::SeqCst);
            self.gate.finished.notify_one();
            Ok(())
        })
    }
}

impl Attach for Joined {
    type Error = std::io::Error;
    fn owner(&self) -> &str {
        forwarded!(self, owner)
    }
    fn operations(&self) -> harmos_domain::Result<Vec<harmos_runtime::operations::Definition>> {
        forwarded!(self, operations)
    }
    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<harmos_runtime::operations::Contract>>> {
        forwarded!(self, operation_contracts)
    }
    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::inventory::contracts::Required>>> {
        forwarded!(self, required_contracts)
    }
    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[harmos_runtime::operations::lifecycle::Descriptor; 2]>> {
        forwarded!(self, sdk_lifecycle)
    }
    fn publications(&self) -> Vec<harmos_sidecar::StreamDeclaration> {
        forwarded!(self, publications)
    }
    fn observers(&self) -> Vec<harmos_domain::inventory::observers::Declaration> {
        forwarded!(self, observers)
    }
    fn observe_contract(&self) -> harmos_domain::Result<Option<harmos_runtime::operations::Contract>> {
        forwarded!(self, observe_contract)
    }
    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
        forwarded!(self, listener_declarations)
    }
    async fn attach(&self) -> Result<Attached, Self::Error> {
        self.gate.attached.fetch_add(1, Ordering::SeqCst);
        let attached = match &self.source {
            Source::Provider(source) => source.attach().await?,
            Source::Client(source) => source.attach().await?,
            Source::Independent(source) => source.attach().await?,
        };
        let (write, read, control, diagnostics, original) = attached.into_parts();
        let ending = Ending::new(GatedEnding {
            original: original.expect("original SDK physical ending"),
            gate: self.gate.clone(),
            started: false,
            waiting: None,
        });
        let cut = self.gate.cut.clone();
        let read = CutRead { original: read, cut: Box::pin(async move { cut.notified().await }), closed: false };
        let mut attached = Attached::new(write, read).ending(ending);
        if let Some(control) = control {
            attached = attached.control(control);
        }
        if let Some(diagnostics) = diagnostics {
            attached = attached.diagnostics(diagnostics);
        }
        Ok(attached)
    }
}

impl Fixture {
    async fn new() -> Self {
        let fixture = Self::start(false).await;
        service_status(fixture.runtime.operations.watch_residents(), "local-graph", |status| {
            matches!(status, ServiceStatus::Running { .. })
        })
        .await;
        fixture
    }

    // Submit The Graph Without Awaiting Admission; `held` Holds The Client's Initialization.
    async fn start(held: bool) -> Self {
        let marks: BTreeMap<_, _> =
            ["process-provider", "process-client", "process-independent"].into_iter().map(|owner| (owner, Witness::open())).collect();
        if held {
            marks["process-client"].mark("hold-initialization");
        }
        let capacity = NonZeroUsize::new(2).unwrap();
        let provider = Arc::new(Gate::default());
        let client = Arc::new(Gate::default());
        let independent = Arc::new(Gate::default());
        let sources = [
            Joined { source: Source::Provider(harmos_sidecar::PluginNative::new(capacity).unwrap()), gate: provider.clone() },
            Joined { source: Source::Client(harmos_sidecar::PluginNative::new(capacity).unwrap()), gate: client.clone() },
            Joined { source: Source::Independent(harmos_sidecar::PluginNative::new(capacity).unwrap()), gate: independent.clone() },
        ];
        let definitions = sources
            .iter()
            .map(|source| {
                let key = marks[source.owner()].key();
                operations::definition(source, serde_json::to_vec(key).unwrap())
            })
            .collect::<Vec<_>>();
        let encoding_bytes = definitions.iter().map(|definition| definition.configuration.len() * 2).max().unwrap()
            + std::mem::size_of::<harmos_runtime::operations::lifecycle::Initialize>()
            + 84;
        assert!(encoding_bytes <= 512);
        let definitions = harmos_runtime::plugin::PluginDefinitions::new(definitions, []).unwrap();
        let residents = sources.map(|source| Supervised::new(source, notices as fn(&Service<Bank>, Notice)));
        let runtime = assembly()
            .operation_settings(harmos_runtime::operations::Settings {
                preparation_bytes: Some(NonZeroUsize::new(4096).unwrap()),
                ..Default::default()
            })
            .runtime()
            .await
            .unwrap();
        runtime
            .operations
            .register_service(
                harmos_runtime::services::ServiceDefinition {
                    id: "local-graph".into(),
                    policy: harmos_runtime::operations::OperationPolicy {
                        restart: harmos_runtime::operations::RestartPolicy::Never,
                        ..harmos_runtime::services::POLICY
                    },
                },
                move |_: Arc<()>| {
                    harmos_runtime::sidecar::Managed::new_graph(
                        residents.clone(),
                        definitions.clone(),
                        HostLanes::Open,
                        capacity,
                        NonZeroUsize::new(encoding_bytes).unwrap(),
                    )
                    .unwrap()
                },
            )
            .unwrap();
        let lease = runtime.operations.submit_resident("local-graph").await.unwrap();
        Self { runtime, _lease: lease, marks, client, provider, independent }
    }
    fn marks(&self, owner: &str) -> &Witness {
        &self.marks[owner]
    }
    fn settled(&self) {
        for gate in [&self.client, &self.provider, &self.independent] {
            assert_eq!(gate.attached.load(Ordering::SeqCst), 1);
            assert_eq!(gate.starts.load(Ordering::SeqCst), 1);
            assert!(gate.joined.load(Ordering::SeqCst));
        }
        for owner in ["process-provider", "process-client", "process-independent"] {
            assert_eq!(self.marks(owner).count("shutdown"), 1);
            assert_eq!(self.marks(owner).count("cleaned"), 1);
        }
        assert_eq!(self.marks("process-provider").count("read-finalized-2"), 1);
    }
}

/* === Public Functions === */

/* === Private Functions === */

fn notices(_: &Service<Bank>, _: Notice) {}

#[tokio::test]
async fn cancelled_stop_retains() {
    let fixture = Fixture::new().await;
    fixture.client.hold.store(true, Ordering::SeqCst);
    {
        let stop = fixture.runtime.stop();
        tokio::pin!(stop);
        tokio::select! {
            result = &mut stop => panic!("stop returned before original ending: {result:?}"),
            _ = async {
                fixture.client.entered.notified().await;
                fixture.independent.finished.notified().await;
            } => {}
        }
    }
    assert!(fixture.independent.joined.load(Ordering::SeqCst));
    assert!(!fixture.client.joined.load(Ordering::SeqCst));
    assert!(!fixture.provider.joined.load(Ordering::SeqCst));
    assert!(!fixture.marks("process-provider").seen("shutdown"));
    assert_eq!(fixture.marks("process-provider").count("read-finalized-2"), 1);
    fixture.client.release.notify_one();
    fixture.runtime.stop().await.unwrap();
    fixture.settled();
}

#[tokio::test]
async fn failed_ending_retires() {
    let fixture = Fixture::new().await;
    fixture.client.fail.store(true, Ordering::SeqCst);
    let result = fixture.runtime.stop().await;
    assert!(matches!(result, Err(harmos_runtime::Error::TaskFailed { domain: "service", .. })));
    assert!(!fixture.client.joined.load(Ordering::SeqCst));
    assert!(!fixture.provider.joined.load(Ordering::SeqCst));
    assert!(!fixture.marks("process-provider").seen("shutdown"));
    assert!(fixture.independent.joined.load(Ordering::SeqCst));
    assert_eq!(fixture.marks("process-independent").count("cleaned"), 1);
    let failures = fixture.client.failures.load(Ordering::SeqCst);
    assert!(failures > 0);
    let result = fixture.runtime.stop().await;
    assert!(matches!(result, Err(harmos_runtime::Error::TaskFailed { domain: "service", .. })));
    assert!(fixture.client.failures.load(Ordering::SeqCst) > failures);
    assert_eq!(fixture.client.starts.load(Ordering::SeqCst), 1);
    assert_eq!(fixture.marks("process-independent").count("cleaned"), 1);
    assert!(!fixture.marks("process-provider").seen("shutdown"));
    let failures = fixture.client.failures.load(Ordering::SeqCst);
    fixture.client.fail.store(false, Ordering::SeqCst);
    fixture.runtime.stop().await.unwrap();
    fixture.settled();
    assert_eq!(fixture.client.failures.load(Ordering::SeqCst), failures);
}

#[tokio::test]
async fn unrelated_failure_retires() {
    let fixture = Fixture::new().await;
    fixture.independent.fail.store(true, Ordering::SeqCst);
    let result = fixture.runtime.stop().await;
    assert!(matches!(result, Err(harmos_runtime::Error::TaskFailed { domain: "service", .. })));
    assert!(!fixture.independent.joined.load(Ordering::SeqCst));
    assert!(fixture.client.joined.load(Ordering::SeqCst));
    assert!(fixture.provider.joined.load(Ordering::SeqCst));
    assert_eq!(fixture.marks("process-provider").count("read-finalized-2"), 1);
    for owner in ["process-client", "process-provider"] {
        assert_eq!(fixture.marks(owner).count("shutdown"), 1);
        assert_eq!(fixture.marks(owner).count("cleaned"), 1);
    }
    fixture.independent.fail.store(false, Ordering::SeqCst);
    fixture.runtime.stop().await.unwrap();
    fixture.settled();
}

#[tokio::test]
async fn eof_retains_child() {
    let fixture = Fixture::new().await;
    let provider = fixture.marks("process-provider");
    provider.mark("hold-read-2");
    fixture.provider.hold.store(true, Ordering::SeqCst);
    {
        let stop = fixture.runtime.stop();
        tokio::pin!(stop);
        tokio::select! {
            result = &mut stop => panic!("stop returned before child request: {result:?}"),
            _ = provider.marked("read-started-2") => {}
        }
        fixture.provider.cut.notify_one();
        tokio::select! {
            result = &mut stop => panic!("stop returned before physical ending: {result:?}"),
            _ = fixture.provider.entered.notified() => {}
        }
        assert!(!fixture.provider.joined.load(Ordering::SeqCst));
        assert!(!fixture.client.joined.load(Ordering::SeqCst));
        assert!(!fixture.marks("process-client").seen("dependency-shutdown"));
        // Cancel Only The Wait; The Same Original Child And Ending Remain Owned.
    }
    fixture.provider.fail.store(true, Ordering::SeqCst);
    fixture.provider.release.notify_one();
    let failed = fixture.runtime.stop().await;
    assert!(matches!(failed, Err(harmos_runtime::Error::TaskFailed { domain: "service", .. })));
    assert!(fixture.provider.failures.load(Ordering::SeqCst) > 0);
    assert!(!fixture.provider.joined.load(Ordering::SeqCst));
    assert!(!fixture.client.joined.load(Ordering::SeqCst));
    fixture.provider.fail.store(false, Ordering::SeqCst);
    for _ in 0..3 {
        let result = fixture.runtime.stop().await;
        assert!(result.is_ok() || matches!(result, Err(harmos_runtime::Error::TaskFailed { domain: "service", .. })));
        if fixture.provider.joined.load(Ordering::SeqCst) && fixture.client.joined.load(Ordering::SeqCst) {
            break;
        }
    }
    for gate in [&fixture.client, &fixture.provider, &fixture.independent] {
        assert!(gate.joined.load(Ordering::SeqCst));
        assert_eq!(gate.attached.load(Ordering::SeqCst), 1);
        assert_eq!(gate.starts.load(Ordering::SeqCst), 1);
    }
    assert_eq!(provider.count("read-prepared-2"), 1);
    assert_eq!(provider.count("read-started-2"), 1);
    assert_eq!(provider.count("read-finalized-2"), 1);
    for owner in ["process-provider", "process-client", "process-independent"] {
        assert_eq!(fixture.marks(owner).count("cleaned"), 1);
    }
}

#[tokio::test]
async fn dependency_precedes_admission() {
    let fixture = Fixture::start(true).await;
    let client = fixture.marks("process-client");
    let provider = fixture.marks("process-provider");

    // The Lifecycle Loan Reaches Its Provider Before Ordinary Admission
    client.marked("dependency-initialized").await;
    assert_eq!(provider.count("initialized"), 1);
    assert_eq!(provider.count("read-finalized-1"), 1);
    let status = service_status(fixture.runtime.operations.watch_residents(), "local-graph", |_| true).await;
    assert_eq!(status, ServiceStatus::Starting);
    assert!(matches!(
        fixture.runtime.operations.submit_encoded(Probe::ID, Step { phase: 3 }.encode_to_vec()),
        Err(harmos_runtime::Error::NotReady { domain: "operation plugin" }),
    ));
    assert!(!client.seen("probe-initialized"));

    // Released Initialization Admits Ordinary Calls
    client.mark("release-initialization");
    service_status(fixture.runtime.operations.watch_residents(), "local-graph", |status| matches!(status, ServiceStatus::Running { .. }))
        .await;
    let response = fixture.runtime.operations.call_encoded(Probe::ID, Step { phase: 3 }.encode_to_vec()).await.unwrap();
    assert_eq!(Step::decode(&response).unwrap().phase, 3);
    assert_eq!(client.count("probe-initialized"), 1);
    assert_eq!(client.count("probe-finalized"), 1);

    // Reverse Retirement Keeps The Provider Through The Client's Shutdown Call
    fixture.runtime.stop().await.unwrap();
    assert_eq!(client.count("dependency-shutdown"), 1);
    assert_eq!(provider.count("read-started-2"), 1);
    fixture.settled();
}

/* === Tests === */