Coverage / crates/hub/src/bus.rs 97.8%

//! The event bus (docs/roles.md, #273): publish and subscribe for the

//! events live pages and workers react to, in this process and in every

//! other process sharing the hub's database.

//!

//! An event says that something happened (a push moved a bookmark, a

//! proposal's check was recorded, a run's jobs changed, a webhook delivery

//! is waiting, a choice in a conflict-resolution room) and carries ids and

//! kinds, never the data: subscribers read the rows they need. Delivery is

//! at most once. Anything that must not be lost is a row in a table (a

//! delivery, a job), and its event only says "look now"; every subscriber

//! also looks by itself from time to time, or when it starts.

//!

//! In process, each [`Topic`] is a `tokio::sync::broadcast` channel: what

//! the single-node edition has always done, and all there is on SQLite (one

//! process by definition). On PostgreSQL the bus also carries each event to

//! the other processes: [`Bus::publish`] delivers it here at once and sends

//! it with `pg_notify` through the pool, on one channel per topic named for

//! the hub's schema (two hubs in two schemas of one database never hear

//! each other). Each process listens on those channels on one connection of

//! its own, outside the pool, and relays what it hears into its local

//! channels without publishing it again. Every event is sent with this

//! process's origin id, and a process skips its own: each subscriber hears

//! an event once, in every process. If the listening connection is lost, the

//! process listens again on a new one, says that events published meanwhile

//! were missed, and nudges its own webhook worker to look now.

use crate::events::{hook_topics, LiveBus};

use crate::live::{RoomEvent, RoomKey};

use crate::Shared;

use hub_api::ProposalEvent;

use tokio::sync::broadcast;

/// What the bus carries: one local channel per topic and, on PostgreSQL,

/// one notification channel per topic.

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]

pub enum Topic {

    /// A webhook delivery is waiting (the worker's nudge).

    Deliveries,

    /// A proposal's event: a check recorded on a projected state.

    Proposals,

    /// Live pages' topics (docs/live.md): pushes, checks, runs' jobs,

    /// comments, reviews, notifications, Diffusion's bumps.

    Live,

    /// Choices and landings in conflict-resolution rooms.

    Rooms,

}

impl Topic {

    pub const ALL: [Topic; 4] = [Topic::Deliveries, Topic::Proposals, Topic::Live, Topic::Rooms];

        }

    /// How many events a slow local subscriber may fall behind by.

        }

}

/// An event: its topic is its kind, and it carries ids and kinds only.

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]

#[serde(tag = "t", rename_all = "snake_case")]

pub enum Event {

    /// A webhook delivery is waiting: the worker should look now.

    DeliveryWaiting,

    /// Something about a proposal of repository `repo` (the proposal

    /// page's stream, hub spec §13.1): a check recorded on a state.

    Proposal { repo: i64, event: ProposalEvent },

    /// A live page's topic changed (`repo:1`, `run:5`, …) with a kind

    /// (`push`, `check`, `jobs`, `comment`, …).

    Live { topic: String, kind: String },

    /// A viewer's choice, or the landing, in a conflict-resolution room.

    Room { room: RoomKey, event: RoomEvent },

}

impl Event {

        }

}

/// This process's end of the bus.

pub struct Bus {

    /// Sent with every event this process publishes, so it skips its own

    /// when they come back from the database.

    #[cfg_attr(not(feature = "postgres"), allow(dead_code))]

    origin: String,

    /// One broadcast per topic, in [`Topic::ALL`]'s order.

    channels: Vec<broadcast::Sender<Event>>,

    /// Live pages' events as this process numbers them, with the last few

    /// kept for `Last-Event-ID` (docs/live.md).

    pub live: LiveBus,

    #[cfg(feature = "postgres")]

    remote: pg::Remote,

}

impl Default for Bus {

        Bus {

            #[cfg(feature = "postgres")]

            remote: pg::Remote::default(),

        }

}

impl Bus {

    /// Publish `event`: this process's subscribers hear it now, and every

    /// other process's soon after (PostgreSQL), at most once.

        #[cfg(feature = "postgres")]

        self.remote.send(&self.origin, &event);

    /// Hear the events on `topic` published from now on, in this process

    /// and the others.

    /// A live page's `topic` changed with `kind` (docs/live.md).

    /// Live pages hear of a webhook event: the repository's topic, and the

    /// proposal's or issue's its data names.

    /// A webhook delivery is waiting.

    /// Announce something about a proposal of a repository.

    /// Tell this process's subscribers only: [`Bus::publish`]'s local half,

    /// and what the relay does with another process's event.

    /// The process id of the database session listening for other

    /// processes' events, while it listens (for tests that end it).

    #[cfg(feature = "postgres")]

    pub fn listener_pid(&self) -> Option<i32> {

        *self.remote.pid.lock().unwrap_or_else(|p| p.into_inner())

    }

}

/// Start carrying events between processes: on PostgreSQL, the sender and

/// the listener; on SQLite, nothing (one process).

    #[cfg(feature = "postgres")]

    pg::start(app).await?;

    #[cfg(not(feature = "postgres"))]

/// The most a notification's payload may be: PostgreSQL takes fewer than

/// 8000 bytes.

pub const MAX_PAYLOAD: usize = 7999;

/// An event as sent between processes: who sent it, and the event.

#[derive(serde::Serialize, serde::Deserialize)]

struct Envelope {

    o: String,

    e: Event,

}

/// The payload for `event` from `origin`, within [`MAX_PAYLOAD`]: a check's

/// summary too long for it is cut (the proposal page reads the check for

/// the rest); anything else too long (a choice's own text) is not sent to

/// other processes (`None`).

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

    loop {

        // Escaping may make the summary longer in JSON than as text: cut

        // by what is over, and again if still over.

    }

/// The event in `payload`, unless this process (`origin`) sent it or it

/// cannot be read.

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

        }

    }

/// The notification channel for `topic` of the hub in `schema`: a stable

/// hash of the schema, so channel names are plain and short whatever the

/// schema is called.

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

#[cfg(feature = "postgres")]

mod pg {

    use super::{channel, decode, encode, Event, Topic};

    use crate::{App, Shared};

    use sqlx::postgres::{PgListener, PgPool, PgPoolOptions};

    use std::sync::{Mutex, OnceLock};

    use std::time::Duration;

    use tokio::sync::mpsc;

    /// How long a quiet listening connection goes before it is asked to

    /// answer, and how long it has to answer.

    const CHECK: Duration = Duration::from_secs(5);

    /// How long to wait before listening again after failing to connect.

    const RETRY: Duration = Duration::from_secs(2);

    /// Events waiting to be sent to other processes; past this, new ones

    /// are dropped (and said so): the bus is at most once.

    const QUEUE: usize = 4096;

    #[derive(Default)]

    pub(super) struct Remote {

        /// The hub's schema and the queue to the sender, once started.

        out: OnceLock<(String, mpsc::Sender<(String, String)>)>,

        /// The listening session's process id, while it listens.

        pub(super) pid: Mutex<Option<i32>>,

    }

    impl Remote {

        /// Queue `event` for the other processes. Before [`start`], or

        /// for an event too large to send, it stays in this process.

        pub(super) fn send(&self, origin: &str, event: &Event) {

            let Some((schema, tx)) = self.out.get() else { return };

            let Some(payload) = encode(origin, event) else {

                tracing::warn!("bus: a {} event too large to send to other processes stays in this one", event.topic().name());

                return;

            };

            if tx.try_send((channel(schema, event.topic()), payload)).is_err() {

                tracing::warn!("bus: too many events waiting to go to other processes; one was dropped");

            }

        }

    }

    pub(super) async fn start(app: &Shared) -> anyhow::Result<()> {

        let schema: String = sqlx::query_scalar("SELECT CAST(current_schema() AS TEXT)").fetch_one(&app.db).await?;

        let (tx, rx) = mpsc::channel(QUEUE);

        if app.bus.remote.out.set((schema.clone(), tx)).is_err() {

            return Ok(());

        }

        tokio::spawn(send_all(app.db.clone(), rx));

        tokio::spawn(listen(app.clone(), schema));

        Ok(())

    }

    /// Send queued events in order, through the pool.

    async fn send_all(db: crate::db::Pool, mut rx: mpsc::Receiver<(String, String)>) {

        while let Some((channel, payload)) = rx.recv().await {

            if let Err(e) = sqlx::query("SELECT pg_notify($1, $2)").bind(&channel).bind(&payload).execute(&db).await {

                tracing::warn!("bus: sending an event to other processes: {e}");

            }

        }

    }

    /// A connection of its own, outside the pool, listening on every topic.

    async fn connect(app: &App, channels: &[String]) -> anyhow::Result<(PgPool, PgListener, i32)> {

        let options = (*app.db.connect_options()).clone();

        let pool = PgPoolOptions::new().max_connections(1).max_lifetime(None).idle_timeout(None).connect_with(options).await?;

        let mut listener = PgListener::connect_with(&pool).await?;

        listener.ignore_pool_close_event(true);

        // A lost connection ends this listener; [`listen`] makes another.

        listener.eager_reconnect(false);

        listener.listen_all(channels.iter().map(String::as_str)).await?;

        let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&mut listener).await?;

        Ok((pool, listener, pid))

    }

    /// Listen for other processes' events and relay them here, for as long

    /// as the process runs, on a new connection whenever one is lost.

    async fn listen(app: Shared, schema: String) {

        let channels: Vec<String> = Topic::ALL.iter().map(|t| channel(&schema, *t)).collect();

        let mut lost = false;

        loop {

            let (pool, mut listener, pid) = match tokio::time::timeout(CHECK * 2, connect(&app, &channels)).await {

                Ok(Ok(c)) => c,

                Ok(Err(e)) => {

                    tracing::warn!("bus: listening for other processes' events: {e}");

                    tokio::time::sleep(RETRY).await;

                    continue;

                }

                Err(_) => {

                    tracing::warn!("bus: listening for other processes' events: timed out");

                    tokio::time::sleep(RETRY).await;

                    continue;

                }

            };

            *app.bus.remote.pid.lock().unwrap_or_else(|p| p.into_inner()) = Some(pid);

            if lost {

                tracing::warn!("bus: listening again; events other processes published meanwhile were missed");

                // Look now for what a missed nudge would have said.

                app.bus.deliver(Event::DeliveryWaiting);

            }

            loop {

                tokio::select! {

                    heard = listener.try_recv() => match heard {

                        Ok(Some(n)) => relay(&app, n.payload()),

                        Ok(None) => {

                            tracing::warn!("bus: the connection listening for other processes' events was lost");

                            break;

                        }

                        Err(e) => {

                            tracing::warn!("bus: the connection listening for other processes' events failed: {e}");

                            break;

                        }

                    },

                    _ = tokio::time::sleep(CHECK) => {

                        let ping = sqlx::query_scalar::<_, i32>("SELECT 1").fetch_one(&mut listener);

                        if !matches!(tokio::time::timeout(CHECK, ping).await, Ok(Ok(_))) {

                            tracing::warn!("bus: the connection listening for other processes' events stopped answering");

                            break;

                        }

                    }

                }

            }

            lost = true;

            *app.bus.remote.pid.lock().unwrap_or_else(|p| p.into_inner()) = None;

            drop(listener);

            let _ = tokio::time::timeout(CHECK, pool.close()).await;

        }

    }

    /// Another process's event, told to this one's subscribers (and its

    /// rooms), never published again.

    fn relay(app: &App, payload: &str) {

        let Some(event) = decode(&app.bus.origin, payload) else { return };

        if let Event::Room { room, event } = &event {

            app.rooms.apply(room, event);

        }

        app.bus.deliver(event);

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    use crate::events::Replay;

    #[test]

        // Live events are numbered for replay as they are published.

    #[test]

    #[test]