bamboo_engine/events/
root_actor.rs

1//! One immutable Root execution's actual per-session/account publication.
2
3use super::AccountEventSink;
4use bamboo_agent_core::AgentEvent;
5use bamboo_domain::{RootActorRuntimeWrite, Storage};
6use std::sync::Arc;
7
8#[derive(Clone)]
9pub struct RootActorEventPublication {
10    storage: Arc<dyn Storage>,
11    account_sink: Arc<AccountEventSink>,
12    owner: RootActorRuntimeWrite,
13}
14
15impl RootActorEventPublication {
16    pub fn new(
17        storage: Arc<dyn Storage>,
18        account_sink: Arc<AccountEventSink>,
19        owner: RootActorRuntimeWrite,
20    ) -> Self {
21        Self {
22            storage,
23            account_sink,
24            owner,
25        }
26    }
27
28    pub fn matches_execution(&self, session_id: &str, run_id: &str) -> bool {
29        self.owner.fence.actor_id == session_id && self.owner.fence.run_id == run_id
30    }
31
32    /// Numeric observations and a process-local opaque identifier, never the
33    /// owner capability or lease-owner value.
34    pub(crate) fn diagnostic_identity(&self) -> (u64, u64, String) {
35        (
36            self.owner.fence.lease_epoch,
37            self.owner.fence.attempt,
38            crate::execution::workspace_diagnostics::fence_fingerprint(&self.owner.fence),
39        )
40    }
41
42    pub async fn publish(
43        &self,
44        session_id: &str,
45        event: &AgentEvent,
46        publish: Box<dyn FnOnce() -> bool + Send>,
47    ) -> std::io::Result<()> {
48        let storage = self.storage.clone();
49        let sink = self.account_sink.clone();
50        let owner_for_queue = self.owner.clone();
51        let event = event.clone();
52        let session_id = session_id.to_owned();
53        let (queued, receipt) = tokio::sync::oneshot::channel();
54        self.storage
55            .publish_root_actor_runtime_event(
56                &self.owner,
57                Box::new(move |check_current| {
58                    check_current()?;
59                    if !publish() {
60                        return Err(std::io::Error::other(
61                            "Root per-session event publication was retired",
62                        ));
63                    }
64                    check_current()?;
65                    let confirmation = if event.is_durable_change() {
66                        Some(
67                            sink.record_root_actor(
68                                storage,
69                                owner_for_queue,
70                                Some(&session_id),
71                                &event,
72                            )
73                            .ok_or_else(|| {
74                                std::io::Error::other(
75                                    "Root account event queue rejected publication",
76                                )
77                            })?,
78                        )
79                    } else {
80                        None
81                    };
82                    let _ = queued.send(confirmation);
83                    Ok(())
84                }),
85            )
86            .await?;
87        // The account writer needs the same physical Root guards. Await only
88        // after the per-session publication job has released those guards.
89        let confirmation = receipt
90            .await
91            .map_err(|_| std::io::Error::other("Root account admission receipt closed"))?;
92        if let Some(confirmation) = confirmation {
93            if !tokio::time::timeout(std::time::Duration::from_secs(30), confirmation)
94                .await
95                .ok()
96                .and_then(Result::ok)
97                .unwrap_or(false)
98            {
99                // The account writer reports a boolean receipt, so an owner
100                // conflict there otherwise loses its typed cause. Revalidate
101                // without publishing or writing runtime state before returning the
102                // generic receipt failure; a lost owner must still interrupt
103                // its exact live transport through the event forwarder.
104                if let Err(error) = self
105                    .storage
106                    .publish_root_actor_runtime_event(
107                        &self.owner,
108                        Box::new(|check_current| check_current()),
109                    )
110                    .await
111                {
112                    if error
113                        .get_ref()
114                        .is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
115                    {
116                        return Err(error);
117                    }
118                }
119                return Err(std::io::Error::other(
120                    "Root account final publication was not confirmed",
121                ));
122            }
123        }
124        Ok(())
125    }
126}
bamboo_engine/events/root_actor.rs

bamboo_engine/events/
root_actor.rs
