diff --git a/.github/fixtures/fusillade-migration-sha384.txt b/.github/fixtures/fusillade-migration-sha384.txt index 520d6ecbd..af861a476 100644 --- a/.github/fixtures/fusillade-migration-sha384.txt +++ b/.github/fixtures/fusillade-migration-sha384.txt @@ -226,3 +226,9 @@ e7e9fb0732e4f0755fe452c6463391da45615f4800c737061946deaadd11199bdeeb53ca3705a46d 936f98aa92536a6d99439d8a96d0f07e18c8962feb4901602ef1b33c9e5a0ccd2d8dd289ae6cebc80b1d1c596966b3f9 fusillade-arsenal/migrations/20261013100020_reindex_batches_finished_late_index.up.sql d34c3ccb5ca7b28858866fd16115615853c49736d0cbe842fa9e67b651aeaddf98e3906d471e728239c7ec5e596ad622 fusillade-arsenal/migrations/20261013100030_validate_batches_finished_late_index.down.sql e8526863e7214449572b9fa4a4cec20f08ea8ebf5ba71e905f8d5845f4b2387b864e96be6ed2a8b993eacb182a891436 fusillade-arsenal/migrations/20261013100030_validate_batches_finished_late_index.up.sql +b4b17aeda491ce00f00cc1d69d885f09522bb6bd315dee7ad4b2ef4ea2775ce6348b2b1864cf9b7686f75f299d3d84a2 fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.down.sql +b6f9ecc0b4af2ceb0dd45437e9faf19c2162e670120fc25551b575b2ffd766e0c4b34a0f90ba064ebf9a8da7892b556f fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.up.sql +922156d0d67b2c8e53e9277713e2f1f58a59cae4e0b953b9a6bf10bf5804857eca5628b7d1064916775d089b7bc04f1a fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.down.sql +412a0cd1a7174692745f5713e63fbd3fbed2c82e4923a78028965f7c47e1a1327ccac0a258a55f56246220ac2a86e8b1 fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.up.sql +45c71e555f9cb92c2113014e3a28e793ffd39d9ad08d610a178b9d9608992237d127e3386aa72f8462f44b7a0b9041c1 fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.down.sql +90e1ebe2c2f229dc48aeb30c3c17d352ee856ec6ccab20675479d52b8e638f90461852ebcd3b6f456b54006ff83591d1 fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.up.sql diff --git a/config.yaml b/config.yaml index af861033e..c05d0257f 100644 --- a/config.yaml +++ b/config.yaml @@ -923,24 +923,38 @@ background_services: # configured until every journaled retirement has completed, even if the # enable flag is turned back off. # retained_response_retirement_enabled: false - # Weekly batch-archive retirement irreversibly drops whole expired - # archive partitions once every batch in the week is past the - # finalization-anchored retention period below. Batch metadata rows are - # stamped retention-expired, never deleted. Requires the same - # session-capable maintenance endpoint as response retirement. + # Content retention for both weekly families below is R = + # batches.files.max_expiry_seconds rounded up to whole days. A week drops + # once it is R days past its end, so no file's expires_at can outrun its + # content. While either family is enabled, a batch can no longer be + # retried once it, or its input file, is R - 1 days old (a retry would + # hold its weeks open past the period); the API answers 400. + # + # Weekly batch-archive retirement irreversibly drops a whole archive + # partition once the week is past R, measured from the week itself, and + # no batch created in that week is still running. The oldest completion + # in a dropped week is 7 + R days old, the newest R days. Batch metadata + # rows are never touched. Requires the same session-capable maintenance + # endpoint as response retirement. # batch_archive_retirement_enabled: false - # batch_archive_retention_days: + # DEPRECATED and ignored: batch_archive_retention_days and + # template_retention_days no longer set retention, which follows + # batches.files.max_expiry_seconds. The keys still parse so existing + # values files keep loading, but log a warning at startup. Remove them. + # batch_archive_retention_days: # DEPRECATED and ignored: request templates are always written to the # weekly generation-2 partitions now that the generation-1 heap has been # retired. The key still parses so existing values files keep loading, # but it logs a warning at startup. Remove it. # template_generation_writes_enabled: false - # File-content expiry + weekly template partition retirement. Tombstones - # aged input-file metadata (rows survive) once every referencing batch's - # content is expired, then drops whole template weeks that no live file - # references. Requires the same session-capable maintenance endpoint. + # Weekly template retirement irreversibly drops a whole week of request + # templates (batch input content) once the week is past R, measured from + # the week itself, and no batch still running reads input uploaded in + # that week. File metadata rows are never touched; an input file reports + # expired from its own expires_at. Requires the same session-capable + # maintenance endpoint. # template_retirement_enabled: false - # template_retention_days: + # template_retention_days: # Schema mode reuses the main direct credentials automatically. Override for # dedicated databases or a different Fusillade query role. Use that role on # the DIRECT host; startup attests database, role, schema, and ownership. diff --git a/docs/src/reference/configuration.md b/docs/src/reference/configuration.md index 72d0b1fde..94b03a913 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -378,6 +378,8 @@ The expiry settings must be positive and ordered `min_expiry_seconds <= default_ - `GET /ai/v1/batches/{batch_id}/results` returns 410 once the batch's output or error file is past its `expires_at`. - Output and error files expire `default_expiry_seconds` after the batch is created. OpenAI counts from when the batch completes instead, so here they expire up to one completion window earlier. Output and error files created before they were given an expiry are treated as expiring `default_expiry_seconds` after their creation. +Stored batch content is deleted in weekly partitions by two retirement switches under `background_services.batch_daemon`, both `false` by default: `template_retirement_enabled` deletes batch inputs (request templates) and `batch_archive_retirement_enabled` deletes archived batch results. An enabled family deletes a week no earlier than `max_expiry_seconds` (rounded up to whole days) after the end of the week the content was created in. While either switch is on, a batch retry is refused with 400 once the batch or its input file is older than that period minus one day. The former `background_services.batch_daemon.batch_archive_retention_days` and `background_services.batch_daemon.template_retention_days` keys are deprecated and ignored; remove them. + ## Background Services ### Onwards Sync diff --git a/dwctl/src/api/handlers/batches.rs b/dwctl/src/api/handlers/batches.rs index ba633b52a..5643a02df 100644 --- a/dwctl/src/api/handlers/batches.rs +++ b/dwctl/src/api/handlers/batches.rs @@ -1788,7 +1788,7 @@ pub async fn delete_batch( description = "Retry a batch: failed AND canceled requests are reset to pending and will be processed again; completed work is never redone. Retrying a cancelled batch overturns the cancellation (cancel-then-retry acts as pause/resume), so this also succeeds on a just-cancelled batch whose requests had not yet settled. Use it after fixing transient issues, increasing rate limits, or to resume a cancelled batch.", responses( (status = 200, description = "Requests queued for retry (or a cancellation was overturned and the batch resumed).", body = BatchResponse), - (status = 400, description = "Nothing to retry: no failed or canceled requests, and the batch was not cancelled."), + (status = 400, description = "The batch cannot be retried: it has no failed or canceled requests and was not cancelled, it is past the retry age cut-off, or its input or results have been deleted by content retention."), (status = 404, description = "Batch not found or you don't have access to it."), (status = 500, description = "An unexpected error occurred. Retry the request or contact support if the issue persists.") ), @@ -1830,8 +1830,13 @@ pub async fn retry_failed_batch_requests( .request_manager .retry_failed_requests_for_batch(fusillade::BatchId(batch_id)) .await - .map_err(|e| Error::Internal { - operation: format!("retry failed requests: {}", e), + .map_err(|e| match e { + // Refusals the client can act on: the batch is past its retry + // age cut-off, or its input or results have been retired. + fusillade::FusilladeError::ValidationError(message) => Error::BadRequest { message }, + e => Error::Internal { + operation: format!("retry failed requests: {}", e), + }, })?; // retried_count == 0 is still SUCCESS when the retry overturned a @@ -1880,7 +1885,7 @@ Use this for fine-grained control over which requests to retry, rather than retr request_body = RetryRequestsRequest, responses( (status = 200, description = "Specified requests queued for retry.", body = BatchResponse), - (status = 400, description = "No valid request IDs provided."), + (status = 400, description = "No valid request IDs provided, or the batch is past the retry age cut-off. Requests whose input has been deleted by content retention are left as they are; the batch is returned with status 200."), (status = 404, description = "Batch not found or you don't have access to it."), (status = 500, description = "An unexpected error occurred. Retry the request or contact support if the issue persists.") ), @@ -1942,8 +1947,13 @@ pub async fn retry_specific_requests( .request_manager .retry_failed_requests(fusillade::BatchId(batch_id), request_ids.clone()) .await - .map_err(|e| Error::Internal { - operation: format!("retry failed requests: {}", e), + .map_err(|e| match e { + // The batch is past its retry age cut-off. A request whose input + // has been retired is refused on its own, in `results`. + fusillade::FusilladeError::ValidationError(message) => Error::BadRequest { message }, + e => Error::Internal { + operation: format!("retry failed requests: {}", e), + }, })?; // Check for any failures diff --git a/dwctl/src/config.rs b/dwctl/src/config.rs index 590608926..770919c2c 100644 --- a/dwctl/src/config.rs +++ b/dwctl/src/config.rs @@ -1275,7 +1275,8 @@ pub struct FilesConfig { /// Smallest `expires_after[seconds]` an upload may request (default: 1 hour) pub min_expiry_seconds: i64, /// Largest `expires_after[seconds]` an upload may request (default: 30 - /// days). Must not exceed what content retention honours. + /// days). Content retention is this rounded up to whole days, so no + /// `expires_at` outlives the content. pub max_expiry_seconds: i64, /// Buffer size for file upload streams (default: 100) pub upload_buffer_size: usize, @@ -2338,8 +2339,9 @@ pub struct DaemonConfig { #[serde(default)] pub retained_response_retirement_enabled: bool, - /// Allow weekly batch-archive partition retirement. Disabled by default - /// and additionally requires an explicit retention period. + /// Allow weekly batch-archive partition retirement. Disabled by default. + /// Retention follows `batches.files.max_expiry_seconds`; see + /// [`content_retention_days`]. #[serde(default)] pub batch_archive_retirement_enabled: bool, @@ -2350,20 +2352,21 @@ pub struct DaemonConfig { #[serde(default, skip_serializing_if = "Option::is_none")] pub template_generation_writes_enabled: Option, - /// Allow file-content expiry and weekly template partition retirement. - /// Disabled by default and additionally requires an explicit retention - /// period. + /// Allow weekly template partition retirement. Disabled by default. + /// Retention follows `batches.files.max_expiry_seconds`; see + /// [`content_retention_days`]. #[serde(default)] pub template_retirement_enabled: bool, - /// Creation-anchored input-content retention period in days. No default: - /// enabling template retirement without it fails startup. - #[serde(default)] + /// Deprecated and ignored: content retention follows + /// `batches.files.max_expiry_seconds`. The key is still accepted so + /// existing values files keep parsing; a warning is logged at startup + /// when it is set. Remove it from your configuration. + #[serde(default, skip_serializing_if = "Option::is_none")] pub template_retention_days: Option, - /// Finalization-anchored batch content retention period in days. No - /// default: enabling batch-archive retirement without it fails startup. - #[serde(default)] + /// Deprecated and ignored; see `template_retention_days`. + #[serde(default, skip_serializing_if = "Option::is_none")] pub batch_archive_retention_days: Option, /// Explicit direct/session-capable primary endpoint used only for @@ -2754,8 +2757,24 @@ impl Default for DaemonConfig { } } +/// Content retention in whole days, derived from the longest file expiry an +/// upload may ask for, rounded up: `ceil(max_expiry_seconds / 1 day)`. +/// +/// Both weekly content families (batch archive and request templates) drop a +/// week once `week_start + 7 days + this` has passed. Every file's +/// `expires_at` is at most `created_at + max_expiry_seconds`, which is at +/// most the week's end plus this, so the date the API shows can never +/// outrun the content. +pub fn content_retention_days(max_expiry_seconds: i64) -> u32 { + const SECONDS_PER_DAY: i64 = 24 * 60 * 60; + let days = max_expiry_seconds.max(1).saturating_add(SECONDS_PER_DAY - 1) / SECONDS_PER_DAY; + u32::try_from(days.min(i64::from(i32::MAX))).unwrap_or(i32::MAX as u32) +} + impl DaemonConfig { - pub fn to_fusillade_retention_maintenance_config(&self) -> fusillade::RetentionMaintenanceConfig { + /// Build the daemon's retention maintenance controls. Content retention + /// is derived from `max_expiry_seconds`; see [`content_retention_days`]. + pub fn to_fusillade_retention_maintenance_config(&self, max_expiry_seconds: i64) -> fusillade::RetentionMaintenanceConfig { fusillade::RetentionMaintenanceConfig::new(self.retention.clone()) .with_batchless_archive_sweep_enabled(self.batchless_archive_sweep_enabled) .with_batchless_archive_backfill_enabled(self.batchless_archive_backfill_enabled) @@ -2764,9 +2783,8 @@ impl DaemonConfig { .with_retained_response_partitions_days_ahead(self.retained_response_partitions_days_ahead) .with_retained_response_retirement_enabled(self.retained_response_retirement_enabled) .with_batch_archive_retirement_enabled(self.batch_archive_retirement_enabled) - .with_batch_archive_retention_days(self.batch_archive_retention_days) .with_template_retirement_enabled(self.template_retirement_enabled) - .with_template_retention_days(self.template_retention_days) + .with_content_retention_days(Some(content_retention_days(max_expiry_seconds))) } /// Convert to fusillade daemon config @@ -3862,22 +3880,6 @@ impl Config { operation: "Config validation: partition retirement on a dedicated database requires an explicit direct session endpoint using the same role as fusillade query traffic".to_string(), }); } - if owns_archive_maintenance - && daemon.batch_archive_retirement_enabled - && daemon.batch_archive_retention_days.is_none_or(|days| days < 1) - { - return Err(Error::Internal { - operation: - "Config validation: batch-archive partition retirement requires an explicit positive batch_archive_retention_days" - .to_string(), - }); - } - if owns_archive_maintenance && daemon.template_retirement_enabled && daemon.template_retention_days.is_none_or(|days| days < 1) { - return Err(Error::Internal { - operation: "Config validation: template partition retirement requires an explicit positive template_retention_days" - .to_string(), - }); - } if owns_archive_maintenance && batchless_movement_enabled && (daemon.batchless_archive_groups_per_tick <= 0 || daemon.batchless_archive_bytes_per_tick <= 0) @@ -4403,7 +4405,7 @@ mod tests { assert_eq!(daemon.batchless_archive_backfill_concurrency, 1); assert!(daemon.retained_response_partitions_days_ahead > 0); - let mapped = daemon.to_fusillade_retention_maintenance_config(); + let mapped = daemon.to_fusillade_retention_maintenance_config(30 * 24 * 60 * 60 + 1); assert_eq!(mapped.policy(), &daemon.retention); assert_eq!(mapped.batchless_archive_sweep_enabled(), daemon.batchless_archive_sweep_enabled); assert_eq!( @@ -4425,7 +4427,12 @@ mod tests { daemon.retained_response_retirement_enabled ); assert_eq!(mapped.batch_archive_retirement_enabled(), daemon.batch_archive_retirement_enabled); - assert_eq!(mapped.batch_archive_retention_days(), daemon.batch_archive_retention_days); + assert_eq!(mapped.template_retirement_enabled(), daemon.template_retirement_enabled); + assert_eq!( + mapped.content_retention_days(), + Some(31), + "retention rounds max expiry up to whole days" + ); } #[test] @@ -5159,6 +5166,16 @@ secret_key: "test-secret-key" assert!(result.is_err()); } + #[test] + fn content_retention_rounds_max_expiry_up_to_whole_days() { + let day = 24 * 60 * 60; + assert_eq!(content_retention_days(1), 1); + assert_eq!(content_retention_days(day), 1); + assert_eq!(content_retention_days(day + 1), 2); + assert_eq!(content_retention_days(30 * day), 30); + assert_eq!(content_retention_days(i64::MAX), i32::MAX as u32); + } + #[test] fn retention_config_is_accepted_on_an_instance_that_never_runs_the_daemon() { // Split topology: API pods share the daemon pods' environment but run @@ -5174,10 +5191,12 @@ secret_key: "test-secret-key" daemon.batchless_archive_sweep_enabled = true; daemon.retained_response_retirement_enabled = true; daemon.batch_archive_retirement_enabled = true; + // Deprecated no-op key: still parses, never rejects. daemon.batch_archive_retention_days = Some(1); // Deprecated no-op key: still parses, never rejects. daemon.template_generation_writes_enabled = Some(true); daemon.template_retirement_enabled = true; + // Deprecated no-op key: still parses, never rejects. daemon.template_retention_days = Some(1); assert!(config.validate().is_ok(), "{:?}", config.validate().err()); } diff --git a/dwctl/src/lib.rs b/dwctl/src/lib.rs index e1c80a3ac..076043825 100644 --- a/dwctl/src/lib.rs +++ b/dwctl/src/lib.rs @@ -4473,7 +4473,10 @@ impl Application { .background_services .batch_daemon .to_fusillade_config_with_limits(Some(model_capacity_limits.clone())); - let retention_maintenance_config = config.background_services.batch_daemon.to_fusillade_retention_maintenance_config(); + let retention_maintenance_config = config + .background_services + .batch_daemon + .to_fusillade_retention_maintenance_config(config.batches.files.max_expiry_seconds); if let Some(value) = config.background_services.batch_daemon.template_generation_writes_enabled { tracing::warn!( @@ -4482,6 +4485,26 @@ impl Application { the weekly generation-2 template store is the only write path. Remove the key." ); } + for (key, value) in [ + ( + "template_retention_days", + config.background_services.batch_daemon.template_retention_days, + ), + ( + "batch_archive_retention_days", + config.background_services.batch_daemon.batch_archive_retention_days, + ), + ] { + if let Some(days) = value { + tracing::warn!( + key, + days, + max_expiry_seconds = config.batches.files.max_expiry_seconds, + "batch_daemon.{key} is deprecated and ignored: content retention follows \ + batches.files.max_expiry_seconds. Remove the key." + ); + } + } let mut request_manager = fusillade_arsenal::PostgresRequestManager::new( fusillade_pools.clone(), fusillade_arsenal::PostgresStorageConfig::from(&fusillade_daemon_config), @@ -4511,7 +4534,8 @@ impl Application { }) .with_batch_insert_strategy(fusillade_arsenal::BatchInsertStrategy::Batched { batch_size: config.batches.files.batch_insert_size, - }); + }) + .with_retry_age_limit(retention_maintenance_config.retry_age_limit()); if pools.fusillade.is_split() && let config::ComponentDb::Schema { name, diff --git a/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.down.sql b/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.down.sql new file mode 100644 index 000000000..e36343fe7 --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.down.sql @@ -0,0 +1,3 @@ +-- no-transaction +-- Reverting returns the retirement guards to scanning all of batches. +DROP INDEX CONCURRENTLY IF EXISTS idx_batches_unfinished_created_at; diff --git a/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.up.sql b/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.up.sql new file mode 100644 index 000000000..ea12ca809 --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100010_add_batches_unfinished_created_at_index.up.sql @@ -0,0 +1,20 @@ +-- no-transaction +-- +-- Driving set for the retirement guards on both content families. A weekly +-- partition of batch_requests_archive or request_templates_g2 drops once it +-- is past its retention period and no unfinished batch can still write into +-- or read from it: +-- +-- WHERE counts_frozen_at IS NULL AND deleted_at IS NULL +-- AND created_at >= AND created_at < +-- +-- Almost every batch is frozen within a day, so the predicate keeps the +-- index to the few dozen batches still running, and the guard reads those +-- rather than scanning all of batches once per retirement tick. +-- +-- One statement only: CONCURRENTLY cannot run in a transaction. The following +-- migrations reindex (repairing an interrupted build) and validate it. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_batches_unfinished_created_at +ON batches (created_at) +WHERE counts_frozen_at IS NULL + AND deleted_at IS NULL; diff --git a/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.down.sql b/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.down.sql new file mode 100644 index 000000000..d3a7ae65d --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.down.sql @@ -0,0 +1,6 @@ +-- no-transaction +-- An interrupted REINDEX INDEX CONCURRENTLY can leave an invalid +-- idx_batches_unfinished_created_at_ccnew behind. Dropping it here, before +-- 20261023100010's down migration drops the index itself, leaves nothing +-- behind. DROP INDEX CONCURRENTLY accepts one name per statement. +DROP INDEX CONCURRENTLY IF EXISTS idx_batches_unfinished_created_at_ccnew; diff --git a/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.up.sql b/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.up.sql new file mode 100644 index 000000000..90b30d41d --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100020_reindex_batches_unfinished_created_at_index.up.sql @@ -0,0 +1,6 @@ +-- no-transaction +-- Rebuild the index created by 20261023100010. On a clean build this is one +-- extra concurrent pass; after an interrupted build it turns the INVALID index +-- left behind into a valid one. See docs/migrations.md, "Concurrent index +-- builds". +REINDEX INDEX CONCURRENTLY idx_batches_unfinished_created_at; diff --git a/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.down.sql b/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.down.sql new file mode 100644 index 000000000..52334b555 --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.down.sql @@ -0,0 +1 @@ +COMMENT ON INDEX idx_batches_unfinished_created_at IS NULL; diff --git a/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.up.sql b/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.up.sql new file mode 100644 index 000000000..29c598bb1 --- /dev/null +++ b/fusillade-arsenal/migrations/20261023100030_validate_batches_unfinished_created_at_index.up.sql @@ -0,0 +1,24 @@ +DO $$ +DECLARE + unfinished_index regclass := to_regclass(format('%I.idx_batches_unfinished_created_at', current_schema())); +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_am am ON am.oid = c.relam + WHERE i.indexrelid = unfinished_index + AND i.indrelid = to_regclass(format('%I.batches', current_schema())) + AND am.amname = 'btree' + AND i.indisvalid AND i.indisready AND NOT i.indisunique + AND i.indnkeyatts = 1 AND i.indnatts = 1 + AND pg_get_indexdef(i.indexrelid, 1, true) = 'created_at' + AND i.indoption::text = '0' + AND pg_get_expr(i.indpred, i.indrelid) = '((counts_frozen_at IS NULL) AND (deleted_at IS NULL))' + ) THEN + RAISE EXCEPTION 'idx_batches_unfinished_created_at is missing, invalid, or has the wrong definition; repair the concurrent build before upgrading'; + END IF; +END +$$; + +COMMENT ON INDEX idx_batches_unfinished_created_at IS +'Batches still running (not frozen, not deleted) by creation time. The weekly retirement guards for batch_requests_archive and request_templates_g2 read this to find any unfinished batch created in, or reading input uploaded in, the week being dropped.'; diff --git a/fusillade-arsenal/src/postgres.rs b/fusillade-arsenal/src/postgres.rs index 72d6add65..395ca1a9f 100644 --- a/fusillade-arsenal/src/postgres.rs +++ b/fusillade-arsenal/src/postgres.rs @@ -39,10 +39,9 @@ use crate::daemon::{ }; use crate::error::{FusilladeError, Result}; use crate::manager::{ - FileExpiryCursor, FileExpiryStep, RetainedResponseArchiveCutoffs, - RetainedResponseArchiveOutcome, RetainedResponsePartitionRunway, - RetainedResponseRetirementOutcome, RetainedResponseWriteError, RetentionPolicy, - TrailingDemandCount, + RetainedResponseArchiveCutoffs, RetainedResponseArchiveOutcome, + RetainedResponsePartitionRunway, RetainedResponseRetirementOutcome, RetainedResponseWriteError, + RetentionPolicy, TrailingDemandCount, }; use crate::request::{ Canceled, CascadeTargetState, Claimed, Completed, CreateBackgroundInput, CreateFlexInput, @@ -243,6 +242,7 @@ pub struct PostgresRequestManager { download_buffer_size: usize, batch_insert_strategy: BatchInsertStrategy, file_expiry: FileExpiry, + retry_age_limit: Option, /// TRANSITIONAL ZDR hook - see [`crate::transform`]. Transforms response/ /// error bodies before persistence; `None` is identity. Remove when stream /// reassembly moves into dwctl. @@ -480,6 +480,7 @@ impl PostgresRequestManager

{ download_buffer_size: 100, batch_insert_strategy: BatchInsertStrategy::default(), file_expiry: FileExpiry::default(), + retry_age_limit: None, response_transformer: std::sync::OnceLock::new(), maintenance_budget: maintenance_budget(DEFAULT_MAINTENANCE_QUERY_TIMEOUT), } @@ -835,24 +836,36 @@ impl PostgresRequestManager

{ // Serialize against file deletion/expiry and reject inaccessible // sources before creating the batch or its virtual files. An FK alone - // only proves the soft-deleted file row still exists. - let source_available = sqlx::query_scalar::<_, bool>( - r#"SELECT EXISTS( - SELECT 1 FROM files - WHERE id = $1 - AND deleted_at IS NULL - AND status = 'processed' - AND (expires_at IS NULL OR expires_at > NOW()) - FOR UPDATE - )"#, + // only proves the soft-deleted file row still exists, and a file past + // its template week's retirement has no content left to run. + // + // The template week's registry row is share-locked until commit. + // Retirement locks it FOR UPDATE before fencing the week and then + // re-checks for unfinished batches, so either it waits for this + // batch to commit and sees it, or it fenced first and this lock + // re-reads the row as no longer active. + let source_available = sqlx::query_scalar::<_, Uuid>( + r#" + SELECT f.id FROM files f + JOIN request_template_buckets bucket + ON bucket.week_start = date_trunc('week', f.created_at AT TIME ZONE 'UTC')::date + WHERE f.id = $1 + AND f.deleted_at IS NULL + AND f.status = 'processed' + AND (f.expires_at IS NULL OR f.expires_at > NOW()) + AND bucket.state = 'active' + FOR UPDATE OF f + FOR SHARE OF bucket + "#, ) .bind(*input.file_id) - .fetch_one(&mut *tx) + .fetch_optional(&mut *tx) .await - .map_err(|e| FusilladeError::Other(anyhow!("Failed to lock source file: {e}")))?; + .map_err(|e| FusilladeError::Other(anyhow!("Failed to lock source file: {e}")))? + .is_some(); if !source_available { return Err(FusilladeError::ValidationError( - "source file is deleted or expired".to_string(), + "source file is deleted, expired, or its content has been retired".to_string(), )); } @@ -1162,6 +1175,61 @@ impl PostgresRequestManager

{ self } + /// Refuse to retry a batch near the end of its content retention. + /// + /// A retry makes a batch unfinished again, and an unfinished batch holds + /// its archive week (results) and its input's template week open past + /// the retention period `limit`. To keep that hold short and + /// predictable, a batch is not retried once + /// `now > created_at + limit - 1 day`, measured from the batch's + /// creation and from its input file's upload (a retry re-reads the + /// input templates). A week drops no earlier than + /// `week_start + 7 days + limit`, and everything in it was created at or + /// after `week_start`, so the one-day margin puts the cut-off before the + /// drop, never after. `None` (the default) means content is not retired + /// and retries have no age limit. + pub fn with_retry_age_limit(mut self, limit: Option) -> Self { + self.retry_age_limit = limit; + self + } + + /// Refuse a retry once the batch is within a day of the end of its + /// content retention; see [`Self::with_retry_age_limit`]. A missing + /// batch passes, so callers keep their own not-found handling. + async fn check_retry_age<'e, E>(&self, executor: E, batch_id: Uuid) -> Result<()> + where + E: sqlx::Executor<'e, Database = sqlx::Postgres>, + { + let Some(limit) = self.retry_age_limit else { + return Ok(()); + }; + // created_at never changes, so this needs no lock. A batch is created + // after its input file, so the file's upload is the earlier clock + // whenever there is one. + let past: Option = sqlx::query_scalar( + r#" + SELECT NOW() > LEAST(b.created_at, COALESCE(f.created_at, b.created_at)) + + $2 * INTERVAL '1 second' - INTERVAL '1 day' + FROM batches b + LEFT JOIN files f ON f.id = b.file_id + WHERE b.id = $1 + "#, + ) + .bind(batch_id) + .bind(limit.num_seconds() as f64) + .fetch_optional(executor) + .await + .map_err(|e| FusilladeError::Other(anyhow!("Failed to check batch retry age: {e}")))?; + if past == Some(true) { + return Err(FusilladeError::ValidationError(format!( + "batch is too old to retry: retries are accepted until one day before its \ + content reaches the end of its {}-day retention period", + limit.num_days() + ))); + } + Ok(()) + } + /// Output and error files created before they carried an expiry read as /// expiring `file_expiry.default` after creation: the value they would /// have been given. @@ -5706,6 +5774,9 @@ impl Storage for PostgresRequestManager

{ ) -> Result>> { tracing::debug!(%batch_id, count = requested_ids.len(), "Retrying failed requests"); + self.check_retry_age(self.write_executor(), *batch_id) + .await?; + // One batch-driven lookup, matching retry_failed_requests_for_batch: // resolve the batch first, then read only its own rows (live, plus // archived rows in its bucket) filtered by the requested ids. Ids from @@ -5774,6 +5845,7 @@ impl Storage for PostgresRequestManager

{ .map(|(id, _)| (*id, Some(*batch_id))) .collect(); + let mut source_unavailable = false; let repended: std::collections::HashSet = if retryable.is_empty() { Default::default() } else { @@ -5820,15 +5892,21 @@ impl Storage for PostgresRequestManager

{ let available_file_ids: Vec = if source_file_ids.is_empty() { Vec::new() } else { + // Share-lock each source's template week, as batch creation + // does, so retirement cannot fence it under this retry. sqlx::query_scalar( r#" - SELECT id FROM files - WHERE id = ANY($1) - AND deleted_at IS NULL - AND status = 'processed' - AND (expires_at IS NULL OR expires_at > NOW()) - ORDER BY id - FOR UPDATE + SELECT f.id FROM files f + JOIN request_template_buckets bucket + ON bucket.week_start = date_trunc('week', f.created_at AT TIME ZONE 'UTC')::date + WHERE f.id = ANY($1) + AND f.deleted_at IS NULL + AND f.status = 'processed' + AND (f.expires_at IS NULL OR f.expires_at > NOW()) + AND bucket.state = 'active' + ORDER BY f.id + FOR UPDATE OF f + FOR SHARE OF bucket "#, ) .bind(&source_file_ids) @@ -5836,6 +5914,9 @@ impl Storage for PostgresRequestManager

{ .await .map_err(|_| FusilladeError::Other(anyhow!("Failed to lock retry source files")))? }; + // Every requested row belongs to this one batch; its rows fall + // out below and each reports the unavailable source. + source_unavailable = available_file_ids.len() < source_file_ids.len(); let safe_batch_ids: std::collections::HashSet = if available_file_ids.is_empty() { Default::default() } else { @@ -6000,6 +6081,11 @@ impl Storage for PostgresRequestManager

{ Some((state, true)) if state == "failed" => { if repended.contains(&**id) { Ok(()) + } else if source_unavailable { + Err(FusilladeError::ValidationError( + "batch source file is deleted, expired, or its content has been retired" + .to_string(), + )) } else { Err(FusilladeError::InvalidState( *id, @@ -6050,25 +6136,36 @@ impl Storage for PostgresRequestManager

{ )); }; - let source_available = sqlx::query_scalar::<_, bool>( - r#"SELECT EXISTS( - SELECT 1 FROM files - WHERE id = $1 - AND deleted_at IS NULL - AND status = 'processed' - AND (expires_at IS NULL OR expires_at > NOW()) - FOR UPDATE - )"#, + self.check_retry_age(&mut *tx, *batch_id).await?; + + // Share-lock the input's template week until commit, as batch + // creation does: retirement either waits and then sees this batch + // unfinished, or fenced first and the week reads as not active. + let source_available = sqlx::query_scalar::<_, Uuid>( + r#" + SELECT f.id FROM files f + JOIN request_template_buckets bucket + ON bucket.week_start = date_trunc('week', f.created_at AT TIME ZONE 'UTC')::date + WHERE f.id = $1 + AND f.deleted_at IS NULL + AND f.status = 'processed' + AND (f.expires_at IS NULL OR f.expires_at > NOW()) + AND bucket.state = 'active' + FOR UPDATE OF f + FOR SHARE OF bucket + "#, ) .bind(source_file_id) - .fetch_one(&mut *tx) + .fetch_optional(&mut *tx) .await .map_err(|e| { FusilladeError::Other(anyhow!("Failed to lock batch source for retry: {}", e)) - })?; + })? + .is_some(); if !source_available { return Err(FusilladeError::ValidationError( - "batch source file is deleted or expired".to_string(), + "batch source file is deleted, expired, or its content has been retired" + .to_string(), )); } @@ -6114,6 +6211,28 @@ impl Storage for PostgresRequestManager

{ routing.0 ))); }; + // The failed/canceled rows and the completions the batch keeps + // live in this archive week. Share-lock its registry row until + // commit, as archive_batch does: once retirement has fenced the + // week, the move-back below would read a missing partition, find + // nothing, and reset the batch over completions that are gone. + // A fence that comes later waits for this commit, then sees the + // batch unfinished. + let bucket_state: Option = sqlx::query_scalar( + "SELECT state FROM batch_archive_buckets WHERE week_start = $1 FOR SHARE", + ) + .bind(bucket) + .fetch_optional(&mut *tx) + .await + .map_err(|e| { + FusilladeError::Other(anyhow!("Failed to lock archive bucket for retry: {}", e)) + })?; + if bucket_state.as_deref() != Some("active") { + return Err(FusilladeError::ValidationError( + "batch results have been retired; the batch can no longer be retried" + .to_string(), + )); + } let moved = sqlx::query!( r#" INSERT INTO requests (id, batch_id, template_id, state, retry_attempt, not_before, @@ -8811,15 +8930,6 @@ impl DaemonStorage for PostgresRequestManager

{ .await } - async fn expire_file_content( - &self, - retention_days: i32, - scan_limit: i64, - after: Option, - ) -> Result { - template_retirement::expire_file_content(self, retention_days, scan_limit, after).await - } - async fn cleanup_retired_template_routes(&self, limit: i64) -> Result { template_retirement::cleanup_retired_template_routes(self, limit).await } diff --git a/fusillade-arsenal/src/postgres/batch_archive_retirement.rs b/fusillade-arsenal/src/postgres/batch_archive_retirement.rs index 162f471fe..ebfc21a24 100644 --- a/fusillade-arsenal/src/postgres/batch_archive_retirement.rs +++ b/fusillade-arsenal/src/postgres/batch_archive_retirement.rs @@ -1,10 +1,17 @@ //! Weekly batch-archive partition retirement. //! //! Deletes archived batch response content by dropping whole weekly -//! `batch_requests_archive` partitions once every batch in the week is past -//! its content-retention period. Batch metadata rows are never deleted: -//! completion atomically stamps `batches.retention_expired_at` so the API -//! reports expiry while the account-lifetime metadata survives. +//! `batch_requests_archive` partitions once the week is past the retention +//! period measured from the week itself, so a batch's completions live a +//! fixed time from its creation, like its input file. Batch metadata rows are +//! never deleted. The API reports results files as expired from the +//! `expires_at` they were given at creation, so finished batches are not +//! stamped at retirement: a write to every batch in the week grows with the +//! week (a million rows in one statement, cancelled by the statement timeout) +//! and the week could never retire. Only the week's explicitly deleted +//! batches with nothing left in `requests` are stamped, which takes them out +//! of the erasure purge's tombstone set once their archived rows are gone; +//! there are few of them. use crate::error::Result; use crate::manager::RetainedResponseRetirementOutcome; @@ -65,12 +72,24 @@ const BATCH_ARCHIVE_FAMILY: FamilySpec = FamilySpec { ) ) "#, - // A weekly bucket is eligible only when the whole week is past the - // retention horizon AND no batch bucketed in it can still need its - // content: every one is fully archived, frozen, and individually past - // its own finalization-anchored retention period. Any live, split, - // unfrozen, or younger batch fails the gate closed. The scan is over - // batch metadata only. + // A weekly bucket is eligible when the whole week is past the retention + // horizon and no batch created in that week is still running: such a + // batch may yet write results into the partition. A finished batch holds + // nothing open, however late it finished; one retry no longer extends + // the week for every batch in it. The guard reads only unfinished + // batches, a few dozen rows under idx_batches_unfinished_created_at. + // + // A retried archived batch ('split') also holds the week its rows were + // archived into, finished or not: its retried rows are back in + // `requests` and archive_batch re-archives them into that same + // `archive_bucket`, which it refuses once the week is retired. Without + // this the rows would stay in `requests`, outside retention, for good. + // Retry share-locks the bucket row and refuses unless the week is + // active, so a week cannot be fenced under a batch going 'split'. The + // two arms read split batches through the partial indexes that already + // hold them: idx_batches_unfinished_created_at for running ones, + // idx_batches_archivable for finished ones awaiting re-archive. A + // deleted split batch holds nothing: the erasure purge removes its rows. candidate_sql: r#" SELECT bucket.partition_schema, namespace.oid AS partition_schema_oid, @@ -91,27 +110,75 @@ const BATCH_ARCHIVE_FAMILY: FamilySpec = FamilySpec { AND NOT EXISTS ( SELECT 1 FROM batches b - WHERE b.archive_bucket >= bucket.week_start - AND b.archive_bucket < bucket.week_start + 7 - AND ( - b.location <> 'archive' - OR b.counts_frozen_at IS NULL - OR (b.counts_frozen_at + make_interval(days => $1)) - > statement_timestamp() - ) + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.created_at >= (bucket.week_start::timestamp AT TIME ZONE 'UTC') + AND b.created_at < ((bucket.week_start + 7)::timestamp AT TIME ZONE 'UTC') + ) + AND NOT EXISTS ( + SELECT 1 + FROM batches b + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.location = 'split' + AND b.archive_bucket = bucket.week_start + ) + AND NOT EXISTS ( + SELECT 1 + FROM batches b + WHERE b.location IN ('live', 'split') + AND b.counts_frozen_at IS NOT NULL + AND b.deleted_at IS NULL + AND b.location = 'split' + AND b.archive_bucket = bucket.week_start ) ORDER BY bucket.week_start LIMIT 1 "#, candidate_binds_retention: true, - // Committed atomically with the journal/bucket completion: batch - // metadata survives, stamped with the exact deletion timestamp so the - // API reports expired content and the explicit-erasure purge never - // treats these tombstones as its own. + // The candidate's guards, re-run under the bucket lock. + recheck_sql: Some( + r#" + SELECT EXISTS ( + SELECT 1 + FROM batches b + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.created_at >= ($1::date::timestamp AT TIME ZONE 'UTC') + AND b.created_at < (($1::date + 7)::timestamp AT TIME ZONE 'UTC') + ) OR EXISTS ( + SELECT 1 + FROM batches b + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.location = 'split' + AND b.archive_bucket = $1::date + ) OR EXISTS ( + SELECT 1 + FROM batches b + WHERE b.location IN ('live', 'split') + AND b.counts_frozen_at IS NOT NULL + AND b.deleted_at IS NULL + AND b.location = 'split' + AND b.archive_bucket = $1::date + ) + "#, + ), + // Takes the week's explicitly deleted batches out of the erasure purge's + // tombstone set (`deleted_at IS NOT NULL AND retention_expired_at IS + // NULL`): their archived rows have just gone with the partition, and + // otherwise every purge tick would probe them forever. Only deleted + // batches are stamped, read through that set's partial index + // idx_batches_explicitly_deleted, so the write stays small however large + // the week; finished batches are left alone. A deleted batch that was + // retried can still have rows in `requests`, and the purge erases those + // only while the batch is in the set, so it is not stamped: it stays a + // tombstone, at the cost of one empty probe per purge tick. completion_sql: Some( "UPDATE batches SET retention_expired_at = $3 \ - WHERE archive_bucket >= $1 AND archive_bucket < $2 \ - AND retention_expired_at IS NULL", + WHERE deleted_at IS NOT NULL AND retention_expired_at IS NULL \ + AND archive_bucket >= $1 AND archive_bucket < $2 \ + AND NOT EXISTS (SELECT 1 FROM requests r WHERE r.batch_id = batches.id)", ), }; diff --git a/fusillade-arsenal/src/postgres/partition_retirement.rs b/fusillade-arsenal/src/postgres/partition_retirement.rs index 49366b307..90567dbc2 100644 --- a/fusillade-arsenal/src/postgres/partition_retirement.rs +++ b/fusillade-arsenal/src/postgres/partition_retirement.rs @@ -44,6 +44,12 @@ pub(super) struct FamilySpec { /// when `candidate_binds_retention`. pub candidate_sql: &'static str, pub candidate_binds_retention: bool, + /// Optional boolean query, `$1 = lower bound`, that is true while + /// something still holds the bucket open (the family's unfinished-batch + /// guard). It is re-run after the bucket row is locked and before the + /// bucket is fenced, so a writer that committed while the claim waited + /// for that lock is not missed; see [`claim`]. + pub recheck_sql: Option<&'static str>, /// Optional metadata statement committed atomically with journal/bucket /// completion. Binds: $1 lower bound, $2 upper bound, $3 completion /// timestamp. Must touch only content-free metadata. @@ -476,6 +482,30 @@ async fn claim( return Err(mismatch()); } + // The candidate query read the guard before the bucket row was locked. + // Writers that make content depend on a bucket (batch creation, retry) + // share-lock its row until they commit, so the FOR UPDATE above waited + // for any of them still in flight. Under READ COMMITTED each statement + // takes a fresh snapshot, so this new statement sees what they + // committed; reusing the candidate's answer would not. A writer that + // arrives after this point waits for the fence to commit and then reads + // the bucket as no longer active. A recovering claim already fenced the + // bucket earlier and must finish. + if !recovering && let Some(recheck_sql) = family.recheck_sql { + let held: bool = sqlx::query_scalar(recheck_sql) + .bind(identity.lower) + .fetch_one(&mut *transaction) + .await + .map_err(retirement_database_error)?; + if held { + transaction + .rollback() + .await + .map_err(retirement_database_error)?; + return Ok(Claim::None); + } + } + if recovering { let renew_statement = format!( "UPDATE retention_partition_retirements \ diff --git a/fusillade-arsenal/src/postgres/retained_response_retirement.rs b/fusillade-arsenal/src/postgres/retained_response_retirement.rs index 4c3d6b4e6..5f020b083 100644 --- a/fusillade-arsenal/src/postgres/retained_response_retirement.rs +++ b/fusillade-arsenal/src/postgres/retained_response_retirement.rs @@ -82,6 +82,7 @@ const DAILY_RESPONSE_FAMILY: FamilySpec = FamilySpec { LIMIT 1 "#, candidate_binds_retention: false, + recheck_sql: None, completion_sql: None, }; diff --git a/fusillade-arsenal/src/postgres/template_retirement.rs b/fusillade-arsenal/src/postgres/template_retirement.rs index 30f6e5764..680584fb5 100644 --- a/fusillade-arsenal/src/postgres/template_retirement.rs +++ b/fusillade-arsenal/src/postgres/template_retirement.rs @@ -1,18 +1,16 @@ -//! Weekly generation-2 template retirement and the file-content expiry that -//! unpins it. +//! Weekly generation-2 template retirement. //! //! Request input payloads are deleted by dropping whole weekly -//! `request_templates_g2` partitions. A week is eligible only when it is past -//! the retention horizon AND no live (undeleted) file still owns templates in -//! it — file metadata rows are never deleted by this lifecycle; scheduled -//! expiry only tombstones them once every referencing batch's content is -//! itself expired or erased. +//! `request_templates_g2` partitions. A week is eligible when it is past the +//! retention horizon, measured from the week itself, and no unfinished batch +//! still reads templates from it. File metadata rows are never touched: the +//! API reports an input file as expired from its own `expires_at`, which the +//! upload cap keeps no later than the week's drop. use crate::error::Result; -use crate::manager::{FileExpiryCursor, FileExpiryStep, RetainedResponseRetirementOutcome}; +use crate::manager::RetainedResponseRetirementOutcome; use crate::{PoolProvider, PostgresRequestManager}; -use chrono::{DateTime, Datelike, NaiveDate, Utc, Weekday}; -use uuid::Uuid; +use chrono::{Datelike, NaiveDate, Weekday}; use super::partition_retirement::{FamilySpec, failed, retire}; @@ -68,11 +66,16 @@ const TEMPLATE_FAMILY: FamilySpec = FamilySpec { ) ) "#, - // Reference-gated eligibility: the week must be past the horizon, no - // LIVE file may still own templates in it, and any row without a - // resolvable owning file fails the gate closed (nothing unmanaged is - // ever dropped implicitly). The scans touch one candidate partition and - // metadata tables only. + // The week must be past the horizon and no unfinished batch may still + // read templates from it: a batch's templates are in the week its input + // file was uploaded, so the guard joins the few dozen unfinished batches + // to their files. A batch is always created after its input file, so + // only batches created since the week began can read it; that bound is + // what lets the guard range-scan idx_batches_unfinished_created_at + // rather than read every unfinished batch. A finished batch + // holds nothing open; its results do not need the templates. A dedicated + // (file-less) template row still fails the gate closed: nothing + // unmanaged is dropped implicitly. candidate_sql: r#" SELECT bucket.partition_schema, namespace.oid AS partition_schema_oid, @@ -92,28 +95,40 @@ const TEMPLATE_FAMILY: FamilySpec = FamilySpec { <= statement_timestamp() AT TIME ZONE 'UTC' AND NOT EXISTS ( SELECT 1 - FROM request_templates_g2 template - JOIN files f ON f.id = template.file_id - WHERE template.created_on >= bucket.week_start - AND template.created_on < bucket.week_start + 7 - AND f.deleted_at IS NULL + FROM batches b + JOIN files f ON f.id = b.file_id + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.created_at >= (bucket.week_start::timestamp AT TIME ZONE 'UTC') + AND f.created_at >= (bucket.week_start::timestamp AT TIME ZONE 'UTC') + AND f.created_at < ((bucket.week_start + 7)::timestamp AT TIME ZONE 'UTC') ) AND NOT EXISTS ( SELECT 1 FROM request_templates_g2 template WHERE template.created_on >= bucket.week_start AND template.created_on < bucket.week_start + 7 - AND ( - template.file_id IS NULL - OR NOT EXISTS ( - SELECT 1 FROM files f WHERE f.id = template.file_id - ) - ) + AND template.file_id IS NULL ) ORDER BY bucket.week_start LIMIT 1 "#, candidate_binds_retention: true, + // The candidate's unfinished-batch guard, re-run under the bucket lock. + recheck_sql: Some( + r#" + SELECT EXISTS ( + SELECT 1 + FROM batches b + JOIN files f ON f.id = b.file_id + WHERE b.counts_frozen_at IS NULL + AND b.deleted_at IS NULL + AND b.created_at >= ($1::date::timestamp AT TIME ZONE 'UTC') + AND f.created_at >= ($1::date::timestamp AT TIME ZONE 'UTC') + AND f.created_at < (($1::date + 7)::timestamp AT TIME ZONE 'UTC') + ) + "#, + ), completion_sql: None, }; @@ -125,110 +140,6 @@ pub(super) async fn retire_expired_template_partition( retire(manager, &TEMPLATE_FAMILY, select_new, Some(retention_days)).await } -/// One step of [`expire_file_content`]: take the next `$2` due files after -/// the cursor (`$3`, `$4`), oldest upload first, and tombstone those that no -/// live batch still pins. -/// -/// The slice bounds the files examined, not the files expired. Batches keep -/// their content for longer than their input files, so most due files stay -/// pinned for weeks, and a limit on expired rows alone would examine every -/// one of them in each statement. -/// -/// The age bound and the cursor are written against the bare columns so -/// `idx_files_content_expiry_due` serves the slice as a range read. -const EXPIRE_FILE_CONTENT: &str = r#" - WITH slice AS MATERIALIZED ( - SELECT id, created_at - FROM files - WHERE purpose = 'batch' - AND deleted_at IS NULL - AND created_at <= statement_timestamp() - make_interval(days => $1) - AND (created_at, id) > ( - COALESCE($3::timestamptz, '-infinity'::timestamptz), - COALESCE($4::uuid, '00000000-0000-0000-0000-000000000000'::uuid) - ) - ORDER BY created_at, id - LIMIT $2 - ), - candidates AS MATERIALIZED ( - SELECT files.id - FROM slice - JOIN files ON files.id = slice.id - WHERE files.deleted_at IS NULL - AND NOT EXISTS ( - SELECT 1 FROM batches b - WHERE b.file_id = files.id - AND b.deleted_at IS NULL - AND b.retention_expired_at IS NULL - ) - FOR UPDATE OF files SKIP LOCKED - ), - expired AS ( - UPDATE files - SET deleted_at = statement_timestamp(), - retention_expired_at = statement_timestamp(), - status = 'expired' - FROM candidates - WHERE files.id = candidates.id - AND files.deleted_at IS NULL - RETURNING 1 - ) - SELECT (SELECT COUNT(*) FROM expired) AS expired, - (SELECT COUNT(*) FROM slice) AS examined, - (SELECT created_at FROM slice ORDER BY created_at DESC, id DESC LIMIT 1) AS last_created_at, - (SELECT id FROM slice ORDER BY created_at DESC, id DESC LIMIT 1) AS last_id - "#; - -/// Tombstone input files whose content has aged out: the file row survives -/// with `retention_expired_at` set, downloads and new batch creation fail, -/// and its template window becomes droppable. A file is eligible only when -/// every batch that referenced it has already had its own content expired or -/// explicitly erased. -/// -/// Examines at most `scan_limit` due files after `after`. A slice shorter -/// than `scan_limit` reached the retention horizon, so `next` is `None`. -pub(super) async fn expire_file_content( - manager: &PostgresRequestManager

, - retention_days: i32, - scan_limit: i64, - after: Option, -) -> Result { - if retention_days < 1 || scan_limit < 1 { - return Ok(FileExpiryStep { - expired: 0, - next: None, - }); - } - let mut transaction = manager - .begin_maintenance_write() - .await - .map_err(|_| failed())?; - let (expired, examined, last_created_at, last_id): ( - i64, - i64, - Option>, - Option, - ) = sqlx::query_as(EXPIRE_FILE_CONTENT) - .bind(retention_days) - .bind(scan_limit) - .bind(after.map(|cursor| cursor.created_at)) - .bind(after.map(|cursor| cursor.id)) - .fetch_one(&mut *transaction) - .await - .map_err(|_| failed())?; - transaction.commit().await.map_err(|_| failed())?; - let next = match (last_created_at, last_id) { - (Some(created_at), Some(id)) if examined >= scan_limit => { - Some(FileExpiryCursor { created_at, id }) - } - _ => None, - }; - Ok(FileExpiryStep { - expired: u64::try_from(expired).map_err(|_| failed())?, - next, - }) -} - /// Bounded removal of location-oracle rows whose weekly bucket has been /// physically retired. Templates need no resurrection fences: their ids are /// never re-targeted by late writers. @@ -275,157 +186,3 @@ pub(super) async fn cleanup_retired_template_routes( transaction.commit().await.map_err(|_| failed())?; u64::try_from(deleted).map_err(|_| failed()) } - -#[cfg(test)] -mod tests { - use super::*; - use crate::TestDbPools; - - /// A long history of recent uploads must not be scanned to find the few - /// files past the retention horizon. - #[sqlx::test] - async fn expiry_reads_only_files_past_the_horizon(pool: sqlx::PgPool) { - sqlx::query( - r#" - INSERT INTO files (name, purpose, created_at) - SELECT 'upload-' || i, 'batch', - CASE WHEN i <= 5 THEN now() - interval '60 days' ELSE now() END - FROM generate_series(1, 20005) i - "#, - ) - .execute(&pool) - .await - .unwrap(); - sqlx::query("ANALYZE files").execute(&pool).await.unwrap(); - - let mut tx = pool.begin().await.unwrap(); - sqlx::query("SET LOCAL plan_cache_mode = force_generic_plan") - .execute(&mut *tx) - .await - .unwrap(); - sqlx::raw_sql(&format!( - "PREPARE expire_file_content(int, bigint, timestamptz, uuid) AS {EXPIRE_FILE_CONTENT}" - )) - .execute(&mut *tx) - .await - .unwrap(); - let plan: serde_json::Value = sqlx::query_scalar( - "EXPLAIN (FORMAT JSON) EXECUTE expire_file_content(30, 100, NULL, NULL)", - ) - .fetch_one(&mut *tx) - .await - .unwrap(); - tx.rollback().await.unwrap(); - // The index must bound the scan at the horizon and start it at the - // cursor, not merely supply the ordering: both are index conditions. - fn find_node<'a>( - node: &'a serde_json::Value, - index: &str, - ) -> Option<&'a serde_json::Value> { - if node["Index Name"] == index { - return Some(node); - } - node["Plans"] - .as_array()? - .iter() - .find_map(|child| find_node(child, index)) - } - let node = - find_node(&plan[0]["Plan"], "idx_files_content_expiry_due").unwrap_or_else(|| { - panic!("expiry candidates must come from idx_files_content_expiry_due: {plan}") - }); - assert!( - matches!( - node["Node Type"].as_str(), - Some("Index Scan" | "Index Only Scan") - ), - "{plan}" - ); - let cond = node["Index Cond"].as_str().unwrap_or_default(); - assert!( - cond.contains("created_at <="), - "age bound must be an index condition: {plan}" - ); - assert!( - cond.contains("ROW(created_at, id) >"), - "cursor must be an index condition: {plan}" - ); - - let manager = PostgresRequestManager::new( - TestDbPools::new(pool.clone()).await.unwrap(), - Default::default(), - ); - let step = expire_file_content(&manager, 30, 100, None).await.unwrap(); - assert_eq!( - step, - FileExpiryStep { - expired: 5, - next: None - } - ); - let step = expire_file_content(&manager, 30, 100, None).await.unwrap(); - assert_eq!(step.expired, 0); - } - - /// Due files that live batches still pin must not make a step examine - /// more than its slice: the walk covers them in bounded steps and expires - /// only the unpinned files among them. - #[sqlx::test] - async fn expiry_walks_pinned_files_in_bounded_steps(pool: sqlx::PgPool) { - // Seven due files, oldest first; every other one is pinned by a batch - // whose content has not expired. - sqlx::query( - r#" - INSERT INTO files (name, purpose, created_at) - SELECT 'upload-' || i, 'batch', now() - interval '60 days' + i * interval '1 minute' - FROM generate_series(1, 7) i - "#, - ) - .execute(&pool) - .await - .unwrap(); - let files: Vec = sqlx::query_scalar("SELECT id FROM files ORDER BY created_at") - .fetch_all(&pool) - .await - .unwrap(); - let pinned: Vec = files.iter().copied().step_by(2).collect(); - sqlx::query( - "INSERT INTO batches (file_id, endpoint, completion_window, expires_at) \ - SELECT id, '/v1/chat/completions', '24h', now() FROM UNNEST($1::uuid[]) AS id", - ) - .bind(&pinned) - .execute(&pool) - .await - .unwrap(); - - let manager = PostgresRequestManager::new( - TestDbPools::new(pool.clone()).await.unwrap(), - Default::default(), - ); - let mut steps = Vec::new(); - let mut after = None; - loop { - let step = expire_file_content(&manager, 30, 3, after).await.unwrap(); - steps.push(step.expired); - match step.next { - Some(cursor) => after = Some(cursor), - None => break, - } - } - // Slices of 3, 3 and 1 files; the last one is short, ending the walk. - assert_eq!(steps, vec![1, 2, 0]); - - let expired: Vec = sqlx::query_scalar( - "SELECT id FROM files WHERE id = ANY($1) AND retention_expired_at IS NOT NULL", - ) - .bind(&files) - .fetch_all(&pool) - .await - .unwrap(); - let mut unpinned: Vec = files.iter().copied().skip(1).step_by(2).collect(); - let mut expired = expired; - unpinned.sort(); - expired.sort(); - assert_eq!(expired, unpinned); - } -} diff --git a/fusillade-arsenal/tests/batch_archive_retirement.rs b/fusillade-arsenal/tests/batch_archive_retirement.rs index b43bf6283..03b2cfeee 100644 --- a/fusillade-arsenal/tests/batch_archive_retirement.rs +++ b/fusillade-arsenal/tests/batch_archive_retirement.rs @@ -70,8 +70,9 @@ async fn archived_batch( .unwrap(); let batch_id: Uuid = sqlx::query_scalar( "INSERT INTO batches (file_id, endpoint, completion_window, expires_at, \ - location, archive_bucket, counts_frozen_at) \ - VALUES ($1, '/v1/x', '24h', $3 + INTERVAL '24 hours', 'archive', $2, $3) RETURNING id", + location, archive_bucket, counts_frozen_at, created_at) \ + VALUES ($1, '/v1/x', '24h', $3 + INTERVAL '24 hours', 'archive', $2, $3, \ + ($2::timestamp AT TIME ZONE 'UTC') + INTERVAL '1 day') RETURNING id", ) .bind(file_id) .bind(week_start) @@ -115,10 +116,16 @@ async fn manager(pool: &PgPool) -> PostgresRequestManager { } #[sqlx::test] -async fn an_expired_week_retires_and_stamps_batch_metadata(pool: PgPool) { +async fn an_expired_week_retires_without_touching_batch_metadata(pool: PgPool) { let week = monday(10); ensure_week(&pool, week).await; let batch = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; + let deleted = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; + sqlx::query("UPDATE batches SET deleted_at = NOW() WHERE id = $1") + .bind(deleted.batch_id) + .execute(&pool) + .await + .unwrap(); let outcome = manager(&pool) .await @@ -149,19 +156,26 @@ async fn an_expired_week_retires_and_stamps_batch_metadata(pool: PgPool) { assert_eq!(state, "retired"); assert!(journal_done); - // The batch metadata row survives, stamped with the deletion timestamp. - let (expired_at, journal_completed): (Option>, DateTime) = sqlx::query_as( - "SELECT b.retention_expired_at, journal.completed_at \ - FROM batches b, retention_partition_retirements journal \ - WHERE b.id = $1 AND journal.parent_table = 'batch_requests_archive' \ - AND journal.lower_bound = $2", - ) - .bind(batch.batch_id) - .bind(week) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(expired_at, Some(journal_completed)); + // The batch metadata row survives untouched: results files report + // expiry from their own expires_at, and a per-batch write here would + // grow with the week. + let expired_at: Option> = + sqlx::query_scalar("SELECT retention_expired_at FROM batches WHERE id = $1") + .bind(batch.batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(expired_at, None); + + // An explicitly deleted batch leaves the erasure purge's tombstone set: + // its archived rows went with the partition. + let deleted_expired_at: Option> = + sqlx::query_scalar("SELECT retention_expired_at FROM batches WHERE id = $1") + .bind(deleted.batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(deleted_expired_at.is_some()); let further = manager(&pool) .await @@ -171,35 +185,57 @@ async fn an_expired_week_retires_and_stamps_batch_metadata(pool: PgPool) { assert_eq!(further, RetainedResponseRetirementOutcome::NoCandidate); } +/// Retention is measured from the week, not from when each batch finished: +/// one batch retried weeks later must not keep every other batch's +/// completions in the week alive. #[sqlx::test] -async fn a_batch_inside_its_retention_period_blocks_the_week(pool: PgPool) { +async fn a_late_finish_does_not_hold_the_week(pool: PgPool) { let week = monday(10); ensure_week(&pool, week).await; archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; - // A second batch in the same week finalized recently. + // A second batch in the same week finalized yesterday. archived_batch(&pool, week, Utc::now() - Duration::days(1)).await; + let outcome = manager(&pool) + .await + .retire_expired_batch_archive_partition(true, 30) + .await + .unwrap(); + assert_eq!(outcome, RetainedResponseRetirementOutcome::Retired); +} + +/// The week is not yet past `week end + retention days`. +#[sqlx::test] +async fn a_young_week_is_not_a_candidate(pool: PgPool) { + let week = monday(4); + ensure_week(&pool, week).await; + archived_batch(&pool, week, Utc::now() - Duration::days(20)).await; + let outcome = manager(&pool) .await .retire_expired_batch_archive_partition(true, 30) .await .unwrap(); assert_eq!(outcome, RetainedResponseRetirementOutcome::NoCandidate); - let bucket_state: String = - sqlx::query_scalar("SELECT state FROM batch_archive_buckets WHERE week_start = $1") - .bind(week) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(bucket_state, "active"); } +/// A batch that is still running and was created in the week holds it open: +/// it may yet write results into the partition. So does a retried archived +/// batch ('split'), running or finished, whose rows must be re-archived into +/// the week. A finished batch otherwise holds nothing open, a running batch +/// from another week is not this week's concern, and a deleted split batch +/// is the erasure purge's. #[sqlx::test] -async fn live_split_or_unfrozen_batches_fail_the_gate_closed(pool: PgPool) { - for (index, (location, frozen)) in [ - ("live", Some(Utc::now() - Duration::days(60))), - ("split", Some(Utc::now() - Duration::days(60))), - ("archive", None), +async fn only_running_or_split_batches_block(pool: PgPool) { + use RetainedResponseRetirementOutcome::{NoCandidate, Retired}; + let frozen = Some(Utc::now() - Duration::days(60)); + for (index, (location, frozen, created_in_week, deleted, expected)) in [ + ("live", frozen, true, false, Retired), + ("split", frozen, true, false, NoCandidate), + ("split", frozen, true, true, Retired), + ("split", None, false, false, NoCandidate), + ("archive", None, true, false, NoCandidate), + ("archive", None, false, false, Retired), ] .into_iter() .enumerate() @@ -207,13 +243,20 @@ async fn live_split_or_unfrozen_batches_fail_the_gate_closed(pool: PgPool) { let week = monday(20 + index as i64); ensure_week(&pool, week).await; let batch = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; - sqlx::query("UPDATE batches SET location = $2, counts_frozen_at = $3 WHERE id = $1") - .bind(batch.batch_id) - .bind(location) - .bind(frozen) - .execute(&pool) - .await - .unwrap(); + sqlx::query( + "UPDATE batches SET location = $2, counts_frozen_at = $3, \ + created_at = CASE WHEN $4 THEN created_at ELSE NOW() END, \ + deleted_at = CASE WHEN $5 THEN NOW() END \ + WHERE id = $1", + ) + .bind(batch.batch_id) + .bind(location) + .bind(frozen) + .bind(created_in_week) + .bind(deleted) + .execute(&pool) + .await + .unwrap(); let outcome = manager(&pool) .await @@ -221,10 +264,12 @@ async fn live_split_or_unfrozen_batches_fail_the_gate_closed(pool: PgPool) { .await .unwrap(); assert_eq!( - outcome, - RetainedResponseRetirementOutcome::NoCandidate, - "location={location} frozen={frozen:?} must block retirement" + outcome, expected, + "location={location} frozen={frozen:?} created_in_week={created_in_week} deleted={deleted}" ); + if expected == Retired { + continue; + } sqlx::query(&format!( "ALTER TABLE batch_requests_archive DETACH PARTITION {}", child_name(week) @@ -244,6 +289,195 @@ async fn live_split_or_unfrozen_batches_fail_the_gate_closed(pool: PgPool) { } } +/// Make `batch` a retried archived batch: its completed row stays in the +/// archive and a retried row is back in `requests`. +async fn split_with_live_row(pool: &PgPool, batch: &ArchivedBatch) -> Uuid { + sqlx::query("UPDATE batches SET location = 'split' WHERE id = $1") + .bind(batch.batch_id) + .execute(pool) + .await + .unwrap(); + sqlx::query_scalar( + "INSERT INTO requests (id, batch_id, model, state, retry_attempt, created_at, updated_at, \ + completed_at, response_status, response_body, response_size) \ + VALUES (gen_random_uuid(), $1, 'test-model', 'completed', 1, NOW(), NOW(), NOW(), \ + 200, '{\"retried\":true}', 16) RETURNING id", + ) + .bind(batch.batch_id) + .fetch_one(pool) + .await + .unwrap() +} + +/// A finished retried batch holds its archive week until its retried rows +/// are archived again: archive_batch refuses a retired week, so retiring +/// first would leave them in `requests` past retention for good. +#[sqlx::test] +async fn a_retried_batch_holds_its_week_until_rearchived(pool: PgPool) { + let week = monday(10); + ensure_week(&pool, week).await; + let batch = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; + split_with_live_row(&pool, &batch).await; + let manager = manager(&pool).await; + + assert_eq!( + manager + .retire_expired_batch_archive_partition(true, 30) + .await + .unwrap(), + RetainedResponseRetirementOutcome::NoCandidate + ); + + assert_eq!( + manager + .archive_batch(fusillade_arsenal::batch::BatchId(batch.batch_id)) + .await + .unwrap(), + fusillade_arsenal::ArchiveOutcome::Archived { rows: 1 } + ); + assert_eq!( + manager + .retire_expired_batch_archive_partition(true, 30) + .await + .unwrap(), + RetainedResponseRetirementOutcome::Retired + ); + let live: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM requests WHERE batch_id = $1") + .bind(batch.batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(live, 0); +} + +/// A deleted batch that was retried still has rows in `requests` when its +/// archive week drops. The purge erases those only while the batch is in its +/// tombstone set, so retirement must leave it there. +#[sqlx::test] +async fn a_deleted_batch_with_live_rows_stays_in_the_purge_set(pool: PgPool) { + let week = monday(10); + ensure_week(&pool, week).await; + let batch = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; + let live_row = split_with_live_row(&pool, &batch).await; + sqlx::query("UPDATE batches SET deleted_at = NOW() WHERE id = $1") + .bind(batch.batch_id) + .execute(&pool) + .await + .unwrap(); + let manager = manager(&pool).await; + + assert_eq!( + manager + .retire_expired_batch_archive_partition(true, 30) + .await + .unwrap(), + RetainedResponseRetirementOutcome::Retired + ); + let expired_at: Option> = + sqlx::query_scalar("SELECT retention_expired_at FROM batches WHERE id = $1") + .bind(batch.batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(expired_at, None, "still a tombstone while it has live rows"); + + while manager.purge_orphaned_rows(100).await.unwrap() > 0 {} + let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM requests WHERE id = $1") + .bind(live_row) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(remaining, 0, "the purge erased the live row"); +} + +/// The candidate query runs before the week's registry row is locked. A +/// writer that share-locked the row (as batch creation and retry do) and +/// commits a running batch while the claim waits for that lock must still +/// stop the week from being fenced: the claim re-checks under the lock. +#[sqlx::test] +async fn a_batch_committed_while_the_claim_waits_holds_the_week(pool: PgPool) { + let week = monday(10); + ensure_week(&pool, week).await; + archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; + + let mut writer = pool.begin().await.unwrap(); + let state: String = sqlx::query_scalar( + "SELECT state FROM batch_archive_buckets WHERE week_start = $1 FOR SHARE", + ) + .bind(week) + .fetch_one(&mut *writer) + .await + .unwrap(); + assert_eq!(state, "active"); + let file_id: Uuid = sqlx::query_scalar( + "INSERT INTO files (name, size_bytes, size_finalized, status, purpose) \ + VALUES ('race-' || gen_random_uuid(), 0, TRUE, 'processed', 'batch') RETURNING id", + ) + .fetch_one(&mut *writer) + .await + .unwrap(); + sqlx::query( + "INSERT INTO batches (file_id, endpoint, completion_window, expires_at, created_at) \ + VALUES ($1, '/v1/x', '24h', NOW() + INTERVAL '24 hours', \ + ($2::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')", + ) + .bind(file_id) + .bind(week) + .execute(&mut *writer) + .await + .unwrap(); + + let manager = manager(&pool).await; + let retirement = tokio::spawn(async move { + manager + .retire_expired_batch_archive_partition(true, 30) + .await + }); + + // Poll until the claim is blocked on the registry row, well inside its + // 5 s lock timeout. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(4); + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity + WHERE datname = current_database() + AND pid <> pg_backend_pid() + AND wait_event_type = 'Lock' + AND query LIKE '%batch_archive_buckets%FOR UPDATE%' + )", + ) + .fetch_one(&pool) + .await + .unwrap(); + if waiting { + break; + } + assert!( + std::time::Instant::now() < deadline, + "the claim never waited for the week's registry row" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + writer.commit().await.unwrap(); + + let outcome = retirement.await.unwrap().unwrap(); + assert_eq!(outcome, RetainedResponseRetirementOutcome::NoCandidate); + let (state, journaled): (String, bool) = sqlx::query_as( + "SELECT bucket.state, EXISTS ( \ + SELECT 1 FROM retention_partition_retirements journal \ + WHERE journal.parent_table = 'batch_requests_archive' \ + AND journal.lower_bound = bucket.week_start) \ + FROM batch_archive_buckets bucket WHERE bucket.week_start = $1", + ) + .bind(week) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "active", "the week must not be fenced"); + assert!(!journaled, "no retirement may be journaled"); +} + #[sqlx::test] async fn recovery_completes_a_pending_journal_without_the_selection_flag(pool: PgPool) { let week = monday(10); @@ -349,37 +583,3 @@ async fn a_renamed_pending_child_fails_closed(pool: PgPool) { "the renamed relation must survive" ); } - -#[sqlx::test] -async fn stamping_never_regresses_an_earlier_expiry(pool: PgPool) { - let week = monday(10); - ensure_week(&pool, week).await; - let batch = archived_batch(&pool, week, Utc::now() - Duration::days(60)).await; - // Truncate to PostgreSQL's microsecond resolution so the round-tripped - // stamp compares exactly equal. - let earlier = chrono::SubsecRound::trunc_subsecs(Utc::now() - Duration::days(5), 6); - sqlx::query("UPDATE batches SET retention_expired_at = $2 WHERE id = $1") - .bind(batch.batch_id) - .bind(earlier) - .execute(&pool) - .await - .unwrap(); - - let outcome = manager(&pool) - .await - .retire_expired_batch_archive_partition(true, 30) - .await - .unwrap(); - assert_eq!(outcome, RetainedResponseRetirementOutcome::Retired); - let stamped: Option> = - sqlx::query_scalar("SELECT retention_expired_at FROM batches WHERE id = $1") - .bind(batch.batch_id) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!( - stamped, - Some(earlier), - "an existing stamp is never rewritten" - ); -} diff --git a/fusillade-arsenal/tests/retry_content_retention.rs b/fusillade-arsenal/tests/retry_content_retention.rs new file mode 100644 index 000000000..f12a576e0 --- /dev/null +++ b/fusillade-arsenal/tests/retry_content_retention.rs @@ -0,0 +1,383 @@ +//! Batch creation and retry refuse input or results whose content week is +//! being, or has been, retired, and retry refuses a batch near the end of +//! its content retention (`with_retry_age_limit`). + +use chrono::{Datelike, Duration, NaiveDate, Utc}; +use fusillade_arsenal::batch::{BatchId, BatchInput, FileId, RequestTemplateInput}; +use fusillade_arsenal::error::FusilladeError; +use fusillade_arsenal::request::RequestId; +use fusillade_arsenal::{PostgresRequestManager, PostgresStorageConfig, Storage, TestDbPools}; +use sqlx::PgPool; +use uuid::Uuid; + +async fn manager(pool: &PgPool, limit: Option) -> PostgresRequestManager { + PostgresRequestManager::new( + TestDbPools::new(pool.clone()).await.unwrap(), + PostgresStorageConfig::default(), + ) + .with_retry_age_limit(limit) +} + +async fn upload(manager: &PostgresRequestManager) -> FileId { + manager + .create_file( + "retention-input".to_string(), + None, + vec![RequestTemplateInput { + custom_id: Some("retention-req".to_string()), + endpoint: "https://example.invalid".to_string(), + method: "POST".to_string(), + path: "/v1/x".to_string(), + body: "{}".to_string(), + model: "test-model".to_string(), + api_key: "test-key".to_string(), + }], + ) + .await + .unwrap() +} + +fn batch_input(file_id: FileId) -> BatchInput { + BatchInput { + file_id, + endpoint: "/v1/chat/completions".to_string(), + completion_window: "24h".to_string(), + metadata: None, + created_by: None, + api_key_id: None, + api_key: None, + total_requests: None, + } +} + +/// A live batch whose only request has failed, ready to retry. +async fn failed_batch(pool: &PgPool, manager: &PostgresRequestManager) -> BatchId { + let file_id = upload(manager).await; + let batch = manager.create_batch(batch_input(file_id)).await.unwrap(); + sqlx::query( + "UPDATE requests SET state = 'failed', failed_at = NOW(), error = 'boom' \ + WHERE batch_id = $1", + ) + .bind(*batch.id) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "UPDATE batches SET failed_at = NOW(), counts_frozen_at = NOW(), failed_requests = 1 \ + WHERE id = $1", + ) + .bind(*batch.id) + .execute(pool) + .await + .unwrap(); + batch.id +} + +async fn request_ids(pool: &PgPool, batch_id: BatchId) -> Vec { + sqlx::query_scalar::<_, Uuid>("SELECT id FROM requests WHERE batch_id = $1") + .bind(*batch_id) + .fetch_all(pool) + .await + .unwrap() + .into_iter() + .map(RequestId) + .collect() +} + +/// Set the state of the template week a file's content was written to. +async fn set_input_week_state(pool: &PgPool, file_id: Uuid, state: &str) { + let updated = sqlx::query( + "UPDATE request_template_buckets SET state = $2 \ + WHERE week_start = (SELECT date_trunc('week', created_at AT TIME ZONE 'UTC')::date \ + FROM files WHERE id = $1)", + ) + .bind(file_id) + .bind(state) + .execute(pool) + .await + .unwrap() + .rows_affected(); + assert_eq!(updated, 1); +} + +async fn batch_file(pool: &PgPool, batch_id: BatchId) -> Uuid { + sqlx::query_scalar("SELECT file_id FROM batches WHERE id = $1") + .bind(*batch_id) + .fetch_one(pool) + .await + .unwrap() +} + +fn assert_validation(result: Result, context: &str) { + match result { + Err(FusilladeError::ValidationError(_)) => {} + other => panic!("{context}: expected a validation error, got {other:?}"), + } +} + +#[sqlx::test] +async fn batch_creation_requires_an_active_input_week(pool: PgPool) { + let manager = manager(&pool, None).await; + let file_id = upload(&manager).await; + + for state in ["retiring", "retired"] { + set_input_week_state(&pool, *file_id, state).await; + assert_validation( + manager.create_batch(batch_input(file_id)).await, + &format!("input week {state}"), + ); + } + let batches: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM batches WHERE file_id = $1") + .bind(*file_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(batches, 0, "a refused creation must leave no batch behind"); + + set_input_week_state(&pool, *file_id, "active").await; + manager.create_batch(batch_input(file_id)).await.unwrap(); +} + +#[sqlx::test] +async fn batch_retry_requires_an_active_input_week(pool: PgPool) { + let manager = manager(&pool, None).await; + let batch_id = failed_batch(&pool, &manager).await; + let file_id = batch_file(&pool, batch_id).await; + + for state in ["retiring", "retired"] { + set_input_week_state(&pool, file_id, state).await; + assert_validation( + manager.retry_failed_requests_for_batch(batch_id).await, + &format!("input week {state}"), + ); + } + + set_input_week_state(&pool, file_id, "active").await; + assert_eq!( + manager + .retry_failed_requests_for_batch(batch_id) + .await + .unwrap(), + 1 + ); +} + +#[sqlx::test] +async fn request_retry_requires_an_active_input_week(pool: PgPool) { + let manager = manager(&pool, None).await; + let batch_id = failed_batch(&pool, &manager).await; + let file_id = batch_file(&pool, batch_id).await; + let ids = request_ids(&pool, batch_id).await; + + // Request retry reports per request, so an unavailable source shows up + // as a validation error on each requested row. + for state in ["retiring", "retired"] { + set_input_week_state(&pool, file_id, state).await; + let results = manager + .retry_failed_requests(batch_id, ids.clone()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + assert_validation( + results.into_iter().next().unwrap(), + &format!("input week {state}"), + ); + } + let state: String = sqlx::query_scalar("SELECT state FROM requests WHERE batch_id = $1") + .bind(*batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "failed", "a refused retry must re-pend nothing"); + + set_input_week_state(&pool, file_id, "active").await; + let results = manager + .retry_failed_requests(batch_id, ids.clone()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + assert!(results[0].is_ok(), "{:?}", results[0]); +} + +async fn set_batch_created_at(pool: &PgPool, batch_id: BatchId, age: &str) { + sqlx::query(&format!( + "UPDATE batches SET created_at = NOW() - INTERVAL '{age}' WHERE id = $1" + )) + .bind(*batch_id) + .execute(pool) + .await + .unwrap(); +} + +/// With a 30-day retention period, retries are accepted until the batch is +/// 29 days old: one day before its archive week could drop. +#[sqlx::test] +async fn batch_retry_stops_a_day_before_results_retention_ends(pool: PgPool) { + let manager = manager(&pool, Some(Duration::days(30))).await; + + let inside = failed_batch(&pool, &manager).await; + set_batch_created_at(&pool, inside, "28 days 23 hours").await; + assert_eq!( + manager + .retry_failed_requests_for_batch(inside) + .await + .unwrap(), + 1 + ); + + let past = failed_batch(&pool, &manager).await; + set_batch_created_at(&pool, past, "29 days 1 hour").await; + assert_validation( + manager.retry_failed_requests_for_batch(past).await, + "batch past the cut-off", + ); + let ids = request_ids(&pool, past).await; + assert_validation( + manager.retry_failed_requests(past, ids).await, + "request retry past the cut-off", + ); + let state: String = sqlx::query_scalar("SELECT state FROM requests WHERE batch_id = $1") + .bind(*past) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "failed", "a refused retry must re-pend nothing"); +} + +/// Move a batch's input file to `age` ago, keeping its (possibly earlier) +/// template week registered and active so only the age decides. +async fn set_input_created_at(pool: &PgPool, batch_id: BatchId, age: &str) { + let file_id = batch_file(pool, batch_id).await; + let week: NaiveDate = sqlx::query_scalar(&format!( + "SELECT date_trunc('week', (NOW() - INTERVAL '{age}') AT TIME ZONE 'UTC')::date" + )) + .fetch_one(pool) + .await + .unwrap(); + sqlx::query("SELECT ensure_request_template_partition($1, NULL)") + .bind(week) + .execute(pool) + .await + .unwrap(); + sqlx::query(&format!( + "UPDATE files SET created_at = NOW() - INTERVAL '{age}' WHERE id = $1" + )) + .bind(file_id) + .execute(pool) + .await + .unwrap(); +} + +/// The same period also runs from the input file's upload: with 10 days, +/// retries are accepted until the input file is 9 days old, one day before +/// its template week could drop, even though the batch itself is new. +#[sqlx::test] +async fn batch_retry_stops_a_day_before_input_retention_ends(pool: PgPool) { + let manager = manager(&pool, Some(Duration::days(10))).await; + + let inside = failed_batch(&pool, &manager).await; + set_input_created_at(&pool, inside, "8 days 23 hours").await; + assert_eq!( + manager + .retry_failed_requests_for_batch(inside) + .await + .unwrap(), + 1 + ); + + let past = failed_batch(&pool, &manager).await; + set_input_created_at(&pool, past, "9 days 1 hour").await; + assert_validation( + manager.retry_failed_requests_for_batch(past).await, + "input past the cut-off", + ); +} + +fn archive_child(week_start: NaiveDate) -> String { + format!( + "batch_requests_archive_y{}w{:02}", + week_start.iso_week().year(), + week_start.iso_week().week() + ) +} + +/// Register a historical archive week (child, attach, registry row). +async fn ensure_archive_week(pool: &PgPool, week_start: NaiveDate) { + let child = archive_child(week_start); + sqlx::query(&format!( + "CREATE TABLE {child} (LIKE batch_requests_archive INCLUDING ALL)" + )) + .execute(pool) + .await + .unwrap(); + sqlx::query(&format!( + "ALTER TABLE batch_requests_archive ATTACH PARTITION {child} \ + FOR VALUES FROM ('{week_start}') TO ('{}')", + week_start + Duration::days(7) + )) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO batch_archive_buckets ( + week_start, partition_schema, partition_table, partition_oid + ) SELECT $1, current_schema(), $2, to_regclass($2)::oid", + ) + .bind(week_start) + .bind(&child) + .execute(pool) + .await + .unwrap(); +} + +/// Retrying an archived batch whose archive week is fenced or dropped would +/// move back nothing and reset the batch over completions that are gone. +#[sqlx::test] +async fn archived_batch_retry_requires_an_active_archive_week(pool: PgPool) { + let manager = manager(&pool, None).await; + let today = Utc::now().date_naive(); + let week = + today - Duration::days(today.weekday().num_days_from_monday().into()) - Duration::days(70); + ensure_archive_week(&pool, week).await; + + let batch_id = failed_batch(&pool, &manager).await; + sqlx::query("UPDATE batches SET location = 'archive', archive_bucket = $2 WHERE id = $1") + .bind(*batch_id) + .bind(week) + .execute(&pool) + .await + .unwrap(); + + for state in ["retiring", "retired"] { + sqlx::query("UPDATE batch_archive_buckets SET state = $2 WHERE week_start = $1") + .bind(week) + .bind(state) + .execute(&pool) + .await + .unwrap(); + assert_validation( + manager.retry_failed_requests_for_batch(batch_id).await, + &format!("archive week {state}"), + ); + let frozen: bool = + sqlx::query_scalar("SELECT counts_frozen_at IS NOT NULL FROM batches WHERE id = $1") + .bind(*batch_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(frozen, "a refused retry must not reset the batch"); + } + + sqlx::query("UPDATE batch_archive_buckets SET state = 'active' WHERE week_start = $1") + .bind(week) + .execute(&pool) + .await + .unwrap(); + assert_eq!( + manager + .retry_failed_requests_for_batch(batch_id) + .await + .unwrap(), + 1 + ); +} diff --git a/fusillade-arsenal/tests/template_generation_migration.rs b/fusillade-arsenal/tests/template_generation_migration.rs index d31e042cf..35d5aadde 100644 --- a/fusillade-arsenal/tests/template_generation_migration.rs +++ b/fusillade-arsenal/tests/template_generation_migration.rs @@ -342,85 +342,24 @@ async fn retirement_manager( .unwrap() } -async fn aged_file(pool: &PgPool, days_old: i64) -> Uuid { +/// A processed input file uploaded on the first day of `week`. +async fn file_uploaded_in(pool: &PgPool, week: chrono::NaiveDate) -> Uuid { sqlx::query_scalar( "INSERT INTO files (name, size_bytes, size_finalized, status, purpose, created_at) \ - VALUES ('aged-' || gen_random_uuid(), 0, TRUE, 'processed', 'batch', \ - NOW() - make_interval(days => $1::int)) RETURNING id", + VALUES ('upload-' || gen_random_uuid(), 0, TRUE, 'processed', 'batch', \ + ($1::timestamp AT TIME ZONE 'UTC') + INTERVAL '1 day') RETURNING id", ) - .bind(days_old as i32) + .bind(week) .fetch_one(pool) .await .unwrap() } +/// A template week drops on age alone; neither the file row nor any finished +/// batch holds it. Only a batch still running on input from that week does, +/// since its requests still read the templates. #[sqlx::test] -async fn file_content_expiry_tombstones_only_released_files(pool: PgPool) { - use fusillade_arsenal::manager::DaemonStorage; - - let released = aged_file(&pool, 400).await; - let pinned = aged_file(&pool, 400).await; - sqlx::query( - "INSERT INTO batches (file_id, endpoint, completion_window, expires_at, \ - location, archive_bucket, counts_frozen_at) \ - VALUES ($1, '/v1/x', '24h', NOW(), 'archive', \ - date_trunc('week', NOW() AT TIME ZONE 'UTC')::date, \ - NOW() - INTERVAL '10 days')", - ) - .bind(pinned) - .execute(&pool) - .await - .unwrap(); - let young = aged_file(&pool, 5).await; - - let manager = retirement_manager(&pool).await; - let step = manager.expire_file_content(365, 100, None).await.unwrap(); - assert_eq!(step.expired, 1, "only the released aged file may expire"); - assert_eq!(step.next, None, "one short slice covers every due file"); - - let rows: Vec<(Uuid, Option, bool, bool)> = sqlx::query_as( - "SELECT id, name, deleted_at IS NOT NULL, retention_expired_at IS NOT NULL \ - FROM files WHERE id = ANY($1) ORDER BY created_at", - ) - .bind(vec![released, pinned, young]) - .fetch_all(&pool) - .await - .unwrap(); - for (id, name, deleted, stamped) in rows { - assert!(name.is_some(), "file metadata (name) must be retained"); - if id == released { - assert!(deleted && stamped, "released aged file must be tombstoned"); - } else { - assert!(!deleted && !stamped, "pinned/young files must survive"); - } - } - - // Once the pinned file's batch content expires, the next pass releases it. - sqlx::query("UPDATE batches SET retention_expired_at = NOW() WHERE file_id = $1") - .bind(pinned) - .execute(&pool) - .await - .unwrap(); - assert_eq!( - manager - .expire_file_content(365, 100, None) - .await - .unwrap() - .expired, - 1 - ); - assert_eq!( - manager - .expire_file_content(365, 100, None) - .await - .unwrap() - .expired, - 0 - ); -} - -#[sqlx::test] -async fn template_week_retires_only_after_every_file_releases(pool: PgPool) { +async fn template_week_retires_on_age_unless_an_unfinished_batch_reads_it(pool: PgPool) { use fusillade_arsenal::manager::{DaemonStorage, RetainedResponseRetirementOutcome}; let week = monday(&pool, -60).await; @@ -429,31 +368,38 @@ async fn template_week_retires_only_after_every_file_releases(pool: PgPool) { .execute(&pool) .await .unwrap(); - let file = aged_file(&pool, 500).await; + let file = file_uploaded_in(&pool, week).await; let template_id = insert_g2_template(&pool, week, file).await; + let batch_id: Uuid = sqlx::query_scalar( + "INSERT INTO batches (file_id, endpoint, completion_window, expires_at, created_at) \ + VALUES ($1, '/v1/x', '24h', NOW() + INTERVAL '24 hours', \ + ($2::timestamp AT TIME ZONE 'UTC') + INTERVAL '1 day') RETURNING id", + ) + .bind(file) + .bind(week) + .fetch_one(&pool) + .await + .unwrap(); let manager = retirement_manager(&pool).await; let blocked = manager - .retire_expired_template_partition(true, 365) + .retire_expired_template_partition(true, 30) .await .unwrap(); assert_eq!( blocked, RetainedResponseRetirementOutcome::NoCandidate, - "a live file must pin its template week" + "an unfinished batch must pin its input's template week" ); - // File-content expiry releases the window (no batches reference it). - assert_eq!( - manager - .expire_file_content(365, 100, None) - .await - .unwrap() - .expired, - 1 - ); + // Finishing the batch releases the week; the file row is not consulted. + sqlx::query("UPDATE batches SET counts_frozen_at = NOW(), completed_at = NOW() WHERE id = $1") + .bind(batch_id) + .execute(&pool) + .await + .unwrap(); let retired = manager - .retire_expired_template_partition(true, 365) + .retire_expired_template_partition(true, 30) .await .unwrap(); assert_eq!(retired, RetainedResponseRetirementOutcome::Retired); @@ -476,6 +422,14 @@ async fn template_week_retires_only_after_every_file_releases(pool: PgPool) { child_exists, None, "the weekly template partition must drop" ); + let (deleted, status): (bool, String) = + sqlx::query_as("SELECT deleted_at IS NOT NULL, status FROM files WHERE id = $1") + .bind(file) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!deleted, "file metadata is never tombstoned by retirement"); + assert_eq!(status, "processed"); // Routes survive the drop and are removed by the bounded cleanup phase. let routed: i64 = @@ -495,6 +449,90 @@ async fn template_week_retires_only_after_every_file_releases(pool: PgPool) { ); } +/// The week is not yet past `week end + retention days`, so nothing holds it +/// but its age. +#[sqlx::test] +async fn a_young_template_week_is_not_a_candidate(pool: PgPool) { + use fusillade_arsenal::manager::{DaemonStorage, RetainedResponseRetirementOutcome}; + + let week = monday(&pool, -4).await; + sqlx::query("SELECT ensure_request_template_partition($1, NULL)") + .bind(week) + .execute(&pool) + .await + .unwrap(); + let file = file_uploaded_in(&pool, week).await; + insert_g2_template(&pool, week, file).await; + + let outcome = retirement_manager(&pool) + .await + .retire_expired_template_partition(true, 30) + .await + .unwrap(); + assert_eq!(outcome, RetainedResponseRetirementOutcome::NoCandidate); + let state: String = + sqlx::query_scalar("SELECT state FROM request_template_buckets WHERE week_start = $1") + .bind(week) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "active"); +} + +/// An unfinished batch holds only the week its own input was uploaded in. +#[sqlx::test] +async fn an_unfinished_batch_on_input_from_another_week_does_not_block(pool: PgPool) { + use fusillade_arsenal::manager::{DaemonStorage, RetainedResponseRetirementOutcome}; + + let week = monday(&pool, -60).await; + let other_week = monday(&pool, -59).await; + for monday in [week, other_week] { + sqlx::query("SELECT ensure_request_template_partition($1, NULL)") + .bind(monday) + .execute(&pool) + .await + .unwrap(); + } + let file = file_uploaded_in(&pool, week).await; + insert_g2_template(&pool, week, file).await; + let other_file = file_uploaded_in(&pool, other_week).await; + insert_g2_template(&pool, other_week, other_file).await; + // Still running, but on input uploaded the following week. + sqlx::query( + "INSERT INTO batches (file_id, endpoint, completion_window, expires_at, created_at) \ + VALUES ($1, '/v1/x', '24h', NOW() + INTERVAL '24 hours', \ + ($2::timestamp AT TIME ZONE 'UTC') + INTERVAL '1 day')", + ) + .bind(other_file) + .bind(other_week) + .execute(&pool) + .await + .unwrap(); + + let manager = retirement_manager(&pool).await; + let outcome = manager + .retire_expired_template_partition(true, 30) + .await + .unwrap(); + assert_eq!(outcome, RetainedResponseRetirementOutcome::Retired); + let states: Vec<(chrono::NaiveDate, String)> = sqlx::query_as( + "SELECT week_start, state FROM request_template_buckets \ + WHERE week_start = ANY($1) ORDER BY week_start", + ) + .bind(vec![week, other_week]) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + states, + vec![ + (week, "retired".to_string()), + (other_week, "active".to_string()) + ], + "only the week the running batch reads stays" + ); +} + #[sqlx::test] async fn an_unowned_template_row_fails_the_gate_closed(pool: PgPool) { use fusillade_arsenal::manager::{DaemonStorage, RetainedResponseRetirementOutcome}; diff --git a/fusillade-core/src/lib.rs b/fusillade-core/src/lib.rs index 036db1085..dcfc891b9 100644 --- a/fusillade-core/src/lib.rs +++ b/fusillade-core/src/lib.rs @@ -37,10 +37,9 @@ pub use daemon_record::{ }; pub use error::{FusilladeError, Result}; pub use manager::{ - DaemonStorage, FileExpiryCursor, FileExpiryStep, ModelFilter, ModelFilterState, - RetainedResponseArchiveOutcome, RetainedResponseMaintenanceError, - RetainedResponsePartitionRunway, RetainedResponseRetirementOutcome, RetentionPolicy, Storage, - TrailingDemandCount, + DaemonStorage, ModelFilter, ModelFilterState, RetainedResponseArchiveOutcome, + RetainedResponseMaintenanceError, RetainedResponsePartitionRunway, + RetainedResponseRetirementOutcome, RetentionPolicy, Storage, TrailingDemandCount, }; pub use request::*; diff --git a/fusillade-core/src/manager.rs b/fusillade-core/src/manager.rs index 94752ee2c..2e4cb8532 100644 --- a/fusillade-core/src/manager.rs +++ b/fusillade-core/src/manager.rs @@ -339,24 +339,6 @@ pub enum RetainedResponseRetirementOutcome { Retired, } -/// Position in the oldest-first walk over files due for content expiry: the -/// `(created_at, id)` of the last file a step examined. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct FileExpiryCursor { - pub created_at: DateTime, - pub id: uuid::Uuid, -} - -/// Content-free result of one bounded file-content expiry step. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct FileExpiryStep { - /// Files tombstoned by this step. - pub expired: u64, - /// Where the next step continues, or `None` once the walk has reached - /// the retention horizon. - pub next: Option, -} - #[cfg(test)] mod retention_policy_tests { use super::*; @@ -1870,10 +1852,10 @@ pub trait DaemonStorage: Send + Sync { Err(RetainedResponseMaintenanceError::Disabled.into_fusillade_error()) } - /// Retire one weekly batch-archive partition whose every batch is past - /// the finalization-anchored `retention_days`. Batch metadata rows are - /// stamped, never deleted. Storage backends opt in explicitly; the - /// default is disabled. + /// Retire one weekly batch-archive partition once the week is past + /// `retention_days`, measured from the week itself, and no batch created + /// in it is unfinished. Batch metadata rows are never touched. Storage + /// backends opt in explicitly; the default is disabled. async fn retire_expired_batch_archive_partition( &self, _select_new: bool, @@ -1882,9 +1864,10 @@ pub trait DaemonStorage: Send + Sync { Err(RetainedResponseMaintenanceError::Disabled.into_fusillade_error()) } - /// Retire one weekly generation-2 template partition once it is past the - /// horizon and no live file still owns templates in it. Storage backends - /// opt in explicitly; the default is disabled. + /// Retire one weekly generation-2 template partition once it is past + /// `retention_days`, measured from the week itself, and no unfinished + /// batch reads input uploaded in it. Storage backends opt in explicitly; + /// the default is disabled. async fn retire_expired_template_partition( &self, _select_new: bool, @@ -1893,23 +1876,6 @@ pub trait DaemonStorage: Send + Sync { Err(RetainedResponseMaintenanceError::Disabled.into_fusillade_error()) } - /// Tombstone aged-out input file metadata (content-free bounded update) - /// once every referencing batch's content is expired or erased. - /// - /// One call examines at most `scan_limit` due files after `after`, - /// oldest first, so its cost stays bounded however many due files live - /// batches still pin. Callers walk the due range by passing each step's - /// `next` back in. Storage backends opt in explicitly; the default is - /// disabled. - async fn expire_file_content( - &self, - _retention_days: i32, - _scan_limit: i64, - _after: Option, - ) -> Result { - Err(RetainedResponseMaintenanceError::Disabled.into_fusillade_error()) - } - /// Remove bounded stale template routes whose bucket has retired. Storage /// backends opt in explicitly; the default is disabled. async fn cleanup_retired_template_routes(&self, _limit: i64) -> Result { diff --git a/fusillade/src/daemon/config.rs b/fusillade/src/daemon/config.rs index ab27b16c8..b8fa8563b 100644 --- a/fusillade/src/daemon/config.rs +++ b/fusillade/src/daemon/config.rs @@ -156,9 +156,8 @@ pub struct RetentionMaintenanceConfig { retained_response_partitions_days_ahead: i32, retained_response_retirement_enabled: bool, batch_archive_retirement_enabled: bool, - batch_archive_retention_days: Option, template_retirement_enabled: bool, - template_retention_days: Option, + content_retention_days: Option, } impl Default for RetentionMaintenanceConfig { @@ -173,9 +172,8 @@ impl Default for RetentionMaintenanceConfig { retained_response_partitions_days_ahead: 7, retained_response_retirement_enabled: false, batch_archive_retirement_enabled: false, - batch_archive_retention_days: None, template_retirement_enabled: false, - template_retention_days: None, + content_retention_days: None, } } } @@ -237,11 +235,15 @@ impl RetentionMaintenanceConfig { self } - /// Set the finalization-anchored batch content retention period. There is - /// deliberately no default; enabling batch-archive retirement without an - /// explicit period fails validation. - pub fn with_batch_archive_retention_days(mut self, days: Option) -> Self { - self.batch_archive_retention_days = days; + /// Set the content retention period in days, shared by the weekly batch + /// archive (results) and template (input) families. A week drops once + /// `week_start + 7 days + days` has passed and no unfinished batch is + /// tied to it. Callers derive it from the longest file expiry they + /// allow, rounded up to whole days, so no file's `expires_at` can + /// outrun its content. There is deliberately no default; enabling + /// either family's retirement without a period fails validation. + pub fn with_content_retention_days(mut self, days: Option) -> Self { + self.content_retention_days = days; self } @@ -290,35 +292,33 @@ impl RetentionMaintenanceConfig { self.batch_archive_retirement_enabled } - /// The finalization-anchored batch content retention period, when set. - pub fn batch_archive_retention_days(&self) -> Option { - self.batch_archive_retention_days - } - - /// Enable or disable file-content expiry and selection of newly eligible - /// weekly template buckets. Durable unfinished retirements remain + /// Enable or disable selection of newly eligible weekly template + /// buckets. Durable unfinished retirements remain /// recoverable while this is false. pub fn with_template_retirement_enabled(mut self, enabled: bool) -> Self { self.template_retirement_enabled = enabled; self } - /// Set the creation-anchored input-content retention period. There is - /// deliberately no default; enabling template retirement without an - /// explicit period fails validation. - pub fn with_template_retention_days(mut self, days: Option) -> Self { - self.template_retention_days = days; - self - } - - /// Whether file-content expiry and template bucket selection are enabled. + /// Whether template bucket selection is enabled. pub fn template_retirement_enabled(&self) -> bool { self.template_retirement_enabled } - /// The creation-anchored input-content retention period, when set. - pub fn template_retention_days(&self) -> Option { - self.template_retention_days + /// The content retention period in days, when set. + pub fn content_retention_days(&self) -> Option { + self.content_retention_days + } + + /// The store's retry age limit: the content retention period while + /// either weekly family's retirement is enabled, else `None`, since + /// content that is never dropped needs no cut-off. See + /// [`fusillade_arsenal::PostgresRequestManager::with_retry_age_limit`]. + pub fn retry_age_limit(&self) -> Option { + (self.batch_archive_retirement_enabled || self.template_retirement_enabled) + .then_some(self.content_retention_days) + .flatten() + .map(|days| chrono::Duration::days(days.into())) } } diff --git a/fusillade/src/daemon/mod.rs b/fusillade/src/daemon/mod.rs index dd2509962..ccc469dc9 100644 --- a/fusillade/src/daemon/mod.rs +++ b/fusillade/src/daemon/mod.rs @@ -633,17 +633,6 @@ async fn run_retained_response_retirement_loop( } } -/// Due files examined by one file-content expiry statement. Each costs a batch -/// lookup, so this bounds how long the statement holds its write lock on -/// `files`. -const FILE_EXPIRY_SCAN_LIMIT: i64 = 10_000; - -/// Minimum time between complete file-content expiry walks. Files mostly -/// become eligible in bulk, when a batch-archive week retires and expires its -/// batches, so re-examining every pinned file on each retirement pass would -/// repeat the same work for little gain. -const FILE_EXPIRY_WALK_INTERVAL: Duration = Duration::from_secs(3_600); - async fn run_template_retirement_loop( storage: Arc, shutdown: tokio_util::sync::CancellationToken, @@ -654,49 +643,8 @@ async fn run_template_retirement_loop( ) where S: DaemonStorage + 'static, { - let mut last_expiry_walk: Option = None; loop { - // Phase 1: bounded file-content expiry unpins template windows. A - // failure here must not suppress retirement recovery below; a failed - // walk starts over on the next pass. - let expiry_due = - last_expiry_walk.is_none_or(|walked| walked.elapsed() >= FILE_EXPIRY_WALK_INTERVAL); - if select_new && expiry_due { - let mut after = None; - loop { - match maintenance_query( - &shutdown, - "file content expiry", - query_timeout, - storage.expire_file_content(retention_days, FILE_EXPIRY_SCAN_LIMIT, after), - ) - .await - { - Ok(Some(step)) => { - counter!("fusillade_file_content_expired_total").increment(step.expired); - match step.next { - Some(cursor) => after = Some(cursor), - None => { - last_expiry_walk = Some(tokio::time::Instant::now()); - break; - } - } - } - Ok(None) => return, - Err(error) => { - crate::background_error!( - "file_content_expiry_failed", - Error, - error_class = %maintenance_error_class(&error), - "Failed to expire aged file content" - ); - break; - } - } - } - } - - // Phase 2: at most one weekly template partition per pass. + // At most one weekly template partition per pass. let outcome = maintenance_query( &shutdown, "template partition retirement", @@ -1427,7 +1375,7 @@ fn validate_retirement_capability( } if config.batch_archive_retirement_enabled() && !config - .batch_archive_retention_days() + .content_retention_days() .is_some_and(|days| (1..=i32::MAX as u32).contains(&days)) { return Err(FusilladeError::ValidationError( @@ -1443,7 +1391,7 @@ fn validate_retirement_capability( } if config.template_retirement_enabled() && !config - .template_retention_days() + .content_retention_days() .is_some_and(|days| (1..=i32::MAX as u32).contains(&days)) { return Err(FusilladeError::ValidationError( @@ -3555,39 +3503,36 @@ where )); daemon_handles.push(("retained_response_retirement", retirement_handle)); - // Recovery of an unfinished batch-archive retirement must - // survive the flag or period being withdrawn, so the loop - // always runs; it selects NEW buckets only when both are set. - let batch_retention_days = self + // Recovery of an unfinished weekly retirement must survive + // the flag or period being withdrawn, so each loop always + // runs; it selects NEW buckets only when its flag and the + // shared content retention period are both set. + let content_retention_days = self .retention_maintenance - .batch_archive_retention_days() + .content_retention_days() .and_then(|days| i32::try_from(days).ok()); let batch_select_new = self .retention_maintenance .batch_archive_retirement_enabled() - && batch_retention_days.is_some(); + && content_retention_days.is_some(); let batch_handle = tokio::spawn(run_batch_archive_retirement_loop( self.storage.clone(), self.shutdown_token.clone(), query_timeout, batch_select_new, - batch_retention_days.unwrap_or(0), + content_retention_days.unwrap_or(0), Duration::from_secs(86_400), )); daemon_handles.push(("batch_archive_retirement", batch_handle)); - let template_retention_days = self - .retention_maintenance - .template_retention_days() - .and_then(|days| i32::try_from(days).ok()); let template_select_new = self.retention_maintenance.template_retirement_enabled() - && template_retention_days.is_some(); + && content_retention_days.is_some(); let template_handle = tokio::spawn(run_template_retirement_loop( self.storage.clone(), self.shutdown_token.clone(), query_timeout, template_select_new, - template_retention_days.unwrap_or(0), + content_retention_days.unwrap_or(0), Duration::from_secs(86_400), )); daemon_handles.push(("template_retirement", template_handle)); @@ -4097,8 +4042,6 @@ mod tests { retained_contiguous_ahead: AtomicUsize, retained_required: AtomicUsize, batch_candidates: AtomicUsize, - expiry_steps: AtomicUsize, - expiry_walk_steps: AtomicUsize, events: std::sync::Mutex>, } @@ -4132,8 +4075,6 @@ mod tests { retained_contiguous_ahead: AtomicUsize::new(7), retained_required: AtomicUsize::new(7), batch_candidates: AtomicUsize::new(0), - expiry_steps: AtomicUsize::new(0), - expiry_walk_steps: AtomicUsize::new(1), events: std::sync::Mutex::new(Vec::new()), } } @@ -4293,29 +4234,6 @@ mod tests { Ok(crate::RetainedResponseRetirementOutcome::NoCandidate) } - async fn expire_file_content( - &self, - _retention_days: i32, - _scan_limit: i64, - after: Option, - ) -> Result { - self.record("file_content_expiry"); - let step = self.expiry_steps.fetch_add(1, Ordering::SeqCst); - let walk_steps = self.expiry_walk_steps.load(Ordering::SeqCst).max(1); - // Each walk must resume from the previous step's cursor and - // restart from the oldest file once complete. - let position = step % walk_steps; - assert_eq!(after.is_none(), position == 0, "step {step}"); - let next = (position + 1 < walk_steps).then(|| crate::manager::FileExpiryCursor { - created_at: chrono::DateTime::UNIX_EPOCH, - id: uuid::Uuid::from_u128(position as u128 + 1), - }); - if let Some(cursor) = after { - assert_eq!(cursor.id, uuid::Uuid::from_u128(position as u128)); - } - Ok(crate::manager::FileExpiryStep { expired: 0, next }) - } - async fn cleanup_retired_template_routes(&self, _limit: i64) -> Result { self.record("template_route_cleanup"); Ok(0) @@ -4522,52 +4440,6 @@ mod tests { retained_handle.await.unwrap(); } - #[tokio::test(start_paused = true)] - async fn file_expiry_walks_in_steps_at_most_hourly() { - let storage = Arc::new(FakeMaintenanceStorage::default()); - storage.expiry_walk_steps.store(3, Ordering::SeqCst); - let shutdown = tokio_util::sync::CancellationToken::new(); - let handle = tokio::spawn(run_template_retirement_loop( - storage.clone(), - shutdown.clone(), - Duration::from_secs(1), - true, - 23, - Duration::from_secs(86_400), - )); - let count = |event: &str| { - storage - .events - .lock() - .unwrap() - .iter() - .filter(|recorded| **recorded == event) - .count() - }; - - // The first pass walks the whole due range, one bounded step at a time. - tokio::task::yield_now().await; - assert_eq!(count("file_content_expiry"), 3); - assert_eq!(count("template_retirement"), 1); - - // Retirement keeps its five-minute cadence without repeating the walk. - for pass in 1..12 { - tokio::time::advance(Duration::from_secs(300)).await; - tokio::task::yield_now().await; - assert_eq!(count("template_retirement"), 1 + pass); - assert_eq!(count("file_content_expiry"), 3); - } - - // An hour after the completed walk, the next pass walks again. - tokio::time::advance(Duration::from_secs(300)).await; - tokio::task::yield_now().await; - assert_eq!(count("template_retirement"), 13); - assert_eq!(count("file_content_expiry"), 6); - - shutdown.cancel(); - handle.await.unwrap(); - } - #[tokio::test(start_paused = true)] async fn recovery_and_route_cleanup_run_even_when_runway_creation_fails() { let storage = Arc::new(FakeMaintenanceStorage::default()); diff --git a/fusillade/src/manager/mod.rs b/fusillade/src/manager/mod.rs index f59883546..329306816 100644 --- a/fusillade/src/manager/mod.rs +++ b/fusillade/src/manager/mod.rs @@ -14,8 +14,8 @@ mod postgres; pub use postgres::PostgresDaemon; pub use fusillade_core::manager::{ - ArchiveOutcome, BackgroundClaimKind, DaemonStorage, FileExpiryCursor, FileExpiryStep, - ModelFilter, ModelFilterState, RetainedResponseArchiveCutoffs, RetainedResponseArchiveOutcome, + ArchiveOutcome, BackgroundClaimKind, DaemonStorage, ModelFilter, ModelFilterState, + RetainedResponseArchiveCutoffs, RetainedResponseArchiveOutcome, RetainedResponseMaintenanceError, RetainedResponsePartitionRunway, RetainedResponseRetirementOutcome, RetainedResponseWriteError, RetentionPolicy, Storage, TrailingDemandCount, diff --git a/fusillade/src/manager/postgres.rs b/fusillade/src/manager/postgres.rs index e184c7308..5cf3861a3 100644 --- a/fusillade/src/manager/postgres.rs +++ b/fusillade/src/manager/postgres.rs @@ -106,7 +106,8 @@ where config.claim_query_timeout_ms, )) .with_retained_response_fence_seconds(fence_seconds) - .with_retained_response_retention_bounds_seconds(retention_bounds), + .with_retained_response_retention_bounds_seconds(retention_bounds) + .with_retry_age_limit(retention.retry_age_limit()), ); Ok(Self::from_store(storage, config).with_retention_maintenance(retention)) }