diff --git a/CHANGELOG.md b/CHANGELOG.md index 3003c068f..288461586 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Changed - Syncing a periodic job declaration that matches the stored schedule no longer rewrites its `awa.cron_jobs` row or moves `updated_at`, so runtimes that sync on a timer stop generating a row version per schedule per interval. +- **Claim transactions commit asynchronously (0.6 port of #507).** Claims and the claim-cursor advance that follows them run with `synchronous_commit = off`, so a lane's claim head is no longer locked across a WAL fsync. A claim lost to a server crash leaves the job claimable again, consistent with at-least-once delivery; enqueues, completions and every other transaction stay synchronous. The setting is `SET LOCAL` in those transactions only. A post-claim cursor advance now applies only while the claim it follows is visible on the server it runs against, so a retry that lands on a promoted standby cannot move the cursor past a claim that never reached it. A worker still running a claim lost this way is not fenced by its run lease, because the job's next claim reuses it; see [architecture](docs/architecture.md). No schema or configuration change. + +- **Claims reuse the instance's claimer lease (0.6 port of #508).** A queue-storage claim no longer re-reads the queue's claimer target and its own claimer lease on every call; the last read or refreshed lease is reused for up to 250 ms (never within a second of expiry), removing two round trips and two transactions from each claim. On `main` at 4 workers this cut transactions per completed job from 2.35 to 1.39. ### Fixed diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 64a894de7..9b9cf77f4 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -12,7 +12,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::hash::{Hash, Hasher}; use std::sync::atomic::{AtomicBool, AtomicU16, AtomicUsize, Ordering}; use std::sync::Mutex; -use std::time::Duration; +use std::time::{Duration, Instant}; use uuid::Uuid; const DEFAULT_SCHEMA: &str = "awa"; @@ -23,6 +23,10 @@ const DEFAULT_QUEUE_STRIPE_COUNT: usize = 1; const QUEUE_STRIPE_DELIMITER: &str = "#"; const COPY_NULL_SENTINEL: &str = "__AWA_NULL__"; const COPY_CHUNK_TARGET_BYTES: usize = 256 * 1024; +/// Claim and claim-cursor transactions commit without waiting for the WAL +/// flush. A claim lost to a server crash leaves its job claimable again, and +/// any later synchronous commit flushes every earlier claim first. +const CLAIM_TRANSACTION_BEGIN: &str = "BEGIN; SET LOCAL synchronous_commit = off"; const TERMINAL_COUNTER_BUCKETS: i16 = 256; const RECEIPT_RESCUE_BATCH_LIMIT: i64 = 500; const RESCUED_DUPLICATE_ERROR: &str = "rescued as duplicate: unique claim held by a newer job"; @@ -216,6 +220,16 @@ pub struct QueueClaimerLease { pub lease_epoch: i64, } +/// How long a claim may reuse this instance's cached claimer lease before +/// re-reading the lease row and the queue's claimer target. +const CLAIMER_GATE_CACHE_TTL: Duration = Duration::from_millis(250); + +#[derive(Debug, Clone, Copy)] +struct CachedClaimerGate { + checked_at: Instant, + lease: QueueClaimerLeaseRow, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::FromRow)] struct QueueClaimerLeaseRow { claimer_slot: i16, @@ -1400,6 +1414,7 @@ mod claim_cursor_advance_tests { enqueue_shard: 0, next_seq, only_if_current, + evidence: None, } } @@ -2046,12 +2061,32 @@ struct ClaimCursorAdvance { enqueue_shard: i16, next_seq: i64, only_if_current: Option, + /// Claim batch row that must exist for the advance to apply: the cursor + /// only moves past a lane position whose claim is visible on the server + /// the advance runs against. + evidence: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ClaimEvidence { + ready_slot: i32, + ready_generation: i64, + lane_seq: i64, + /// `(job_id, run_lease)` of a claim written straight into `leases`. A + /// receipt claim, with or without a deadline, is proven by its + /// `ready_claim_attempt_batches` row instead. + legacy_attempt: Option<(i64, i64)>, } type ClaimCursorLaneKey = (String, i16, i16); type ConditionalClaimCursorAdvances = BTreeMap; -type GroupedClaimCursorAdvances = - BTreeMap, ConditionalClaimCursorAdvances)>; +type GroupedClaimCursorAdvances = BTreeMap< + ClaimCursorLaneKey, + ( + Option<(i64, Option)>, + ConditionalClaimCursorAdvances, + ), +>; type TerminalCounterKey = (i32, i64, String, i16, i16, i16); struct CancelJobTxResult { @@ -3014,6 +3049,9 @@ pub struct QueueStorage { /// `reset()`. With the default `enqueue_shards = 1`, the cache holds 1 /// and `pick_shard` returns 0 unconditionally. enqueue_shards_cache: Mutex>, + /// This instance's claimer lease per queue, as last read or written, so a + /// claim does not re-read the lease and claimer target on every call. + claimer_gates: Mutex>, /// Lane-presence cache: `(physical_queue, priority, enqueue_shard)` /// triples whose three lane rows (queue_lanes, queue_enqueue_heads, /// queue_claim_heads) we have previously inserted. Skips the three @@ -3072,6 +3110,7 @@ impl QueueStorage { next_stripe_probe: AtomicUsize::new(0), shard_rotor: AtomicU16::new(0), enqueue_shards_cache: Mutex::new(HashMap::new()), + claimer_gates: Mutex::new(HashMap::new()), ensured_lanes: Mutex::new(HashSet::new()), lane_provisioner_present: AtomicBool::new(false), lane_provisioner_warned: AtomicBool::new(false), @@ -4017,15 +4056,19 @@ impl QueueStorage { tx.commit().await.map_err(map_sqlx_error)?; - // queue_lanes was TRUNCATEd above and queue_meta may have a new - // shard configuration for the next round. Clear both caches so - // the next ensure_lane / shard_for_enqueue calls re-observe DB - // state. + // queue_lanes and queue_claimer_leases were TRUNCATEd above and + // queue_meta may have a new shard configuration for the next round. + // Clear the caches so the next ensure_lane / shard_for_enqueue / + // claim calls re-observe DB state. self.clear_lane_cache(); self.enqueue_shards_cache .lock() .expect("enqueue_shards_cache poisoned") .clear(); + self.claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .clear(); Ok(()) } @@ -4528,25 +4571,37 @@ impl QueueStorage { } fn claim_cursor_advances(rows: &[ReadyJobLeaseRow]) -> Vec { - let mut next_by_lane: BTreeMap = BTreeMap::new(); + let mut last_by_lane: BTreeMap = BTreeMap::new(); for row in rows { let key = (row.queue.clone(), row.lane_priority, row.enqueue_shard); - let next = row.lane_seq + 1; - next_by_lane + last_by_lane .entry(key) - .and_modify(|current| *current = (*current).max(next)) - .or_insert(next); + .and_modify(|current| { + if row.lane_seq > current.lane_seq { + *current = row; + } + }) + .or_insert(row); } - next_by_lane + last_by_lane .into_iter() .map( - |((queue, priority, enqueue_shard), next_seq)| ClaimCursorAdvance { + |((queue, priority, enqueue_shard), last)| ClaimCursorAdvance { queue, priority, enqueue_shard, - next_seq, + next_seq: last.lane_seq + 1, only_if_current: None, + evidence: Some(ClaimEvidence { + ready_slot: last.ready_slot, + ready_generation: last.ready_generation, + lane_seq: last.lane_seq, + legacy_attempt: last + .receipt_id + .is_none() + .then_some((last.job_id, last.run_lease)), + }), }, ) .collect() @@ -4567,24 +4622,21 @@ impl QueueStorage { .entry(only_if_current) .and_modify(|next| *next = (*next).max(advance.next_seq)) .or_insert(advance.next_seq); - } else { - *unconditional = Some( - unconditional - .map(|next| next.max(advance.next_seq)) - .unwrap_or(advance.next_seq), - ); + } else if unconditional.is_none_or(|(next, _)| advance.next_seq > next) { + *unconditional = Some((advance.next_seq, advance.evidence)); } } let mut normalized = Vec::with_capacity(advances.len()); for ((queue, priority, enqueue_shard), (unconditional, conditional)) in grouped { - if let Some(next_seq) = unconditional { + if let Some((next_seq, evidence)) = unconditional { normalized.push(ClaimCursorAdvance { queue, priority, enqueue_shard, next_seq, only_if_current: None, + evidence, }); continue; } @@ -4596,6 +4648,7 @@ impl QueueStorage { enqueue_shard, next_seq, only_if_current: Some(only_if_current), + evidence: None, }); } } @@ -4603,6 +4656,36 @@ impl QueueStorage { normalized } + /// Replay the post-commit advance for an actual claim on the current server. + #[doc(hidden)] + pub async fn replay_claim_cursor_advance( + &self, + pool: &PgPool, + claimed: &ClaimedRuntimeJob, + ) -> Result<(), AwaError> { + let claim = &claimed.claim; + self.advance_claim_cursors_strict( + pool, + &[ClaimCursorAdvance { + queue: claim.queue.clone(), + priority: claim.priority, + enqueue_shard: claim.enqueue_shard, + next_seq: claim.lane_seq + 1, + only_if_current: None, + evidence: Some(ClaimEvidence { + ready_slot: claim.ready_slot, + ready_generation: claim.ready_generation, + lane_seq: claim.lane_seq, + legacy_attempt: claim + .receipt_id + .is_none() + .then_some((claimed.job.id, claimed.job.run_lease)), + }), + }], + ) + .await + } + async fn advance_claim_cursors(&self, pool: &PgPool, advances: &[ClaimCursorAdvance]) { let advances = Self::normalize_claim_cursor_advances(advances); // PostgreSQL sequence state is not rolled back with the surrounding @@ -4643,7 +4726,10 @@ impl QueueStorage { } let schema = self.schema(); - let mut tx = pool.begin().await.map_err(map_sqlx_error)?; + let mut tx = pool + .begin_with(CLAIM_TRANSACTION_BEGIN) + .await + .map_err(map_sqlx_error)?; for advance in advances { sqlx::query(&format!( r#" @@ -4657,8 +4743,30 @@ impl QueueStorage { ) SELECT {schema}.set_sequence_next(seq_name, $4) FROM head - WHERE $5::bigint IS NULL - OR {schema}.sequence_next_value(seq_name) = $5 + WHERE ($5::bigint IS NULL OR {schema}.sequence_next_value(seq_name) = $5) + AND ( + $6::int IS NULL + OR ($9::bigint IS NULL AND EXISTS ( + SELECT 1 + FROM {schema}.ready_claim_attempt_batches AS attempts + WHERE attempts.ready_slot = $6 + AND attempts.ready_generation = $7 + AND attempts.queue = $1 + AND attempts.priority = $2 + AND attempts.enqueue_shard = $3 + AND attempts.first_lane_seq <= $8 + AND attempts.next_lane_seq > $8 + AND attempts.lane_ranges @> int8range($8, $8 + 1, '[)') + )) + OR EXISTS ( + SELECT 1 FROM {schema}.leases AS leases + WHERE $9::bigint IS NOT NULL + AND leases.job_id = $9 AND leases.run_lease = $10 + AND leases.ready_slot = $6 AND leases.ready_generation = $7 + AND leases.queue = $1 AND leases.enqueue_shard = $3 + AND leases.lane_seq = $8 + ) + ) "# )) .bind(&advance.queue) @@ -4666,6 +4774,19 @@ impl QueueStorage { .bind(advance.enqueue_shard) .bind(advance.next_seq) .bind(advance.only_if_current) + .bind(advance.evidence.map(|evidence| evidence.ready_slot)) + .bind(advance.evidence.map(|evidence| evidence.ready_generation)) + .bind(advance.evidence.map(|evidence| evidence.lane_seq)) + .bind( + advance + .evidence + .and_then(|evidence| evidence.legacy_attempt.map(|attempt| attempt.0)), + ) + .bind( + advance + .evidence + .and_then(|evidence| evidence.legacy_attempt.map(|attempt| attempt.1)), + ) .execute(tx.as_mut()) .await .map_err(map_sqlx_error)?; @@ -6091,7 +6212,10 @@ impl QueueStorage { return Ok(Vec::new()); } - let mut tx = pool.begin().await.map_err(map_sqlx_error)?; + let mut tx = pool + .begin_with(CLAIM_TRANSACTION_BEGIN) + .await + .map_err(map_sqlx_error)?; let mut claimed_rows = Vec::new(); let stripe_queues = self.physical_queues_for_logical(queue); let start = self.stripe_probe_start(stripe_queues.len()); @@ -6214,7 +6338,10 @@ impl QueueStorage { return Ok(Vec::new()); } - let mut tx = pool.begin().await.map_err(map_sqlx_error)?; + let mut tx = pool + .begin_with(CLAIM_TRANSACTION_BEGIN) + .await + .map_err(map_sqlx_error)?; let mut claimed = Vec::new(); claimed.extend( self.claim_ready_rows_tx(&mut tx, queue, max_batch, deadline_duration, aging_interval) @@ -6406,6 +6533,27 @@ impl QueueStorage { Ok(None) } + #[doc(hidden)] + pub fn cached_claimer_lease_count(&self) -> usize { + self.claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .len() + } + + fn cache_claimer_gate(&self, key: (String, Uuid), lease: QueueClaimerLeaseRow) { + self.claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .insert( + key, + CachedClaimerGate { + checked_at: Instant::now(), + lease, + }, + ); + } + #[tracing::instrument(skip(self, pool), fields(queue = %queue, instance_id = %instance_id, claimer_slot = lease.claimer_slot), name = "queue_storage.mark_queue_claimer_active")] pub async fn mark_queue_claimer_active( &self, @@ -6628,22 +6776,43 @@ impl QueueStorage { lease_ttl: Duration, idle_threshold: Duration, ) -> Result, AwaError> { - let target_claimers = self - .queue_claimer_target(pool, queue, max_claimers, Duration::from_millis(500)) - .await?; - - let Some(lease) = self - .acquire_queue_claimer_row( - pool, - queue, - instance_id, - target_claimers, - lease_ttl, - idle_threshold, - ) - .await? - else { - return Ok(Vec::new()); + let gate_key = (queue.to_string(), instance_id); + let cached_lease = self + .claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .get(&gate_key) + .filter(|gate| { + gate.checked_at.elapsed() < CLAIMER_GATE_CACHE_TTL + && gate.lease.expires_at > Utc::now() + TimeDelta::seconds(1) + }) + .map(|gate| gate.lease); + let lease = match cached_lease { + Some(lease) => lease, + None => { + let target_claimers = self + .queue_claimer_target(pool, queue, max_claimers, Duration::from_millis(500)) + .await?; + let Some(lease) = self + .acquire_queue_claimer_row( + pool, + queue, + instance_id, + target_claimers, + lease_ttl, + idle_threshold, + ) + .await? + else { + self.claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .remove(&gate_key); + return Ok(Vec::new()); + }; + self.cache_claimer_gate(gate_key.clone(), lease); + lease + } }; let claimed = self @@ -6656,10 +6825,27 @@ impl QueueStorage { ) .await?; - if !claimed.is_empty() && lease.needs_refresh(Utc::now(), lease_ttl, idle_threshold) { - let _ = self + let now = Utc::now(); + if !claimed.is_empty() && lease.needs_refresh(now, lease_ttl, idle_threshold) { + let still_owned = self .mark_queue_claimer_active(pool, queue, instance_id, lease.lease(), lease_ttl) .await?; + if still_owned { + let refreshed = QueueClaimerLeaseRow { + last_claimed_at: now, + expires_at: now + + TimeDelta::from_std(lease_ttl).map_err(|err| { + AwaError::Validation(format!("invalid claimer lease ttl: {err}")) + })?, + ..lease + }; + self.cache_claimer_gate(gate_key, refreshed); + } else { + self.claimer_gates + .lock() + .expect("claimer gate cache poisoned") + .remove(&gate_key); + } } Ok(claimed) @@ -9092,6 +9278,7 @@ impl QueueStorage { enqueue_shard: ready.enqueue_shard, next_seq: ready.lane_seq + 1, only_if_current: Some(ready.lane_seq), + evidence: None, }; return Ok(Some(CancelJobTxResult { row: done.into_job_row()?, diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 3c1ea28da..501665d23 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -13147,14 +13147,19 @@ async fn test_queue_storage_striped_runtime_claims_do_not_deadlock_with_enqueues }; let store = Arc::new(create_store_with_config(&pool, config).await); + let (first_enqueue_tx, first_enqueue_rx) = tokio::sync::oneshot::channel(); let producer_pool = pool.clone(); let producer_store = Arc::clone(&store); let producer = tokio::spawn(async move { + let mut first_enqueue_tx = Some(first_enqueue_tx); for _ in 0..64 { producer_store .enqueue_batch(&producer_pool, queue, 1, 16) .await .expect("striped enqueue should not deadlock"); + if let Some(start) = first_enqueue_tx.take() { + start.send(()).expect("claimer start receiver dropped"); + } tokio::task::yield_now().await; } }); @@ -13162,6 +13167,9 @@ async fn test_queue_storage_striped_runtime_claims_do_not_deadlock_with_enqueues let claimer_pool = pool.clone(); let claimer_store = Arc::clone(&store); let claimer = tokio::spawn(async move { + // Otherwise fast empty claims can finish before the first producer + // commit, making the progress assertion a scheduler race. + first_enqueue_rx.await.expect("producer did not enqueue"); let mut claimed_total = 0usize; for _ in 0..128 { let claimed = claimer_store @@ -13453,6 +13461,170 @@ async fn test_queue_storage_aged_completion_stays_compact_and_keeps_lane_priorit assert_eq!(stored_priority, 4); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_queue_storage_reset_forgets_cached_claimer_lease() { + let (_db_guard, pool) = setup_pool(10).await; + let schema = "awa_qs_reset_claimer_cache"; + let store = create_store(&pool, schema).await; + let queue = "qs_reset_claimer_cache"; + let instance = Uuid::new_v4(); + let ttl = Duration::from_secs(300); + let idle_threshold = Duration::from_secs(120); + + store + .enqueue_batch(&pool, queue, 1, 1) + .await + .expect("failed to enqueue job"); + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + Duration::from_secs(300), + Duration::from_secs(60), + instance, + 1, + ttl, + idle_threshold, + ) + .await + .expect("claim should succeed"); + assert_eq!(claimed.len(), 1); + assert_eq!(store.cached_claimer_lease_count(), 1); + + store.reset(&pool).await.expect("failed to reset store"); + assert_eq!( + store.cached_claimer_lease_count(), + 0, + "reset truncates queue_claimer_leases, so cached leases must go too" + ); + store + .enqueue_batch(&pool, queue, 1, 1) + .await + .expect("failed to enqueue job after reset"); + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + Duration::from_secs(300), + Duration::from_secs(60), + instance, + 1, + ttl, + idle_threshold, + ) + .await + .expect("claim after reset should succeed"); + assert_eq!(claimed.len(), 1); + + let owned: i64 = sqlx::query_scalar(&format!( + "SELECT count(*)::bigint FROM {schema}.queue_claimer_leases WHERE queue = $1 AND owner_instance_id = $2" + )) + .bind(queue) + .bind(instance) + .fetch_one(&pool) + .await + .expect("failed to count claimer leases"); + assert_eq!( + owned, 1, + "a claim right after reset must take a real claimer lease, not a cached one" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_queue_storage_claimer_cache_rechecks_reduced_target_after_lease_loss() { + let (_db_guard, pool) = setup_pool(10).await; + let schema = "awa_qs_claimer_cache_target"; + let store = create_store(&pool, schema).await; + let queue = "qs_claimer_cache_target"; + let instance = Uuid::new_v4(); + let other = Uuid::new_v4(); + let ttl = Duration::from_secs(300); + let idle = Duration::from_secs(120); + store.enqueue_batch(&pool, queue, 1, 3).await.unwrap(); + // Own slot 1 while another active instance occupies slot 0. + sqlx::query(&format!( + "INSERT INTO {schema}.queue_claimer_state (queue, target_claimers, updated_at) + VALUES ($1, 2, now() + interval '1 hour'); + " + )) + .bind(queue) + .execute(&pool) + .await + .unwrap(); + for (slot, owner) in [(0_i16, other), (1_i16, instance)] { + sqlx::query(&format!( + "INSERT INTO {schema}.queue_claimer_leases + (queue, claimer_slot, owner_instance_id, lease_epoch, leased_at, last_claimed_at, expires_at) + VALUES ($1, $2, $3, 0, now(), now(), now() + interval '1 hour')" + )) + .bind(queue).bind(slot).bind(owner).execute(&pool).await.unwrap(); + } + let claimed = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + ttl, + Duration::ZERO, + instance, + 2, + ttl, + idle, + ) + .await + .unwrap(); + assert_eq!(claimed.len(), 1); + assert_eq!(store.cached_claimer_lease_count(), 1); + // An external coordination change is allowed to lag until the cache TTL. + // After it expires, the old owner must not acquire the now-disallowed + // second slot or bypass the active owner in slot 0. + sqlx::query(&format!( + "UPDATE {schema}.queue_claimer_state SET target_claimers = 1 WHERE queue = $1" + )) + .bind(queue) + .execute(&pool) + .await + .unwrap(); + sqlx::query(&format!( + "DELETE FROM {schema}.queue_claimer_leases WHERE queue = $1 AND claimer_slot = 1" + )) + .bind(queue) + .execute(&pool) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + let denied = store + .claim_runtime_batch_with_aging_for_instance( + &pool, + queue, + 1, + ttl, + Duration::ZERO, + instance, + 2, + ttl, + idle, + ) + .await + .unwrap(); + assert!( + denied.is_empty(), + "expired cache must recheck target and ownership" + ); + assert_eq!(store.cached_claimer_lease_count(), 0); + assert_eq!( + store + .claim_runtime_batch(&pool, queue, 2, ttl) + .await + .unwrap() + .len(), + 2, + "coordination denial must leave the remaining work claimable" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_storage_bounded_claimers_limit_active_claimers_per_queue() { let (_db_guard, pool) = setup_pool(10).await; @@ -13592,6 +13764,8 @@ async fn test_queue_storage_claimer_heartbeat_skips_fresh_lease() { .execute(&pool) .await .expect("failed to age claimer lease heartbeat"); + // Claims reuse this instance's cached claimer lease for a short TTL. + tokio::time::sleep(Duration::from_millis(300)).await; store .enqueue_batch(&pool, queue, 1, 1) @@ -22544,3 +22718,118 @@ async fn test_claim_batch_job_index_patch_leaves_indexing_to_claim_prune() { assert_eq!(job.state, JobState::Completed); } } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_queue_storage_cursor_advance_requires_claim_evidence() { + cursor_advance_requires_claim_evidence(true).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_queue_storage_legacy_cursor_advance_requires_claim_evidence() { + cursor_advance_requires_claim_evidence(false).await; +} + +async fn cursor_advance_requires_claim_evidence(receipts: bool) { + let (_db_guard, pool) = setup_pool(10).await; + let schema = "awa_qs_cursor_claim_evidence"; + let store = create_store_with_config( + &pool, + QueueStorageConfig { + schema: schema.to_string(), + queue_slot_count: 4, + lease_slot_count: 2, + lease_claim_receipts: receipts, + ..Default::default() + }, + ) + .await; + let queue = "qs_cursor_claim_evidence"; + + store + .enqueue_batch(&pool, queue, 2, 1) + .await + .expect("failed to enqueue job"); + let claimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::from_secs(30)) + .await + .expect("failed to claim job") + .into_iter() + .next() + .expect("missing claimed job"); + let claim = claimed.claim.clone(); + assert_eq!(claim.receipt_id.is_some(), receipts); + assert_eq!( + claim_cursor_for(&pool, &store, queue, claim.priority, claim.enqueue_shard).await, + claim.lane_seq + 1, + "a recorded claim advances the cursor past its row" + ); + + // A committed claim with a lost cursor advance remains spent. Replaying + // the real production advance repairs its cursor without emitting twice. + sqlx::query(&format!( + "SELECT setval(format('%I.%I', $1::text, seq_name)::regclass, $3, false) + FROM {schema}.queue_claim_heads WHERE queue = $2" + )) + .bind(schema) + .bind(queue) + .bind(claim.lane_seq) + .execute(&pool) + .await + .unwrap(); + store + .replay_claim_cursor_advance(&pool, &claimed) + .await + .unwrap(); + assert_eq!( + claim_cursor_for(&pool, &store, queue, claim.priority, claim.enqueue_shard).await, + claim.lane_seq + 1, + "committed evidence permits the replay" + ); + + // A server that never received the claim (e.g. a standby promoted before + // the asynchronously committed claim replicated): no claim rows, and the + // cursor still at the claimed row. + for table in [ + "ready_claim_attempt_batches", + "lease_claim_batches", + "lease_claims", + "leases", + ] { + sqlx::query(&format!("DELETE FROM {schema}.{table} WHERE queue = $1")) + .bind(queue) + .execute(&pool) + .await + .expect("failed to remove claim evidence"); + } + sqlx::query(&format!( + "SELECT setval(format('%I.%I', $1::text, seq_name)::regclass, $3, false) + FROM {schema}.queue_claim_heads + WHERE queue = $2" + )) + .bind(schema) + .bind(queue) + .bind(claim.lane_seq) + .execute(&pool) + .await + .expect("failed to rewind claim cursor"); + + store + .replay_claim_cursor_advance(&pool, &claimed) + .await + .expect("cursor advance should not fail"); + assert_eq!( + claim_cursor_for(&pool, &store, queue, claim.priority, claim.enqueue_shard).await, + claim.lane_seq, + "the cursor must not move past a row whose claim is not recorded" + ); + + let reclaimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::from_secs(30)) + .await + .expect("failed to reclaim job"); + assert_eq!( + reclaimed.iter().map(|job| job.job.id).collect::>(), + vec![claimed.job.id], + "the job whose claim was lost stays claimable" + ); +} diff --git a/correctness/run-tlc-suite.sh b/correctness/run-tlc-suite.sh index 007d27b82..106e78b49 100755 --- a/correctness/run-tlc-suite.sh +++ b/correctness/run-tlc-suite.sh @@ -17,6 +17,8 @@ set -u SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) SUITE=' +AwaAsyncClaimCursor.tla ;; - ;; pass ;; evidence-guarded async claim replay +AwaAsyncClaimCursor.tla ;; AwaAsyncClaimCursorUnguarded.cfg ;; NoSkippedJob is violated ;; lost claim on promoted standby witness AwaSegmentedStorage.tla ;; - ;; pass ;; segmented storage, 1 worker AwaSegmentedStorage.tla ;; AwaSegmentedStorageInterleavings.cfg ;; pass ;; segmented storage, 2-worker interleavings AwaSegmentedStorageRaces.tla ;; AwaSegmentedStorageRaces.cfg ;; PrunedLeaseSegmentsAreEmpty is violated|PrunedClaimSegmentsAreEmpty is violated ;; claim/rotate/prune race witness diff --git a/correctness/storage/AwaAsyncClaimCursor.cfg b/correctness/storage/AwaAsyncClaimCursor.cfg new file mode 100644 index 000000000..b2ed63215 --- /dev/null +++ b/correctness/storage/AwaAsyncClaimCursor.cfg @@ -0,0 +1,5 @@ +CONSTANT GuardEvidence = TRUE +CONSTANT AllowCrash = TRUE +SPECIFICATION Spec +INVARIANTS TypeOK NoSkippedJob DurableOrder ReplicaOrder +CHECK_DEADLOCK FALSE diff --git a/correctness/storage/AwaAsyncClaimCursor.tla b/correctness/storage/AwaAsyncClaimCursor.tla new file mode 100644 index 000000000..c9ec28028 --- /dev/null +++ b/correctness/storage/AwaAsyncClaimCursor.tla @@ -0,0 +1,74 @@ +---- MODULE AwaAsyncClaimCursor ---- +EXTENDS Naturals, FiniteSets, TLC + +\* Focused refinement of Claim into acknowledged claim and post-commit +\* sequence movement. Claims and cursor advances can be lost independently +\* after reconnecting to a promoted standby. Local WAL flush and replication +\* preserve commit order; replayed advances must use evidence on this server. +\* Receipt evidence is a covering ready_claim_attempt_batches range; legacy +\* evidence is the exact materialized lease attempt. Both abstract to evidence. +\* Completion/rescue/prune are outside this focused model (covered by the +\* lifecycle model); disappearing evidence conservatively prevents replay. +CONSTANTS GuardEvidence, AllowCrash +Jobs == 1..2 +VARIABLES evidence, cursor, durableEvidence, durableCursor, + standbyEvidence, standbyCursor, pending +vars == <> +Init == + /\ evidence = {} + /\ cursor = 1 + /\ durableEvidence = {} + /\ durableCursor = 1 + /\ standbyEvidence = {} + /\ standbyCursor = 1 + /\ pending = {} +Claim == + /\ cursor \in Jobs + /\ cursor \notin evidence + /\ evidence' = evidence \cup {cursor} + /\ pending' = pending \cup {cursor} + /\ UNCHANGED <> +Advance(j) == + /\ j \in pending + /\ ~GuardEvidence \/ j \in evidence + /\ cursor' = IF cursor > j THEN cursor ELSE j + 1 + /\ pending' = pending \ {j} + /\ UNCHANGED <> +Repair == + /\ cursor \in evidence + /\ cursor' = cursor + 1 + /\ UNCHANGED <> +Flush == + /\ durableEvidence' = evidence + /\ durableCursor' = cursor + /\ UNCHANGED <> +Replicate == + /\ standbyEvidence' = evidence + /\ standbyCursor' = cursor + /\ UNCHANGED <> +Crash == + /\ evidence' = durableEvidence + /\ cursor' = durableCursor + /\ UNCHANGED <> +Promote == + /\ evidence' = standbyEvidence + /\ cursor' = standbyCursor + /\ durableEvidence' = standbyEvidence + /\ durableCursor' = standbyCursor + /\ UNCHANGED <> +Next == Claim \/ (\E j \in Jobs : Advance(j)) \/ Repair \/ Flush + \/ Replicate \/ (AllowCrash /\ Crash) \/ Promote +Spec == Init /\ [][Next]_vars +TypeOK == + /\ evidence \subseteq Jobs /\ durableEvidence \subseteq Jobs + /\ standbyEvidence \subseteq Jobs /\ pending \subseteq Jobs + /\ cursor \in 1..3 /\ durableCursor \in 1..3 /\ standbyCursor \in 1..3 +NoSkippedJob == \A j \in Jobs : j < cursor => j \in evidence +DurableOrder == \A j \in Jobs : j < durableCursor => j \in durableEvidence +ReplicaOrder == \A j \in Jobs : j < standbyCursor => j \in standbyEvidence +============================================================================= diff --git a/correctness/storage/AwaAsyncClaimCursorUnguarded.cfg b/correctness/storage/AwaAsyncClaimCursorUnguarded.cfg new file mode 100644 index 000000000..1c4319f72 --- /dev/null +++ b/correctness/storage/AwaAsyncClaimCursorUnguarded.cfg @@ -0,0 +1,5 @@ +CONSTANT GuardEvidence = FALSE +CONSTANT AllowCrash = FALSE +SPECIFICATION Spec +INVARIANTS TypeOK NoSkippedJob DurableOrder ReplicaOrder +CHECK_DEADLOCK FALSE diff --git a/correctness/storage/MAPPING.md b/correctness/storage/MAPPING.md index c132f7e03..93ad5ea8d 100644 --- a/correctness/storage/MAPPING.md +++ b/correctness/storage/MAPPING.md @@ -140,6 +140,10 @@ The model deliberately keeps `MixedHasQueueExecutor` as an entry-gate property, The TLA+ storage model does not represent local worker-capacity accounting. Rust now releases local queue capacity immediately after handler execution and progress snapshotting, while durable completion continues asynchronously through the completion batcher. That changes throughput and scheduling behavior, but it does not change the modeled storage safety boundary because the `run_lease`-guarded finalization and rescue semantics are unchanged. +## Claimer coordination cache (#508) + +The per-instance 250 ms claimer cache is a throughput gate, not job ownership evidence. A stale target/lease may allow another claim during that window, but the same database claim transaction and attempt fences still govern each job. Existing owned slots were already retained across target reductions; the target bounds new acquisition. Cache expiry rechecks acquisition, near-expiry leases bypass the cache, lost refresh/acquisition evicts it, and reset clears it after truncating leases. Rust tests cover reset/reacquisition and a reduced target combined with loss of a cached slot. The storage lifecycle model needs no new ownership transition for this hint cache. Concurrent reset and active claims are outside the reset API's test-isolation use; clearing this cache does not establish a fleet reset barrier. + ## Producer batching note The TLA+ storage model treats `EnqueueReady` and `EnqueueDeferred` as logical per-job state transitions. Rust may batch the SQL implementation of producer side effects: allocating a contiguous lane sequence range, syncing enqueue-time `job_unique_claims` with one array-backed statement, and inserting rows with multi-row `INSERT` or COPY. Those batching choices refine the same logical actions as long as they commit in the same transaction as the ready/deferred append. @@ -349,3 +353,9 @@ These implementation behaviours refine existing actions or sit outside the model The shared LISTEN hub, capped empty-poll backoff and runtime semaphore change scheduling, not `Claim` ownership. Durable SQL claim evidence remains authoritative. A saturated semaphore is cancellable and refunds worker permits/rate tokens. A claim already in progress can finish during shutdown; each spawned execution observes the dispatcher cancellation token, including before in-flight registration. Dispatcher exit and worker drain share a timeout, after which pending dispatchers are aborted and unexecuted claims rely on lease rescue. Executable scheduler regressions cover saturated-gate cancellation (including a simultaneously available permit), late task registration, cancellation during pending execution, bounded dispatcher shutdown and draining both queues after terminating the shared listener. These tests establish these concrete scheduling behaviors; storage TLC does not prove eventual execution under arbitrary database outages, unfair runtime scheduling or handlers that ignore cancellation. + +## Asynchronous claim commits (#507) + +`AwaSegmentedStorage.Claim` is atomic and does not describe acknowledgment before WAL durability or a post-claim cursor retry on another server. `AwaAsyncClaimCursor` supplies that refinement: a claim may be lost on local crash or standby promotion, a committed claim may survive a lost cursor advance, and a queued advance can be replayed after promotion. WAL flush and replica application preserve order. `NoSkippedJob` forbids advancing past an attempt without evidence on the current server. The unguarded config is an expected counterexample, not a passing proof. + +Receipt advances probe the covering slot/generation/lane range in `ready_claim_attempt_batches`; legacy advances require the exact `(job_id, run_lease)` lease and ready identity. If completion or maintenance has removed that evidence, replay conservatively leaves the cursor behind; the existing spent-prefix recovery recognizes committed dispositions. Conditional cancellation advances are derived from synchronous terminal commits and remain separate. Rust regressions replay both real receipt and legacy identities with evidence present, then remove the claim and rewind the cursor to represent a promoted standby and verify the job is claimable again. This is a deterministic state reconstruction, not a real replication/failover rehearsal. diff --git a/correctness/storage/README.md b/correctness/storage/README.md index f28ab0ed7..4ed7be41c 100644 --- a/correctness/storage/README.md +++ b/correctness/storage/README.md @@ -276,3 +276,5 @@ See [`../README.md`](../README.md) for the full Known Divergences list. Specific - **Enqueue-shard routing is split across specs.** The base storage model uses one `(queue, priority, enqueue_shard)` lane. Cross-shard `lane_seq` collisions are checked by `AwaShardedPrune`, but the producer shard chooser, per-queue `enqueue_shards` changes, and projection exactness across shards remain code-tested rather than represented in a single unified TLA+ state. - **DLQ bulk ops are modelled as quantified single-row actions.** The Rust `bulk_retry_from_dlq` / `purge_dlq` paths are transactionally atomic across the matching rows; the spec explores arbitrary interleavings of individual `RetryFromDlq(j)` / `PurgeDlq(j)` which is a strictly weaker claim. Safety invariants hold under both framings; a refinement with a `BulkScope` set variable could tighten this. - **Unique-claim conflicts on `retry_from_dlq`.** The Rust contract says retry-from-dlq returns `UniqueConflict` and leaves the DLQ row intact when a replacement owns the unique slot. The model doesn't have unique keys, so TLC simply explores `RetryFromDlq(j)` whenever the precondition holds. Preserving-the-DLQ-row-on-conflict is enforced in SQL (`awa-model/src/queue_storage.rs::sync_unique_claim`). + +`AwaAsyncClaimCursor.tla` refines async claim acknowledgment, WAL durability, standby promotion and post-commit sequence replay. Its safe config checks no skipped jobs; `AwaAsyncClaimCursorUnguarded.cfg` preserves the lost-claim cursor-skip counterexample. diff --git a/docs/architecture.md b/docs/architecture.md index 80317e554..9059c3b46 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -70,7 +70,7 @@ Most applications should use the Rust, Python, CLI, or UI APIs rather than query | `awa.jobs` / `awa.insert_job_compat()` | Canonical compatibility surface used during the storage transition and by SQL adapters that need the public job shape. When queue storage is active, the runtime does not claim or complete from `awa.jobs`; compatibility inserts route into the active backend, and SQL deletes of ready rows append ready tombstones. | Public compatibility surface. Prefer Rust/Python/adapter APIs for ordinary writes. For high-volume queue-storage producers, prefer configured direct COPY through `QueueStorage::enqueue_params_copy()` or Python `enqueue_many_copy()` rather than the compat COPY path. | | `{schema}.terminal_jobs` | Read-only hydrated view over queue-storage terminal history. It joins narrow `done_entries_*` rows and compact `receipt_completion_batches_*` rows back to retained `ready_entries_*` bodies. | Public read surface for SQL inspection and reporting of terminal queue-storage rows. It is not a write or transition surface. | | `ready_entries_*` | Physical runnable queue ring. | Internal storage. Read only for low-level debugging; writes must go through Awa enqueue, claim, retry, cancel, or maintenance paths. | -| `ready_claim_attempt_batches_*` | Physical attempt-emission ledger keyed by ready slot/generation and lane ranges. | Internal storage. Claim writes one compact batch row in the same transaction as the receipt/lease claims; stale claim-cursor recovery treats covered lanes as durable evidence that those lanes already emitted `run_lease + 1`. Queue prune truncates it with the matching ready segment. | +| `ready_claim_attempt_batches_*` | Physical attempt-emission ledger keyed by ready slot/generation and lane ranges. | Internal storage. Receipt-mode claim writes one compact batch row in the same transaction as its receipt claims; stale claim-cursor recovery treats committed covered lanes as evidence that those lanes already emitted `run_lease + 1`. Legacy materialized claims use their exact lease attempt as evidence. Queue prune truncates it with the matching ready segment. | | `ready_tombstones_*` | Physical ledger for ready lanes made unavailable by cancellation, priority aging, or similar out-of-band ready mutations. | Internal storage. Claim treats matching rows as spent lane evidence and exact-count paths skip them; maintenance truncates it with the matching ready segment. | | `ready_segments_*` | Compact control-plane map from committed ready lane ranges to their ready slot/generation. | Internal storage. Claim uses it to choose the target ready segment before validating `ready_entries_*`; ranges are split when the lane-head `run_at` timestamp changes so claim-time priority aging stays exact as the cursor advances; queue prune truncates the matching ready-slot child. | | `queue_claim_heads` ready-segment columns | Legacy nullable ready-slot/generation cache beside the per-lane claim cursor. | **Unused.** `claim_ready_runtime` resolves the target ready slot from `ready_segments_*` on every claim instead of caching it here — the per-claim `UPDATE` of this singleton row was the dominant dead-tuple source under a pinned MVCC horizon. The columns are retained for rolling-upgrade compatibility; dropping them is deferred to a major version. | @@ -169,6 +169,8 @@ Claim is cursor-based rather than heap-scan based: - `PartitionedQueue` is a Rust and Python helper for mapping one hot logical queue to several ordinary physical queue names. The storage engine does not add a separate group table for this; each physical queue keeps the normal lane, lease, DLQ, descriptor, and terminal-history contract. - Queue striping and bounded claimers reduce contention on very hot logical queues. Per-queue `claimers` can add dispatcher/claimer loops inside one runtime, but those loops share the queue's worker permits and rate limiter; they do not own jobs. Recovery still follows the receipt/lease state in Postgres. +Claim transactions and their post-commit cursor advances use `synchronous_commit = off`. An acknowledged claim can be lost before WAL durability or standby replication, leaving its job claimable again under the at-least-once contract. Enqueues and completions stay synchronous. Cursor replays require claim evidence visible on the current server: a covering ready-attempt range for receipt claims or the exact materialized lease attempt for legacy claims. Missing evidence leaves the cursor behind; the spent-prefix recovery path repairs it from committed dispositions. A later local synchronous commit flushes preceding claim WAL, while failover survival still depends on replication configuration. See the [async-claim refinement mapping](https://github.com/hardbyte/awa/blob/main/correctness/storage/MAPPING.md#asynchronous-claim-commits-507). A worker still running a claim that was lost this way is not fenced: the job's next claim reuses the lost claim's run lease, so that worker's heartbeat or completion can be accepted against the new attempt. Delivery stays at-least-once; before asynchronous claim commits this needed a failover with asynchronous replication, and a primary crash can now cause it until the next synchronous commit or WAL writer flush. + Priority ordering is by effective priority first. Within one enqueue shard the lane sequence is FIFO; across enqueue shards, strict global lane order is not promised. With queue storage, priority aging is applied at claim time rather than by physically rewriting ready rows. The ready/done/lease partitions carry shard-aware lane indexes on `(queue, priority, enqueue_shard, lane_seq)` so deep backlog claim probes do not scan a non-shard-selective lane index and post-filter most rows. ## Completion And Callbacks diff --git a/docs/configuration.md b/docs/configuration.md index c6e0e2747..dba17cf96 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -120,7 +120,7 @@ With LISTEN available, consecutive empty polls back off to a 2 s cap (or `poll_i A logical queue is the application concept: for example `email` or `customer-updates`. A physical queue is the queue name Awa stores in Postgres and workers claim from. Most applications use one physical queue per logical queue. -Very hot workloads can partition one logical queue over multiple physical queues when the ordering contract allows it. This creates independent durable claim and completion streams. It is the most direct way to reduce hot-head coordination without weakening durability: each job is still written, claimed, leased, completed, retried, and rescued through the normal Awa tables. +Very hot workloads can partition one logical queue over multiple physical queues when the ordering contract allows it. This creates independent claim and completion streams. It is the most direct way to reduce hot-head coordination without weakening durability: each job is still written, claimed, leased, completed, retried, and rescued through the normal Awa tables. Rust producers and workers can share `PartitionedQueue`: @@ -668,7 +668,7 @@ Lowering the value is safe at any time — see [`docs/upgrade-0.5-to-0.6.md`](up ### Hot-Queue Claim Control -Queue storage also uses a bounded-claimer control plane (`queue_claimer_state` / `queue_claimer_leases`) so not every replica hammers a hot queue's claim path at once. For a single hot queue, raising `QueueConfig.claimers` lets one runtime run multiple dispatcher/claimer loops while still sharing the queue's `max_workers` / `min_workers` permits. +Queue storage also uses a bounded-claimer control plane (`queue_claimer_state` / `queue_claimer_leases`) so not every replica hammers a hot queue's claim path at once. Each instance may reuse its claimer lease and target check for up to 250 ms, bypassing the cache within one second of lease expiry. Target reductions constrain new acquisition; existing owned slots remain valid. These are coordination hints: job ownership still comes from the database claim transaction. Reset clears the local cache after truncating the coordination leases. For a single hot queue, raising `QueueConfig.claimers` lets one runtime run multiple dispatcher/claimer loops while still sharing the queue's `max_workers` / `min_workers` permits. Keep `claimers` modest. In the hot-queue reference shape, `enqueue_shards = 4` is the change that turns overload into bounded end-to-end throughput; raising `claimers` to `4` increases transaction pressure and worsens tail latency in that shape. For extreme single-queue workloads, raise `enqueue_shards` first when the ordering contract allows partitioned FIFO. Benchmark `claimers = 2` or `4` only for a workload-specific reason, and judge the result by durable completion rate, p99 end-to-end latency, queue depth, WAL bytes per completed job, transaction commits per completed job, and dead tuples. Lower `claim_batch_size` from `512` for small or latency-sensitive queues; raise claimers only with workload-specific evidence.