tilo_app/framework/
query.rs

1//! Reading another provider's action result, kept current.
2//!
3//! [`RecordEnv::query`](super::RecordEnv::query) invokes a view action of
4//! any provider through the session's action door, as the person using the
5//! app, and shows its answer as data that arrives
6//! ([`Arriving`](ui_core::arriving::Arriving)): the view says what shows
7//! once it is there ([`ViewCx::arriving`](super::ViewCx::arriving)), and the
8//! framework shows the placeholder until then. A live action
9//! ([`tilo_actions::ExecutionMode::Live`]) sends a new result whenever its
10//! provider's answer changes, and the answer follows; an action that
11//! answers once is read once. The door decides the read as it decides any
12//! invocation: a caller without the permission gets
13//! [`ReadFailure::Denied`], never the data.
14
15use std::cell::RefCell;
16use std::collections::HashMap;
17use std::rc::{Rc, Weak};
18use std::sync::Arc;
19use std::sync::atomic::{AtomicU64, Ordering};
20
21use serde::de::DeserializeOwned;
22use tilo_actions::InvocationUpdate;
23use tilo_actions::remote::{ActionHostError, InvokeArgument, RemoteActionTarget};
24use ui_core::Followed;
25use ui_core::arriving::{Arrival, Arriving, Source};
26
27/// Why a [`Query`] has no answer to show: the caller may not read it, or
28/// the read failed. Either is the read's failure ([`Arrival::Failed`]).
29#[derive(Clone, Debug, PartialEq, Eq)]
30pub enum ReadFailure {
31    /// The caller may not read it; why.
32    Denied(String),
33    /// The read failed; why.
34    Failed(String),
35}
36
37impl std::fmt::Display for ReadFailure {
38    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        match self {
40            Self::Denied(why) | Self::Failed(why) => f.write_str(why),
41        }
42    }
43}
44
45/// The action denial codes a provider fails a read with.
46const DENIED: &[&str] = &["not_permitted", "permission_denied"];
47
48/// What `call`'s latest update says about the read; `None` while it has
49/// nothing new to show.
50fn answered<T: DeserializeOwned>(
51    call: &Arrival<InvocationUpdate, ActionHostError>,
52) -> Option<Result<T, ReadFailure>> {
53    let decode = |value: &serde_json::Value| {
54        T::deserialize(value)
55            .map_err(|error| ReadFailure::Failed(format!("the answer was not understood: {error}")))
56    };
57    match call {
58        Arrival::Pending { .. }
59        | Arrival::Ready(
60            InvocationUpdate::Accepted { .. }
61            | InvocationUpdate::Progress { .. }
62            | InvocationUpdate::Cancelled { .. },
63        ) => None,
64        Arrival::Ready(InvocationUpdate::Item { value, .. }) => Some(decode(value)),
65        Arrival::Ready(InvocationUpdate::Succeeded { output, .. }) => Some(decode(
66            output.value.as_ref().unwrap_or(&serde_json::Value::Null),
67        )),
68        Arrival::Ready(InvocationUpdate::Failed { error, .. }) => {
69            Some(Err(if DENIED.contains(&error.code.as_str()) {
70                ReadFailure::Denied(error.message.clone())
71            } else {
72                ReadFailure::Failed(error.message.clone())
73            }))
74        }
75        Arrival::Failed(ActionHostError::NotPermitted) => Some(Err(ReadFailure::Denied(
76            ActionHostError::NotPermitted.to_string(),
77        ))),
78        Arrival::Failed(error) => Some(Err(ReadFailure::Failed(error.to_string()))),
79    }
80}
81
82/// One read [`RecordEnv::query`](super::RecordEnv::query) follows. Clones
83/// share it. It is followed while the view scope that asked for it lives,
84/// or a clone of it does; then it is cancelled.
85pub struct Query<T> {
86    answer: Arriving<T, ReadFailure>,
87    inner: Rc<Inner>,
88}
89
90impl<T> Clone for Query<T> {
91    fn clone(&self) -> Self {
92        Self {
93            answer: self.answer.clone(),
94            inner: Rc::clone(&self.inner),
95        }
96    }
97}
98
99struct Inner {
100    /// The door it reads through; none for a read that reaches no
101    /// provider ([`Query::fixed`]).
102    opener: Option<super::reference::Opener>,
103    target: RemoteActionTarget,
104    /// The argument it reads with now: what a retry reads again.
105    argument: RefCell<InvokeArgument>,
106    /// Which read's answers count: each [`Query::set_argument`] starts the
107    /// next, and an answer the one before sends late is not shown.
108    generation: Arc<AtomicU64>,
109    followed: RefCell<Option<Followed>>,
110    /// The answer it lands in; none for a read that reaches no provider.
111    answer: Option<Box<dyn Landing>>,
112}
113
114/// What lands each update of a followed call in its answer.
115type Lander = Box<dyn Fn(&Arrival<InvocationUpdate, ActionHostError>) + Send>;
116
117/// What following a read does with its answer, whatever the answer's type:
118/// the following is not generic, so the runtime holds it once for every
119/// app and every answer type.
120trait Landing {
121    /// What lands each update of the call in the answer, decoded.
122    fn lander(&self) -> Lander;
123    /// Whether an answer shows.
124    fn shown(&self) -> bool;
125    /// The answer is on its way from `from`.
126    fn begin(&self, from: Source);
127}
128
129impl<T: DeserializeOwned + Send + Sync + 'static> Landing for Arriving<T, ReadFailure> {
130    fn lander(&self) -> Lander {
131        let writer = self.writer();
132        Box::new(move |call| {
133            if let Some(answered) = answered::<T>(call) {
134                writer.finish(answered);
135            }
136        })
137    }
138
139    fn shown(&self) -> bool {
140        self.with(|read| read.ready().is_some())
141    }
142
143    fn begin(&self, from: Source) {
144        // ignored-failure: the restart's completion is the query's own; its lander publishes the answer
145        let _ = Arriving::begin(self, from);
146    }
147}
148
149thread_local! {
150    /// The queries of this thread a retry can reach, by their number: a
151    /// retry is asked through the read's handle, which may cross threads,
152    /// and the query it reads again lives on this one.
153    static QUERIES: RefCell<HashMap<u64, Weak<Inner>>> = RefCell::new(HashMap::new());
154}
155
156static NEXT_QUERY: AtomicU64 = AtomicU64::new(1);
157
158/// Read query `number` again, if it still lives on this thread.
159fn asked_again(number: u64) {
160    let inner = QUERIES.with(|queries| queries.borrow().get(&number).and_then(Weak::upgrade));
161    if let Some(inner) = inner {
162        follow(&inner, true);
163    }
164}
165
166/// Follow a query numbered `number` through `opener` while `alive` lives,
167/// landing its answers in `answer`.
168fn following(
169    opener: &super::reference::Opener,
170    alive: &Weak<()>,
171    target: RemoteActionTarget,
172    argument: InvokeArgument,
173    number: u64,
174    answer: Box<dyn Landing>,
175) -> Rc<Inner> {
176    let inner = Rc::new(Inner {
177        opener: Some(opener.clone()),
178        target,
179        argument: RefCell::new(argument),
180        generation: Arc::new(AtomicU64::new(0)),
181        followed: RefCell::new(None),
182        answer: Some(answer),
183    });
184    QUERIES.with(|queries| {
185        let mut queries = queries.borrow_mut();
186        queries.retain(|_, query| query.strong_count() > 0);
187        queries.insert(number, Rc::downgrade(&inner));
188    });
189    follow(&inner, false);
190    opener.keep(alive.clone(), Rc::clone(&inner) as Rc<dyn std::any::Any>);
191    inner
192}
193
194impl<T: DeserializeOwned + Send + Sync + 'static> Query<T> {
195    pub(super) fn start(
196        opener: &super::reference::Opener,
197        alive: &Weak<()>,
198        target: RemoteActionTarget,
199        argument: InvokeArgument,
200    ) -> Self {
201        let number = NEXT_QUERY.fetch_add(1, Ordering::Relaxed);
202        let answer = Arriving::pending(Source::Local).on_retry(move |_| asked_again(number));
203        let landing = Box::new(answer.clone());
204        let inner = following(opener, alive, target, argument, number, landing);
205        Self { answer, inner }
206    }
207
208    /// A read that reaches no provider and holds `answer` (a capture's
209    /// stand-in for the action's result), or nothing.
210    pub(super) fn fixed(answer: Option<&serde_json::Value>) -> Self {
211        let answer = match answer {
212            Some(value) => {
213                let read = Arriving::pending(Source::Local);
214                read.writer().finish(T::deserialize(value).map_err(|error| {
215                    ReadFailure::Failed(format!("the answer was not understood: {error}"))
216                }));
217                read
218            }
219            None => Arriving::pending(Source::Local),
220        };
221        Self {
222            answer,
223            inner: Rc::new(Inner {
224                opener: None,
225                target: RemoteActionTarget {
226                    provider: String::new(),
227                    address: tilo_actions::ActionAddress::new("", ""),
228                },
229                argument: RefCell::new(InvokeArgument::None),
230                generation: Arc::new(AtomicU64::new(0)),
231                followed: RefCell::new(None),
232                answer: None,
233            }),
234        }
235    }
236
237    /// Read again with `argument` in place of the last one (a filter
238    /// changed). The answer shows the last result until the new one lands.
239    pub fn set_argument(&self, argument: InvokeArgument) {
240        if self.inner.opener.is_none() {
241            return;
242        }
243        *self.inner.argument.borrow_mut() = argument;
244        follow(&self.inner, false);
245    }
246
247    /// What the read has answered, kept current: build what shows once it
248    /// is there with [`ViewCx::arriving`](super::ViewCx::arriving).
249    pub fn answer(&self) -> Arriving<T, ReadFailure> {
250        self.answer.clone()
251    }
252}
253
254/// Follow `inner`'s read with its current argument, landing it in its
255/// answer; the read it replaces is cancelled as it drops. A read asked
256/// again (`again`) shows its placeholder until it answers; a new argument
257/// keeps the last result shown until the new one lands.
258fn follow(inner: &Inner, again: bool) {
259    let (Some(opener), Some(answer)) = (&inner.opener, &inner.answer) else {
260        return;
261    };
262    let generation = inner
263        .generation
264        .fetch_add(1, Ordering::AcqRel)
265        .wrapping_add(1);
266    let current = Arc::clone(&inner.generation);
267    let land = answer.lander();
268    let followed = opener.follow(
269        inner.target.clone(),
270        inner.argument.borrow().clone(),
271        move |call: &Arrival<InvocationUpdate, ActionHostError>| {
272            if current.load(Ordering::Acquire) == generation {
273                land(call);
274            }
275        },
276    );
277    // Where the answer comes from is where the door says the call goes.
278    let from = followed
279        .call()
280        .with(|call| call.source().cloned())
281        .unwrap_or(Source::Local);
282    if again || !answer.shown() {
283        answer.begin(from);
284    }
285    inner.followed.replace(Some(followed));
286}
287
288#[cfg(test)]
289#[path = "query_tests.rs"]
290mod tests;
tilo_app/framework/query.rs

tilo_app/framework/
query.rs
