From b700769fa7949f15ea5c79892920c2d683318e0a Mon Sep 17 00:00:00 2001 From: HazyForge Automation <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:05:02 -0500 Subject: [PATCH] fix(hosted): fence artifact abandonment cleanup --- .github/workflows/ci.yml | 16 + Cargo.lock | 2 + Cargo.toml | 2 +- docs/deployment/hosted-worker.md | 49 +- docs/product/RUNTIME_DATABASE.md | 9 + ...40000_hosted_delivery_terminal_privacy.sql | 138 +++ src/hosted_control.rs | 254 +++++ src/main.rs | 875 +++++++++++++++++- 8 files changed, 1323 insertions(+), 22 deletions(-) create mode 100644 migrations/20260812040000_hosted_delivery_terminal_privacy.sql diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 206ee4b..086bd71 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,6 +12,20 @@ permissions: jobs: rust: runs-on: ubuntu-latest + services: + postgres: + image: postgres:17-alpine + env: + POSTGRES_DB: call_scribe_test + POSTGRES_USER: call_scribe_test + POSTGRES_PASSWORD: call_scribe_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U call_scribe_test -d call_scribe_test" + --health-interval 2s + --health-timeout 5s + --health-retries 15 steps: - name: Check out repository uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 @@ -32,6 +46,8 @@ jobs: run: cargo clippy --all-features --all-targets --locked -- -D warnings - name: Run tests + env: + CALL_SCRIBE_TEST_DATABASE_URL: postgres://call_scribe_test:call_scribe_test@127.0.0.1:5432/call_scribe_test run: cargo test --all-features --locked - name: Audit dependencies diff --git a/Cargo.lock b/Cargo.lock index cece1bc..7507e91 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4046,6 +4046,7 @@ dependencies = [ "tokio-stream", "tracing", "url", + "uuid", "webpki-roots 0.26.11", ] @@ -4084,6 +4085,7 @@ dependencies = [ "stringprep", "thiserror 2.0.19", "tracing", + "uuid", "whoami", ] diff --git a/Cargo.toml b/Cargo.toml index a65e274..cdfb6cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,7 +35,7 @@ rustls = { version = "0.23", default-features = false, features = ["ring"], opti serenity = { version = "0.12", default-features = false, features = ["client", "gateway", "model", "native_tls_backend"], optional = true } songbird = { version = "0.6", default-features = false, features = ["driver", "gateway", "native", "serenity", "tungstenite", "receive"], optional = true } sqlx = { package = "sqlx-core", version = "0.8", default-features = false, features = ["_rt-tokio", "_tls-rustls-ring-webpki", "chrono", "json"] } -sqlx-postgres = { version = "0.8", features = ["chrono", "json"] } +sqlx-postgres = { version = "0.8", features = ["chrono", "json", "uuid"] } tokio = { version = "1.47", features = ["fs", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] } tokio-util = { version = "0.7", features = ["io"] } tower-http = { version = "0.6", features = ["fs"] } diff --git a/docs/deployment/hosted-worker.md b/docs/deployment/hosted-worker.md index fdc3f23..9117337 100644 --- a/docs/deployment/hosted-worker.md +++ b/docs/deployment/hosted-worker.md @@ -285,8 +285,53 @@ deletion. The provider `HEAD` must return the exact content length and the local WAV, and mark delivery final. It never stores either presigned URL or its headers. Failed uploads or verification retain the local WAV for fenced retry, and an unsafe delivery older than ten minutes backpressures later hosted -starts for that organization/guild. After twenty attempts the job is terminal -and still backpressures capture for operator reconciliation. +starts for that organization/guild. After twenty attempts the worker clears the +encrypted reservation lease and durably retries one idempotent abandonment +notification: + +```json +{ + "notificationId": "7754ab7e-c0f7-41a1-8373-9c213f27efb5", + "operationId": "7f634aa5-f8cb-41b9-8c92-b23e5a44ae53", + "generation": 7, + "reservationId": "c5596189-da7b-4437-bd21-4541f877ee8a", + "recordingId": "recording_01", + "artifactId": "artifact_01", + "artifactKind": "raw_audio_wav", + "segmentIndex": 1, + "objectKey": "objects/artifact_01.wav", + "destinationId": "84b890cf-88d2-4979-b099-08978d2faedf", + "destinationRevision": "c5b70242-8402-4120-a26d-122687ec9c06", + "provider": "customer_s3", + "allowedUploadHost": "bucket.s3.us-east-1.amazonaws.com", + "contentLength": 123456, + "sha256": "64-lowercase-hex-characters", + "reason": "retry_budget_exhausted" +} +``` + +The endpoint is +`POST /internal/v1/worker/artifact-deliveries/abandon`. `notificationId` is +generated once and reused for every retry. Operation ID, generation, and object +key are either all present or all absent. A null operation tuple also covers a +lost prepare response: the control plane resolves the exact historical +reservation/artifact/destination tuple and never reveals the internally +resolved operation in its echo. The control plane tuple-digest fences +notification replay and exactly echoes every identity, destination, size, and +hash field. A `cleanup_pending`/`tombstone_queued` response is only an +acknowledgement: the worker retains the local WAV and raw locator metadata and +continues backpressure while the customer-provider object may exist. Only a +`provider_absent` response with `no_operation`, `not_required`, or +`provider_absence_verified` authorizes local deletion and terminal metadata +minimization. Thus neither service may race a delayed upload with a time-only +deletion. + +After verified delivery/local purge or authoritative provider absence/local +purge, the worker atomically replaces the raw terminal outbox, runtime artifact, +receipt, provider host/key, filesystem path, and customer identifiers with +domain-separated SHA-256 audit evidence. A bounded sweep processes 25 legacy +terminal rows per worker pass. Self-hosted runtime artifacts and audit behavior +are unchanged. ## Rollout diff --git a/docs/product/RUNTIME_DATABASE.md b/docs/product/RUNTIME_DATABASE.md index c07c277..c1e2122 100644 --- a/docs/product/RUNTIME_DATABASE.md +++ b/docs/product/RUNTIME_DATABASE.md @@ -122,6 +122,15 @@ Stores events such as: This table provides an append-only operational history for self-hosted deployments. +Hosted raw-audio delivery uses a separate durable outbox while delivery or +provider cleanup is active. Once verified delivery has removed the local file, +or an idempotent control-plane abandonment handshake has authoritatively proved +the provider object absent and the local file is removed, raw outbox, artifact, +receipt, locator, and customer-identity fields are deleted. The hosted terminal +audit table retains only domain-separated SHA-256 identity evidence, artifact +shape, attempt counts, and completion timestamps. Cleanup-pending abandonment +never deletes or minimizes the remaining local copy. + ## Capture mode Set `CALL_SCRIBE_CAPTURE_MODE`: diff --git a/migrations/20260812040000_hosted_delivery_terminal_privacy.sql b/migrations/20260812040000_hosted_delivery_terminal_privacy.sql new file mode 100644 index 0000000..8775aea --- /dev/null +++ b/migrations/20260812040000_hosted_delivery_terminal_privacy.sql @@ -0,0 +1,138 @@ +-- Preserve an authoritative worker-abandonment handshake after upload retry +-- exhaustion. Local audio remains recoverable until the control plane proves +-- that the customer provider object is absent. +ALTER TABLE call_scribe_hosted_artifact_delivery_outbox + ADD COLUMN IF NOT EXISTS abandonment_notification_id UUID, + ADD COLUMN IF NOT EXISTS abandonment_notification_attempt_count BIGINT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS abandonment_notified_at TIMESTAMPTZ; + +-- Replace the prerelease lifecycle checks additively without depending on the +-- automatically generated constraint suffixes used by a particular Postgres +-- version. +DO $$ +DECLARE + constraint_name TEXT; +BEGIN + FOR constraint_name IN + SELECT constraints.conname + FROM pg_constraint constraints + WHERE conrelid = 'call_scribe_hosted_artifact_delivery_outbox'::regclass + AND contype = 'c' + AND ( + constraints.conname = 'call_scribe_hosted_artifact_delivery_status_v2_check' + OR constraints.conname = 'call_scribe_hosted_artifact_delivery_lifecycle_v2_check' + OR pg_get_constraintdef(constraints.oid) LIKE '%status = ANY%pending%in_progress%verified%delivered%failed%' + OR pg_get_constraintdef(constraints.oid) LIKE '%status = ''pending''%encrypted_lease_token%status = ''failed''%' + ) + LOOP + EXECUTE format( + 'ALTER TABLE call_scribe_hosted_artifact_delivery_outbox DROP CONSTRAINT %I', + constraint_name + ); + END LOOP; +END $$; + +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'abandonment_pending', + abandonment_notification_id = COALESCE(abandonment_notification_id, gen_random_uuid()), + updated_at = now() +WHERE status = 'failed'; + +ALTER TABLE call_scribe_hosted_artifact_delivery_outbox + ADD CONSTRAINT call_scribe_hosted_artifact_delivery_status_v2_check + CHECK (status IN ( + 'pending', 'in_progress', 'verified', 'delivered', + 'abandonment_pending', 'cleanup_pending', 'abandoned' + )), + ADD CONSTRAINT call_scribe_hosted_artifact_delivery_lifecycle_v2_check CHECK ( + (status = 'pending' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL + AND claim_token IS NULL + AND claim_until IS NULL + AND abandonment_notification_id IS NULL) + OR (status = 'in_progress' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL + AND claim_owner IS NOT NULL + AND claim_token IS NOT NULL + AND claim_until IS NOT NULL + AND abandonment_notification_id IS NULL) + OR (status = 'verified' + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND receipt IS NOT NULL + AND verified_at IS NOT NULL + AND abandonment_notification_id IS NULL + AND ((claim_owner IS NULL AND claim_token IS NULL AND claim_until IS NULL) + OR (claim_owner IS NOT NULL AND claim_token IS NOT NULL + AND claim_until IS NOT NULL))) + OR (status = 'delivered' + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND receipt IS NOT NULL + AND verified_at IS NOT NULL + AND local_deleted_at IS NOT NULL + AND completed_at IS NOT NULL + AND claim_token IS NULL + AND claim_until IS NULL + AND abandonment_notification_id IS NULL) + OR (status IN ('abandonment_pending', 'cleanup_pending') + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND abandonment_notification_id IS NOT NULL + AND ((claim_owner IS NULL AND claim_token IS NULL AND claim_until IS NULL) + OR (claim_owner IS NOT NULL AND claim_token IS NOT NULL + AND claim_until IS NOT NULL))) + OR (status = 'abandoned' + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND abandonment_notification_id IS NOT NULL + AND local_deleted_at IS NOT NULL + AND completed_at IS NOT NULL + AND claim_token IS NULL + AND claim_until IS NULL) + ); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_artifact_abandonment_claim_idx + ON call_scribe_hosted_artifact_delivery_outbox(next_attempt_at, claim_until, created_at) + WHERE status IN ('abandonment_pending', 'cleanup_pending'); + +-- Raw tenant, provider-locator, receipt, and filesystem fields live only in the +-- active outbox. Terminal rows are moved here and retain domain-separated +-- SHA-256 evidence only. +CREATE TABLE IF NOT EXISTS call_scribe_hosted_artifact_delivery_terminal_audit ( + id UUID PRIMARY KEY, + notification_id UUID UNIQUE, + terminal_state TEXT NOT NULL CHECK (terminal_state IN ('delivered', 'abandoned')), + organization_id_sha256 TEXT NOT NULL CHECK (organization_id_sha256 ~ '^[0-9a-f]{64}$'), + guild_id_sha256 TEXT NOT NULL CHECK (guild_id_sha256 ~ '^[0-9a-f]{64}$'), + recording_id_sha256 TEXT NOT NULL CHECK (recording_id_sha256 ~ '^[0-9a-f]{64}$'), + artifact_id_sha256 TEXT NOT NULL CHECK (artifact_id_sha256 ~ '^[0-9a-f]{64}$'), + reservation_id_sha256 TEXT NOT NULL CHECK (reservation_id_sha256 ~ '^[0-9a-f]{64}$'), + operation_id_sha256 TEXT CHECK (operation_id_sha256 IS NULL OR operation_id_sha256 ~ '^[0-9a-f]{64}$'), + object_key_sha256 TEXT CHECK (object_key_sha256 IS NULL OR object_key_sha256 ~ '^[0-9a-f]{64}$'), + destination_id_sha256 TEXT NOT NULL CHECK (destination_id_sha256 ~ '^[0-9a-f]{64}$'), + destination_revision_sha256 TEXT NOT NULL CHECK (destination_revision_sha256 ~ '^[0-9a-f]{64}$'), + allowed_upload_host_sha256 TEXT NOT NULL CHECK (allowed_upload_host_sha256 ~ '^[0-9a-f]{64}$'), + provider TEXT NOT NULL CHECK (provider IN ('customer_s3', 'customer_r2')), + artifact_kind TEXT NOT NULL CHECK (artifact_kind = 'raw_audio_wav'), + segment_index INTEGER NOT NULL CHECK (segment_index BETWEEN 1 AND 16), + content_length BIGINT NOT NULL CHECK (content_length > 0), + content_sha256 TEXT NOT NULL CHECK (content_sha256 ~ '^[0-9a-f]{64}$'), + receipt_sha256 TEXT CHECK (receipt_sha256 IS NULL OR receipt_sha256 ~ '^[0-9a-f]{64}$'), + delivery_attempt_count INTEGER NOT NULL CHECK (delivery_attempt_count >= 0), + abandonment_notification_attempt_count BIGINT NOT NULL CHECK (abandonment_notification_attempt_count >= 0), + provider_absence_verified_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ NOT NULL, + archived_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ( + (terminal_state = 'delivered' AND notification_id IS NULL + AND receipt_sha256 IS NOT NULL AND provider_absence_verified_at IS NULL) + OR (terminal_state = 'abandoned' AND notification_id IS NOT NULL + AND receipt_sha256 IS NULL AND provider_absence_verified_at IS NOT NULL) + ) +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_terminal_audit_artifact_idx + ON call_scribe_hosted_artifact_delivery_terminal_audit(artifact_id_sha256, archived_at); diff --git a/src/hosted_control.rs b/src/hosted_control.rs index 6540956..4f318ca 100644 --- a/src/hosted_control.rs +++ b/src/hosted_control.rs @@ -20,6 +20,7 @@ const CONFIG_PATH: &str = "internal/v1/worker/guild-configurations"; const COMMANDS_PATH: &str = "internal/v1/worker/commands"; const USAGE_RESERVATIONS_PATH: &str = "internal/v1/worker/usage/reservations"; const ARTIFACT_DELIVERY_PREPARE_PATH: &str = "internal/v1/worker/artifact-deliveries/prepare"; +const ARTIFACT_DELIVERY_ABANDON_PATH: &str = "internal/v1/worker/artifact-deliveries/abandon"; const MAX_SIGNED_UPLOAD_URL_BYTES: usize = 16 * 1024; const MAX_SIGNED_UPLOAD_HEADERS: usize = 16; const MAX_UPLOAD_EXPIRY: Duration = Duration::from_secs(10 * 60); @@ -406,6 +407,51 @@ pub(crate) struct ArtifactDeliveryReceipt { pub verified_at: DateTime, } +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ArtifactDeliveryAbandonRequest { + pub notification_id: String, + pub operation_id: Option, + pub generation: Option, + pub reservation_id: String, + pub recording_id: String, + pub artifact_id: String, + pub artifact_kind: String, + pub segment_index: u32, + pub object_key: Option, + pub destination_id: String, + pub destination_revision: String, + pub provider: String, + pub allowed_upload_host: String, + pub content_length: u64, + pub sha256: String, + pub reason: String, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct ArtifactDeliveryAbandonResponse { + pub notification_id: String, + pub operation_id: Option, + pub generation: Option, + pub reservation_id: String, + pub recording_id: String, + pub artifact_id: String, + pub artifact_kind: String, + pub segment_index: u32, + pub object_key: Option, + pub destination_id: String, + pub destination_revision: String, + pub provider: String, + pub allowed_upload_host: String, + pub content_length: u64, + pub sha256: String, + pub reason: String, + pub accepted_at: DateTime, + pub terminal_state: String, + pub cleanup_disposition: String, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct CommandLeaseRequest<'a> { @@ -946,6 +992,117 @@ impl HostedControlPlaneClient { proof.receipt, ))) } + + pub(crate) async fn abandon_artifact_delivery( + &self, + request: &ArtifactDeliveryAbandonRequest, + ) -> Result { + validate_artifact_abandon_request(request)?; + let response = self + .authenticated_request( + Method::POST, + self.base_url.join(ARTIFACT_DELIVERY_ABANDON_PATH)?, + ) + .json(request) + .send() + .await + .context("hosted artifact abandonment notification failed")?; + if !response.status().is_success() { + bail!( + "hosted artifact abandonment notification returned HTTP {}", + response.status() + ); + } + let response: ArtifactDeliveryAbandonResponse = decode_json_bounded(response, 32 * 1024) + .await + .context("hosted artifact abandonment response was invalid")?; + validate_artifact_abandon_response(&response, request)?; + Ok(response) + } +} + +fn validate_artifact_abandon_request(request: &ArtifactDeliveryAbandonRequest) -> Result<()> { + let operation_scope_valid = match ( + request.operation_id.as_deref(), + request.generation, + request.object_key.as_deref(), + ) { + (Some(operation_id), Some(generation), Some(object_key)) => { + valid_opaque_id(operation_id, 200) && generation > 0 && valid_object_key(object_key) + } + (None, None, None) => true, + _ => false, + }; + if uuid::Uuid::parse_str(&request.notification_id) + .ok() + .is_none_or(|id| id.to_string() != request.notification_id) + || !operation_scope_valid + || !valid_opaque_id(&request.reservation_id, 200) + || !valid_opaque_id(&request.recording_id, 200) + || !valid_opaque_id(&request.artifact_id, 200) + || request.artifact_kind != "raw_audio_wav" + || !(1..=16).contains(&request.segment_index) + || !valid_opaque_id(&request.destination_id, 200) + || !valid_opaque_id(&request.destination_revision, 200) + || !storage_destination_supported(&request.provider) + || !valid_pinned_provider_host(&request.provider, &request.allowed_upload_host) + || request.content_length == 0 + || decode_sha256_hex(&request.sha256).is_err() + || request.reason != "retry_budget_exhausted" + { + bail!("hosted artifact abandonment notification was invalid"); + } + Ok(()) +} + +fn validate_artifact_abandon_response( + response: &ArtifactDeliveryAbandonResponse, + request: &ArtifactDeliveryAbandonRequest, +) -> Result<()> { + let identity_matches = response.notification_id == request.notification_id + && response.operation_id == request.operation_id + && response.generation == request.generation + && response.reservation_id == request.reservation_id + && response.recording_id == request.recording_id + && response.artifact_id == request.artifact_id + && response.artifact_kind == request.artifact_kind + && response.segment_index == request.segment_index + && response.object_key == request.object_key + && response.destination_id == request.destination_id + && response.destination_revision == request.destination_revision + && response.provider == request.provider + && response.allowed_upload_host == request.allowed_upload_host + && response.content_length == request.content_length + && response.sha256 == request.sha256 + && response.reason == request.reason; + let outcome_valid = match request.operation_id.as_ref() { + None => matches!( + ( + response.terminal_state.as_str(), + response.cleanup_disposition.as_str(), + ), + ("cleanup_pending", "tombstone_queued") + | ("provider_absent", "no_operation") + | ("provider_absent", "not_required") + | ("provider_absent", "provider_absence_verified") + ), + Some(_) => matches!( + ( + response.terminal_state.as_str(), + response.cleanup_disposition.as_str(), + ), + ("cleanup_pending", "tombstone_queued") + | ("provider_absent", "not_required") + | ("provider_absent", "provider_absence_verified") + ), + }; + if !identity_matches + || !outcome_valid + || response.accepted_at > Utc::now() + chrono::Duration::minutes(1) + { + bail!("hosted artifact abandonment response did not bind the exact artifact"); + } + Ok(()) } fn validate_artifact_manifest_request(request: &ArtifactDeliveryPrepareRequest<'_>) -> Result<()> { @@ -1659,6 +1816,27 @@ mod tests { } } + fn artifact_abandonment_request() -> ArtifactDeliveryAbandonRequest { + ArtifactDeliveryAbandonRequest { + notification_id: "7754ab7e-c0f7-41a1-8373-9c213f27efb5".to_string(), + operation_id: Some("7f634aa5-f8cb-41b9-8c92-b23e5a44ae53".to_string()), + generation: Some(7), + reservation_id: "c5596189-da7b-4437-bd21-4541f877ee8a".to_string(), + recording_id: "rec_01".to_string(), + artifact_id: "art_01".to_string(), + artifact_kind: "raw_audio_wav".to_string(), + segment_index: 1, + object_key: Some("objects/art_01.wav".to_string()), + destination_id: "84b890cf-88d2-4979-b099-08978d2faedf".to_string(), + destination_revision: "c5b70242-8402-4120-a26d-122687ec9c06".to_string(), + provider: "customer_s3".to_string(), + allowed_upload_host: "bucket.s3.us-east-1.amazonaws.com".to_string(), + content_length: 44, + sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".to_string(), + reason: "retry_budget_exhausted".to_string(), + } + } + fn artifact_destination(provider: &str) -> HostedStorageDestination { let allowed_host = match provider { "customer_s3" => "bucket.s3.us-east-1.amazonaws.com", @@ -2531,6 +2709,82 @@ mod tests { ); } + #[test] + fn abandonment_contract_requires_complete_operation_identity_and_exact_echo() { + let request = artifact_abandonment_request(); + assert!(validate_artifact_abandon_request(&request).is_ok()); + let response = ArtifactDeliveryAbandonResponse { + notification_id: request.notification_id.clone(), + operation_id: request.operation_id.clone(), + generation: request.generation, + reservation_id: request.reservation_id.clone(), + recording_id: request.recording_id.clone(), + artifact_id: request.artifact_id.clone(), + artifact_kind: request.artifact_kind.clone(), + segment_index: request.segment_index, + object_key: request.object_key.clone(), + destination_id: request.destination_id.clone(), + destination_revision: request.destination_revision.clone(), + provider: request.provider.clone(), + allowed_upload_host: request.allowed_upload_host.clone(), + content_length: request.content_length, + sha256: request.sha256.clone(), + reason: request.reason.clone(), + accepted_at: Utc::now(), + terminal_state: "cleanup_pending".to_string(), + cleanup_disposition: "tombstone_queued".to_string(), + }; + assert!(validate_artifact_abandon_response(&response, &request).is_ok()); + let mut partial_scope = request.clone(); + partial_scope.object_key = None; + assert!(validate_artifact_abandon_request(&partial_scope).is_err()); + + let mut mismatched = response.clone(); + mismatched.artifact_id = "another-artifact".to_string(); + assert!(validate_artifact_abandon_response(&mismatched, &request).is_err()); + + let mut unsafe_outcome = response; + unsafe_outcome.terminal_state = "provider_absent".to_string(); + unsafe_outcome.cleanup_disposition = "tombstone_queued".to_string(); + assert!(validate_artifact_abandon_response(&unsafe_outcome, &request).is_err()); + } + + #[test] + fn no_operation_abandonment_is_explicit_and_still_exactly_bound() { + let mut request = artifact_abandonment_request(); + request.operation_id = None; + request.generation = None; + request.object_key = None; + assert!(validate_artifact_abandon_request(&request).is_ok()); + + let response = ArtifactDeliveryAbandonResponse { + notification_id: request.notification_id.clone(), + operation_id: None, + generation: None, + reservation_id: request.reservation_id.clone(), + recording_id: request.recording_id.clone(), + artifact_id: request.artifact_id.clone(), + artifact_kind: request.artifact_kind.clone(), + segment_index: request.segment_index, + object_key: None, + destination_id: request.destination_id.clone(), + destination_revision: request.destination_revision.clone(), + provider: request.provider.clone(), + allowed_upload_host: request.allowed_upload_host.clone(), + content_length: request.content_length, + sha256: request.sha256.clone(), + reason: request.reason.clone(), + accepted_at: Utc::now(), + terminal_state: "provider_absent".to_string(), + cleanup_disposition: "no_operation".to_string(), + }; + assert!(validate_artifact_abandon_response(&response, &request).is_ok()); + let mut response_loss_cleanup = response; + response_loss_cleanup.terminal_state = "cleanup_pending".to_string(); + response_loss_cleanup.cleanup_disposition = "tombstone_queued".to_string(); + assert!(validate_artifact_abandon_response(&response_loss_cleanup, &request).is_ok()); + } + #[test] fn only_customer_s3_and_r2_are_supported() { assert!(storage_destination_supported("customer_s3")); diff --git a/src/main.rs b/src/main.rs index b5b30d7..ccb3395 100644 --- a/src/main.rs +++ b/src/main.rs @@ -32,10 +32,10 @@ mod providers; #[cfg(feature = "discord")] use hosted_control::{ - ArtifactDeliveryOperationRef, ArtifactDeliveryPrepareRequest, ArtifactDeliveryReceipt, - ArtifactDeliveryVerification, HostedConfigurationStore, HostedControlPlaneClient, - HostedStorageDestination, RESERVATION_EXPIRY_MARGIN, RESERVATION_SETTLEMENT_GRACE, - UsageReservation, WorkerCommand, + ArtifactDeliveryAbandonRequest, ArtifactDeliveryOperationRef, ArtifactDeliveryPrepareRequest, + ArtifactDeliveryReceipt, ArtifactDeliveryVerification, HostedConfigurationStore, + HostedControlPlaneClient, HostedStorageDestination, RESERVATION_EXPIRY_MARGIN, + RESERVATION_SETTLEMENT_GRACE, UsageReservation, WorkerCommand, }; use providers::{ ElevenLabsSttConfig, ElevenLabsSttProvider, OpenAiSttConfig, OpenAiSttProvider, @@ -407,6 +407,7 @@ pub(crate) async fn migrate_runtime_schema(pool: &PgPool) -> Result<()> { include_str!("../migrations/20260812010000_hosted_worker_command_executions.sql"), include_str!("../migrations/20260812020000_hosted_capture_crash_recovery.sql"), include_str!("../migrations/20260812030000_hosted_artifact_delivery_outbox.sql"), + include_str!("../migrations/20260812040000_hosted_delivery_terminal_privacy.sql"), ] { sqlx::raw_sql::raw_sql(migration) .execute(pool) @@ -613,6 +614,116 @@ struct HostedArtifactDeliveryRow { operation_object_key: Option, } +#[cfg(feature = "discord")] +struct HostedArtifactAbandonmentRow { + artifact_id: String, + recording_id: String, + reservation_id: String, + segment_index: i32, + local_path: String, + content_length: i64, + sha256: String, + storage_provider: String, + storage_destination_id: String, + storage_destination_revision: String, + storage_allowed_host: String, + operation_id: Option, + operation_generation: Option, + operation_object_key: Option, + abandonment_notification_id: Uuid, +} + +#[cfg(feature = "discord")] +impl<'r> sqlx::from_row::FromRow<'r, sqlx_postgres::PgRow> for HostedArtifactAbandonmentRow { + fn from_row(row: &'r sqlx_postgres::PgRow) -> std::result::Result { + Ok(Self { + artifact_id: row.try_get("artifact_id")?, + recording_id: row.try_get("recording_id")?, + reservation_id: row.try_get("reservation_id")?, + segment_index: row.try_get("segment_index")?, + local_path: row.try_get("local_path")?, + content_length: row.try_get("content_length")?, + sha256: row.try_get("sha256")?, + storage_provider: row.try_get("storage_provider")?, + storage_destination_id: row.try_get("storage_destination_id")?, + storage_destination_revision: row.try_get("storage_destination_revision")?, + storage_allowed_host: row.try_get("storage_allowed_host")?, + operation_id: row.try_get("operation_id")?, + operation_generation: row.try_get("operation_generation")?, + operation_object_key: row.try_get("operation_object_key")?, + abandonment_notification_id: row.try_get("abandonment_notification_id")?, + }) + } +} + +#[cfg(feature = "discord")] +struct HostedArtifactTerminalRow { + artifact_id: String, + organization_id: String, + guild_id: String, + recording_id: String, + reservation_id: String, + artifact_kind: String, + segment_index: i32, + content_length: i64, + sha256: String, + storage_provider: String, + storage_destination_id: String, + storage_destination_revision: String, + storage_allowed_host: String, + operation_id: Option, + operation_object_key: Option, + receipt: Option, + attempt_count: i32, + abandonment_notification_id: Option, + abandonment_notification_attempt_count: i64, + completed_at: DateTime, +} + +#[cfg(feature = "discord")] +impl<'r> sqlx::from_row::FromRow<'r, sqlx_postgres::PgRow> for HostedArtifactTerminalRow { + fn from_row(row: &'r sqlx_postgres::PgRow) -> std::result::Result { + Ok(Self { + artifact_id: row.try_get("artifact_id")?, + organization_id: row.try_get("organization_id")?, + guild_id: row.try_get("guild_id")?, + recording_id: row.try_get("recording_id")?, + reservation_id: row.try_get("reservation_id")?, + artifact_kind: row.try_get("artifact_kind")?, + segment_index: row.try_get("segment_index")?, + content_length: row.try_get("content_length")?, + sha256: row.try_get("sha256")?, + storage_provider: row.try_get("storage_provider")?, + storage_destination_id: row.try_get("storage_destination_id")?, + storage_destination_revision: row.try_get("storage_destination_revision")?, + storage_allowed_host: row.try_get("storage_allowed_host")?, + operation_id: row.try_get("operation_id")?, + operation_object_key: row.try_get("operation_object_key")?, + receipt: row.try_get("receipt")?, + attempt_count: row.try_get("attempt_count")?, + abandonment_notification_id: row.try_get("abandonment_notification_id")?, + abandonment_notification_attempt_count: row + .try_get("abandonment_notification_attempt_count")?, + completed_at: row.try_get("completed_at")?, + }) + } +} + +#[cfg(feature = "discord")] +fn hash_hosted_terminal_field(label: &str, value: &[u8]) -> String { + let mut digest = Sha256::new(); + digest.update(b"call-scribe-hosted-terminal-audit-v1\0"); + digest.update(label.as_bytes()); + digest.update(b"\0"); + digest.update(value); + format!("{:x}", digest.finalize()) +} + +#[cfg(feature = "discord")] +fn hash_hosted_terminal_text(label: &str, value: &str) -> String { + hash_hosted_terminal_field(label, value.as_bytes()) +} + #[cfg(feature = "discord")] impl<'r> sqlx::from_row::FromRow<'r, sqlx_postgres::PgRow> for HostedArtifactDeliveryRow { fn from_row(row: &'r sqlx_postgres::PgRow) -> std::result::Result { @@ -2099,7 +2210,8 @@ SELECT EXISTS ( sqlx::query::query( r#" UPDATE call_scribe_hosted_artifact_delivery_outbox -SET status = 'failed', encrypted_lease_token = NULL, encryption_nonce = NULL, +SET status = 'abandonment_pending', encrypted_lease_token = NULL, encryption_nonce = NULL, + abandonment_notification_id = COALESCE(abandonment_notification_id, gen_random_uuid()), claim_owner = NULL, claim_token = NULL, claim_until = NULL, last_error = 'hosted artifact delivery exhausted its retry budget', updated_at = now() WHERE status IN ('pending', 'in_progress') @@ -2114,6 +2226,9 @@ WHERE status IN ('pending', 'in_progress') self.finish_verified_hosted_artifact_deletions(capture_dir) .await?; + self.archive_legacy_hosted_artifact_terminal_rows().await?; + self.notify_hosted_artifact_abandonments(client, capture_dir, claim_owner) + .await?; let claim_token = Uuid::new_v4().to_string(); let claim_seconds = i64::try_from(HOSTED_DELIVERY_CLAIM_TTL.as_secs()) @@ -2172,6 +2287,177 @@ RETURNING delivery.artifact_id, delivery.recording_id, delivery.reservation_id, .await } + async fn notify_hosted_artifact_abandonments( + &self, + client: &HostedControlPlaneClient, + capture_dir: &Path, + claim_owner: &str, + ) -> Result<()> { + let claim_token = Uuid::new_v4().to_string(); + let jobs: Vec = sqlx::query_as::query_as( + r#" +WITH candidates AS ( + SELECT artifact_id + FROM call_scribe_hosted_artifact_delivery_outbox + WHERE (status IN ('abandonment_pending', 'cleanup_pending') + AND (claim_until IS NULL OR claim_until <= now())) + AND next_attempt_at <= now() + ORDER BY next_attempt_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT 10 +) +UPDATE call_scribe_hosted_artifact_delivery_outbox AS delivery +SET claim_owner = $1, claim_token = $2, claim_until = now() + interval '1 minute', + abandonment_notification_attempt_count = abandonment_notification_attempt_count + 1, + updated_at = now() +FROM candidates +WHERE delivery.artifact_id = candidates.artifact_id +RETURNING delivery.artifact_id, delivery.recording_id, delivery.reservation_id, + delivery.segment_index, + delivery.local_path, delivery.content_length, delivery.sha256, + delivery.storage_provider, delivery.storage_destination_id, + delivery.storage_destination_revision, delivery.storage_allowed_host, + delivery.operation_id, delivery.operation_generation, + delivery.operation_object_key, delivery.abandonment_notification_id +"#, + ) + .bind(claim_owner) + .bind(&claim_token) + .fetch_all(&self.pool) + .await + .context("failed to claim hosted artifact abandonment notifications")?; + + for job in jobs { + let artifact_id = job.artifact_id.clone(); + if let Err(error) = self + .notify_claimed_hosted_artifact_abandonment(client, capture_dir, &claim_token, job) + .await + { + eprintln!("hosted artifact {artifact_id} abandonment remains pending: {error:#}"); + self.defer_claimed_hosted_artifact_abandonment(&artifact_id, &claim_token, &error) + .await?; + } + } + Ok(()) + } + + async fn notify_claimed_hosted_artifact_abandonment( + &self, + client: &HostedControlPlaneClient, + capture_dir: &Path, + claim_token: &str, + job: HostedArtifactAbandonmentRow, + ) -> Result<()> { + let operation_generation = job + .operation_generation + .map(u64::try_from) + .transpose() + .context("hosted abandonment operation generation was invalid")?; + let segment_index = u32::try_from(job.segment_index) + .context("hosted abandonment segment index was invalid")?; + let content_length = u64::try_from(job.content_length) + .context("hosted abandonment content length was invalid")?; + let request = ArtifactDeliveryAbandonRequest { + notification_id: job.abandonment_notification_id.to_string(), + operation_id: job.operation_id.clone(), + generation: operation_generation, + reservation_id: job.reservation_id.clone(), + recording_id: job.recording_id.clone(), + artifact_id: job.artifact_id.clone(), + artifact_kind: "raw_audio_wav".to_string(), + segment_index, + object_key: job.operation_object_key.clone(), + destination_id: job.storage_destination_id.clone(), + destination_revision: job.storage_destination_revision.clone(), + provider: job.storage_provider.clone(), + allowed_upload_host: job.storage_allowed_host.clone(), + content_length, + sha256: job.sha256.clone(), + reason: "retry_budget_exhausted".to_string(), + }; + let response = client.abandon_artifact_delivery(&request).await?; + if response.terminal_state == "cleanup_pending" { + let updated = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'cleanup_pending', abandonment_notified_at = $3, + claim_owner = NULL, claim_token = NULL, claim_until = NULL, + next_attempt_at = now() + interval '30 seconds', last_error = NULL, updated_at = now() +WHERE artifact_id = $1 AND status IN ('abandonment_pending', 'cleanup_pending') + AND claim_token = $2 AND abandonment_notification_id = $4 +"#, + ) + .bind(&job.artifact_id) + .bind(claim_token) + .bind(response.accepted_at) + .bind(job.abandonment_notification_id) + .execute(&self.pool) + .await + .context("failed to persist pending provider cleanup acknowledgement")?; + if updated.rows_affected() != 1 { + bail!("hosted artifact abandonment claim fence was lost"); + } + return Ok(()); + } + + let local_path = PathBuf::from(&job.local_path); + if !local_path.starts_with(capture_dir) { + bail!("hosted abandonment artifact escaped the configured capture directory"); + } + match tokio::fs::remove_file(&local_path).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error).context("failed to remove abandoned hosted artifact"), + } + let updated = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'abandoned', local_deleted_at = now(), completed_at = now(), + abandonment_notified_at = $3, claim_owner = NULL, claim_token = NULL, + claim_until = NULL, last_error = NULL, updated_at = now() +WHERE artifact_id = $1 AND status IN ('abandonment_pending', 'cleanup_pending') + AND claim_token = $2 AND abandonment_notification_id = $4 +"#, + ) + .bind(&job.artifact_id) + .bind(claim_token) + .bind(response.accepted_at) + .bind(job.abandonment_notification_id) + .execute(&self.pool) + .await + .context("failed to persist authoritative hosted artifact abandonment")?; + if updated.rows_affected() != 1 { + bail!("hosted artifact abandonment claim fence was lost before completion"); + } + self.archive_hosted_artifact_terminal_row(&job.artifact_id, "abandoned", Some(Utc::now())) + .await + } + + async fn defer_claimed_hosted_artifact_abandonment( + &self, + artifact_id: &str, + claim_token: &str, + error: &anyhow::Error, + ) -> Result<()> { + let message: String = format!("{error:#}").chars().take(500).collect(); + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET claim_owner = NULL, claim_token = NULL, claim_until = NULL, + next_attempt_at = now() + interval '30 seconds', last_error = $3, updated_at = now() +WHERE artifact_id = $1 AND status IN ('abandonment_pending', 'cleanup_pending') + AND claim_token = $2 +"#, + ) + .bind(artifact_id) + .bind(claim_token) + .bind(message) + .execute(&self.pool) + .await + .context("failed to defer hosted artifact abandonment notification")?; + Ok(()) + } + async fn deliver_claimed_hosted_artifact( &self, client: &HostedControlPlaneClient, @@ -2391,7 +2677,7 @@ WHERE artifact_id = $1 AND status = 'in_progress' AND claim_token = $2 WITH candidates AS ( SELECT artifact_id FROM call_scribe_hosted_artifact_delivery_outbox - WHERE status = 'verified' + WHERE status = 'verified' AND (claim_until IS NULL OR claim_until <= now()) ORDER BY verified_at FOR UPDATE SKIP LOCKED LIMIT 25 @@ -2431,7 +2717,7 @@ RETURNING delivery.artifact_id, delivery.local_path continue; } } - sqlx::query::query( + let completed = sqlx::query::query( r#" UPDATE call_scribe_hosted_artifact_delivery_outbox SET status = 'delivered', local_deleted_at = now(), completed_at = now(), @@ -2444,21 +2730,189 @@ WHERE artifact_id = $1 AND status = 'verified' AND claim_token = $2 .execute(&self.pool) .await .context("failed to mark hosted artifact delivery complete")?; - sqlx::query::query( - r#" -UPDATE call_scribe_artifacts -SET metadata = metadata || '{"hosted_delivery_state":"delivered","local_deleted":true}'::jsonb -WHERE id = $1 + if completed.rows_affected() != 1 { + bail!("hosted artifact deletion claim fence was lost before completion"); + } + self.archive_hosted_artifact_terminal_row(&artifact_id, "delivered", None) + .await?; + } + Ok(()) + } + + async fn archive_legacy_hosted_artifact_terminal_rows(&self) -> Result<()> { + let rows: Vec<(String, String)> = sqlx::query_as::query_as( + r#" +SELECT artifact_id, status +FROM call_scribe_hosted_artifact_delivery_outbox +WHERE status IN ('delivered', 'abandoned') +ORDER BY completed_at, artifact_id +LIMIT 25 "#, + ) + .fetch_all(&self.pool) + .await + .context("failed to select legacy hosted terminal delivery metadata")?; + for (artifact_id, terminal_state) in rows { + let provider_absence_verified_at = if terminal_state == "abandoned" { + Some(Utc::now()) + } else { + None + }; + self.archive_hosted_artifact_terminal_row( + &artifact_id, + &terminal_state, + provider_absence_verified_at, ) - .bind(&artifact_id) - .execute(&self.pool) - .await - .context("failed to update hosted artifact delivery metadata")?; + .await?; } Ok(()) } + async fn archive_hosted_artifact_terminal_row( + &self, + artifact_id: &str, + terminal_state: &str, + provider_absence_verified_at: Option>, + ) -> Result<()> { + if !matches!(terminal_state, "delivered" | "abandoned") { + bail!("hosted terminal archive state was invalid"); + } + let mut tx = self.pool.begin().await?; + let Some(row) = sqlx::query_as::query_as::<_, HostedArtifactTerminalRow>( + r#" +SELECT artifact_id, organization_id, guild_id, recording_id, reservation_id, + artifact_kind, segment_index, content_length, sha256, storage_provider, + storage_destination_id, storage_destination_revision, storage_allowed_host, + operation_id, operation_object_key, receipt, attempt_count, + abandonment_notification_id, abandonment_notification_attempt_count, + completed_at +FROM call_scribe_hosted_artifact_delivery_outbox +WHERE artifact_id = $1 AND status = $2 +FOR UPDATE +"#, + ) + .bind(artifact_id) + .bind(terminal_state) + .fetch_optional(&mut *tx) + .await? + else { + tx.rollback().await?; + return Ok(()); + }; + let receipt_sha256 = row + .receipt + .as_ref() + .map(serde_json::to_vec) + .transpose() + .context("failed to encode hosted delivery receipt for minimization")? + .map(|receipt| hash_hosted_terminal_field("receipt", &receipt)); + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_artifact_delivery_terminal_audit + (id, notification_id, terminal_state, organization_id_sha256, guild_id_sha256, + recording_id_sha256, artifact_id_sha256, reservation_id_sha256, + operation_id_sha256, object_key_sha256, destination_id_sha256, + destination_revision_sha256, allowed_upload_host_sha256, provider, + artifact_kind, segment_index, content_length, content_sha256, receipt_sha256, + delivery_attempt_count, abandonment_notification_attempt_count, + provider_absence_verified_at, completed_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, + $15, $16, $17, $18, $19, $20, $21, $22, $23) +ON CONFLICT (notification_id) WHERE notification_id IS NOT NULL DO NOTHING +"#, + ) + .bind(Uuid::new_v4()) + .bind(row.abandonment_notification_id) + .bind(terminal_state) + .bind(hash_hosted_terminal_text( + "organization_id", + &row.organization_id, + )) + .bind(hash_hosted_terminal_text("guild_id", &row.guild_id)) + .bind(hash_hosted_terminal_text("recording_id", &row.recording_id)) + .bind(hash_hosted_terminal_text("artifact_id", &row.artifact_id)) + .bind(hash_hosted_terminal_text( + "reservation_id", + &row.reservation_id, + )) + .bind( + row.operation_id + .as_deref() + .map(|value| hash_hosted_terminal_text("operation_id", value)), + ) + .bind( + row.operation_object_key + .as_deref() + .map(|value| hash_hosted_terminal_text("object_key", value)), + ) + .bind(hash_hosted_terminal_text( + "destination_id", + &row.storage_destination_id, + )) + .bind(hash_hosted_terminal_text( + "destination_revision", + &row.storage_destination_revision, + )) + .bind(hash_hosted_terminal_text( + "allowed_upload_host", + &row.storage_allowed_host, + )) + .bind(&row.storage_provider) + .bind(&row.artifact_kind) + .bind(row.segment_index) + .bind(row.content_length) + .bind(&row.sha256) + .bind(receipt_sha256) + .bind(row.attempt_count) + .bind(row.abandonment_notification_attempt_count) + .bind(provider_absence_verified_at) + .bind(row.completed_at) + .execute(&mut *tx) + .await + .context("failed to archive minimized hosted delivery evidence")?; + + let minimized_metadata = serde_json::json!({ + "terminal_state": terminal_state, + "artifact_id_sha256": hash_hosted_terminal_text("artifact_id", &row.artifact_id), + "recording_id_sha256": hash_hosted_terminal_text("recording_id", &row.recording_id), + "destination_id_sha256": hash_hosted_terminal_text( + "destination_id", + &row.storage_destination_id, + ), + }); + sqlx::query::query( + r#" +UPDATE call_scribe_audit_events +SET organization_id = NULL, session_id = NULL, guild_id = NULL, channel_id = NULL, + metadata = $3 +WHERE session_id = $1 + AND event_type = 'hosted_artifact_delivery_queued' + AND metadata ->> 'artifact_id' = $2 +"#, + ) + .bind(&row.recording_id) + .bind(&row.artifact_id) + .bind(minimized_metadata) + .execute(&mut *tx) + .await + .context("failed to minimize hosted delivery audit metadata")?; + sqlx::query::query( + "DELETE FROM call_scribe_hosted_artifact_delivery_outbox WHERE artifact_id = $1 AND status = $2", + ) + .bind(&row.artifact_id) + .bind(terminal_state) + .execute(&mut *tx) + .await + .context("failed to delete raw hosted terminal outbox metadata")?; + sqlx::query::query("DELETE FROM call_scribe_artifacts WHERE id = $1") + .bind(&row.artifact_id) + .execute(&mut *tx) + .await + .context("failed to delete raw hosted terminal artifact metadata")?; + tx.commit().await?; + Ok(()) + } + async fn release_verified_deletion_claim( &self, artifact_id: &str, @@ -6506,6 +6960,97 @@ VALUES ('reservation-invalid', NULL, NULL, 'recording-invalid', Ok(()) } + #[cfg(feature = "discord")] + #[tokio::test] + async fn terminal_privacy_migration_upgrades_failed_delivery_without_deleting_recovery_data() + -> Result<()> { + let Some((admin, pool, _second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping terminal migration proof"); + return Ok(()); + }; + for migration in [ + include_str!("../migrations/20260601183000_runtime_sessions_artifacts_audit.sql"), + include_str!("../migrations/20260601190000_drop_legacy_runtime_migrations.sql"), + include_str!("../migrations/20260803120000_multi_tenant_recordings_transcripts.sql"), + include_str!("../migrations/20260803140000_github_connections_and_issue_jobs.sql"), + include_str!("../migrations/20260803160000_browser_sessions.sql"), + include_str!("../migrations/20260812010000_hosted_worker_command_executions.sql"), + include_str!("../migrations/20260812020000_hosted_capture_crash_recovery.sql"), + include_str!("../migrations/20260812030000_hosted_artifact_delivery_outbox.sql"), + ] { + sqlx::raw_sql::raw_sql(migration).execute(&pool).await?; + } + sqlx::raw_sql::raw_sql( + r#" +INSERT INTO call_scribe_capture_sessions + (id, source, title, status, started_at, organization_id, mode, metadata) +VALUES ('recording-legacy-failed', 'discord', 'Legacy failed', 'completed', now(), + 'org_private_alpha', 'record_only', '{}'::jsonb); +INSERT INTO call_scribe_artifacts + (id, organization_id, session_id, kind, path, byte_size, metadata) +VALUES ('artifact-legacy-failed', 'org_private_alpha', 'recording-legacy-failed', + 'raw_audio_wav', '/captures/legacy-failed.wav', 44, '{}'::jsonb); +INSERT INTO call_scribe_hosted_artifact_delivery_outbox + (artifact_id, organization_id, guild_id, recording_id, reservation_id, + artifact_kind, segment_index, local_path, content_length, sha256, content_type, + storage_provider, storage_destination_id, storage_destination_revision, + storage_allowed_host, storage_object_key_prefix, transient_delete_policy, + status, attempt_count) +VALUES ('artifact-legacy-failed', 'org_private_alpha', '1', + 'recording-legacy-failed', 'reservation-legacy-failed', 'raw_audio_wav', 1, + '/captures/legacy-failed.wav', 44, + '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef', + 'audio/wav', 'customer_s3', 'destination-legacy', 'revision-legacy', + 'bucket.s3.us-east-1.amazonaws.com', 'objects/', + 'delete_after_verified_delivery', 'failed', 20) +"#, + ) + .execute(&pool) + .await?; + + let migration = + include_str!("../migrations/20260812040000_hosted_delivery_terminal_privacy.sql"); + sqlx::raw_sql::raw_sql(migration).execute(&pool).await?; + let first: (String, String, String, Option>) = sqlx::query_as::query_as( + r#" +SELECT status, abandonment_notification_id::text, local_path, local_deleted_at +FROM call_scribe_hosted_artifact_delivery_outbox +WHERE artifact_id = 'artifact-legacy-failed' +"#, + ) + .fetch_one(&pool) + .await?; + assert_eq!(first.0, "abandonment_pending"); + assert_eq!(first.2, "/captures/legacy-failed.wav"); + assert!(first.3.is_none()); + + sqlx::raw_sql::raw_sql(migration).execute(&pool).await?; + let replay: (String, String) = sqlx::query_as::query_as( + r#" +SELECT status, abandonment_notification_id::text +FROM call_scribe_hosted_artifact_delivery_outbox +WHERE artifact_id = 'artifact-legacy-failed' +"#, + ) + .fetch_one(&pool) + .await?; + assert_eq!(replay, (first.0, first.1)); + assert!( + sqlx::query::query( + "UPDATE call_scribe_hosted_artifact_delivery_outbox SET status = 'abandoned' WHERE artifact_id = 'artifact-legacy-failed'", + ) + .execute(&pool) + .await + .is_err(), + "migration must not permit terminal abandonment before local deletion" + ); + + pool.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + #[cfg(feature = "discord")] #[tokio::test] async fn two_instances_cannot_recover_a_live_heartbeating_capture() -> Result<()> { @@ -6938,12 +7483,47 @@ WHERE recording_id = 'recording-delivery' .await?; assert_eq!(verify_calls.load(Ordering::SeqCst), 1); assert_eq!(unexpected_calls.load(Ordering::SeqCst), 0); - let delivered: String = sqlx::query_scalar::query_scalar( - "SELECT status FROM call_scribe_hosted_artifact_delivery_outbox WHERE recording_id = 'recording-delivery'", + let raw_rows: i64 = sqlx::query_scalar::query_scalar( + "SELECT count(*) FROM call_scribe_hosted_artifact_delivery_outbox WHERE recording_id = 'recording-delivery'", + ) + .fetch_one(&first) + .await?; + assert_eq!(raw_rows, 0, "terminal raw locator metadata must be removed"); + let delivered: (String, String, String, String, Option) = sqlx::query_as::query_as( + r#" +SELECT terminal_state, artifact_id_sha256, recording_id_sha256, + destination_id_sha256, receipt_sha256 +FROM call_scribe_hosted_artifact_delivery_terminal_audit +WHERE artifact_id_sha256 = $1 +"#, + ) + .bind(hash_hosted_terminal_text( + "artifact_id", + &manifests[0].artifact_id, + )) + .fetch_one(&first) + .await?; + assert_eq!(delivered.0, "delivered"); + assert_eq!( + delivered.1, + hash_hosted_terminal_text("artifact_id", &manifests[0].artifact_id) + ); + assert_eq!( + delivered.2, + hash_hosted_terminal_text("recording_id", "recording-delivery") + ); + assert_eq!( + delivered.3, + hash_hosted_terminal_text("destination_id", &destination.destination_id) + ); + assert!(delivered.4.is_some()); + let raw_artifact_rows: i64 = sqlx::query_scalar::query_scalar( + "SELECT count(*) FROM call_scribe_artifacts WHERE id = $1", ) + .bind(&manifests[0].artifact_id) .fetch_one(&first) .await?; - assert_eq!(delivered, "delivered"); + assert_eq!(raw_artifact_rows, 0); assert!( !wav_path.exists(), "only the verified retry may delete local audio" @@ -6958,6 +7538,263 @@ WHERE recording_id = 'recording-delivery' Ok(()) } + #[cfg(feature = "discord")] + #[tokio::test] + async fn exhausted_delivery_waits_for_provider_absence_then_minimizes_terminal_metadata() + -> Result<()> { + let Some((admin, pool, _second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping abandonment proof"); + return Ok(()); + }; + migrate_runtime_schema(&pool).await?; + let store = SqlxRuntimeStore { + pool: pool.clone(), + organization_id: DEFAULT_ORGANIZATION_ID.to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + let test_dir = std::env::temp_dir().join(format!( + "call-scribe-abandonment-{}", + Uuid::new_v4().simple() + )); + std::fs::create_dir_all(&test_dir)?; + let wav_path = test_dir.join("exhausted.wav"); + std::fs::write(&wav_path, b"recoverable-audio")?; + let artifact_id = format!("artifact-{}", Uuid::new_v4().simple()); + let recording_id = format!("recording-{}", Uuid::new_v4().simple()); + let reservation_id = Uuid::new_v4().to_string(); + let destination_id = Uuid::new_v4().to_string(); + let destination_revision = Uuid::new_v4().to_string(); + let content_sha256 = hash_hosted_terminal_text("fixture-content", "recoverable-audio"); + sqlx::query::query( + r#" +INSERT INTO call_scribe_capture_sessions + (id, source, guild_id, channel_id, title, status, started_at, organization_id, + mode, metadata) +VALUES ($1, 'discord', '1', '2', 'Abandonment proof', 'completed', now(), + 'org_private_alpha', 'record_only', '{}'::jsonb) +"#, + ) + .bind(&recording_id) + .execute(&pool) + .await?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_artifacts + (id, organization_id, session_id, kind, path, byte_size, metadata) +VALUES ($1, 'org_private_alpha', $2, 'raw_audio_wav', $3, 17, $4) +"#, + ) + .bind(&artifact_id) + .bind(&recording_id) + .bind(wav_path.to_str().context("test path must be UTF-8")?) + .bind(json!({"artifact_id": artifact_id, "sha256": content_sha256})) + .execute(&pool) + .await?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_artifact_delivery_outbox + (artifact_id, organization_id, guild_id, recording_id, reservation_id, + encrypted_lease_token, encryption_nonce, artifact_kind, segment_index, + local_path, content_length, sha256, content_type, storage_provider, + storage_destination_id, storage_destination_revision, storage_allowed_host, + storage_object_key_prefix, transient_delete_policy, status, attempt_count, + next_attempt_at) +VALUES ($1, 'org_private_alpha', '1', $2, $3, decode('00','hex'), decode('00','hex'), + 'raw_audio_wav', 1, $4, 17, $5, 'audio/wav', 'customer_s3', $6, $7, + 'bucket.s3.us-east-1.amazonaws.com', 'objects/', + 'delete_after_verified_delivery', 'pending', $8, now()) +"#, + ) + .bind(&artifact_id) + .bind(&recording_id) + .bind(&reservation_id) + .bind(wav_path.to_str().context("test path must be UTF-8")?) + .bind(&content_sha256) + .bind(&destination_id) + .bind(&destination_revision) + .bind(HOSTED_DELIVERY_MAX_ATTEMPTS) + .execute(&pool) + .await?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_audit_events + (id, organization_id, session_id, event_type, actor_kind, guild_id, metadata) +VALUES ($1, 'org_private_alpha', $2, 'hosted_artifact_delivery_queued', 'system', + '1', $3) +"#, + ) + .bind(Uuid::new_v4().to_string()) + .bind(&recording_id) + .bind(json!({ + "artifact_id": artifact_id, + "recording_id": recording_id, + "object_key": "objects/exhausted.wav", + "allowed_upload_host": "bucket.s3.us-east-1.amazonaws.com", + })) + .execute(&pool) + .await?; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let calls = Arc::new(AtomicU64::new(0)); + let notification_ids = Arc::new(Mutex::new(Vec::::new())); + let handler_calls = calls.clone(); + let handler_notifications = notification_ids.clone(); + let app = axum::Router::new().route( + "/internal/v1/worker/artifact-deliveries/abandon", + axum::routing::post( + move |axum::Json(mut input): axum::Json| { + let handler_calls = handler_calls.clone(); + let handler_notifications = handler_notifications.clone(); + async move { + let notification_id = input["notificationId"] + .as_str() + .expect("notification id must be present") + .to_string(); + handler_notifications + .lock() + .expect("notification capture lock") + .push(notification_id); + assert!(input["operationId"].is_null()); + assert!(input["generation"].is_null()); + assert!(input["objectKey"].is_null()); + let call = handler_calls.fetch_add(1, Ordering::SeqCst); + let object = input.as_object_mut().expect("request must be an object"); + object.insert("acceptedAt".into(), json!(Utc::now())); + if call == 0 { + object.insert("terminalState".into(), json!("cleanup_pending")); + object.insert("cleanupDisposition".into(), json!("tombstone_queued")); + } else { + object.insert("terminalState".into(), json!("provider_absent")); + object.insert( + "cleanupDisposition".into(), + json!("provider_absence_verified"), + ); + } + axum::Json(input) + } + }, + ), + ); + let server = tokio::spawn(async move { axum::serve(listener, app).await }); + let client = HostedControlPlaneClient::new( + &format!("http://{address}"), + "test-token-with-at-least-thirty-two-bytes".to_string(), + "abandonment-worker".to_string(), + "test-outbox-key-with-at-least-thirty-two-bytes".to_string(), + )?; + + store + .retry_hosted_artifact_delivery_outbox(&client, &test_dir, "abandonment-worker") + .await?; + assert!( + wav_path.exists(), + "cleanup-pending must preserve local audio" + ); + let pending: (String, String, Option>, Option>) = sqlx::query_as::query_as( + r#" +SELECT status, abandonment_notification_id::text, encrypted_lease_token, + encryption_nonce +FROM call_scribe_hosted_artifact_delivery_outbox WHERE artifact_id = $1 +"#, + ) + .bind(&artifact_id) + .fetch_one(&pool) + .await?; + assert_eq!(pending.0, "cleanup_pending"); + assert!(pending.2.is_none() && pending.3.is_none()); + assert!( + store + .has_unsafe_hosted_deliveries("org_private_alpha", "1", Duration::from_secs(0),) + .await?, + "provider cleanup must keep hosted capture backpressured" + ); + + sqlx::query::query( + "UPDATE call_scribe_hosted_artifact_delivery_outbox SET next_attempt_at = now() WHERE artifact_id = $1", + ) + .bind(&artifact_id) + .execute(&pool) + .await?; + store + .retry_hosted_artifact_delivery_outbox(&client, &test_dir, "abandonment-worker") + .await?; + assert!( + !wav_path.exists(), + "provider-absence proof permits local purge" + ); + assert_eq!( + sqlx::query_scalar::query_scalar::<_, i64>( + "SELECT count(*) FROM call_scribe_hosted_artifact_delivery_outbox WHERE artifact_id = $1", + ) + .bind(&artifact_id) + .fetch_one(&pool) + .await?, + 0 + ); + let audit: (String, String, Option>, Option) = + sqlx::query_as::query_as( + r#" +SELECT terminal_state, artifact_id_sha256, provider_absence_verified_at, + receipt_sha256 +FROM call_scribe_hosted_artifact_delivery_terminal_audit +WHERE notification_id::text = $1 +"#, + ) + .bind(&pending.1) + .fetch_one(&pool) + .await?; + assert_eq!(audit.0, "abandoned"); + assert_eq!( + audit.1, + hash_hosted_terminal_text("artifact_id", &artifact_id) + ); + assert!(audit.2.is_some()); + assert!(audit.3.is_none()); + let minimized_event: ( + Option, + Option, + Option, + serde_json::Value, + ) = sqlx::query_as::query_as( + r#" +SELECT organization_id, session_id, guild_id, metadata +FROM call_scribe_audit_events +WHERE event_type = 'hosted_artifact_delivery_queued' + AND metadata ->> 'artifact_id_sha256' = $1 +"#, + ) + .bind(hash_hosted_terminal_text("artifact_id", &artifact_id)) + .fetch_one(&pool) + .await?; + assert_eq!( + (minimized_event.0, minimized_event.1, minimized_event.2), + (None, None, None) + ); + let event_text = minimized_event.3.to_string(); + for sensitive in [ + artifact_id.as_str(), + recording_id.as_str(), + "objects/exhausted.wav", + "bucket.s3.us-east-1.amazonaws.com", + ] { + assert!(!event_text.contains(sensitive)); + } + let captured = notification_ids + .lock() + .expect("notification capture lock") + .clone(); + assert_eq!(captured, vec![pending.1.clone(), pending.1]); + assert_eq!(calls.load(Ordering::SeqCst), 2); + server.abort(); + + std::fs::remove_dir_all(&test_dir)?; + pool.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + #[cfg(feature = "discord")] #[tokio::test] async fn zero_duration_hosted_capture_closes_without_delivery_or_backpressure() -> Result<()> {