frame_host/
doc_binding.rs

1//! The document-service transport binding (IRIDIUM-A3 R11, C1 as ruled):
2//! frame-host is the thin embedding host — the service logic lives in
3//! `frame-doc-service`; THIS module owns only the concrete bus wiring:
4//!
5//! - a [`FeedSink`] over the bus SDK's TCP push client, publishing every
6//!   feed frame to the declared feed channel;
7//! - the authoring pump: a TCP channel subscription on the declared
8//!   authoring channel whose delivered bytes are pushed into
9//!   [`DocumentService::receive_command`], every refusal logged loudly.
10//!
11//! Disconnect-release (C8, verified at liminal 0.3.0): the channel
12//! delivery surface exposes NO participant-lifecycle observability for a
13//! channel publisher, so v1 releases via the DECLARED policy expiry
14//! (`[document].lease_expiry_ms`, observed lazily by the authority) plus
15//! the page's best-effort `release-lease` on unload. The gap is recorded
16//! as an upstream liminal ask in the leg's report; epoch fencing keeps
17//! every failure mode safe (a dead holder's lease expires; its ghost
18//! proposals fence).
19
20use std::sync::Arc;
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::thread::JoinHandle;
23use std::time::{Duration, Instant};
24
25use frame_authority::{AuthorityConfig, DocumentId};
26use frame_core::component::ComponentId;
27use frame_doc_service::{DocumentBinding, DocumentService, FeedSink, SinkError};
28use frame_editor_wire::snapshot::PresentationInputs;
29use frame_state::{ComponentSchema, ComponentStateStore, ComponentStoreHandle};
30use liminal_sdk::remote::{PushClient, SubscriptionStream};
31
32use crate::config::{DocumentSection, FrameConfig};
33use crate::embedded::EmbeddedLiminal;
34use crate::error::HostError;
35use crate::syntax_theme::default_syntax_theme;
36
37/// The authoring pump's receive quantum. Delivery is server push over the
38/// blocking SDK stream; this bound only caps how long one idle block lasts
39/// (and therefore the shutdown latency) — it is NOT a polling interval
40/// (the frame-host lifecycle-logger precedent, `runtime.rs`).
41const PUMP_QUANTUM: Duration = Duration::from_millis(500);
42
43/// The live document binding: the service plus its bus wiring. Shutdown
44/// order: stop the pump, join it, then tear the service down.
45pub struct DocBinding {
46    service: Arc<DocumentService>,
47    pump: JoinHandle<()>,
48    stop: Arc<AtomicBool>,
49}
50
51/// The feed publisher: one TCP push client on the declared feed channel.
52/// The client is `!Sync` (it owns its ack receiver), and two threads
53/// legitimately publish — the service's feed pump and its resync path —
54/// so the mutex serializes them.
55struct LiminalFeedSink {
56    client: std::sync::Mutex<PushClient>,
57    channel: String,
58}
59
60impl FeedSink for LiminalFeedSink {
61    fn publish(&self, frame: &str) -> Result<(), SinkError> {
62        self.client
63            .lock()
64            .map_err(|_| SinkError::new("feed publisher mutex poisoned"))?
65            .publish(&self.channel, frame.as_bytes().to_vec())
66            .map_err(|error| SinkError::new(format!("feed publish failed: {error}")))
67    }
68}
69
70impl DocBinding {
71    /// Boots the binding when `[document]` is configured: the frame-state
72    /// store, the document service, the feed publisher, and the authoring
73    /// pump — each failure typed and loud, nothing half-up.
74    ///
75    /// # Errors
76    ///
77    /// Returns the first typed failure from content reading, state-store
78    /// opening, service boot, bus client connection, or thread start.
79    pub fn boot(
80        liminal: &EmbeddedLiminal,
81        config: &FrameConfig,
82    ) -> Result<Option<Self>, HostError> {
83        let Some(section) = &config.document else {
84            return Ok(None);
85        };
86        let initial_content = std::fs::read_to_string(&section.content_path).map_err(|source| {
87            HostError::DocumentContent {
88                path: section.content_path.clone(),
89                source,
90            }
91        })?;
92        let handle = open_store(section)?;
93        let authority_config = AuthorityConfig::declare(
94            Duration::from_millis(section.lease_expiry_ms),
95            section.journal_length_bound,
96            Duration::from_millis(section.quiesce_window_ms),
97        )
98        .map_err(|source| HostError::DocumentAuthorityConfig { source })?;
99        let document =
100            DocumentId::new(section.id.clone()).map_err(|error| HostError::ConfigContract {
101                detail: format!("[document].id rejected by the model: {error}"),
102            })?;
103        let binding = DocumentBinding {
104            document,
105            language: section.language.clone(),
106            initial_content,
107            component_id: section.component_id.clone(),
108            presentation: PresentationInputs {
109                dark_theme: section.dark_theme,
110                // C5, amended 2026-07-21: a highlighter must visibly
111                // highlight out of the box. [document].syntax_theme, when
112                // declared, REPLACES the built-in default wholesale (no
113                // per-key merging); when absent, the built-in default
114                // palette (copied from the proven example theme, never
115                // invented) is used instead. The snapshot therefore always
116                // carries Some(theme) — never the perpetually-absent member
117                // C5 originally read as monochrome-by-default.
118                syntax_theme: Some(
119                    section
120                        .syntax_theme
121                        .clone()
122                        .unwrap_or_else(default_syntax_theme),
123                ),
124            },
125        };
126        let address = liminal.tcp_addr().to_string();
127        let auth_token = config
128            .bus
129            .auth
130            .as_ref()
131            .map(|auth| auth.token.clone().into_bytes());
132        let sink = LiminalFeedSink {
133            client: std::sync::Mutex::new(connect_publisher(&address, auth_token.as_deref())?),
134            channel: section.feed_channel.clone(),
135        };
136        let service = Arc::new(
137            DocumentService::start(handle, authority_config, binding, Arc::new(sink))
138                .map_err(|source| HostError::DocumentService { source })?,
139        );
140        // The pinned SDK's SubscriptionStream sends an empty auth token by
141        // construction (0.3.0); a token-gated bus + [document] is refused
142        // at config load, so this open is always against an open server.
143        let stream = SubscriptionStream::open(&address, &section.authoring_channel, Vec::new())
144            .map_err(|error| HostError::DocumentBusClient {
145                component: "authoring subscriber",
146                detail: error.to_string(),
147            })?;
148        let stop = Arc::new(AtomicBool::new(false));
149        let pump = spawn_authoring_pump(stream, Arc::clone(&service), Arc::clone(&stop))?;
150        tracing::info!(
151            document = %section.id,
152            feed_channel = %section.feed_channel,
153            authoring_channel = %section.authoring_channel,
154            "document service live behind the embedded bus"
155        );
156        Ok(Some(Self {
157            service,
158            pump,
159            stop,
160        }))
161    }
162
163    /// Ordered teardown: stop the pump, join it, tear the service down.
164    ///
165    /// # Errors
166    ///
167    /// Returns typed pump-join and service-shutdown failures.
168    pub fn shutdown(self) -> Result<(), HostError> {
169        self.stop.store(true, Ordering::SeqCst);
170        self.pump.join().map_err(|_| HostError::AuthoringPump {
171            detail: "the authoring pump thread panicked".to_owned(),
172        })?;
173        let service = Arc::try_unwrap(self.service).map_err(|_| HostError::AuthoringPump {
174            detail: "the authoring pump still holds the service after join".to_owned(),
175        })?;
176        service
177            .shutdown()
178            .map_err(|source| HostError::DocumentService { source })
179    }
180}
181
182/// Connects the feed publisher, with the bus auth token when one gates the
183/// server.
184fn connect_publisher(address: &str, auth_token: Option<&[u8]>) -> Result<PushClient, HostError> {
185    let connected = match auth_token {
186        Some(token) => PushClient::connect_with_auth(address, token),
187        None => PushClient::connect(address),
188    };
189    connected.map_err(|error| HostError::DocumentBusClient {
190        component: "feed publisher",
191        detail: error.to_string(),
192    })
193}
194
195/// Opens (or reopens) the document service's frame-state namespace.
196fn open_store(section: &DocumentSection) -> Result<ComponentStoreHandle, HostError> {
197    let store = ComponentStateStore::open(&section.state_dir)
198        .map_err(|source| HostError::DocumentState { source })?;
199    let component = ComponentId::derive("frame-doc-service", &section.id);
200    if let Ok(handle) = store.component_handle(component) {
201        return Ok(handle);
202    }
203    // First boot for this namespace: declare and install.
204    store
205        .declare_schema(component, ComponentSchema::lww())
206        .map_err(|source| HostError::DocumentState { source })?;
207    store
208        .install_storage(component)
209        .map_err(|source| HostError::DocumentState { source })
210}
211
212/// The authoring pump: delivered command bytes go into the service; every
213/// refusal is logged loudly; the thread exits on the stop flag or on
214/// stream death (a dead wire is a loud death, never a silent one).
215fn spawn_authoring_pump(
216    stream: SubscriptionStream,
217    service: Arc<DocumentService>,
218    stop: Arc<AtomicBool>,
219) -> Result<JoinHandle<()>, HostError> {
220    std::thread::Builder::new()
221        .name("frame-doc-service-authoring-pump".to_owned())
222        .spawn(move || {
223            loop {
224                if stop.load(Ordering::SeqCst) {
225                    tracing::info!("authoring pump exiting on shutdown");
226                    return;
227                }
228                let waited = Instant::now();
229                match stream.recv_timeout(PUMP_QUANTUM) {
230                    Ok(delivered) => match String::from_utf8(delivered.payload().to_vec()) {
231                        Ok(bytes) => match service.receive_command(&bytes) {
232                            Ok(outcome) => {
233                                tracing::info!(?outcome, "authoring command admitted");
234                            }
235                            Err(refusal) => {
236                                tracing::warn!(%refusal, "authoring command REFUSED (loud, nothing applied)");
237                            }
238                        },
239                        Err(error) => {
240                            tracing::warn!(%error, "authoring frame is not UTF-8; refused loudly");
241                        }
242                    },
243                    Err(error) => {
244                        // The SDK folds the idle timeout and reader death into
245                        // one Connection error; a sub-quantum return means the
246                        // reader is gone (an mpsc disconnect returns
247                        // immediately). Flagged as an upstream SDK seam in the
248                        // leg's report.
249                        if waited.elapsed() < PUMP_QUANTUM / 2 && !stop.load(Ordering::SeqCst) {
250                            tracing::error!(
251                                %error,
252                                "authoring subscription DIED; the pump exits loudly — commands no longer reach the service"
253                            );
254                            return;
255                        }
256                    }
257                }
258            }
259        })
260        .map_err(|source| HostError::AuthoringPump {
261            detail: format!("failed to start the authoring pump: {source}"),
262        })
263}
frame_host/doc_binding.rs

frame_host/
doc_binding.rs
