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    pub async fn publish(
33        &self,
34        session_id: &str,
35        event: &AgentEvent,
36        publish: Box<dyn FnOnce() -> bool + Send>,
37    ) -> std::io::Result<()> {
38        let storage = self.storage.clone();
39        let sink = self.account_sink.clone();
40        let owner_for_queue = self.owner.clone();
41        let event = event.clone();
42        let session_id = session_id.to_owned();
43        let (queued, receipt) = tokio::sync::oneshot::channel();
44        self.storage
45            .publish_root_actor_runtime_event(
46                &self.owner,
47                Box::new(move |check_current| {
48                    check_current()?;
49                    if !publish() {
50                        return Err(std::io::Error::other(
51                            "Root per-session event publication was retired",
52                        ));
53                    }
54                    check_current()?;
55                    let confirmation = if event.is_durable_change() {
56                        Some(
57                            sink.record_root_actor(
58                                storage,
59                                owner_for_queue,
60                                Some(&session_id),
61                                &event,
62                            )
63                            .ok_or_else(|| {
64                                std::io::Error::other(
65                                    "Root account event queue rejected publication",
66                                )
67                            })?,
68                        )
69                    } else {
70                        None
71                    };
72                    let _ = queued.send(confirmation);
73                    Ok(())
74                }),
75            )
76            .await?;
77        // The account writer needs the same physical Root guards. Await only
78        // after the per-session publication job has released those guards.
79        let confirmation = receipt
80            .await
81            .map_err(|_| std::io::Error::other("Root account admission receipt closed"))?;
82        if let Some(confirmation) = confirmation {
83            if !tokio::time::timeout(std::time::Duration::from_secs(30), confirmation)
84                .await
85                .ok()
86                .and_then(Result::ok)
87                .unwrap_or(false)
88            {
89                // The account writer reports a boolean receipt, so an owner
90                // conflict there otherwise loses its typed cause. Revalidate
91                // without publishing or writing runtime state before returning the
92                // generic receipt failure; a lost owner must still interrupt
93                // its exact live transport through the event forwarder.
94                if let Err(error) = self
95                    .storage
96                    .publish_root_actor_runtime_event(
97                        &self.owner,
98                        Box::new(|check_current| check_current()),
99                    )
100                    .await
101                {
102                    if error
103                        .get_ref()
104                        .is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
105                    {
106                        return Err(error);
107                    }
108                }
109                return Err(std::io::Error::other(
110                    "Root account final publication was not confirmed",
111                ));
112            }
113        }
114        Ok(())
115    }
116}
bamboo_engine/events/root_actor.rs

bamboo_engine/events/
root_actor.rs
