gwk_kernel/
writer.rs

1//! Singleton write authority: the advisory boot lock and the durable epoch.
2//!
3//! Two independent guards, because they fail in different directions.
4//!
5//! The **advisory lock** is held for the process lifetime on its own dedicated
6//! connection, taken with the NONBLOCKING variant so a second kernel exits
7//! immediately with a clear message instead of hanging on a queue nobody is
8//! watching. PostgreSQL releases a session advisory lock when its session ends,
9//! so the lock is exactly as alive as that connection — which is why losing the
10//! connection and losing the lock are the same event, and why a dead probe is
11//! sufficient evidence of both.
12//!
13//! The **durable epoch** covers what the lock cannot. A process that was fenced
14//! out may still hold pooled connections with in-flight work; the lock says
15//! nothing about those. Bumping a row at boot and comparing it inside every
16//! mutating transaction makes such a write fail at COMMIT time rather than
17//! land behind its successor's back.
18
19use std::time::Duration;
20
21use secrecy::{ExposeSecret, SecretString};
22use sqlx::{Connection, Executor, PgConnection, PgPool, Row};
23use tokio::sync::watch;
24
25use crate::error::{KernelError, Result};
26
27/// The advisory-lock key for "this database's kernel writer".
28///
29/// Advisory locks share one namespace per database, so the key is a constant
30/// anything else in this database must avoid. It spells `gwk` in the high bytes
31/// and `wr1` in the low ones, which makes an unexpected holder identifiable in
32/// `pg_locks` instead of anonymous.
33pub const WRITER_LOCK_KEY: i64 = 0x0067_776b_0077_7231;
34
35/// How often the holder proves its connection is still alive.
36pub const LOCK_PROBE_INTERVAL: Duration = Duration::from_secs(5);
37
38/// The process-lifetime advisory lock, plus the signal that it is gone.
39///
40/// Dropping this aborts the probe and closes the connection, which releases the
41/// lock — so scope IS lifetime, and there is no "forgot to unlock" path.
42#[derive(Debug)]
43pub struct WriterLock {
44    cancelled: watch::Receiver<bool>,
45    probe: tokio::task::JoinHandle<()>,
46}
47
48impl WriterLock {
49    /// Take the lock, or refuse. Never waits: another live kernel means this
50    /// one must not start, and blocking would hide that behind a hang.
51    pub async fn acquire(database_url: &SecretString) -> Result<Self> {
52        Self::acquire_with_interval(database_url, LOCK_PROBE_INTERVAL).await
53    }
54
55    pub async fn acquire_with_interval(
56        database_url: &SecretString,
57        probe_interval: Duration,
58    ) -> Result<Self> {
59        let mut conn = PgConnection::connect(database_url.expose_secret()).await?;
60        let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
61            .bind(WRITER_LOCK_KEY)
62            .fetch_one(&mut conn)
63            .await?;
64        if !acquired {
65            return Err(KernelError::Writer(
66                "another kernel already holds the writer lock on this database; \
67                 exactly one process may write per store"
68                    .to_owned(),
69            ));
70        }
71
72        let (tx, cancelled) = watch::channel(false);
73        let probe = tokio::spawn(async move {
74            loop {
75                tokio::time::sleep(probe_interval).await;
76                // A session advisory lock dies with its session, so a failed
77                // round trip on THIS connection is proof the lock is gone.
78                // Nothing here tries to re-acquire: a kernel that lost write
79                // authority must stop, not race the process that took it.
80                if conn.execute("SELECT 1").await.is_err() {
81                    let _ = tx.send(true);
82                    return;
83                }
84                if tx.is_closed() {
85                    return;
86                }
87            }
88        });
89
90        Ok(Self { cancelled, probe })
91    }
92
93    /// True once write authority is gone. Callers stop accepting work.
94    pub fn is_cancelled(&self) -> bool {
95        *self.cancelled.borrow()
96    }
97
98    /// Resolves when write authority is lost. Never resolves while healthy, so
99    /// it composes in a `select!` beside the accept loop.
100    pub async fn cancelled(&self) {
101        let mut rx = self.cancelled.clone();
102        while !*rx.borrow_and_update() {
103            if rx.changed().await.is_err() {
104                return; // the probe is gone: treat that as lost authority too
105            }
106        }
107    }
108}
109
110impl Drop for WriterLock {
111    fn drop(&mut self) {
112        self.probe.abort();
113    }
114}
115
116/// Claim this boot's writer epoch: bump the durable counter under a row lock
117/// and return what this process must present from now on.
118///
119/// Every later mutating transaction compares against this value, so a process
120/// that booted before another one finds its epoch superseded and cannot commit.
121///
122/// No explicit `FOR UPDATE`: an `UPDATE` already takes a row lock, and under
123/// READ COMMITTED a blocked one re-evaluates against the committed row rather
124/// than its original snapshot — so simultaneous boots serialize and cannot be
125/// handed the same number. Verified against 16 with 40 concurrent claims, which
126/// returned exactly 1..40.
127pub async fn claim_epoch(pool: &PgPool) -> Result<i64> {
128    let row = sqlx::query(
129        "UPDATE gwk_internal.writer SET epoch = epoch + 1 WHERE id = 1 RETURNING epoch",
130    )
131    .fetch_optional(pool)
132    .await?
133    .ok_or_else(|| {
134        KernelError::Schema(
135            "gwk_internal.writer has no singleton row: this database was not initialized by \
136             `gw admin init`"
137                .to_owned(),
138        )
139    })?;
140    Ok(row.try_get("epoch")?)
141}
gwk_kernel/writer.rs

gwk_kernel/
writer.rs
