apiplant_queue/
listener.rs

1//! The half of the queue that makes it prompt: a Postgres `LISTEN`.
2//!
3//! This needs a connection of its own, which is why it is not simply another
4//! method on [`Queue`](crate::Queue). `LISTEN` is a property of a *session* —
5//! it lasts until that connection ends — and every other query in apiplant goes
6//! through a pool that hands the connection back the moment the statement
7//! finishes. Issuing `LISTEN` there would register interest on a connection
8//! immediately returned to the pool and reused for something else.
9//!
10//! So the listener holds one connection open for the life of the process. That
11//! is the cost of the feature: one connection per replica, whether or not
12//! anything is ever published.
13//!
14//! Losing it is survivable and deliberately not fatal. [`sqlx`]'s listener
15//! reconnects and re-subscribes on its own, and any notification that lands in
16//! the gap costs nothing but latency, because the message is a committed row
17//! that the subscriber's periodic sweep will find regardless. That is the whole
18//! reason the row exists as well as the notification.
19
20use sea_orm::sqlx::postgres::PgListener;
21
22use crate::QueueError;
23
24/// A live `LISTEN` on the app's queue channel.
25pub struct Listener {
26    inner: PgListener,
27    channel: String,
28}
29
30impl Listener {
31    /// Open a dedicated connection and subscribe to `channel`.
32    ///
33    /// Fails when the database is unreachable *right now*, which the caller
34    /// should treat as "run without notifications" rather than as a reason not
35    /// to start: a subscriber that polls every `poll_secs` still handles every
36    /// message, just later.
37    pub async fn connect(url: &str, channel: &str) -> Result<Self, QueueError> {
38        let mut inner = PgListener::connect(url)
39            .await
40            .map_err(|e| QueueError::Backend(format!("cannot open a listener connection: {e}")))?;
41        inner
42            .listen(channel)
43            .await
44            .map_err(|e| QueueError::Backend(format!("cannot LISTEN on `{channel}`: {e}")))?;
45        Ok(Listener {
46            inner,
47            channel: channel.to_string(),
48        })
49    }
50
51    /// Wait for the next notification and return the topic it names.
52    ///
53    /// The topic is a hint, not an instruction: the caller sweeps for whatever
54    /// is claimable rather than trusting it, because a notification can be
55    /// coalesced, arrive twice, or be for a topic whose row another replica has
56    /// already taken. Treating it as data to act on would make the queue's
57    /// correctness depend on the one part of it that is allowed to be lost.
58    pub async fn recv(&mut self) -> Result<String, QueueError> {
59        // `recv` reconnects and re-issues the LISTEN internally, so this only
60        // errors when the database is properly gone.
61        let notification = self.inner.recv().await.map_err(|e| {
62            QueueError::Backend(format!("listening on `{}` failed: {e}", self.channel))
63        })?;
64        Ok(notification.payload().to_string())
65    }
66}
apiplant_queue/listener.rs

apiplant_queue/
listener.rs
