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        claim_known(&self.db, lease, limit, &Known::default(), now).await
169    }
170
171    /// A conflict means the claim was lost: another worker took the message after its lease ran out.
172    pub async fn renew(&self, message: &Claimed, lease: Duration) -> Result<()> {
173        let params = message.fenced([milliseconds(lease)?]);
174        let step = Step::statement(self.sql().renew, params, Some(1));
175        self.db.transaction(vec![step]).await
176    }
177
178    /// Committed with its job's outcome, it makes the outcome happen once.
179    #[must_use]
180    pub fn completion(&self, message: &Claimed) -> Step {
181        Step::statement(self.sql().complete, message.fenced([]), Some(1))
182    }
183
184    /// `listener` names the lone listener an event ran as; the message becomes its job.
185    pub async fn retry(
186        &self,
187        message: &Claimed,
188        listener: Option<&str>,
189        error: &str,
190        delay: Duration,
191    ) -> Result<()> {
192        let due = later(self.db.clock().now().timestamp(), delay)?;
193        let due = Some(SqlValue::Timestamp(due));
194        self.release(self.sql().retry, message, listener, error, due)
195            .await
196    }
197
198    pub async fn bury(&self, message: &Claimed, listener: Option<&str>, error: &str) -> Result<()> {
199        self.release(self.sql().bury, message, listener, error, None)
200            .await
201    }
202
203    /// A lone listener's message first becomes its job, so its retries and its death name the
204    /// listener that failed, not the event.
205    async fn release(
206        &self,
207        sql: &'static str,
208        message: &Claimed,
209        listener: Option<&str>,
210        error: &str,
211        due: Option<SqlValue>,
212    ) -> Result<()> {
213        let params = message.fenced(std::iter::once(SqlValue::Text(error.to_owned())).chain(due));
214        let mut steps = Vec::with_capacity(2);
215        if let Some(listener) = listener {
216            let params = message.fenced([SqlValue::Text(listener.to_owned())]);
217            steps.push(Step::statement(self.sql().to_job, params, Some(1)));
218        }
219        steps.push(Step::statement(sql, params, Some(1)));
220        self.db.transaction(steps).await
221    }
222
223    pub async fn add_schedule(
224        &self,
225        name: &str,
226        first_at: Timestamp,
227        cadence: &str,
228        previous_cadence: &str,
229    ) -> Result<()> {
230        let params = [
231            SqlValue::Text(name.to_owned()),
232            SqlValue::Timestamp(first_at),
233            SqlValue::Text(cadence.to_owned()),
234            SqlValue::Text(previous_cadence.to_owned()),
235        ];
236        let sql = self.sql().add_schedule;
237        query_values(self.db.driver(), sql, &params).await.map(drop)
238    }
239
240    /// A run by hand: a running worker keeps the next due time in memory, but takes a job at once.
241    pub async fn queue_schedule(&self, name: &str) -> Result<()> {
242        let dialect = self.db.driver().dialect();
243        let payload = Value::Object(serde_json::Map::new());
244        let now = self.db.clock().now().timestamp();
245        let mut steps = Vec::new();
246        let queuing = Queuing {
247            kind: MessageKind::Job,
248            at: now,
249            actor: None,
250        };
251        queue_steps(dialect, &queuing, [(name, payload)], None, &mut steps)?;
252        // In the same transaction, as the process may end before a notification sent later.
253        if let Some(notify) = dialect.messages().notify {
254            steps.push(Step::Script(notify.to_owned()));
255        }
256        self.db.transaction(steps).await
257    }
258
259    /// Only one of any number of concurrent callers is told it fired.
260    pub async fn fire_schedule(
261        &self,
262        name: &str,
263        now: Timestamp,
264        next_at: Timestamp,
265    ) -> Result<Option<Fired>> {
266        let payload = Value::Object(serde_json::Map::new());
267        let mut params = Message {
268            kind: MessageKind::Job,
269            name,
270            payload,
271            available_at: now,
272            actor: None,
273        }
274        .columns()?;
275        params.push(SqlValue::Timestamp(next_at));
276        let sql = self.sql().fire_schedule;
277        let fired =
278            query_decoded(self.db.driver(), sql, &params, Vec::new(), Fired::decode).await?;
279        Ok(fired.into_iter().next())
280    }
281
282    pub(crate) async fn arrived(&self) {
283        self.db.arrivals().arrived().await;
284    }
285
286    #[must_use]
287    pub fn wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
288        self.db.driver().wakeups()
289    }
290}
291
292pub(crate) async fn claim_known(
293    db: &Db,
294    lease: Duration,
295    limit: i64,
296    mine: &Known,
297    held_until: Timestamp,
298) -> Result<Vec<Claimed>> {
299    let params = [
300        milliseconds(lease)?,
301        SqlValue::Timestamp(db.clock().now().timestamp()),
302        SqlValue::Int(limit),
303        SqlValue::Json(mine.to_json()),
304        SqlValue::Timestamp(held_until),
305    ];
306    let sql = db.driver().dialect().messages().claim;
307    query_decoded(db.driver(), sql, &params, Vec::new(), Claimed::decode).await
308}
309
310pub async fn ping_queue(db: &Db) -> Result<()> {
311    let messages = db.driver().dialect().qualified("sofie", "messages");
312    let sql = format!("SELECT 1 FROM {messages} LIMIT 1");
313    query_values(db.driver(), &sql, &[]).await.map(drop)
314}
315
316pub(crate) struct Queuing {
317    kind: MessageKind,
318    at: Timestamp,
319    actor: Option<Uuid>,
320}
321
322impl Stamp {
323    pub(crate) fn queuing(&self, kind: MessageKind, at: Timestamp) -> Queuing {
324        Queuing {
325            kind,
326            at,
327            actor: self.actor,
328        }
329    }
330}
331
332pub(crate) fn queue_steps<'a>(
333    dialect: &dyn Dialect,
334    queuing: &Queuing,
335    messages: impl IntoIterator<Item = (&'a str, Value)>,
336    heard: Option<&Known>,
337    steps: &mut Vec<Step>,
338) -> Result<bool> {
339    if let Some(heard) = heard.filter(|_| queuing.kind == MessageKind::Event) {
340        return fan_out_raised(dialect, queuing, messages, &heard.listeners, steps);
341    }
342    let mut queued = false;
343    for (name, payload) in messages {
344        let params = Message {
345            kind: queuing.kind,
346            name,
347            payload,
348            available_at: queuing.at,
349            actor: queuing.actor,
350        }
351        .columns()?;
352        steps.push(Step::statement(dialect.messages().insert, params, None));
353        queued = true;
354    }
355    Ok(queued)
356}
357
358fn fan_out_raised<'a>(
359    dialect: &dyn Dialect,
360    queuing: &Queuing,
361    events: impl IntoIterator<Item = (&'a str, Value)>,
362    listeners: &Listeners,
363    steps: &mut Vec<Step>,
364) -> Result<bool> {
365    let traceparent = sofie_telemetry::current_traceparent().map(|parent| parent.to_string());
366    let mut raised = Vec::new();
367    for (name, payload) in events {
368        raised.push(Fanned {
369            id: Uuid::now_v7()?,
370            name,
371            payload,
372            traceparent: traceparent.clone(),
373        });
374    }
375    let Some(step) = fan_out_events(dialect, raised, queuing.at, queuing.actor, listeners)? else {
376        return Ok(false);
377    };
378    steps.push(step);
379    Ok(true)
380}
381
382pub(crate) struct Fanned<'a> {
383    pub(crate) id: Uuid,
384    pub(crate) name: &'a str,
385    pub(crate) payload: Value,
386    pub(crate) traceparent: Option<String>,
387}
388
389pub(crate) fn fan_out_events<'a>(
390    dialect: &dyn Dialect,
391    events: impl IntoIterator<Item = Fanned<'a>>,
392    at: Timestamp,
393    actor: Option<Uuid>,
394    listeners: &Listeners,
395) -> Result<Option<Step>> {
396    let (mut taken, mut jobs) = (Vec::new(), Vec::new());
397    for event in events {
398        let taking = listeners.taking(event.name, &event.payload);
399        if taking.is_empty() {
400            continue;
401        }
402        let id = event.id.to_string();
403        for listener in taking.iter() {
404            let job = Uuid::now_v7()?.to_string();
405            jobs.push(json!({ "id": job, "name": listener, "event": id }));
406        }
407        taken.push(json!({ "id": id, "payload": event.payload, "traceparent": event.traceparent }));
408    }
409    if jobs.is_empty() {
410        return Ok(None);
411    }
412    let params = vec![
413        SqlValue::Json(Value::Array(taken)),
414        SqlValue::Json(Value::Array(jobs)),
415        SqlValue::Timestamp(at),
416        actor.map_or(SqlValue::Null, SqlValue::Uuid),
417    ];
418    let sql = dialect.messages().fan_out_events;
419    Ok(Some(Step::statement(sql, params, None)))
420}
421
422pub(crate) fn current_traceparent() -> SqlValue {
423    sofie_telemetry::current_traceparent().map_or(SqlValue::Null, |traceparent| {
424        SqlValue::Text(traceparent.to_string())
425    })
426}
427
428fn milliseconds(lease: Duration) -> Result<SqlValue> {
429    i64::try_from(lease.as_millis())
430        .map(SqlValue::Int)
431        .map_err(|error| Error::internal("a lease too long to store", error))
432}
433
434fn later(now: Timestamp, by: Duration) -> Result<Timestamp> {
435    now.checked_add(by)
436        .map_err(|error| Error::internal("computing a message time", error))
437}
sofie_db/messages.rs

sofie_db/
messages.rs
