sim_lib_expr_tree_server/
server.rs

1//! Bounded authoritative session registry and request routing.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    sync::{
6        Arc, Mutex, MutexGuard,
7        atomic::{AtomicU64, Ordering},
8    },
9};
10
11use sim_kernel::{Cx, Env, Error, Expr, Symbol, Value};
12use sim_lib_expr_tree::TreeHandle;
13use sim_lib_server::{ServerAddress, SystemWallClock, WallClock};
14use sim_lib_view::SurfaceCodec;
15use sim_lib_view_expr_tree::ExpressionTreeSurfaceCodec;
16use sim_value::access;
17
18use crate::error::{ExpressionTreeServerError, ServerResult, internal};
19use crate::model::{ExpressionTreeServerLimits, SessionId, WatchBatch, WatchId};
20use crate::protocol;
21use crate::session::SessionRecord;
22
23mod route;
24
25static NEXT_SERVER_NONCE: AtomicU64 = AtomicU64::new(1);
26
27/// Authoritative bounded expression-tree session server.
28pub struct ExpressionTreeServer {
29    address: ServerAddress,
30    codecs: Vec<Symbol>,
31    clock: Arc<dyn WallClock>,
32    limits: ExpressionTreeServerLimits,
33    nonce: u64,
34    registry: Mutex<Registry>,
35}
36
37struct Registry {
38    sessions: BTreeMap<SessionId, SessionRecord>,
39    in_flight: BTreeSet<SessionId>,
40    next_session: u64,
41    next_tick: u64,
42}
43
44struct RuntimeTarget {
45    tree: TreeHandle,
46    resource: Symbol,
47}
48
49impl RuntimeTarget {
50    fn new(record: &SessionRecord) -> Self {
51        Self {
52            tree: record.tree.clone(),
53            resource: record.resource(),
54        }
55    }
56}
57
58impl ExpressionTreeServer {
59    /// Creates a server with explicit address, codecs, wall clock, and hard
60    /// lifecycle limits.
61    pub fn new(
62        address: ServerAddress,
63        codecs: Vec<Symbol>,
64        clock: Arc<dyn WallClock>,
65        limits: ExpressionTreeServerLimits,
66    ) -> ServerResult<Self> {
67        if codecs.is_empty() {
68            return Err(ExpressionTreeServerError::new(
69                "invalid-config",
70                "at least one server codec is required",
71            ));
72        }
73        if !limits.validate() {
74            return Err(ExpressionTreeServerError::new(
75                "invalid-config",
76                "all expression-tree server limits must be nonzero",
77            ));
78        }
79        Ok(Self {
80            address,
81            codecs,
82            clock,
83            limits,
84            nonce: NEXT_SERVER_NONCE.fetch_add(1, Ordering::Relaxed),
85            registry: Mutex::new(Registry {
86                sessions: BTreeMap::new(),
87                in_flight: BTreeSet::new(),
88                next_session: 1,
89                next_tick: 1,
90            }),
91        })
92    }
93
94    /// Creates a local server using the system wall clock and binary server
95    /// frames.
96    pub fn local() -> Self {
97        Self::new(
98            ServerAddress::Local,
99            vec![Symbol::qualified("codec", "binary")],
100            Arc::new(SystemWallClock),
101            ExpressionTreeServerLimits::default(),
102        )
103        .expect("default expression-tree server configuration is valid")
104    }
105
106    /// Returns the configured server address.
107    pub fn address(&self) -> &ServerAddress {
108        &self.address
109    }
110
111    /// Returns the configured frame codecs.
112    pub fn codecs(&self) -> &[Symbol] {
113        &self.codecs
114    }
115
116    /// Returns the configured lifecycle limits.
117    pub const fn limits(&self) -> ExpressionTreeServerLimits {
118        self.limits
119    }
120
121    /// Creates one authoritative session, capturing the creator's runtime and
122    /// immutable authority ceiling in the underlying expression tree.
123    pub fn create_session(&self, cx: &mut Cx, storage_name: &str) -> ServerResult<SessionId> {
124        let (id, tick) = {
125            let mut registry = self.lock_registry()?;
126            let tick = begin_request(&mut registry, self.limits);
127            if registry.sessions.len() >= self.limits.max_sessions {
128                return Err(ExpressionTreeServerError::new(
129                    "session-limit",
130                    format!("server session limit {} reached", self.limits.max_sessions),
131                ));
132            }
133            let id = SessionId(format!(
134                "{:016x}-{:016x}",
135                self.nonce, registry.next_session
136            ));
137            registry.next_session = registry.next_session.saturating_add(1);
138            (id, tick)
139        };
140
141        let value = cx
142            .eval_expr(Expr::Call {
143                operator: Box::new(Expr::Symbol(Symbol::qualified("expr-tree", "open"))),
144                args: vec![Expr::String(storage_name.to_owned())],
145            })
146            .map_err(classify_kernel_error)?;
147        let tree = value
148            .object()
149            .downcast_ref::<TreeHandle>()
150            .cloned()
151            .ok_or_else(|| {
152                ExpressionTreeServerError::new(
153                    "runtime-contract",
154                    "expr-tree/open did not return a live TreeHandle",
155                )
156            })?;
157        let clock = Arc::clone(&self.clock);
158        tree.set_wall_clock(move || clock.now().ok().map(|time| time.unix_millis()))
159            .map_err(classify_kernel_error)?;
160
161        let mut registry = self.lock_registry()?;
162        expire_idle(&mut registry, self.limits);
163        if registry.sessions.len() >= self.limits.max_sessions {
164            return Err(ExpressionTreeServerError::new(
165                "session-limit",
166                "session capacity changed while opening the tree",
167            ));
168        }
169        registry
170            .sessions
171            .insert(id.clone(), SessionRecord::new(id.clone(), tree, tick));
172        Ok(id)
173    }
174
175    /// Returns the current bounded snapshot for a session.
176    pub fn snapshot(&self, session: &SessionId) -> ServerResult<Expr> {
177        let mut registry = self.lock_registry()?;
178        let tick = begin_request(&mut registry, self.limits);
179        let record = session_mut(&mut registry, session)?;
180        record.last_activity_tick = tick;
181        record.snapshot(self.limits)
182    }
183
184    /// Returns the current optimistic revision.
185    pub fn revision(&self, session: &SessionId) -> ServerResult<u64> {
186        let mut registry = self.lock_registry()?;
187        let tick = begin_request(&mut registry, self.limits);
188        let record = session_mut(&mut registry, session)?;
189        record.last_activity_tick = tick;
190        Ok(record.revision)
191    }
192
193    /// Decodes and commits one standard Intent through the existing
194    /// expression-tree `SurfaceCodec`.
195    pub fn apply_intent(
196        &self,
197        cx: &mut Cx,
198        session: &SessionId,
199        expected_revision: u64,
200        intent: &Expr,
201    ) -> ServerResult<Expr> {
202        let snapshot = self.snapshot(session)?;
203        let current = snapshot_revision(&snapshot)?;
204        if current != expected_revision {
205            return Err(stale(expected_revision, current));
206        }
207        let codec = ExpressionTreeSurfaceCodec::new();
208        let draft = codec
209            .decode(cx, &snapshot, intent)
210            .map_err(classify_kernel_error)?;
211        let operation = codec.commit(cx, &draft).map_err(classify_kernel_error)?;
212        cx.require_all(&operation.required_capabilities)
213            .map_err(classify_kernel_error)?;
214        self.commit_surface_operation(cx, session, Some(expected_revision), &operation.form)
215    }
216
217    /// Closes and removes one authoritative session.
218    pub fn close_session(&self, session: &SessionId) -> ServerResult<bool> {
219        let mut registry = self.lock_registry()?;
220        begin_request(&mut registry, self.limits);
221        if registry.in_flight.contains(session) {
222            return Err(session_busy());
223        }
224        Ok(registry.sessions.remove(session).is_some())
225    }
226
227    /// Subscribes one bounded independent watch.
228    pub fn subscribe(&self, session: &SessionId) -> ServerResult<WatchId> {
229        let mut registry = self.lock_registry()?;
230        let tick = begin_request(&mut registry, self.limits);
231        let record = session_mut(&mut registry, session)?;
232        record.last_activity_tick = tick;
233        record.subscribe(self.limits)
234    }
235
236    /// Drains at most `limit` changes from one watch.
237    pub fn poll_watch(
238        &self,
239        session: &SessionId,
240        watch: &WatchId,
241        limit: usize,
242    ) -> ServerResult<WatchBatch> {
243        let mut registry = self.lock_registry()?;
244        let tick = begin_request(&mut registry, self.limits);
245        let record = session_mut(&mut registry, session)?;
246        record.last_activity_tick = tick;
247        record.poll_watch(watch, limit.min(self.limits.watch_capacity))
248    }
249
250    /// Cancels a watch idempotently with respect to future event delivery.
251    pub fn cancel_watch(&self, session: &SessionId, watch: &WatchId) -> ServerResult<()> {
252        let mut registry = self.lock_registry()?;
253        let tick = begin_request(&mut registry, self.limits);
254        let record = session_mut(&mut registry, session)?;
255        record.last_activity_tick = tick;
256        record.cancel_watch(watch)
257    }
258
259    /// Advances the server's mandatory logical lifecycle clock and expires idle
260    /// sessions. No wall-clock value participates in the comparison.
261    pub fn maintenance_tick(&self, steps: u64) -> ServerResult<usize> {
262        let mut registry = self.lock_registry()?;
263        registry.next_tick = registry.next_tick.saturating_add(steps);
264        let before = registry.sessions.len();
265        expire_idle(&mut registry, self.limits);
266        Ok(before.saturating_sub(registry.sessions.len()))
267    }
268
269    fn wall_observation(&self) -> Option<u64> {
270        self.clock.now().ok().map(|time| time.unix_millis())
271    }
272
273    fn lock_registry(&self) -> ServerResult<MutexGuard<'_, Registry>> {
274        self.registry.lock().map_err(internal)
275    }
276
277    #[cfg(test)]
278    pub(crate) fn registry_is_unlocked_for_test(&self) -> bool {
279        self.registry.try_lock().is_ok()
280    }
281}
282
283impl Default for ExpressionTreeServer {
284    fn default() -> Self {
285        Self::local()
286    }
287}
288
289fn begin_request(registry: &mut Registry, limits: ExpressionTreeServerLimits) -> u64 {
290    let tick = registry.next_tick;
291    registry.next_tick = registry.next_tick.saturating_add(1);
292    expire_idle(registry, limits);
293    tick
294}
295
296fn expire_idle(registry: &mut Registry, limits: ExpressionTreeServerLimits) {
297    let now = registry.next_tick;
298    let Registry {
299        sessions,
300        in_flight,
301        ..
302    } = registry;
303    sessions.retain(|id, session| {
304        in_flight.contains(id)
305            || now.saturating_sub(session.last_activity_tick) <= limits.max_idle_ticks
306    });
307}
308
309fn session_mut<'a>(
310    registry: &'a mut Registry,
311    session: &SessionId,
312) -> ServerResult<&'a mut SessionRecord> {
313    if registry.in_flight.contains(session) {
314        return Err(session_busy());
315    }
316    reserved_session_mut(registry, session)
317}
318
319fn reserved_session_mut<'a>(
320    registry: &'a mut Registry,
321    session: &SessionId,
322) -> ServerResult<&'a mut SessionRecord> {
323    registry.sessions.get_mut(session).ok_or_else(|| {
324        ExpressionTreeServerError::new(
325            "unknown-session",
326            "session is absent, expired, cancelled, or belongs to another server",
327        )
328    })
329}
330
331fn session_busy() -> ExpressionTreeServerError {
332    ExpressionTreeServerError::new(
333        "session-busy",
334        "another operation is already evaluating for this session",
335    )
336}
337
338fn execute_runtime(cx: &mut Cx, target: &RuntimeTarget, operation: &Expr) -> ServerResult<Value> {
339    validate_runtime_target(target, operation)?;
340    let tree = cx
341        .factory()
342        .opaque(Arc::new(target.tree.clone()))
343        .map_err(classify_kernel_error)?;
344    let mut env = Env::child(Arc::new(cx.env().clone()));
345    env.define(target.resource.clone(), tree);
346    cx.with_env(env, |cx| cx.eval_expr(operation.clone()))
347        .map_err(classify_kernel_error)
348}
349
350fn validate_runtime_target(target: &RuntimeTarget, operation: &Expr) -> ServerResult<()> {
351    let Expr::Call { operator, args } = operation else {
352        return Err(ExpressionTreeServerError::new(
353            "invalid-operation",
354            "surface operation must be a local map or expression-tree call",
355        ));
356    };
357    let Expr::Symbol(operator) = operator.as_ref() else {
358        return Err(ExpressionTreeServerError::new(
359            "invalid-operation",
360            "runtime operation must name an expression-tree function",
361        ));
362    };
363    if operator.namespace.as_deref() != Some("expr-tree") {
364        return Err(ExpressionTreeServerError::new(
365            "invalid-operation",
366            "runtime operation is outside the expression-tree family",
367        ));
368    }
369    if !matches!(args.first(), Some(Expr::Symbol(resource)) if resource == &target.resource) {
370        return Err(ExpressionTreeServerError::new(
371            "session-mismatch",
372            "runtime operation targets another expression-tree session",
373        ));
374    }
375    Ok(())
376}
377
378fn operation_metadata(operation: &Expr) -> ServerResult<(String, Option<String>)> {
379    if let Some(op) = protocol::operation(operation) {
380        let path = access::field_str(operation, "path").map(str::to_owned);
381        return Ok((op.name.to_string(), path));
382    }
383    let Expr::Call { operator, args } = operation else {
384        return Err(ExpressionTreeServerError::new(
385            "invalid-operation",
386            "operation is neither a map nor a call",
387        ));
388    };
389    let Expr::Symbol(operator) = operator.as_ref() else {
390        return Err(ExpressionTreeServerError::new(
391            "invalid-operation",
392            "operation call has a non-symbol operator",
393        ));
394    };
395    let path = args.get(1).and_then(|arg| match arg {
396        Expr::String(path) => Some(path.clone()),
397        _ => None,
398    });
399    Ok((operator.name.to_string(), path))
400}
401
402fn is_surface_local(operation: &Expr) -> bool {
403    protocol::operation(operation)
404        .is_some_and(|op| op.namespace.as_deref() == Some("expr-tree-view"))
405}
406
407fn is_revision_change(kind: &str) -> bool {
408    !matches!(kind, "ref" | "list" | "status" | "explain" | "open-policy")
409}
410
411fn target_session(expr: &Expr) -> Option<SessionId> {
412    let Expr::Call { operator, args } = expr else {
413        return None;
414    };
415    let Expr::Symbol(operator) = operator.as_ref() else {
416        return None;
417    };
418    if operator.namespace.as_deref() != Some("expr-tree") {
419        return None;
420    }
421    match args.first() {
422        Some(Expr::Symbol(resource)) => SessionId::from_resource(resource),
423        _ => None,
424    }
425}
426
427fn snapshot_revision(snapshot: &Expr) -> ServerResult<u64> {
428    protocol::uint(snapshot, "revision").map_err(|_| {
429        ExpressionTreeServerError::new(
430            "invalid-expected-revision",
431            "expected-current is not an expression-tree snapshot",
432        )
433    })
434}
435
436fn stale(expected: u64, current: u64) -> ExpressionTreeServerError {
437    ExpressionTreeServerError::new(
438        "stale-revision",
439        format!("expected revision {expected}, current revision is {current}"),
440    )
441}
442
443fn classify_kernel_error(error: Error) -> ExpressionTreeServerError {
444    match error {
445        Error::CapabilityDenied { capability } => ExpressionTreeServerError::new(
446            "authority-denied",
447            format!("caller lacks capability {capability}"),
448        ),
449        Error::TrustDenied { capability, .. } => ExpressionTreeServerError::new(
450            "trust-denied",
451            format!("caller trust does not permit capability {capability}"),
452        ),
453        other => ExpressionTreeServerError::new("operation-failed", other.to_string()),
454    }
455}
sim_lib_expr_tree_server/server.rs

sim_lib_expr_tree_server/
server.rs
