sofie_db/
messages.rs

1use std::time::Duration;
2
3use serde_json::{Value, json};
4use sofie_core::variants::{self, Variants};
5use sofie_core::{Cast, ColumnReader, Error, Result, SqlValue, Timestamp, Uuid};
6use sofie_telemetry::TraceParent;
7
8use crate::heard::{Known, Listeners};
9use crate::{BoxFuture, Db, Dialect, Stamp, Step, Wakeups, query_decoded, query_values};
10
11/// Every insert binds the message columns first, as `$1` to `$7`: id, kind,
12/// name, payload, `available_at`, traceparent, actor.
13pub struct MessageSql {
14    pub insert: &'static str,
15    /// Message columns, then a key; skipped while a job of that name waits for it.
16    pub fan_out: &'static str,
17    /// `(events, jobs, available_at, actor)`: events `{id, payload, traceparent}`, jobs
18    /// `{id, name, event}`.
19    pub fan_out_events: &'static str,
20    /// `(lease_milliseconds, now, limit, mine, held_until)`, counting attempts past the last;
21    /// `mine` names by kind, and any other waits until `held_until`.
22    pub claim: &'static str,
23    /// `(id, attempts, lease_milliseconds)`: extends a claim still held, on the database's clock.
24    pub renew: &'static str,
25    pub complete: &'static str,
26    pub retry: &'static str,
27    /// Clears the key, so a dead job holds back no new one for it.
28    pub bury: &'static str,
29    /// `(id, attempts, name)`: turns an event its lone listener ran into that listener's job.
30    pub to_job: &'static str,
31    pub add_schedule: &'static str,
32    /// Message columns, then the next time; returns it and whether this call queued a run (none
33    /// while one waits or retries), no row when the schedule does not exist.
34    pub fire_schedule: &'static str,
35    /// Sent after a commit that queued messages, never within it: Postgres
36    /// serializes the commits of notifying transactions on one lock.
37    pub notify: Option<&'static str>,
38}
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum MessageKind {
42    Event,
43    Job,
44}
45
46impl Variants for MessageKind {
47    const TYPE: &'static str = "message kind";
48    const VARIANTS: &'static [(&'static str, Self)] = &[("event", Self::Event), ("job", Self::Job)];
49}
50
51/// Its `attempt` fences every later write, so an expired lease can neither finish nor fail it.
52#[derive(Debug)]
53pub struct Claimed {
54    pub id: Uuid,
55    pub kind: MessageKind,
56    pub name: String,
57    pub payload: Value,
58    pub attempt: i64,
59    pub traceparent: Option<TraceParent>,
60    pub available_at: Timestamp,
61    pub actor: Option<Uuid>,
62}
63
64impl Claimed {
65    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
66        Ok(Self {
67            id: row.uuid()?,
68            kind: variants::decode(row)?,
69            name: row.text()?.to_owned(),
70            payload: row.json()?,
71            attempt: row.int()?,
72            traceparent: traceparent(row)?,
73            available_at: row.timestamp()?,
74            actor: Option::<Uuid>::decode(row)?,
75        })
76    }
77
78    fn fenced(&self, more: impl IntoIterator<Item = SqlValue>) -> Vec<SqlValue> {
79        let mut params = vec![SqlValue::Uuid(self.id), SqlValue::Int(self.attempt)];
80        params.extend(more);
81        params
82    }
83}
84
85fn traceparent(row: &mut dyn ColumnReader) -> Result<Option<TraceParent>> {
86    if row.is_null()? {
87        row.skip()?;
88        return Ok(None);
89    }
90    Ok(TraceParent::parse(row.text()?))
91}
92
93#[derive(Clone, Copy, Debug, PartialEq, Eq)]
94pub struct Fired {
95    pub fired: bool,
96    pub next_at: Timestamp,
97}
98
99impl Fired {
100    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
101        Ok(Self {
102            next_at: row.timestamp()?,
103            fired: row.bool()?,
104        })
105    }
106}
107
108struct Message<'a> {
109    kind: MessageKind,
110    name: &'a str,
111    payload: Value,
112    available_at: Timestamp,
113    actor: Option<Uuid>,
114}
115
116impl Message<'_> {
117    fn columns(self) -> Result<Vec<SqlValue>> {
118        Ok(vec![
119            SqlValue::Uuid(Uuid::now_v7()?),
120            SqlValue::Text(self.kind.name().to_owned()),
121            SqlValue::Text(self.name.to_owned()),
122            SqlValue::Json(self.payload),
123            SqlValue::Timestamp(self.available_at),
124            current_traceparent(),
125            self.actor.map_or(SqlValue::Null, SqlValue::Uuid),
126        ])
127    }
128}
129
130pub(crate) fn keyed_job(
131    dialect: &dyn Dialect,
132    name: &str,
133    payload: Value,
134    key: Uuid,
135    available_at: Timestamp,
136    actor: Option<Uuid>,
137) -> Result<Step> {
138    let mut params = Message {
139        kind: MessageKind::Job,
140        name,
141        payload,
142        available_at,
143        actor,
144    }
145    .columns()?;
146    params.push(SqlValue::Uuid(key));
147    Ok(Step::statement(dialect.messages().fan_out, params, None))
148}
149
150/// Due times follow the app's clock, so a test can move them; leases follow the database's.
151#[derive(Clone)]
152pub struct Messages {
153    db: Db,
154}
155
156impl Messages {
157    #[must_use]
158    pub fn new(db: Db) -> Self {
159        Self { db }
160    }
161
162    fn sql(&self) -> &'static MessageSql {
163        self.db.driver().dialect().messages()
164    }
165
166    pub async fn claim(&self, lease: Duration, limit: i64) -> Result<Vec<Claimed>> {
167        let now = self.db.clock().now().timestamp();
168        let nothing_known = SqlValue::Json(Known::default().to_json());
169        claim_known(&self.db, lease, limit, &nothing_known, now).await
170    }
171
172    /// A conflict means the claim was lost: another worker took the message after its lease ran out.
173    pub async fn renew(&self, message: &Claimed, lease: Duration) -> Result<()> {
174        let params = message.fenced([milliseconds(lease)?]);
175        let step = Step::statement(self.sql().renew, params, Some(1));
176        self.db.transaction(vec![step]).await
177    }
178
179    /// Committed with its job's outcome, it makes the outcome happen once.
180    #[must_use]
181    pub fn completion(&self, message: &Claimed) -> Step {
182        Step::statement(self.sql().complete, message.fenced([]), Some(1))
183    }
184
185    /// `listener` names the lone listener an event ran as; the message becomes its job.
186    pub async fn retry(
187        &self,
188        message: &Claimed,
189        listener: Option<&str>,
190        error: &str,
191        delay: Duration,
192    ) -> Result<()> {
193        let due = later(self.db.clock().now().timestamp(), delay)?;
194        let due = Some(SqlValue::Timestamp(due));
195        self.release(self.sql().retry, message, listener, error, due)
196            .await
197    }
198
199    pub async fn bury(&self, message: &Claimed, listener: Option<&str>, error: &str) -> Result<()> {
200        self.release(self.sql().bury, message, listener, error, None)
201            .await
202    }
203
204    /// A lone listener's message first becomes its job, so its retries and its death name the
205    /// listener that failed, not the event.
206    async fn release(
207        &self,
208        sql: &'static str,
209        message: &Claimed,
210        listener: Option<&str>,
211        error: &str,
212        due: Option<SqlValue>,
213    ) -> Result<()> {
214        let params = message.fenced(std::iter::once(SqlValue::Text(error.to_owned())).chain(due));
215        let mut steps = Vec::with_capacity(2);
216        if let Some(listener) = listener {
217            let params = message.fenced([SqlValue::Text(listener.to_owned())]);
218            steps.push(Step::statement(self.sql().to_job, params, Some(1)));
219        }
220        steps.push(Step::statement(sql, params, Some(1)));
221        self.db.transaction(steps).await
222    }
223
224    pub async fn add_schedule(
225        &self,
226        name: &str,
227        first_at: Timestamp,
228        cadence: &str,
229        previous_cadence: &str,
230    ) -> Result<()> {
231        let params = [
232            SqlValue::Text(name.to_owned()),
233            SqlValue::Timestamp(first_at),
234            SqlValue::Text(cadence.to_owned()),
235            SqlValue::Text(previous_cadence.to_owned()),
236        ];
237        let sql = self.sql().add_schedule;
238        query_values(self.db.driver(), sql, &params).await.map(drop)
239    }
240
241    /// A run by hand: a running worker keeps the next due time in memory, but takes a job at once.
242    pub async fn queue_schedule(&self, name: &str) -> Result<()> {
243        let dialect = self.db.driver().dialect();
244        let payload = Value::Object(serde_json::Map::new());
245        let now = self.db.clock().now().timestamp();
246        let mut steps = Vec::new();
247        let queuing = Queuing {
248            kind: MessageKind::Job,
249            at: now,
250            actor: None,
251        };
252        queue_steps(dialect, &queuing, [(name, payload)], None, &mut steps)?;
253        // In the same transaction, as the process may end before a notification sent later.
254        if let Some(notify) = dialect.messages().notify {
255            steps.push(Step::Script(notify.to_owned()));
256        }
257        self.db.transaction(steps).await
258    }
259
260    /// Only one of any number of concurrent callers is told it fired.
261    pub async fn fire_schedule(
262        &self,
263        name: &str,
264        now: Timestamp,
265        next_at: Timestamp,
266    ) -> Result<Option<Fired>> {
267        let payload = Value::Object(serde_json::Map::new());
268        let mut params = Message {
269            kind: MessageKind::Job,
270            name,
271            payload,
272            available_at: now,
273            actor: None,
274        }
275        .columns()?;
276        params.push(SqlValue::Timestamp(next_at));
277        let sql = self.sql().fire_schedule;
278        let fired =
279            query_decoded(self.db.driver(), sql, &params, Vec::new(), Fired::decode).await?;
280        Ok(fired.into_iter().next())
281    }
282
283    pub(crate) async fn arrived(&self) {
284        self.db.arrivals().arrived().await;
285    }
286
287    #[must_use]
288    pub fn wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
289        self.db.driver().wakeups()
290    }
291}
292
293pub(crate) async fn claim_known(
294    db: &Db,
295    lease: Duration,
296    limit: i64,
297    mine: &SqlValue,
298    held_until: Timestamp,
299) -> Result<Vec<Claimed>> {
300    let params = [
301        milliseconds(lease)?,
302        SqlValue::Timestamp(db.clock().now().timestamp()),
303        SqlValue::Int(limit),
304        mine.clone(),
305        SqlValue::Timestamp(held_until),
306    ];
307    let sql = db.driver().dialect().messages().claim;
308    query_decoded(db.driver(), sql, &params, Vec::new(), Claimed::decode).await
309}
310
311pub async fn ping_queue(db: &Db) -> Result<()> {
312    let messages = db.driver().dialect().qualified("sofie", "messages");
313    let sql = format!("SELECT 1 FROM {messages} LIMIT 1");
314    query_values(db.driver(), &sql, &[]).await.map(drop)
315}
316
317pub(crate) struct Queuing {
318    kind: MessageKind,
319    at: Timestamp,
320    actor: Option<Uuid>,
321}
322
323impl Stamp {
324    pub(crate) fn queuing(&self, kind: MessageKind, at: Timestamp) -> Queuing {
325        Queuing {
326            kind,
327            at,
328            actor: self.actor,
329        }
330    }
331}
332
333pub(crate) fn queue_steps<'a>(
334    dialect: &dyn Dialect,
335    queuing: &Queuing,
336    messages: impl IntoIterator<Item = (&'a str, Value)>,
337    heard: Option<&Known>,
338    steps: &mut Vec<Step>,
339) -> Result<bool> {
340    if let Some(heard) = heard.filter(|_| queuing.kind == MessageKind::Event) {
341        return fan_out_raised(dialect, queuing, messages, &heard.listeners, steps);
342    }
343    let mut queued = false;
344    for (name, payload) in messages {
345        let params = Message {
346            kind: queuing.kind,
347            name,
348            payload,
349            available_at: queuing.at,
350            actor: queuing.actor,
351        }
352        .columns()?;
353        steps.push(Step::statement(dialect.messages().insert, params, None));
354        queued = true;
355    }
356    Ok(queued)
357}
358
359fn fan_out_raised<'a>(
360    dialect: &dyn Dialect,
361    queuing: &Queuing,
362    events: impl IntoIterator<Item = (&'a str, Value)>,
363    listeners: &Listeners,
364    steps: &mut Vec<Step>,
365) -> Result<bool> {
366    let traceparent = sofie_telemetry::current_traceparent().map(|parent| parent.to_string());
367    let mut raised = Vec::new();
368    for (name, payload) in events {
369        raised.push(Fanned {
370            id: Uuid::now_v7()?,
371            name,
372            payload,
373            traceparent: traceparent.clone(),
374        });
375    }
376    let Some(step) = fan_out_events(dialect, raised, queuing.at, queuing.actor, listeners)? else {
377        return Ok(false);
378    };
379    steps.push(step);
380    Ok(true)
381}
382
383pub(crate) struct Fanned<'a> {
384    pub(crate) id: Uuid,
385    pub(crate) name: &'a str,
386    pub(crate) payload: Value,
387    pub(crate) traceparent: Option<String>,
388}
389
390pub(crate) fn fan_out_events<'a>(
391    dialect: &dyn Dialect,
392    events: impl IntoIterator<Item = Fanned<'a>>,
393    at: Timestamp,
394    actor: Option<Uuid>,
395    listeners: &Listeners,
396) -> Result<Option<Step>> {
397    let (mut taken, mut jobs) = (Vec::new(), Vec::new());
398    for event in events {
399        let taking = listeners.taking(event.name, &event.payload);
400        if taking.is_empty() {
401            continue;
402        }
403        let id = event.id.to_string();
404        for listener in taking.iter() {
405            let job = Uuid::now_v7()?.to_string();
406            jobs.push(json!({ "id": job, "name": listener, "event": id }));
407        }
408        taken.push(json!({ "id": id, "payload": event.payload, "traceparent": event.traceparent }));
409    }
410    if jobs.is_empty() {
411        return Ok(None);
412    }
413    let params = vec![
414        SqlValue::Json(Value::Array(taken)),
415        SqlValue::Json(Value::Array(jobs)),
416        SqlValue::Timestamp(at),
417        actor.map_or(SqlValue::Null, SqlValue::Uuid),
418    ];
419    let sql = dialect.messages().fan_out_events;
420    Ok(Some(Step::statement(sql, params, None)))
421}
422
423pub(crate) fn current_traceparent() -> SqlValue {
424    sofie_telemetry::current_traceparent().map_or(SqlValue::Null, |traceparent| {
425        SqlValue::Text(traceparent.to_string())
426    })
427}
428
429fn milliseconds(lease: Duration) -> Result<SqlValue> {
430    i64::try_from(lease.as_millis())
431        .map(SqlValue::Int)
432        .map_err(|error| Error::internal("a lease too long to store", error))
433}
434
435fn later(now: Timestamp, by: Duration) -> Result<Timestamp> {
436    now.checked_add(by)
437        .map_err(|error| Error::internal("computing a message time", error))
438}
sofie_db/messages.rs

sofie_db/
messages.rs
