sim_relation_site/
site.rs

1use std::{
2    sync::Arc,
3    time::{Duration, Instant},
4};
5
6use sim_kernel::{
7    Cx, Datum, DatumStore, Effect, Ref, Symbol,
8    effect::{effect_abort_op_key, effect_resume_op_key, resolve_effect},
9};
10use sim_relation_core::{Row, RowType, ToRelationDatum};
11use sim_relation_migrate::CheckedProgram;
12use sim_relation_plan::{CheckedMutation, CheckedQuery};
13
14use crate::{
15    Bindings, Driver, Limits, Operation, ProviderStats, Receipt, RelationPlacement, RowSink,
16    Session, SiteError, Transaction,
17};
18
19/// Provider-neutral bounded site.
20pub struct RelationSite {
21    pub(crate) placement: RelationPlacement,
22    pub(crate) driver: Arc<dyn Driver>,
23}
24impl RelationSite {
25    /// Binds a loaded placement to its driver.
26    pub fn new(placement: RelationPlacement, driver: Arc<dyn Driver>) -> Self {
27        Self { placement, driver }
28    }
29
30    /// Execute a checked query.
31    pub fn query(
32        &self,
33        cx: &mut Cx,
34        plan: &CheckedQuery,
35        bindings: &Bindings,
36        limits: Limits,
37        sink: &mut dyn RowSink,
38    ) -> Result<Receipt, SiteError> {
39        self.rows_effect(
40            cx,
41            Operation::Read,
42            RowsSpec {
43                id: content_id_text(plan.plan_id().content_id()),
44                expected: plan.output(),
45            },
46            limits,
47            sink,
48            |session, bounded| session.query(plan, bindings, &limits, bounded),
49        )
50    }
51    /// Execute a checked mutation.
52    pub fn mutate(
53        &self,
54        cx: &mut Cx,
55        plan: &CheckedMutation,
56        bindings: &Bindings,
57        limits: Limits,
58        sink: &mut dyn RowSink,
59    ) -> Result<Receipt, SiteError> {
60        self.rows_effect(
61            cx,
62            Operation::Write,
63            RowsSpec {
64                id: content_id_text(plan.plan_id().content_id()),
65                expected: plan.output(),
66            },
67            limits,
68            sink,
69            |session, bounded| session.mutate(plan, bindings, &limits, bounded),
70        )
71    }
72    /// Apply a checked migration.
73    pub fn migrate(
74        &self,
75        cx: &mut Cx,
76        program: &CheckedProgram,
77        limits: Limits,
78    ) -> Result<Receipt, SiteError> {
79        self.simple_effect(
80            cx,
81            Operation::Migrate,
82            "checked-migration".into(),
83            limits,
84            |s| s.migrate(program, &limits),
85        )
86    }
87    /// Apply an admitted schema program under the schema capability.
88    pub fn schema(
89        &self,
90        cx: &mut Cx,
91        program: &CheckedProgram,
92        limits: Limits,
93    ) -> Result<Receipt, SiteError> {
94        self.simple_effect(
95            cx,
96            Operation::Schema,
97            "checked-schema".into(),
98            limits,
99            |s| s.schema(program, &limits),
100        )
101    }
102    /// Attach a provider locator under the attach capability.
103    pub fn attach(
104        &self,
105        cx: &mut Cx,
106        locator: &Datum,
107        limits: Limits,
108    ) -> Result<Receipt, SiteError> {
109        self.simple_effect(cx, Operation::Attach, "attach".into(), limits, |s| {
110            s.attach(locator, &limits)
111        })
112    }
113    /// Run a closure-managed transaction. Provider unwind must roll back.
114    pub fn transaction(
115        &self,
116        cx: &mut Cx,
117        limits: Limits,
118        mut body: impl FnMut(&mut dyn Transaction) -> Result<(), SiteError>,
119    ) -> Result<Receipt, SiteError> {
120        self.simple_effect(
121            cx,
122            Operation::Transaction,
123            "transaction".into(),
124            limits,
125            |s| {
126                s.transaction(&mut body)?;
127                Ok(ProviderStats::default())
128            },
129        )
130    }
131
132    fn connect(&self, limits: &Limits) -> Result<Box<dyn Session>, SiteError> {
133        self.driver
134            .connect(self.placement.locator(), limits)
135            .map_err(|_| SiteError::Locator)
136    }
137    fn simple_effect(
138        &self,
139        cx: &mut Cx,
140        op: Operation,
141        id: String,
142        limits: Limits,
143        perform: impl FnOnce(&mut dyn Session) -> Result<ProviderStats, SiteError>,
144    ) -> Result<Receipt, SiteError> {
145        let mut stats = None;
146        self.effect(cx, op, |cx| {
147            let mut session = self.connect(&limits)?;
148            let started = Instant::now();
149            let got = perform(session.as_mut())?;
150            enforce_deadline(&limits, started.elapsed())?;
151            enforce_work(&limits, got.work)?;
152            stats = Some(got);
153            cx.datum_store_mut()
154                .intern(Datum::Symbol(Symbol::new("ok")))
155                .map(Ref::Content)
156                .map_err(kernel)
157        })?;
158        let got = stats.unwrap_or_default();
159        Ok(Receipt {
160            operation_id: id,
161            operation: op,
162            rows: 0,
163            cells: 0,
164            bytes: 0,
165            work: got.work,
166            affected: got.affected,
167        })
168    }
169    fn rows_effect(
170        &self,
171        cx: &mut Cx,
172        op: Operation,
173        spec: RowsSpec<'_>,
174        limits: Limits,
175        sink: &mut dyn RowSink,
176        perform: impl FnOnce(&mut dyn Session, &mut dyn RowSink) -> Result<ProviderStats, SiteError>,
177    ) -> Result<Receipt, SiteError> {
178        let mut counts = Counts::default();
179        let mut stats = None;
180        self.effect(cx, op, |cx| {
181            let mut session = self.connect(&limits)?;
182            let mut bounded = BoundedSink {
183                expected: spec.expected,
184                limits: &limits,
185                inner: sink,
186                counts: &mut counts,
187            };
188            let started = Instant::now();
189            let got = perform(session.as_mut(), &mut bounded)?;
190            enforce_deadline(&limits, started.elapsed())?;
191            enforce_work(&limits, got.work)?;
192            stats = Some(got);
193            cx.datum_store_mut()
194                .intern(Datum::Symbol(Symbol::new("ok")))
195                .map(Ref::Content)
196                .map_err(kernel)
197        })?;
198        let got = stats.unwrap_or_default();
199        Ok(Receipt {
200            operation_id: spec.id,
201            operation: op,
202            rows: counts.rows,
203            cells: counts.cells,
204            bytes: counts.bytes,
205            work: got.work,
206            affected: got.affected,
207        })
208    }
209    pub(crate) fn effect(
210        &self,
211        cx: &mut Cx,
212        op: Operation,
213        perform: impl FnOnce(&mut Cx) -> Result<Ref, SiteError>,
214    ) -> Result<(), SiteError> {
215        let input = cx
216            .datum_store_mut()
217            .intern(Datum::Node {
218                tag: Symbol::qualified("relation", "request"),
219                fields: vec![(
220                    Symbol::new("site"),
221                    Datum::Symbol(self.placement.site.clone()),
222                )],
223            })
224            .map(Ref::Content)
225            .map_err(kernel)?;
226        let effect = Effect::new(
227            cx.fresh_handle(),
228            op.effect(),
229            Ref::Symbol(self.placement.site.clone()),
230            input,
231            Ref::Symbol(Symbol::qualified("core", "Any")),
232            effect_resume_op_key(),
233            effect_abort_op_key(),
234        )
235        .requiring(op.capability());
236        let mut operation_error = None;
237        if let Err(error) = resolve_effect(cx, effect, |cx, _| match perform(cx) {
238            Ok(value) => Ok(value),
239            Err(error) => {
240                operation_error = Some(error.clone());
241                Err(sim_kernel::Error::Eval(error.to_string()))
242            }
243        }) {
244            return Err(operation_error.unwrap_or_else(|| kernel(error)));
245        }
246        Ok(())
247    }
248}
249
250struct RowsSpec<'a> {
251    id: String,
252    expected: &'a RowType,
253}
254
255#[derive(Default)]
256pub(crate) struct Counts {
257    rows: u64,
258    cells: u64,
259    bytes: u64,
260}
261pub(crate) struct BoundedSink<'a> {
262    pub(crate) expected: &'a RowType,
263    pub(crate) limits: &'a Limits,
264    pub(crate) inner: &'a mut dyn RowSink,
265    pub(crate) counts: &'a mut Counts,
266}
267impl RowSink for BoundedSink<'_> {
268    fn push(&mut self, row: Row) -> Result<(), SiteError> {
269        if row.row_type() != self.expected {
270            return Err(SiteError::RowType);
271        }
272        let cells = row.cells().len() as u64;
273        let bytes = format!("{:?}", row.to_datum()).len() as u64;
274        let rows = self
275            .counts
276            .rows
277            .checked_add(1)
278            .ok_or(SiteError::Limit(crate::LimitKind::Rows))?;
279        let cells = self
280            .counts
281            .cells
282            .checked_add(cells)
283            .ok_or(SiteError::Limit(crate::LimitKind::Cells))?;
284        let bytes = self
285            .counts
286            .bytes
287            .checked_add(bytes)
288            .ok_or(SiteError::Limit(crate::LimitKind::Bytes))?;
289        if rows > self.limits.rows {
290            return Err(SiteError::Limit(crate::LimitKind::Rows));
291        }
292        if cells > self.limits.cells {
293            return Err(SiteError::Limit(crate::LimitKind::Cells));
294        }
295        if bytes > self.limits.bytes {
296            return Err(SiteError::Limit(crate::LimitKind::Bytes));
297        }
298        self.inner.push(row)?;
299        self.counts.rows = rows;
300        self.counts.cells = cells;
301        self.counts.bytes = bytes;
302        Ok(())
303    }
304}
305pub(crate) fn enforce_work(l: &Limits, work: u64) -> Result<(), SiteError> {
306    if work > l.work {
307        Err(SiteError::Limit(crate::LimitKind::Work))
308    } else {
309        Ok(())
310    }
311}
312fn enforce_deadline(limits: &Limits, elapsed: Duration) -> Result<(), SiteError> {
313    if limits.deadline.is_some_and(|deadline| elapsed > deadline) {
314        Err(SiteError::Limit(crate::LimitKind::Deadline))
315    } else {
316        Ok(())
317    }
318}
319fn content_id_text(id: &sim_kernel::ContentId) -> String {
320    let digest: String = id.bytes.iter().map(|byte| format!("{byte:02x}")).collect();
321    format!("{}:{digest}", id.algorithm)
322}
323pub(crate) fn kernel(e: sim_kernel::Error) -> SiteError {
324    SiteError::Kernel(e.to_string())
325}
sim_relation_site/site.rs

sim_relation_site/
site.rs
