diff --git a/Cargo.lock b/Cargo.lock index 9a41996..34cf8bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -818,6 +818,7 @@ dependencies = [ name = "crank-registry" version = "0.3.1" dependencies = [ + "crank-artifacts", "crank-core", "crank-mapping", "crank-schema", diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index 0ab28aa..465155d 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -588,6 +588,36 @@ impl From for ApiError { format!("import job {job_id} was already applied with different parameters"), json!({ "job_id": job_id }), ), + RegistryError::SourceNotFound { source_id } => Self::not_found_with_context( + "artifact source was not found", + json!({ + "source_id": source_id, + "error_code": "artifact_source_not_found" + }), + ), + RegistryError::SourceConflict { source_id } => Self::conflict_with_context( + "artifact source metadata or lifecycle conflicts with the request", + json!({ + "source_id": source_id, + "error_code": "artifact_source_conflict", + "recovery": "reload" + }), + ), + RegistryError::SourceUnavailable => Self::unprocessable_with_context( + "artifact source is unavailable", + json!({ "error_code": "artifact_source_unavailable" }), + ), + RegistryError::SourceIntegrity => Self::unprocessable_with_context( + "artifact source failed integrity verification", + json!({ "error_code": "artifact_source_integrity" }), + ), + RegistryError::InvalidArtifactSource { field } => Self::validation_with_context( + "artifact source metadata is invalid", + json!({ + "field": field, + "error_code": "artifact_source_invalid" + }), + ), RegistryError::Storage(_) => Self::internal("registry operation failed"), RegistryError::Migration(_) | RegistryError::Serialization(_) diff --git a/apps/admin-api/tests/migration_command.rs b/apps/admin-api/tests/migration_command.rs index 3f651b1..ae6d5ea 100644 --- a/apps/admin-api/tests/migration_command.rs +++ b/apps/admin-api/tests/migration_command.rs @@ -40,7 +40,7 @@ fn plan_is_deterministic_and_committed_contract_is_current() { ); assert_eq!(first.stdout, second.stdout); let plan: serde_json::Value = serde_json::from_slice(&first.stdout).unwrap(); - assert_eq!(plan["sequence"].as_array().unwrap().len(), 11); + assert_eq!(plan["sequence"].as_array().unwrap().len(), 12); let checked = command(&["plan", "--check"], None); assert!( @@ -82,7 +82,7 @@ async fn database_only_config_can_apply_and_preflight_a_fresh_schema() { ); let result: serde_json::Value = serde_json::from_slice(&preflight.stdout).unwrap(); assert_eq!(result["status"], "current"); - assert_eq!(result["version"], 11); + assert_eq!(result["version"], 12); } #[tokio::test] diff --git a/crates/crank-artifacts/src/lib.rs b/crates/crank-artifacts/src/lib.rs index d6173a0..320ed72 100644 --- a/crates/crank-artifacts/src/lib.rs +++ b/crates/crank-artifacts/src/lib.rs @@ -19,5 +19,7 @@ pub mod test_support; pub use error::ArtifactError; pub use housekeeping::{StaleTemp, TempScan}; -pub use model::{ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, StoredArtifact}; +pub use model::{ + ArtifactRef, MAX_ARTIFACT_BYTES, MAX_SOURCE_BYTES, RegisteredArtifact, StoredArtifact, +}; pub use store::ArtifactStore; diff --git a/crates/crank-artifacts/src/model.rs b/crates/crank-artifacts/src/model.rs index a711485..3195b5b 100644 --- a/crates/crank-artifacts/src/model.rs +++ b/crates/crank-artifacts/src/model.rs @@ -66,3 +66,23 @@ pub struct StoredArtifact { pub artifact_ref: ArtifactRef, pub size_bytes: usize, } + +/// Non-forgeable proof that this process published and verified an artifact. +/// +/// Registry metadata creation accepts this capability instead of a caller-built +/// digest, preventing knowledge of a shared digest from granting read access. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RegisteredArtifact { + pub(crate) artifact_ref: ArtifactRef, + pub(crate) size_bytes: usize, +} + +impl RegisteredArtifact { + pub fn artifact_ref(&self) -> &ArtifactRef { + &self.artifact_ref + } + + pub fn size_bytes(&self) -> usize { + self.size_bytes + } +} diff --git a/crates/crank-artifacts/src/store.rs b/crates/crank-artifacts/src/store.rs index 9bf5162..4e80278 100644 --- a/crates/crank-artifacts/src/store.rs +++ b/crates/crank-artifacts/src/store.rs @@ -13,7 +13,7 @@ use std::{ use rand::random; use sha2::{Digest, Sha256}; -use crate::{ArtifactError, ArtifactRef, MAX_ARTIFACT_BYTES, StoredArtifact}; +use crate::{ArtifactError, ArtifactRef, MAX_ARTIFACT_BYTES, RegisteredArtifact, StoredArtifact}; const SHA_DIR: &[u8] = b"sha256"; const TEMP_PREFIX: &str = ".crank-artifact-tmp-v1-"; @@ -106,6 +106,16 @@ impl ArtifactStore { self.put_reader(&mut io::Cursor::new(bytes)) } + /// Stores bytes and returns a non-forgeable capability suitable for + /// creating authoritative registry metadata. + pub fn put_registered(&self, bytes: &[u8]) -> Result { + let stored = self.put(bytes)?; + Ok(RegisteredArtifact { + artifact_ref: stored.artifact_ref, + size_bytes: stored.size_bytes, + }) + } + /// Historical spelling retained for source compatibility; bytes are not /// restricted to UTF-8 at this filesystem boundary. pub fn put_utf8(&self, bytes: &[u8]) -> Result { diff --git a/crates/crank-registry/Cargo.toml b/crates/crank-registry/Cargo.toml index fcd8dc5..6cb30bc 100644 --- a/crates/crank-registry/Cargo.toml +++ b/crates/crank-registry/Cargo.toml @@ -7,6 +7,7 @@ publish.workspace = true version.workspace = true [dependencies] +crank-artifacts = { path = "../crank-artifacts" } crank-core = { path = "../crank-core" } crank-mapping = { path = "../crank-mapping" } crank-schema = { path = "../crank-schema" } @@ -16,8 +17,8 @@ sha2.workspace = true sqlx.workspace = true thiserror.workspace = true time.workspace = true +tokio.workspace = true uuid.workspace = true [dev-dependencies] crank-test-support = { path = "../crank-test-support" } -tokio.workspace = true diff --git a/crates/crank-registry/src/error.rs b/crates/crank-registry/src/error.rs index d0eb05d..39b6ed1 100644 --- a/crates/crank-registry/src/error.rs +++ b/crates/crank-registry/src/error.rs @@ -148,4 +148,14 @@ pub enum RegistryError { OnboardingStaleRevision, #[error("onboarding domain steps are not complete")] OnboardingIncomplete, + #[error("artifact source {source_id} was not found")] + SourceNotFound { source_id: String }, + #[error("artifact source {source_id} conflicts with immutable metadata or lifecycle")] + SourceConflict { source_id: String }, + #[error("artifact source is unavailable")] + SourceUnavailable, + #[error("artifact source integrity verification failed")] + SourceIntegrity, + #[error("artifact source metadata is invalid for field {field}")] + InvalidArtifactSource { field: &'static str }, } diff --git a/crates/crank-registry/src/lib.rs b/crates/crank-registry/src/lib.rs index f55dd64..31d773e 100644 --- a/crates/crank-registry/src/lib.rs +++ b/crates/crank-registry/src/lib.rs @@ -14,21 +14,23 @@ pub use migrations::{ pub mod records { pub use crate::model::{ AdminBootstrapContractRecord, AgentSummary, AgentVersionRecord, AppendProductEventOutcome, - AppliedImportOperation, ApprovalRequestRecord, AuthUserRecord, DescriptorKind, - DescriptorMetadata, ImportJob, ImportJobApplyResult, ImportJobId, ImportJobKind, - ImportJobStatus, InvitationRecord, InvocationHistoryLoss, InvocationHistoryLossCategory, - InvocationHistoryWriteOutcome, InvocationLogRecord, InvocationRetentionOutcome, - InvocationRetentionPolicy, InvocationRetentionStatus, MasterKeyIdentityRecord, - MasterKeyRotationRecord, MasterKeyRotationStatus, MembershipRecord, - OnboardingMilestoneResult, OnboardingPresentationMilestone, OperationAgentRef, - OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord, - Page, PlatformApiKeyRecord, ProductEventRecord, PublishedAgentCatalog, PublishedAgentTool, - RegistryOperation, SampleKind, SecretRecord, SecretVersionRecord, SessionRecord, - SkippedImportOperation, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, - UsageOutcomeBreakdown, UsageOutcomeGroup, UsageRollupRecord, UsageSummary, - UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, - WorkspaceUpstreamId, YamlImportJob, YamlImportJobCompletion, YamlImportJobId, - YamlImportJobStatus, + AppliedImportOperation, ApprovalRequestRecord, ArtifactBlobLifecycle, ArtifactBlobRecord, + ArtifactDigest, ArtifactSourceCursor, ArtifactSourceId, ArtifactSourceLifecycle, + ArtifactSourcePage, ArtifactSourceRecord, ArtifactSourceSensitivity, AuthUserRecord, + DescriptorKind, DescriptorMetadata, ImportJob, ImportJobApplyResult, ImportJobId, + ImportJobKind, ImportJobStatus, InvitationRecord, InvocationHistoryLoss, + InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, InvocationLogRecord, + InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, + MasterKeyIdentityRecord, MasterKeyRotationRecord, MasterKeyRotationStatus, + MembershipRecord, OnboardingMilestoneResult, OnboardingPresentationMilestone, + OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, + OperationVersionRecord, Page, PlatformApiKeyRecord, ProductEventRecord, + PublishedAgentCatalog, PublishedAgentTool, RegistryOperation, SampleKind, SecretRecord, + SecretVersionRecord, SessionRecord, SkippedImportOperation, UsageAgentBreakdown, + UsageBucket, UsageOperationBreakdown, UsageOutcomeBreakdown, UsageOutcomeGroup, + UsageRollupRecord, UsageSummary, UsageTimelinePoint, VerifiedArtifactSource, + WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, + YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }; } @@ -37,11 +39,12 @@ pub mod requests { AdminSecurityAuditRequest, AppendProductEventRequest, ApplyImportJobRequest, ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateApprovalRequest, - CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest, - CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest, - CreateWorkspaceRequest, CreateYamlImportJobRequest, DecideApprovalRequest, - ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest, ImportConflictMode, - ImportOperationDraft, ListApprovalRequestsQuery, ListInvocationLogsQuery, + CreateArtifactSourceRequest, CreateImportJobRequest, CreateInvitationRequest, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, + CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, + DecideApprovalRequest, DetachArtifactSourceRequest, ExpireApprovalRequest, + FinishApprovalRequest, FinishImportJobRequest, ImportConflictMode, ImportOperationDraft, + ListApprovalRequestsQuery, ListArtifactSourcesQuery, ListInvocationLogsQuery, ListProductEventsQuery, MasterKeyIdentityCandidate, PublishAgentRequest, PublishRequest, RecordOnboardingCompletionRequest, RecordOnboardingMilestoneRequest, RecoverAdminPasswordRequest, RotateSecretRequest, SaveAgentBindingsRequest, @@ -63,32 +66,35 @@ pub mod infrastructure { pub use model::{ AdminBootstrapContractRecord, AdminSecurityAuditRequest, AgentStateExpectation, AgentSummary, AgentVersionRecord, AppendProductEventOutcome, AppendProductEventRequest, - AppliedImportOperation, ApplyImportJobRequest, ApprovalRequestRecord, AuthUserRecord, - ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, + AppliedImportOperation, ApplyImportJobRequest, ApprovalRequestRecord, ArtifactBlobLifecycle, + ArtifactBlobRecord, ArtifactDigest, ArtifactSourceCursor, ArtifactSourceId, + ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, ArtifactSourceSensitivity, + AuthUserRecord, ConsumeAdminBootstrapContractRequest, CreateAdminBootstrapContractRequest, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateApprovalRequest, - CreateImportJobRequest, CreateInvitationRequest, CreateInvocationLogRequest, - CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest, CreateWorkspaceRequest, - CreateYamlImportJobRequest, DecideApprovalRequest, DescriptorKind, DescriptorMetadata, + CreateArtifactSourceRequest, CreateImportJobRequest, CreateInvitationRequest, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, + CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, + DecideApprovalRequest, DescriptorKind, DescriptorMetadata, DetachArtifactSourceRequest, ExpireApprovalRequest, FinishApprovalRequest, FinishImportJobRequest, ImportConflictMode, ImportJob, ImportJobApplyResult, ImportJobId, ImportJobKind, ImportJobStatus, ImportOperationDraft, InvitationRecord, InvocationHistoryLoss, InvocationHistoryLossCategory, InvocationHistoryWriteOutcome, InvocationLogRecord, InvocationRetentionOutcome, InvocationRetentionPolicy, InvocationRetentionStatus, ListApprovalRequestsQuery, - ListInvocationLogsQuery, ListProductEventsQuery, MASTER_KEY_CIPHER_CONTRACT, - MasterKeyIdentityCandidate, MasterKeyIdentityRecord, MasterKeyRotationRecord, - MasterKeyRotationStatus, MembershipRecord, OnboardingMilestoneResult, - OnboardingPresentationMilestone, OperationAgentRef, OperationSampleMetadata, - OperationStateExpectation, OperationSummary, OperationUsageSummary, OperationVersionRecord, - Page, PlatformApiKeyRecord, ProductEventRecord, PublishAgentRequest, PublishRequest, - PublishedAgentCatalog, PublishedAgentTool, RecordOnboardingCompletionRequest, + ListArtifactSourcesQuery, ListInvocationLogsQuery, ListProductEventsQuery, + MASTER_KEY_CIPHER_CONTRACT, MAX_ARTIFACT_SOURCE_PAGE_SIZE, MasterKeyIdentityCandidate, + MasterKeyIdentityRecord, MasterKeyRotationRecord, MasterKeyRotationStatus, MembershipRecord, + OnboardingMilestoneResult, OnboardingPresentationMilestone, OperationAgentRef, + OperationSampleMetadata, OperationStateExpectation, OperationSummary, OperationUsageSummary, + OperationVersionRecord, Page, PlatformApiKeyRecord, ProductEventRecord, PublishAgentRequest, + PublishRequest, PublishedAgentCatalog, PublishedAgentTool, RecordOnboardingCompletionRequest, RecordOnboardingMilestoneRequest, RecoverAdminPasswordRequest, RegistryOperation, RotateSecretRequest, SampleKind, SaveAgentBindingsRequest, SaveAgentCatalogConfigRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, SecretRecord, SecretVersionRecord, SessionRecord, SkippedImportOperation, UpdateAgentSummaryRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageOutcomeBreakdown, UsageOutcomeGroup, UsageQuery, - UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, - WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, YamlImportJob, - YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, + UsageRollupRecord, UsageSummary, UsageTimelinePoint, VerifiedArtifactSource, + WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, + YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }; pub use postgres::{PostgresPoolConfig, PostgresPoolConfigError, PostgresRegistry}; diff --git a/crates/crank-registry/src/migrations.rs b/crates/crank-registry/src/migrations.rs index 52b5ea0..92619cc 100644 --- a/crates/crank-registry/src/migrations.rs +++ b/crates/crank-registry/src/migrations.rs @@ -1,6 +1,7 @@ mod admin_auth_lifecycle_v8; mod agent_catalog_lifecycle_v9; mod approval_side_effects_v10; +mod artifact_metadata_v12; mod authority; mod baseline_v1; mod execution_outcome_v5; @@ -11,6 +12,7 @@ mod platform_key_name_reuse_v6; mod schema_guard; mod schema_guard_v10; mod schema_guard_v11; +mod schema_guard_v12; mod schema_guard_v7; mod schema_guard_v8; mod schema_guard_v9; diff --git a/crates/crank-registry/src/migrations/artifact_metadata_v12.rs b/crates/crank-registry/src/migrations/artifact_metadata_v12.rs new file mode 100644 index 0000000..93ede9c --- /dev/null +++ b/crates/crank-registry/src/migrations/artifact_metadata_v12.rs @@ -0,0 +1,44 @@ +use sqlx::{Postgres, Transaction, query}; + +use super::authority::{MigrationDescriptor, MigrationError}; + +pub(super) const SOURCE: &str = include_str!("artifact_metadata_v12.sql"); +pub(super) const SOURCE_SHA256: &str = + "052058da53cd861b2a1af81243b34cc35204d665a9276f8ece563e061f8ebbfe"; + +pub(super) async fn apply( + transaction: &mut Transaction<'_, Postgres>, + descriptor: &MigrationDescriptor, +) -> Result<(), MigrationError> { + sqlx::raw_sql(SOURCE) + .execute(&mut **transaction) + .await + .map_err(|_| { + MigrationError::new( + "apply_failed", + "apply.artifact_metadata", + Some(12), + "restore_known_good_backup", + ) + })?; + query( + "insert into __crank_migrations (version, name, checksum, phase, compatibility) + values ($1, $2, $3, $4, $5)", + ) + .bind(descriptor.version) + .bind(descriptor.name) + .bind(&descriptor.checksum) + .bind(descriptor.phase) + .bind(descriptor.compatibility) + .execute(&mut **transaction) + .await + .map_err(|_| { + MigrationError::new( + "apply_failed", + "apply.canonical_ledger", + Some(12), + "restore_known_good_backup", + ) + })?; + Ok(()) +} diff --git a/crates/crank-registry/src/migrations/artifact_metadata_v12.sql b/crates/crank-registry/src/migrations/artifact_metadata_v12.sql new file mode 100644 index 0000000..4590216 --- /dev/null +++ b/crates/crank-registry/src/migrations/artifact_metadata_v12.sql @@ -0,0 +1,61 @@ +create table artifact_blobs ( + digest text primary key, + artifact_ref text not null unique, + size_bytes bigint not null, + storage_lifecycle text not null default 'available', + claim_token text null, + claim_expires_at timestamptz null, + created_at timestamptz not null, + updated_at timestamptz not null, + constraint artifact_blobs_digest_check check (digest ~ '^[0-9a-f]{64}$'), + constraint artifact_blobs_ref_check check (artifact_ref = 'sha256:' || digest), + constraint artifact_blobs_size_check check (size_bytes between 1 and 262144), + constraint artifact_blobs_lifecycle_check check ( + storage_lifecycle in ('available', 'unavailable') + ), + constraint artifact_blobs_claim_token_check check ( + claim_token is null or octet_length(claim_token) between 1 and 128 + ), + constraint artifact_blobs_claim_shape_check check ( + (claim_token is null and claim_expires_at is null) + or (claim_token is not null and claim_expires_at is not null) + ), + constraint artifact_blobs_timestamps_check check (updated_at >= created_at) +); + +create table artifact_sources ( + workspace_id text not null references workspaces(id) on delete cascade, + source_id text not null, + blob_digest text not null references artifact_blobs(digest), + mime_type text not null, + sensitivity text not null, + lifecycle text not null default 'active', + created_at timestamptz not null, + updated_at timestamptz not null, + detached_at timestamptz null, + constraint artifact_sources_id_check check ( + source_id ~ '^src_[A-Za-z0-9_-]{1,128}$' + ), + constraint artifact_sources_mime_type_check check ( + octet_length(mime_type) between 3 and 255 + and mime_type ~ '^[A-Za-z0-9!#$&^_.+-]+/[A-Za-z0-9!#$&^_.+-]+$' + ), + constraint artifact_sources_sensitivity_check check ( + sensitivity in ('public', 'internal', 'secret') + ), + constraint artifact_sources_lifecycle_check check ( + lifecycle in ('active', 'detached') + ), + constraint artifact_sources_state_check check ( + (lifecycle = 'active' and detached_at is null) + or (lifecycle = 'detached' and detached_at is not null) + ), + constraint artifact_sources_timestamps_check check ( + updated_at >= created_at + and (detached_at is null or (detached_at >= created_at and updated_at >= detached_at)) + ), + primary key (workspace_id, source_id) +); + +create index artifact_sources_workspace_created_idx + on artifact_sources(workspace_id, created_at, source_id); diff --git a/crates/crank-registry/src/migrations/authority.rs b/crates/crank-registry/src/migrations/authority.rs index fece934..85fe7c2 100644 --- a/crates/crank-registry/src/migrations/authority.rs +++ b/crates/crank-registry/src/migrations/authority.rs @@ -1,6 +1,7 @@ use super::admin_auth_lifecycle_v8; use super::agent_catalog_lifecycle_v9; use super::approval_side_effects_v10; +use super::artifact_metadata_v12; use super::execution_outcome_v5; use super::master_key_identity_v7; use super::onboarding_product_events_v11; @@ -11,12 +12,17 @@ use super::schema_guard::{ }; use super::{BASELINE_CHECKSUM, BASELINE_VERSION, apply_baseline}; use crate::ext::ExtensionMigration; -use sha2::{Digest, Sha256}; use sqlx::{PgConnection, PgPool, Row, Transaction, query}; use std::fmt; + +mod contract; + +#[cfg(test)] +use contract::baseline_source_digest; +use contract::validate_descriptors; const MIGRATION_LOCK_ID: i64 = 0x4352_414E_4B4D_4947; -const CURRENT_VERSION: i64 = 11; -const IMPLEMENTED_VERSIONS: &[i64] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]; +const CURRENT_VERSION: i64 = 12; +const IMPLEMENTED_VERSIONS: &[i64] = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]; const BASELINE_SOURCE_SHA256: &str = "eb1656fc5b4b5be9ee390d237d1d58e4b2274ae5ba9b7ba06a2f3f860dfda675"; const CONSOLIDATION_SOURCE: &str = include_str!("consolidation_v2.sql"); @@ -242,6 +248,12 @@ impl MigrationAuthority { onboarding_product_events_v11::SOURCE_SHA256, 10, ), + expand_descriptor( + 12, + "artifact-metadata-v12", + artifact_metadata_v12::SOURCE_SHA256, + 11, + ), ] } pub fn validate_sequence() -> Result<(), MigrationError> { @@ -364,6 +376,9 @@ impl MigrationAuthority { if from < 11 { onboarding_product_events_v11::apply(&mut transaction, &Self::sequence()[10]).await?; } + if from < 12 { + artifact_metadata_v12::apply(&mut transaction, &Self::sequence()[11]).await?; + } transaction .commit() .await @@ -374,134 +389,6 @@ impl MigrationAuthority { }) } } -fn sha256_hex(bytes: &[u8]) -> String { - format!("{:x}", Sha256::digest(bytes)) -} -#[cfg(test)] -fn baseline_source_digest() -> String { - let source = include_str!("baseline_v1.rs"); - let (_, baseline) = source - .split_once("// baseline-v1:start\n") - .expect("baseline start marker must exist"); - let (baseline, _) = baseline - .split_once("// baseline-v1:end") - .expect("baseline end marker must exist"); - sha256_hex(baseline.as_bytes()) -} -fn validate_descriptors(descriptors: &[MigrationDescriptor]) -> Result<(), MigrationError> { - if descriptors.is_empty() || descriptors.len() > 1_024 { - return Err(MigrationError::new( - "invalid_contract", - "contract.sequence", - None, - "contact_operator", - )); - } - for (index, descriptor) in descriptors.iter().enumerate() { - let expected_version = i64::try_from(index + 1).unwrap_or(i64::MAX); - let valid_name = !descriptor.name.is_empty() - && descriptor.name.len() <= 128 - && descriptor - .name - .bytes() - .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') - && !descriptors[..index] - .iter() - .any(|prior| prior.name == descriptor.name); - let valid_source_digest = descriptor.source_digest.len() == 64 - && descriptor - .source_digest - .bytes() - .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()); - let valid_checksum = if descriptor.version == 1 { - descriptor.checksum == BASELINE_CHECKSUM - } else { - descriptor.checksum == descriptor.source_digest - }; - let valid_window = descriptor.readable_schema_min >= 1 - && descriptor.readable_schema_min <= descriptor.readable_schema_max - && descriptor.readable_schema_max <= descriptor.version; - let valid_phase = match descriptor.phase { - "expand" => descriptor.backfill == BackfillPolicy::None, - "migrate" => matches!( - descriptor.backfill, - BackfillPolicy::Bounded { - max_batch_rows: 1..=10_000, - max_batch_ms: 1..=60_000, - resumable: true, - } - ), - "contract" => { - descriptor.compatibility == "window-closed" - && descriptor.contract_evidence.is_some_and(|evidence| { - !evidence.is_empty() - && evidence.len() <= 256 - && !evidence.starts_with('/') - && !evidence.contains("..") - }) - && descriptor.backfill == BackfillPolicy::None - } - _ => false, - }; - let valid_compatibility = matches!( - descriptor.compatibility, - "legacy-baseline" | "n-minus-one-readable" | "window-open" | "window-closed" - ); - let valid_owner = !descriptor.owner.is_empty() - && descriptor.owner.len() <= 128 - && descriptor - .owner - .bytes() - .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-'); - if descriptor.version != expected_version - || !valid_name - || !valid_source_digest - || !valid_checksum - || !valid_phase - || !valid_compatibility - || !valid_owner - || !valid_window - || !descriptor.transactional - { - return Err(MigrationError::new( - "invalid_contract", - "contract.sequence", - Some(descriptor.version), - "contact_operator", - )); - } - } - if descriptors.len() != usize::try_from(CURRENT_VERSION).unwrap_or_default() - || descriptors - .iter() - .map(|descriptor| descriptor.version) - .ne(IMPLEMENTED_VERSIONS.iter().copied()) - || sha256_hex(CONSOLIDATION_SOURCE.as_bytes()) != CONSOLIDATION_SOURCE_SHA256 - || sha256_hex(REQUEST_TRACE_IDENTITY_SOURCE.as_bytes()) - != REQUEST_TRACE_IDENTITY_SOURCE_SHA256 - || sha256_hex(OPERATION_LIFECYCLE_SOURCE.as_bytes()) != OPERATION_LIFECYCLE_SOURCE_SHA256 - || sha256_hex(execution_outcome_v5::SOURCE.as_bytes()) - != execution_outcome_v5::SOURCE_SHA256 - || sha256_hex(platform_key_name_reuse_v6::SOURCE.as_bytes()) - != platform_key_name_reuse_v6::SOURCE_SHA256 - || sha256_hex(master_key_identity_v7::SOURCE.as_bytes()) - != master_key_identity_v7::SOURCE_SHA256 - || sha256_hex(admin_auth_lifecycle_v8::SOURCE.as_bytes()) - != admin_auth_lifecycle_v8::SOURCE_SHA256 - || sha256_hex(agent_catalog_lifecycle_v9::SOURCE.as_bytes()) - != agent_catalog_lifecycle_v9::SOURCE_SHA256 - || sha256_hex(approval_side_effects_v10::SOURCE.as_bytes()) - != approval_side_effects_v10::SOURCE_SHA256 - { - return Err(MigrationError::new( - "invalid_contract", - "contract.implementation", - Some(CURRENT_VERSION), - "contact_operator", - )); - } - Ok(()) -} async fn inspect(connection: &mut PgConnection) -> Result { let core_exists = relation_exists(connection, "__crank_core_migrations").await?; let canonical_exists = relation_exists(connection, "__crank_migrations").await?; @@ -633,6 +520,12 @@ async fn inspect(connection: &mut PgConnection) -> Result String { + format!("{:x}", Sha256::digest(bytes)) +} + +#[cfg(test)] +pub(super) fn baseline_source_digest() -> String { + let source = include_str!("../baseline_v1.rs"); + let (_, baseline) = source + .split_once("// baseline-v1:start\n") + .expect("baseline start marker must exist"); + let (baseline, _) = baseline + .split_once("// baseline-v1:end") + .expect("baseline end marker must exist"); + sha256_hex(baseline.as_bytes()) +} + +pub(super) fn validate_descriptors( + descriptors: &[MigrationDescriptor], +) -> Result<(), MigrationError> { + if descriptors.is_empty() || descriptors.len() > 1_024 { + return Err(MigrationError::new( + "invalid_contract", + "contract.sequence", + None, + "contact_operator", + )); + } + for (index, descriptor) in descriptors.iter().enumerate() { + let expected_version = i64::try_from(index + 1).unwrap_or(i64::MAX); + let valid_name = !descriptor.name.is_empty() + && descriptor.name.len() <= 128 + && descriptor + .name + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') + && !descriptors[..index] + .iter() + .any(|prior| prior.name == descriptor.name); + let valid_source_digest = descriptor.source_digest.len() == 64 + && descriptor + .source_digest + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()); + let valid_checksum = if descriptor.version == 1 { + descriptor.checksum == BASELINE_CHECKSUM + } else { + descriptor.checksum == descriptor.source_digest + }; + let valid_window = descriptor.readable_schema_min >= 1 + && descriptor.readable_schema_min <= descriptor.readable_schema_max + && descriptor.readable_schema_max <= descriptor.version; + let valid_phase = match descriptor.phase { + "expand" => descriptor.backfill == BackfillPolicy::None, + "migrate" => matches!( + descriptor.backfill, + BackfillPolicy::Bounded { + max_batch_rows: 1..=10_000, + max_batch_ms: 1..=60_000, + resumable: true, + } + ), + "contract" => { + descriptor.compatibility == "window-closed" + && descriptor.contract_evidence.is_some_and(|evidence| { + !evidence.is_empty() + && evidence.len() <= 256 + && !evidence.starts_with('/') + && !evidence.contains("..") + }) + && descriptor.backfill == BackfillPolicy::None + } + _ => false, + }; + let valid_compatibility = matches!( + descriptor.compatibility, + "legacy-baseline" | "n-minus-one-readable" | "window-open" | "window-closed" + ); + let valid_owner = !descriptor.owner.is_empty() + && descriptor.owner.len() <= 128 + && descriptor + .owner + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-'); + if descriptor.version != expected_version + || !valid_name + || !valid_source_digest + || !valid_checksum + || !valid_phase + || !valid_compatibility + || !valid_owner + || !valid_window + || !descriptor.transactional + { + return Err(MigrationError::new( + "invalid_contract", + "contract.sequence", + Some(descriptor.version), + "contact_operator", + )); + } + } + if descriptors.len() != usize::try_from(CURRENT_VERSION).unwrap_or_default() + || descriptors + .iter() + .map(|descriptor| descriptor.version) + .ne(IMPLEMENTED_VERSIONS.iter().copied()) + || sha256_hex(CONSOLIDATION_SOURCE.as_bytes()) != CONSOLIDATION_SOURCE_SHA256 + || sha256_hex(REQUEST_TRACE_IDENTITY_SOURCE.as_bytes()) + != REQUEST_TRACE_IDENTITY_SOURCE_SHA256 + || sha256_hex(OPERATION_LIFECYCLE_SOURCE.as_bytes()) != OPERATION_LIFECYCLE_SOURCE_SHA256 + || sha256_hex(execution_outcome_v5::SOURCE.as_bytes()) + != execution_outcome_v5::SOURCE_SHA256 + || sha256_hex(platform_key_name_reuse_v6::SOURCE.as_bytes()) + != platform_key_name_reuse_v6::SOURCE_SHA256 + || sha256_hex(master_key_identity_v7::SOURCE.as_bytes()) + != master_key_identity_v7::SOURCE_SHA256 + || sha256_hex(admin_auth_lifecycle_v8::SOURCE.as_bytes()) + != admin_auth_lifecycle_v8::SOURCE_SHA256 + || sha256_hex(agent_catalog_lifecycle_v9::SOURCE.as_bytes()) + != agent_catalog_lifecycle_v9::SOURCE_SHA256 + || sha256_hex(approval_side_effects_v10::SOURCE.as_bytes()) + != approval_side_effects_v10::SOURCE_SHA256 + || sha256_hex(onboarding_product_events_v11::SOURCE.as_bytes()) + != onboarding_product_events_v11::SOURCE_SHA256 + || sha256_hex(artifact_metadata_v12::SOURCE.as_bytes()) + != artifact_metadata_v12::SOURCE_SHA256 + { + return Err(MigrationError::new( + "invalid_contract", + "contract.implementation", + Some(CURRENT_VERSION), + "contact_operator", + )); + } + Ok(()) +} diff --git a/crates/crank-registry/src/migrations/authority_tests.rs b/crates/crank-registry/src/migrations/authority_tests.rs index 135734c..ba1aa2c 100644 --- a/crates/crank-registry/src/migrations/authority_tests.rs +++ b/crates/crank-registry/src/migrations/authority_tests.rs @@ -8,7 +8,7 @@ fn sequence_is_deterministic_and_append_only() { MigrationAuthority::validate_sequence().unwrap(); assert_eq!( first.iter().map(|item| item.version).collect::>(), - vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11] + vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] ); assert_eq!(first[0].checksum, "crank-community-baseline-v1"); assert_eq!(first[0].source_digest, BASELINE_SOURCE_SHA256); diff --git a/crates/crank-registry/src/migrations/owned_relations.rs b/crates/crank-registry/src/migrations/owned_relations.rs index a4880f2..b851b4e 100644 --- a/crates/crank-registry/src/migrations/owned_relations.rs +++ b/crates/crank-registry/src/migrations/owned_relations.rs @@ -31,6 +31,8 @@ pub(super) const ONBOARDING_PRODUCT_EVENTS: &[&str] = &[ "onboarding_selections", ]; +pub(super) const ARTIFACT_METADATA: &[&str] = &["artifact_blobs", "artifact_sources"]; + pub(super) const CONSOLIDATION: &[&str] = &[ "__crank_migrations", "__crank_migration_legacy_audit", diff --git a/crates/crank-registry/src/migrations/schema_guard.rs b/crates/crank-registry/src/migrations/schema_guard.rs index fed9154..06d6a0c 100644 --- a/crates/crank-registry/src/migrations/schema_guard.rs +++ b/crates/crank-registry/src/migrations/schema_guard.rs @@ -35,6 +35,8 @@ pub(super) const OWNED_RELATIONS: &[&str] = &[ "approval_requests", "invocation_logs", "usage_rollups", + "artifact_blobs", + "artifact_sources", ]; const REQUIRED_COLUMNS: &[(&str, &[&str])] = &[ @@ -615,6 +617,13 @@ pub(super) async fn validate_schema_fingerprint( } else { super::schema_guard_v11::validate_v11_onboarding_product_events(connection).await?; } + if current_version < 12 { + if !super::schema_guard_v12::validate_v12_absent(connection).await? { + return Err(schema_error(current_version)); + } + } else { + super::schema_guard_v12::validate_v12_artifact_metadata(connection).await?; + } Ok(()) } diff --git a/crates/crank-registry/src/migrations/schema_guard_v12.rs b/crates/crank-registry/src/migrations/schema_guard_v12.rs new file mode 100644 index 0000000..1f951af --- /dev/null +++ b/crates/crank-registry/src/migrations/schema_guard_v12.rs @@ -0,0 +1,406 @@ +use sqlx::{PgConnection, Row, query}; + +use super::authority::MigrationError; +use super::schema_guard::schema_error; + +const OWNED_RELATIONS: &[(&str, &str)] = &[ + ("artifact_blobs", "r"), + ("artifact_blobs_artifact_ref_key", "i"), + ("artifact_blobs_pkey", "i"), + ("artifact_sources", "r"), + ("artifact_sources_pkey", "i"), + ("artifact_sources_workspace_created_idx", "i"), +]; + +const BLOB_COLUMNS: &[(&str, &str, bool, Option<&str>)] = &[ + ("digest", "text", false, None), + ("artifact_ref", "text", false, None), + ("size_bytes", "bigint", false, None), + ( + "storage_lifecycle", + "text", + false, + Some("'available'::text"), + ), + ("claim_token", "text", true, None), + ("claim_expires_at", "timestamp with time zone", true, None), + ("created_at", "timestamp with time zone", false, None), + ("updated_at", "timestamp with time zone", false, None), +]; + +const SOURCE_COLUMNS: &[(&str, &str, bool, Option<&str>)] = &[ + ("workspace_id", "text", false, None), + ("source_id", "text", false, None), + ("blob_digest", "text", false, None), + ("mime_type", "text", false, None), + ("sensitivity", "text", false, None), + ("lifecycle", "text", false, Some("'active'::text")), + ("created_at", "timestamp with time zone", false, None), + ("updated_at", "timestamp with time zone", false, None), + ("detached_at", "timestamp with time zone", true, None), +]; + +const BLOB_CONSTRAINTS: &[(&str, &str, &str)] = &[ + ("artifact_blobs_pkey", "p", "PRIMARY KEY (digest)"), + ( + "artifact_blobs_artifact_ref_key", + "u", + "UNIQUE (artifact_ref)", + ), + ( + "artifact_blobs_digest_check", + "c", + "CHECK (digest ~ '^[0-9a-f]{64}$'::text)", + ), + ( + "artifact_blobs_ref_check", + "c", + "CHECK (artifact_ref = ('sha256:'::text || digest))", + ), + ( + "artifact_blobs_size_check", + "c", + "CHECK (size_bytes >= 1 AND size_bytes <= 262144)", + ), + ( + "artifact_blobs_lifecycle_check", + "c", + "CHECK (storage_lifecycle = ANY (ARRAY['available'::text, 'unavailable'::text]))", + ), + ( + "artifact_blobs_claim_token_check", + "c", + "CHECK (claim_token IS NULL OR octet_length(claim_token) >= 1 AND octet_length(claim_token) <= 128)", + ), + ( + "artifact_blobs_claim_shape_check", + "c", + "CHECK (claim_token IS NULL AND claim_expires_at IS NULL OR claim_token IS NOT NULL AND claim_expires_at IS NOT NULL)", + ), + ( + "artifact_blobs_timestamps_check", + "c", + "CHECK (updated_at >= created_at)", + ), +]; + +const SOURCE_CONSTRAINTS: &[(&str, &str, &str)] = &[ + ( + "artifact_sources_pkey", + "p", + "PRIMARY KEY (workspace_id, source_id)", + ), + ( + "artifact_sources_workspace_id_fkey", + "f", + "FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE", + ), + ( + "artifact_sources_blob_digest_fkey", + "f", + "FOREIGN KEY (blob_digest) REFERENCES artifact_blobs(digest)", + ), + ( + "artifact_sources_id_check", + "c", + "CHECK (source_id ~ '^src_[A-Za-z0-9_-]{1,128}$'::text)", + ), + ( + "artifact_sources_mime_type_check", + "c", + "CHECK (octet_length(mime_type) >= 3 AND octet_length(mime_type) <= 255 AND mime_type ~ '^[A-Za-z0-9!#$&^_.+-]+/[A-Za-z0-9!#$&^_.+-]+$'::text)", + ), + ( + "artifact_sources_sensitivity_check", + "c", + "CHECK (sensitivity = ANY (ARRAY['public'::text, 'internal'::text, 'secret'::text]))", + ), + ( + "artifact_sources_lifecycle_check", + "c", + "CHECK (lifecycle = ANY (ARRAY['active'::text, 'detached'::text]))", + ), + ( + "artifact_sources_state_check", + "c", + "CHECK (lifecycle = 'active'::text AND detached_at IS NULL OR lifecycle = 'detached'::text AND detached_at IS NOT NULL)", + ), + ( + "artifact_sources_timestamps_check", + "c", + "CHECK (updated_at >= created_at AND (detached_at IS NULL OR detached_at >= created_at AND updated_at >= detached_at))", + ), +]; + +pub(super) async fn validate_v12_absent( + connection: &mut PgConnection, +) -> Result { + Ok(owned_relations(connection).await?.is_empty()) +} + +pub(super) async fn validate_v12_artifact_metadata( + connection: &mut PgConnection, +) -> Result<(), MigrationError> { + if owned_relations(connection).await? != OWNED_RELATIONS { + return Err(schema_error(12)); + } + validate_table_properties(connection).await?; + validate_columns(connection, "artifact_blobs", BLOB_COLUMNS).await?; + validate_columns(connection, "artifact_sources", SOURCE_COLUMNS).await?; + validate_constraints(connection, "artifact_blobs", BLOB_CONSTRAINTS).await?; + validate_constraints(connection, "artifact_sources", SOURCE_CONSTRAINTS).await?; + validate_indexes(connection).await?; + validate_no_runtime_objects(connection).await +} + +async fn validate_table_properties(connection: &mut PgConnection) -> Result<(), MigrationError> { + let rows = query( + "select c.relname, c.relpersistence::text as persistence, + c.relrowsecurity, c.relforcerowsecurity, c.reloptions + from pg_catalog.pg_class c + join pg_catalog.pg_namespace n on n.oid = c.relnamespace + where n.nspname = current_schema() + and c.relname in ('artifact_blobs', 'artifact_sources') + order by c.relname", + ) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let valid = rows.len() == 2 + && rows.iter().all(|row| { + row.try_get::("persistence").ok().as_deref() == Some("p") + && row.try_get::("relrowsecurity").ok() == Some(false) + && row.try_get::("relforcerowsecurity").ok() == Some(false) + && row + .try_get::>, _>("reloptions") + .ok() + .flatten() + .is_none() + }); + if valid { Ok(()) } else { Err(schema_error(12)) } +} + +async fn owned_relations( + connection: &mut PgConnection, +) -> Result, MigrationError> { + let rows = query( + "select c.relname, c.relkind::text as relkind + from pg_catalog.pg_class c + join pg_catalog.pg_namespace n on n.oid = c.relnamespace + where n.nspname = current_schema() + and c.relname like 'artifact\\_%' escape '\\' + order by c.relname", + ) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + rows.into_iter() + .map(|row| { + let name = row + .try_get::("relname") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let kind = row + .try_get::("relkind") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let expected = OWNED_RELATIONS + .iter() + .copied() + .find(|(expected_name, expected_kind)| { + name == *expected_name && kind == *expected_kind + }) + .ok_or_else(|| schema_error(12))?; + Ok(expected) + }) + .collect() +} + +async fn validate_columns( + connection: &mut PgConnection, + table: &str, + expected: &[(&str, &str, bool, Option<&str>)], +) -> Result<(), MigrationError> { + let rows = query( + "select column_name, data_type, is_nullable, column_default, + datetime_precision, collation_name + from information_schema.columns + where table_schema = current_schema() and table_name = $1 + order by ordinal_position", + ) + .bind(table) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + if rows.len() != expected.len() { + return Err(schema_error(12)); + } + for (row, (name, data_type, nullable, default)) in rows.iter().zip(expected) { + let actual_default = row + .try_get::, _>("column_default") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let datetime_precision = row + .try_get::, _>("datetime_precision") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let collation = row + .try_get::, _>("collation_name") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + if row.try_get::("column_name").ok().as_deref() != Some(*name) + || row.try_get::("data_type").ok().as_deref() != Some(*data_type) + || row.try_get::("is_nullable").ok().as_deref() + != Some(if *nullable { "YES" } else { "NO" }) + || actual_default.as_deref() != *default + || (data_type == &"timestamp with time zone" && datetime_precision != Some(6)) + || collation.is_some() + { + return Err(schema_error(12)); + } + } + Ok(()) +} + +async fn validate_constraints( + connection: &mut PgConnection, + table: &str, + expected: &[(&str, &str, &str)], +) -> Result<(), MigrationError> { + let rows = query( + "select c.conname, c.contype::text as contype, pg_get_constraintdef(c.oid, true) as definition + from pg_catalog.pg_constraint c + join pg_catalog.pg_class t on t.oid = c.conrelid + join pg_catalog.pg_namespace n on n.oid = t.relnamespace + where n.nspname = current_schema() and t.relname = $1 + order by c.conname", + ) + .bind(table) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + if rows.len() != expected.len() { + return Err(schema_error(12)); + } + for row in rows { + let name = row + .try_get::("conname") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let kind = row + .try_get::("contype") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let definition = row + .try_get::("definition") + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let Some((_, _, expected_definition)) = + expected.iter().find(|(expected_name, expected_kind, _)| { + name == *expected_name && kind == *expected_kind + }) + else { + return Err(schema_error(12)); + }; + if canonical_definition(&definition) != canonical_definition(expected_definition) { + return Err(schema_error(12)); + } + } + Ok(()) +} + +fn canonical_definition(value: &str) -> String { + let mut quoted = false; + value + .chars() + .filter_map(|character| { + if character == '\'' { + quoted = !quoted; + return Some(character); + } + if character.is_ascii_whitespace() { + None + } else if quoted { + Some(character) + } else { + Some(character.to_ascii_lowercase()) + } + }) + .collect() +} + +async fn validate_indexes(connection: &mut PgConnection) -> Result<(), MigrationError> { + let rows = query( + "select idx.relname as index_name, t.relname as table_name, am.amname as access_method, + i.indisvalid, i.indisready, i.indisunique, + i.indnkeyatts, i.indnatts, + pg_get_indexdef(i.indexrelid, 1, true) as first_column, + pg_get_indexdef(i.indexrelid, 2, true) as second_column, + pg_get_indexdef(i.indexrelid, 3, true) as third_column, + pg_get_expr(i.indpred, i.indrelid) as predicate + from pg_catalog.pg_index i + join pg_catalog.pg_class idx on idx.oid = i.indexrelid + join pg_catalog.pg_class t on t.oid = i.indrelid + join pg_catalog.pg_namespace n on n.oid = t.relnamespace + join pg_catalog.pg_am am on am.oid = idx.relam + where n.nspname = current_schema() + and t.relname in ('artifact_blobs', 'artifact_sources') + order by idx.relname", + ) + .fetch_all(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + let names = rows + .iter() + .filter_map(|row| row.try_get::("index_name").ok()) + .collect::>(); + if names + != [ + "artifact_blobs_artifact_ref_key", + "artifact_blobs_pkey", + "artifact_sources_pkey", + "artifact_sources_workspace_created_idx", + ] + { + return Err(schema_error(12)); + } + let row = rows.iter().find(|row| { + row.try_get::("index_name").ok().as_deref() + == Some("artifact_sources_workspace_created_idx") + }); + let valid = row.is_some_and(|row| { + row.try_get::("table_name").ok().as_deref() == Some("artifact_sources") + && row.try_get::("access_method").ok().as_deref() == Some("btree") + && row.try_get::("indisvalid").ok() == Some(true) + && row.try_get::("indisready").ok() == Some(true) + && row.try_get::("indisunique").ok() == Some(false) + && row.try_get::("indnkeyatts").ok() == Some(3) + && row.try_get::("indnatts").ok() == Some(3) + && row.try_get::("first_column").ok().as_deref() == Some("workspace_id") + && row.try_get::("second_column").ok().as_deref() == Some("created_at") + && row.try_get::("third_column").ok().as_deref() == Some("source_id") + && row + .try_get::, _>("predicate") + .ok() + .flatten() + .is_none() + }); + if valid { Ok(()) } else { Err(schema_error(12)) } +} + +async fn validate_no_runtime_objects(connection: &mut PgConnection) -> Result<(), MigrationError> { + let count: i64 = sqlx::query_scalar( + "select + (select count(*) from pg_catalog.pg_trigger tr + join pg_catalog.pg_class t on t.oid = tr.tgrelid + join pg_catalog.pg_namespace n on n.oid = t.relnamespace + where n.nspname = current_schema() + and t.relname in ('artifact_blobs', 'artifact_sources') + and not tr.tgisinternal) + + (select count(*) from pg_catalog.pg_policy p + join pg_catalog.pg_class t on t.oid = p.polrelid + join pg_catalog.pg_namespace n on n.oid = t.relnamespace + where n.nspname = current_schema() + and t.relname in ('artifact_blobs', 'artifact_sources'))", + ) + .fetch_one(connection) + .await + .map_err(|_| MigrationError::storage("preflight.schema"))?; + if count == 0 { + Ok(()) + } else { + Err(schema_error(12)) + } +} diff --git a/crates/crank-registry/src/model.rs b/crates/crank-registry/src/model.rs index 71e1c48..7c0de8d 100644 --- a/crates/crank-registry/src/model.rs +++ b/crates/crank-registry/src/model.rs @@ -43,6 +43,10 @@ macro_rules! define_registry_id { }; } +mod artifact_source; + +pub use artifact_source::*; + define_registry_id!(YamlImportJobId); define_registry_id!(ImportJobId); define_registry_id!(WorkspaceUpstreamId); diff --git a/crates/crank-registry/src/model/artifact_source.rs b/crates/crank-registry/src/model/artifact_source.rs new file mode 100644 index 0000000..2e711d0 --- /dev/null +++ b/crates/crank-registry/src/model/artifact_source.rs @@ -0,0 +1,106 @@ +use crank_artifacts::{ArtifactRef, RegisteredArtifact}; +use crank_core::WorkspaceId; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; + +define_registry_id!(ArtifactSourceId); +define_registry_id!(ArtifactDigest); + +pub const MAX_ARTIFACT_SOURCE_PAGE_SIZE: u32 = 100; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactBlobLifecycle { + Available, + Unavailable, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactSourceSensitivity { + Public, + Internal, + Secret, +} + +impl ArtifactSourceSensitivity { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Public => "public", + Self::Internal => "internal", + Self::Secret => "secret", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactSourceLifecycle { + Active, + Detached, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ArtifactBlobRecord { + pub digest: ArtifactDigest, + pub artifact_ref: ArtifactRef, + pub size_bytes: u64, + pub lifecycle: ArtifactBlobLifecycle, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ArtifactSourceRecord { + pub workspace_id: WorkspaceId, + pub source_id: ArtifactSourceId, + pub blob: ArtifactBlobRecord, + pub mime_type: String, + pub sensitivity: ArtifactSourceSensitivity, + pub lifecycle: ArtifactSourceLifecycle, + pub created_at: OffsetDateTime, + pub updated_at: OffsetDateTime, + pub detached_at: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ArtifactSourceCursor { + pub workspace_id: WorkspaceId, + pub created_at: OffsetDateTime, + pub source_id: ArtifactSourceId, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ArtifactSourcePage { + pub items: Vec, + pub next_cursor: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CreateArtifactSourceRequest<'a> { + pub workspace_id: &'a WorkspaceId, + pub source_id: &'a ArtifactSourceId, + pub artifact: &'a RegisteredArtifact, + pub mime_type: &'a str, + pub sensitivity: ArtifactSourceSensitivity, + pub created_at: OffsetDateTime, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ListArtifactSourcesQuery<'a> { + pub workspace_id: &'a WorkspaceId, + pub cursor: Option<&'a ArtifactSourceCursor>, + pub limit: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DetachArtifactSourceRequest<'a> { + pub workspace_id: &'a WorkspaceId, + pub source_id: &'a ArtifactSourceId, + pub expected_updated_at: Option, + pub detached_at: OffsetDateTime, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct VerifiedArtifactSource { + pub source: ArtifactSourceRecord, + pub bytes: Vec, +} diff --git a/crates/crank-registry/src/postgres/artifact_source.rs b/crates/crank-registry/src/postgres/artifact_source.rs new file mode 100644 index 0000000..fc0bc97 --- /dev/null +++ b/crates/crank-registry/src/postgres/artifact_source.rs @@ -0,0 +1,507 @@ +use crank_artifacts::{ArtifactError, ArtifactRef, ArtifactStore, MAX_ARTIFACT_BYTES}; +use crank_core::WorkspaceId; +use sqlx::{Postgres, Row, Transaction, postgres::PgRow}; + +use crate::{ + ArtifactBlobLifecycle, ArtifactBlobRecord, ArtifactDigest, ArtifactSourceCursor, + ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourcePage, ArtifactSourceRecord, + ArtifactSourceSensitivity, CreateArtifactSourceRequest, DetachArtifactSourceRequest, + ListArtifactSourcesQuery, MAX_ARTIFACT_SOURCE_PAGE_SIZE, RegistryError, VerifiedArtifactSource, +}; + +use super::PostgresRegistry; + +const SOURCE_ID_PREFIX: &str = "src_"; +const MAX_SOURCE_ID_SUFFIX_BYTES: usize = 128; +const MAX_MIME_TYPE_BYTES: usize = 255; + +impl PostgresRegistry { + pub async fn create_artifact_source( + &self, + request: CreateArtifactSourceRequest<'_>, + ) -> Result { + validate_create_request(&request)?; + + let mut transaction = self.pool().begin().await?; + require_workspace(&mut transaction, request.workspace_id).await?; + ensure_blob(&mut transaction, &request).await?; + + sqlx::query( + "insert into artifact_sources ( + workspace_id, source_id, blob_digest, mime_type, sensitivity, + lifecycle, created_at, updated_at, detached_at + ) values ($1, $2, $3, $4, $5, 'active', $6, $6, null) + on conflict (workspace_id, source_id) do nothing", + ) + .bind(request.workspace_id.as_str()) + .bind(request.source_id.as_str()) + .bind(request.artifact.artifact_ref().digest_hex()) + .bind(request.mime_type) + .bind(request.sensitivity.as_str()) + .bind(request.created_at) + .execute(&mut *transaction) + .await?; + + let recorded = + get_source_in_transaction(&mut transaction, request.workspace_id, request.source_id) + .await? + .ok_or_else(|| source_not_found(request.source_id))?; + if !is_create_replay(&recorded, &request) { + return Err(source_conflict(request.source_id)); + } + + transaction.commit().await?; + Ok(recorded) + } + + pub async fn get_artifact_source( + &self, + workspace_id: &WorkspaceId, + source_id: &ArtifactSourceId, + ) -> Result { + validate_source_id(source_id)?; + let row = sqlx::query(SOURCE_SELECT) + .bind(workspace_id.as_str()) + .bind(source_id.as_str()) + .fetch_optional(self.pool()) + .await?; + row.map(map_artifact_source) + .transpose()? + .ok_or_else(|| source_not_found(source_id)) + } + + pub async fn list_artifact_sources( + &self, + query: ListArtifactSourcesQuery<'_>, + ) -> Result { + if query.cursor.is_some_and(|cursor| { + cursor.workspace_id != *query.workspace_id + || !valid_source_id(cursor.source_id.as_str()) + }) { + return Err(RegistryError::InvalidArtifactSource { field: "cursor" }); + } + let limit = query.limit.min(MAX_ARTIFACT_SOURCE_PAGE_SIZE); + if limit == 0 { + return Ok(ArtifactSourcePage { + items: Vec::new(), + next_cursor: None, + }); + } + let cursor_created_at = query.cursor.map(|cursor| cursor.created_at); + let cursor_source_id = query.cursor.map(|cursor| cursor.source_id.as_str()); + let fetch_limit = i64::from(limit) + 1; + let rows = sqlx::query( + "select + s.workspace_id, s.source_id, s.blob_digest, s.mime_type, + s.sensitivity, s.lifecycle as source_lifecycle, + s.created_at as source_created_at, s.updated_at as source_updated_at, + s.detached_at, + b.artifact_ref, b.size_bytes, + b.storage_lifecycle as blob_lifecycle + from artifact_sources s + join artifact_blobs b on b.digest = s.blob_digest + where s.workspace_id = $1 + and ( + $2::timestamptz is null + or (s.created_at, s.source_id) > ($2::timestamptz, $3::text) + ) + order by s.created_at asc, s.source_id asc + limit $4", + ) + .bind(query.workspace_id.as_str()) + .bind(cursor_created_at) + .bind(cursor_source_id) + .bind(fetch_limit) + .fetch_all(self.pool()) + .await?; + + let mut items = rows + .into_iter() + .map(map_artifact_source) + .collect::, _>>()?; + let has_more = items.len() > limit as usize; + if has_more { + items.pop(); + } + let next_cursor = has_more.then(|| { + let last = items + .last() + .expect("a non-zero bounded page with an extra row retains one item"); + ArtifactSourceCursor { + workspace_id: query.workspace_id.clone(), + created_at: last.created_at, + source_id: last.source_id.clone(), + } + }); + Ok(ArtifactSourcePage { items, next_cursor }) + } + + pub async fn detach_artifact_source( + &self, + request: DetachArtifactSourceRequest<'_>, + ) -> Result { + validate_source_id(request.source_id)?; + let mut transaction = self.pool().begin().await?; + let mut recorded = + get_source_in_transaction(&mut transaction, request.workspace_id, request.source_id) + .await? + .ok_or_else(|| source_not_found(request.source_id))?; + + if recorded.lifecycle == ArtifactSourceLifecycle::Detached { + if recorded.detached_at == Some(request.detached_at) + && request.expected_updated_at == Some(recorded.created_at) + { + transaction.commit().await?; + return Ok(recorded); + } + return Err(source_conflict(request.source_id)); + } + if request.detached_at < recorded.created_at + || request + .expected_updated_at + .is_some_and(|expected| expected != recorded.updated_at) + { + return Err(source_conflict(request.source_id)); + } + + let updated = sqlx::query( + "update artifact_sources + set lifecycle = 'detached', updated_at = $1, detached_at = $1 + where workspace_id = $2 and source_id = $3 + and lifecycle = 'active' and updated_at = $4", + ) + .bind(request.detached_at) + .bind(request.workspace_id.as_str()) + .bind(request.source_id.as_str()) + .bind(recorded.updated_at) + .execute(&mut *transaction) + .await? + .rows_affected(); + if updated != 1 { + return Err(source_conflict(request.source_id)); + } + + recorded.lifecycle = ArtifactSourceLifecycle::Detached; + recorded.updated_at = request.detached_at; + recorded.detached_at = Some(request.detached_at); + transaction.commit().await?; + Ok(recorded) + } + + pub async fn read_artifact_source( + &self, + store: &ArtifactStore, + workspace_id: &WorkspaceId, + source_id: &ArtifactSourceId, + ) -> Result { + let source = self.get_artifact_source(workspace_id, source_id).await?; + if source.lifecycle != ArtifactSourceLifecycle::Active + || source.blob.lifecycle != ArtifactBlobLifecycle::Available + { + return Err(RegistryError::SourceUnavailable); + } + + let store = store.clone(); + let artifact_ref = source.blob.artifact_ref.clone(); + let bytes = tokio::task::spawn_blocking(move || store.read(&artifact_ref)) + .await + .map_err(|_| RegistryError::SourceUnavailable)? + .map_err(map_artifact_read_error)?; + let actual_size = u64::try_from(bytes.len()).map_err(|_| RegistryError::SourceIntegrity)?; + if actual_size != source.blob.size_bytes { + return Err(RegistryError::SourceIntegrity); + } + Ok(VerifiedArtifactSource { source, bytes }) + } +} + +const SOURCE_SELECT: &str = "select + s.workspace_id, s.source_id, s.blob_digest, s.mime_type, + s.sensitivity, s.lifecycle as source_lifecycle, + s.created_at as source_created_at, s.updated_at as source_updated_at, + s.detached_at, + b.artifact_ref, b.size_bytes, + b.storage_lifecycle as blob_lifecycle + from artifact_sources s + join artifact_blobs b on b.digest = s.blob_digest + where s.workspace_id = $1 and s.source_id = $2"; + +async fn require_workspace( + transaction: &mut Transaction<'_, Postgres>, + workspace_id: &WorkspaceId, +) -> Result<(), RegistryError> { + let exists = + sqlx::query_scalar::<_, bool>("select exists(select 1 from workspaces where id = $1)") + .bind(workspace_id.as_str()) + .fetch_one(&mut **transaction) + .await?; + if exists { + Ok(()) + } else { + Err(RegistryError::WorkspaceNotFound { + workspace_id: workspace_id.as_str().to_owned(), + }) + } +} + +async fn ensure_blob( + transaction: &mut Transaction<'_, Postgres>, + request: &CreateArtifactSourceRequest<'_>, +) -> Result<(), RegistryError> { + let digest = request.artifact.artifact_ref().digest_hex(); + let size_bytes = i64::try_from(request.artifact.size_bytes()).map_err(|_| { + RegistryError::InvalidArtifactSource { + field: "size_bytes", + } + })?; + sqlx::query( + "insert into artifact_blobs ( + digest, artifact_ref, size_bytes, storage_lifecycle, + claim_token, claim_expires_at, created_at, updated_at + ) values ($1, $2, $3, 'available', null, null, $4, $4) + on conflict (digest) do nothing", + ) + .bind(digest) + .bind(request.artifact.artifact_ref().as_str()) + .bind(size_bytes) + .bind(request.created_at) + .execute(&mut **transaction) + .await?; + + let row = sqlx::query( + "select artifact_ref, size_bytes from artifact_blobs where digest = $1 for update", + ) + .bind(digest) + .fetch_one(&mut **transaction) + .await?; + let recorded_ref = row.try_get::("artifact_ref")?; + let recorded_size = row.try_get::("size_bytes")?; + if recorded_ref != request.artifact.artifact_ref().as_str() || recorded_size != size_bytes { + return Err(source_conflict(request.source_id)); + } + Ok(()) +} + +async fn get_source_in_transaction( + transaction: &mut Transaction<'_, Postgres>, + workspace_id: &WorkspaceId, + source_id: &ArtifactSourceId, +) -> Result, RegistryError> { + let row = sqlx::query( + "select + s.workspace_id, s.source_id, s.blob_digest, s.mime_type, + s.sensitivity, s.lifecycle as source_lifecycle, + s.created_at as source_created_at, s.updated_at as source_updated_at, + s.detached_at, + b.artifact_ref, b.size_bytes, + b.storage_lifecycle as blob_lifecycle + from artifact_sources s + join artifact_blobs b on b.digest = s.blob_digest + where s.workspace_id = $1 and s.source_id = $2 + for update of s", + ) + .bind(workspace_id.as_str()) + .bind(source_id.as_str()) + .fetch_optional(&mut **transaction) + .await?; + row.map(map_artifact_source).transpose() +} + +fn map_artifact_source(row: PgRow) -> Result { + let digest = row.try_get::("blob_digest")?; + let artifact_ref_text = row.try_get::("artifact_ref")?; + let artifact_ref = ArtifactRef::parse(&artifact_ref_text).map_err(|_| { + RegistryError::InvalidArtifactSource { + field: "artifact_ref", + } + })?; + if digest != artifact_ref.digest_hex() { + return Err(RegistryError::InvalidArtifactSource { + field: "blob_digest", + }); + } + let size_bytes = row.try_get::("size_bytes")?; + let size_bytes = + u64::try_from(size_bytes).map_err(|_| RegistryError::InvalidArtifactSource { + field: "size_bytes", + })?; + let blob_lifecycle = match row.try_get::("blob_lifecycle")?.as_str() { + "available" => ArtifactBlobLifecycle::Available, + "unavailable" => ArtifactBlobLifecycle::Unavailable, + _ => { + return Err(RegistryError::InvalidArtifactSource { + field: "storage_lifecycle", + }); + } + }; + let sensitivity = match row.try_get::("sensitivity")?.as_str() { + "public" => ArtifactSourceSensitivity::Public, + "internal" => ArtifactSourceSensitivity::Internal, + "secret" => ArtifactSourceSensitivity::Secret, + _ => { + return Err(RegistryError::InvalidArtifactSource { + field: "sensitivity", + }); + } + }; + let lifecycle = match row.try_get::("source_lifecycle")?.as_str() { + "active" => ArtifactSourceLifecycle::Active, + "detached" => ArtifactSourceLifecycle::Detached, + _ => { + return Err(RegistryError::InvalidArtifactSource { field: "lifecycle" }); + } + }; + let detached_at = row.try_get::, _>("detached_at")?; + if (lifecycle == ArtifactSourceLifecycle::Active) != detached_at.is_none() { + return Err(RegistryError::InvalidArtifactSource { + field: "detached_at", + }); + } + + Ok(ArtifactSourceRecord { + workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), + source_id: ArtifactSourceId::new(row.try_get::("source_id")?), + blob: ArtifactBlobRecord { + digest: ArtifactDigest::new(digest), + artifact_ref, + size_bytes, + lifecycle: blob_lifecycle, + }, + mime_type: row.try_get("mime_type")?, + sensitivity, + lifecycle, + created_at: row.try_get("source_created_at")?, + updated_at: row.try_get("source_updated_at")?, + detached_at, + }) +} + +fn validate_create_request(request: &CreateArtifactSourceRequest<'_>) -> Result<(), RegistryError> { + if !valid_source_id(request.source_id.as_str()) { + return Err(RegistryError::InvalidArtifactSource { field: "source_id" }); + } + if !valid_mime_type(request.mime_type) { + return Err(RegistryError::InvalidArtifactSource { field: "mime_type" }); + } + if !valid_artifact_size(request.artifact.size_bytes()) { + return Err(RegistryError::InvalidArtifactSource { + field: "size_bytes", + }); + } + Ok(()) +} + +fn valid_artifact_size(size_bytes: usize) -> bool { + (1..=MAX_ARTIFACT_BYTES).contains(&size_bytes) +} + +fn validate_source_id(source_id: &ArtifactSourceId) -> Result<(), RegistryError> { + if valid_source_id(source_id.as_str()) { + Ok(()) + } else { + Err(RegistryError::InvalidArtifactSource { field: "source_id" }) + } +} + +fn valid_mime_type(value: &str) -> bool { + if value.len() > MAX_MIME_TYPE_BYTES { + return false; + } + let Some((type_name, subtype)) = value.split_once('/') else { + return false; + }; + !type_name.is_empty() + && !subtype.is_empty() + && !subtype.contains('/') + && type_name.bytes().all(is_mime_token_byte) + && subtype.bytes().all(is_mime_token_byte) +} + +fn is_mime_token_byte(byte: u8) -> bool { + byte.is_ascii_alphanumeric() + || matches!( + byte, + b'!' | b'#' | b'$' | b'&' | b'^' | b'_' | b'.' | b'+' | b'-' + ) +} + +fn valid_source_id(value: &str) -> bool { + let Some(suffix) = value.strip_prefix(SOURCE_ID_PREFIX) else { + return false; + }; + !suffix.is_empty() + && suffix.len() <= MAX_SOURCE_ID_SUFFIX_BYTES + && suffix + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-') +} + +fn is_create_replay( + recorded: &ArtifactSourceRecord, + request: &CreateArtifactSourceRequest<'_>, +) -> bool { + recorded.lifecycle == ArtifactSourceLifecycle::Active + && recorded.blob.digest.as_str() == request.artifact.artifact_ref().digest_hex() + && recorded.blob.artifact_ref == *request.artifact.artifact_ref() + && recorded.blob.size_bytes == request.artifact.size_bytes() as u64 + && recorded.mime_type == request.mime_type + && recorded.sensitivity == request.sensitivity +} + +fn map_artifact_read_error(error: ArtifactError) -> RegistryError { + match error { + ArtifactError::NotFound | ArtifactError::UnsafeRoot | ArtifactError::Storage => { + RegistryError::SourceUnavailable + } + ArtifactError::Integrity + | ArtifactError::InvalidReference + | ArtifactError::EmptySource + | ArtifactError::SourceTooLarge => RegistryError::SourceIntegrity, + } +} + +fn source_not_found(source_id: &ArtifactSourceId) -> RegistryError { + RegistryError::SourceNotFound { + source_id: source_id.as_str().to_owned(), + } +} + +fn source_conflict(source_id: &ArtifactSourceId) -> RegistryError { + RegistryError::SourceConflict { + source_id: source_id.as_str().to_owned(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validates_bounded_source_identity_and_metadata() { + assert!(valid_source_id("src_openapi-1")); + assert!(valid_mime_type("application/yaml")); + assert!(valid_artifact_size(1)); + assert!(valid_artifact_size(MAX_ARTIFACT_BYTES)); + + for invalid in ["source", "src_", "src_bad.value"] { + assert!(!valid_source_id(invalid)); + } + for invalid in ["text/plain\n", "text", "text/plain/extra", "text/(plain)"] { + assert!(!valid_mime_type(invalid)); + } + assert!(!valid_artifact_size(0)); + assert!(!valid_artifact_size(MAX_ARTIFACT_BYTES + 1)); + } + + #[test] + fn artifact_failures_map_to_redacted_outcomes() { + assert!(matches!( + map_artifact_read_error(ArtifactError::NotFound), + RegistryError::SourceUnavailable + )); + assert!(matches!( + map_artifact_read_error(ArtifactError::Integrity), + RegistryError::SourceIntegrity + )); + } +} diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index 4dfde21..9806a8c 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -2,6 +2,7 @@ mod agent; mod agent_catalog; mod api_key; mod approval; +mod artifact_source; mod auth; mod connection; mod import_job; diff --git a/crates/crank-registry/tests/integration.rs b/crates/crank-registry/tests/integration.rs index cbe70bf..906a9f4 100644 --- a/crates/crank-registry/tests/integration.rs +++ b/crates/crank-registry/tests/integration.rs @@ -1,6 +1,7 @@ mod integration { mod agents_usage; mod approval; + mod artifact_sources; mod common; mod credential_touch; mod master_key_identity; diff --git a/crates/crank-registry/tests/integration/artifact_sources.rs b/crates/crank-registry/tests/integration/artifact_sources.rs new file mode 100644 index 0000000..31a1499 --- /dev/null +++ b/crates/crank-registry/tests/integration/artifact_sources.rs @@ -0,0 +1,536 @@ +use super::common::TestDatabase; + +use std::{ + fs, + os::unix::fs::PermissionsExt, + path::PathBuf, + sync::atomic::{AtomicU64, Ordering}, +}; + +use crank_artifacts::{ArtifactRef, ArtifactStore}; +use crank_core::{Workspace, WorkspaceId, WorkspaceStatus}; +use crank_registry::{ + ArtifactSourceId, ArtifactSourceLifecycle, ArtifactSourceSensitivity, + CreateArtifactSourceRequest, CreateWorkspaceRequest, DetachArtifactSourceRequest, + ListArtifactSourcesQuery, RegistryError, +}; +use serde_json::json; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; + +static NEXT_ROOT: AtomicU64 = AtomicU64::new(0); + +struct TestRoot(PathBuf); + +impl TestRoot { + fn new(name: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "crank-registry-artifacts-{name}-{}-{}", + std::process::id(), + NEXT_ROOT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + Self(path) + } +} + +impl Drop for TestRoot { + fn drop(&mut self) { + let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700)); + let _ = fs::remove_dir_all(&self.0); + } +} + +fn timestamp(value: &str) -> OffsetDateTime { + OffsetDateTime::parse(value, &Rfc3339).unwrap() +} + +async fn create_workspace(registry: &crank_registry::PostgresRegistry, id: &str) -> WorkspaceId { + let workspace_id = WorkspaceId::new(id); + let created_at = timestamp("2026-08-26T10:00:00Z"); + registry + .create_workspace(CreateWorkspaceRequest { + workspace: &Workspace { + id: workspace_id.clone(), + slug: id.replace('_', "-"), + display_name: id.to_owned(), + status: WorkspaceStatus::Active, + settings: json!({}), + created_at, + updated_at: created_at, + }, + }) + .await + .unwrap(); + workspace_id +} + +#[tokio::test] +async fn source_relations_are_scoped_replayable_pageable_and_detachable() { + let database = TestDatabase::new().await; + let registry = database.registry().await; + let workspace_a = create_workspace(®istry, "ws_artifacts_a").await; + let workspace_b = create_workspace(®istry, "ws_artifacts_b").await; + let root = TestRoot::new("relations"); + let store = ArtifactStore::open(&root.0).unwrap(); + let registered = store.put_registered(b"openapi: 3.1.0\n").unwrap(); + let source_id = ArtifactSourceId::new("src_shared"); + let created_at = timestamp("2026-08-26T10:01:00Z"); + + let created = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at, + }) + .await + .unwrap(); + let replayed = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: timestamp("2026-08-26T10:02:00Z"), + }) + .await + .unwrap(); + assert_eq!(replayed, created); + + assert!(matches!( + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + artifact: ®istered, + mime_type: "application/json", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at, + }) + .await, + Err(RegistryError::SourceConflict { .. }) + )); + assert!(matches!( + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Secret, + created_at, + }) + .await, + Err(RegistryError::SourceConflict { .. }) + )); + let different = store.put_registered(b"openapi: 3.0.3\n").unwrap(); + assert!(matches!( + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + artifact: &different, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at, + }) + .await, + Err(RegistryError::SourceConflict { .. }) + )); + assert!(matches!( + registry.get_artifact_source(&workspace_b, &source_id).await, + Err(RegistryError::SourceNotFound { .. }) + )); + + let shared_in_b = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_b, + source_id: &source_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Secret, + created_at, + }) + .await + .unwrap(); + assert_eq!(shared_in_b.blob.digest, created.blob.digest); + assert_ne!(shared_in_b.sensitivity, created.sensitivity); + let raw_pool = database.raw_pool().await; + let blob_count = + sqlx::query_scalar::<_, i64>("select count(*) from artifact_blobs where digest = $1") + .bind(registered.artifact_ref().digest_hex()) + .fetch_one(&raw_pool) + .await + .unwrap(); + assert_eq!(blob_count, 1); + let total_blob_count: i64 = sqlx::query_scalar("select count(*) from artifact_blobs") + .fetch_one(&raw_pool) + .await + .unwrap(); + assert_eq!( + total_blob_count, 1, + "conflicting replay must roll back blob metadata" + ); + + for id in ["src_page_a", "src_page_b"] { + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &ArtifactSourceId::new(id), + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Public, + created_at, + }) + .await + .unwrap(); + } + let first = registry + .list_artifact_sources(ListArtifactSourcesQuery { + workspace_id: &workspace_a, + cursor: None, + limit: 2, + }) + .await + .unwrap(); + assert_eq!(first.items.len(), 2); + let second = registry + .list_artifact_sources(ListArtifactSourcesQuery { + workspace_id: &workspace_a, + cursor: first.next_cursor.as_ref(), + limit: 2, + }) + .await + .unwrap(); + assert_eq!(second.items.len(), 1); + assert!(second.next_cursor.is_none()); + let listed_ids = first + .items + .iter() + .chain(second.items.iter()) + .map(|source| source.source_id.as_str()) + .collect::>(); + assert_eq!(listed_ids, vec!["src_page_a", "src_page_b", "src_shared"]); + assert!( + first + .items + .iter() + .chain(second.items.iter()) + .all(|source| source.workspace_id == workspace_a) + ); + assert!(matches!( + registry + .list_artifact_sources(ListArtifactSourcesQuery { + workspace_id: &workspace_b, + cursor: first.next_cursor.as_ref(), + limit: 2, + }) + .await, + Err(RegistryError::InvalidArtifactSource { field: "cursor" }) + )); + + sqlx::query( + "insert into artifact_sources + (workspace_id, source_id, blob_digest, mime_type, sensitivity, lifecycle, + created_at, updated_at) + select $1, 'src_bulk_' || lpad(value::text, 3, '0'), $2, + 'application/yaml', 'public', 'active', $3, $3 + from generate_series(0, 100) as value", + ) + .bind(workspace_a.as_str()) + .bind(registered.artifact_ref().digest_hex()) + .bind(created_at) + .execute(&raw_pool) + .await + .unwrap(); + let capped = registry + .list_artifact_sources(ListArtifactSourcesQuery { + workspace_id: &workspace_a, + cursor: None, + limit: u32::MAX, + }) + .await + .unwrap(); + assert_eq!(capped.items.len(), 100); + let tail = registry + .list_artifact_sources(ListArtifactSourcesQuery { + workspace_id: &workspace_a, + cursor: capped.next_cursor.as_ref(), + limit: u32::MAX, + }) + .await + .unwrap(); + assert_eq!(tail.items.len(), 4); + assert!(tail.next_cursor.is_none()); + + let invalid_id = ArtifactSourceId::new(format!("src_{}", "x".repeat(129))); + assert!(matches!( + registry + .get_artifact_source(&workspace_a, &invalid_id) + .await, + Err(RegistryError::InvalidArtifactSource { field: "source_id" }) + )); + assert!(matches!( + registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &invalid_id, + expected_updated_at: None, + detached_at: created_at, + }) + .await, + Err(RegistryError::InvalidArtifactSource { field: "source_id" }) + )); + + let detached_at = timestamp("2026-08-26T10:03:00Z"); + let detached = registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + expected_updated_at: Some(created.updated_at), + detached_at, + }) + .await + .unwrap(); + assert_eq!(detached.lifecycle, ArtifactSourceLifecycle::Detached); + let retry = registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + expected_updated_at: Some(created.updated_at), + detached_at, + }) + .await + .unwrap(); + assert_eq!(retry, detached); + assert!(matches!( + registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &source_id, + expected_updated_at: Some(created.updated_at), + detached_at: timestamp("2026-08-26T10:04:00Z"), + }) + .await, + Err(RegistryError::SourceConflict { .. }) + )); + assert_eq!( + registry + .get_artifact_source(&workspace_a, &source_id) + .await + .unwrap() + .lifecycle, + ArtifactSourceLifecycle::Detached + ); + assert!(matches!( + registry + .read_artifact_source(&store, &workspace_a, &source_id) + .await, + Err(RegistryError::SourceUnavailable) + )); + assert_eq!( + store.read(registered.artifact_ref()).unwrap(), + b"openapi: 3.1.0\n" + ); + + let concurrent_id = ArtifactSourceId::new("src_concurrent"); + let concurrent = registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_a, + source_id: &concurrent_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at, + }) + .await + .unwrap(); + let first_registry = registry.clone(); + let second_registry = registry.clone(); + let first_workspace = workspace_a.clone(); + let second_workspace = workspace_a.clone(); + let first_id = concurrent_id.clone(); + let second_id = concurrent_id.clone(); + let (first_detach, second_detach) = tokio::join!( + async move { + first_registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &first_workspace, + source_id: &first_id, + expected_updated_at: Some(concurrent.updated_at), + detached_at: timestamp("2026-08-26T10:05:00Z"), + }) + .await + }, + async move { + second_registry + .detach_artifact_source(DetachArtifactSourceRequest { + workspace_id: &second_workspace, + source_id: &second_id, + expected_updated_at: Some(concurrent.updated_at), + detached_at: timestamp("2026-08-26T10:06:00Z"), + }) + .await + } + ); + assert_eq!( + usize::from(first_detach.is_ok()) + usize::from(second_detach.is_ok()), + 1 + ); + assert_eq!( + usize::from(matches!( + first_detach, + Err(RegistryError::SourceConflict { .. }) + )) + usize::from(matches!( + second_detach, + Err(RegistryError::SourceConflict { .. }) + )), + 1 + ); + + database.cleanup().await; +} + +#[tokio::test] +async fn verified_read_returns_only_digest_and_size_verified_bytes() { + let database = TestDatabase::new().await; + let registry = database.registry().await; + let workspace_id = create_workspace(®istry, "ws_verified_read").await; + let root = TestRoot::new("verified-read"); + let store = ArtifactStore::open(&root.0).unwrap(); + let bytes = b"openapi: 3.1.0\ninfo: {}\n"; + let registered = store.put_registered(bytes).unwrap(); + let valid_id = ArtifactSourceId::new("src_valid"); + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_id, + source_id: &valid_id, + artifact: ®istered, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Secret, + created_at: timestamp("2026-08-26T11:00:00Z"), + }) + .await + .unwrap(); + let verified = registry + .read_artifact_source(&store, &workspace_id, &valid_id) + .await + .unwrap(); + assert_eq!(verified.bytes, bytes); + + let raw_pool = database.raw_pool().await; + let missing_ref = ArtifactRef::from_digest_hex(&"f".repeat(64)).unwrap(); + let missing_id = ArtifactSourceId::new("src_missing"); + sqlx::query( + "insert into artifact_blobs + (digest, artifact_ref, size_bytes, storage_lifecycle, created_at, updated_at) + values ($1, $2, 12, 'available', $3, $3)", + ) + .bind(missing_ref.digest_hex()) + .bind(missing_ref.as_str()) + .bind(timestamp("2026-08-26T11:01:00Z")) + .execute(&raw_pool) + .await + .unwrap(); + sqlx::query( + "insert into artifact_sources + (workspace_id, source_id, blob_digest, mime_type, sensitivity, lifecycle, + created_at, updated_at) + values ($1, $2, $3, 'application/yaml', 'internal', 'active', $4, $4)", + ) + .bind(workspace_id.as_str()) + .bind(missing_id.as_str()) + .bind(missing_ref.digest_hex()) + .bind(timestamp("2026-08-26T11:01:00Z")) + .execute(&raw_pool) + .await + .unwrap(); + assert!(matches!( + registry + .read_artifact_source(&store, &workspace_id, &missing_id) + .await, + Err(RegistryError::SourceUnavailable) + )); + assert!( + registry + .get_artifact_source(&workspace_id, &missing_id) + .await + .is_ok() + ); + + let wrong_size_id = ArtifactSourceId::new("src_wrong_size"); + let wrong_size = store.put_registered(b"different source").unwrap(); + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_id, + source_id: &wrong_size_id, + artifact: &wrong_size, + mime_type: "application/yaml", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: timestamp("2026-08-26T11:02:00Z"), + }) + .await + .unwrap(); + sqlx::query("update artifact_blobs set size_bytes = size_bytes + 1 where digest = $1") + .bind(wrong_size.artifact_ref().digest_hex()) + .execute(&raw_pool) + .await + .unwrap(); + assert!(matches!( + registry + .read_artifact_source(&store, &workspace_id, &wrong_size_id) + .await, + Err(RegistryError::SourceIntegrity) + )); + + let unavailable = store.put_registered(b"temporarily unavailable").unwrap(); + let unavailable_id = ArtifactSourceId::new("src_unavailable"); + registry + .create_artifact_source(CreateArtifactSourceRequest { + workspace_id: &workspace_id, + source_id: &unavailable_id, + artifact: &unavailable, + mime_type: "text/plain", + sensitivity: ArtifactSourceSensitivity::Internal, + created_at: timestamp("2026-08-26T11:03:00Z"), + }) + .await + .unwrap(); + sqlx::query("update artifact_blobs set storage_lifecycle = 'unavailable' where digest = $1") + .bind(unavailable.artifact_ref().digest_hex()) + .execute(&raw_pool) + .await + .unwrap(); + assert!(matches!( + registry + .read_artifact_source(&store, &workspace_id, &unavailable_id) + .await, + Err(RegistryError::SourceUnavailable) + )); + + let tampered = root + .0 + .join("sha256") + .join(registered.artifact_ref().digest_hex().get(..2).unwrap()) + .join(registered.artifact_ref().digest_hex()); + fs::set_permissions(&tampered, fs::Permissions::from_mode(0o600)).unwrap(); + fs::write(&tampered, vec![b'x'; bytes.len()]).unwrap(); + assert!(matches!( + registry + .read_artifact_source(&store, &workspace_id, &valid_id) + .await, + Err(RegistryError::SourceIntegrity) + )); + assert!( + registry + .get_artifact_source(&workspace_id, &valid_id) + .await + .is_ok() + ); + + database.cleanup().await; +} diff --git a/crates/crank-registry/tests/integration/migrations.rs b/crates/crank-registry/tests/integration/migrations.rs index 3b51a73..7db908d 100644 --- a/crates/crank-registry/tests/integration/migrations.rs +++ b/crates/crank-registry/tests/integration/migrations.rs @@ -1,6 +1,7 @@ use crank_registry::{MigrationAuthority, MigrationPreflight, PostgresRegistry}; use sqlx::Row; +mod artifact_metadata; mod rollback; static EVENT_TRIGGER_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); @@ -49,6 +50,7 @@ async fn controlled_authority_is_versioned_and_safe_under_concurrent_apply() { (9, "agent-catalog-lifecycle-v9"), (10, "approval-side-effects-v10"), (11, "onboarding-product-events-v11"), + (12, "artifact-metadata-v12"), ]; assert_eq!(rows.len(), expected.len()); for (row, (version, name)) in rows.iter().zip(expected) { @@ -91,9 +93,20 @@ async fn controlled_authority_is_versioned_and_safe_under_concurrent_apply() { .await .expect("V11 key provenance column must be readable"); assert!(key_scope_column); + let artifact_relations = sqlx::query( + "select table_name + from information_schema.tables + where table_schema = current_schema() + and table_name in ('artifact_blobs', 'artifact_sources') + order by table_name", + ) + .fetch_all(first.pool()) + .await + .expect("V12 artifact metadata relations must be readable"); + assert_eq!(artifact_relations.len(), 2); assert_eq!( MigrationAuthority::preflight(first.pool()).await.unwrap(), - MigrationPreflight::Current { version: 11 } + MigrationPreflight::Current { version: 12 } ); } #[tokio::test] @@ -216,7 +229,7 @@ async fn legacy_core_baseline_is_consolidated_without_data_loss() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 1, - target: 11, + target: 12, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -295,7 +308,7 @@ async fn future_sequence_fails_closed() { let database_url = crank_test_support::postgres_schema_url("test_future_sequence").await; let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); MigrationAuthority::apply(&pool).await.unwrap(); - sqlx::query("update __crank_migrations set version = 12 where version = 11") + sqlx::query("update __crank_migrations set version = 13 where version = 12") .execute(&pool) .await .unwrap(); @@ -307,6 +320,7 @@ async fn future_sequence_fails_closed() { "future_version" ); } + #[tokio::test] async fn healthy_v2_is_reported_as_migration_required_and_upgrades_to_v3() { let database_url = crank_test_support::postgres_schema_url("test_v2_to_v3_identity").await; @@ -322,13 +336,13 @@ async fn healthy_v2_is_reported_as_migration_required_and_upgrades_to_v3() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 2, - target: 11, + target: 12, } ); MigrationAuthority::apply(&pool).await.unwrap(); assert_eq!( MigrationAuthority::preflight(&pool).await.unwrap(), - MigrationPreflight::Current { version: 11 } + MigrationPreflight::Current { version: 12 } ); let trace_column: bool = sqlx::query_scalar( "select exists ( @@ -371,7 +385,7 @@ async fn healthy_v3_upgrades_to_v4_with_honest_legacy_snapshot_provenance() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 3, - target: 11, + target: 12, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -431,7 +445,7 @@ async fn healthy_v4_upgrades_to_v5_without_fabricating_legacy_outcomes() { MigrationAuthority::preflight(&pool).await.unwrap(), MigrationPreflight::MigrationRequired { current: 4, - target: 11, + target: 12, } ); MigrationAuthority::apply(&pool).await.unwrap(); @@ -685,7 +699,7 @@ async fn v3_upgrade_ignores_oversized_legacy_request_ids_in_partial_index() { MigrationAuthority::apply(&pool).await.unwrap(); assert_eq!( MigrationAuthority::preflight(&pool).await.unwrap(), - MigrationPreflight::Current { version: 11 } + MigrationPreflight::Current { version: 12 } ); } async fn remove_v3_schema(pool: &sqlx::PgPool) { @@ -822,6 +836,7 @@ async fn remove_v10_schema(pool: &sqlx::PgPool) { .unwrap(); } async fn remove_v11_schema(pool: &sqlx::PgPool) { + remove_v12_schema(pool).await; sqlx::raw_sql( "drop table if exists onboarding_selections; drop trigger if exists product_events_append_only_guard on product_events; @@ -838,6 +853,16 @@ async fn remove_v11_schema(pool: &sqlx::PgPool) { .await .unwrap(); } +async fn remove_v12_schema(pool: &sqlx::PgPool) { + sqlx::raw_sql( + "drop table if exists artifact_sources; + drop table if exists artifact_blobs; + delete from __crank_migrations where version = 12;", + ) + .execute(pool) + .await + .unwrap(); +} #[tokio::test] async fn partial_sequence_fails_closed() { let database_url = crank_test_support::postgres_schema_url("test_partial_sequence").await; diff --git a/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs b/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs new file mode 100644 index 0000000..f3de502 --- /dev/null +++ b/crates/crank-registry/tests/integration/migrations/artifact_metadata.rs @@ -0,0 +1,177 @@ +use super::*; + +#[tokio::test] +async fn healthy_v11_upgrades_to_v12_without_rewriting_prior_ledger() { + let database_url = crank_test_support::postgres_schema_url("test_v11_to_v12_artifacts").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + let prior = sqlx::query( + "select version, name, checksum, applied_at + from __crank_migrations where version <= 11 order by version", + ) + .fetch_all(&pool) + .await + .unwrap() + .into_iter() + .map(|row| { + ( + row.get::("version"), + row.get::("name"), + row.get::("checksum"), + row.get::("applied_at"), + ) + }) + .collect::>(); + remove_v12_schema(&pool).await; + assert_eq!( + MigrationAuthority::preflight(&pool).await.unwrap(), + MigrationPreflight::MigrationRequired { + current: 11, + target: 12, + } + ); + + MigrationAuthority::apply(&pool).await.unwrap(); + let after = sqlx::query( + "select version, name, checksum, applied_at + from __crank_migrations where version <= 11 order by version", + ) + .fetch_all(&pool) + .await + .unwrap() + .into_iter() + .map(|row| { + ( + row.get::("version"), + row.get::("name"), + row.get::("checksum"), + row.get::("applied_at"), + ) + }) + .collect::>(); + assert_eq!(after, prior); + + let v12_applied_at: time::OffsetDateTime = + sqlx::query_scalar("select applied_at from __crank_migrations where version = 12") + .fetch_one(&pool) + .await + .unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + let replayed_at: time::OffsetDateTime = + sqlx::query_scalar("select applied_at from __crank_migrations where version = 12") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(replayed_at, v12_applied_at); +} + +#[tokio::test] +async fn v12_exact_guard_rejects_column_constraint_index_and_relation_drift() { + let database_url = crank_test_support::postgres_schema_url("test_v12_artifact_drift").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + + for (drift, restore) in [ + ( + "alter table artifact_blobs add column unexpected text null;", + "alter table artifact_blobs drop column unexpected;", + ), + ( + "create index unrelated_idx on artifact_sources(source_id) include (mime_type);", + "drop index unrelated_idx;", + ), + ( + "alter table artifact_blobs alter column storage_lifecycle set default 'unavailable';", + "alter table artifact_blobs alter column storage_lifecycle set default 'available';", + ), + ( + "alter table artifact_sources alter column mime_type drop not null;", + "alter table artifact_sources alter column mime_type set not null;", + ), + ( + "alter table artifact_blobs drop constraint artifact_blobs_size_check; + alter table artifact_blobs add constraint artifact_blobs_size_check + check (size_bytes between 1 and 262145);", + "alter table artifact_blobs drop constraint artifact_blobs_size_check; + alter table artifact_blobs add constraint artifact_blobs_size_check + check (size_bytes between 1 and 262144);", + ), + ( + "drop index artifact_sources_workspace_created_idx; + create index artifact_sources_workspace_created_idx + on artifact_sources(workspace_id, source_id, created_at);", + "drop index artifact_sources_workspace_created_idx; + create index artifact_sources_workspace_created_idx + on artifact_sources(workspace_id, created_at, source_id);", + ), + ( + "alter table artifact_sources alter column created_at type timestamptz(3);", + "alter table artifact_sources alter column created_at type timestamptz;", + ), + ( + "alter table artifact_sources enable row level security;", + "alter table artifact_sources disable row level security;", + ), + ( + "create policy unexpected_policy on artifact_sources using (true);", + "drop policy unexpected_policy on artifact_sources;", + ), + ( + "create function unrelated_trigger_fn() returns trigger language plpgsql as $$ + begin return new; end + $$; + create trigger unexpected_trigger before insert on artifact_sources + for each row execute function unrelated_trigger_fn();", + "drop trigger unexpected_trigger on artifact_sources; + drop function unrelated_trigger_fn();", + ), + ( + "alter table artifact_sources set unlogged;", + "alter table artifact_sources set logged;", + ), + ( + "alter table artifact_sources drop constraint artifact_sources_id_check; + alter table artifact_sources add constraint artifact_sources_id_check + check (source_id ~ '^src_[a-zA-Z0-9_-]{1,128}$');", + "alter table artifact_sources drop constraint artifact_sources_id_check; + alter table artifact_sources add constraint artifact_sources_id_check + check (source_id ~ '^src_[A-Za-z0-9_-]{1,128}$');", + ), + ( + "alter table artifact_blobs drop constraint artifact_blobs_claim_shape_check; + alter table artifact_blobs add constraint artifact_blobs_claim_shape_check check ( + claim_token is null and + (claim_expires_at is null or claim_token is not null) and + claim_expires_at is not null + );", + "alter table artifact_blobs drop constraint artifact_blobs_claim_shape_check; + alter table artifact_blobs add constraint artifact_blobs_claim_shape_check check ( + (claim_token is null and claim_expires_at is null) + or (claim_token is not null and claim_expires_at is not null) + );", + ), + ] { + sqlx::raw_sql(drift).execute(&pool).await.unwrap(); + let error = MigrationAuthority::preflight(&pool).await.unwrap_err(); + assert_eq!(error.code(), "partial_sequence", "drift: {drift}"); + assert_eq!(error.version(), Some(12), "drift: {drift}"); + sqlx::raw_sql(restore).execute(&pool).await.unwrap(); + assert_eq!( + MigrationAuthority::preflight(&pool).await.unwrap(), + MigrationPreflight::Current { version: 12 }, + "restore: {restore}" + ); + } + + sqlx::query("alter table artifact_sources rename to artifact_sources_table") + .execute(&pool) + .await + .unwrap(); + sqlx::query("create view artifact_sources as select * from artifact_sources_table") + .execute(&pool) + .await + .unwrap(); + let error = MigrationAuthority::preflight(&pool).await.unwrap_err(); + assert_eq!(error.code(), "partial_sequence"); + assert_eq!(error.version(), Some(12)); +} diff --git a/crates/crank-registry/tests/integration/migrations/rollback.rs b/crates/crank-registry/tests/integration/migrations/rollback.rs index f4d2f98..c9364a2 100644 --- a/crates/crank-registry/tests/integration/migrations/rollback.rs +++ b/crates/crank-registry/tests/integration/migrations/rollback.rs @@ -1,5 +1,73 @@ use super::*; +#[tokio::test] +async fn failed_artifact_metadata_migration_rolls_back_schema_and_ledger() { + let _event_trigger_guard = EVENT_TRIGGER_TEST_LOCK.lock().await; + let database_url = + crank_test_support::postgres_schema_url("test_artifact_metadata_rollback").await; + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + MigrationAuthority::apply(&pool).await.unwrap(); + remove_v12_schema(&pool).await; + let schema: String = sqlx::query_scalar("select current_schema()") + .fetch_one(&pool) + .await + .unwrap(); + let failure_trigger = format!( + "create function reject_story21b_v12() returns event_trigger language plpgsql as $$ + begin + if current_schema() = '{schema}' and current_query() like '%artifact_sources%' then + raise exception 'injected v12 ddl failure'; + end if; + end $$; + create event trigger reject_story21b_v12 on ddl_command_start + execute function reject_story21b_v12();" + ); + sqlx::raw_sql(sqlx::AssertSqlSafe(failure_trigger)) + .execute(&pool) + .await + .unwrap(); + let error = MigrationAuthority::apply(&pool).await.unwrap_err(); + sqlx::raw_sql( + "drop event trigger reject_story21b_v12; + drop function reject_story21b_v12();", + ) + .execute(&pool) + .await + .unwrap(); + assert_eq!(error.code(), "apply_failed"); + assert_eq!(error.version(), Some(12)); + for relation in [ + "artifact_blobs", + "artifact_blobs_pkey", + "artifact_blobs_artifact_ref_key", + "artifact_sources", + "artifact_sources_pkey", + "artifact_sources_workspace_created_idx", + ] { + let present: bool = sqlx::query_scalar( + "select to_regclass(format('%I.%I', current_schema(), $1)) is not null", + ) + .bind(relation) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!present, "{relation} must roll back with failed V12 DDL"); + } + let ledger_v12: i64 = + sqlx::query_scalar("select count(*) from __crank_migrations where version = 12") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(ledger_v12, 0); + assert_eq!( + MigrationAuthority::preflight(&pool).await.unwrap(), + MigrationPreflight::MigrationRequired { + current: 11, + target: 12, + } + ); +} + #[tokio::test] async fn failed_consolidation_rolls_back_all_changes() { let _event_trigger_guard = EVENT_TRIGGER_TEST_LOCK.lock().await; diff --git a/docs/migrations.md b/docs/migrations.md index e309b0d..089b1df 100644 --- a/docs/migrations.md +++ b/docs/migrations.md @@ -16,7 +16,7 @@ docker compose -f deploy/community/docker-compose.yml --env-file deploy/communit 2. Создайте и проверьте согласованный backup PostgreSQL и artifact storage. Для первой пустой установки зафиксируйте, что восстанавливать нечего. 3. Проверьте immutable plan: `cargo run -p admin-api --bin crank-migrate -- plan --check` в source checkout либо ` run --rm migrate crank-migrate plan` для образа. 4. Примените sequence: ` run --rm migrate crank-migrate apply`. -5. Повторите preflight и убедитесь в `{"status":"current","version":11}`. +5. Повторите preflight и убедитесь в `{"status":"current","version":12}`. 6. Только теперь запускайте long-running services: ` up -d`. Обычный `up` также содержит обязательный migration job, но при upgrade он не заменяет предварительные preflight и backup. Migrator делает до десяти bounded попыток подключения с секундной паузой и затем безопасно завершается ошибкой. @@ -85,6 +85,11 @@ cargo run -p admin-api --bin crank-migrate -- plan --check `invocation_logs.platform_api_key_id`, immutable local `product_events`, daily denominator rollups и partial success index. Legacy history не backfill-ится; external/unscoped credentials не получают fabricated key identity. +- V12 — global immutable artifact blob metadata и workspace-owned source + relations. Digest/ref и size фиксируются canonical constraints; MIME, + sensitivity и source lifecycle принадлежат scoped relation. Claim token/expiry + зарезервированы bounded all-or-none contract для последующего reconciliation, + но эта версия не запускает cleanup и не удаляет physical blobs. - Каждая версия имеет contiguous `i64` version, стабильное имя, lowercase SHA-256, owner, phase, explicit readable schema min/max и backfill policy. - `migrate` требует bounded cursor/batch policy; `contract` дополнительно требует tracked compatibility evidence и закрытого окна. - Добавление descriptor без executable implementation блокируется `invalid_contract` до DB I/O. diff --git a/docs/schemas/migration-sequence.json b/docs/schemas/migration-sequence.json index b9ad646..ffac4f7 100644 --- a/docs/schemas/migration-sequence.json +++ b/docs/schemas/migration-sequence.json @@ -176,6 +176,22 @@ "source_digest": "a439bbcdc9cc909d717ed5a868c7a51bd1ad3c49c4e85fd20933743ee3f31166", "transactional": true, "version": 11 + }, + { + "backfill": { + "kind": "none" + }, + "checksum": "052058da53cd861b2a1af81243b34cc35204d665a9276f8ece563e061f8ebbfe", + "compatibility": "n-minus-one-readable", + "contract_evidence": null, + "name": "artifact-metadata-v12", + "owner": "crank-registry", + "phase": "expand", + "readable_schema_max": 12, + "readable_schema_min": 11, + "source_digest": "052058da53cd861b2a1af81243b34cc35204d665a9276f8ece563e061f8ebbfe", + "transactional": true, + "version": 12 } ] }