[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_179":-5,"_182":183,"_184":185},"routes/shell",{"_41":42,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":32,"_44":32,"_45":32,"_46":32,"_47":32,"_48":32,"_49":50,"_51":50,"_52":32,"_53":-7,"_54":-7,"_55":56,"_57":58,"_59":60,"_61":-5,"_62":63,"_64":65,"_66":67,"_68":69},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":22,"_23":24,"_25":-5,"_26":27,"_28":29,"_29":30,"_31":32,"_33":14},"ref","main","path","crates/logic/queue.rs","oid","c36b432ed38555c52359bfff32c3961b8892af36","size",13008,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","// Copyright 2026 Deno Land Inc. Apache-2.0 license.\n\n//! Cloudflare Queue policy, reified sans I/O.\n//!\n//! A queue is one reserved cell. The cell owns SQL and dispatch, while this\n//! module owns the values and transitions that another caller must reproduce:\n//! the stable address, the public bounds, the alarm deadline, concurrency\n//! admission, retry timing, lease-generation advancement, settlement fencing,\n//! purge classification, and deploy-time configuration validation.\n//!\n//! The lease generation travels in [`PlannedLease`]. A caller cannot select a\n//! row and then forget to advance its generation, because this module never\n//! returns a bare sequence number for a deliverable row. That prevents a late\n//! settlement from an expired lease from acknowledging a newer delivery.\n\nuse crate::Ms;\n\n/// The runtime-supplied Durable Object class that stores every queue.\npub const RESERVED_CLASS: &str = \"__Queue\";\n\n/// Cloudflare measures a Queue kilobyte as 1000 bytes.\npub const MAX_MESSAGE_BYTES: usize = 128_000;\npub const MAX_SEND_BATCH_BYTES: usize = 256_000;\npub const MAX_BATCH_MESSAGES: usize = 100;\npub const MAX_BATCH_TIMEOUT_SECONDS: u16 = 60;\npub const MAX_RETRIES: u16 = 100;\npub const MAX_CONCURRENCY: u16 = 250;\npub const MAX_DELAY_SECONDS: u32 = 86_400;\n\npub const DEFAULT_MAX_BATCH_SIZE: u16 = 10;\npub const DEFAULT_MAX_BATCH_TIMEOUT_SECONDS: u16 = 5;\npub const DEFAULT_MAX_RETRIES: u16 = 3;\n\n/// Queue retention is deployment-independent in celld v1. Wrangler has no\n/// configuration key for it, so accepting one would make a celld project\n/// invalid on Cloudflare.\npub const RETENTION_MS: Ms = 4 * 24 * 60 * 60 * 1000;\n\n/// The Durable Object name that `getByName` hashes for a queue.\n///\n/// The identity function is intentional. The queue name is already the\n/// fleet-wide resource identity, and prefixing it here would make a project\n/// address a different queue from Cloudflare. Both the binding and the\n/// operator CLI call this function before the runtime applies the namespace\n/// HMAC, so even an identity address has one source.\npub fn cell_name(queue_name: &str) -> &str {\n queue_name\n}\n\n/// Choose the one durable alarm from every Queue deadline.\n///\n/// A deadline already past becomes `now`. SQLite can retain an overdue row\n/// across a crash, and installing its old timestamp as a new alarm can leave\n/// the wake index behind the clock instead of scheduling immediate work.\npub fn rearm(\n now: Ms,\n batch_deadline: Option,\n earliest_visible: Option,\n earliest_lease_expiry: Option,\n next_sweep: Option,\n) -> Option {\n [\n batch_deadline,\n earliest_visible,\n earliest_lease_expiry,\n next_sweep,\n ]\n .into_iter()\n .flatten()\n .min()\n .map(|deadline| deadline.max(now))\n}\n\n/// Resolve a retry deadline from Cloudflare's precedence rule.\n///\n/// A per-message or per-batch delay wins, including an explicit zero. The\n/// configured consumer delay is a fixed default; it does not compound with\n/// the attempt number. Cloudflare documents exponential backoff as code an\n/// application can implement with `message.attempts`, not as broker behavior.\npub fn retry_at(now: Ms, explicit_seconds: Option, configured_seconds: Option) -> Ms {\n let seconds = explicit_seconds.or(configured_seconds).unwrap_or(0);\n now.saturating_add(Ms::from(seconds).saturating_mul(1000))\n}\n\n/// Whether a failed current delivery has spent the configured retry count.\n///\n/// `attempt` is one for the first delivery. `max_retries = 3` therefore allows\n/// attempts one through four, and the fourth failure exhausts the message.\npub fn exhausted(attempt: u16, max_retries: u16) -> bool {\n attempt > max_retries\n}\n\n/// The durable transition owed when one persisted lease expires.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct ExpiredLease {\n /// The delivery that just failed. The first delivery is attempt one.\n pub attempt: u16,\n pub action: ExpiredLeaseAction,\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub enum ExpiredLeaseAction {\n RetryAt(Ms),\n Exhausted,\n DeletePurged,\n}\n\n/// Spend one failed delivery and decide what replaces its expired lease.\n///\n/// Purge wins over retry because a purge is irreversible. Otherwise the same\n/// fixed consumer delay and retry ceiling used by an explicit settlement\n/// apply to a consumer that did not settle before its lease deadline.\npub fn expire_lease(\n now: Ms,\n prior_failures: u16,\n max_retries: u16,\n configured_delay_seconds: Option,\n purge_on_settle: bool,\n) -> ExpiredLease {\n let attempt = prior_failures.saturating_add(1);\n let action = if purge_on_settle {\n ExpiredLeaseAction::DeletePurged\n } else if exhausted(attempt, max_retries) {\n ExpiredLeaseAction::Exhausted\n } else {\n ExpiredLeaseAction::RetryAt(retry_at(now, None, configured_delay_seconds))\n };\n ExpiredLease { attempt, action }\n}\n\n/// The fixed lifetime of one persisted Queue lease.\n///\n/// Admission, handler execution, and settlement each have a runtime-enforced\n/// budget. A lease must cover all three or a live consumer can be reclaimed.\n/// The broker stores the resulting absolute deadline once, so a restart or a\n/// configuration change cannot lengthen an existing lease.\npub fn lease_duration_ms(admission_wait_ms: Ms, handler_budget_ms: Ms, settle_budget_ms: Ms) -> Ms {\n admission_wait_ms\n .saturating_add(handler_budget_ms)\n .saturating_add(settle_budget_ms)\n}\n\n/// The row facts needed to plan one lease transaction.\n///\n/// The SQL query supplies rows in its chosen best-effort delivery order. This\n/// function preserves that order and never upgrades it into a FIFO promise.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct BatchRow {\n pub seq: i64,\n pub visible_at: Ms,\n pub lease_generation: u64,\n pub leased_until: Option,\n /// Purge marks a live lease and lets its consumer finish. The row is\n /// deleted when the lease settles or expires, and is never redelivered.\n pub purge_on_settle: bool,\n}\n\n/// A row and the generation the new consumer must settle.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct PlannedLease {\n pub seq: i64,\n pub generation: u64,\n /// True when the previous lease expired. The cell increments the failed\n /// attempt in the same transaction that installs this generation.\n pub reclaimed: bool,\n}\n\n/// One message identity captured by a Queue lease.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct LeaseMember<'a> {\n pub seq: i64,\n pub message_id: &'a str,\n pub generation: u64,\n}\n\n/// Decide whether a settlement names the complete current lease batch.\n///\n/// A consumer can retain an old settlement after the broker has re-leased a\n/// message. The lease ID selects the candidate rows, and this comparison then\n/// fences every member by its message identity, sequence, and generation. A\n/// complete one-to-one match also rejects a truncated or duplicated batch.\npub fn settlement_matches(current: &[LeaseMember<'_>], submitted: &[LeaseMember<'_>]) -> bool {\n if current.is_empty() || current.len() != submitted.len() {\n return false;\n }\n if current.iter().enumerate().any(|(index, member)| {\n current[..index]\n .iter()\n .any(|prior| prior.message_id == member.message_id)\n }) {\n return false;\n }\n\n let mut matched = vec![false; current.len()];\n for submitted in submitted {\n let Some((index, _)) = current.iter().enumerate().find(|(index, current)| {\n !matched[*index]\n && current.message_id == submitted.message_id\n && current.seq == submitted.seq\n && current.generation == submitted.generation\n }) else {\n return false;\n };\n matched[index] = true;\n }\n true\n}\n\n/// Decide whether another batch lease fits the configured concurrency bound.\n///\n/// `active_batches` counts persisted lease IDs, not completed settlements. A\n/// consumer batch occupies its slot from lease installation until settlement\n/// or expiry removes the lease.\npub fn can_install_lease(active_batches: usize, max_concurrency: u16) -> bool {\n active_batches < usize::from(max_concurrency)\n}\n\n/// The lease fact needed to classify one row during a Queue purge.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct PurgeRow {\n pub seq: i64,\n pub lease_id_present: bool,\n pub leased_until: Option,\n}\n\n/// All mutations one purge transaction owes.\n#[derive(Clone, Debug, Default, PartialEq, Eq)]\npub struct PurgePlan {\n /// Rows with no live lease can be deleted immediately.\n pub delete: Vec,\n /// Rows with a live lease remain until settlement or expiry.\n pub mark_for_settle: Vec,\n}\n\n/// Partition Queue rows without deleting a row under a live consumer.\npub fn purge_plan(now: Ms, rows: &[PurgeRow]) -> PurgePlan {\n let mut plan = PurgePlan::default();\n for row in rows {\n if row.lease_id_present && row.leased_until.is_some_and(|until| until > now) {\n plan.mark_for_settle.push(row.seq);\n } else {\n plan.delete.push(row.seq);\n }\n }\n plan\n}\n\n/// All mutations one batch-planning transaction owes.\n#[derive(Clone, Debug, Default, PartialEq, Eq)]\npub struct BatchPlan {\n pub leases: Vec,\n /// Purged rows remain while a live consumer can still settle them. Once\n /// that lease expires, deletion replaces reclamation and redelivery.\n pub delete_purged: Vec,\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub enum PlanError {\n GenerationExhausted { seq: i64 },\n}\n\nimpl std::fmt::Display for PlanError {\n fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {\n match self {\n Self::GenerationExhausted { seq } => {\n write!(f, \"queue message {seq} exhausted its lease generation\")\n }\n }\n }\n}\n\nimpl std::error::Error for PlanError {}\n\n/// Select visible, unleased rows and reclaim expired leases under a new\n/// generation. A live lease is never returned.\npub fn batch_plan(\n now: Ms,\n rows: &[BatchRow],\n max_batch_size: usize,\n) -> Result {\n let mut plan = BatchPlan::default();\n for row in rows {\n let live_lease = row.leased_until.is_some_and(|until| until > now);\n if row.purge_on_settle {\n if !live_lease {\n plan.delete_purged.push(row.seq);\n }\n continue;\n }\n if plan.leases.len() >= max_batch_size || row.visible_at > now || live_lease {\n continue;\n }\n let generation = row\n .lease_generation\n .checked_add(1)\n .ok_or(PlanError::GenerationExhausted { seq: row.seq })?;\n plan.leases.push(PlannedLease {\n seq: row.seq,\n generation,\n reclaimed: row.leased_until.is_some(),\n });\n }\n Ok(plan)\n}\n\n/// The Wrangler Queue values whose bounds celld enforces at deploy time.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct QueueConfig {\n pub max_batch_size: u16,\n pub max_batch_timeout_seconds: u16,\n pub max_retries: u16,\n pub max_concurrency: Option,\n pub delivery_delay_seconds: u32,\n pub retry_delay_seconds: Option,\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct ConfigError {\n pub field: &'static str,\n pub message: &'static str,\n}\n\nimpl std::fmt::Display for ConfigError {\n fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {\n write!(f, \"{} {}\", self.field, self.message)\n }\n}\n\nimpl std::error::Error for ConfigError {}\n\n/// Validate the Queue bounds that a bad deployment must report up front.\npub fn validate_config(config: &QueueConfig) -> Result<(), ConfigError> {\n let fail = |field, message| ConfigError { field, message };\n if !(1..=MAX_BATCH_MESSAGES as u16).contains(&config.max_batch_size) {\n return Err(fail(\"max_batch_size\", \"must be between 1 and 100\"));\n }\n if config.max_batch_timeout_seconds > MAX_BATCH_TIMEOUT_SECONDS {\n return Err(fail(\n \"max_batch_timeout\",\n \"must be between 0 and 60 seconds\",\n ));\n }\n if config.max_retries > MAX_RETRIES {\n return Err(fail(\"max_retries\", \"must be between 0 and 100\"));\n }\n if config\n .max_concurrency\n .is_some_and(|value| !(1..=MAX_CONCURRENCY).contains(&value))\n {\n return Err(fail(\"max_concurrency\", \"must be between 1 and 250\"));\n }\n if config.delivery_delay_seconds > MAX_DELAY_SECONDS {\n return Err(fail(\n \"delivery_delay\",\n \"must be between 0 and 86400 seconds\",\n ));\n }\n if config\n .retry_delay_seconds\n .is_some_and(|value| value > MAX_DELAY_SECONDS)\n {\n return Err(fail(\"retry_delay\", \"must be between 0 and 86400 seconds\"));\n }\n Ok(())\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[134,135,136,137,138,139,140,141,142,143,144,145,146],"navigation",{"_55":56,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_43":32,"_89":14,"_90":32,"_49":50,"_51":50},"branches",[14],"tags",[70,71,72,73,74,75,76,77,78,79,80,81,82],"cloneUrl","https://git.nexthub.sh/denoland/celld.git","v0.0.1","v0.0.2","v0.1.0","v0.2.0","v0.2.1","v0.3.0","v0.4.0","v0.4.1","v0.5.0","v0.5.1","v0.6.0","v0.6.1","v0.6.2","description","self-hosted, distributed Durable Objects","owner","denoland","section","","head","empty","pins",[116,117,118],"documents",{"_111":112,"_113":114},"code",{"_108":109,"_110":104},"automations",{"_106":50,"_107":-5},"delivery",{"_103":104,"_105":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_115":114},"total",38,"issue",{"_119":110,"_121":132,"_15":110,"_123":133,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_53":125,"_126":32,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_53":125,"_126":32,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_53":125,"_126":32,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_53":125,"_126":32,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_53":32,"_126":32,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_53":32,"_126":32,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_53":32,"_126":32,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_53":32,"_126":32,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_53":32,"_126":32,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_53":32,"_126":32,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_53":32,"_126":32,"_127":32},{"_119":153,"_121":154,"_15":153,"_123":155,"_53":32,"_126":32,"_127":32},{"_119":150,"_121":151,"_15":150,"_123":152,"_53":32,"_126":32,"_127":32},{"_119":147,"_121":148,"_15":147,"_123":149,"_53":32,"_126":32,"_127":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794435047574,"sidebar","expanded"]