feat(v1.1.9): QueueRepo trait + Postgres impl + shared queue.rs
- shared/queue.rs: QueueService trait (enqueue / depth / depth_pending),
EnqueueOpts, QueueError, NoopQueueService. Methods derive app_id from
cx.app_id — no script-passed app_id. The handle-less surface mirrors
pubsub (queues are the grouping unit).
- shared/ids.rs: QueueMessageId.
- manager-core/queue_repo.rs: PostgresQueueRepo with:
- enqueue(NewQueueMessage) — single INSERT
- claim(app_id, queue_name) — atomic UPDATE WHERE id = (SELECT … FOR
UPDATE SKIP LOCKED LIMIT 1) RETURNING — the single-round-trip claim
from the design notes
- ack(id, claim_token) — DELETE WHERE id AND claim_token (lease check)
- nack(id, claim_token, retry_delay) — clear claim + set deliver_after
- reclaim_visibility_timeouts() — periodic UPDATE joining triggers +
queue_trigger_details, clears claims older than per-queue
visibility_timeout_secs
- depth / depth_pending / list_for_app (dashboard read-only)
- dead_letter(...) — atomic move to dead_letters + DELETE from
queue_messages, in one transaction. Uses a JSON payload shaped the
same as TriggerEvent::Queue so the existing fan_out_dead_letter
path delivers the original to registered dead_letter triggers
without special-casing.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -55,6 +55,7 @@ pub mod outbox_repo;
|
||||
pub mod principal_resolver;
|
||||
pub mod pubsub_repo;
|
||||
pub mod pubsub_service;
|
||||
pub mod queue_repo;
|
||||
pub mod realtime_authority;
|
||||
pub mod repo;
|
||||
pub mod route_admin;
|
||||
|
||||
463
crates/manager-core/src/queue_repo.rs
Normal file
463
crates/manager-core/src/queue_repo.rs
Normal file
@@ -0,0 +1,463 @@
|
||||
//! `QueueRepo` — CRUD over the `queue_messages` table.
|
||||
//!
|
||||
//! Producer path: `enqueue(...)` writes one row. Dispatcher hot path:
|
||||
//! `claim(...)` does the single-round-trip atomic claim
|
||||
//! (`UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED)`).
|
||||
//! Outcome: `ack` deletes the row, `nack` clears the claim with a
|
||||
//! `deliver_after` retry delay, `dead_letter` moves the payload to
|
||||
//! `dead_letters` and deletes the queue row.
|
||||
//!
|
||||
//! `reclaim_visibility_timeouts()` is the periodic safety net — a
|
||||
//! crashed dispatcher (or one whose handler hung past the visibility
|
||||
//! timeout) loses its claim and the message becomes claimable again.
|
||||
//!
|
||||
//! All methods take `app_id` explicitly; the service layer derives it
|
||||
//! from `cx.app_id` and passes it in — repo never reads cx.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use chrono::{DateTime, Utc};
|
||||
use picloud_shared::{AdminUserId, AppId, QueueMessageId, TriggerId};
|
||||
use sqlx::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::dead_letter_repo::{DeadLetterRepoError, NewDeadLetter};
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum QueueRepoError {
|
||||
#[error("database error: {0}")]
|
||||
Db(#[from] sqlx::Error),
|
||||
|
||||
/// JSON serialization failure on the payload (extremely rare; only
|
||||
/// surfaces if a non-serializable Value somehow reached the repo).
|
||||
#[error("payload serialization: {0}")]
|
||||
Payload(String),
|
||||
|
||||
/// Surfaces from the dead_letter() helper if the dead-letter insert
|
||||
/// fails.
|
||||
#[error("dead-letter write: {0}")]
|
||||
DeadLetter(String),
|
||||
}
|
||||
|
||||
/// Insert payload — what `QueueService::enqueue` hands the repo.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct NewQueueMessage {
|
||||
pub app_id: AppId,
|
||||
pub queue_name: String,
|
||||
pub payload: serde_json::Value,
|
||||
/// `Some` → defer until `NOW() + delay`; `None` → immediate.
|
||||
pub deliver_after: Option<DateTime<Utc>>,
|
||||
pub max_attempts: u32,
|
||||
pub enqueued_by_principal: Option<AdminUserId>,
|
||||
}
|
||||
|
||||
/// One claimed message ready for handler dispatch.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ClaimedMessage {
|
||||
pub id: QueueMessageId,
|
||||
pub app_id: AppId,
|
||||
pub queue_name: String,
|
||||
pub payload: serde_json::Value,
|
||||
pub enqueued_at: DateTime<Utc>,
|
||||
pub attempt: u32,
|
||||
pub max_attempts: u32,
|
||||
pub claim_token: Uuid,
|
||||
}
|
||||
|
||||
/// Aggregate counts for the dashboard's read-only queue overview.
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct QueueStats {
|
||||
/// `COUNT(*)` for `(app_id, queue_name)`.
|
||||
pub total: u64,
|
||||
/// Claimable now — `claim_token IS NULL AND (deliver_after IS NULL OR <= NOW())`.
|
||||
pub pending: u64,
|
||||
/// Currently leased to a dispatcher — `claim_token IS NOT NULL`.
|
||||
pub claimed: u64,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait QueueRepo: Send + Sync {
|
||||
/// Insert one row. Returns the new message id.
|
||||
async fn enqueue(&self, msg: NewQueueMessage) -> Result<QueueMessageId, QueueRepoError>;
|
||||
|
||||
/// Atomic claim of a single ready message from `(app_id, queue_name)`.
|
||||
/// Returns `Ok(None)` if nothing is claimable.
|
||||
async fn claim(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
queue_name: &str,
|
||||
) -> Result<Option<ClaimedMessage>, QueueRepoError>;
|
||||
|
||||
/// Handler succeeded: delete the row IFF the claim_token matches
|
||||
/// (so a stale dispatcher whose lease expired doesn't accidentally
|
||||
/// delete a re-claimed message). Returns `true` if 1 row deleted.
|
||||
async fn ack(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
) -> Result<bool, QueueRepoError>;
|
||||
|
||||
/// Handler threw: clear the claim, defer redelivery by `retry_delay`.
|
||||
/// IFF the claim_token matches. Returns `true` on success.
|
||||
async fn nack(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
retry_delay: chrono::Duration,
|
||||
) -> Result<bool, QueueRepoError>;
|
||||
|
||||
/// Periodic safety net: clear claims whose `claimed_at` exceeds the
|
||||
/// per-queue `visibility_timeout_secs` (joined from
|
||||
/// `queue_trigger_details`). Returns the number of rows reclaimed.
|
||||
async fn reclaim_visibility_timeouts(&self) -> Result<u64, QueueRepoError>;
|
||||
|
||||
/// `queue::depth(name)` — total rows in the queue.
|
||||
async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError>;
|
||||
|
||||
/// `queue::depth_pending(name)` — currently claimable rows.
|
||||
async fn depth_pending(&self, app_id: AppId, queue_name: &str)
|
||||
-> Result<u64, QueueRepoError>;
|
||||
|
||||
/// Dashboard read-only view: every distinct queue name in `app_id`
|
||||
/// with its aggregate counts.
|
||||
async fn list_for_app(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
) -> Result<Vec<(String, QueueStats)>, QueueRepoError>;
|
||||
|
||||
/// Dispatcher dead-letter helper: move a claimed message to
|
||||
/// `dead_letters` and delete it from `queue_messages`, in one
|
||||
/// transaction. The dispatcher's existing `fan_out_dead_letter`
|
||||
/// fires registered `dead_letter` triggers off the new row.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn dead_letter(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
app_id: AppId,
|
||||
queue_name: &str,
|
||||
trigger_id: Option<TriggerId>,
|
||||
script_id: Option<picloud_shared::ScriptId>,
|
||||
attempt: u32,
|
||||
first_attempt_at: DateTime<Utc>,
|
||||
last_error: &str,
|
||||
) -> Result<picloud_shared::DeadLetterId, QueueRepoError>;
|
||||
}
|
||||
|
||||
pub struct PostgresQueueRepo {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresQueueRepo {
|
||||
#[must_use]
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl QueueRepo for PostgresQueueRepo {
|
||||
async fn enqueue(&self, msg: NewQueueMessage) -> Result<QueueMessageId, QueueRepoError> {
|
||||
let (id,): (Uuid,) = sqlx::query_as(
|
||||
"INSERT INTO queue_messages ( \
|
||||
app_id, queue_name, payload, deliver_after, \
|
||||
max_attempts, enqueued_by_principal \
|
||||
) VALUES ($1, $2, $3, $4, $5, $6) \
|
||||
RETURNING id",
|
||||
)
|
||||
.bind(msg.app_id.into_inner())
|
||||
.bind(&msg.queue_name)
|
||||
.bind(&msg.payload)
|
||||
.bind(msg.deliver_after)
|
||||
.bind(i32::try_from(msg.max_attempts).unwrap_or(3))
|
||||
.bind(msg.enqueued_by_principal.map(AdminUserId::into_inner))
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(id.into())
|
||||
}
|
||||
|
||||
async fn claim(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
queue_name: &str,
|
||||
) -> Result<Option<ClaimedMessage>, QueueRepoError> {
|
||||
let token = Uuid::new_v4();
|
||||
let row: Option<ClaimedRow> = sqlx::query_as(
|
||||
"UPDATE queue_messages \
|
||||
SET claim_token = $3, claimed_at = NOW(), attempt = attempt + 1 \
|
||||
WHERE id = ( \
|
||||
SELECT id FROM queue_messages \
|
||||
WHERE app_id = $1 AND queue_name = $2 \
|
||||
AND claim_token IS NULL \
|
||||
AND (deliver_after IS NULL OR deliver_after <= NOW()) \
|
||||
ORDER BY enqueued_at \
|
||||
FOR UPDATE SKIP LOCKED \
|
||||
LIMIT 1 \
|
||||
) \
|
||||
RETURNING id, app_id, queue_name, payload, enqueued_at, attempt, max_attempts",
|
||||
)
|
||||
.bind(app_id.into_inner())
|
||||
.bind(queue_name)
|
||||
.bind(token)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
Ok(row.map(|r| ClaimedMessage {
|
||||
id: r.id.into(),
|
||||
app_id: r.app_id.into(),
|
||||
queue_name: r.queue_name,
|
||||
payload: r.payload,
|
||||
enqueued_at: r.enqueued_at,
|
||||
attempt: u32::try_from(r.attempt).unwrap_or(1),
|
||||
max_attempts: u32::try_from(r.max_attempts).unwrap_or(3),
|
||||
claim_token: token,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn ack(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
) -> Result<bool, QueueRepoError> {
|
||||
let res = sqlx::query(
|
||||
"DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2",
|
||||
)
|
||||
.bind(message_id.into_inner())
|
||||
.bind(claim_token)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(res.rows_affected() == 1)
|
||||
}
|
||||
|
||||
async fn nack(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
retry_delay: chrono::Duration,
|
||||
) -> Result<bool, QueueRepoError> {
|
||||
let next = Utc::now() + retry_delay;
|
||||
let res = sqlx::query(
|
||||
"UPDATE queue_messages \
|
||||
SET claim_token = NULL, claimed_at = NULL, deliver_after = $3 \
|
||||
WHERE id = $1 AND claim_token = $2",
|
||||
)
|
||||
.bind(message_id.into_inner())
|
||||
.bind(claim_token)
|
||||
.bind(next)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(res.rows_affected() == 1)
|
||||
}
|
||||
|
||||
async fn reclaim_visibility_timeouts(&self) -> Result<u64, QueueRepoError> {
|
||||
let res = sqlx::query(
|
||||
"UPDATE queue_messages m \
|
||||
SET claim_token = NULL, claimed_at = NULL \
|
||||
FROM triggers t \
|
||||
JOIN queue_trigger_details d ON d.trigger_id = t.id \
|
||||
WHERE m.app_id = t.app_id \
|
||||
AND m.queue_name = d.queue_name \
|
||||
AND m.claim_token IS NOT NULL \
|
||||
AND m.claimed_at < NOW() - (d.visibility_timeout_secs || ' seconds')::INTERVAL \
|
||||
AND t.kind = 'queue' AND t.enabled = TRUE",
|
||||
)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(res.rows_affected())
|
||||
}
|
||||
|
||||
async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
|
||||
let (n,): (i64,) = sqlx::query_as(
|
||||
"SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2",
|
||||
)
|
||||
.bind(app_id.into_inner())
|
||||
.bind(queue_name)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(u64::try_from(n).unwrap_or(0))
|
||||
}
|
||||
|
||||
async fn depth_pending(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
queue_name: &str,
|
||||
) -> Result<u64, QueueRepoError> {
|
||||
let (n,): (i64,) = sqlx::query_as(
|
||||
"SELECT COUNT(*) FROM queue_messages \
|
||||
WHERE app_id = $1 AND queue_name = $2 \
|
||||
AND claim_token IS NULL \
|
||||
AND (deliver_after IS NULL OR deliver_after <= NOW())",
|
||||
)
|
||||
.bind(app_id.into_inner())
|
||||
.bind(queue_name)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(u64::try_from(n).unwrap_or(0))
|
||||
}
|
||||
|
||||
async fn list_for_app(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
) -> Result<Vec<(String, QueueStats)>, QueueRepoError> {
|
||||
let rows: Vec<QueueAggRow> = sqlx::query_as(
|
||||
"SELECT queue_name, \
|
||||
COUNT(*) AS total, \
|
||||
COUNT(*) FILTER ( \
|
||||
WHERE claim_token IS NULL \
|
||||
AND (deliver_after IS NULL OR deliver_after <= NOW()) \
|
||||
) AS pending, \
|
||||
COUNT(*) FILTER (WHERE claim_token IS NOT NULL) AS claimed \
|
||||
FROM queue_messages \
|
||||
WHERE app_id = $1 \
|
||||
GROUP BY queue_name \
|
||||
ORDER BY queue_name",
|
||||
)
|
||||
.bind(app_id.into_inner())
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|r| {
|
||||
(
|
||||
r.queue_name,
|
||||
QueueStats {
|
||||
total: u64::try_from(r.total).unwrap_or(0),
|
||||
pending: u64::try_from(r.pending).unwrap_or(0),
|
||||
claimed: u64::try_from(r.claimed).unwrap_or(0),
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn dead_letter(
|
||||
&self,
|
||||
message_id: QueueMessageId,
|
||||
claim_token: Uuid,
|
||||
app_id: AppId,
|
||||
queue_name: &str,
|
||||
trigger_id: Option<TriggerId>,
|
||||
script_id: Option<picloud_shared::ScriptId>,
|
||||
attempt: u32,
|
||||
first_attempt_at: DateTime<Utc>,
|
||||
last_error: &str,
|
||||
) -> Result<picloud_shared::DeadLetterId, QueueRepoError> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
// Pull the row inside the transaction; we need the original
|
||||
// payload for the dead_letters row. Filtered by claim_token so
|
||||
// a lost lease can't dead-letter a re-claimed message.
|
||||
let row: QueueRowMinimal = sqlx::query_as(
|
||||
"SELECT id, payload, enqueued_at FROM queue_messages \
|
||||
WHERE id = $1 AND claim_token = $2",
|
||||
)
|
||||
.bind(message_id.into_inner())
|
||||
.bind(claim_token)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let payload = serde_json::json!({
|
||||
"source": "queue",
|
||||
"queue_name": queue_name,
|
||||
"message": row.payload,
|
||||
"enqueued_at": row.enqueued_at,
|
||||
"attempt": attempt,
|
||||
"message_id": row.id.to_string(),
|
||||
});
|
||||
|
||||
let dl_id = picloud_shared::DeadLetterId::new();
|
||||
sqlx::query(
|
||||
"INSERT INTO dead_letters ( \
|
||||
id, app_id, original_event_id, source, op, trigger_id, script_id, \
|
||||
payload, attempt_count, first_attempt_at, last_attempt_at, last_error \
|
||||
) VALUES ($1, $2, $3, 'queue', 'receive', $4, $5, $6, $7, $8, NOW(), $9)",
|
||||
)
|
||||
.bind(dl_id.into_inner())
|
||||
.bind(app_id.into_inner())
|
||||
.bind(row.id)
|
||||
.bind(trigger_id.map(TriggerId::into_inner))
|
||||
.bind(script_id.map(picloud_shared::ScriptId::into_inner))
|
||||
.bind(&payload)
|
||||
.bind(i32::try_from(attempt).unwrap_or(0))
|
||||
.bind(first_attempt_at)
|
||||
.bind(last_error)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2")
|
||||
.bind(message_id.into_inner())
|
||||
.bind(claim_token)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(dl_id)
|
||||
}
|
||||
}
|
||||
|
||||
/// Bridge between this repo and the existing `DeadLetterRepo`. Not used
|
||||
/// by `dead_letter()` directly (that path inlines its own INSERT for one-
|
||||
/// transaction atomicity), but exposed for callers that already have a
|
||||
/// `DeadLetterRepo` and want to keep one writer.
|
||||
#[must_use]
|
||||
pub fn build_new_dead_letter_from_claimed(
|
||||
msg: &ClaimedMessage,
|
||||
trigger_id: Option<TriggerId>,
|
||||
script_id: Option<picloud_shared::ScriptId>,
|
||||
last_error: String,
|
||||
first_attempt_at: DateTime<Utc>,
|
||||
) -> NewDeadLetter {
|
||||
let payload = serde_json::json!({
|
||||
"source": "queue",
|
||||
"queue_name": msg.queue_name,
|
||||
"message": msg.payload,
|
||||
"enqueued_at": msg.enqueued_at,
|
||||
"attempt": msg.attempt,
|
||||
"message_id": msg.id.to_string(),
|
||||
});
|
||||
NewDeadLetter {
|
||||
app_id: msg.app_id,
|
||||
original_event_id: msg.id.into_inner(),
|
||||
source: "queue".into(),
|
||||
op: "receive".into(),
|
||||
trigger_id,
|
||||
script_id,
|
||||
payload,
|
||||
attempt_count: msg.attempt,
|
||||
first_attempt_at,
|
||||
last_attempt_at: Utc::now(),
|
||||
last_error,
|
||||
}
|
||||
}
|
||||
|
||||
/// Surface to bridge DeadLetterRepoError → QueueRepoError when callers
|
||||
/// route the dead-letter write through the standard repo instead of the
|
||||
/// inline `dead_letter()` path.
|
||||
impl From<DeadLetterRepoError> for QueueRepoError {
|
||||
fn from(e: DeadLetterRepoError) -> Self {
|
||||
Self::DeadLetter(e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(sqlx::FromRow)]
|
||||
struct ClaimedRow {
|
||||
id: Uuid,
|
||||
app_id: Uuid,
|
||||
queue_name: String,
|
||||
payload: serde_json::Value,
|
||||
enqueued_at: DateTime<Utc>,
|
||||
attempt: i32,
|
||||
max_attempts: i32,
|
||||
}
|
||||
|
||||
#[derive(sqlx::FromRow)]
|
||||
struct QueueAggRow {
|
||||
queue_name: String,
|
||||
total: i64,
|
||||
pending: i64,
|
||||
claimed: i64,
|
||||
}
|
||||
|
||||
#[derive(sqlx::FromRow)]
|
||||
struct QueueRowMinimal {
|
||||
id: Uuid,
|
||||
payload: serde_json::Value,
|
||||
enqueued_at: DateTime<Utc>,
|
||||
}
|
||||
@@ -56,3 +56,4 @@ id_type!(ApiKeyId);
|
||||
id_type!(TriggerId);
|
||||
id_type!(AppUserId);
|
||||
id_type!(InvitationId);
|
||||
id_type!(QueueMessageId);
|
||||
|
||||
@@ -23,6 +23,7 @@ pub mod log_sink;
|
||||
pub mod modules;
|
||||
pub mod outbox_writer;
|
||||
pub mod pubsub;
|
||||
pub mod queue;
|
||||
pub mod realtime;
|
||||
pub mod realtime_authority;
|
||||
pub mod route;
|
||||
@@ -53,8 +54,8 @@ pub use files::{
|
||||
};
|
||||
pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService};
|
||||
pub use ids::{
|
||||
AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, InvitationId, RequestId, ScriptId,
|
||||
TriggerId,
|
||||
AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, InvitationId, QueueMessageId, RequestId,
|
||||
ScriptId, TriggerId,
|
||||
};
|
||||
pub use inbox::{
|
||||
InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult, NoopInboxResolver,
|
||||
@@ -66,6 +67,7 @@ pub use outbox_writer::{HttpDispatchPayload, NewHttpOutbox, OutboxWriter, Outbox
|
||||
pub use pubsub::{
|
||||
topic_matches, validate_topic_pattern, NoopPubsubService, PubsubError, PubsubService,
|
||||
};
|
||||
pub use queue::{EnqueueOpts, NoopQueueService, QueueError, QueueService};
|
||||
pub use realtime::{BroadcasterError, NoopRealtimeBroadcaster, RealtimeBroadcaster, RealtimeEvent};
|
||||
pub use realtime_authority::{DenyAllRealtimeAuthority, RealtimeAuthority, SubscribeDenied};
|
||||
pub use route::{DispatchMode, HostKind, PathKind, Route};
|
||||
|
||||
103
crates/shared/src/queue.rs
Normal file
103
crates/shared/src/queue.rs
Normal file
@@ -0,0 +1,103 @@
|
||||
//! `QueueService` — the v1.1.9 durable queue contract.
|
||||
//!
|
||||
//! `queue::enqueue(name, message, opts?)` writes one row to
|
||||
//! `queue_messages`; the dispatcher's queue arm claims it later via
|
||||
//! `FOR UPDATE SKIP LOCKED` and fires the registered `queue:receive`
|
||||
//! trigger (commit 6). On handler success the row is deleted (ack);
|
||||
//! on throw it's nacked (claim cleared, retry per the trigger's policy
|
||||
//! with `deliver_after = NOW() + backoff`); when `attempt >= max_attempts`
|
||||
//! the message is moved to `dead_letters`.
|
||||
//!
|
||||
//! No `peek`/`dequeue`/`purge` script-side surface — consumers are
|
||||
//! triggers. Exposing manual dequeue would force PiCloud to expose
|
||||
//! visibility-timeout + nack semantics to script-land, which is the
|
||||
//! whole reason queue is trigger-based.
|
||||
//!
|
||||
//! Identity tuple: `(cx.app_id, queue_name)`. Cross-app isolation is
|
||||
//! enforced because every method derives `app_id` from `cx.app_id` —
|
||||
//! never from a script argument.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::{QueueMessageId, SdkCallCx};
|
||||
|
||||
/// Optional per-enqueue overrides passed to [`QueueService::enqueue`].
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct EnqueueOpts {
|
||||
/// Defer delivery until `NOW() + delay_ms`. Default: immediate.
|
||||
pub delay_ms: Option<i64>,
|
||||
/// Override the default `max_attempts` (3) for THIS message. Clamped
|
||||
/// to `[1, 20]` at the SDK boundary so a script can't enqueue a
|
||||
/// retry-forever message.
|
||||
pub max_attempts: Option<u32>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait QueueService: Send + Sync {
|
||||
/// Enqueue one message onto `(cx.app_id, queue_name)`. Returns the
|
||||
/// new message id (forensic; not surfaced to scripts in the
|
||||
/// script-side return).
|
||||
async fn enqueue(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
queue_name: &str,
|
||||
payload: serde_json::Value,
|
||||
opts: EnqueueOpts,
|
||||
) -> Result<QueueMessageId, QueueError>;
|
||||
|
||||
/// Total queued messages (claimed + unclaimed + retrying-but-not-deleted)
|
||||
/// for `(cx.app_id, queue_name)`. Surfaces as `queue::depth(name)`.
|
||||
async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError>;
|
||||
|
||||
/// Currently-claimable count — `claim_token IS NULL` AND `deliver_after`
|
||||
/// has passed. Surfaces as `queue::depth_pending(name)`. Diverges from
|
||||
/// `depth` for delayed / in-flight messages.
|
||||
async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError>;
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum QueueError {
|
||||
#[error("queue_name must not be empty")]
|
||||
EmptyName,
|
||||
|
||||
/// Payload could not be serialized (e.g. a Rhai FnPtr / closure
|
||||
/// captured into the message). Rejected at the SDK boundary; surface
|
||||
/// the wording verbatim so scripts see the documented message.
|
||||
#[error("queue rejected: {0}")]
|
||||
Rejected(String),
|
||||
|
||||
/// max_attempts (or delay_ms) outside the allowed bounds.
|
||||
#[error("{0}")]
|
||||
InvalidOpts(String),
|
||||
|
||||
/// Backend unavailable (Postgres down, etc.).
|
||||
#[error("queue backend error: {0}")]
|
||||
Unavailable(String),
|
||||
}
|
||||
|
||||
/// Test-only stub: every call errors `Unavailable`. Used by harnesses
|
||||
/// that build a `Services` bundle without a database.
|
||||
#[derive(Debug, Default, Clone, Copy)]
|
||||
pub struct NoopQueueService;
|
||||
|
||||
#[async_trait]
|
||||
impl QueueService for NoopQueueService {
|
||||
async fn enqueue(
|
||||
&self,
|
||||
_cx: &SdkCallCx,
|
||||
_queue_name: &str,
|
||||
_payload: serde_json::Value,
|
||||
_opts: EnqueueOpts,
|
||||
) -> Result<QueueMessageId, QueueError> {
|
||||
Err(QueueError::Unavailable("queue is not wired in".into()))
|
||||
}
|
||||
|
||||
async fn depth(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
|
||||
Err(QueueError::Unavailable("queue is not wired in".into()))
|
||||
}
|
||||
|
||||
async fn depth_pending(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
|
||||
Err(QueueError::Unavailable("queue is not wired in".into()))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user