diff --git a/Cargo.lock b/Cargo.lock index 0b14a85..cece1bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -341,6 +341,7 @@ dependencies = [ "sqlx-core", "sqlx-postgres", "tokio", + "tokio-util", "tower-http", "tracing", "urlencoding", diff --git a/Cargo.toml b/Cargo.toml index 41611ee..a65e274 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ hound = { version = "3.5", optional = true } jsonwebtoken = { version = "9", default-features = false, features = ["use_pem"] } mime_guess = "2.0" rand = { version = "0.9", features = ["std", "std_rng"] } -reqwest = { version = "0.13", features = ["form", "json", "multipart"] } +reqwest = { version = "0.13", features = ["form", "json", "multipart", "stream"] } schemars = "1.0" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" @@ -37,6 +37,7 @@ songbird = { version = "0.6", default-features = false, features = ["driver", "g 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"] } 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"] } tracing = "0.1" urlencoding = "2" diff --git a/docs/deployment/hosted-worker.md b/docs/deployment/hosted-worker.md index a26a070..fdc3f23 100644 --- a/docs/deployment/hosted-worker.md +++ b/docs/deployment/hosted-worker.md @@ -27,6 +27,9 @@ conditions: - a notice channel is configured; - a retention period from 1 through 365 days is configured; - the monthly recording cap exists and has remaining seconds; and +- a versioned opaque `customer_s3` or `customer_r2` destination is pinned with + its exact normalized account/bucket host and + `delete_after_verified_delivery`; and - at least one human participant is currently in the requested channel. Missing configuration, a failed first fetch, stale configuration, malformed @@ -43,12 +46,13 @@ All requests use `Authorization: Bearer ` and include the server must bind the token to the configured worker ID and should keep these internal routes on private ingress or require workload mTLS. -This incremental adapter does not yet implement a hosted storage destination. -Every current provider (`customer_s3`, `customer_r2`, `google_drive`, and -`managed_transient`) therefore fails closed at the worker, even when the -control plane reports it as ready. Hosted recording remains disabled until a -provider adapter uploads, verifies, and deletes or retains transient local data -according to policy. Self-hosted local/PVC capture is unchanged. +Hosted capture remains raw-audio record-only. The worker can deliver finalized +WAV segments to `customer_s3` or `customer_r2` through short-lived, +exact-object grants issued by the private control plane. It never receives or +stores the customer's reusable storage credentials. Unsupported providers, +missing destination ID/revision, and any transient policy except +`delete_after_verified_delivery` fail closed. Self-hosted local/PVC capture and +transcript content behavior are unchanged. `GET /internal/v1/worker/guild-configurations` @@ -71,6 +75,11 @@ according to policy. Self-hosted local/PVC capture is unchanged. "remainingRecordingSeconds": 35900, "storageProvider": "customer_s3", "storageDestinationLabel": "Customer archive", + "storageDestinationId": "dst_01", + "storageDestinationRevision": "dstrev_04", + "storageAllowedHost": "customer-audio.s3.us-east-1.amazonaws.com", + "storageObjectKeyPrefix": "tenant-opaque-prefix/", + "transientDeletePolicy": "delete_after_verified_delivery", "ready": true, "blockedReasons": [], "desiredRecordingGeneration": 42 @@ -200,13 +209,84 @@ capture releases the reservation. Invalid or missing audio stays pending only inside the bounded settlement window and becomes a visible terminal operator-reconciliation item afterward. Recovery settles an abandoned capture; it never resumes capture or renews recording authority after a process crash. +Header-only zero-duration WAVs never enter artifact delivery or delivery +backpressure. Their paths are atomically placed in a separate crash-safe local +cleanup outbox with session-stop evidence; cleanup revalidates that each target +is a regular zero-sample WAV inside `captureDir` before removing it. Recovery requires durable shared access to both Postgres and `captureDir` across pod replacement. A hard crash can discard audio written after the most recent WAV checkpoint, so recovery intentionally may undercount by less than the five-second checkpoint interval rather than bill unpersisted wall-clock -time. All hosted storage providers remain disabled in this branch, so no -reservation or audio capture can enter this path yet. +time. + +At finalization, the worker hashes each regular, non-symlink WAV and atomically +persists the runtime artifact, immutable manifest, encrypted reservation lease, +delivery job, usage job, stop evidence, and removal of capture recovery. It +then calls `POST /internal/v1/worker/artifact-deliveries/prepare` with only the +reservation authority and exact object manifest: + +```json +{ + "reservationId": "reservation_01", + "leaseToken": "opaque-reservation-lease", + "recordingId": "recording_01", + "artifactId": "artifact_01", + "artifactKind": "raw_audio_wav", + "segmentIndex": 1, + "contentLength": 123456, + "sha256": "64-lowercase-hex-characters", + "contentType": "audio/wav" +} +``` + +Hosted capture disables per-SSRC source-stem creation, so the mixed WAV segment +list is the exhaustive sensitive-audio manifest. Self-hosted source-stem +behavior remains available and unchanged. + +The control plane derives organization, guild, destination, and object key from +the reservation and responds with `operationId`, positive `generation`, exact +echoes of `recordingId`, `artifactId`, `artifactKind`, and `segmentIndex`, an +opaque safe `objectKey`, pinned destination ID/revision/provider, the exact +configuration-pinned `allowedUploadHost`, and a short-lived HTTPS `PUT` URL +plus signed headers. The URL must use effective HTTPS port 443, and its path +must be exactly `/`; an explicit alternate port is a different +origin and is rejected. The worker accepts no redirects and does not let the +response choose the value used to approve its own host. Only AWS S3 hosts +ending in `amazonaws.com` with an `s3` DNS label or the exact +`.r2.cloudflarestorage.com` host shape may be pinned by configuration. +S3 uses bucket-qualified virtual-host style. Because R2 uses an +account-qualified host with path-style buckets, the independently pinned +`storageObjectKeyPrefix` begins with the destination bucket and tenant prefix. +Every returned object key must remain under the pinned prefix for both +providers. + +The worker parses the actual SigV4 query, not only the detached JSON expiry. It +requires `AWS4-HMAC-SHA256`, a valid `s3/aws4_request` credential scope, +`X-Amz-Date`, `X-Amz-Expires` of at most ten minutes, a 64-character lowercase +signature, and a sorted `X-Amz-SignedHeaders` containing `host` plus every +outgoing header. The SigV4 expiry must exactly equal `expiresAt` and remain +inside the live policy window. Signed upload headers bind content type, length, +and `x-amz-meta-callscribe-sha256`. AWS S3 additionally requires +`x-amz-checksum-sha256` containing the base64 raw digest; R2 rejects that +unsupported full-object checksum header. Uploads time out within eight minutes +under a fifteen-minute database claim. + +After upload, the worker posts the generation and exact recording, artifact, +kind, segment, and object-key identity to +`/internal/v1/worker/artifact-deliveries//verify`. The response +contains an audit receipt carrying the same full identity and a separate +short-lived presigned provider `HEAD` capability. The worker validates that +capability with the same exact host, object path, and SigV4 rules, then calls +the provider directly. A matching control-plane receipt alone never authorizes +deletion. The provider `HEAD` must return the exact content length and +`x-amz-meta-callscribe-sha256`; AWS S3 must also return its provider-verified +`x-amz-checksum-sha256`. Only then does the worker persist verification, delete +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. ## Rollout diff --git a/migrations/20260812030000_hosted_artifact_delivery_outbox.sql b/migrations/20260812030000_hosted_artifact_delivery_outbox.sql new file mode 100644 index 0000000..56e8c23 --- /dev/null +++ b/migrations/20260812030000_hosted_artifact_delivery_outbox.sql @@ -0,0 +1,156 @@ +-- Pin the customer-owned storage destination and transient deletion policy to +-- the crash-recovery row. A later configuration refresh must never redirect an +-- already-recorded artifact to a different destination. +ALTER TABLE call_scribe_hosted_capture_recovery + ADD COLUMN IF NOT EXISTS organization_id TEXT, + ADD COLUMN IF NOT EXISTS guild_id TEXT, + ADD COLUMN IF NOT EXISTS storage_provider TEXT, + ADD COLUMN IF NOT EXISTS storage_destination_id TEXT, + ADD COLUMN IF NOT EXISTS storage_destination_revision TEXT, + ADD COLUMN IF NOT EXISTS storage_allowed_host TEXT, + ADD COLUMN IF NOT EXISTS storage_object_key_prefix TEXT, + ADD COLUMN IF NOT EXISTS transient_delete_policy TEXT; + +-- This is only an upgrade guard for rows created by a prerelease worker which +-- could not enable hosted storage. The unsupported marker remains fail-closed. +UPDATE call_scribe_hosted_capture_recovery +SET organization_id = COALESCE(organization_id, 'org_private_alpha'), + guild_id = COALESCE(guild_id, 'unsupported_legacy'), + storage_provider = COALESCE(storage_provider, 'unsupported_legacy'), + storage_destination_id = COALESCE(storage_destination_id, 'unsupported_legacy'), + storage_destination_revision = COALESCE(storage_destination_revision, 'unsupported_legacy'), + storage_allowed_host = COALESCE(storage_allowed_host, 'unsupported.invalid'), + storage_object_key_prefix = COALESCE(storage_object_key_prefix, 'unsupported/'), + transient_delete_policy = COALESCE(transient_delete_policy, 'retain_pending_operator') +WHERE organization_id IS NULL + OR guild_id IS NULL + OR storage_provider IS NULL + OR storage_destination_id IS NULL + OR storage_destination_revision IS NULL + OR storage_allowed_host IS NULL + OR storage_object_key_prefix IS NULL + OR transient_delete_policy IS NULL; + +ALTER TABLE call_scribe_hosted_capture_recovery + ALTER COLUMN organization_id SET NOT NULL, + ALTER COLUMN guild_id SET NOT NULL, + ALTER COLUMN storage_provider SET NOT NULL, + ALTER COLUMN storage_destination_id SET NOT NULL, + ALTER COLUMN storage_destination_revision SET NOT NULL, + ALTER COLUMN storage_allowed_host SET NOT NULL, + ALTER COLUMN storage_object_key_prefix SET NOT NULL, + ALTER COLUMN transient_delete_policy SET NOT NULL; + +CREATE TABLE IF NOT EXISTS call_scribe_hosted_artifact_delivery_outbox ( + artifact_id TEXT PRIMARY KEY + REFERENCES call_scribe_artifacts(id) ON DELETE RESTRICT, + organization_id TEXT NOT NULL, + guild_id TEXT NOT NULL, + recording_id TEXT NOT NULL + REFERENCES call_scribe_capture_sessions(id) ON DELETE RESTRICT, + reservation_id TEXT NOT NULL, + encrypted_lease_token BYTEA, + encryption_nonce BYTEA, + artifact_kind TEXT NOT NULL CHECK (artifact_kind = 'raw_audio_wav'), + segment_index INTEGER NOT NULL CHECK (segment_index BETWEEN 1 AND 16), + local_path TEXT NOT NULL, + content_length BIGINT NOT NULL CHECK (content_length > 0), + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + content_type TEXT NOT NULL CHECK (content_type = 'audio/wav'), + storage_provider TEXT NOT NULL CHECK (storage_provider IN ('customer_s3', 'customer_r2')), + storage_destination_id TEXT NOT NULL, + storage_destination_revision TEXT NOT NULL, + storage_allowed_host TEXT NOT NULL, + storage_object_key_prefix TEXT NOT NULL, + transient_delete_policy TEXT NOT NULL + CHECK (transient_delete_policy = 'delete_after_verified_delivery'), + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'in_progress', 'verified', 'delivered', 'failed')), + operation_id TEXT, + operation_generation BIGINT CHECK (operation_generation IS NULL OR operation_generation > 0), + operation_object_key TEXT, + receipt JSONB, + claim_owner TEXT, + claim_token TEXT, + claim_until TIMESTAMPTZ, + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + first_attempt_at TIMESTAMPTZ, + verified_at TIMESTAMPTZ, + local_deleted_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (reservation_id, artifact_kind, segment_index), + CHECK ( + (operation_id IS NULL AND operation_generation IS NULL AND operation_object_key IS NULL) + OR (operation_id IS NOT NULL AND operation_generation IS NOT NULL + AND operation_object_key IS NOT NULL) + ), + 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) + 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) + 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 ((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) + OR (status = 'failed' + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND claim_token IS NULL + AND claim_until IS NULL) + ) +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_artifact_delivery_claim_idx + ON call_scribe_hosted_artifact_delivery_outbox(next_attempt_at, claim_until, created_at) + WHERE status IN ('pending', 'in_progress', 'verified'); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_artifact_delivery_backpressure_idx + ON call_scribe_hosted_artifact_delivery_outbox(organization_id, guild_id, created_at) + WHERE status <> 'delivered'; + +-- Header-only zero-duration WAVs have no deliverable audio and no remaining +-- reservation authority. Keep their local removal crash-safe without putting +-- an impossible artifact into the delivery/backpressure state machine. +CREATE TABLE IF NOT EXISTS call_scribe_hosted_zero_duration_cleanup_outbox ( + local_path TEXT PRIMARY KEY, + organization_id TEXT NOT NULL, + recording_id TEXT NOT NULL + REFERENCES call_scribe_capture_sessions(id) ON DELETE RESTRICT, + claim_token TEXT, + claim_until TIMESTAMPTZ, + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((claim_token IS NULL AND claim_until IS NULL) + OR (claim_token IS NOT NULL AND claim_until IS NOT NULL)) +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_zero_duration_cleanup_claim_idx + ON call_scribe_hosted_zero_duration_cleanup_outbox(next_attempt_at, claim_until, created_at); diff --git a/src/hosted_control.rs b/src/hosted_control.rs index 0f8e35e..6540956 100644 --- a/src/hosted_control.rs +++ b/src/hosted_control.rs @@ -1,19 +1,28 @@ use std::collections::{HashMap, HashSet}; +use std::io::SeekFrom; +use std::path::Path; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; use anyhow::{Context, Result, bail}; +use base64::Engine; use chacha20poly1305::aead::{Aead, KeyInit, Payload}; use chacha20poly1305::{ChaCha20Poly1305, Nonce}; use chrono::{DateTime, Utc}; -use reqwest::{Client, Method, StatusCode, Url, redirect}; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; +use reqwest::{Body, Client, Method, StatusCode, Url, redirect}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; 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 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); const WORKER_ID_HEADER: &str = "X-Call-Scribe-Worker-Id"; pub(crate) const RESERVATION_EXPIRY_MARGIN: Duration = Duration::from_secs(15); pub(crate) const RESERVATION_MAX_LEASE: Duration = Duration::from_secs(90); @@ -23,6 +32,7 @@ const RESERVATION_CLOCK_SKEW_TOLERANCE: Duration = Duration::from_secs(5); #[derive(Clone)] pub(crate) struct HostedControlPlaneClient { http: Client, + upload_http: Client, base_url: Url, workload_token: String, worker_id: String, @@ -47,6 +57,11 @@ pub(crate) struct GuildConfiguration { pub remaining_recording_seconds: Option, pub storage_provider: Option, pub storage_destination_label: Option, + pub storage_destination_id: Option, + pub storage_destination_revision: Option, + pub storage_allowed_host: Option, + pub storage_object_key_prefix: Option, + pub transient_delete_policy: Option, pub ready: bool, #[serde(default)] pub blocked_reasons: Vec, @@ -54,6 +69,34 @@ pub(crate) struct GuildConfiguration { } impl GuildConfiguration { + pub(crate) fn storage_destination(&self) -> Option { + let provider = self.storage_provider.as_deref()?; + let destination_id = self.storage_destination_id.as_deref()?; + let destination_revision = self.storage_destination_revision.as_deref()?; + let allowed_host = self.storage_allowed_host.as_deref()?; + let object_key_prefix = self.storage_object_key_prefix.as_deref()?; + let transient_delete_policy = self.transient_delete_policy.as_deref()?; + if !storage_destination_supported(provider) + || !valid_opaque_id(destination_id, 200) + || !valid_opaque_id(destination_revision, 200) + || !valid_pinned_provider_host(provider, allowed_host) + || !valid_object_key_prefix(object_key_prefix) + || transient_delete_policy != "delete_after_verified_delivery" + { + return None; + } + Some(HostedStorageDestination { + organization_id: self.organization_id.clone(), + guild_id: self.guild_id.clone(), + provider: provider.to_string(), + destination_id: destination_id.to_string(), + destination_revision: destination_revision.to_string(), + allowed_host: allowed_host.to_string(), + object_key_prefix: object_key_prefix.to_string(), + transient_delete_policy: transient_delete_policy.to_string(), + }) + } + fn non_usage_controls_permit_recording(&self, channel_id: u64) -> bool { self.entitlement_active && self.recording_enabled @@ -109,16 +152,26 @@ impl GuildConfiguration { } pub(crate) fn permits_recording(&self, channel_id: u64) -> bool { - self.core_controls_permit_recording(channel_id) - && storage_destination_supported(self.storage_provider.as_deref()) + self.core_controls_permit_recording(channel_id) && self.storage_destination().is_some() } pub(crate) fn permits_continuation(&self, channel_id: u64) -> bool { - self.core_controls_permit_continuation(channel_id) - && storage_destination_supported(self.storage_provider.as_deref()) + self.core_controls_permit_continuation(channel_id) && self.storage_destination().is_some() } } +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct HostedStorageDestination { + pub organization_id: String, + pub guild_id: String, + pub provider: String, + pub destination_id: String, + pub destination_revision: String, + pub allowed_host: String, + pub object_key_prefix: String, + pub transient_delete_policy: String, +} + #[derive(Clone, Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct GuildConfigurationResponse { @@ -225,6 +278,134 @@ pub(crate) struct UsageReservation { pub expires_at: String, } +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ArtifactDeliveryPrepareRequest<'a> { + pub reservation_id: &'a str, + pub lease_token: &'a str, + pub recording_id: &'a str, + pub artifact_id: &'a str, + pub artifact_kind: &'static str, + pub segment_index: u32, + pub content_length: u64, + pub sha256: &'a str, + pub content_type: &'static str, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct ArtifactDeliveryPrepareResponse { + pub operation_id: String, + pub generation: u64, + pub recording_id: String, + pub artifact_id: String, + pub artifact_kind: String, + pub segment_index: u32, + pub object_key: String, + pub destination_id: String, + pub destination_revision: String, + pub provider: String, + pub allowed_upload_host: String, + pub upload: SignedArtifactUpload, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ArtifactDeliveryOperationRef { + pub operation_id: String, + pub generation: u64, + pub recording_id: String, + pub artifact_id: String, + pub artifact_kind: String, + pub segment_index: u32, + pub object_key: String, + pub destination_id: String, + pub destination_revision: String, + pub provider: String, + pub allowed_upload_host: String, +} + +impl From<&ArtifactDeliveryPrepareResponse> for ArtifactDeliveryOperationRef { + fn from(prepared: &ArtifactDeliveryPrepareResponse) -> Self { + Self { + operation_id: prepared.operation_id.clone(), + generation: prepared.generation, + recording_id: prepared.recording_id.clone(), + artifact_id: prepared.artifact_id.clone(), + artifact_kind: prepared.artifact_kind.clone(), + segment_index: prepared.segment_index, + object_key: prepared.object_key.clone(), + destination_id: prepared.destination_id.clone(), + destination_revision: prepared.destination_revision.clone(), + provider: prepared.provider.clone(), + allowed_upload_host: prepared.allowed_upload_host.clone(), + } + } +} + +pub(crate) enum ArtifactDeliveryVerification { + Verified(Box), + NotReady, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct SignedArtifactUpload { + pub url: String, + pub method: String, + #[serde(default)] + pub headers: HashMap, + pub expires_at: DateTime, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ArtifactDeliveryVerifyRequest { + generation: u64, + recording_id: String, + artifact_id: String, + artifact_kind: String, + segment_index: u32, + object_key: String, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ArtifactDeliveryVerificationResponse { + receipt: ArtifactDeliveryReceipt, + verification: SignedArtifactVerification, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct SignedArtifactVerification { + url: String, + method: String, + #[serde(default)] + headers: HashMap, + expires_at: DateTime, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct ArtifactDeliveryReceipt { + pub receipt_id: String, + pub operation_id: String, + pub generation: u64, + pub recording_id: String, + pub artifact_id: String, + pub artifact_kind: String, + pub segment_index: u32, + pub object_key: String, + pub destination_id: String, + pub destination_revision: String, + pub provider: String, + pub allowed_upload_host: String, + pub verified: bool, + pub content_length: u64, + pub sha256: String, + pub verified_at: DateTime, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct CommandLeaseRequest<'a> { @@ -340,12 +521,18 @@ impl HostedControlPlaneClient { .redirect(redirect::Policy::none()) .build() .context("failed to build hosted control-plane HTTP client")?; + let upload_http = Client::builder() + .timeout(Duration::from_secs(8 * 60)) + .redirect(redirect::Policy::none()) + .build() + .context("failed to build hosted artifact upload HTTP client")?; let mut key_derivation = Sha256::new(); key_derivation.update(b"call-scribe-hosted-usage-outbox-v1\0"); key_derivation.update(outbox_encryption_secret.as_bytes()); let outbox_encryption_key: [u8; 32] = key_derivation.finalize().into(); Ok(Self { http, + upload_http, base_url, workload_token, worker_id, @@ -607,6 +794,600 @@ impl HostedControlPlaneClient { } Ok(()) } + + pub(crate) async fn prepare_artifact_delivery( + &self, + request: &ArtifactDeliveryPrepareRequest<'_>, + destination: &HostedStorageDestination, + ) -> Result { + validate_artifact_manifest_request(request)?; + let response = self + .authenticated_request( + Method::POST, + self.base_url.join(ARTIFACT_DELIVERY_PREPARE_PATH)?, + ) + .json(request) + .send() + .await + .context("hosted artifact delivery preparation failed")?; + if response.status() != StatusCode::OK && response.status() != StatusCode::CREATED { + bail!( + "hosted artifact delivery preparation returned HTTP {}", + response.status() + ); + } + let prepared: ArtifactDeliveryPrepareResponse = decode_json_bounded(response, 64 * 1024) + .await + .context("hosted artifact delivery preparation response was invalid")?; + validate_prepared_upload(&prepared, request, destination)?; + Ok(prepared) + } + + pub(crate) async fn upload_artifact( + &self, + prepared: &ArtifactDeliveryPrepareResponse, + request: &ArtifactDeliveryPrepareRequest<'_>, + local_path: &Path, + destination: &HostedStorageDestination, + ) -> Result<()> { + let (url, headers) = validate_prepared_upload(prepared, request, destination)?; + let metadata = tokio::fs::symlink_metadata(local_path) + .await + .with_context(|| { + format!( + "failed to inspect local hosted artifact {}", + local_path.display() + ) + })?; + if metadata.file_type().is_symlink() + || !metadata.is_file() + || metadata.len() != request.content_length + { + bail!("local hosted artifact changed after its manifest was persisted"); + } + let mut file = tokio::fs::File::open(local_path).await.with_context(|| { + format!( + "failed to open local hosted artifact {}", + local_path.display() + ) + })?; + let mut digest = Sha256::new(); + let mut buffer = vec![0_u8; 1024 * 1024]; + loop { + let read = file + .read(&mut buffer) + .await + .context("failed to revalidate hosted artifact before upload")?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + if format!("{:x}", digest.finalize()) != request.sha256 { + bail!("local hosted artifact changed after its manifest was persisted"); + } + file.seek(SeekFrom::Start(0)) + .await + .context("failed to rewind hosted artifact for upload")?; + let body = Body::wrap_stream(tokio_util::io::ReaderStream::new(file)); + let response = self + .upload_http + .put(url) + .headers(headers) + .body(body) + .send() + .await + .map_err(|_| anyhow::anyhow!("customer storage upload transport failed"))?; + if !response.status().is_success() { + bail!( + "customer storage upload returned HTTP {}", + response.status() + ); + } + Ok(()) + } + + pub(crate) async fn verify_artifact_delivery( + &self, + operation: &ArtifactDeliveryOperationRef, + request: &ArtifactDeliveryPrepareRequest<'_>, + destination: &HostedStorageDestination, + ) -> Result { + let path = format!( + "internal/v1/worker/artifact-deliveries/{}/verify", + urlencoding::encode(&operation.operation_id) + ); + let response = self + .authenticated_request(Method::POST, self.base_url.join(&path)?) + .json(&ArtifactDeliveryVerifyRequest { + generation: operation.generation, + recording_id: operation.recording_id.clone(), + artifact_id: operation.artifact_id.clone(), + artifact_kind: operation.artifact_kind.clone(), + segment_index: operation.segment_index, + object_key: operation.object_key.clone(), + }) + .send() + .await + .context("hosted artifact delivery verification failed")?; + if response.status() == StatusCode::CONFLICT { + return Ok(ArtifactDeliveryVerification::NotReady); + } + if response.status() != StatusCode::OK { + bail!( + "hosted artifact delivery verification returned HTTP {}", + response.status() + ); + } + let proof: ArtifactDeliveryVerificationResponse = decode_json_bounded(response, 64 * 1024) + .await + .context("hosted artifact delivery verification response was invalid")?; + validate_delivery_receipt(&proof.receipt, operation, request, destination)?; + let (url, headers) = + validate_signed_verification(&proof.verification, operation, destination)?; + let provider_response = self + .upload_http + .head(url) + .headers(headers) + .send() + .await + .map_err(|_| anyhow::anyhow!("customer storage verification transport failed"))?; + if provider_response.status() == StatusCode::NOT_FOUND { + return Ok(ArtifactDeliveryVerification::NotReady); + } + if !provider_response.status().is_success() { + bail!( + "customer storage verification returned HTTP {}", + provider_response.status() + ); + } + validate_provider_head_response(provider_response.headers(), request, destination)?; + Ok(ArtifactDeliveryVerification::Verified(Box::new( + proof.receipt, + ))) + } +} + +fn validate_artifact_manifest_request(request: &ArtifactDeliveryPrepareRequest<'_>) -> Result<()> { + if !valid_opaque_id(request.reservation_id, 200) + || request.lease_token.len() < 16 + || request.lease_token.len() > 1_024 + || !valid_opaque_id(request.recording_id, 200) + || !valid_opaque_id(request.artifact_id, 200) + || request.artifact_kind != "raw_audio_wav" + || request.segment_index == 0 + || request.segment_index > 16 + || request.content_length == 0 + || request.sha256.len() != 64 + || !request + .sha256 + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + || request.content_type != "audio/wav" + { + bail!("hosted artifact manifest was invalid"); + } + Ok(()) +} + +fn validate_prepared_upload( + prepared: &ArtifactDeliveryPrepareResponse, + request: &ArtifactDeliveryPrepareRequest<'_>, + destination: &HostedStorageDestination, +) -> Result<(Url, HeaderMap)> { + if !valid_opaque_id(&prepared.operation_id, 200) + || prepared.generation == 0 + || prepared.recording_id != request.recording_id + || prepared.artifact_id != request.artifact_id + || prepared.artifact_kind != request.artifact_kind + || prepared.segment_index != request.segment_index + || !valid_object_key(&prepared.object_key) + || !prepared + .object_key + .starts_with(&destination.object_key_prefix) + || prepared.destination_id != destination.destination_id + || prepared.destination_revision != destination.destination_revision + || prepared.provider != destination.provider + || prepared.allowed_upload_host != destination.allowed_host + || prepared.upload.method != "PUT" + { + bail!("hosted artifact delivery grant did not match its pinned destination"); + } + let url = Url::parse(&prepared.upload.url).context("hosted upload URL was invalid")?; + if prepared.upload.url.len() > MAX_SIGNED_UPLOAD_URL_BYTES + || !provider_url_origin_is_allowed(&url) + || !url.username().is_empty() + || url.password().is_some() + || url.fragment().is_some() + || !url_path_matches_object_key(&url, &prepared.object_key) + || url.host_str() != Some(prepared.allowed_upload_host.as_str()) + { + bail!("hosted upload URL escaped the provider host boundary"); + } + if prepared.upload.headers.is_empty() + || prepared.upload.headers.len() > MAX_SIGNED_UPLOAD_HEADERS + { + bail!("hosted upload headers were missing or excessive"); + } + let mut headers = HeaderMap::new(); + for (name, value) in &prepared.upload.headers { + let normalized = name.to_ascii_lowercase(); + if name != &normalized + || !allowed_upload_header(&normalized) + || value.len() > 4_096 + || value.contains(['\r', '\n']) + { + bail!("hosted upload grant contained an unsafe signed header"); + } + let name = HeaderName::from_bytes(normalized.as_bytes()) + .context("hosted upload header name was invalid")?; + let value = + HeaderValue::from_str(value).context("hosted upload header value was invalid")?; + if headers.insert(name, value).is_some() { + bail!("hosted upload grant duplicated a signed header"); + } + } + if headers + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + != Some(request.content_type) + || headers + .get(reqwest::header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + != Some(request.content_length) + { + bail!("hosted upload grant did not bind the artifact type and length"); + } + validate_sigv4_capability(&url, &headers, prepared.upload.expires_at)?; + let hash_is_bound = headers + .get("x-amz-meta-callscribe-sha256") + .and_then(|value| value.to_str().ok()) + == Some(request.sha256); + if !hash_is_bound { + bail!("hosted upload grant did not bind the artifact SHA-256"); + } + let provider_checksum = headers + .get("x-amz-checksum-sha256") + .and_then(|value| value.to_str().ok()); + match prepared.provider.as_str() { + "customer_s3" => { + let expected = base64::engine::general_purpose::STANDARD + .encode(decode_sha256_hex(request.sha256)?); + if provider_checksum != Some(expected.as_str()) { + bail!("AWS S3 upload grant did not bind the provider-verified SHA-256 checksum"); + } + } + "customer_r2" if provider_checksum.is_some() => { + bail!("R2 upload grant requested an unsupported full-object SHA-256 checksum"); + } + "customer_r2" => {} + _ => bail!("hosted upload provider was unsupported"), + } + Ok((url, headers)) +} + +fn validate_signed_verification( + verification: &SignedArtifactVerification, + operation: &ArtifactDeliveryOperationRef, + destination: &HostedStorageDestination, +) -> Result<(Url, HeaderMap)> { + if verification.method != "HEAD" { + bail!("hosted verification capability did not use HEAD"); + } + let url = Url::parse(&verification.url).context("hosted verification URL was invalid")?; + if verification.url.len() > MAX_SIGNED_UPLOAD_URL_BYTES + || !provider_url_origin_is_allowed(&url) + || !url.username().is_empty() + || url.password().is_some() + || url.fragment().is_some() + || url.host_str() != Some(destination.allowed_host.as_str()) + || destination.allowed_host != operation.allowed_upload_host + || !operation + .object_key + .starts_with(&destination.object_key_prefix) + || !url_path_matches_object_key(&url, &operation.object_key) + || verification.headers.len() > MAX_SIGNED_UPLOAD_HEADERS + { + bail!("hosted verification capability escaped the exact object boundary"); + } + let mut headers = HeaderMap::new(); + for (name, value) in &verification.headers { + let normalized = name.to_ascii_lowercase(); + if name != &normalized + || !matches!(normalized.as_str(), "x-amz-checksum-mode") + || value.len() > 100 + || value.contains(['\r', '\n']) + { + bail!("hosted verification capability contained an unsafe header"); + } + let name = HeaderName::from_bytes(normalized.as_bytes()) + .context("hosted verification header name was invalid")?; + let value = + HeaderValue::from_str(value).context("hosted verification header value was invalid")?; + if headers.insert(name, value).is_some() { + bail!("hosted verification capability duplicated a header"); + } + } + match destination.provider.as_str() { + "customer_s3" + if headers + .get("x-amz-checksum-mode") + .and_then(|value| value.to_str().ok()) + != Some("ENABLED") => + { + bail!("AWS S3 verification did not request provider checksum evidence"); + } + "customer_r2" if !headers.is_empty() => { + bail!("R2 verification requested unsupported provider headers"); + } + "customer_s3" | "customer_r2" => {} + _ => bail!("hosted verification provider was unsupported"), + } + validate_sigv4_capability(&url, &headers, verification.expires_at)?; + Ok((url, headers)) +} + +fn validate_provider_head_response( + headers: &HeaderMap, + request: &ArtifactDeliveryPrepareRequest<'_>, + destination: &HostedStorageDestination, +) -> Result<()> { + let content_length = headers + .get(reqwest::header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); + let metadata_sha256 = headers + .get("x-amz-meta-callscribe-sha256") + .and_then(|value| value.to_str().ok()); + if content_length != Some(request.content_length) || metadata_sha256 != Some(request.sha256) { + bail!("customer storage did not contain the exact uploaded artifact"); + } + match destination.provider.as_str() { + "customer_s3" => { + let expected = base64::engine::general_purpose::STANDARD + .encode(decode_sha256_hex(request.sha256)?); + if headers + .get("x-amz-checksum-sha256") + .and_then(|value| value.to_str().ok()) + != Some(expected.as_str()) + { + bail!("AWS S3 HEAD did not prove the provider-verified artifact checksum"); + } + } + "customer_r2" => {} + _ => bail!("hosted verification provider was unsupported"), + } + Ok(()) +} + +fn decode_sha256_hex(value: &str) -> Result<[u8; 32]> { + if value.len() != 64 + || !value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + bail!("artifact SHA-256 was not lowercase hexadecimal"); + } + let mut decoded = [0_u8; 32]; + for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() { + let pair = std::str::from_utf8(pair).expect("ASCII hexadecimal remains UTF-8"); + decoded[index] = u8::from_str_radix(pair, 16) + .context("artifact SHA-256 was not lowercase hexadecimal")?; + } + Ok(decoded) +} + +fn validate_delivery_receipt( + receipt: &ArtifactDeliveryReceipt, + operation: &ArtifactDeliveryOperationRef, + request: &ArtifactDeliveryPrepareRequest<'_>, + destination: &HostedStorageDestination, +) -> Result<()> { + if !receipt.verified + || !valid_opaque_id(&receipt.receipt_id, 200) + || receipt.operation_id != operation.operation_id + || receipt.generation != operation.generation + || receipt.recording_id != request.recording_id + || receipt.artifact_id != request.artifact_id + || receipt.artifact_kind != request.artifact_kind + || receipt.segment_index != request.segment_index + || receipt.object_key != operation.object_key + || receipt.destination_id != destination.destination_id + || receipt.destination_revision != destination.destination_revision + || receipt.provider != destination.provider + || receipt.allowed_upload_host != destination.allowed_host + || receipt.content_length != request.content_length + || receipt.sha256 != request.sha256 + || receipt.verified_at > Utc::now() + chrono::Duration::minutes(1) + { + bail!("hosted artifact delivery receipt did not verify the exact artifact"); + } + Ok(()) +} + +fn allowed_upload_header(name: &str) -> bool { + matches!( + name, + "content-type" + | "content-length" + | "x-amz-checksum-sha256" + | "x-amz-meta-callscribe-sha256" + | "x-amz-server-side-encryption" + | "x-amz-server-side-encryption-aws-kms-key-id" + | "x-amz-server-side-encryption-context" + ) +} + +fn provider_host_is_allowed(provider: &str, host: &str) -> bool { + if host.is_empty() + || host != host.to_ascii_lowercase() + || host.starts_with('.') + || host.ends_with('.') + || !host + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')) + { + return false; + } + let labels: Vec<_> = host.split('.').collect(); + match provider { + "customer_r2" => { + labels.len() == 4 + && labels[1..] == ["r2", "cloudflarestorage", "com"] + && !labels[0].is_empty() + } + "customer_s3" => { + labels.ends_with(&["amazonaws", "com"]) + && labels[..labels.len().saturating_sub(2)] + .iter() + .position(|label| *label == "s3") + .is_some_and(|s3_index| s3_index > 0) + } + _ => false, + } +} + +fn valid_pinned_provider_host(provider: &str, host: &str) -> bool { + host.len() <= 253 && provider_host_is_allowed(provider, host) +} + +fn provider_url_origin_is_allowed(url: &Url) -> bool { + (url.scheme() == "https" && url.port_or_known_default() == Some(443)) + || (cfg!(test) + && url.scheme() == "http" + && url.port().is_some() + && url.host_str().is_some_and(is_loopback_host)) +} + +fn valid_object_key(value: &str) -> bool { + !value.is_empty() + && value.len() <= 1_024 + && !value.starts_with('/') + && !value.ends_with('/') + && value.split('/').all(|segment| { + !segment.is_empty() + && !matches!(segment, "." | "..") + && segment + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + }) +} + +fn valid_object_key_prefix(value: &str) -> bool { + value.strip_suffix('/').is_some_and(valid_object_key) && value.len() <= 512 +} + +fn url_path_matches_object_key(url: &Url, object_key: &str) -> bool { + valid_object_key(object_key) && url.path() == format!("/{object_key}") +} + +fn validate_sigv4_capability( + url: &Url, + request_headers: &HeaderMap, + detached_expires_at: DateTime, +) -> Result<()> { + if url.query().is_none_or(|query| query.len() > 12 * 1024) { + bail!("signed provider capability query was missing or excessive"); + } + let mut parameters = HashMap::new(); + for (name, value) in url.query_pairs() { + if name.len() > 100 || value.len() > 8 * 1024 { + bail!("signed provider capability parameter was excessive"); + } + if parameters + .insert(name.into_owned(), value.into_owned()) + .is_some() + { + bail!("signed provider capability duplicated a query parameter"); + } + } + if parameters.get("X-Amz-Algorithm").map(String::as_str) != Some("AWS4-HMAC-SHA256") { + bail!("signed provider capability algorithm was unsupported"); + } + let signature = parameters + .get("X-Amz-Signature") + .context("signed provider capability omitted its signature")?; + if signature.len() != 64 + || !signature + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + bail!("signed provider capability signature was malformed"); + } + + let signed_at_raw = parameters + .get("X-Amz-Date") + .context("signed provider capability omitted its signing time")?; + let signed_at = chrono::NaiveDateTime::parse_from_str(signed_at_raw, "%Y%m%dT%H%M%SZ") + .context("signed provider capability signing time was malformed")? + .and_utc(); + let expires_seconds = parameters + .get("X-Amz-Expires") + .and_then(|value| value.parse::().ok()) + .filter(|seconds| *seconds > 0 && *seconds <= MAX_UPLOAD_EXPIRY.as_secs()) + .context("signed provider capability lifetime was outside policy")?; + let signed_expires_at = signed_at + + chrono::Duration::seconds( + i64::try_from(expires_seconds).expect("bounded SigV4 expiry must fit i64"), + ); + let now = Utc::now(); + let maximum_expiry = chrono::Duration::from_std(MAX_UPLOAD_EXPIRY) + .expect("upload expiry bound must fit chrono duration"); + if signed_at > now + chrono::Duration::minutes(5) + || signed_expires_at <= now + chrono::Duration::seconds(15) + || signed_expires_at > now + maximum_expiry + || signed_expires_at != detached_expires_at + { + bail!("signed provider capability timestamps did not match their declared lifetime"); + } + + let credential = parameters + .get("X-Amz-Credential") + .context("signed provider capability omitted its credential scope")?; + let scope: Vec<_> = credential.split('/').collect(); + if scope.len() != 5 + || scope[0].is_empty() + || scope[0].len() > 256 + || signed_at_raw.get(..8) != Some(scope[1]) + || scope[2].is_empty() + || scope[2].len() > 100 + || scope[3] != "s3" + || scope[4] != "aws4_request" + { + bail!("signed provider capability credential scope was malformed"); + } + + let signed_headers_raw = parameters + .get("X-Amz-SignedHeaders") + .context("signed provider capability omitted its signed headers")?; + let signed_headers: Vec<_> = signed_headers_raw.split(';').collect(); + if signed_headers.is_empty() + || signed_headers.iter().any(|name| { + name.is_empty() + || name.bytes().any(|byte| byte.is_ascii_uppercase()) + || !name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + }) + || !signed_headers.windows(2).all(|pair| pair[0] < pair[1]) + || !signed_headers.contains(&"host") + || request_headers + .keys() + .any(|name| !signed_headers.contains(&name.as_str())) + { + bail!("signed provider capability did not bind every required header"); + } + Ok(()) +} + +fn valid_opaque_id(value: &str, maximum_len: usize) -> bool { + !value.is_empty() + && value.len() <= maximum_len + && value.trim() == value + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':')) } fn validate_reservation_expiry(expires_at: DateTime, source: &str) -> Result<()> { @@ -749,15 +1530,10 @@ fn is_loopback_host(host: &str) -> bool { matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1") } -fn storage_destination_supported(provider: Option<&str>) -> bool { - // A provider is only enabled after this crate can upload the completed - // artifacts, verify delivery, and apply local deletion/retention. Keeping - // this exhaustive and fail-closed prevents control-plane configuration - // from silently degrading to local PVC retention. - match provider { - Some("customer_s3" | "customer_r2" | "google_drive" | "managed_transient") | None => false, - Some(_) => false, - } +fn storage_destination_supported(provider: &str) -> bool { + // Only object providers supported by the exact-object grant/verification + // contract are accepted. Other values remain fail-closed. + matches!(provider, "customer_s3" | "customer_r2") } #[cfg(test)] @@ -766,7 +1542,7 @@ mod tests { use axum::extract::Path; use axum::http::HeaderMap; use axum::response::IntoResponse; - use axum::routing::post; + use axum::routing::{head, post}; use axum::{Json, Router}; fn reservation(expires_at: DateTime) -> UsageReservation { @@ -858,17 +1634,112 @@ mod tests { remaining_recording_seconds: Some(30), storage_provider: Some("customer_s3".to_string()), storage_destination_label: Some("worker-pvc".to_string()), + storage_destination_id: Some("dst_01".to_string()), + storage_destination_revision: Some("rev_01".to_string()), + storage_allowed_host: Some("bucket.s3.us-east-1.amazonaws.com".to_string()), + storage_object_key_prefix: Some("objects/".to_string()), + transient_delete_policy: Some("delete_after_verified_delivery".to_string()), ready: true, blocked_reasons: Vec::new(), desired_recording_generation: 1, } } + fn artifact_request<'a>() -> ArtifactDeliveryPrepareRequest<'a> { + ArtifactDeliveryPrepareRequest { + reservation_id: "res_01", + lease_token: "opaque-lease-token-with-sufficient-length", + recording_id: "rec_01", + artifact_id: "art_01", + artifact_kind: "raw_audio_wav", + segment_index: 1, + content_length: 44, + sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + content_type: "audio/wav", + } + } + + fn artifact_destination(provider: &str) -> HostedStorageDestination { + let allowed_host = match provider { + "customer_s3" => "bucket.s3.us-east-1.amazonaws.com", + "customer_r2" => "accountid.r2.cloudflarestorage.com", + _ => "unsupported.invalid", + }; + HostedStorageDestination { + organization_id: "org_01".to_string(), + guild_id: "1".to_string(), + provider: provider.to_string(), + destination_id: "dst_01".to_string(), + destination_revision: "rev_01".to_string(), + allowed_host: allowed_host.to_string(), + object_key_prefix: "objects/".to_string(), + transient_delete_policy: "delete_after_verified_delivery".to_string(), + } + } + + fn sigv4_url(host: &str, object_key: &str, signed_headers: &str) -> (String, DateTime) { + let signed_at = DateTime::::from_timestamp(Utc::now().timestamp(), 0) + .expect("current timestamp must be valid"); + let expires_at = signed_at + chrono::Duration::minutes(5); + let scope_date = signed_at.format("%Y%m%d"); + let signing_time = signed_at.format("%Y%m%dT%H%M%SZ"); + ( + format!( + "https://{host}/{object_key}?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=test-access/{scope_date}/us-east-1/s3/aws4_request&X-Amz-Date={signing_time}&X-Amz-Expires=300&X-Amz-SignedHeaders={signed_headers}&X-Amz-Signature={}", + "0".repeat(64) + ), + expires_at, + ) + } + + fn prepared_upload(provider: &str, host: &str) -> ArtifactDeliveryPrepareResponse { + let request = artifact_request(); + let mut headers = HashMap::from([ + ("content-type".to_string(), "audio/wav".to_string()), + ("content-length".to_string(), "44".to_string()), + ( + "x-amz-meta-callscribe-sha256".to_string(), + request.sha256.to_string(), + ), + ]); + if provider == "customer_s3" { + headers.insert( + "x-amz-checksum-sha256".to_string(), + base64::engine::general_purpose::STANDARD + .encode(decode_sha256_hex(request.sha256).expect("test checksum is valid")), + ); + } + let mut signed_header_names: Vec<_> = headers.keys().map(String::as_str).collect(); + signed_header_names.push("host"); + signed_header_names.sort_unstable(); + let object_key = "objects/art_01.wav"; + let (url, expires_at) = sigv4_url(host, object_key, &signed_header_names.join(";")); + ArtifactDeliveryPrepareResponse { + operation_id: "op_01".to_string(), + generation: 1, + recording_id: request.recording_id.to_string(), + artifact_id: request.artifact_id.to_string(), + artifact_kind: request.artifact_kind.to_string(), + segment_index: request.segment_index, + object_key: object_key.to_string(), + destination_id: "dst_01".to_string(), + destination_revision: "rev_01".to_string(), + provider: provider.to_string(), + allowed_upload_host: host.to_string(), + upload: SignedArtifactUpload { + url, + method: "PUT".to_string(), + headers, + expires_at, + }, + } + } + #[test] fn core_controls_require_fully_authorized_approved_channel() { let policy = enabled_policy(); assert!(policy.core_controls_permit_recording(2)); - assert!(!policy.permits_recording(2)); + assert!(policy.permits_recording(2)); assert!(!policy.core_controls_permit_recording(9)); } @@ -893,6 +1764,28 @@ mod tests { assert!(!policy.core_controls_permit_recording(2)); } + #[test] + fn incomplete_or_unsupported_storage_contract_fails_closed() { + let mut policy = enabled_policy(); + policy.storage_destination_revision = None; + assert!(!policy.permits_recording(2)); + policy.storage_destination_revision = Some("rev_01".to_string()); + policy.transient_delete_policy = Some("retain_locally".to_string()); + assert!(!policy.permits_recording(2)); + policy.transient_delete_policy = Some("delete_after_verified_delivery".to_string()); + policy.storage_provider = Some("google_drive".to_string()); + assert!(!policy.permits_recording(2)); + policy.storage_provider = Some("customer_s3".to_string()); + policy.storage_allowed_host = Some("s3.us-east-1.amazonaws.com".to_string()); + assert!( + !policy.permits_recording(2), + "path-style S3 cannot independently pin the customer bucket" + ); + policy.storage_allowed_host = Some("bucket.s3.us-east-1.amazonaws.com".to_string()); + policy.storage_object_key_prefix = None; + assert!(!policy.permits_recording(2)); + } + #[test] fn matching_reservation_can_continue_only_through_usage_exhaustion() { let mut policy = enabled_policy(); @@ -1100,6 +1993,11 @@ mod tests { "remainingRecordingSeconds": 3600, "storageProvider": "customer_s3", "storageDestinationLabel": "Customer archive", + "storageDestinationId": "dst_01", + "storageDestinationRevision": "rev_04", + "storageAllowedHost": "bucket.s3.us-east-1.amazonaws.com", + "storageObjectKeyPrefix": "objects/", + "transientDeletePolicy": "delete_after_verified_delivery", "ready": true, "blockedReasons": [], "desiredRecordingGeneration": 4 @@ -1107,7 +2005,7 @@ mod tests { })) .expect("control-plane snapshot should deserialize"); assert!(validate_configuration_response(&response).is_ok()); - assert!(!response.guilds[0].permits_recording(456)); + assert!(response.guilds[0].permits_recording(456)); } #[test] @@ -1314,15 +2212,330 @@ mod tests { } #[test] - fn all_current_hosted_storage_providers_fail_closed() { - for provider in [ - None, - Some("customer_s3"), - Some("customer_r2"), - Some("google_drive"), - Some("managed_transient"), - Some("unknown"), - ] { + fn artifact_prepare_contract_uses_exact_object_manifest() { + assert_eq!( + serde_json::to_value(artifact_request()).expect("artifact request should serialize"), + serde_json::json!({ + "reservationId": "res_01", + "leaseToken": "opaque-lease-token-with-sufficient-length", + "recordingId": "rec_01", + "artifactId": "art_01", + "artifactKind": "raw_audio_wav", + "segmentIndex": 1, + "contentLength": 44, + "sha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "contentType": "audio/wav" + }) + ); + } + + #[test] + fn signed_upload_is_exact_destination_host_and_manifest_bound() { + let request = artifact_request(); + let s3 = artifact_destination("customer_s3"); + let prepared = prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + assert!(validate_prepared_upload(&prepared, &request, &s3).is_ok()); + + let mut explicit_default_port = prepared.clone(); + explicit_default_port.upload.url = explicit_default_port.upload.url.replacen( + "bucket.s3.us-east-1.amazonaws.com/", + "bucket.s3.us-east-1.amazonaws.com:443/", + 1, + ); + assert!(validate_prepared_upload(&explicit_default_port, &request, &s3).is_ok()); + + let mut nonstandard_port = prepared.clone(); + nonstandard_port.upload.url = nonstandard_port.upload.url.replacen( + "bucket.s3.us-east-1.amazonaws.com/", + "bucket.s3.us-east-1.amazonaws.com:8443/", + 1, + ); + assert!( + validate_prepared_upload(&nonstandard_port, &request, &s3).is_err(), + "the pinned provider host must not authorize a different HTTPS origin port" + ); + + let r2 = artifact_destination("customer_r2"); + let prepared = prepared_upload("customer_r2", "accountid.r2.cloudflarestorage.com"); + assert!(validate_prepared_upload(&prepared, &request, &r2).is_ok()); + + let mut exfiltration = prepared.clone(); + exfiltration.upload.url = + "https://attacker.example/object?X-Amz-Signature=opaque".to_string(); + assert!(validate_prepared_upload(&exfiltration, &request, &r2).is_err()); + + let attacker_s3 = prepared_upload("customer_s3", "attacker.s3.us-east-1.amazonaws.com"); + assert!( + validate_prepared_upload(&attacker_s3, &request, &s3).is_err(), + "an in-family attacker bucket must not replace the independently pinned host" + ); + let attacker_r2 = prepared_upload("customer_r2", "attackerid.r2.cloudflarestorage.com"); + assert!( + validate_prepared_upload(&attacker_r2, &request, &r2).is_err(), + "an in-family attacker account must not replace the independently pinned host" + ); + + let mut wrong_artifact = + prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + wrong_artifact.artifact_id = "art_02".to_string(); + assert!(validate_prepared_upload(&wrong_artifact, &request, &s3).is_err()); + + let mut wrong_object = prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + wrong_object.object_key = "objects/different.wav".to_string(); + assert!(validate_prepared_upload(&wrong_object, &request, &s3).is_err()); + + let mut outside_prefix = + prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + outside_prefix.object_key = "outside/art_01.wav".to_string(); + outside_prefix.upload.url = outside_prefix + .upload + .url + .replace("/objects/art_01.wav", "/outside/art_01.wav"); + assert!(validate_prepared_upload(&outside_prefix, &request, &s3).is_err()); + + let mut unbound_hash = prepared; + unbound_hash + .upload + .headers + .remove("x-amz-meta-callscribe-sha256"); + assert!(validate_prepared_upload(&unbound_hash, &request, &r2).is_err()); + + let mut r2_with_unsupported_checksum = + prepared_upload("customer_r2", "accountid.r2.cloudflarestorage.com"); + r2_with_unsupported_checksum.upload.headers.insert( + "x-amz-checksum-sha256".to_string(), + "not-supported-on-r2-full-object-put".to_string(), + ); + assert!(validate_prepared_upload(&r2_with_unsupported_checksum, &request, &r2).is_err()); + + let mut aws_without_checksum = + prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + aws_without_checksum + .upload + .headers + .remove("x-amz-checksum-sha256"); + assert!(validate_prepared_upload(&aws_without_checksum, &request, &s3).is_err()); + } + + #[test] + fn sigv4_capability_lifetime_and_required_headers_are_signed() { + let request = artifact_request(); + let destination = artifact_destination("customer_s3"); + let valid = prepared_upload("customer_s3", &destination.allowed_host); + assert!(validate_prepared_upload(&valid, &request, &destination).is_ok()); + assert_eq!(valid.upload.expires_at.timestamp_subsec_nanos(), 0); + + let mut overlong = valid.clone(); + overlong.upload.url = overlong + .upload + .url + .replace("X-Amz-Expires=300", "X-Amz-Expires=900"); + overlong.upload.expires_at += chrono::Duration::minutes(10); + assert!(validate_prepared_upload(&overlong, &request, &destination).is_err()); + + let mut detached_mismatch = valid.clone(); + detached_mismatch.upload.expires_at += chrono::Duration::seconds(1); + assert!(validate_prepared_upload(&detached_mismatch, &request, &destination).is_err()); + + let mut fractional_detached_expiry = valid.clone(); + fractional_detached_expiry.upload.expires_at += chrono::Duration::nanoseconds(1); + assert!( + validate_prepared_upload(&fractional_detached_expiry, &request, &destination).is_err(), + "detached expiry must be the exact whole-second SigV4 expiry" + ); + + let mut unsigned_integrity = valid; + unsigned_integrity.upload.url = + unsigned_integrity.upload.url.replace("content-length;", ""); + assert!(validate_prepared_upload(&unsigned_integrity, &request, &destination).is_err()); + } + + #[test] + fn signed_verification_rejects_nonstandard_https_port() { + let destination = artifact_destination("customer_s3"); + let prepared = prepared_upload("customer_s3", &destination.allowed_host); + let operation = ArtifactDeliveryOperationRef::from(&prepared); + let (url, expires_at) = sigv4_url( + &destination.allowed_host, + &operation.object_key, + "host;x-amz-checksum-mode", + ); + let verification = SignedArtifactVerification { + url: url.replacen( + "bucket.s3.us-east-1.amazonaws.com/", + "bucket.s3.us-east-1.amazonaws.com:8443/", + 1, + ), + method: "HEAD".to_string(), + headers: HashMap::from([("x-amz-checksum-mode".to_string(), "ENABLED".to_string())]), + expires_at, + }; + assert!( + validate_signed_verification(&verification, &operation, &destination).is_err(), + "the pinned provider host must not authorize HEAD on another HTTPS origin port" + ); + } + + #[test] + fn provider_head_evidence_must_match_exact_bytes() { + let request = artifact_request(); + let destination = artifact_destination("customer_s3"); + let checksum = base64::engine::general_purpose::STANDARD + .encode(decode_sha256_hex(request.sha256).expect("checksum must decode")); + let mut headers = HeaderMap::new(); + headers.insert( + reqwest::header::CONTENT_LENGTH, + HeaderValue::from_static("44"), + ); + headers.insert( + HeaderName::from_static("x-amz-meta-callscribe-sha256"), + HeaderValue::from_str(request.sha256).expect("digest header must parse"), + ); + headers.insert( + HeaderName::from_static("x-amz-checksum-sha256"), + HeaderValue::from_str(&checksum).expect("checksum header must parse"), + ); + assert!(validate_provider_head_response(&headers, &request, &destination).is_ok()); + headers.insert( + reqwest::header::CONTENT_LENGTH, + HeaderValue::from_static("45"), + ); + assert!(validate_provider_head_response(&headers, &request, &destination).is_err()); + } + + #[tokio::test] + async fn matching_control_plane_receipt_without_provider_object_is_not_verified() { + let request = artifact_request(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("verification listener must bind"); + let address = listener + .local_addr() + .expect("listener must have an address"); + let host = "127.0.0.1"; + let object_key = "objects/art_01.wav"; + let (verification_url, expires_at) = + sigv4_url(&address.to_string(), object_key, "host;x-amz-checksum-mode"); + let verification_url = verification_url.replacen("https://", "http://", 1); + let receipt = serde_json::json!({ + "receiptId": "receipt_01", + "operationId": "op_01", + "generation": 1, + "recordingId": request.recording_id, + "artifactId": request.artifact_id, + "artifactKind": request.artifact_kind, + "segmentIndex": request.segment_index, + "objectKey": object_key, + "destinationId": "dst_01", + "destinationRevision": "rev_01", + "provider": "customer_s3", + "allowedUploadHost": host, + "verified": true, + "contentLength": request.content_length, + "sha256": request.sha256, + "verifiedAt": Utc::now().to_rfc3339(), + }); + let response = serde_json::json!({ + "receipt": receipt, + "verification": { + "url": verification_url, + "method": "HEAD", + "headers": {"x-amz-checksum-mode": "ENABLED"}, + "expiresAt": expires_at.to_rfc3339(), + } + }); + let app = Router::new() + .route( + "/internal/v1/worker/artifact-deliveries/op_01/verify", + post(move || { + let response = response.clone(); + async move { Json(response) } + }), + ) + .route( + "/objects/art_01.wav", + head(|| async { StatusCode::NOT_FOUND }), + ); + tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("verification server must run"); + }); + + let destination = HostedStorageDestination { + organization_id: "org_01".to_string(), + guild_id: "1".to_string(), + provider: "customer_s3".to_string(), + destination_id: "dst_01".to_string(), + destination_revision: "rev_01".to_string(), + allowed_host: host.to_string(), + object_key_prefix: "objects/".to_string(), + transient_delete_policy: "delete_after_verified_delivery".to_string(), + }; + let operation = ArtifactDeliveryOperationRef { + operation_id: "op_01".to_string(), + generation: 1, + recording_id: request.recording_id.to_string(), + artifact_id: request.artifact_id.to_string(), + artifact_kind: request.artifact_kind.to_string(), + segment_index: request.segment_index, + object_key: object_key.to_string(), + destination_id: destination.destination_id.clone(), + destination_revision: destination.destination_revision.clone(), + provider: destination.provider.clone(), + allowed_upload_host: host.to_string(), + }; + let client = test_client(&format!("http://{address}"), Duration::from_secs(2)); + assert!(matches!( + client + .verify_artifact_delivery(&operation, &request, &destination) + .await + .expect("missing provider object is a not-ready result"), + ArtifactDeliveryVerification::NotReady + )); + } + + #[test] + fn verified_receipt_must_match_the_exact_operation_and_manifest() { + let request = artifact_request(); + let destination = artifact_destination("customer_s3"); + let prepared = prepared_upload("customer_s3", "bucket.s3.us-east-1.amazonaws.com"); + let receipt = ArtifactDeliveryReceipt { + receipt_id: "receipt_01".to_string(), + operation_id: prepared.operation_id.clone(), + generation: prepared.generation, + recording_id: request.recording_id.to_string(), + artifact_id: request.artifact_id.to_string(), + artifact_kind: request.artifact_kind.to_string(), + segment_index: request.segment_index, + object_key: prepared.object_key.clone(), + destination_id: destination.destination_id.clone(), + destination_revision: destination.destination_revision.clone(), + provider: destination.provider.clone(), + allowed_upload_host: destination.allowed_host.clone(), + verified: true, + content_length: request.content_length, + sha256: request.sha256.to_string(), + verified_at: Utc::now(), + }; + let operation = ArtifactDeliveryOperationRef::from(&prepared); + assert!(validate_delivery_receipt(&receipt, &operation, &request, &destination).is_ok()); + let mut wrong = receipt; + wrong.content_length += 1; + assert!(validate_delivery_receipt(&wrong, &operation, &request, &destination).is_err()); + let mut wrong_object = wrong; + wrong_object.content_length = request.content_length; + wrong_object.object_key = "objects/another.wav".to_string(); + assert!( + validate_delivery_receipt(&wrong_object, &operation, &request, &destination).is_err() + ); + } + + #[test] + fn only_customer_s3_and_r2_are_supported() { + assert!(storage_destination_supported("customer_s3")); + assert!(storage_destination_supported("customer_r2")); + for provider in ["google_drive", "managed_transient", "unknown"] { assert!(!storage_destination_supported(provider)); } } diff --git a/src/main.rs b/src/main.rs index 8dce124..b5b30d7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,7 @@ use std::time::{Duration, Instant}; use std::{ collections::HashMap, fs::File, - io::BufWriter, + io::{BufWriter, Read}, num::NonZeroU8, ops::DerefMut, sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, @@ -15,6 +15,8 @@ use std::{ }; use anyhow::{Context as AnyhowContext, Result, bail}; +#[cfg(all(feature = "discord", test))] +use base64::Engine; use chrono::Local; #[cfg(feature = "discord")] use chrono::{DateTime, Utc}; @@ -30,8 +32,10 @@ mod providers; #[cfg(feature = "discord")] use hosted_control::{ - HostedConfigurationStore, HostedControlPlaneClient, RESERVATION_EXPIRY_MARGIN, - RESERVATION_SETTLEMENT_GRACE, UsageReservation, WorkerCommand, + ArtifactDeliveryOperationRef, ArtifactDeliveryPrepareRequest, ArtifactDeliveryReceipt, + ArtifactDeliveryVerification, HostedConfigurationStore, HostedControlPlaneClient, + HostedStorageDestination, RESERVATION_EXPIRY_MARGIN, RESERVATION_SETTLEMENT_GRACE, + UsageReservation, WorkerCommand, }; use providers::{ ElevenLabsSttConfig, ElevenLabsSttProvider, OpenAiSttConfig, OpenAiSttProvider, @@ -50,11 +54,15 @@ use serenity::client::{Client, Context as SerenityContext, EventHandler}; #[cfg(feature = "discord")] use serenity::prelude::GatewayIntents; #[cfg(feature = "discord")] +use sha2::{Digest, Sha256}; +#[cfg(feature = "discord")] use songbird::driver::{DecodeConfig, DecodeMode}; #[cfg(feature = "discord")] use songbird::model::payload::{ClientDisconnect, Speaking}; #[cfg(feature = "discord")] use songbird::{Config as SongbirdConfig, CoreEvent, EventContext, SerenityInit, Songbird}; +#[cfg(feature = "discord")] +use sqlx::row::Row; use sqlx_postgres::PgPool; #[cfg(feature = "discord")] use sqlx_postgres::PgPoolOptions; @@ -96,6 +104,12 @@ const HOSTED_START_STEP_TIMEOUT: Duration = Duration::from_secs(10); #[cfg(feature = "discord")] const HOSTED_AUTHORITY_WATCHDOG_INTERVAL: Duration = Duration::from_secs(1); #[cfg(feature = "discord")] +const HOSTED_DELIVERY_CLAIM_TTL: Duration = Duration::from_secs(15 * 60); +#[cfg(feature = "discord")] +const HOSTED_DELIVERY_BACKPRESSURE_AGE: Duration = Duration::from_secs(10 * 60); +#[cfg(feature = "discord")] +const HOSTED_DELIVERY_MAX_ATTEMPTS: i32 = 20; +#[cfg(feature = "discord")] const DISCORD_PLAYOUT_BUFFER_PACKETS: u8 = 12; #[cfg(feature = "discord")] const DISCORD_PLAYOUT_SPIKE_PACKETS: u8 = 8; @@ -392,6 +406,7 @@ pub(crate) async fn migrate_runtime_schema(pool: &PgPool) -> Result<()> { 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) @@ -552,8 +567,88 @@ type HostedCaptureRecoveryRow = ( i64, DateTime, DateTime, + String, + String, + String, + String, + String, + String, + String, + String, +); + +#[cfg(feature = "discord")] +type HostedPinnedRecoveryRow = ( + Vec, + Vec, + String, + String, + String, + String, + String, + String, + String, + String, ); +#[cfg(feature = "discord")] +struct HostedArtifactDeliveryRow { + artifact_id: String, + recording_id: String, + reservation_id: String, + encrypted_lease_token: Vec, + encryption_nonce: Vec, + 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, + storage_object_key_prefix: String, + transient_delete_policy: String, + operation_id: Option, + operation_generation: Option, + operation_object_key: Option, +} + +#[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 { + Ok(Self { + artifact_id: row.try_get("artifact_id")?, + recording_id: row.try_get("recording_id")?, + reservation_id: row.try_get("reservation_id")?, + encrypted_lease_token: row.try_get("encrypted_lease_token")?, + encryption_nonce: row.try_get("encryption_nonce")?, + 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")?, + storage_object_key_prefix: row.try_get("storage_object_key_prefix")?, + transient_delete_policy: row.try_get("transient_delete_policy")?, + operation_id: row.try_get("operation_id")?, + operation_generation: row.try_get("operation_generation")?, + operation_object_key: row.try_get("operation_object_key")?, + }) + } +} + +#[cfg(feature = "discord")] +#[derive(Clone, Debug)] +struct HostedArtifactManifest { + artifact_id: String, + segment_index: u32, + local_path: PathBuf, + content_length: u64, + sha256: String, +} + #[cfg(feature = "discord")] struct FinalizingRecordingGuard { finalizing: Arc>, @@ -601,6 +696,7 @@ struct ActiveCapture { runtime_store: Option, capture_mode: CaptureMode, hosted_usage: Option<(HostedControlPlaneClient, UsageReservation)>, + hosted_storage: Option, hosted_generation: Option, } @@ -1096,6 +1192,7 @@ WHERE command_id = $1 AND status = 'started' Ok(()) } + #[allow(clippy::too_many_arguments)] async fn persist_hosted_capture_recovery( &self, client: &HostedControlPlaneClient, @@ -1104,6 +1201,7 @@ WHERE command_id = $1 AND status = 'started' base_wav_path: &Path, started_at: DateTime, owner_instance_id: &str, + destination: &HostedStorageDestination, ) -> Result<()> { let base_wav_path = base_wav_path .to_str() @@ -1123,8 +1221,11 @@ WHERE command_id = $1 AND status = 'started' r#" INSERT INTO call_scribe_hosted_capture_recovery (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, - base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id, + organization_id, guild_id, storage_provider, storage_destination_id, + storage_destination_revision, storage_allowed_host, + storage_object_key_prefix, transient_delete_policy) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) "#, ) .bind(&reservation.reservation_id) @@ -1136,6 +1237,14 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) .bind(started_at) .bind(expires_at) .bind(owner_instance_id) + .bind(&destination.organization_id) + .bind(&destination.guild_id) + .bind(&destination.provider) + .bind(&destination.destination_id) + .bind(&destination.destination_revision) + .bind(&destination.allowed_host) + .bind(&destination.object_key_prefix) + .bind(&destination.transient_delete_policy) .execute(&self.pool) .await .context("failed to persist hosted capture recovery state")?; @@ -1206,6 +1315,7 @@ RETURNING expires_at Ok((recovery_claim_token, effective_authorization_end)) } + #[cfg(test)] async fn remove_claimed_hosted_capture_recovery( &self, reservation_id: &str, @@ -1339,7 +1449,11 @@ WHERE recovery.reservation_id = candidates.reservation_id RETURNING recovery.reservation_id, recovery.encrypted_lease_token, recovery.encryption_nonce, recovery.recording_id, recovery.base_wav_path, recovery.reserved_seconds, - recovery.started_at, recovery.expires_at + recovery.started_at, recovery.expires_at, recovery.organization_id, + recovery.guild_id, recovery.storage_provider, + recovery.storage_destination_id, recovery.storage_destination_revision, + recovery.storage_allowed_host, recovery.storage_object_key_prefix, + recovery.transient_delete_policy "#, ) .bind(stale_before) @@ -1394,6 +1508,14 @@ RETURNING reservation_id, recording_id reserved_seconds, started_at, expires_at, + organization_id, + guild_id, + storage_provider, + storage_destination_id, + storage_destination_revision, + storage_allowed_host, + storage_object_key_prefix, + transient_delete_policy, ) in recoveries { let recovery_result = async { @@ -1414,31 +1536,59 @@ RETURNING reservation_id, recording_id reserved_seconds, expires_at: expires_at.to_rfc3339(), }; + let destination = HostedStorageDestination { + organization_id, + guild_id, + provider: storage_provider, + destination_id: storage_destination_id, + destination_revision: storage_destination_revision, + allowed_host: storage_allowed_host, + object_key_prefix: storage_object_key_prefix, + transient_delete_policy, + }; + if destination.transient_delete_policy != "delete_after_verified_delivery" + || !matches!(destination.provider.as_str(), "customer_s3" | "customer_r2") + { + bail!("hosted recovery pinned an unsupported storage destination"); + } let actual_seconds = recovered_usage_seconds(&base_wav_path, reserved_seconds)?.min( authorized_usage_seconds(started_at, expires_at, reserved_seconds), ); + let wav_paths = checkpointed_wav_paths(&base_wav_path)?; if actual_seconds == 0 { client.release_usage(&reservation).await?; - } else { - let occurred_at = started_at - + chrono::Duration::seconds( - i64::try_from(actual_seconds) - .context("hosted recovery duration exceeded timestamp range")?, - ); - self.enqueue_hosted_usage( - client, + self.finalize_zero_duration_hosted_capture_transaction( &reservation, &recording_id, - actual_seconds, - occurred_at, + &recovery_claim_token, + &destination, + &wav_paths, + Utc::now(), ) .await?; + if let Err(err) = self.finish_zero_duration_hosted_cleanup(capture_dir).await { + eprintln!( + "zero-duration hosted cleanup remains durably queued for recording {recording_id}: {err:#}" + ); + } + return Ok(()); } - self.remove_claimed_hosted_capture_recovery( - &reservation_id, + let manifests = hosted_artifact_manifests(&wav_paths).await?; + let occurred_at = started_at + + chrono::Duration::seconds( + i64::try_from(actual_seconds) + .context("hosted recovery duration exceeded timestamp range")?, + ); + self.finalize_hosted_capture_transaction( + &reservation, &recording_id, &recovery_claim_token, + actual_seconds, + occurred_at, + &destination, + &manifests, + Utc::now(), ) .await } @@ -1453,14 +1603,24 @@ RETURNING reservation_id, recording_id Ok(()) } - async fn enqueue_hosted_usage( + #[allow(clippy::too_many_arguments)] + async fn finalize_hosted_capture_transaction( &self, - client: &HostedControlPlaneClient, reservation: &UsageReservation, recording_id: &str, + recovery_claim_token: &str, actual_seconds: u64, occurred_at: DateTime, + destination: &HostedStorageDestination, + manifests: &[HostedArtifactManifest], + stopped_at: DateTime, ) -> Result<()> { + if actual_seconds == 0 { + bail!("zero-duration hosted capture must not enter artifact delivery"); + } + if manifests.is_empty() || manifests.len() > MAX_HOSTED_RECOVERY_WAV_SEGMENTS as usize { + bail!("hosted capture did not produce a bounded raw-audio manifest"); + } let actual_seconds = i64::try_from(actual_seconds) .context("hosted usage duration exceeded database range")?; let authorization_expires_at = DateTime::parse_from_rfc3339(&reservation.expires_at) @@ -1475,52 +1635,890 @@ RETURNING reservation_id, recording_id if expires_at <= Utc::now() { bail!("hosted usage settlement window expired before it could be queued"); } - let (encrypted_lease_token, encryption_nonce) = client - .encrypt_reservation_lease(&reservation.reservation_id, &reservation.lease_token)?; - let inserted = sqlx::query::query( + let mut tx = self + .pool + .begin() + .await + .context("failed to begin hosted finalization")?; + let recovery: HostedPinnedRecoveryRow = sqlx::query_as::query_as( + r#" +SELECT encrypted_lease_token, encryption_nonce, organization_id, guild_id, + storage_provider, storage_destination_id, storage_destination_revision, + storage_allowed_host, storage_object_key_prefix, transient_delete_policy +FROM call_scribe_hosted_capture_recovery +WHERE reservation_id = $1 + AND recording_id = $2 + AND status = 'reconciling' + AND recovery_claim_token = $3 +FOR UPDATE +"#, + ) + .bind(&reservation.reservation_id) + .bind(recording_id) + .bind(recovery_claim_token) + .fetch_one(&mut *tx) + .await + .context("hosted finalization lost its recovery fence")?; + if recovery.2 != destination.organization_id + || recovery.3 != destination.guild_id + || recovery.4 != destination.provider + || recovery.5 != destination.destination_id + || recovery.6 != destination.destination_revision + || recovery.7 != destination.allowed_host + || recovery.8 != destination.object_key_prefix + || recovery.9 != destination.transient_delete_policy + { + bail!("hosted finalization destination differed from its pinned recovery snapshot"); + } + + let stopped = sqlx::query::query( + r#" +UPDATE call_scribe_capture_sessions +SET status = 'captured', stopped_at = $2, updated_at = now() +WHERE id = $1 AND organization_id = $3 +"#, + ) + .bind(recording_id) + .bind(stopped_at) + .bind(&destination.organization_id) + .execute(&mut *tx) + .await + .context("failed to persist hosted session stop")?; + if stopped.rows_affected() != 1 { + bail!("hosted capture session was missing during atomic finalization"); + } + sqlx::query::query( r#" +INSERT INTO call_scribe_audit_events + (id, organization_id, session_id, event_type, actor_kind, guild_id, metadata) +VALUES ($1, $2, $3, 'recording_stopped', 'system', $4, + '{"mode":"record_only","hosted":true}'::jsonb) +"#, + ) + .bind(Uuid::new_v4().to_string()) + .bind(&destination.organization_id) + .bind(recording_id) + .bind(&destination.guild_id) + .execute(&mut *tx) + .await + .context("failed to persist hosted recording-stop evidence")?; + + if actual_seconds > 0 { + let inserted = sqlx::query::query( + r#" INSERT INTO call_scribe_hosted_usage_outbox (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, actual_seconds, occurred_at, expires_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (reservation_id) DO NOTHING +"#, + ) + .bind(&reservation.reservation_id) + .bind(&recovery.0) + .bind(&recovery.1) + .bind(recording_id) + .bind(actual_seconds) + .bind(occurred_at) + .bind(expires_at) + .execute(&mut *tx) + .await + .context("failed to durably enqueue hosted usage")?; + if inserted.rows_affected() == 0 { + let existing: (String, i64, DateTime, DateTime) = + sqlx::query_as::query_as( + r#" +SELECT recording_id, actual_seconds, occurred_at, expires_at +FROM call_scribe_hosted_usage_outbox +WHERE reservation_id = $1 +"#, + ) + .bind(&reservation.reservation_id) + .fetch_one(&mut *tx) + .await + .context("failed to verify existing hosted usage")?; + if existing + != ( + recording_id.to_string(), + actual_seconds, + occurred_at, + expires_at, + ) + { + bail!("hosted reservation id was reused with different usage"); + } + } + } + + for manifest in manifests { + let segment_index = i32::try_from(manifest.segment_index) + .context("hosted segment index exceeded database range")?; + let content_length = i64::try_from(manifest.content_length) + .context("hosted artifact length exceeded database range")?; + let local_path = manifest + .local_path + .to_str() + .context("hosted artifact path was not valid UTF-8")?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_artifacts + (id, organization_id, session_id, kind, path, byte_size, metadata) +VALUES ($1, $2, $3, 'raw_audio_wav', $4, $5, $6) +"#, + ) + .bind(&manifest.artifact_id) + .bind(&destination.organization_id) + .bind(recording_id) + .bind(local_path) + .bind(content_length) + .bind(serde_json::json!({ + "segment_index": manifest.segment_index, + "sha256": &manifest.sha256, + "content_type": "audio/wav", + "hosted_delivery_state": "pending", + })) + .execute(&mut *tx) + .await + .context("failed to persist hosted raw-audio artifact")?; + 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) +VALUES ($1, $2, $3, $4, $5, $6, $7, 'raw_audio_wav', $8, $9, $10, $11, + 'audio/wav', $12, $13, $14, $15, $16, $17) +"#, + ) + .bind(&manifest.artifact_id) + .bind(&destination.organization_id) + .bind(&destination.guild_id) + .bind(recording_id) + .bind(&reservation.reservation_id) + .bind(&recovery.0) + .bind(&recovery.1) + .bind(segment_index) + .bind(local_path) + .bind(content_length) + .bind(&manifest.sha256) + .bind(&destination.provider) + .bind(&destination.destination_id) + .bind(&destination.destination_revision) + .bind(&destination.allowed_host) + .bind(&destination.object_key_prefix) + .bind(&destination.transient_delete_policy) + .execute(&mut *tx) + .await + .context("failed to persist hosted artifact delivery job")?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_audit_events + (id, organization_id, session_id, event_type, actor_kind, guild_id, metadata) +VALUES ($1, $2, $3, 'hosted_artifact_delivery_queued', 'system', $4, $5) +"#, + ) + .bind(Uuid::new_v4().to_string()) + .bind(&destination.organization_id) + .bind(recording_id) + .bind(&destination.guild_id) + .bind(serde_json::json!({ + "artifact_id": &manifest.artifact_id, + "artifact_kind": "raw_audio_wav", + "segment_index": manifest.segment_index, + "content_length": manifest.content_length, + "sha256": &manifest.sha256, + "storage_provider": &destination.provider, + "storage_destination_id": &destination.destination_id, + "storage_destination_revision": &destination.destination_revision, + "storage_allowed_host": &destination.allowed_host, + "storage_object_key_prefix": &destination.object_key_prefix, + "transient_delete_policy": &destination.transient_delete_policy, + })) + .execute(&mut *tx) + .await + .context("failed to persist hosted artifact delivery evidence")?; + } + + let deleted = sqlx::query::query( + r#" +DELETE FROM call_scribe_hosted_capture_recovery +WHERE reservation_id = $1 + AND recording_id = $2 + AND status = 'reconciling' + AND recovery_claim_token = $3 "#, ) .bind(&reservation.reservation_id) - .bind(encrypted_lease_token) - .bind(encryption_nonce) .bind(recording_id) - .bind(actual_seconds) - .bind(occurred_at) - .bind(expires_at) - .execute(&self.pool) + .bind(recovery_claim_token) + .execute(&mut *tx) .await - .context("failed to enqueue hosted usage reconciliation")?; - if inserted.rows_affected() == 1 { - return Ok(()); + .context("failed to clear atomically finalized hosted recovery")?; + if deleted.rows_affected() != 1 { + bail!("hosted finalization recovery fence was lost"); } + tx.commit() + .await + .context("failed to commit hosted finalization")?; + Ok(()) + } - let existing: (String, i64, DateTime, DateTime) = sqlx::query_as::query_as( + async fn finalize_zero_duration_hosted_capture_transaction( + &self, + reservation: &UsageReservation, + recording_id: &str, + recovery_claim_token: &str, + destination: &HostedStorageDestination, + cleanup_paths: &[PathBuf], + stopped_at: DateTime, + ) -> Result<()> { + if cleanup_paths.is_empty() + || cleanup_paths.len() > MAX_HOSTED_RECOVERY_WAV_SEGMENTS as usize + { + bail!("zero-duration hosted capture had an invalid cleanup set"); + } + let mut tx = self + .pool + .begin() + .await + .context("failed to begin zero-duration hosted finalization")?; + let recovery: ( + String, + String, + String, + String, + String, + String, + String, + String, + ) = sqlx::query_as::query_as( r#" -SELECT recording_id, actual_seconds, occurred_at, expires_at -FROM call_scribe_hosted_usage_outbox +SELECT organization_id, guild_id, storage_provider, storage_destination_id, + storage_destination_revision, storage_allowed_host, + storage_object_key_prefix, transient_delete_policy +FROM call_scribe_hosted_capture_recovery WHERE reservation_id = $1 + AND recording_id = $2 + AND status = 'reconciling' + AND recovery_claim_token = $3 +FOR UPDATE "#, ) .bind(&reservation.reservation_id) - .fetch_one(&self.pool) + .bind(recording_id) + .bind(recovery_claim_token) + .fetch_one(&mut *tx) .await - .context("failed to verify existing hosted usage reconciliation")?; - if existing + .context("zero-duration finalization lost its recovery fence")?; + if recovery != ( - recording_id.to_string(), - actual_seconds, - occurred_at, - expires_at, + destination.organization_id.clone(), + destination.guild_id.clone(), + destination.provider.clone(), + destination.destination_id.clone(), + destination.destination_revision.clone(), + destination.allowed_host.clone(), + destination.object_key_prefix.clone(), + destination.transient_delete_policy.clone(), ) { - bail!("hosted reservation id was reused with different usage"); + bail!("zero-duration finalization destination differed from its pinned snapshot"); + } + + let stopped = sqlx::query::query( + r#" +UPDATE call_scribe_capture_sessions +SET status = 'captured', stopped_at = $2, updated_at = now() +WHERE id = $1 AND organization_id = $3 +"#, + ) + .bind(recording_id) + .bind(stopped_at) + .bind(&destination.organization_id) + .execute(&mut *tx) + .await + .context("failed to close zero-duration hosted session")?; + if stopped.rows_affected() != 1 { + bail!("zero-duration hosted session was missing during finalization"); + } + sqlx::query::query( + r#" +INSERT INTO call_scribe_audit_events + (id, organization_id, session_id, event_type, actor_kind, guild_id, metadata) +VALUES ($1, $2, $3, 'recording_stopped', 'system', $4, + '{"mode":"record_only","hosted":true,"zero_duration":true}'::jsonb) +"#, + ) + .bind(Uuid::new_v4().to_string()) + .bind(&destination.organization_id) + .bind(recording_id) + .bind(&destination.guild_id) + .execute(&mut *tx) + .await + .context("failed to persist zero-duration stop evidence")?; + for path in cleanup_paths { + let path = path + .to_str() + .context("zero-duration cleanup path was not valid UTF-8")?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_zero_duration_cleanup_outbox + (local_path, organization_id, recording_id) +VALUES ($1, $2, $3) +ON CONFLICT (local_path) DO NOTHING +"#, + ) + .bind(path) + .bind(&destination.organization_id) + .bind(recording_id) + .execute(&mut *tx) + .await + .context("failed to queue zero-duration local cleanup")?; + } + let deleted = sqlx::query::query( + r#" +DELETE FROM call_scribe_hosted_capture_recovery +WHERE reservation_id = $1 AND recording_id = $2 + AND status = 'reconciling' AND recovery_claim_token = $3 +"#, + ) + .bind(&reservation.reservation_id) + .bind(recording_id) + .bind(recovery_claim_token) + .execute(&mut *tx) + .await + .context("failed to clear zero-duration recovery")?; + if deleted.rows_affected() != 1 { + bail!("zero-duration finalization recovery fence was lost"); } + tx.commit() + .await + .context("failed to commit zero-duration hosted finalization")?; + Ok(()) + } + + async fn finish_zero_duration_hosted_cleanup(&self, capture_dir: &Path) -> Result<()> { + let claim_token = Uuid::new_v4().to_string(); + let rows: Vec<(String,)> = sqlx::query_as::query_as( + r#" +WITH candidates AS ( + SELECT local_path + FROM call_scribe_hosted_zero_duration_cleanup_outbox + WHERE next_attempt_at <= now() + AND (claim_token IS NULL OR claim_until <= now()) + ORDER BY next_attempt_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT 25 +) +UPDATE call_scribe_hosted_zero_duration_cleanup_outbox AS cleanup +SET claim_token = $1, claim_until = now() + interval '5 minutes', + attempt_count = attempt_count + 1, updated_at = now() +FROM candidates +WHERE cleanup.local_path = candidates.local_path +RETURNING cleanup.local_path +"#, + ) + .bind(&claim_token) + .fetch_all(&self.pool) + .await + .context("failed to claim zero-duration cleanup")?; + for (local_path,) in rows { + let path = PathBuf::from(&local_path); + let cleanup_result = remove_zero_duration_wav(&path, capture_dir).await; + match cleanup_result { + Ok(()) => { + sqlx::query::query( + "DELETE FROM call_scribe_hosted_zero_duration_cleanup_outbox WHERE local_path = $1 AND claim_token = $2", + ) + .bind(&local_path) + .bind(&claim_token) + .execute(&self.pool) + .await + .context("failed to complete zero-duration cleanup")?; + } + Err(err) => { + let message: String = format!("{err:#}").chars().take(500).collect(); + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_zero_duration_cleanup_outbox +SET claim_token = NULL, claim_until = NULL, + next_attempt_at = now() + interval '30 seconds', last_error = $3, updated_at = now() +WHERE local_path = $1 AND claim_token = $2 +"#, + ) + .bind(&local_path) + .bind(&claim_token) + .bind(message) + .execute(&self.pool) + .await + .context("failed to defer zero-duration cleanup")?; + } + } + } + Ok(()) + } + + async fn has_unsafe_hosted_deliveries( + &self, + organization_id: &str, + guild_id: &str, + maximum_age: Duration, + ) -> Result { + let maximum_age = chrono::Duration::from_std(maximum_age) + .context("hosted delivery backpressure age exceeded timestamp range")?; + let unsafe_before = Utc::now() - maximum_age; + sqlx::query_scalar::query_scalar( + r#" +SELECT EXISTS ( + SELECT 1 + FROM call_scribe_hosted_artifact_delivery_outbox + WHERE organization_id = $1 + AND guild_id = $2 + AND status <> 'delivered' + AND created_at <= $3 +) +"#, + ) + .bind(organization_id) + .bind(guild_id) + .bind(unsafe_before) + .fetch_one(&self.pool) + .await + .context("failed to check hosted delivery backpressure") + } + + async fn retry_hosted_artifact_delivery_outbox( + &self, + client: &HostedControlPlaneClient, + capture_dir: &Path, + claim_owner: &str, + ) -> Result<()> { + self.finish_zero_duration_hosted_cleanup(capture_dir) + .await?; + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'failed', encrypted_lease_token = NULL, encryption_nonce = NULL, + 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') + AND attempt_count >= $1 + AND (status = 'pending' OR claim_until <= now()) +"#, + ) + .bind(HOSTED_DELIVERY_MAX_ATTEMPTS) + .execute(&self.pool) + .await + .context("failed to terminally fence exhausted hosted deliveries")?; + + self.finish_verified_hosted_artifact_deletions(capture_dir) + .await?; + + let claim_token = Uuid::new_v4().to_string(); + let claim_seconds = i64::try_from(HOSTED_DELIVERY_CLAIM_TTL.as_secs()) + .expect("hosted delivery claim TTL must fit i64"); + let jobs: Vec = sqlx::query_as::query_as( + r#" +WITH candidates AS ( + SELECT artifact_id + FROM call_scribe_hosted_artifact_delivery_outbox + WHERE (status = 'pending' OR (status = 'in_progress' AND claim_until <= now())) + AND next_attempt_at <= now() + AND attempt_count < $1 + ORDER BY next_attempt_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT 10 +) +UPDATE call_scribe_hosted_artifact_delivery_outbox AS delivery +SET status = 'in_progress', claim_owner = $2, claim_token = $3, + claim_until = now() + ($4 * interval '1 second'), + attempt_count = attempt_count + 1, + first_attempt_at = COALESCE(first_attempt_at, now()), + updated_at = now() +FROM candidates +WHERE delivery.artifact_id = candidates.artifact_id +RETURNING delivery.artifact_id, delivery.recording_id, delivery.reservation_id, + delivery.encrypted_lease_token, delivery.encryption_nonce, + 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.storage_object_key_prefix, delivery.transient_delete_policy, + delivery.operation_id, + delivery.operation_generation, delivery.operation_object_key +"#, + ) + .bind(HOSTED_DELIVERY_MAX_ATTEMPTS) + .bind(claim_owner) + .bind(&claim_token) + .bind(claim_seconds) + .fetch_all(&self.pool) + .await + .context("failed to claim hosted artifact deliveries")?; + + for job in jobs { + let artifact_id = job.artifact_id.clone(); + if let Err(err) = self + .deliver_claimed_hosted_artifact(client, capture_dir, &claim_token, job) + .await + { + eprintln!("hosted artifact {artifact_id} delivery remains pending: {err:#}"); + self.defer_claimed_hosted_artifact(&artifact_id, &claim_token, &err) + .await?; + } + } + self.finish_verified_hosted_artifact_deletions(capture_dir) + .await + } + + async fn deliver_claimed_hosted_artifact( + &self, + client: &HostedControlPlaneClient, + capture_dir: &Path, + claim_token: &str, + job: HostedArtifactDeliveryRow, + ) -> Result<()> { + let HostedArtifactDeliveryRow { + artifact_id, + recording_id, + reservation_id, + encrypted_lease_token, + encryption_nonce, + segment_index, + local_path, + content_length, + sha256, + storage_provider, + storage_destination_id: destination_id, + storage_destination_revision: destination_revision, + storage_allowed_host: allowed_host, + storage_object_key_prefix: object_key_prefix, + transient_delete_policy, + operation_id: persisted_operation_id, + operation_generation: persisted_operation_generation, + operation_object_key: persisted_operation_object_key, + } = job; + let local_path = PathBuf::from(local_path); + if !local_path.starts_with(capture_dir) { + bail!("hosted delivery artifact escaped the configured capture directory"); + } + let lease_token = client.decrypt_reservation_lease( + &reservation_id, + &encrypted_lease_token, + &encryption_nonce, + )?; + let content_length = + u64::try_from(content_length).context("hosted artifact length was invalid")?; + let segment_index = + u32::try_from(segment_index).context("hosted artifact segment index was invalid")?; + let destination = HostedStorageDestination { + // Organization and guild are not sent to the prepare endpoint; the + // control plane derives them from reservation authority. + organization_id: String::new(), + guild_id: String::new(), + provider: storage_provider, + destination_id, + destination_revision, + allowed_host, + object_key_prefix, + transient_delete_policy, + }; + if destination.transient_delete_policy != "delete_after_verified_delivery" { + bail!("hosted artifact had an unsupported transient deletion policy"); + } + let request = ArtifactDeliveryPrepareRequest { + reservation_id: &reservation_id, + lease_token: &lease_token, + recording_id: &recording_id, + artifact_id: &artifact_id, + artifact_kind: "raw_audio_wav", + segment_index, + content_length, + sha256: &sha256, + content_type: "audio/wav", + }; + let persisted_operation = match ( + persisted_operation_id, + persisted_operation_generation, + persisted_operation_object_key, + ) { + (Some(operation_id), Some(generation), Some(object_key)) => { + Some(ArtifactDeliveryOperationRef { + operation_id, + generation: u64::try_from(generation) + .context("persisted hosted delivery generation was invalid")?, + recording_id: recording_id.clone(), + artifact_id: artifact_id.clone(), + artifact_kind: "raw_audio_wav".to_string(), + segment_index, + object_key, + destination_id: destination.destination_id.clone(), + destination_revision: destination.destination_revision.clone(), + provider: destination.provider.clone(), + allowed_upload_host: destination.allowed_host.clone(), + }) + } + (None, None, None) => None, + _ => bail!("hosted delivery persisted only part of its operation fence"), + }; + if let Some(operation) = &persisted_operation + && let ArtifactDeliveryVerification::Verified(receipt) = client + .verify_artifact_delivery(operation, &request, &destination) + .await? + { + return self + .mark_hosted_artifact_verified(&artifact_id, claim_token, operation, &receipt) + .await; + } + + let prepared = client + .prepare_artifact_delivery(&request, &destination) + .await?; + let operation = ArtifactDeliveryOperationRef::from(&prepared); + if let Some(persisted) = &persisted_operation + && operation != *persisted + { + bail!("hosted control plane replaced an already-persisted delivery operation"); + } + let generation = i64::try_from(operation.generation) + .context("hosted delivery generation exceeded database range")?; + let persisted = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET operation_id = $3, operation_generation = $4, operation_object_key = $5, + updated_at = now() +WHERE artifact_id = $1 AND status = 'in_progress' AND claim_token = $2 + AND (operation_id IS NULL OR (operation_id = $3 AND operation_generation = $4 + AND operation_object_key = $5)) +"#, + ) + .bind(&artifact_id) + .bind(claim_token) + .bind(&operation.operation_id) + .bind(generation) + .bind(&operation.object_key) + .execute(&self.pool) + .await + .context("failed to persist hosted delivery operation fence")?; + if persisted.rows_affected() != 1 { + bail!("hosted artifact delivery claim fence was lost before upload"); + } + client + .upload_artifact(&prepared, &request, &local_path, &destination) + .await?; + let receipt = client + .verify_artifact_delivery(&operation, &request, &destination) + .await?; + match receipt { + ArtifactDeliveryVerification::Verified(receipt) => { + self.mark_hosted_artifact_verified(&artifact_id, claim_token, &operation, &receipt) + .await + } + ArtifactDeliveryVerification::NotReady => { + bail!("control plane could not independently verify the uploaded artifact") + } + } + } + + async fn mark_hosted_artifact_verified( + &self, + artifact_id: &str, + claim_token: &str, + operation: &ArtifactDeliveryOperationRef, + receipt: &ArtifactDeliveryReceipt, + ) -> Result<()> { + let verified_at = receipt.verified_at; + let receipt_json = + serde_json::to_value(receipt).context("failed to encode delivery receipt")?; + let verified = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'verified', encrypted_lease_token = NULL, encryption_nonce = NULL, + receipt = $3, verified_at = $4, claim_owner = NULL, claim_token = NULL, + claim_until = NULL, last_error = NULL, updated_at = now() +WHERE artifact_id = $1 AND status = 'in_progress' AND claim_token = $2 + AND operation_id = $5 AND operation_generation = $6 +"#, + ) + .bind(artifact_id) + .bind(claim_token) + .bind(receipt_json) + .bind(verified_at) + .bind(&operation.operation_id) + .bind(i64::try_from(operation.generation).context("delivery generation overflowed")?) + .execute(&self.pool) + .await + .context("failed to persist verified hosted delivery receipt")?; + if verified.rows_affected() != 1 { + bail!("hosted artifact delivery claim fence was lost before verification"); + } + Ok(()) + } + + async fn defer_claimed_hosted_artifact( + &self, + artifact_id: &str, + claim_token: &str, + error: &anyhow::Error, + ) -> Result<()> { + let message: String = format!("{error:#}").chars().take(500).collect(); + let deferred = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'pending', 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_progress' AND claim_token = $2 +"#, + ) + .bind(artifact_id) + .bind(claim_token) + .bind(message) + .execute(&self.pool) + .await + .context("failed to defer hosted artifact delivery")?; + if deferred.rows_affected() != 1 { + bail!("hosted artifact delivery claim fence was lost before retry"); + } + Ok(()) + } + + async fn finish_verified_hosted_artifact_deletions(&self, capture_dir: &Path) -> Result<()> { + let deletion_owner = Uuid::new_v4().to_string(); + let deletion_claim_token = Uuid::new_v4().to_string(); + let verified: Vec<(String, String)> = sqlx::query_as::query_as( + r#" +WITH candidates AS ( + SELECT artifact_id + FROM call_scribe_hosted_artifact_delivery_outbox + WHERE status = 'verified' + ORDER BY verified_at + FOR UPDATE SKIP LOCKED + LIMIT 25 +) +UPDATE call_scribe_hosted_artifact_delivery_outbox AS delivery +SET claim_owner = $1, claim_token = $2, claim_until = now() + interval '5 minutes', + updated_at = now() +FROM candidates +WHERE delivery.artifact_id = candidates.artifact_id +RETURNING delivery.artifact_id, delivery.local_path +"#, + ) + .bind(&deletion_owner) + .bind(&deletion_claim_token) + .fetch_all(&self.pool) + .await + .context("failed to list verified hosted artifact deletions")?; + for (artifact_id, local_path) in verified { + let local_path = PathBuf::from(local_path); + if !local_path.starts_with(capture_dir) { + eprintln!( + "hosted artifact {artifact_id} deletion denied outside capture directory" + ); + self.release_verified_deletion_claim(&artifact_id, &deletion_claim_token) + .await?; + continue; + } + match tokio::fs::remove_file(&local_path).await { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => { + eprintln!( + "hosted artifact {artifact_id} local deletion remains pending: {err}" + ); + self.release_verified_deletion_claim(&artifact_id, &deletion_claim_token) + .await?; + continue; + } + } + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'delivered', local_deleted_at = now(), completed_at = now(), + claim_owner = NULL, claim_token = NULL, claim_until = NULL, updated_at = now() +WHERE artifact_id = $1 AND status = 'verified' AND claim_token = $2 +"#, + ) + .bind(&artifact_id) + .bind(&deletion_claim_token) + .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 +"#, + ) + .bind(&artifact_id) + .execute(&self.pool) + .await + .context("failed to update hosted artifact delivery metadata")?; + } + Ok(()) + } + + async fn release_verified_deletion_claim( + &self, + artifact_id: &str, + claim_token: &str, + ) -> Result<()> { + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET claim_owner = NULL, claim_token = NULL, claim_until = NULL, updated_at = now() +WHERE artifact_id = $1 AND status = 'verified' AND claim_token = $2 +"#, + ) + .bind(artifact_id) + .bind(claim_token) + .execute(&self.pool) + .await + .context("failed to release hosted artifact deletion claim")?; + Ok(()) + } + + #[cfg(test)] + async fn enqueue_hosted_usage( + &self, + client: &HostedControlPlaneClient, + reservation: &UsageReservation, + recording_id: &str, + actual_seconds: u64, + occurred_at: DateTime, + ) -> Result<()> { + let actual_seconds = i64::try_from(actual_seconds) + .context("hosted usage duration exceeded database range")?; + let authorization_expires_at = DateTime::parse_from_rfc3339(&reservation.expires_at) + .context("hosted usage expiry was invalid")? + .with_timezone(&Utc); + if authorization_expires_at < occurred_at { + bail!("hosted usage exceeded its capture authority"); + } + let expires_at = authorization_expires_at + + chrono::Duration::from_std(RESERVATION_SETTLEMENT_GRACE) + .expect("reservation settlement grace must fit chrono duration"); + let (encrypted_lease_token, encryption_nonce) = client + .encrypt_reservation_lease(&reservation.reservation_id, &reservation.lease_token)?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_usage_outbox + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + actual_seconds, occurred_at, expires_at) +VALUES ($1, $2, $3, $4, $5, $6, $7) +"#, + ) + .bind(&reservation.reservation_id) + .bind(encrypted_lease_token) + .bind(encryption_nonce) + .bind(recording_id) + .bind(actual_seconds) + .bind(occurred_at) + .bind(expires_at) + .execute(&self.pool) + .await + .context("failed to enqueue hosted usage reconciliation")?; Ok(()) } @@ -1670,6 +2668,7 @@ struct SegmentedWavRecorder { paths: Vec, writer: hound::WavWriter>, data_bytes_written: u32, + capture_source_stems: bool, source_paths: HashMap, source_writers: HashMap>>, } @@ -2471,51 +3470,72 @@ impl DiscordCaptureHandler { ) -> Result<()> { self.stop_capture(ctx, guild_id).await?; - let (organization_id, capture_mode, hosted_generation, hosted_reservation_request) = - if let Some(hosted) = &self.config.hosted { - let policy = hosted - .configurations - .policy_for(guild_id.get()) - .context("hosted configuration is missing or stale")?; - if !policy.permits_recording(channel_id.get()) { - bail!("hosted configuration does not authorize this guild and channel"); - } - let requested = self - .requested_channels - .get(&guild_id) - .map(|requested| requested.clone()) - .context("hosted start has no durable requested state")?; - if requested.channel_id != channel_id - || requested.generation != policy.desired_recording_generation - { - bail!("hosted start generation does not match current desired state"); - } - let requested_seconds = policy - .remaining_recording_seconds - .context("hosted remaining usage is not configured")? - .min(3_600); - ( - policy.organization_id, - CaptureMode::RecordOnly, - Some(requested.generation), - Some(( - hosted.client.clone(), - requested.command_id, - requested_seconds, - )), - ) - } else { - ( - self.config.self_hosted_organization_id.clone(), - self.config.self_hosted_capture_mode, - None, - None, - ) - }; + let ( + organization_id, + capture_mode, + hosted_generation, + hosted_reservation_request, + hosted_storage, + ) = if let Some(hosted) = &self.config.hosted { + let policy = hosted + .configurations + .policy_for(guild_id.get()) + .context("hosted configuration is missing or stale")?; + if !policy.permits_recording(channel_id.get()) { + bail!("hosted configuration does not authorize this guild and channel"); + } + let requested = self + .requested_channels + .get(&guild_id) + .map(|requested| requested.clone()) + .context("hosted start has no durable requested state")?; + if requested.channel_id != channel_id + || requested.generation != policy.desired_recording_generation + { + bail!("hosted start generation does not match current desired state"); + } + let requested_seconds = policy + .remaining_recording_seconds + .context("hosted remaining usage is not configured")? + .min(3_600); + let destination = policy + .storage_destination() + .context("hosted storage destination is incomplete or unsupported")?; + ( + policy.organization_id.clone(), + CaptureMode::RecordOnly, + Some(requested.generation), + Some(( + hosted.client.clone(), + requested.command_id, + requested_seconds, + )), + Some(destination), + ) + } else { + ( + self.config.self_hosted_organization_id.clone(), + self.config.self_hosted_capture_mode, + None, + None, + None, + ) + }; let runtime_store = match &self.config.runtime_store { Some(store) => Some(store.scoped(organization_id, capture_mode).await?), None => None, }; + if let (Some(store), Some(destination)) = (&runtime_store, &hosted_storage) + && store + .has_unsafe_hosted_deliveries( + &destination.organization_id, + &destination.guild_id, + HOSTED_DELIVERY_BACKPRESSURE_AGE, + ) + .await? + { + bail!("hosted recording is backpressured by an overdue unsafe local artifact delivery"); + } let session_id = Uuid::new_v4().to_string(); let started_at = Local::now(); @@ -2527,7 +3547,10 @@ impl DiscordCaptureHandler { channel_id.get(), session_id, )); - let recorder = create_wav_recorder(&base_wav_path)?; + // Hosted delivery intentionally records only the mixed artifact. This + // makes the durable manifest exhaustive and prevents participant stems + // from falling outside delivery, deletion, and retention accounting. + let recorder = create_wav_recorder(&base_wav_path, hosted_storage.is_none())?; let known_ssrcs = Arc::new(DashMap::new()); let voice_stats = Arc::new(DiscordVoiceStats::default()); let manager = songbird::get(ctx) @@ -2571,6 +3594,9 @@ impl DiscordCaptureHandler { &base_wav_path, started_at, &self.recovery_owner_id, + hosted_storage + .as_ref() + .context("hosted capture omitted its pinned storage destination")?, ) .await { @@ -2791,6 +3817,7 @@ impl DiscordCaptureHandler { runtime_store, capture_mode, hosted_usage, + hosted_storage, hosted_generation, }, ); @@ -2887,38 +3914,75 @@ impl DiscordCaptureHandler { .release_usage(&settlement_reservation) .await .context("failed to release zero-duration hosted capture")?; + let destination = active + .hosted_storage + .as_ref() + .context("hosted capture omitted its pinned storage destination")?; + store + .finalize_zero_duration_hosted_capture_transaction( + &settlement_reservation, + &active.session_id, + &recovery_claim_token, + destination, + &wav_paths, + stopped_at.with_timezone(&Utc), + ) + .await + .context("failed to finalize zero-duration hosted capture")?; + if let Err(err) = store + .finish_zero_duration_hosted_cleanup(&self.config.capture_dir) + .await + { + eprintln!( + "zero-duration hosted cleanup remains durably queued for recording {}: {err:#}", + active.session_id + ); + } } else { + let manifests = hosted_artifact_manifests(&wav_paths).await?; let occurred_at = active.started_at.with_timezone(&Utc) + chrono::Duration::seconds( i64::try_from(actual_seconds) .context("hosted usage duration exceeded timestamp range")?, ); + let destination = active + .hosted_storage + .as_ref() + .context("hosted capture omitted its pinned storage destination")?; store - .enqueue_hosted_usage( - client, + .finalize_hosted_capture_transaction( &settlement_reservation, &active.session_id, + &recovery_claim_token, actual_seconds, occurred_at, + destination, + &manifests, + stopped_at.with_timezone(&Utc), ) .await - .context("failed to durably enqueue hosted usage")?; + .context("failed to atomically finalize hosted artifacts and delivery jobs")?; } - store - .remove_claimed_hosted_capture_recovery( - &reservation.reservation_id, - &active.session_id, - &recovery_claim_token, - ) - .await?; if let Err(err) = store.retry_hosted_usage_outbox(client).await { eprintln!( "hosted usage for recording {} remains queued for retry: {err:#}", active.session_id ); } - } - if let Some(store) = &active.runtime_store { + if let Err(err) = store + .retry_hosted_artifact_delivery_outbox( + client, + &self.config.capture_dir, + &self.recovery_owner_id, + ) + .await + { + eprintln!( + "hosted artifacts for recording {} remain queued for retry: {err:#}", + active.session_id + ); + } + } else if let Some(store) = &active.runtime_store { if let Err(err) = store .record_session_stopped(&active.session_id, stopped_at) .await @@ -3210,6 +4274,16 @@ impl DiscordCaptureHandler { if let Err(err) = store.retry_hosted_usage_outbox(&hosted.client).await { eprintln!("hosted usage outbox retry failed: {err:#}"); } + if let Err(err) = store + .retry_hosted_artifact_delivery_outbox( + &hosted.client, + &self.config.capture_dir, + &self.recovery_owner_id, + ) + .await + { + eprintln!("hosted artifact delivery retry failed: {err:#}"); + } } let mut reconcile_guilds = std::collections::HashSet::new(); @@ -3750,8 +4824,8 @@ impl DiscordVoiceStats { } #[cfg(feature = "discord")] -fn create_wav_recorder(path: &Path) -> Result { - let recorder = SegmentedWavRecorder::new(path)?; +fn create_wav_recorder(path: &Path, capture_source_stems: bool) -> Result { + let recorder = SegmentedWavRecorder::new(path, capture_source_stems)?; Ok(Arc::new(Mutex::new(Some(recorder)))) } @@ -3821,6 +4895,36 @@ fn recovered_usage_seconds(base_path: &Path, reserved_seconds: u64) -> Result Result<()> { + if !path.starts_with(capture_dir) { + bail!("zero-duration cleanup escaped the configured capture directory"); + } + let metadata = match tokio::fs::symlink_metadata(path).await { + Ok(metadata) => metadata, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(err) => return Err(err).context("failed to inspect zero-duration cleanup path"), + }; + if metadata.file_type().is_symlink() || !metadata.is_file() { + bail!("zero-duration cleanup target was not a regular worker-owned file"); + } + let owned_path = path.to_path_buf(); + let duration = tokio::task::spawn_blocking(move || { + let reader = hound::WavReader::open(&owned_path) + .with_context(|| format!("failed to read cleanup WAV {}", owned_path.display()))?; + Ok::<_, anyhow::Error>(reader.duration()) + }) + .await + .context("zero-duration WAV validation task failed")??; + if duration != 0 { + bail!("zero-duration cleanup refused to delete a WAV containing audio samples"); + } + tokio::fs::remove_file(path) + .await + .with_context(|| format!("failed to remove zero-duration WAV {}", path.display()))?; + Ok(()) +} + #[cfg(feature = "discord")] fn authorized_usage_seconds( started_at: DateTime, @@ -3845,9 +4949,59 @@ fn hosted_authority_requires_fence(expires_at: DateTime, now: DateTime .expect("reservation expiry margin must fit chrono duration") } +#[cfg(feature = "discord")] +async fn hosted_artifact_manifests(paths: &[PathBuf]) -> Result> { + if paths.is_empty() || paths.len() > MAX_HOSTED_RECOVERY_WAV_SEGMENTS as usize { + bail!("hosted raw-audio artifact count was outside the supported bound"); + } + let paths = paths.to_vec(); + tokio::task::spawn_blocking(move || { + paths + .into_iter() + .enumerate() + .map(|(index, path)| { + let symlink_metadata = std::fs::symlink_metadata(&path).with_context(|| { + format!("failed to inspect hosted artifact {}", path.display()) + })?; + if symlink_metadata.file_type().is_symlink() || !symlink_metadata.is_file() { + bail!("hosted artifact was not a regular worker-owned file"); + } + let content_length = symlink_metadata.len(); + if content_length == 0 { + bail!("hosted artifact was empty"); + } + let mut file = File::open(&path).with_context(|| { + format!("failed to open hosted artifact {}", path.display()) + })?; + let mut digest = Sha256::new(); + let mut buffer = vec![0_u8; 1024 * 1024]; + loop { + let read = file.read(&mut buffer).with_context(|| { + format!("failed to hash hosted artifact {}", path.display()) + })?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + Ok(HostedArtifactManifest { + artifact_id: Uuid::new_v4().to_string(), + segment_index: u32::try_from(index + 1) + .context("hosted segment index exceeded u32")?, + local_path: path, + content_length, + sha256: format!("{:x}", digest.finalize()), + }) + }) + .collect::>>() + }) + .await + .context("hosted artifact hashing task failed")? +} + #[cfg(feature = "discord")] impl SegmentedWavRecorder { - fn new(base_path: &Path) -> Result { + fn new(base_path: &Path, capture_source_stems: bool) -> Result { let writer = hound::WavWriter::create(base_path, discord_wav_spec()) .with_context(|| format!("failed to create {}", base_path.display()))?; Ok(Self { @@ -3856,6 +5010,7 @@ impl SegmentedWavRecorder { paths: vec![base_path.to_path_buf()], writer, data_bytes_written: 0, + capture_source_stems, source_paths: HashMap::new(), source_writers: HashMap::new(), }) @@ -3873,19 +5028,21 @@ impl SegmentedWavRecorder { self.write_sample(clamped as i16)?; } - for (ssrc, _) in source_frames { - self.ensure_source_writer(*ssrc)?; - } + if self.capture_source_stems { + for (ssrc, _) in source_frames { + self.ensure_source_writer(*ssrc)?; + } - let source_by_ssrc: HashMap = source_frames.iter().copied().collect(); - for (ssrc, source_writer) in &mut self.source_writers { - let source_samples = source_by_ssrc.get(ssrc).copied(); - for idx in 0..frame_sample_count { - let sample = source_samples - .and_then(|samples| samples.get(idx)) - .copied() - .unwrap_or_default(); - source_writer.write_sample(sample)?; + let source_by_ssrc: HashMap = source_frames.iter().copied().collect(); + for (ssrc, source_writer) in &mut self.source_writers { + let source_samples = source_by_ssrc.get(ssrc).copied(); + for idx in 0..frame_sample_count { + let sample = source_samples + .and_then(|samples| samples.get(idx)) + .copied() + .unwrap_or_default(); + source_writer.write_sample(sample)?; + } } } @@ -4911,6 +6068,11 @@ mod tests { remaining_recording_seconds: Some(300), storage_provider: Some("customer_s3".to_string()), storage_destination_label: Some("worker-pvc".to_string()), + storage_destination_id: Some("dst_01".to_string()), + storage_destination_revision: Some("rev_01".to_string()), + storage_allowed_host: Some("bucket.s3.us-east-1.amazonaws.com".to_string()), + storage_object_key_prefix: Some("objects/".to_string()), + transient_delete_policy: Some("delete_after_verified_delivery".to_string()), ready: true, blocked_reasons: Vec::new(), desired_recording_generation: 1, @@ -5041,7 +6203,7 @@ mod tests { #[cfg(feature = "discord")] #[test] - fn hosted_mode_stays_closed_after_command_until_storage_adapter_exists() { + fn hosted_mode_opens_only_after_durable_command_and_complete_storage_contract() { let guild_id = GuildId::new(1); let channel_id = ChannelId::new(2); let participant_user_id = UserId::new(4); @@ -5059,7 +6221,7 @@ mod tests { command_id: "cmd-1".to_string(), }, ); - assert_eq!(handler.desired_capture_channel(guild_id), None); + assert_eq!(handler.desired_capture_channel(guild_id), Some(channel_id)); } #[cfg(feature = "discord")] @@ -5111,6 +6273,29 @@ mod tests { assert_eq!(handler.desired_capture_channel(guild_id), None); } + #[cfg(feature = "discord")] + #[test] + fn hosted_recorder_disables_untracked_participant_stems() -> Result<()> { + let test_dir = std::env::temp_dir().join(format!("call-scribe-stems-{}", Uuid::new_v4())); + std::fs::create_dir_all(&test_dir)?; + let hosted_path = test_dir.join("hosted.wav"); + let mut hosted = SegmentedWavRecorder::new(&hosted_path, false)?; + hosted.write_tick(&[1, 1], &[(42, &[1, 1])], 2)?; + hosted.finalize()?; + assert!(!source_wav_path(&hosted_path, 42).exists()); + + let self_hosted_path = test_dir.join("self-hosted.wav"); + let mut self_hosted = SegmentedWavRecorder::new(&self_hosted_path, true)?; + self_hosted.write_tick(&[1, 1], &[(42, &[1, 1])], 2)?; + self_hosted.finalize()?; + assert!( + source_wav_path(&self_hosted_path, 42).exists(), + "self-hosted source-stem behavior must remain available" + ); + std::fs::remove_dir_all(test_dir)?; + Ok(()) + } + #[cfg(feature = "discord")] #[test] fn stale_stop_generation_cannot_clear_newer_requested_start() { @@ -5344,10 +6529,15 @@ VALUES ('reservation-invalid', NULL, NULL, 'recording-invalid', r#" INSERT INTO call_scribe_hosted_capture_recovery (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, - base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id) + base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id, + organization_id, guild_id, storage_provider, storage_destination_id, + storage_destination_revision, storage_allowed_host, + storage_object_key_prefix, transient_delete_policy) VALUES ('reservation-live', '\x01', '\x02', 'recording-live', '/captures/live.wav', 300, now() - interval '1 minute', - now() + interval '1 hour', 'instance-a') + now() + interval '1 hour', 'instance-a', 'org-test', '1', + 'customer_s3', 'dst-test', 'rev-1', 'bucket.s3.us-east-1.amazonaws.com', + 'objects/', 'delete_after_verified_delivery') "#, ) .execute(&first) @@ -5453,6 +6643,439 @@ WHERE reservation_id = 'reservation-live' Ok(()) } + #[cfg(feature = "discord")] + #[tokio::test] + async fn hosted_finalization_atomically_hands_raw_audio_to_one_fenced_delivery_owner() + -> Result<()> { + let Some((admin, first, second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping Postgres delivery proof"); + return Ok(()); + }; + migrate_runtime_schema(&first).await?; + let store = SqlxRuntimeStore { + pool: first.clone(), + organization_id: "org-delivery-test".to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + store.ensure_organization().await?; + let client = HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-token-with-at-least-thirty-two-bytes".to_string(), + "delivery-test-worker".to_string(), + "test-outbox-key-with-at-least-thirty-two-bytes".to_string(), + )?; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let destination = HostedStorageDestination { + organization_id: "org-delivery-test".to_string(), + guild_id: "1".to_string(), + provider: "customer_s3".to_string(), + destination_id: "dst-test".to_string(), + destination_revision: "rev-1".to_string(), + allowed_host: "127.0.0.1".to_string(), + object_key_prefix: "objects/".to_string(), + transient_delete_policy: "delete_after_verified_delivery".to_string(), + }; + let reservation = UsageReservation { + reservation_id: "reservation-delivery".to_string(), + lease_token: "opaque-delivery-lease-token".to_string(), + reserved_seconds: 300, + expires_at: (Utc::now() + chrono::Duration::hours(1)).to_rfc3339(), + }; + let test_dir = + std::env::temp_dir().join(format!("call-scribe-delivery-{}", Uuid::new_v4())); + std::fs::create_dir_all(&test_dir)?; + let wav_path = test_dir.join("capture.wav"); + let mut writer = hound::WavWriter::create(&wav_path, discord_wav_spec())?; + for _ in 0..(u64::from(DISCORD_SAMPLE_RATE) * u64::from(DISCORD_CHANNELS)) { + writer.write_sample(0_i16)?; + } + writer.finalize()?; + let started_at = Local::now(); + store + .persist_hosted_capture_recovery( + &client, + &reservation, + "recording-delivery", + &wav_path, + started_at, + "instance-a", + &destination, + ) + .await?; + store + .record_session_started( + "recording-delivery", + GuildId::new(1), + ChannelId::new(2), + started_at, + "Delivery proof", + serde_json::json!({"hosted": true}), + ) + .await?; + let (recovery_claim, authorization_end) = store + .claim_live_hosted_capture_finalization( + &reservation.reservation_id, + "recording-delivery", + "instance-a", + Utc::now() + chrono::Duration::seconds(5), + ) + .await?; + let mut settlement_reservation = reservation.clone(); + settlement_reservation.expires_at = authorization_end.to_rfc3339(); + let manifests = hosted_artifact_manifests(std::slice::from_ref(&wav_path)).await?; + store + .finalize_hosted_capture_transaction( + &settlement_reservation, + "recording-delivery", + &recovery_claim, + 1, + Utc::now(), + &destination, + &manifests, + Utc::now(), + ) + .await?; + + let counts: (i64, i64, i64, i64) = sqlx::query_as::query_as( + r#" +SELECT + (SELECT count(*) FROM call_scribe_hosted_capture_recovery), + (SELECT count(*) FROM call_scribe_artifacts WHERE session_id = 'recording-delivery'), + (SELECT count(*) FROM call_scribe_hosted_artifact_delivery_outbox + WHERE recording_id = 'recording-delivery' AND status = 'pending'), + (SELECT count(*) FROM call_scribe_hosted_usage_outbox + WHERE recording_id = 'recording-delivery' AND status = 'pending') +"#, + ) + .fetch_one(&first) + .await?; + assert_eq!(counts, (0, 1, 1, 1)); + assert!( + wav_path.exists(), + "local WAV must remain until a verified receipt" + ); + let manifest_row: (String, String, String, String, i64, String) = sqlx::query_as::query_as( + r#" +SELECT organization_id, guild_id, storage_destination_id, + storage_destination_revision, content_length, sha256 +FROM call_scribe_hosted_artifact_delivery_outbox +WHERE recording_id = 'recording-delivery' +"#, + ) + .fetch_one(&first) + .await?; + assert_eq!(manifest_row.0, destination.organization_id); + assert_eq!(manifest_row.1, destination.guild_id); + assert_eq!(manifest_row.2, destination.destination_id); + assert_eq!(manifest_row.3, destination.destination_revision); + assert_eq!(manifest_row.4, i64::try_from(manifests[0].content_length)?); + assert_eq!(manifest_row.5, manifests[0].sha256); + + async fn race_claim(pool: &PgPool, owner: &str, token: &str) -> Result { + Ok(sqlx::query::query( + r#" +WITH candidate AS ( + SELECT artifact_id + FROM call_scribe_hosted_artifact_delivery_outbox + WHERE status = 'pending' + FOR UPDATE SKIP LOCKED + LIMIT 1 +) +UPDATE call_scribe_hosted_artifact_delivery_outbox AS delivery +SET status = 'in_progress', claim_owner = $1, claim_token = $2, + claim_until = now() + interval '15 minutes', attempt_count = attempt_count + 1 +FROM candidate +WHERE delivery.artifact_id = candidate.artifact_id +"#, + ) + .bind(owner) + .bind(token) + .execute(pool) + .await? + .rows_affected()) + } + let (first_claim, second_claim) = tokio::join!( + race_claim(&first, "instance-a", "claim-a"), + race_claim(&second, "instance-b", "claim-b") + ); + assert_eq!( + first_claim? + second_claim?, + 1, + "only one replica may claim a job" + ); + let wrong_fence = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET operation_id = 'must-not-write' +WHERE recording_id = 'recording-delivery' AND claim_token = 'wrong-claim' +"#, + ) + .execute(&second) + .await?; + assert_eq!(wrong_fence.rows_affected(), 0); + + // Simulate the provider verification having committed while the + // worker lost the HTTP response/DB update. The next claim must verify + // the persisted operation first and must not call prepare or issue a + // second PUT. + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_artifact_delivery_outbox +SET status = 'pending', claim_owner = NULL, claim_token = NULL, claim_until = NULL, + operation_id = 'op-verified', operation_generation = 7, + operation_object_key = 'objects/delivery.wav', next_attempt_at = now() +WHERE recording_id = 'recording-delivery' +"#, + ) + .execute(&first) + .await?; + let unexpected_calls = Arc::new(AtomicU64::new(0)); + let verify_calls = Arc::new(AtomicU64::new(0)); + let verify_counter = verify_calls.clone(); + let signed_at = DateTime::::from_timestamp(Utc::now().timestamp(), 0) + .context("current timestamp must be valid")?; + let verification_expires_at = signed_at + chrono::Duration::minutes(5); + let verification_url = format!( + "http://{address}/objects/delivery.wav?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=test-access/{}/us-east-1/s3/aws4_request&X-Amz-Date={}&X-Amz-Expires=300&X-Amz-SignedHeaders=host;x-amz-checksum-mode&X-Amz-Signature={}", + signed_at.format("%Y%m%d"), + signed_at.format("%Y%m%dT%H%M%SZ"), + "0".repeat(64), + ); + let checksum_bytes = manifests[0] + .sha256 + .as_bytes() + .chunks_exact(2) + .map(|pair| { + u8::from_str_radix(std::str::from_utf8(pair).expect("hex must be UTF-8"), 16) + .expect("manifest digest must be hex") + }) + .collect::>(); + let provider_checksum = base64::engine::general_purpose::STANDARD.encode(checksum_bytes); + let receipt = serde_json::json!({ + "receiptId": "receipt-verified", + "operationId": "op-verified", + "generation": 7, + "recordingId": "recording-delivery", + "artifactId": manifests[0].artifact_id, + "artifactKind": "raw_audio_wav", + "segmentIndex": manifests[0].segment_index, + "objectKey": "objects/delivery.wav", + "destinationId": destination.destination_id, + "destinationRevision": destination.destination_revision, + "provider": destination.provider, + "allowedUploadHost": destination.allowed_host, + "verified": true, + "contentLength": manifests[0].content_length, + "sha256": manifests[0].sha256, + "verifiedAt": Utc::now().to_rfc3339(), + }); + let verification_response = serde_json::json!({ + "receipt": receipt, + "verification": { + "url": verification_url, + "method": "HEAD", + "headers": {"x-amz-checksum-mode": "ENABLED"}, + "expiresAt": verification_expires_at.to_rfc3339(), + } + }); + let provider_content_length = manifests[0].content_length; + let provider_sha256 = manifests[0].sha256.clone(); + let verify_app = axum::Router::new() + .route( + "/internal/v1/worker/artifact-deliveries/op-verified/verify", + axum::routing::post(move || { + let verify_counter = verify_counter.clone(); + let verification_response = verification_response.clone(); + async move { + verify_counter.fetch_add(1, Ordering::SeqCst); + axum::Json(verification_response) + } + }), + ) + .route( + "/objects/delivery.wav", + axum::routing::head(move || { + let provider_checksum = provider_checksum.clone(); + let provider_sha256 = provider_sha256.clone(); + async move { + axum::http::Response::builder() + .status(axum::http::StatusCode::OK) + .header( + reqwest::header::CONTENT_LENGTH, + provider_content_length.to_string(), + ) + .header("x-amz-meta-callscribe-sha256", provider_sha256) + .header("x-amz-checksum-sha256", provider_checksum) + .body(axum::body::Body::empty()) + .expect("provider HEAD response must build") + } + }), + ) + .fallback({ + let unexpected_calls = unexpected_calls.clone(); + move || { + let unexpected_calls = unexpected_calls.clone(); + async move { + unexpected_calls.fetch_add(1, Ordering::SeqCst); + axum::http::StatusCode::INTERNAL_SERVER_ERROR + } + } + }); + let server = tokio::spawn(async move { axum::serve(listener, verify_app).await }); + let retry_client = HostedControlPlaneClient::new( + &format!("http://{address}"), + "test-token-with-at-least-thirty-two-bytes".to_string(), + "delivery-retry-worker".to_string(), + "test-outbox-key-with-at-least-thirty-two-bytes".to_string(), + )?; + store + .retry_hosted_artifact_delivery_outbox( + &retry_client, + &test_dir, + "delivery-retry-worker", + ) + .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'", + ) + .fetch_one(&first) + .await?; + assert_eq!(delivered, "delivered"); + assert!( + !wav_path.exists(), + "only the verified retry may delete local audio" + ); + server.abort(); + + std::fs::remove_dir_all(&test_dir)?; + first.close().await; + second.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<()> { + let Some((admin, pool, _second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping zero-duration proof"); + return Ok(()); + }; + migrate_runtime_schema(&pool).await?; + let store = SqlxRuntimeStore { + pool: pool.clone(), + organization_id: "org-zero-test".to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + store.ensure_organization().await?; + let client = HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-token-with-at-least-thirty-two-bytes".to_string(), + "zero-test-worker".to_string(), + "test-outbox-key-with-at-least-thirty-two-bytes".to_string(), + )?; + let destination = HostedStorageDestination { + organization_id: "org-zero-test".to_string(), + guild_id: "1".to_string(), + provider: "customer_s3".to_string(), + destination_id: "dst-zero".to_string(), + destination_revision: "rev-1".to_string(), + allowed_host: "bucket.s3.us-east-1.amazonaws.com".to_string(), + object_key_prefix: "objects/".to_string(), + transient_delete_policy: "delete_after_verified_delivery".to_string(), + }; + let reservation = UsageReservation { + reservation_id: "reservation-zero".to_string(), + lease_token: "opaque-zero-duration-lease-token".to_string(), + reserved_seconds: 300, + expires_at: (Utc::now() + chrono::Duration::minutes(1)).to_rfc3339(), + }; + let test_dir = std::env::temp_dir().join(format!("call-scribe-zero-{}", Uuid::new_v4())); + std::fs::create_dir_all(&test_dir)?; + let wav_path = test_dir.join("zero.wav"); + hound::WavWriter::create(&wav_path, discord_wav_spec())?.finalize()?; + let started_at = Local::now(); + store + .persist_hosted_capture_recovery( + &client, + &reservation, + "recording-zero", + &wav_path, + started_at, + "instance-a", + &destination, + ) + .await?; + store + .record_session_started( + "recording-zero", + GuildId::new(1), + ChannelId::new(2), + started_at, + "Zero duration proof", + serde_json::json!({"hosted": true}), + ) + .await?; + let (claim_token, _) = store + .claim_live_hosted_capture_finalization( + &reservation.reservation_id, + "recording-zero", + "instance-a", + Utc::now() + chrono::Duration::seconds(5), + ) + .await?; + store + .finalize_zero_duration_hosted_capture_transaction( + &reservation, + "recording-zero", + &claim_token, + &destination, + std::slice::from_ref(&wav_path), + Utc::now(), + ) + .await?; + let counts: (i64, i64, i64, i64) = sqlx::query_as::query_as( + r#" +SELECT + (SELECT count(*) FROM call_scribe_hosted_capture_recovery), + (SELECT count(*) FROM call_scribe_hosted_usage_outbox), + (SELECT count(*) FROM call_scribe_artifacts WHERE session_id = 'recording-zero'), + (SELECT count(*) FROM call_scribe_hosted_artifact_delivery_outbox + WHERE recording_id = 'recording-zero') +"#, + ) + .fetch_one(&pool) + .await?; + assert_eq!(counts, (0, 0, 0, 0)); + assert!( + !store + .has_unsafe_hosted_deliveries("org-zero-test", "1", Duration::ZERO) + .await? + ); + store.finish_zero_duration_hosted_cleanup(&test_dir).await?; + store.finish_zero_duration_hosted_cleanup(&test_dir).await?; + assert!( + !wav_path.exists(), + "header-only WAV must be removed idempotently" + ); + let cleanup_count: i64 = sqlx::query_scalar::query_scalar( + "SELECT count(*) FROM call_scribe_hosted_zero_duration_cleanup_outbox", + ) + .fetch_one(&pool) + .await?; + assert_eq!(cleanup_count, 0); + + std::fs::remove_dir_all(test_dir)?; + pool.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + #[cfg(feature = "discord")] #[test] fn missing_or_failed_heartbeat_self_fences_only_unowned_guilds() { @@ -5537,6 +7160,16 @@ WHERE reservation_id = 'reservation-live' reserved_seconds: 300, expires_at: initial_expiry.to_rfc3339(), }; + let destination = HostedStorageDestination { + organization_id: "org-test".to_string(), + guild_id: "1".to_string(), + provider: "customer_s3".to_string(), + destination_id: "dst-test".to_string(), + destination_revision: "rev-1".to_string(), + allowed_host: "bucket.s3.us-east-1.amazonaws.com".to_string(), + object_key_prefix: "objects/".to_string(), + transient_delete_policy: "delete_after_verified_delivery".to_string(), + }; let base_wav_path = Path::new("/captures/shortened.wav"); store .persist_hosted_capture_recovery( @@ -5546,6 +7179,7 @@ WHERE reservation_id = 'reservation-live' base_wav_path, started_at.with_timezone(&Local), "instance-a", + &destination, ) .await?;