feat: add a cross-repo dispatcher surface to the postgres effect queue
commit
a329962feat: add a cross-repo dispatcher surface to the postgres effect queue
dispatcher_claim spans every repo_id (one WS7 dispatcher machine serves them all) with a per-repo exclusion for repos at their fairness cap; dispatcher_requeue_stale returns claims older than a timeout to enqueued — the redelivery half of the queue’s at-least-once contract; dispatcher_complete closes a row wherever it came from. The namespace-per-repo doc note gains the queue’s deliberate exception.
Assisted-by: Claude:claude-sonnet-5
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
crates/refstore-postgres/src/lib.rs
@@ -38,7 +38,7 @@
mod ref_store;
mod small_tier;
-pub use queue::{ClaimedEffect, EffectId};
+pub use queue::{ClaimedEffect, DispatchedEffect, EffectId};
use git_backend::{Error, Result};
@@ -55,7 +55,11 @@
/// [`git_backend::RefStore`] over a Postgres database: one row per ref,
/// scoped to a single `repo_id` (`docs/scale-out.adoc`'s "namespace per
/// repo" rule — no cross-repo query this store issues ever omits the
-/// `repo_id` filter).
+/// `repo_id` filter, with one deliberate exception: the effect queue's
+/// `dispatcher_*` surface in [`queue`], which the WS7 dispatcher — one
+/// small machine serving every repository — drains across repos; the
+/// queue table carries jobs, never repository state, so the namespace
+/// rule's GC/oracle concerns do not apply to it).
///
/// Holds one [`tokio_postgres::Client`] behind a [`tokio::sync::Mutex`],
/// per the dependency policy: no connection pool. `transaction` needs
crates/refstore-postgres/src/queue.rs
@@ -14,6 +14,18 @@
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EffectId(i64);
+impl From<EffectId> for i64 {
+ fn from(id: EffectId) -> Self {
+ id.0
+ }
+}
+
+impl From<i64> for EffectId {
+ fn from(raw: i64) -> Self {
+ Self(raw)
+ }
+}
+
/// One row claimed off the effect queue by [`PostgresRefStore::claim_effects`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClaimedEffect {
@@ -24,6 +36,21 @@
pub payload: String,
}
+/// One row claimed off the effect queue by
+/// [`PostgresRefStore::dispatcher_claim`]. Unlike [`ClaimedEffect`], it
+/// carries its `repo_id`: the dispatcher serves every repository from one
+/// loop and accounts its per-repo fairness cap by that attribution.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DispatchedEffect {
+ /// The claimed row's id, needed to later call
+ /// [`PostgresRefStore::dispatcher_complete`].
+ pub id: EffectId,
+ /// The repository the row was enqueued for.
+ pub repo_id: String,
+ /// The effect payload enqueued by [`PostgresRefStore::enqueue_effect`].
+ pub payload: String,
+}
+
impl PostgresRefStore {
/// Append one effect payload to this store's repo-scoped queue in state
/// `enqueued`, returning its id.
@@ -112,6 +139,111 @@
.map(|_rows_affected| ())
}
+ /// Atomically claim up to `limit` of the oldest `enqueued` rows across
+ /// *every* repository, skipping rows whose `repo_id` is in
+ /// `exclude_repos` — the WS7 dispatcher's claim (`docs/scale-out.adoc`,
+ /// "WS7 — Effects and Sprites"): one small machine drains the whole
+ /// queue, and passes the repos currently at their per-repo fairness cap
+ /// as the exclusion so a saturated repository's backlog never starves
+ /// the rest. Same `FOR UPDATE SKIP LOCKED` discipline as
+ /// [`Self::claim_effects`]; deliberately not scoped to this store's
+ /// `repo_id` (see the crate-level note on the dispatcher exception).
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::RefStore`] if the claim query fails.
+ pub fn dispatcher_claim(
+ &self,
+ claimed_by: &str,
+ limit: i64,
+ exclude_repos: &[String],
+ ) -> Result<Vec<DispatchedEffect>> {
+ let exclude: Vec<String> = exclude_repos.to_vec();
+ let rows = self
+ .runtime
+ .block_on(async {
+ let client = self.client.lock().await;
+ client
+ .query(
+ "UPDATE git_ents_effect_queue
+ SET state = 'claimed', claimed_by = $1, claimed_at = now()
+ WHERE id IN (
+ SELECT id FROM git_ents_effect_queue
+ WHERE state = 'enqueued' AND repo_id <> ALL($2)
+ ORDER BY id
+ LIMIT $3
+ FOR UPDATE SKIP LOCKED
+ )
+ RETURNING id, repo_id, payload",
+ &[&claimed_by, &exclude, &limit],
+ )
+ .await
+ })
+ .map_err(pg_err)?;
+
+ rows.into_iter()
+ .map(|row| {
+ let id: i64 = row.try_get(0).map_err(pg_err)?;
+ let repo_id: String = row.try_get(1).map_err(pg_err)?;
+ let payload: String = row.try_get(2).map_err(pg_err)?;
+ Ok(DispatchedEffect {
+ id: EffectId(id),
+ repo_id,
+ payload,
+ })
+ })
+ .collect()
+ }
+
+ /// Return every `claimed` row (any repository) whose claim is older
+ /// than `older_than` to `enqueued`, clearing the claimant — the
+ /// redelivery half of the queue's at-least-once contract: a dispatcher
+ /// that died with claims outstanding loses nothing, its rows come back
+ /// once the timeout passes. Returns how many rows were requeued.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::RefStore`] if the update fails.
+ pub fn dispatcher_requeue_stale(&self, older_than: std::time::Duration) -> Result<u64> {
+ let seconds = older_than.as_secs_f64();
+ self.runtime
+ .block_on(async {
+ let client = self.client.lock().await;
+ client
+ .execute(
+ "UPDATE git_ents_effect_queue
+ SET state = 'enqueued', claimed_by = NULL, claimed_at = NULL
+ WHERE state = 'claimed'
+ AND claimed_at < now() - ($1 * interval '1 second')",
+ &[&seconds],
+ )
+ .await
+ })
+ .map_err(pg_err)
+ }
+
+ /// Mark a [`Self::dispatcher_claim`]-claimed row `done`, whichever
+ /// repository it belongs to.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::RefStore`] if the update fails.
+ pub fn dispatcher_complete(&self, id: EffectId) -> Result<()> {
+ self.runtime
+ .block_on(async {
+ let client = self.client.lock().await;
+ client
+ .execute(
+ "UPDATE git_ents_effect_queue SET state = 'done', done_at = now()
+ WHERE id = $1",
+ &[&id.0],
+ )
+ .await
+ })
+ .map_err(pg_err)
+ .map(|_rows_affected| ())
+ }
+
/// Record one accepted push's op record OID (`docs/scale-out.adoc`,
/// "Attested push": "Push ID = op record OID, uniformly").
///