From ccd3644aa413c93b55e332659721ea6516915c2b Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Thu, 2 Jul 2026 22:08:21 +0200 Subject: [PATCH] feat(shared-queues): group-keyed queue store + enqueue service + SDK (D3.1) The producer side of shared durable queues: - migration 0065 group_queue_messages (mirrors 0034, keyed by (group_id, collection); CASCADE on group delete). - GroupQueueRepo/PostgresGroupQueueRepo: enqueue + the competing-consumer claim (FOR UPDATE SKIP LOCKED) + ack/nack/drop_exhausted/reclaim/depth. - GroupQueueService trait (shared) + GroupQueueServiceImpl: resolve owning group (kind='queue') from cx.app_id's chain, require editor+ (GroupQueueEnqueue, fails closed on anon), size-cap, enqueue. - SDK: `queue::shared_collection("name")` -> GroupQueueHandle with `.enqueue(msg[, opts])` / `.depth()` / `.depth_pending()`; wired through Services + the picloud binary. - authz: Capability::GroupQueueEnqueue(GroupId), editor+ / script:write. Deterministic test proves competing consumers claim each message exactly once. Consumption wiring (materialized consumers + dispatcher branch) lands in D3.2. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/executor-core/src/sdk/queue.rs | 103 +++++- .../migrations/0065_group_queues.sql | 46 +++ crates/manager-core/src/authz.rs | 14 +- crates/manager-core/src/group_queue_repo.rs | 294 ++++++++++++++++++ .../manager-core/src/group_queue_service.rs | 152 +++++++++ crates/manager-core/src/lib.rs | 6 + crates/manager-core/tests/expected_schema.txt | 24 ++ crates/manager-core/tests/group_queue.rs | 141 +++++++++ crates/picloud/src/lib.rs | 45 +-- crates/shared/src/group_queue.rs | 103 ++++++ crates/shared/src/lib.rs | 2 + crates/shared/src/services.rs | 27 +- 12 files changed, 929 insertions(+), 28 deletions(-) create mode 100644 crates/manager-core/migrations/0065_group_queues.sql create mode 100644 crates/manager-core/src/group_queue_repo.rs create mode 100644 crates/manager-core/src/group_queue_service.rs create mode 100644 crates/manager-core/tests/group_queue.rs create mode 100644 crates/shared/src/group_queue.rs diff --git a/crates/executor-core/src/sdk/queue.rs b/crates/executor-core/src/sdk/queue.rs index 37b721e..245c0c9 100644 --- a/crates/executor-core/src/sdk/queue.rs +++ b/crates/executor-core/src/sdk/queue.rs @@ -20,7 +20,9 @@ use std::sync::Arc; use base64::engine::general_purpose::STANDARD; use base64::Engine as _; -use picloud_shared::{EnqueueOpts, QueueError, QueueService, SdkCallCx, Services}; +use picloud_shared::{ + EnqueueOpts, GroupQueueService, QueueError, QueueService, SdkCallCx, Services, +}; use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; use serde_json::Value as Json; use tokio::runtime::Handle as TokioHandle; @@ -87,7 +89,106 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result> { + if name.is_empty() { + return Err("queue::shared_collection name must not be empty".into()); + } + Ok(GroupQueueHandle { + collection: name.to_string(), + service: group_svc.clone(), + cx: cx.clone(), + }) + }, + ); + } + engine.register_static_module("queue", module.into()); + engine.register_type_with_name::("GroupQueueHandle"); + register_shared_queue_methods(engine); +} + +/// §11.6 D3 handle returned by `queue::shared_collection("name")`. +#[derive(Clone)] +pub struct GroupQueueHandle { + collection: String, + service: Arc, + cx: Arc, +} + +fn group_enqueue( + h: &mut GroupQueueHandle, + message: Dynamic, + opts: EnqueueOpts, +) -> Result<(), Box> { + let json = message_to_json(&message)?; + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("queue: no tokio runtime available: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let service = h.service.clone(); + let cx = h.cx.clone(); + let collection = h.collection.clone(); + handle + .block_on(async move { service.enqueue(&cx, &collection, json, opts).await }) + .map(|_id| ()) + .map_err(|err| -> Box { + EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() + }) +} + +fn group_depth(h: &mut GroupQueueHandle, f: F) -> Result> +where + F: FnOnce(Arc, Arc, String) -> Fut, + Fut: std::future::Future>, +{ + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("queue: no tokio runtime available: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let fut = f(h.service.clone(), h.cx.clone(), h.collection.clone()); + let n = handle.block_on(fut).map_err(|err| -> Box { + EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() + })?; + Ok(i64::try_from(n).unwrap_or(i64::MAX)) +} + +fn register_shared_queue_methods(engine: &mut RhaiEngine) { + engine.register_fn("enqueue", |h: &mut GroupQueueHandle, message: Dynamic| { + group_enqueue(h, message, EnqueueOpts::default()) + }); + engine.register_fn( + "enqueue", + |h: &mut GroupQueueHandle, message: Dynamic, opts: Map| { + let opts = parse_opts(&opts)?; + group_enqueue(h, message, opts) + }, + ); + engine.register_fn("depth", |h: &mut GroupQueueHandle| { + group_depth( + h, + |svc, cx, coll| async move { svc.depth(&cx, &coll).await }, + ) + }); + engine.register_fn("depth_pending", |h: &mut GroupQueueHandle| { + group_depth(h, |svc, cx, coll| async move { + svc.depth_pending(&cx, &coll).await + }) + }); } /// Run the async enqueue inside the synchronous Rhai context. Mirrors diff --git a/crates/manager-core/migrations/0065_group_queues.sql b/crates/manager-core/migrations/0065_group_queues.sql new file mode 100644 index 0000000..d91266d --- /dev/null +++ b/crates/manager-core/migrations/0065_group_queues.sql @@ -0,0 +1,46 @@ +-- §11.6 D3: group-shared durable queues. +-- +-- The queue counterpart to the group_kv/docs/files shared stores (0053-0055). +-- A group declares a `queue` shared collection (kind admitted in 0064); any +-- subtree app enqueues into ONE group-owned store via +-- `queue::shared_collection(name).enqueue(...)`. Consumption is by COMPETING +-- CONSUMERS: a group `[[triggers.queue]] shared = true` template materializes a +-- consumer copy per descendant app (like the M5 stateful templates), and all +-- copies claim from this shared store with FOR UPDATE SKIP LOCKED — each message +-- delivered at-most-once across the whole subtree, scaling horizontally. +-- +-- Identity tuple: (group_id, collection). Keyed by the OWNING GROUP, not an app +-- (the shared rows belong to the group). CASCADE on group delete (data dies with +-- its group; an app delete leaves it), matching the other group stores. Mirrors +-- queue_messages (0034) column-for-column, swapping (app_id, queue_name) for +-- (group_id, collection). + +CREATE TABLE group_queue_messages ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + group_id UUID NOT NULL REFERENCES groups(id) ON DELETE CASCADE, + collection TEXT NOT NULL, + payload JSONB NOT NULL, + enqueued_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deliver_after TIMESTAMPTZ NULL, + claim_token UUID NULL, + claimed_at TIMESTAMPTZ NULL, + attempt INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 3, + enqueued_by_principal UUID NULL +); + +-- Dispatch hot path: scan for unclaimed messages in (group_id, collection) +-- order by enqueued_at (the NOW() deliver_after filter is applied on the small +-- partial result, as in 0034). +CREATE INDEX idx_group_queue_messages_dispatch + ON group_queue_messages (group_id, collection, enqueued_at) + WHERE claim_token IS NULL; + +-- depth() / depth_pending() helpers. +CREATE INDEX idx_group_queue_messages_group_collection + ON group_queue_messages (group_id, collection); + +-- Reclaim-task scan: currently-leased messages past their visibility timeout. +CREATE INDEX idx_group_queue_messages_claimed + ON group_queue_messages (claimed_at) + WHERE claim_token IS NOT NULL; diff --git a/crates/manager-core/src/authz.rs b/crates/manager-core/src/authz.rs index 52fca83..6eb14b0 100644 --- a/crates/manager-core/src/authz.rs +++ b/crates/manager-core/src/authz.rs @@ -170,6 +170,10 @@ pub enum Capability { /// group; fails closed for an anonymous principal, like `GroupKvWrite` — a /// publish is a write to shared state. GroupPubsubPublish(GroupId), + /// Enqueue into a group-owned shared QUEUE (§11.6 D3). editor+ on the owning + /// group; fails closed for an anonymous principal, like `GroupKvWrite` — an + /// enqueue is a write to shared state. + GroupQueueEnqueue(GroupId), /// Send an outbound email from a script in this app (v1.1.7). Maps /// to `script:write` on API keys (sending mail is an outbound /// side-effect like an HTTP request). Granted to `editor`+. @@ -243,7 +247,8 @@ impl Capability { | Self::GroupDocsWrite(_) | Self::GroupFilesRead(_) | Self::GroupFilesWrite(_) - | Self::GroupPubsubPublish(_) => None, + | Self::GroupPubsubPublish(_) + | Self::GroupQueueEnqueue(_) => None, Self::AppRead(id) | Self::AppWriteScript(id) | Self::AppWriteRoute(id) @@ -321,6 +326,7 @@ impl Capability { | Self::GroupDocsWrite(_) | Self::GroupFilesWrite(_) | Self::GroupPubsubPublish(_) + | Self::GroupQueueEnqueue(_) | Self::AppInvoke(_) => Scope::ScriptWrite, Self::AppWriteRoute(_) => Scope::RouteWrite, Self::AppManageDomains(_) => Scope::DomainManage, @@ -543,7 +549,8 @@ async fn role_grants( | Capability::GroupDocsWrite(g) | Capability::GroupFilesRead(g) | Capability::GroupFilesWrite(g) - | Capability::GroupPubsubPublish(g) => { + | Capability::GroupPubsubPublish(g) + | Capability::GroupQueueEnqueue(g) => { group_member_grants(repo, principal.user_id, cap, g).await } // Creating a root-level group is an instance act — members @@ -618,7 +625,8 @@ const fn group_role_satisfies(role: AppRole, cap: Capability) -> bool { | Capability::GroupKvWrite(_) | Capability::GroupDocsWrite(_) | Capability::GroupFilesWrite(_) - | Capability::GroupPubsubPublish(_) => { + | Capability::GroupPubsubPublish(_) + | Capability::GroupQueueEnqueue(_) => { matches!(role, AppRole::Editor | AppRole::AppAdmin) } // group_admin manages the group + reads secret VALUES (the diff --git a/crates/manager-core/src/group_queue_repo.rs b/crates/manager-core/src/group_queue_repo.rs new file mode 100644 index 0000000..d1755ca --- /dev/null +++ b/crates/manager-core/src/group_queue_repo.rs @@ -0,0 +1,294 @@ +//! `GroupQueueRepo` — CRUD over `group_queue_messages`, the §11.6 D3 group- +//! shared durable queue store. +//! +//! The group counterpart to [`crate::queue_repo::QueueRepo`], keyed by +//! `(group_id, collection)` instead of `(app_id, queue_name)`. Producers enqueue +//! via `GroupQueueServiceImpl` (`queue::shared_collection(name).enqueue(...)`); +//! the dispatcher's queue arm claims from here for a materialized shared-queue +//! consumer (competing consumers — every descendant app runs a consumer copy, +//! all claiming this one store with `FOR UPDATE SKIP LOCKED`, so each message is +//! delivered at-most-once across the subtree). +//! +//! Deferred (documented): shared-queue dead-lettering — an exhausted message is +//! nacked with backoff by the dispatcher (never silently dropped), but there is +//! no group dead-letter store yet. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use picloud_shared::{AdminUserId, GroupId, QueueMessageId}; +use sqlx::PgPool; +use uuid::Uuid; + +/// Cap on `depth`/`depth_pending` scans (mirrors `queue_repo`). +const QUEUE_DEPTH_SCAN_CAP: i64 = 10_000; + +#[derive(Debug, thiserror::Error)] +pub enum GroupQueueRepoError { + #[error("database error: {0}")] + Db(#[from] sqlx::Error), +} + +/// Insert payload — what `GroupQueueService::enqueue` hands the repo. +#[derive(Debug, Clone)] +pub struct NewGroupQueueMessage { + pub group_id: GroupId, + pub collection: String, + pub payload: serde_json::Value, + pub deliver_after: Option>, + pub max_attempts: u32, + pub enqueued_by_principal: Option, +} + +/// One claimed message ready for handler dispatch. +#[derive(Debug, Clone)] +pub struct ClaimedGroupMessage { + pub id: QueueMessageId, + pub group_id: GroupId, + pub collection: String, + pub payload: serde_json::Value, + pub enqueued_at: DateTime, + pub attempt: u32, + pub max_attempts: u32, + pub claim_token: Uuid, +} + +#[async_trait] +pub trait GroupQueueRepo: Send + Sync { + async fn enqueue( + &self, + msg: NewGroupQueueMessage, + ) -> Result; + + /// Atomic claim of one ready message from `(group_id, collection)` with + /// `FOR UPDATE SKIP LOCKED` — the competing-consumer primitive. `Ok(None)` + /// when nothing is claimable. + async fn claim( + &self, + group_id: GroupId, + collection: &str, + ) -> Result, GroupQueueRepoError>; + + /// Handler succeeded: delete the row iff `claim_token` matches. + async fn ack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + ) -> Result; + + /// Handler threw: clear the claim, defer redelivery by `retry_delay`. + async fn nack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result; + + /// Drop a message that exhausted its attempts (no group dead-letter store + /// yet — see the module deferral note). Deletes iff `claim_token` matches. + async fn drop_exhausted( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + ) -> Result; + + /// Periodic safety net: clear claims older than a shared-queue consumer's + /// `visibility_timeout_secs`. Returns the number of rows reclaimed. + async fn reclaim_visibility_timeouts(&self) -> Result; + + async fn depth(&self, group_id: GroupId, collection: &str) -> Result; + + async fn depth_pending( + &self, + group_id: GroupId, + collection: &str, + ) -> Result; +} + +pub struct PostgresGroupQueueRepo { + pool: PgPool, +} + +impl PostgresGroupQueueRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl GroupQueueRepo for PostgresGroupQueueRepo { + async fn enqueue( + &self, + msg: NewGroupQueueMessage, + ) -> Result { + let (id,): (Uuid,) = sqlx::query_as( + "INSERT INTO group_queue_messages ( \ + group_id, collection, payload, deliver_after, \ + max_attempts, enqueued_by_principal \ + ) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id", + ) + .bind(msg.group_id.into_inner()) + .bind(&msg.collection) + .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, + group_id: GroupId, + collection: &str, + ) -> Result, GroupQueueRepoError> { + let token = Uuid::new_v4(); + let row: Option = sqlx::query_as( + "UPDATE group_queue_messages \ + SET claim_token = $3, claimed_at = NOW(), attempt = attempt + 1 \ + WHERE id = ( \ + SELECT id FROM group_queue_messages \ + WHERE group_id = $1 AND collection = $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, group_id, collection, payload, enqueued_at, attempt, max_attempts", + ) + .bind(group_id.into_inner()) + .bind(collection) + .bind(token) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| ClaimedGroupMessage { + id: r.id.into(), + group_id: r.group_id.into(), + collection: r.collection, + 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 { + let res = + sqlx::query("DELETE FROM group_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 { + let next = Utc::now() + retry_delay; + let res = sqlx::query( + "UPDATE group_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 drop_exhausted( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + ) -> Result { + let res = + sqlx::query("DELETE FROM group_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 reclaim_visibility_timeouts(&self) -> Result { + // A shared-queue consumer is a materialized app-owned trigger + // (`shared = TRUE`, `group_id` on the SOURCE template) whose + // queue_trigger_details.queue_name IS the shared collection name. Join + // the group template to recover the owning group + visibility timeout. + let res = sqlx::query( + "UPDATE group_queue_messages m \ + SET claim_token = NULL, claimed_at = NULL \ + FROM triggers copy \ + JOIN triggers tmpl ON tmpl.id = copy.materialized_from \ + JOIN queue_trigger_details d ON d.trigger_id = copy.id \ + WHERE tmpl.group_id = m.group_id \ + AND d.queue_name = m.collection \ + AND tmpl.shared = TRUE \ + AND m.claim_token IS NOT NULL \ + AND m.claimed_at < NOW() - (d.visibility_timeout_secs || ' seconds')::INTERVAL", + ) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) + } + + async fn depth(&self, group_id: GroupId, collection: &str) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM ( \ + SELECT 1 FROM group_queue_messages \ + WHERE group_id = $1 AND collection = $2 LIMIT $3 \ + ) sub", + ) + .bind(group_id.into_inner()) + .bind(collection) + .bind(QUEUE_DEPTH_SCAN_CAP) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } + + async fn depth_pending( + &self, + group_id: GroupId, + collection: &str, + ) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM ( \ + SELECT 1 FROM group_queue_messages \ + WHERE group_id = $1 AND collection = $2 \ + AND claim_token IS NULL \ + AND (deliver_after IS NULL OR deliver_after <= NOW()) LIMIT $3 \ + ) sub", + ) + .bind(group_id.into_inner()) + .bind(collection) + .bind(QUEUE_DEPTH_SCAN_CAP) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } +} + +#[derive(sqlx::FromRow)] +struct ClaimedRow { + id: Uuid, + group_id: Uuid, + collection: String, + payload: serde_json::Value, + enqueued_at: DateTime, + attempt: i32, + max_attempts: i32, +} diff --git a/crates/manager-core/src/group_queue_service.rs b/crates/manager-core/src/group_queue_service.rs new file mode 100644 index 0000000..0e28180 --- /dev/null +++ b/crates/manager-core/src/group_queue_service.rs @@ -0,0 +1,152 @@ +//! `GroupQueueServiceImpl` — wires `GroupQueueRepo` + the group-collection +//! registry underneath the `picloud_shared::GroupQueueService` trait that +//! scripts reach via the `queue::shared_collection("name")` Rhai handle +//! (§11.6 D3). +//! +//! Resolves the OWNING GROUP (kind `queue`) from `cx.app_id`'s chain (the +//! isolation boundary), enforces editor+ (`GroupQueueEnqueue`, fails closed on +//! anon — enqueue is a shared write), caps the payload size, then writes to the +//! group-keyed store. Consumption is out of band (competing per-descendant +//! materialized consumers, in the dispatcher). + +use std::sync::Arc; + +use async_trait::async_trait; +use chrono::Utc; +use picloud_shared::{ + EnqueueOpts, GroupId, GroupQueueError, GroupQueueService, QueueMessageId, SdkCallCx, +}; + +use crate::authz::{self, AuthzRepo, Capability}; +use crate::group_collection_repo::GroupCollectionResolver; +use crate::group_queue_repo::{GroupQueueRepo, NewGroupQueueMessage}; +use crate::queue_service::queue_max_payload_bytes_from_env; + +/// The registry `kind` this service resolves. +const KIND_QUEUE: &str = "queue"; + +pub struct GroupQueueServiceImpl { + repo: Arc, + resolver: Arc, + authz: Arc, + max_payload_bytes: usize, +} + +impl GroupQueueServiceImpl { + #[must_use] + pub fn new( + repo: Arc, + resolver: Arc, + authz: Arc, + ) -> Self { + Self { + repo, + resolver, + authz, + max_payload_bytes: queue_max_payload_bytes_from_env(), + } + } + + async fn owning_group( + &self, + cx: &SdkCallCx, + collection: &str, + ) -> Result { + if collection.is_empty() { + return Err(GroupQueueError::InvalidCollection); + } + self.resolver + .resolve_owning_group(cx.app_id, collection, KIND_QUEUE) + .await + .map_err(|e| GroupQueueError::Backend(e.to_string()))? + .ok_or_else(|| GroupQueueError::CollectionNotShared(collection.to_string())) + } + + async fn check_write(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupQueueError> { + authz::script_gate_require_principal( + &*self.authz, + cx, + Capability::GroupQueueEnqueue(group_id), + || GroupQueueError::Forbidden, + GroupQueueError::Backend, + ) + .await + } +} + +#[async_trait] +impl GroupQueueService for GroupQueueServiceImpl { + async fn enqueue( + &self, + cx: &SdkCallCx, + collection: &str, + payload: serde_json::Value, + opts: EnqueueOpts, + ) -> Result { + if collection.is_empty() { + return Err(GroupQueueError::InvalidCollection); + } + // Size cap first (no DB) — a cheap reject before resolve/authz. + let encoded_len = serde_json::to_vec(&payload) + .map(|v| v.len()) + .map_err(|e| GroupQueueError::Backend(format!("encode payload: {e}")))?; + if encoded_len > self.max_payload_bytes { + return Err(GroupQueueError::PayloadTooLarge { + limit: self.max_payload_bytes, + actual: encoded_len, + }); + } + let max_attempts = opts.max_attempts.unwrap_or(3); + if !(1..=20).contains(&max_attempts) { + return Err(GroupQueueError::InvalidOpts( + "queue::enqueue: max_attempts must be in [1, 20]".into(), + )); + } + if let Some(delay) = opts.delay_ms { + if !(0..=86_400_000).contains(&delay) { + return Err(GroupQueueError::InvalidOpts( + "queue::enqueue: delay_ms must be in [0, 86_400_000]".into(), + )); + } + } + let group_id = self.owning_group(cx, collection).await?; + self.check_write(cx, group_id).await?; + + let deliver_after = opts + .delay_ms + .and_then(chrono::Duration::try_milliseconds) + .map(|d| Utc::now() + d); + self.repo + .enqueue(NewGroupQueueMessage { + group_id, + collection: collection.to_string(), + payload, + deliver_after, + max_attempts, + enqueued_by_principal: cx.principal.as_ref().map(|p| p.user_id), + }) + .await + .map_err(|e| GroupQueueError::Backend(e.to_string())) + } + + async fn depth(&self, cx: &SdkCallCx, collection: &str) -> Result { + let group_id = self.owning_group(cx, collection).await?; + // Depth is a read — reads are open (any subtree script). + self.repo + .depth(group_id, collection) + .await + .map_err(|e| GroupQueueError::Backend(e.to_string())) + } + + async fn depth_pending( + &self, + cx: &SdkCallCx, + collection: &str, + ) -> Result { + let group_id = self.owning_group(cx, collection).await?; + self.repo + .depth_pending(group_id, collection) + .await + .map_err(|e| GroupQueueError::Backend(e.to_string())) + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index c2bea20..ff404a4 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -60,6 +60,8 @@ pub mod group_kv_repo; pub mod group_kv_service; pub mod group_members_repo; pub mod group_pubsub_service; +pub mod group_queue_repo; +pub mod group_queue_service; pub mod group_quota; pub mod group_repo; pub mod group_scripts_api; @@ -200,6 +202,10 @@ pub use group_members_repo::{ PostgresGroupMembersRepository, }; pub use group_pubsub_service::GroupPubsubServiceImpl; +pub use group_queue_repo::{ + GroupQueueRepo, GroupQueueRepoError, NewGroupQueueMessage, PostgresGroupQueueRepo, +}; +pub use group_queue_service::GroupQueueServiceImpl; pub use group_repo::{ GroupChildCounts, GroupRepository, GroupRepositoryError, PostgresGroupRepository, ROOT_GROUP_SLUG, diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index 44ba662..4d721a4 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/crates/manager-core/tests/expected_schema.txt @@ -264,6 +264,19 @@ table: group_members role: text NOT NULL created_at: timestamp with time zone NOT NULL default=now() +table: group_queue_messages + id: uuid NOT NULL default=gen_random_uuid() + group_id: uuid NOT NULL + collection: text NOT NULL + payload: jsonb NOT NULL + enqueued_at: timestamp with time zone NOT NULL default=now() + deliver_after: timestamp with time zone NULL + claim_token: uuid NULL + claimed_at: timestamp with time zone NULL + attempt: integer NOT NULL default=0 + max_attempts: integer NOT NULL default=3 + enqueued_by_principal: uuid NULL + table: groups id: uuid NOT NULL default=gen_random_uuid() parent_id: uuid NULL @@ -559,6 +572,12 @@ indexes on group_members: group_members_pkey: public.group_members USING btree (group_id, user_id) group_members_user_id_idx: public.group_members USING btree (user_id) +indexes on group_queue_messages: + group_queue_messages_pkey: public.group_queue_messages USING btree (id) + idx_group_queue_messages_claimed: public.group_queue_messages USING btree (claimed_at) WHERE (claim_token IS NOT NULL) + idx_group_queue_messages_dispatch: public.group_queue_messages USING btree (group_id, collection, enqueued_at) WHERE (claim_token IS NULL) + idx_group_queue_messages_group_collection: public.group_queue_messages USING btree (group_id, collection) + indexes on groups: groups_parent_id_idx: public.groups USING btree (parent_id) groups_pkey: public.groups USING btree (id) @@ -793,6 +812,10 @@ constraints on group_members: [FOREIGN KEY] group_members_user_id_fkey: FOREIGN KEY (user_id) REFERENCES admin_users(id) ON DELETE CASCADE [PRIMARY KEY] group_members_pkey: PRIMARY KEY (group_id, user_id) +constraints on group_queue_messages: + [FOREIGN KEY] group_queue_messages_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE + [PRIMARY KEY] group_queue_messages_pkey: PRIMARY KEY (id) + constraints on groups: [FOREIGN KEY] groups_parent_id_fkey: FOREIGN KEY (parent_id) REFERENCES groups(id) ON DELETE RESTRICT [PRIMARY KEY] groups_pkey: PRIMARY KEY (id) @@ -949,3 +972,4 @@ constraints on vars: 0062: materialized triggers 0063: materialized unique 0064: shared topic queue kinds + 0065: group queues diff --git a/crates/manager-core/tests/group_queue.rs b/crates/manager-core/tests/group_queue.rs new file mode 100644 index 0000000..8820075 --- /dev/null +++ b/crates/manager-core/tests/group_queue.rs @@ -0,0 +1,141 @@ +//! §11.6 D3 integration test: group-shared durable queue store. +//! +//! Proves the competing-consumer primitive: N messages enqueued into one +//! group-keyed store are claimed EXACTLY ONCE across concurrent claimers (the +//! `FOR UPDATE SKIP LOCKED` guarantee that makes per-descendant materialized +//! consumers safe). Also checks ack removes a row and nack re-defers it. +//! +//! Deterministic: drives `GroupQueueRepo` directly (no dispatcher). Skips when +//! `DATABASE_URL` is unset. + +#![allow(clippy::too_many_lines)] + +use std::collections::HashSet; + +use picloud_manager_core::group_queue_repo::{ + GroupQueueRepo, NewGroupQueueMessage, PostgresGroupQueueRepo, +}; +use picloud_shared::GroupId; +use sqlx::postgres::PgPoolOptions; +use sqlx::PgPool; +use uuid::Uuid; + +async fn pool_or_skip() -> Option { + let Ok(url) = std::env::var("DATABASE_URL") else { + eprintln!("group_queue: DATABASE_URL unset — skipping"); + return None; + }; + let pool = PgPoolOptions::new() + .max_connections(6) + .connect(&url) + .await + .expect("connect"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("migrate"); + Some(pool) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn competing_consumers_claim_each_message_exactly_once() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let sfx = Uuid::new_v4().simple().to_string(); + let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("gq-g-{sfx}")) + .fetch_one(&pool) + .await + .unwrap(); + let group = GroupId::from(g.0); + let repo = PostgresGroupQueueRepo::new(pool.clone()); + + // Enqueue 50 distinct messages into the shared `tasks` queue. + const N: usize = 50; + for i in 0..N { + repo.enqueue(NewGroupQueueMessage { + group_id: group, + collection: "tasks".into(), + payload: serde_json::json!({ "i": i }), + deliver_after: None, + max_attempts: 3, + enqueued_by_principal: None, + }) + .await + .unwrap(); + } + + // Four concurrent "consumers" claim until the queue is drained, ack-ing each. + // Every claimed payload id must be unique — no message delivered twice. + let claim_loop = |repo: PostgresGroupQueueRepo, group: GroupId| async move { + let mut got: Vec = Vec::new(); + loop { + match repo.claim(group, "tasks").await.unwrap() { + Some(msg) => { + got.push(msg.payload["i"].as_i64().unwrap()); + assert!(repo.ack(msg.id, msg.claim_token).await.unwrap(), "ack ok"); + } + None => break, + } + } + got + }; + let (a, b, c, d) = tokio::join!( + claim_loop(PostgresGroupQueueRepo::new(pool.clone()), group), + claim_loop(PostgresGroupQueueRepo::new(pool.clone()), group), + claim_loop(PostgresGroupQueueRepo::new(pool.clone()), group), + claim_loop(PostgresGroupQueueRepo::new(pool.clone()), group), + ); + + let mut all: Vec = Vec::new(); + all.extend(a); + all.extend(b); + all.extend(c); + all.extend(d); + let unique: HashSet = all.iter().copied().collect(); + assert_eq!( + all.len(), + N, + "every message delivered exactly once (no dupes)" + ); + assert_eq!(unique.len(), N, "all N distinct ids covered"); + assert_eq!( + repo.depth(group, "tasks").await.unwrap(), + 0, + "queue drained" + ); + + // nack re-defers a message (a later claim gets it back). + let id = repo + .enqueue(NewGroupQueueMessage { + group_id: group, + collection: "tasks".into(), + payload: serde_json::json!({ "i": 999 }), + deliver_after: None, + max_attempts: 3, + enqueued_by_principal: None, + }) + .await + .unwrap(); + let msg = repo.claim(group, "tasks").await.unwrap().expect("claimed"); + assert_eq!(msg.id, id); + repo.nack(msg.id, msg.claim_token, chrono::Duration::milliseconds(0)) + .await + .unwrap(); + // After nack (0ms delay) it is claimable again. + let again = repo.claim(group, "tasks").await.unwrap().expect("re-claim"); + assert_eq!(again.id, id, "nacked message is re-delivered"); + assert_eq!(again.attempt, 2, "attempt incremented on re-claim"); + repo.ack(again.id, again.claim_token).await.unwrap(); + + // Cleanup (messages cascade on group delete anyway). + let _ = sqlx::query("DELETE FROM group_queue_messages WHERE group_id = $1") + .bind(g.0) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM groups WHERE id = $1") + .bind(g.0) + .execute(&pool) + .await; +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index f7f7f42..6161e0f 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -20,23 +20,23 @@ use picloud_manager_core::{ AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, FsGroupFilesRepo, GroupDocsServiceImpl, GroupFilesServiceImpl, - GroupKvServiceImpl, GroupMembersRepository, GroupPubsubServiceImpl, GroupRepository, - GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, - OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, - PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository, - PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo, - PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, - PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, - PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo, - PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresGroupCollectionResolver, - PostgresGroupDocsRepo, PostgresGroupKvRepo, PostgresGroupMembersRepository, - PostgresGroupRepository, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, - PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, - PostgresTriggerRepo, PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl, - RealtimeAuthorityImpl, RepoResolver, RouteAdminState, SandboxCeiling, ScriptRepository, - SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, - TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, - VarsServiceImpl, + GroupKvServiceImpl, GroupMembersRepository, GroupPubsubServiceImpl, GroupQueueServiceImpl, + GroupRepository, GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, + KvServiceImpl, OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, + PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository, + PostgresAppDomainRepository, PostgresAppMembersRepository, PostgresAppRepository, + PostgresAppSecretsRepo, PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, + PostgresAppUserRepository, PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, + PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, PostgresDeadLetterService, + PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink, + PostgresGroupCollectionResolver, PostgresGroupDocsRepo, PostgresGroupKvRepo, + PostgresGroupMembersRepository, PostgresGroupQueueRepo, PostgresGroupRepository, + PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, + PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, + PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver, + RouteAdminState, SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl, + SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo, + TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl, }; use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS; use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable}; @@ -268,6 +268,14 @@ pub async fn build_app( Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), )); + // §11.6 D3: the group shared-queue store + enqueue service. + let group_queue_repo = Arc::new(PostgresGroupQueueRepo::new(pool.clone())); + let group_queue: Arc = + Arc::new(GroupQueueServiceImpl::new( + group_queue_repo.clone(), + Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), + authz.clone(), + )); let pubsub: Arc = Arc::new( PubsubServiceImpl::new(pubsub_repo, authz.clone()) .with_max_message_bytes( @@ -393,7 +401,8 @@ pub async fn build_app( .with_group_kv(group_kv) .with_group_docs(group_docs) .with_group_files(group_files) - .with_group_pubsub(group_pubsub); + .with_group_pubsub(group_pubsub) + .with_group_queue(group_queue); // v1.1.9: keep the invoke depth bound aligned with the dispatcher's // trigger-depth bound (same counter under the hood). let engine_limits = Limits { diff --git a/crates/shared/src/group_queue.rs b/crates/shared/src/group_queue.rs new file mode 100644 index 0000000..f461075 --- /dev/null +++ b/crates/shared/src/group_queue.rs @@ -0,0 +1,103 @@ +//! `GroupQueueService` — the §11.6 D3 shared group-QUEUE contract. +//! +//! The durable-queue counterpart to [`crate::GroupKvService`]. A script reaches +//! it via the explicit `queue::shared_collection("name")` handle (distinct from +//! the app-private `queue::enqueue`). Any subtree app ENQUEUES into one group- +//! owned store; CONSUMPTION is by competing per-descendant materialized +//! consumers (handled outside this trait, in the dispatcher). The impl resolves +//! the OWNING GROUP by walking the calling app's ancestor chain for the nearest +//! group that declares the queue shared — the isolation boundary. +//! +//! Trust model (§11.6): enqueue is a WRITE to shared state, so it requires an +//! authenticated principal with editor+ on the owning group (fails closed for +//! an anonymous caller), matching `GroupKvService` writes. + +use async_trait::async_trait; +use thiserror::Error; + +use crate::{EnqueueOpts, QueueMessageId, SdkCallCx}; + +#[async_trait] +pub trait GroupQueueService: Send + Sync { + /// Enqueue `payload` into the group shared queue `collection`. Resolves the + /// owning group (kind `queue`) from `cx.app_id`'s chain and requires editor+. + async fn enqueue( + &self, + cx: &SdkCallCx, + collection: &str, + payload: serde_json::Value, + opts: EnqueueOpts, + ) -> Result; + + /// Total rows in the shared queue. + async fn depth(&self, cx: &SdkCallCx, collection: &str) -> Result; + + /// Currently-claimable rows in the shared queue. + async fn depth_pending(&self, cx: &SdkCallCx, collection: &str) + -> Result; +} + +/// Stub for test bundles that don't exercise shared queues. Every call errors so +/// accidental use surfaces clearly. +#[derive(Debug, Default, Clone, Copy)] +pub struct NoopGroupQueueService; + +#[async_trait] +impl GroupQueueService for NoopGroupQueueService { + async fn enqueue( + &self, + _cx: &SdkCallCx, + _collection: &str, + _payload: serde_json::Value, + _opts: EnqueueOpts, + ) -> Result { + Err(GroupQueueError::Backend( + "group queue is not wired in".into(), + )) + } + + async fn depth(&self, _cx: &SdkCallCx, _collection: &str) -> Result { + Err(GroupQueueError::Backend( + "group queue is not wired in".into(), + )) + } + + async fn depth_pending( + &self, + _cx: &SdkCallCx, + _collection: &str, + ) -> Result { + Err(GroupQueueError::Backend( + "group queue is not wired in".into(), + )) + } +} + +/// Failure modes surfaced to the Rhai bridge. +#[derive(Debug, Error)] +pub enum GroupQueueError { + /// Empty collection name; rejected at the SDK boundary. + #[error("queue name must not be empty")] + InvalidCollection, + + /// No group on the calling app's ancestor chain declares this queue shared. + #[error("queue {0:?} is not a shared group queue for this app")] + CollectionNotShared(String), + + /// Caller lacked the required capability. Enqueue always needs an + /// authenticated editor+ on the owning group (fails closed for anon). + #[error("forbidden")] + Forbidden, + + /// Payload exceeds the per-message JSON-encoded size cap. + #[error("payload too large: {actual} bytes exceeds the {limit}-byte cap")] + PayloadTooLarge { limit: usize, actual: usize }, + + /// Invalid enqueue options (bad max_attempts / delay_ms). + #[error("{0}")] + InvalidOpts(String), + + /// Storage / backend failure. + #[error("backend error: {0}")] + Backend(String), +} diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index ac94c0f..c75d6cf 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -20,6 +20,7 @@ pub mod group_docs; pub mod group_files; pub mod group_kv; pub mod group_pubsub; +pub mod group_queue; pub mod http; pub mod ids; pub mod inbox; @@ -73,6 +74,7 @@ pub use group_docs::{GroupDocsError, GroupDocsService, NoopGroupDocsService}; pub use group_files::{GroupFilesError, GroupFilesService, NoopGroupFilesService}; pub use group_kv::{GroupKvError, GroupKvService, NoopGroupKvService}; pub use group_pubsub::{GroupPubsubError, GroupPubsubService, NoopGroupPubsubService}; +pub use group_queue::{GroupQueueError, GroupQueueService, NoopGroupQueueService}; pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService}; pub use ids::{ AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, QueueMessageId, diff --git a/crates/shared/src/services.rs b/crates/shared/src/services.rs index 4a5c1dc..fc366d5 100644 --- a/crates/shared/src/services.rs +++ b/crates/shared/src/services.rs @@ -21,12 +21,13 @@ use std::sync::Arc; use crate::{ DeadLetterService, DocsService, EmailService, FilesService, GroupDocsService, - GroupFilesService, GroupKvService, GroupPubsubService, HttpService, InvokeService, KvService, - ModuleSource, NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter, - NoopFilesService, NoopGroupDocsService, NoopGroupFilesService, NoopGroupKvService, - NoopGroupPubsubService, NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource, - NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService, NoopVarsService, - PubsubService, QueueService, SecretsService, ServiceEventEmitter, UsersService, VarsService, + GroupFilesService, GroupKvService, GroupPubsubService, GroupQueueService, HttpService, + InvokeService, KvService, ModuleSource, NoopDeadLetterService, NoopDocsService, + NoopEmailService, NoopEventEmitter, NoopFilesService, NoopGroupDocsService, + NoopGroupFilesService, NoopGroupKvService, NoopGroupPubsubService, NoopGroupQueueService, + NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource, NoopPubsubService, + NoopQueueService, NoopSecretsService, NoopUsersService, NoopVarsService, PubsubService, + QueueService, SecretsService, ServiceEventEmitter, UsersService, VarsService, }; /// SDK service bundle. See module docs for the lifecycle and the v1.1.x @@ -139,6 +140,12 @@ pub struct Services { /// `shared = true` group pubsub triggers. Wired via /// [`Services::with_group_pubsub`]; defaults to `NoopGroupPubsubService`. pub group_pubsub: Arc, + + /// Shared cross-app group QUEUES (§11.6 D3). Scripts get + /// `queue::shared_collection(name)` — enqueue into a group-keyed store + /// drained by competing per-descendant consumers. Wired via + /// [`Services::with_group_queue`]; defaults to `NoopGroupQueueService`. + pub group_queue: Arc, } impl Services { @@ -185,6 +192,7 @@ impl Services { group_docs: Arc::new(NoopGroupDocsService), group_files: Arc::new(NoopGroupFilesService), group_pubsub: Arc::new(NoopGroupPubsubService), + group_queue: Arc::new(NoopGroupQueueService), } } @@ -217,6 +225,13 @@ impl Services { self } + /// Set the §11.6 D3 shared group-QUEUE service (picloud binary; tests leave noop). + #[must_use] + pub fn with_group_queue(mut self, group_queue: Arc) -> Self { + self.group_queue = group_queue; + self + } + /// All-noop bundle for tests that build an `Engine` but don't /// exercise the stateful services. Returns the same shape as /// `Services::new` so callers can't accidentally rely on a stub