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    /// A schedule stored under another cadence starts over from `first_at`.
224    pub async fn add_schedule(&self, name: &str, first_at: Timestamp, cadence: &str) -> Result<()> {
225        let params = [
226            SqlValue::Text(name.to_owned()),
227            SqlValue::Timestamp(first_at),
228            SqlValue::Text(cadence.to_owned()),
229        ];
230        let sql = self.sql().add_schedule;
231        query_values(self.db.driver(), sql, &params).await.map(drop)
232    }
233
234    /// A run by hand: a running worker keeps the next due time in memory, but takes a job at once.
235    pub async fn queue_schedule(&self, name: &str) -> Result<()> {
236        let dialect = self.db.driver().dialect();
237        let payload = Value::Object(serde_json::Map::new());
238        let now = self.db.clock().now().timestamp();
239        let mut steps = Vec::new();
240        let queuing = Queuing {
241            kind: MessageKind::Job,
242            at: now,
243            actor: None,
244        };
245        queue_steps(dialect, &queuing, [(name, payload)], None, &mut steps)?;
246        // In the same transaction, as the process may end before a notification sent later.
247        if let Some(notify) = dialect.messages().notify {
248            steps.push(Step::Script(notify.to_owned()));
249        }
250        self.db.transaction(steps).await
251    }
252
253    /// Only one of any number of concurrent callers is told it fired.
254    pub async fn fire_schedule(
255        &self,
256        name: &str,
257        now: Timestamp,
258        next_at: Timestamp,
259    ) -> Result<Option<Fired>> {
260        let payload = Value::Object(serde_json::Map::new());
261        let mut params = Message {
262            kind: MessageKind::Job,
263            name,
264            payload,
265            available_at: now,
266            actor: None,
267        }
268        .columns()?;
269        params.push(SqlValue::Timestamp(next_at));
270        let sql = self.sql().fire_schedule;
271        let fired =
272            query_decoded(self.db.driver(), sql, &params, Vec::new(), Fired::decode).await?;
273        Ok(fired.into_iter().next())
274    }
275
276    pub(crate) async fn arrived(&self) {
277        self.db.arrivals().arrived().await;
278    }
279
280    #[must_use]
281    pub fn wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
282        self.db.driver().wakeups()
283    }
284}
285
286pub(crate) async fn claim_known(
287    db: &Db,
288    lease: Duration,
289    limit: i64,
290    mine: &Known,
291    held_until: Timestamp,
292) -> Result<Vec<Claimed>> {
293    let params = [
294        milliseconds(lease)?,
295        SqlValue::Timestamp(db.clock().now().timestamp()),
296        SqlValue::Int(limit),
297        SqlValue::Json(mine.to_json()),
298        SqlValue::Timestamp(held_until),
299    ];
300    let sql = db.driver().dialect().messages().claim;
301    query_decoded(db.driver(), sql, &params, Vec::new(), Claimed::decode).await
302}
303
304pub async fn ping_queue(db: &Db) -> Result<()> {
305    let messages = db.driver().dialect().qualified("sofie", "messages");
306    let sql = format!("SELECT 1 FROM {messages} LIMIT 1");
307    query_values(db.driver(), &sql, &[]).await.map(drop)
308}
309
310pub(crate) struct Queuing {
311    kind: MessageKind,
312    at: Timestamp,
313    actor: Option<Uuid>,
314}
315
316impl Stamp {
317    pub(crate) fn queuing(&self, kind: MessageKind, at: Timestamp) -> Queuing {
318        Queuing {
319            kind,
320            at,
321            actor: self.actor,
322        }
323    }
324}
325
326pub(crate) fn queue_steps<'a>(
327    dialect: &dyn Dialect,
328    queuing: &Queuing,
329    messages: impl IntoIterator<Item = (&'a str, Value)>,
330    heard: Option<&Known>,
331    steps: &mut Vec<Step>,
332) -> Result<bool> {
333    if let Some(heard) = heard.filter(|_| queuing.kind == MessageKind::Event) {
334        return fan_out_raised(dialect, queuing, messages, &heard.listeners, steps);
335    }
336    let mut queued = false;
337    for (name, payload) in messages {
338        let params = Message {
339            kind: queuing.kind,
340            name,
341            payload,
342            available_at: queuing.at,
343            actor: queuing.actor,
344        }
345        .columns()?;
346        steps.push(Step::statement(dialect.messages().insert, params, None));
347        queued = true;
348    }
349    Ok(queued)
350}
351
352fn fan_out_raised<'a>(
353    dialect: &dyn Dialect,
354    queuing: &Queuing,
355    events: impl IntoIterator<Item = (&'a str, Value)>,
356    listeners: &Listeners,
357    steps: &mut Vec<Step>,
358) -> Result<bool> {
359    let traceparent = sofie_telemetry::current_traceparent().map(|parent| parent.to_string());
360    let mut raised = Vec::new();
361    for (name, payload) in events {
362        raised.push(Fanned {
363            id: Uuid::now_v7()?,
364            name,
365            payload,
366            traceparent: traceparent.clone(),
367        });
368    }
369    let Some(step) = fan_out_events(dialect, raised, queuing.at, queuing.actor, listeners)? else {
370        return Ok(false);
371    };
372    steps.push(step);
373    Ok(true)
374}
375
376pub(crate) struct Fanned<'a> {
377    pub(crate) id: Uuid,
378    pub(crate) name: &'a str,
379    pub(crate) payload: Value,
380    pub(crate) traceparent: Option<String>,
381}
382
383pub(crate) fn fan_out_events<'a>(
384    dialect: &dyn Dialect,
385    events: impl IntoIterator<Item = Fanned<'a>>,
386    at: Timestamp,
387    actor: Option<Uuid>,
388    listeners: &Listeners,
389) -> Result<Option<Step>> {
390    let (mut taken, mut jobs) = (Vec::new(), Vec::new());
391    for event in events {
392        let taking = listeners.taking(event.name, &event.payload);
393        if taking.is_empty() {
394            continue;
395        }
396        let id = event.id.to_string();
397        for listener in taking.iter() {
398            let job = Uuid::now_v7()?.to_string();
399            jobs.push(json!({ "id": job, "name": listener, "event": id }));
400        }
401        taken.push(json!({ "id": id, "payload": event.payload, "traceparent": event.traceparent }));
402    }
403    if jobs.is_empty() {
404        return Ok(None);
405    }
406    let params = vec![
407        SqlValue::Json(Value::Array(taken)),
408        SqlValue::Json(Value::Array(jobs)),
409        SqlValue::Timestamp(at),
410        actor.map_or(SqlValue::Null, SqlValue::Uuid),
411    ];
412    let sql = dialect.messages().fan_out_events;
413    Ok(Some(Step::statement(sql, params, None)))
414}
415
416pub(crate) fn current_traceparent() -> SqlValue {
417    sofie_telemetry::current_traceparent().map_or(SqlValue::Null, |traceparent| {
418        SqlValue::Text(traceparent.to_string())
419    })
420}
421
422fn milliseconds(lease: Duration) -> Result<SqlValue> {
423    i64::try_from(lease.as_millis())
424        .map(SqlValue::Int)
425        .map_err(|error| Error::internal("a lease too long to store", error))
426}
427
428fn later(now: Timestamp, by: Duration) -> Result<Timestamp> {
429    now.checked_add(by)
430        .map_err(|error| Error::internal("computing a message time", error))
431}
sofie_db/messages.rs

sofie_db/
messages.rs
