diff --git a/distribution/self-hosted/release-manifest.schema.json b/distribution/self-hosted/release-manifest.schema.json index 1ec8639..4c64b5d 100644 --- a/distribution/self-hosted/release-manifest.schema.json +++ b/distribution/self-hosted/release-manifest.schema.json @@ -129,9 +129,9 @@ "required": ["sha256", "count", "latest", "forwardOnly"], "properties": { "sha256": { "$ref": "#/$defs/sha256" }, - "count": { "const": 10 }, - "latest": { "const": "010" }, - "forwardOnly": { "const": ["010"] } + "count": { "const": 11 }, + "latest": { "const": "011" }, + "forwardOnly": { "const": ["010", "011"] } } }, "catalog": { diff --git a/docs/operations.md b/docs/operations.md index 625d10a..6e4f755 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -233,6 +233,24 @@ schema. An active forgotten writer causes a bounded lock-timeout failure; the transaction and migration registration roll back, leaving schema 009 usable. Stop that writer and rerun the unchanged migration. +Migration 011 reconciles the historical +`external_source_snapshots.decoded_bytes` counter. Native ingestion before 011 +counted UTF-8 instruction bytes plus resource bytes, while the restore gate and +migration-006 backfill defined the same derived counter as the sum of immutable +canonical revision-object bytes. The migration accepts only an exact match to +one of those two representations, recomputes the canonical total through +immutable snapshot observations (including reused revisions), and records every +prior and resulting value in the append-only +`external_snapshot_byte_total_reconciliations` ledger. It does not change bundle +or content hashes, revision identities, provenance, resources, classifications, +or advisory history. Any other total, malformed object, hash mismatch, overflow, +or duplicate accounting aborts the transaction and leaves schema 010 unchanged. +Migration 011 is safe to rerun through the normal idempotent migrator, but it is +also a forward-only boundary. A deferred database constraint rejects every new +snapshot unless its canonical projection and reconciliation evidence agree at +commit, so a pre-011 writer fails rather than advancing mixed-format history. +Never start that writer against schema 011. + ### Backup, restore, and rollback for migration 010 The pre-upgrade backup must include the complete PostgreSQL database and be @@ -287,13 +305,14 @@ corrections. Do not edit the migration table. `Database migration ... is newer than binary migration ...` means the selected image is too old for that database; select a compatible post-migration image. Do -not bypass the guard. A lock or statement timeout during migration 010 means -either a writer was not drained or the rehearsed bound is too small. Confirm -writers are stopped before changing a bound. Because each migration and its -registration are transactional, a failed 010 leaves schema 009 in place and is -safe to rerun after the cause is fixed. A historical candidate-attribution -failure is a data-integrity stop: keep traffic down, preserve evidence, and -repair it only through a separately reviewed forward migration or restore. +not bypass the guard. A lock or statement timeout during migration 010 or 011 +means either a writer was not drained or the rehearsed bound is too small. +Confirm writers are stopped before changing a bound. Because each migration and +its registration are transactional, a failed migration leaves the preceding +schema in place and is safe to rerun after the cause is fixed. A historical +candidate-attribution failure is a data-integrity stop: keep traffic down, +preserve evidence, and repair it only through a separately reviewed forward +migration or restore. ### Catalog verification fails diff --git a/docs/upgrades/v0.2.0.md b/docs/upgrades/v0.2.0.md new file mode 100644 index 0000000..d54eac0 --- /dev/null +++ b/docs/upgrades/v0.2.0.md @@ -0,0 +1,38 @@ +# v0.2.0 upgrade: canonical snapshot byte totals + +v0.2.0 adds migration 011 to reconcile one derived imported-snapshot counter. +The immutable catalog objects, hashes, revisions, provenance, classifications, +and advisory history remain authoritative and unchanged. + +Before migration 011, two exact counter representations can exist: + +- native ingestion counted UTF-8 instruction bytes plus resource bytes; +- the migration-006 backfill counted canonical revision-object bytes. + +Both are derived exclusively from immutable objects. Migration 011 accepts only +an exact match to one of them, records the prior representation and both totals +in an append-only ledger, and stores the canonical total. Shared revisions are +accounted through their immutable per-snapshot observations, not only through +the snapshot that first created the revision. Quarantined and missing +observations contribute no revision bytes. After migration, a deferred database +constraint recomputes the immutable projection at commit and rejects missing or +fabricated reconciliation evidence, including writes from a pre-011 binary. + +Treat this as a maintenance upgrade: + +1. Stop and drain every SkillWire, ingestion, and administration writer. +2. Take a protected PostgreSQL backup and restore-validate it in an isolated + PostgreSQL 17.10 instance while it is still on schema 009 or 010. +3. Run the v0.2.0 migrator. Schema 009 upgrades through 010 and 011; schema 010 + applies only 011. +4. Rerun the migrator to prove idempotence, then require schema 011 readiness, + catalog verification, advisory verification, and the exact MCP smoke tests. +5. Start only a schema-011-compatible application. + +An arbitrary counter difference is not a compatibility case. Malformed or +missing catalog objects, content or bundle hash drift, overflow, duplicate +revision accounting, invalid attribution, or inconsistent audit evidence aborts +the migration. Preserve the failed database and backup; do not edit either in +place. Crossing back over migration 010 or 011 requires the validated +pre-upgrade database backup and its matching application image, not an +image-only rollback. diff --git a/migrations/011_reconcile_snapshot_byte_totals.sql b/migrations/011_reconcile_snapshot_byte_totals.sql new file mode 100644 index 0000000..973cb32 --- /dev/null +++ b/migrations/011_reconcile_snapshot_byte_totals.sql @@ -0,0 +1,429 @@ +-- Reconcile legacy decoded-payload totals to immutable canonical revision bytes. +-- +-- Before migration 011, native ingestion stored the UTF-8 byte length of each +-- observed revision's instructions plus resources. Migration 006 and the v0.2.0 +-- restore gate instead interpreted decoded_bytes as canonical revision bytes. +-- Both values are derived from immutable catalog objects. Accept only either +-- exact representation, record the transition, and make canonical bytes the +-- sole stored representation after this migration. + +ALTER TABLE external_source_snapshots + DISABLE TRIGGER external_snapshots_immutable; + +CREATE TABLE external_snapshot_byte_total_reconciliations ( + snapshot_id uuid PRIMARY KEY + REFERENCES external_source_snapshots(id) ON DELETE RESTRICT, + prior_decoded_bytes bigint NOT NULL CHECK (prior_decoded_bytes >= 0), + legacy_payload_decoded_bytes bigint NOT NULL + CHECK (legacy_payload_decoded_bytes >= 0), + reconciled_decoded_bytes bigint NOT NULL + CHECK (reconciled_decoded_bytes >= 0), + prior_representation text NOT NULL + CHECK (prior_representation IN ('canonical', 'legacy-payload')), + migration_version text NOT NULL DEFAULT '011' + CHECK (migration_version = '011'), + reconciled_at timestamptz NOT NULL DEFAULT transaction_timestamp(), + CHECK ( + (prior_representation = 'canonical' + AND prior_decoded_bytes = reconciled_decoded_bytes) + OR + (prior_representation = 'legacy-payload' + AND prior_decoded_bytes = legacy_payload_decoded_bytes) + ) +); + +COMMENT ON TABLE external_snapshot_byte_total_reconciliations IS + 'Append-only migration-011 evidence for exact legacy-payload to canonical snapshot byte-total reconciliation.'; + +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM external_content_objects content + WHERE content.byte_length <> octet_length(content.content) + OR content.sha256 <> + encode(sha256(convert_to(content.content, 'UTF8')), 'hex') + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found malformed content objects'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM external_skill_revisions revision + WHERE revision.bundle_sha256 <> + encode(sha256(convert_to(revision.canonical_bytes, 'UTF8')), 'hex') + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found malformed canonical revisions'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM external_skill_revisions revision + LEFT JOIN external_content_objects instructions + ON instructions.sha256=revision.instructions_sha256 + LEFT JOIN external_content_objects license + ON license.sha256=revision.license_sha256 + LEFT JOIN external_content_objects notice + ON notice.sha256=revision.notice_sha256 + WHERE instructions.sha256 IS NULL + OR instructions.kind<>'instructions' + OR license.sha256 IS NULL + OR license.kind<>'license' + OR revision.notice_sha256 IS NOT NULL + AND (notice.sha256 IS NULL OR notice.kind<>'notice') + OR EXISTS ( + SELECT 1 + FROM external_revision_resources resource + LEFT JOIN external_content_objects content + ON content.sha256=resource.content_sha256 + WHERE resource.revision_id=revision.id + AND (content.sha256 IS NULL + OR content.kind<>'resource' + OR content.byte_length<>resource.byte_length) + ) + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found malformed catalog objects'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM external_snapshot_skill_observations observation + LEFT JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + LEFT JOIN external_import_candidates candidate + ON candidate.id=observation.candidate_id + WHERE (observation.revision_id IS NULL AND ( + observation.result NOT IN ('missing','quarantined') + OR observation.observed_content_identity_sha256 IS NOT NULL + )) + OR (observation.revision_id IS NOT NULL AND ( + revision.id IS NULL + OR candidate.id IS NULL + OR observation.result NOT IN ('published','reused') + OR observation.skill_identity_id <> revision.skill_identity_id + OR candidate.snapshot_id <> observation.snapshot_id + OR candidate.published_revision_id IS NOT NULL + AND candidate.published_revision_id <> observation.revision_id + OR candidate.skill_identity_id IS NOT NULL + AND candidate.skill_identity_id <> observation.skill_identity_id + OR observation.observed_content_identity_sha256 IS DISTINCT FROM + revision.content_identity_sha256 + )) + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found malformed observation attribution'; + END IF; + + IF EXISTS ( + SELECT observation.snapshot_id + FROM external_snapshot_skill_observations observation + WHERE observation.revision_id IS NOT NULL + GROUP BY observation.snapshot_id + HAVING count(*) <> count(DISTINCT observation.revision_id) + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found duplicate revision accounting'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM external_source_snapshots snapshot + CROSS JOIN LATERAL ( + SELECT + count(observation.revision_id)::bigint AS observed_revision_count, + COALESCE(sum(octet_length(revision.canonical_bytes)::numeric), 0) + AS canonical_total, + COALESCE( + sum( + instructions.byte_length::numeric + + COALESCE(resources.total_bytes, 0) + ), + 0 + ) AS legacy_total + FROM external_snapshot_skill_observations observation + JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + JOIN external_content_objects instructions + ON instructions.sha256=revision.instructions_sha256 + LEFT JOIN LATERAL ( + SELECT COALESCE(sum(resource.byte_length::numeric), 0) AS total_bytes + FROM external_revision_resources resource + WHERE resource.revision_id=revision.id + ) resources ON true + WHERE observation.snapshot_id=snapshot.id + ) totals + WHERE snapshot.revision_count <> totals.observed_revision_count + OR snapshot.candidate_count <> ( + SELECT count(*) FROM external_import_candidates candidate + WHERE candidate.snapshot_id=snapshot.id + ) + OR snapshot.quarantine_count <> ( + SELECT count(*) + FROM external_import_candidates candidate + JOIN external_current_classifications current + ON current.candidate_id=candidate.id + WHERE candidate.snapshot_id=snapshot.id + AND current.classification='quarantined' + ) + OR snapshot.resource_count <> ( + SELECT count(*) + FROM external_snapshot_skill_observations observation + JOIN external_revision_resources resource + ON resource.revision_id=observation.revision_id + WHERE observation.snapshot_id=snapshot.id + ) + OR snapshot.dependency_count <> ( + SELECT count(*) + FROM external_snapshot_skill_observations observation + JOIN external_revision_dependencies dependency + ON dependency.revision_id=observation.revision_id + WHERE observation.snapshot_id=snapshot.id + ) + OR totals.canonical_total > 9223372036854775807::numeric + OR totals.legacy_total > 9223372036854775807::numeric + OR snapshot.decoded_bytes NOT IN ( + totals.canonical_total::bigint, + totals.legacy_total::bigint + ) + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation found unsupported totals'; + END IF; +END; +$$; + +CREATE TEMPORARY TABLE snapshot_byte_total_reconciliation_work +ON COMMIT DROP AS +SELECT + snapshot.id AS snapshot_id, + snapshot.decoded_bytes AS prior_decoded_bytes, + totals.legacy_total::bigint AS legacy_payload_decoded_bytes, + totals.canonical_total::bigint AS reconciled_decoded_bytes, + CASE + WHEN snapshot.decoded_bytes=totals.canonical_total::bigint + THEN 'canonical' + ELSE 'legacy-payload' + END AS prior_representation +FROM external_source_snapshots snapshot +CROSS JOIN LATERAL ( + SELECT + COALESCE(sum(octet_length(revision.canonical_bytes)::numeric), 0) + AS canonical_total, + COALESCE( + sum( + instructions.byte_length::numeric + + COALESCE(resources.total_bytes, 0) + ), + 0 + ) AS legacy_total + FROM external_snapshot_skill_observations observation + JOIN external_skill_revisions revision ON revision.id=observation.revision_id + JOIN external_content_objects instructions + ON instructions.sha256=revision.instructions_sha256 + LEFT JOIN LATERAL ( + SELECT COALESCE(sum(resource.byte_length::numeric), 0) AS total_bytes + FROM external_revision_resources resource + WHERE resource.revision_id=revision.id + ) resources ON true + WHERE observation.snapshot_id=snapshot.id +) totals; + +UPDATE external_source_snapshots snapshot +SET decoded_bytes=reconciliation.reconciled_decoded_bytes +FROM snapshot_byte_total_reconciliation_work reconciliation +WHERE reconciliation.snapshot_id=snapshot.id; +SET CONSTRAINTS external_snapshot_finalization_required IMMEDIATE; +ALTER TABLE external_source_snapshots + ENABLE TRIGGER external_snapshots_immutable; + +INSERT INTO external_snapshot_byte_total_reconciliations ( + snapshot_id,prior_decoded_bytes,legacy_payload_decoded_bytes, + reconciled_decoded_bytes,prior_representation +) +SELECT + snapshot_id,prior_decoded_bytes,legacy_payload_decoded_bytes, + reconciled_decoded_bytes,prior_representation +FROM snapshot_byte_total_reconciliation_work; + +CREATE TRIGGER external_snapshot_byte_total_reconciliations_immutable + BEFORE UPDATE OR DELETE ON external_snapshot_byte_total_reconciliations + FOR EACH ROW EXECUTE FUNCTION reject_external_history_mutation(); +CREATE TRIGGER external_snapshot_byte_total_reconciliations_truncate_rejected + BEFORE TRUNCATE ON external_snapshot_byte_total_reconciliations + FOR EACH STATEMENT EXECUTE FUNCTION reject_external_history_mutation(); + +-- Bind every post-011 snapshot to its canonical projection and immutable +-- reconciliation evidence at commit. This rejects pre-011 writers even when +-- their legacy total happens to equal the canonical total, and prevents direct +-- SQL from advancing a projection with missing or fabricated evidence. +CREATE FUNCTION validate_external_snapshot_byte_total_projection() +RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + totals record; + reconciliation record; +BEGIN + SELECT + snapshot.revision_count, + snapshot.candidate_count, + snapshot.quarantine_count, + snapshot.resource_count, + snapshot.dependency_count, + snapshot.decoded_bytes, + count(observation.revision_id)::numeric AS observed_revision_count, + count(DISTINCT observation.revision_id)::numeric + AS distinct_observed_revision_count, + COALESCE(sum( + CASE WHEN observation.revision_id IS NULL THEN 0 + ELSE octet_length(revision.canonical_bytes)::numeric END + ), 0) AS canonical_total, + COALESCE(sum( + CASE WHEN observation.revision_id IS NULL THEN 0 + ELSE instructions.byte_length::numeric + + COALESCE(resources.total_bytes, 0) END + ), 0) AS legacy_total, + COALESCE(sum(COALESCE(resources.resource_count, 0)), 0)::numeric + AS observed_resource_count, + COALESCE(sum(COALESCE(dependencies.dependency_count, 0)), 0)::numeric + AS observed_dependency_count, + count(*) FILTER ( + WHERE (observation.revision_id IS NULL + AND observation.snapshot_id IS NOT NULL + AND ( + observation.result NOT IN ('missing','quarantined') + OR observation.observed_content_identity_sha256 IS NOT NULL + )) + OR (observation.revision_id IS NOT NULL AND ( + revision.id IS NULL + OR revision.bundle_sha256 <> + encode(sha256(convert_to(revision.canonical_bytes, 'UTF8')), 'hex') + OR instructions.sha256 IS NULL + OR instructions.kind <> 'instructions' + OR instructions.byte_length <> octet_length(instructions.content) + OR instructions.sha256 <> + encode(sha256(convert_to(instructions.content, 'UTF8')), 'hex') + OR license.sha256 IS NULL + OR license.kind <> 'license' + OR license.byte_length <> octet_length(license.content) + OR license.sha256 <> + encode(sha256(convert_to(license.content, 'UTF8')), 'hex') + OR revision.notice_sha256 IS NOT NULL AND ( + notice.sha256 IS NULL + OR notice.kind <> 'notice' + OR notice.byte_length <> octet_length(notice.content) + OR notice.sha256 <> + encode(sha256(convert_to(notice.content, 'UTF8')), 'hex') + ) + OR candidate.id IS NULL + OR candidate.snapshot_id <> observation.snapshot_id + OR candidate.skill_identity_id IS NOT NULL + AND candidate.skill_identity_id <> observation.skill_identity_id + OR candidate.published_revision_id IS NOT NULL + AND candidate.published_revision_id <> observation.revision_id + OR observation.result NOT IN ('published','reused') + OR observation.skill_identity_id <> revision.skill_identity_id + OR observation.observed_content_identity_sha256 IS DISTINCT FROM + revision.content_identity_sha256 + OR COALESCE(resources.invalid_object_count, 0) <> 0 + )) + )::numeric AS invalid_object_count + INTO totals + FROM external_source_snapshots snapshot + LEFT JOIN external_snapshot_skill_observations observation + ON observation.snapshot_id=snapshot.id + LEFT JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + LEFT JOIN external_content_objects instructions + ON instructions.sha256=revision.instructions_sha256 + LEFT JOIN external_content_objects license + ON license.sha256=revision.license_sha256 + LEFT JOIN external_content_objects notice + ON notice.sha256=revision.notice_sha256 + LEFT JOIN external_import_candidates candidate + ON candidate.id=observation.candidate_id + LEFT JOIN LATERAL ( + SELECT + count(*)::numeric AS resource_count, + COALESCE(sum(resource.byte_length::numeric), 0) AS total_bytes, + count(*) FILTER ( + WHERE content.sha256 IS NULL + OR content.kind <> 'resource' + OR content.byte_length <> resource.byte_length + OR content.byte_length <> octet_length(content.content) + OR content.sha256 <> + encode(sha256(convert_to(content.content, 'UTF8')), 'hex') + )::numeric AS invalid_object_count + FROM external_revision_resources resource + LEFT JOIN external_content_objects content + ON content.sha256=resource.content_sha256 + WHERE resource.revision_id=revision.id + ) resources ON true + LEFT JOIN LATERAL ( + SELECT count(*)::numeric AS dependency_count + FROM external_revision_dependencies dependency + WHERE dependency.revision_id=revision.id + ) dependencies ON true + WHERE snapshot.id=NEW.id + GROUP BY snapshot.id; + + SELECT * INTO reconciliation + FROM external_snapshot_byte_total_reconciliations + WHERE snapshot_id=NEW.id; + + IF totals IS NULL + OR reconciliation IS NULL + OR totals.observed_revision_count <> + totals.distinct_observed_revision_count + OR totals.revision_count <> totals.observed_revision_count + OR totals.candidate_count <> ( + SELECT count(*) FROM external_import_candidates candidate + WHERE candidate.snapshot_id=NEW.id + ) + OR totals.quarantine_count <> ( + SELECT count(*) + FROM external_import_candidates candidate + JOIN external_current_classifications current + ON current.candidate_id=candidate.id + WHERE candidate.snapshot_id=NEW.id + AND current.classification='quarantined' + ) + OR totals.resource_count <> totals.observed_resource_count + OR totals.dependency_count <> totals.observed_dependency_count + OR totals.invalid_object_count <> 0 + OR totals.canonical_total > 9223372036854775807::numeric + OR totals.legacy_total > 9223372036854775807::numeric + OR totals.decoded_bytes <> totals.canonical_total + OR reconciliation.migration_version <> '011' + OR reconciliation.reconciled_decoded_bytes <> totals.canonical_total + OR reconciliation.reconciled_decoded_bytes <> totals.decoded_bytes + OR reconciliation.legacy_payload_decoded_bytes <> totals.legacy_total + OR reconciliation.prior_representation='canonical' + AND reconciliation.prior_decoded_bytes <> totals.canonical_total + OR reconciliation.prior_representation='legacy-payload' + AND reconciliation.prior_decoded_bytes <> totals.legacy_total + OR reconciliation.prior_representation NOT IN + ('canonical','legacy-payload') THEN + RAISE EXCEPTION 'snapshot byte-total projection is invalid'; + END IF; + RETURN NULL; +END; +$$; + +CREATE CONSTRAINT TRIGGER external_snapshot_byte_total_projection_valid + AFTER INSERT OR UPDATE ON external_source_snapshots + DEFERRABLE INITIALLY DEFERRED + FOR EACH ROW + EXECUTE FUNCTION validate_external_snapshot_byte_total_projection(); + +DO $$ +BEGIN + IF (SELECT count(*) FROM external_snapshot_byte_total_reconciliations) <> + (SELECT count(*) FROM external_source_snapshots) + OR EXISTS ( + SELECT 1 + FROM external_source_snapshots snapshot + JOIN external_snapshot_byte_total_reconciliations reconciliation + ON reconciliation.snapshot_id=snapshot.id + WHERE snapshot.decoded_bytes <> reconciliation.reconciled_decoded_bytes + ) THEN + RAISE EXCEPTION 'snapshot byte-total reconciliation is incomplete'; + END IF; +END; +$$; diff --git a/scripts/build-self-hosted-release.ts b/scripts/build-self-hosted-release.ts index e8ec280..4eeca28 100644 --- a/scripts/build-self-hosted-release.ts +++ b/scripts/build-self-hosted-release.ts @@ -276,8 +276,8 @@ export async function buildSelfHostedRelease( compatibility: { node: "24.18.0", postgresql: "17.10", - schemaMinimum: 10, - schemaMaximum: 10, + schemaMinimum: 9, + schemaMaximum: 11, }, feature003Integrity: { path: feature003Integrity.path, diff --git a/src/onboarding/adapters/postgres/backup.ts b/src/onboarding/adapters/postgres/backup.ts index 94e3ff1..74555fe 100644 --- a/src/onboarding/adapters/postgres/backup.ts +++ b/src/onboarding/adapters/postgres/backup.ts @@ -307,7 +307,7 @@ export class PostgresBackupAdapter { ); if ( validation.latestMigration !== - (this.options.expectedLatestMigration ?? "010") || + (this.options.expectedLatestMigration ?? "011") || !validation.migrationInventoryValid || !validation.constraintsValid || !validation.catalogValid || diff --git a/src/onboarding/adapters/postgres/restore-validation.ts b/src/onboarding/adapters/postgres/restore-validation.ts index 32bdd24..9b9c6b8 100644 --- a/src/onboarding/adapters/postgres/restore-validation.ts +++ b/src/onboarding/adapters/postgres/restore-validation.ts @@ -75,6 +75,8 @@ const TriggerEvidenceSchema = z functionBodySha256: z.string().regex(/^[0-9a-f]{64}$/), timing: z.enum(["BEFORE", "AFTER", "INSTEAD OF"]), level: z.enum(["ROW", "STATEMENT"]), + deferrable: z.boolean(), + initiallyDeferred: z.boolean(), events: z.array(TriggerEventSchema).min(1).max(4), enabled: z.enum(["origin", "disabled", "replica", "always"]), definition: z @@ -98,6 +100,12 @@ const EvidenceSchema = z.object({ dependencyCount: z.number().int().nonnegative(), contentObjectCount: z.number().int().nonnegative(), identitySha256: z.string().regex(/^[0-9a-f]{64}$/), + legacySnapshotByteTotals: z.number().int().nonnegative(), + invalidSnapshotByteTotals: z.number().int().nonnegative(), + invalidSnapshotByteOverflows: z.number().int().nonnegative(), + invalidSnapshotObjectGraph: z.number().int().nonnegative(), + invalidRevisionHashes: z.number().int().nonnegative(), + invalidSnapshotReconciliations: z.number().int().nonnegative(), invalidSnapshotCounts: z.number().int().nonnegative(), invalidPublishedPointers: z.number().int().nonnegative(), invalidContentLengths: z.number().int().nonnegative(), @@ -181,6 +189,8 @@ const FUNCTION_BODY_SHA256 = { "e26c0466292fb76c9f4cce6af78ea7f617617d7a5f117fd20e15f3bfbbfdde8c", validate_external_revision_classification_transition: "01461630956a69c1ead4bfd7bf1df621e217a352052123fcfe79e401dea840b8", + validate_external_snapshot_byte_total_projection: + "6c866c631670b92a1ea5082ad0440ce207c0a53274073871acdcf45de5d18877", } as const; const REQUIRED_TRIGGERS = [ @@ -307,6 +317,8 @@ const REQUIRED_TRIGGERS = [ FUNCTION_BODY_SHA256.require_external_snapshot_finalization, events: ["INSERT", "UPDATE"], timing: "AFTER", + deferrable: true, + initiallyDeferred: true, }, { minimumMigration: 10, @@ -325,6 +337,36 @@ const REQUIRED_TRIGGERS = [ functionBodySha256: FUNCTION_BODY_SHA256.reject_external_history_mutation, events: ["DELETE", "UPDATE"], }, + { + minimumMigration: 11, + triggerName: "external_snapshot_byte_total_projection_valid", + tableName: "external_source_snapshots", + functionName: "validate_external_snapshot_byte_total_projection", + functionBodySha256: + FUNCTION_BODY_SHA256.validate_external_snapshot_byte_total_projection, + events: ["INSERT", "UPDATE"], + timing: "AFTER", + deferrable: true, + initiallyDeferred: true, + }, + { + minimumMigration: 11, + triggerName: "external_snapshot_byte_total_reconciliations_immutable", + tableName: "external_snapshot_byte_total_reconciliations", + functionName: "reject_external_history_mutation", + functionBodySha256: FUNCTION_BODY_SHA256.reject_external_history_mutation, + events: ["DELETE", "UPDATE"], + }, + { + minimumMigration: 11, + triggerName: + "external_snapshot_byte_total_reconciliations_truncate_rejected", + tableName: "external_snapshot_byte_total_reconciliations", + functionName: "reject_external_history_mutation", + functionBodySha256: FUNCTION_BODY_SHA256.reject_external_history_mutation, + events: ["TRUNCATE"], + level: "STATEMENT", + }, ] as const; export async function expectedMigrationInventory( @@ -461,13 +503,25 @@ export function assessRestoredDatabaseEvidence( : required.functionBodySha256) && trigger.timing === ("timing" in required ? required.timing : "BEFORE") && - trigger.level === "ROW" && + trigger.level === ("level" in required ? required.level : "ROW") && + trigger.deferrable === + ("deferrable" in required ? required.deferrable : false) && + trigger.initiallyDeferred === + ("initiallyDeferred" in required + ? required.initiallyDeferred + : false) && JSON.stringify(trigger.events) === JSON.stringify(required.events), ), ) && JSON.stringify(evidence.triggers) === JSON.stringify(expectations.expectedState.schemaControls.triggers); const catalogValid = + (latestMigration < 11 || evidence.catalog.legacySnapshotByteTotals === 0) && + evidence.catalog.invalidSnapshotByteTotals === 0 && + evidence.catalog.invalidSnapshotByteOverflows === 0 && + evidence.catalog.invalidSnapshotObjectGraph === 0 && + evidence.catalog.invalidRevisionHashes === 0 && + evidence.catalog.invalidSnapshotReconciliations === 0 && evidence.catalog.invalidSnapshotCounts === 0 && evidence.catalog.invalidPublishedPointers === 0 && evidence.catalog.invalidContentLengths === 0 && @@ -556,9 +610,180 @@ export function databaseStateExpectation( }; } +function schemaControlKey(input: { + readonly schemaName: string; + readonly tableName: string; + readonly constraintName?: string; + readonly triggerName?: string; +}): string { + return `${input.schemaName}\0${input.tableName}\0${input.constraintName ?? input.triggerName ?? ""}`; +} + +export function forwardMigrationStateExpectation( + sourceInput: unknown, + targetInput: unknown, +): DatabaseStateExpectation { + const source = EvidenceSchema.parse(sourceInput); + const target = EvidenceSchema.parse(targetInput); + const sourceLatest = Number(source.migrations.at(-1)?.version ?? "0"); + const targetLatest = Number(target.migrations.at(-1)?.version ?? "0"); + if ( + targetLatest <= sourceLatest || + JSON.stringify(target.migrations.slice(0, source.migrations.length)) !== + JSON.stringify(source.migrations) + ) + throw new Error("Forward migration inventory is not an exact extension"); + + const targetConstraints = new Map( + target.constraints.map((control) => [schemaControlKey(control), control]), + ); + if (targetConstraints.size !== target.constraints.length) + throw new Error("Forward migration produced duplicate schema constraints"); + const changedConstraint = source.constraints.find((control) => { + if ( + JSON.stringify(targetConstraints.get(schemaControlKey(control))) === + JSON.stringify(control) + ) + return false; + if ( + control.constraintName !== + "external_current_revision_classifications_latest_event_id_fkey" || + sourceLatest >= 10 || + targetLatest < 10 + ) + return true; + const replacement = targetConstraints.get( + "public\0external_current_revision_classifications\0external_current_revision_event_subject_valid", + ); + return ( + replacement?.constraintType !== "foreign-key" || !replacement.validated + ); + }); + if (changedConstraint !== undefined) + throw new Error( + `Forward migration changed pre-existing constraint ${changedConstraint.constraintName}`, + ); + const targetTriggers = new Map( + target.triggers.map((control) => [schemaControlKey(control), control]), + ); + if (targetTriggers.size !== target.triggers.length) + throw new Error("Forward migration produced duplicate schema triggers"); + const changedTrigger = source.triggers.find((control) => { + const migrated = targetTriggers.get(schemaControlKey(control)); + if (migrated === undefined) return true; + if (JSON.stringify(migrated) === JSON.stringify(control)) return false; + if ( + control.triggerName !== "external_classification_transition_valid" || + sourceLatest >= 10 || + targetLatest < 10 + ) + return true; + const { + functionBodySha256: _sourceBody, + functionDefinition: _sourceDefinition, + ...sourceIdentity + } = control; + const { + functionBodySha256: _targetBody, + functionDefinition: _targetDefinition, + ...targetIdentity + } = migrated; + return JSON.stringify(targetIdentity) !== JSON.stringify(sourceIdentity); + }); + if (changedTrigger !== undefined) + throw new Error( + `Forward migration changed pre-existing trigger ${changedTrigger.triggerName}`, + ); + + return { + ...databaseStateExpectation(source), + schemaControls: databaseStateExpectation(target).schemaControls, + }; +} + function restoredDatabaseEvidenceQuery(installationAccountId: string): string { const accountId = z.uuid().parse(installationAccountId); - return `SELECT json_build_object( + return `WITH snapshot_totals AS ( + SELECT + snapshot.id AS snapshot_id, + snapshot.revision_count, + snapshot.candidate_count, + snapshot.quarantine_count, + snapshot.resource_count, + snapshot.dependency_count, + snapshot.decoded_bytes, + count(observation.revision_id)::numeric AS observed_revision_count, + count(DISTINCT observation.revision_id)::numeric + AS distinct_observed_revision_count, + COALESCE(sum( + CASE WHEN observation.revision_id IS NULL THEN 0 + ELSE octet_length(revision.canonical_bytes)::numeric END + ), 0) AS canonical_total, + COALESCE(sum( + CASE WHEN observation.revision_id IS NULL THEN 0 + ELSE instructions.byte_length::numeric + + COALESCE(resources.total_bytes, 0) END + ), 0) AS legacy_total, + COALESCE(sum(COALESCE(resources.resource_count, 0)), 0)::numeric + AS observed_resource_count, + COALESCE(sum(COALESCE(dependencies.dependency_count, 0)), 0)::numeric + AS observed_dependency_count, + count(*) FILTER ( + WHERE (observation.revision_id IS NULL + AND observation.snapshot_id IS NOT NULL + AND ( + observation.result NOT IN ('missing','quarantined') + OR observation.observed_content_identity_sha256 IS NOT NULL + )) + OR (observation.revision_id IS NOT NULL AND ( + revision.id IS NULL + OR instructions.sha256 IS NULL + OR candidate.id IS NULL + OR candidate.snapshot_id <> observation.snapshot_id + OR candidate.skill_identity_id IS NOT NULL + AND candidate.skill_identity_id <> observation.skill_identity_id + OR candidate.published_revision_id IS NOT NULL + AND candidate.published_revision_id <> observation.revision_id + OR observation.skill_identity_id <> revision.skill_identity_id + OR observation.observed_content_identity_sha256 IS DISTINCT FROM + revision.content_identity_sha256 + OR COALESCE(resources.invalid_object_count, 0) <> 0 + )) + )::numeric AS invalid_object_count + FROM external_source_snapshots snapshot + LEFT JOIN external_snapshot_skill_observations observation + ON observation.snapshot_id=snapshot.id + LEFT JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + LEFT JOIN external_content_objects instructions + ON instructions.sha256=revision.instructions_sha256 + AND instructions.kind='instructions' + LEFT JOIN external_import_candidates candidate + ON candidate.id=observation.candidate_id + LEFT JOIN LATERAL ( + SELECT + count(*)::numeric AS resource_count, + COALESCE(sum(resource.byte_length::numeric), 0) AS total_bytes, + count(*) FILTER ( + WHERE content.sha256 IS NULL + OR content.kind <> 'resource' + OR content.byte_length <> resource.byte_length + )::numeric AS invalid_object_count + FROM external_revision_resources resource + LEFT JOIN external_content_objects content + ON content.sha256=resource.content_sha256 + WHERE resource.revision_id=revision.id + ) resources ON true + LEFT JOIN LATERAL ( + SELECT count(*)::numeric AS dependency_count + FROM external_revision_dependencies dependency + WHERE dependency.revision_id=revision.id + ) dependencies ON true + GROUP BY snapshot.id + ), latest_schema AS ( + SELECT COALESCE(max(version), '000') AS version FROM schema_migrations + ) + SELECT json_build_object( 'currentDatabase', current_database(), 'inRecovery', pg_is_in_recovery(), 'transactionReadOnly', current_setting('transaction_read_only'), @@ -583,9 +808,11 @@ function restoredDatabaseEvidenceQuery(installationAccountId: string): string { 'functionName', function_entry.proname, 'functionArguments', pg_get_function_identity_arguments(function_entry.oid), 'functionDefinition', pg_get_functiondef(function_entry.oid), - 'functionBodySha256', encode(sha256(convert_to(btrim(function_entry.prosrc), 'UTF8')), 'hex'), + 'functionBodySha256', encode(sha256(convert_to(regexp_replace(function_entry.prosrc, '^[[:space:]]+|[[:space:]]+$', '', 'g'), 'UTF8')), 'hex'), 'timing', CASE WHEN (trigger_entry.tgtype & 64)<>0 THEN 'INSTEAD OF' WHEN (trigger_entry.tgtype & 2)<>0 THEN 'BEFORE' ELSE 'AFTER' END, 'level', CASE WHEN (trigger_entry.tgtype & 1)<>0 THEN 'ROW' ELSE 'STATEMENT' END, + 'deferrable', trigger_entry.tgdeferrable, + 'initiallyDeferred', trigger_entry.tginitdeferred, 'events', array_remove(ARRAY[ CASE WHEN (trigger_entry.tgtype & 4)<>0 THEN 'INSERT' END, CASE WHEN (trigger_entry.tgtype & 8)<>0 THEN 'DELETE' END, @@ -608,7 +835,19 @@ function restoredDatabaseEvidenceQuery(installationAccountId: string): string { 'dependencyCount', (SELECT count(*) FROM external_revision_dependencies), 'contentObjectCount', (SELECT count(*) FROM external_content_objects), 'identitySha256', (SELECT encode(sha256(convert_to(COALESCE(string_agg(identity,'\n' ORDER BY identity),''),'UTF8')),'hex') FROM (SELECT 'snapshot:'||id::text||':'||source_id::text||':'||commit_sha||':'||tree_sha||':'||COALESCE(advisory_chain_head_sha256,'null') AS identity FROM external_source_snapshots UNION ALL SELECT 'revision:'||id::text||':'||snapshot_id::text||':'||bundle_sha256||':'||content_identity_sha256 FROM external_skill_revisions UNION ALL SELECT 'resource:'||revision_id::text||':'||resource_path||':'||content_sha256 FROM external_revision_resources UNION ALL SELECT 'dependency:'||revision_id::text||':'||target_revision_id::text||':'||evidence_source_sha256 FROM external_revision_dependencies UNION ALL SELECT 'content:'||sha256||':'||byte_length::text FROM external_content_objects) catalog_identity), - 'invalidSnapshotCounts', (SELECT count(*) FROM external_source_snapshots snapshot WHERE snapshot.revision_count<>(SELECT count(*) FROM external_skill_revisions revision WHERE revision.snapshot_id=snapshot.id) OR snapshot.candidate_count<>(SELECT count(*) FROM external_import_candidates candidate WHERE candidate.snapshot_id=snapshot.id) OR snapshot.quarantine_count<>(SELECT count(*) FROM external_import_candidates candidate JOIN external_current_classifications current ON current.candidate_id=candidate.id WHERE candidate.snapshot_id=snapshot.id AND current.classification='quarantined') OR snapshot.resource_count<>(SELECT count(*) FROM external_revision_resources resource JOIN external_skill_revisions revision ON revision.id=resource.revision_id WHERE revision.snapshot_id=snapshot.id) OR snapshot.dependency_count<>(SELECT count(*) FROM external_revision_dependencies dependency JOIN external_skill_revisions revision ON revision.id=dependency.revision_id WHERE revision.snapshot_id=snapshot.id) OR snapshot.decoded_bytes<>(SELECT COALESCE(sum(octet_length(revision.canonical_bytes)),0) FROM external_skill_revisions revision WHERE revision.snapshot_id=snapshot.id)), + 'legacySnapshotByteTotals', (SELECT count(*) FROM snapshot_totals totals, latest_schema WHERE latest_schema.version<'011' AND totals.legacy_total<>totals.canonical_total AND totals.decoded_bytes=totals.legacy_total), + 'invalidSnapshotByteTotals', (SELECT count(*) FROM snapshot_totals totals, latest_schema WHERE (latest_schema.version<'011' AND totals.decoded_bytes NOT IN (totals.canonical_total,totals.legacy_total)) OR (latest_schema.version>='011' AND totals.decoded_bytes<>totals.canonical_total)), + 'invalidSnapshotByteOverflows', (SELECT count(*) FROM snapshot_totals WHERE canonical_total>9223372036854775807::numeric OR legacy_total>9223372036854775807::numeric), + 'invalidSnapshotObjectGraph', (SELECT count(*) FROM snapshot_totals WHERE invalid_object_count<>0 OR observed_revision_count<>distinct_observed_revision_count), + 'invalidRevisionHashes', (SELECT count(*) FROM external_skill_revisions WHERE bundle_sha256<>encode(sha256(convert_to(canonical_bytes,'UTF8')),'hex')), + 'invalidSnapshotReconciliations', (SELECT CASE + WHEN version<'011' THEN 0 + WHEN to_regclass('public.external_snapshot_byte_total_reconciliations') IS NULL THEN 1 + ELSE cardinality(xpath('/table/row', query_to_xml( + 'SELECT invalid.snapshot_id FROM (SELECT snapshot.id AS snapshot_id FROM external_source_snapshots snapshot LEFT JOIN external_snapshot_byte_total_reconciliations reconciliation ON reconciliation.snapshot_id=snapshot.id CROSS JOIN LATERAL (SELECT COALESCE(sum(octet_length(revision.canonical_bytes)::numeric),0) AS canonical_total,COALESCE(sum(instructions.byte_length::numeric + COALESCE((SELECT sum(resource.byte_length::numeric) FROM external_revision_resources resource WHERE resource.revision_id=revision.id),0)),0) AS legacy_total FROM external_snapshot_skill_observations observation JOIN external_skill_revisions revision ON revision.id=observation.revision_id JOIN external_content_objects instructions ON instructions.sha256=revision.instructions_sha256 WHERE observation.snapshot_id=snapshot.id) totals WHERE reconciliation.snapshot_id IS NULL OR (SELECT count(*) FROM external_snapshot_byte_total_reconciliations duplicate WHERE duplicate.snapshot_id=snapshot.id)<>1 OR reconciliation.migration_version<>''011'' OR reconciliation.reconciled_decoded_bytes<>snapshot.decoded_bytes OR reconciliation.reconciled_decoded_bytes<>totals.canonical_total OR reconciliation.legacy_payload_decoded_bytes<>totals.legacy_total OR reconciliation.prior_representation=''canonical'' AND reconciliation.prior_decoded_bytes<>totals.canonical_total OR reconciliation.prior_representation=''legacy-payload'' AND reconciliation.prior_decoded_bytes<>totals.legacy_total OR reconciliation.prior_representation NOT IN (''canonical'',''legacy-payload'') UNION ALL SELECT reconciliation.snapshot_id FROM external_snapshot_byte_total_reconciliations reconciliation LEFT JOIN external_source_snapshots snapshot ON snapshot.id=reconciliation.snapshot_id WHERE snapshot.id IS NULL) invalid', + false, false, ''))) + END FROM latest_schema), + 'invalidSnapshotCounts', (SELECT count(*) FROM snapshot_totals totals WHERE totals.revision_count<>totals.observed_revision_count OR totals.candidate_count<>(SELECT count(*) FROM external_import_candidates candidate WHERE candidate.snapshot_id=totals.snapshot_id) OR totals.quarantine_count<>(SELECT count(*) FROM external_import_candidates candidate JOIN external_current_classifications current ON current.candidate_id=candidate.id WHERE candidate.snapshot_id=totals.snapshot_id AND current.classification='quarantined') OR totals.resource_count<>totals.observed_resource_count OR totals.dependency_count<>totals.observed_dependency_count), 'invalidPublishedPointers', (SELECT count(*) FROM github_sources source JOIN external_source_snapshots snapshot ON snapshot.id=source.current_published_snapshot_id WHERE snapshot.source_id<>source.id), 'invalidContentLengths', (SELECT count(*) FROM external_content_objects WHERE byte_length<>octet_length(content)), 'invalidContentHashes', (SELECT count(*) FROM external_content_objects WHERE sha256<>encode(sha256(convert_to(content,'UTF8')),'hex')), diff --git a/src/onboarding/adapters/postgres/schema-compatibility.ts b/src/onboarding/adapters/postgres/schema-compatibility.ts index fdd8e07..b992855 100644 --- a/src/onboarding/adapters/postgres/schema-compatibility.ts +++ b/src/onboarding/adapters/postgres/schema-compatibility.ts @@ -41,7 +41,7 @@ export function classifySchemaUpgrade(input: { { length: input.latestMigration - input.liveSchema }, (_, index) => input.liveSchema + index + 1, ); - const forwardOnly = new Set(input.forwardOnlyMigrations ?? [10]); + const forwardOnly = new Set(input.forwardOnlyMigrations ?? [10, 11]); if (!migrations.some((migration) => forwardOnly.has(migration))) throw new Error("Unclassified forward migration is forbidden"); return { diff --git a/src/onboarding/adapters/postgres/service-database.ts b/src/onboarding/adapters/postgres/service-database.ts index 18055cf..5448c8e 100644 --- a/src/onboarding/adapters/postgres/service-database.ts +++ b/src/onboarding/adapters/postgres/service-database.ts @@ -84,9 +84,9 @@ export class ServiceDatabase { if ( version === undefined || !version.startsWith("17.") || - latestMigration !== "010" + latestMigration !== "011" ) - throw new Error("PostgreSQL 17/migration 010 readiness gate failed"); + throw new Error("PostgreSQL 17/migration 011 readiness gate failed"); return { version, latestMigration }; } } diff --git a/src/onboarding/application/production-lifecycle.ts b/src/onboarding/application/production-lifecycle.ts index 40ee7bc..1038ddb 100644 --- a/src/onboarding/application/production-lifecycle.ts +++ b/src/onboarding/application/production-lifecycle.ts @@ -68,6 +68,7 @@ import { assessRestoredDatabaseEvidence, databaseStateExpectation, expectedMigrationInventory, + forwardMigrationStateExpectation, readDatabaseEvidence, validateRestoredDatabaseContainer, } from "../adapters/postgres/restore-validation.js"; @@ -2709,7 +2710,13 @@ async function upgradeOperation( installationAccountId: installation.accountId, expectedActiveApiKeys: activeCredentialReferenceCount, expectedDatabase: "skillwire", - expectedState, + expectedState: + liveSchema === candidate.target.latestMigration + ? expectedState + : forwardMigrationStateExpectation( + sourceEvidence, + targetEvidence, + ), }); }, verifyClients: async () => { diff --git a/src/onboarding/application/upgrade.ts b/src/onboarding/application/upgrade.ts index 9b2849c..d63096f 100644 --- a/src/onboarding/application/upgrade.ts +++ b/src/onboarding/application/upgrade.ts @@ -125,7 +125,7 @@ export async function runUpgrade(options: { schemaMinimum: target.schemaMinimum, schemaMaximum: target.schemaMaximum, latestMigration: target.latestMigration, - forwardOnlyMigrations: [10], + forwardOnlyMigrations: [10, 11], }); const backup = await effect( "upgrade-backup", diff --git a/src/onboarding/domain/release-components.ts b/src/onboarding/domain/release-components.ts index 13f3210..567d990 100644 --- a/src/onboarding/domain/release-components.ts +++ b/src/onboarding/domain/release-components.ts @@ -35,11 +35,11 @@ export function deriveReleaseComponents(payload: Payload): ReleaseComponents { const versions = migrations.map(({ path }) => path.slice("migrations/".length, "migrations/".length + 3), ); - const expectedVersions = Array.from({ length: 10 }, (_value, index) => + const expectedVersions = Array.from({ length: 11 }, (_value, index) => String(index + 1).padStart(3, "0"), ); if (versions.join("\0") !== expectedVersions.join("\0")) { - throw new Error("Release migration set must bind exact migrations 001-010"); + throw new Error("Release migration set must bind exact migrations 001-011"); } const catalogEntries = entriesBelow(payload, ["catalog/"]); const advisory = payload.find( @@ -70,9 +70,9 @@ export function deriveReleaseComponents(payload: Payload): ReleaseComponents { }, migrations: { sha256: aggregate(migrations), - count: 10, - latest: "010", - forwardOnly: ["010"], + count: 11, + latest: "011", + forwardOnly: ["010", "011"], }, catalog: { sha256: aggregate(catalogEntries), diff --git a/src/onboarding/domain/release-manifest.ts b/src/onboarding/domain/release-manifest.ts index 3767edd..1c96cdd 100644 --- a/src/onboarding/domain/release-manifest.ts +++ b/src/onboarding/domain/release-manifest.ts @@ -42,9 +42,9 @@ export const ReleaseComponentsSchema = z migrations: z .object({ sha256: Sha256Schema, - count: z.literal(10), - latest: z.literal("010"), - forwardOnly: z.tuple([z.literal("010")]), + count: z.literal(11), + latest: z.literal("011"), + forwardOnly: z.tuple([z.literal("010"), z.literal("011")]), }) .strict(), catalog: z diff --git a/src/persistence/postgres/external-catalog-store.ts b/src/persistence/postgres/external-catalog-store.ts index c9eb5c9..269ca00 100644 --- a/src/persistence/postgres/external-catalog-store.ts +++ b/src/persistence/postgres/external-catalog-store.ts @@ -330,6 +330,47 @@ async function snapshotMatchesCandidateInputs( ); } +async function authoritativeCanonicalByteTotal( + client: PoolClient, + sourceId: string, + revisions: readonly ExternalSkillRevision[], +): Promise { + const total = await client.query<{ canonical_byte_total: string }>( + `WITH input AS ( + SELECT * + FROM unnest($2::text[],$3::text[],$4::bigint[]) + AS planned(normalized_skill_root,content_identity_sha256,fallback_bytes) + ) + SELECT COALESCE(sum(COALESCE( + octet_length(stored.canonical_bytes)::numeric, + input.fallback_bytes::numeric + )),0)::text AS canonical_byte_total + FROM input + LEFT JOIN LATERAL ( + SELECT revision.canonical_bytes + FROM external_skill_identities identity + JOIN external_skill_revisions revision + ON revision.skill_identity_id=identity.id + WHERE identity.source_id=$1 + AND identity.normalized_skill_root=input.normalized_skill_root + AND revision.content_identity_sha256=input.content_identity_sha256 + ) stored ON true`, + [ + sourceId, + revisions.map(({ provenance }) => skillRoot(provenance.skillPath)), + revisions.map(({ contentIdentitySha256 }) => contentIdentitySha256), + revisions.map(({ canonicalBytes }) => + Buffer.byteLength(canonicalBytes, "utf8"), + ), + ], + ); + const canonicalByteTotal = Number(total.rows[0]?.canonical_byte_total); + if (!Number.isSafeInteger(canonicalByteTotal)) { + throw new Error("INVALID_REVISION_BATCH"); + } + return canonicalByteTotal; +} + export class PostgresExternalCatalogStore implements ExternalCatalogStore { readonly #sources: PostgresGitHubSourceStore; @@ -374,6 +415,27 @@ export class PostgresExternalCatalogStore implements ExternalCatalogStore { const validationInputSha256 = input.validationInputSha256 ?? snapshotValidationInputSha256(input, candidates, true); + const proposedCanonicalByteTotal = input.revisions.reduce( + (total, revision) => + total + Buffer.byteLength(revision.canonicalBytes, "utf8"), + 0, + ); + const legacyPayloadByteTotal = input.revisions.reduce( + (total, revision) => + total + + Buffer.byteLength(revision.instructions, "utf8") + + revision.resources.reduce( + (resourceTotal, resource) => resourceTotal + resource.byteLength, + 0, + ), + 0, + ); + if ( + !Number.isSafeInteger(proposedCanonicalByteTotal) || + !Number.isSafeInteger(legacyPayloadByteTotal) + ) { + throw new Error("INVALID_REVISION_BATCH"); + } return requestTransaction(this.pool, context, async (client) => { if (input.lease !== undefined) await assertLeaseHeld(client, input.lease); const advisoryHead = await client.query<{ last_event_sha256: string }>( @@ -532,6 +594,12 @@ export class PostgresExternalCatalogStore implements ExternalCatalogStore { throw new Error("PUBLICATION_CONFLICT"); } + const canonicalByteTotal = await authoritativeCanonicalByteTotal( + client, + input.sourceId, + input.revisions, + ); + const snapshotId = randomUUID(); await client.query( ` @@ -564,17 +632,7 @@ export class PostgresExternalCatalogStore implements ExternalCatalogStore { (count, revision) => count + revision.dependencies.length, 0, ), - input.revisions.reduce( - (count, revision) => - count + - Buffer.byteLength(revision.instructions, "utf8") + - revision.resources.reduce( - (resourceCount, resource) => - resourceCount + resource.byteLength, - 0, - ), - 0, - ), + canonicalByteTotal, validationInputSha256, null, input.observedRepository?.repositoryId ?? @@ -585,6 +643,13 @@ export class PostgresExternalCatalogStore implements ExternalCatalogStore { input.revisions[0]?.provenance.repository, ], ); + await client.query( + `INSERT INTO external_snapshot_byte_total_reconciliations ( + snapshot_id,prior_decoded_bytes,legacy_payload_decoded_bytes, + reconciled_decoded_bytes,prior_representation + ) VALUES ($1,$2,$3,$2,'canonical')`, + [snapshotId, canonicalByteTotal, legacyPayloadByteTotal], + ); const traces: ImportTraceResult[] = []; const candidateTraces: CandidateTraceResult[] = []; diff --git a/tests/contract/release/self-hosted-release.test.ts b/tests/contract/release/self-hosted-release.test.ts index 05e2aff..2135610 100644 --- a/tests/contract/release/self-hosted-release.test.ts +++ b/tests/contract/release/self-hosted-release.test.ts @@ -66,9 +66,9 @@ describe("reproducible self-hosted release", () => { "distribution/self-hosted/compose.yaml", ); expect(manifest.components.migrations).toMatchObject({ - count: 10, - latest: "010", - forwardOnly: ["010"], + count: 11, + latest: "011", + forwardOnly: ["010", "011"], }); expect(manifest.components.catalog.firstPartyRevisionCount).toBe(10); expect(manifest.components.catalog.advisorySha256).toMatch( diff --git a/tests/e2e/self-hosted-onboarding/first-party-catalog.test.ts b/tests/e2e/self-hosted-onboarding/first-party-catalog.test.ts index ae08159..c26a01e 100644 --- a/tests/e2e/self-hosted-onboarding/first-party-catalog.test.ts +++ b/tests/e2e/self-hosted-onboarding/first-party-catalog.test.ts @@ -56,7 +56,7 @@ async function catalogRelease(root: string): Promise<{ sha256: "0".repeat(64), mode: "0644" as const, }; - const migrations = Array.from({ length: 10 }, (_value, index) => ({ + const migrations = Array.from({ length: 11 }, (_value, index) => ({ path: `migrations/${String(index + 1).padStart(3, "0")}_fixture.sql`, size: 1, sha256: String(index).padStart(64, "0"), diff --git a/tests/helpers/self-hosted-release-fixtures.ts b/tests/helpers/self-hosted-release-fixtures.ts index ec8049d..93a05be 100644 --- a/tests/helpers/self-hosted-release-fixtures.ts +++ b/tests/helpers/self-hosted-release-fixtures.ts @@ -60,7 +60,7 @@ export const RELEASE_PAYLOAD_FILES: Readonly> = { "claude-plugin", ...catalogFixtureFiles(), ...Object.fromEntries( - Array.from({ length: 10 }, (_value, index) => { + Array.from({ length: 11 }, (_value, index) => { const version = String(index + 1).padStart(3, "0"); return [`migrations/${version}_fixture.sql`, `migration-${version}`]; }), @@ -155,8 +155,8 @@ export function releaseManifestFixture( compatibility: { node: "24.18.0", postgresql: "17.10", - schemaMinimum: 10, - schemaMaximum: 10, + schemaMinimum: 9, + schemaMaximum: 11, }, feature003Integrity: { path: "distribution/codex-marketplace/release-integrity.json", diff --git a/tests/integration/github-ingestion/registered-source-ingestion.test.ts b/tests/integration/github-ingestion/registered-source-ingestion.test.ts index 66e028f..c84e270 100644 --- a/tests/integration/github-ingestion/registered-source-ingestion.test.ts +++ b/tests/integration/github-ingestion/registered-source-ingestion.test.ts @@ -133,6 +133,31 @@ describe("registered mattpocock/skills ingestion", () => { row.bundle_sha256, ), ).toBe(true); + const byteTotals = await database.pool.query<{ + decoded_bytes: string; + reconciled_decoded_bytes: string | null; + }>( + ` + SELECT snapshot.decoded_bytes::text, + reconciliation.reconciled_decoded_bytes::text + FROM external_source_snapshots snapshot + LEFT JOIN external_snapshot_byte_total_reconciliations reconciliation + ON reconciliation.snapshot_id=snapshot.id + WHERE snapshot.id=$1 + `, + [published.snapshotId], + ); + const canonicalByteTotal = stored.rows.reduce( + (total, { canonical_bytes }) => + total + Buffer.byteLength(canonical_bytes, "utf8"), + 0, + ); + expect(byteTotals.rows).toEqual([ + { + decoded_bytes: String(canonicalByteTotal), + reconciled_decoded_bytes: String(canonicalByteTotal), + }, + ]); const expectedInvocationModes = new Map( fixture.inventory.skills.map(({ name, userOnly }) => [ name, diff --git a/tests/integration/github-ingestion/synchronization.test.ts b/tests/integration/github-ingestion/synchronization.test.ts index 01d8401..b81d299 100644 --- a/tests/integration/github-ingestion/synchronization.test.ts +++ b/tests/integration/github-ingestion/synchronization.test.ts @@ -1411,6 +1411,34 @@ describe("immutable source synchronization", () => { classification: "quarantined", reasonCodes: ["DEPENDENCY_MISSING"], }); + const secondByteTotals = await database.pool.query<{ + canonical_total: string; + decoded_bytes: string; + prior_decoded_bytes: string; + reconciled_decoded_bytes: string; + }>( + `SELECT + sum(octet_length(revision.canonical_bytes))::text AS canonical_total, + snapshot.decoded_bytes::text, + reconciliation.prior_decoded_bytes::text, + reconciliation.reconciled_decoded_bytes::text + FROM external_source_snapshots snapshot + JOIN external_snapshot_skill_observations observation + ON observation.snapshot_id=snapshot.id + JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + JOIN external_snapshot_byte_total_reconciliations reconciliation + ON reconciliation.snapshot_id=snapshot.id + WHERE snapshot.id=$1 + GROUP BY snapshot.id,reconciliation.snapshot_id`, + [second.snapshotId], + ); + expect(secondByteTotals.rows[0]).toEqual({ + canonical_total: secondByteTotals.rows[0]?.canonical_total, + decoded_bytes: secondByteTotals.rows[0]?.canonical_total, + prior_decoded_bytes: secondByteTotals.rows[0]?.canonical_total, + reconciled_decoded_bytes: secondByteTotals.rows[0]?.canonical_total, + }); const initialTransitions = await database.pool.query<{ previous_classification: string | null; next_classification: string; diff --git a/tests/integration/onboarding/backup-restore-validation.test.ts b/tests/integration/onboarding/backup-restore-validation.test.ts index 3d391e6..d7d3819 100644 --- a/tests/integration/onboarding/backup-restore-validation.test.ts +++ b/tests/integration/onboarding/backup-restore-validation.test.ts @@ -40,7 +40,7 @@ function localDockerContext( }; } -const completeValidation = (latestMigration = "010") => ({ +const completeValidation = (latestMigration = "011") => ({ latestMigration, migrationInventoryValid: true, constraintsValid: true, @@ -501,7 +501,7 @@ describe("restore-validated PostgreSQL backup", () => { environment: dockerEnvironment, stdin: [ "CREATE TABLE schema_migrations (version text PRIMARY KEY);", - "INSERT INTO schema_migrations(version) VALUES ('010');", + "INSERT INTO schema_migrations(version) VALUES ('011');", "CREATE TABLE accounts (id uuid PRIMARY KEY);", "CREATE TABLE external_skill_revisions (id uuid PRIMARY KEY);", "CREATE TABLE external_advisory_chain_head (singleton boolean PRIMARY KEY);", @@ -543,7 +543,7 @@ describe("restore-validated PostgreSQL backup", () => { ...completeValidation(migration ?? ""), constraintsValid: accounts === "true", catalogValid: catalog === "true" && advisory === "true", - ready: migration === "010", + ready: migration === "011", }; }, }); @@ -554,7 +554,7 @@ describe("restore-validated PostgreSQL backup", () => { expect(backup.archiveSha256).toMatch(/^[0-9a-f]{64}$/); expect(backup).toMatchObject({ validation: { - latestMigration: "010", + latestMigration: "011", migrationInventoryValid: true, constraintsValid: true, catalogValid: true, diff --git a/tests/integration/onboarding/service-setup.test.ts b/tests/integration/onboarding/service-setup.test.ts index 0efd97b..006dea1 100644 --- a/tests/integration/onboarding/service-setup.test.ts +++ b/tests/integration/onboarding/service-setup.test.ts @@ -5,6 +5,7 @@ import { resolve } from "node:path"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { DeploymentAdapter } from "../../../src/onboarding/adapters/docker/deployment.js"; +import { ServiceDatabase } from "../../../src/onboarding/adapters/postgres/service-database.js"; import type { CommandOptions, CommandResult, @@ -20,6 +21,23 @@ describe("service-only deployment boundary", () => { afterAll(async () => { await rm(runtimeDirectory, { recursive: true, force: true }); }); + it("requires PostgreSQL 17 with the reconciled migration 011 schema", async () => { + const run = vi.fn((): Promise => + Promise.resolve(result("17.10|011\n")), + ); + const database = new ServiceDatabase({ + dockerExecutable: "/usr/bin/docker", + projectName: "skillwire-test-0123456789abcdef", + volumeName: "skillwire-test-0123456789abcdef_postgres_data", + composePath: "/tmp/disposable/compose.yaml", + run, + }); + + await expect(database.verifySchemaAndReadiness()).resolves.toEqual({ + version: "17.10", + latestMigration: "011", + }); + }); it("pulls each exact digest only when it is absent from a clean local cache", async () => { const cached = new Set(); const calls: string[][] = []; diff --git a/tests/integration/onboarding/upgrade-forward-only-010.test.ts b/tests/integration/onboarding/upgrade-forward-only-010.test.ts index 691a62c..91d428c 100644 --- a/tests/integration/onboarding/upgrade-forward-only-010.test.ts +++ b/tests/integration/onboarding/upgrade-forward-only-010.test.ts @@ -10,6 +10,53 @@ import { import type { UpgradeRecoveryError } from "../../../src/onboarding/application/upgrade.js"; describe("forward-only migration 010 upgrade", () => { + it("treats migration 011 reconciliation as a forward-only boundary", async () => { + const target = { + releaseId: "11-amd64", + releaseSequence: 11, + trustPolicySequence: 4, + schemaMinimum: 9, + schemaMaximum: 11, + latestMigration: 11, + manifestSha256: "1".repeat(64), + imageDigest: `sha256:${"2".repeat(64)}`, + }; + const preview = previewUpgrade({ + installationId: randomUUID(), + currentReleaseSequence: 10, + currentTrustPolicySequence: 4, + liveSchema: 10, + target, + }); + const drainWriters = vi.fn(async () => undefined); + const migrate = vi.fn(async () => undefined); + + await expect( + runUpgrade({ + preview, + confirmation: preview.previewHash, + signal: new AbortController().signal, + verifyTarget: async () => target, + createBackup: async () => ({ + backupId: randomUUID(), + validated: true, + }), + drainWriters, + installApplication: async () => undefined, + migrate, + verifyLiveSchema: async () => 11, + preActivationReadiness: async () => undefined, + verifyClients: async () => undefined, + activateApplication: async () => undefined, + commitSelection: async () => undefined, + rollbackApplication: async () => undefined, + stopWriters: async () => undefined, + }), + ).resolves.toMatchObject({ releaseId: target.releaseId }); + expect(drainWriters).toHaveBeenCalledTimes(1); + expect(migrate).toHaveBeenCalledTimes(1); + }); + it("keeps the public writer stopped through preactivation and client gates", async () => { const target = { releaseId: "10-amd64", diff --git a/tests/integration/postgres/external-policy-migration.test.ts b/tests/integration/postgres/external-policy-migration.test.ts index 8cb9b0c..111ef06 100644 --- a/tests/integration/postgres/external-policy-migration.test.ts +++ b/tests/integration/postgres/external-policy-migration.test.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -34,6 +34,10 @@ interface HistoricalRevisionFixture { readonly terminalEventId: string; } +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + async function insertHistoricalRevision( pool: Pool, sourceId: string, @@ -51,18 +55,21 @@ async function insertHistoricalRevision( classification === "curated" ? randomUUID() : verifiedEventId; const key = ordinal.toString(16); const root = `${classification}-${String(ordinal)}`; + const canonicalBytes = JSON.stringify({ ordinal }); await pool.query( `INSERT INTO external_source_snapshots ( id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + candidate_count,quarantine_count,decoded_bytes, advisory_chain_head_sha256,origin_github_repository_id, origin_owner,origin_repository - ) VALUES ($1,$2,$3,$4,'nested-v1',1,$5,7002,'fixture-org', + ) VALUES ($1,$2,$3,$4,'nested-v1',1,1,$5,2,$6,7002,'fixture-org', 'migration-shapes')`, [ snapshotId, sourceId, key.repeat(40), ((ordinal + 8) % 16).toString(16).repeat(40), + classification === "quarantined" ? 1 : 0, "0".repeat(64), ], ); @@ -79,20 +86,21 @@ async function insertHistoricalRevision( source_owner,spdx_license_id,license_sha256,instructions_sha256, invocation_mode,canonical_bytes,origin_owner,origin_repository ) VALUES ($1,$2,$3,$4,$5,$6,$7,'Migration shape fixture.',$8,$9, - 'Fixture Owner','MIT',$10,$11,'automatic','{}', + 'Fixture Owner','MIT',$10,$11,'automatic',$12, 'fixture-org','migration-shapes')`, [ revisionId, identityId, snapshotId, `gh-${key.repeat(64)}`, - ((ordinal + 1) % 16).toString(16).repeat(64), + sha256(canonicalBytes), ((ordinal + 2) % 16).toString(16).repeat(64), `${root}-skill`, `${root}/SKILL.md`, key.repeat(40), - "e".repeat(64), - "f".repeat(64), + sha256("MIT"), + sha256("ok"), + canonicalBytes, ], ); await pool.query( @@ -202,9 +210,10 @@ async function insertHistoricalSibling( await pool.query( `INSERT INTO external_source_snapshots ( id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + candidate_count,decoded_bytes, advisory_chain_head_sha256,origin_github_repository_id, origin_owner,origin_repository - ) VALUES ($1,$2,$3,$4,'nested-v1',1,$5,7002,'fixture-org', + ) VALUES ($1,$2,$3,$4,'nested-v1',1,1,2,$5,7002,'fixture-org', 'migration-shapes')`, [snapshotId, sourceId, "a".repeat(40), "b".repeat(40), "0".repeat(64)], ); @@ -275,9 +284,10 @@ async function insertUnobservedHistoricalCandidate( await pool.query( `INSERT INTO external_source_snapshots ( id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + candidate_count, advisory_chain_head_sha256,origin_github_repository_id, origin_owner,origin_repository - ) VALUES ($1,$2,$3,$4,'nested-v1',0,$5,7002,'fixture-org', + ) VALUES ($1,$2,$3,$4,'nested-v1',0,1,$5,7002,'fixture-org', 'migration-shapes')`, [snapshotId, sourceId, "c".repeat(40), "d".repeat(40), "0".repeat(64)], ); @@ -339,7 +349,7 @@ describe("external policy migration", () => { const versions = await database.pool.query<{ version: string }>( "SELECT version FROM schema_migrations ORDER BY version", ); - expect(versions.rows.at(-1)?.version).toBe("010"); + expect(versions.rows.at(-1)?.version).toBe("011"); const required = [ "github_discovery_runs", "github_discovery_evidence", @@ -401,8 +411,8 @@ describe("external policy migration", () => { const discoveredEventId = randomUUID(); const verifiedEventId = randomUUID(); const curatedEventId = randomUUID(); - const instructionsSha256 = "1".repeat(64); - const licenseSha256 = "2".repeat(64); + const instructionsSha256 = sha256("ok"); + const licenseSha256 = sha256("MIT"); const contentIdentitySha256 = "3".repeat(64); await legacy.pool.query( `INSERT INTO github_sources ( @@ -415,9 +425,10 @@ describe("external policy migration", () => { await legacy.pool.query( `INSERT INTO external_source_snapshots ( id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + candidate_count,decoded_bytes, advisory_chain_head_sha256,origin_github_repository_id, origin_owner,origin_repository - ) VALUES ($1,$2,$3,$4,'nested-v1',1,$5,7001,'fixture-org', + ) VALUES ($1,$2,$3,$4,'nested-v1',1,1,2,$5,7001,'fixture-org', 'legacy-events')`, [snapshotId, sourceId, "a".repeat(40), "b".repeat(40), "0".repeat(64)], ); @@ -449,7 +460,7 @@ describe("external policy migration", () => { identityId, snapshotId, `gh-${"4".repeat(64)}`, - "5".repeat(64), + sha256("{}"), contentIdentitySha256, "a".repeat(40), licenseSha256, @@ -679,7 +690,7 @@ describe("external policy migration", () => { sha256,kind,media_type,byte_length,content ) VALUES ($1,'license','text/plain',3,'MIT'), ($2,'instructions','text/markdown',2,'ok')`, - ["e".repeat(64), "f".repeat(64)], + [sha256("MIT"), sha256("ok")], ); const verified = await insertHistoricalRevision( legacy.pool, @@ -891,7 +902,7 @@ describe("external policy migration", () => { sha256,kind,media_type,byte_length,content ) VALUES ($1,'license','text/plain',3,'MIT'), ($2,'instructions','text/markdown',2,'ok')`, - ["e".repeat(64), "f".repeat(64)], + [sha256("MIT"), sha256("ok")], ); const verified = await insertHistoricalRevision( legacy.pool, diff --git a/tests/integration/postgres/github-ingestion-migrations.test.ts b/tests/integration/postgres/github-ingestion-migrations.test.ts index 63278d4..798ed3d 100644 --- a/tests/integration/postgres/github-ingestion-migrations.test.ts +++ b/tests/integration/postgres/github-ingestion-migrations.test.ts @@ -35,6 +35,7 @@ describe("GitHub ingestion migrations", () => { "008", "009", "010", + "011", ]); expect( versions.rows.every(({ checksum }) => /^[0-9a-f]{64}$/.test(checksum)), @@ -45,6 +46,12 @@ describe("GitHub ingestion migrations", () => { expect( versions.rows.find(({ version }) => version === "010")?.checksum, ).toBe(createHash("sha256").update(migration010).digest("hex")); + const migration011 = await readFile( + join(process.cwd(), "migrations/011_reconcile_snapshot_byte_totals.sql"), + ); + expect( + versions.rows.find(({ version }) => version === "011")?.checksum, + ).toBe(createHash("sha256").update(migration011).digest("hex")); const legacyTables = await database.pool.query<{ name: string }>( ` @@ -70,15 +77,29 @@ describe("GitHub ingestion migrations", () => { `, [sourceId], ); - await database.pool.query( - ` - INSERT INTO external_source_snapshots ( - id, source_id, commit_sha, tree_sha, manifest_version, revision_count, - origin_github_repository_id, origin_owner, origin_repository - ) VALUES ($1, $2, $3, $4, '1.2.3', 25, 1148788086, 'mattpocock', 'skills') - `, - [snapshotId, sourceId, "8".repeat(40), "1".repeat(40)], - ); + const client = await database.pool.connect(); + try { + await client.query("BEGIN"); + await client.query( + ` + INSERT INTO external_source_snapshots ( + id, source_id, commit_sha, tree_sha, manifest_version, revision_count, + origin_github_repository_id, origin_owner, origin_repository + ) VALUES ($1, $2, $3, $4, '1.2.3', 0, 1148788086, 'mattpocock', 'skills') + `, + [snapshotId, sourceId, "8".repeat(40), "1".repeat(40)], + ); + await client.query( + `INSERT INTO external_snapshot_byte_total_reconciliations ( + snapshot_id,prior_decoded_bytes,legacy_payload_decoded_bytes, + reconciled_decoded_bytes,prior_representation + ) VALUES ($1,0,0,0,'canonical')`, + [snapshotId], + ); + await client.query("COMMIT"); + } finally { + client.release(); + } await expect( database.pool.query( "UPDATE external_source_snapshots SET manifest_version = '9.9.9' WHERE id = $1", diff --git a/tests/integration/postgres/legacy-snapshot-byte-totals.test.ts b/tests/integration/postgres/legacy-snapshot-byte-totals.test.ts new file mode 100644 index 0000000..4eb4cde --- /dev/null +++ b/tests/integration/postgres/legacy-snapshot-byte-totals.test.ts @@ -0,0 +1,859 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { Pool, PoolClient } from "pg"; +import { describe, expect, it } from "vitest"; + +import { runMigrations } from "../../../src/persistence/postgres/migration-runner.js"; +import { + assessRestoredDatabaseEvidence, + databaseStateExpectation, + expectedMigrationInventory, + forwardMigrationStateExpectation, + readDatabaseEvidence, +} from "../../../src/onboarding/adapters/postgres/restore-validation.js"; +import type { CommandOptions } from "../../../src/onboarding/adapters/process/command-runner.js"; +import { createTestDatabase } from "../../helpers/database.js"; + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +async function copyMigrationsThrough009(): Promise { + const source = join(process.cwd(), "migrations"); + const target = await mkdtemp(join(tmpdir(), "skillwire-byte-total-009-")); + for (const name of (await readdir(source)).filter((name) => + /^00[1-9]_.*\.sql$/.test(name), + )) { + await writeFile( + join(target, name), + await readFile(join(source, name), "utf8"), + ); + } + return target; +} + +async function insertLegacyVerifiedSnapshot( + pool: Pool | PoolClient, + options: { + readonly corruptBundleHash?: boolean; + readonly corruptContentHash?: boolean; + readonly storedByteTotal?: number; + } = {}, +): Promise<{ + readonly canonicalBytes: string; + readonly canonicalByteTotal: number; + readonly contentIdentitySha256: string; + readonly identityId: string; + readonly legacyByteTotal: number; + readonly revisionId: string; + readonly snapshotId: string; + readonly sourceId: string; +}> { + const sourceId = randomUUID(); + const snapshotId = randomUUID(); + const identityId = randomUUID(); + const revisionId = randomUUID(); + const candidateId = randomUUID(); + const reportId = randomUUID(); + const discoveredEventId = randomUUID(); + const verifiedEventId = randomUUID(); + const instructions = "legacy instructions"; + const resource = "legacy resource"; + const license = "MIT"; + const canonicalBytes = JSON.stringify({ + schemaVersion: 2, + skillId: "legacy-byte-total-skill", + instructions, + resources: [{ path: "references/example.md", content: resource }], + }); + const legacyByteTotal = + Buffer.byteLength(instructions, "utf8") + + Buffer.byteLength(resource, "utf8"); + const canonicalByteTotal = Buffer.byteLength(canonicalBytes, "utf8"); + const instructionsSha256 = options.corruptContentHash + ? "1".repeat(64) + : sha256(instructions); + expect(legacyByteTotal).not.toBe(canonicalByteTotal); + + await pool.query( + `INSERT INTO github_sources ( + id,github_repository_id,owner,repository,normalized_owner, + normalized_repository,default_branch + ) VALUES ($1,9001,'fixture-owner','fixture-repository','fixture-owner', + 'fixture-repository','main')`, + [sourceId], + ); + await pool.query( + `INSERT INTO external_source_snapshots ( + id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + adapter_kind,candidate_count,quarantine_count,resource_count, + dependency_count,decoded_bytes,validation_input_sha256, + advisory_chain_head_sha256,origin_github_repository_id,origin_owner, + origin_repository + ) VALUES ($1,$2,$3,$4,'nested-v1',1,'nested-skill',1,0,1,0,$5,$6,$7, + 9001,'fixture-owner','fixture-repository')`, + [ + snapshotId, + sourceId, + "a".repeat(40), + "b".repeat(40), + options.storedByteTotal ?? legacyByteTotal, + "c".repeat(64), + "0".repeat(64), + ], + ); + await pool.query( + `INSERT INTO external_content_objects ( + sha256,kind,media_type,byte_length,content + ) VALUES + ($1,'instructions','text/markdown',$2,$3), + ($4,'resource','text/markdown',$5,$6), + ($7,'license','text/plain',$8,$9)`, + [ + instructionsSha256, + Buffer.byteLength(instructions, "utf8"), + instructions, + sha256(resource), + Buffer.byteLength(resource, "utf8"), + resource, + sha256(license), + Buffer.byteLength(license, "utf8"), + license, + ], + ); + await pool.query( + `INSERT INTO external_skill_identities ( + id,source_id,catalog_skill_id,normalized_skill_root + ) VALUES ($1,$2,'legacy-byte-total-skill','skills/legacy-byte-total')`, + [identityId, sourceId], + ); + await pool.query( + `INSERT INTO external_skill_revisions ( + id,skill_identity_id,snapshot_id,revision,bundle_sha256, + content_identity_sha256,name,description,skill_path,commit_sha, + source_owner,spdx_license_id,license_sha256,instructions_sha256, + invocation_mode,canonical_bytes,origin_owner,origin_repository + ) VALUES ($1,$2,$3,$4,$5,$6,'legacy-byte-total-skill', + 'Sanitized legacy byte-total fixture.', + 'skills/legacy-byte-total/SKILL.md',$7,'Fixture Owner','MIT',$8, + $9,'automatic',$10,'fixture-owner','fixture-repository')`, + [ + revisionId, + identityId, + snapshotId, + `gh-${sha256(canonicalBytes)}`, + options.corruptBundleHash ? "2".repeat(64) : sha256(canonicalBytes), + sha256(`identity:${canonicalBytes}`), + "a".repeat(40), + sha256(license), + instructionsSha256, + canonicalBytes, + ], + ); + await pool.query( + `INSERT INTO external_revision_resources ( + revision_id,resource_path,media_type,byte_length,content_sha256,ordinal + ) VALUES ($1,'references/example.md','text/markdown',$2,$3,0)`, + [revisionId, Buffer.byteLength(resource, "utf8"), sha256(resource)], + ); + await pool.query( + `INSERT INTO external_import_candidates ( + id,snapshot_id,adapter_kind,normalized_skill_root,normalized_name, + display_name,description,skill_document_path,source_path_sha256 + ) VALUES ($1,$2,'nested-skill','skills/legacy-byte-total', + 'legacy-byte-total-skill','legacy-byte-total-skill', + 'Sanitized legacy byte-total fixture.', + 'skills/legacy-byte-total/SKILL.md',$3)`, + [candidateId, snapshotId, "d".repeat(64)], + ); + await pool.query( + `INSERT INTO external_verification_reports ( + id,candidate_id,policy_version,validator_version,input_sha256, + report_sha256,result + ) VALUES ($1,$2,'external-policy-v1','external-validator-v1',$3,$4, + 'passed')`, + [reportId, candidateId, "e".repeat(64), "f".repeat(64)], + ); + await pool.query( + `INSERT INTO external_classification_events ( + id,candidate_id,previous_classification,next_classification, + actor_kind,actor_id,reason_code,report_id + ) VALUES ($1,$2,NULL,'discovered','synchronization','legacy-fixture', + 'CANDIDATE_DISCOVERED',NULL)`, + [discoveredEventId, candidateId], + ); + await pool.query( + `INSERT INTO external_current_classifications ( + candidate_id,classification,latest_event_id + ) VALUES ($1,'discovered',$2)`, + [candidateId, discoveredEventId], + ); + await pool.query( + `INSERT INTO external_classification_events ( + id,candidate_id,previous_classification,next_classification, + actor_kind,actor_id,reason_code,report_id + ) VALUES ($1,$2,'discovered','verified','verifier','legacy-fixture', + 'AUTOMATIC_VERIFICATION_PASSED',$3)`, + [verifiedEventId, candidateId, reportId], + ); + await pool.query( + `UPDATE external_current_classifications + SET classification='verified',latest_event_id=$2 + WHERE candidate_id=$1`, + [candidateId, verifiedEventId], + ); + await pool.query( + `INSERT INTO external_snapshot_skill_observations ( + snapshot_id,skill_identity_id,revision_id,result,candidate_id, + observed_content_identity_sha256 + ) VALUES ($1,$2,$3,'published',$4,$5)`, + [ + snapshotId, + identityId, + revisionId, + candidateId, + sha256(`identity:${canonicalBytes}`), + ], + ); + const schema = await pool.query<{ latest: string }>( + "SELECT max(version) AS latest FROM schema_migrations", + ); + let revisionEventId = verifiedEventId; + if ((schema.rows[0]?.latest ?? "000") >= "010") { + revisionEventId = randomUUID(); + await pool.query( + `INSERT INTO external_revision_classification_events ( + id,revision_id,initiating_candidate_id,previous_classification, + next_classification,actor_kind,actor_id,reason_code,report_id + ) VALUES ($1,$2,$3,NULL,'verified','verifier','legacy-fixture', + 'AUTOMATIC_VERIFICATION_PASSED',$4)`, + [revisionEventId, revisionId, candidateId, reportId], + ); + } + await pool.query( + `INSERT INTO external_current_revision_classifications ( + revision_id,classification,latest_event_id + ) VALUES ($1,'verified',$2)`, + [revisionId, revisionEventId], + ); + return { + canonicalBytes, + canonicalByteTotal, + contentIdentitySha256: sha256(`identity:${canonicalBytes}`), + identityId, + legacyByteTotal, + revisionId, + snapshotId, + sourceId, + }; +} + +async function insertSharedAndQuarantinedSnapshots( + pool: Pool, + fixture: Awaited>, +): Promise<{ + readonly quarantinedSnapshotId: string; + readonly sharedSnapshotId: string; +}> { + const sharedSnapshotId = randomUUID(); + const sharedCandidateId = randomUUID(); + const sharedReportId = randomUUID(); + const sharedDiscoveredId = randomUUID(); + const sharedVerifiedId = randomUUID(); + await pool.query( + `INSERT INTO external_source_snapshots ( + id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + adapter_kind,candidate_count,quarantine_count,resource_count, + dependency_count,decoded_bytes,validation_input_sha256, + advisory_chain_head_sha256,origin_github_repository_id,origin_owner, + origin_repository + ) VALUES ($1,$2,$3,$4,'nested-v1',1,'nested-skill',1,0,1,0,$5,$6,$7, + 9001,'fixture-owner','fixture-repository')`, + [ + sharedSnapshotId, + fixture.sourceId, + "1".repeat(40), + "2".repeat(40), + fixture.legacyByteTotal, + "3".repeat(64), + "0".repeat(64), + ], + ); + await pool.query( + `INSERT INTO external_import_candidates ( + id,snapshot_id,adapter_kind,normalized_skill_root,normalized_name, + display_name,description,skill_document_path,source_path_sha256 + ) VALUES ($1,$2,'nested-skill','skills/legacy-byte-total', + 'legacy-byte-total-skill','legacy-byte-total-skill', + 'Sanitized shared-revision fixture.', + 'skills/legacy-byte-total/SKILL.md',$3)`, + [sharedCandidateId, sharedSnapshotId, "4".repeat(64)], + ); + await pool.query( + `INSERT INTO external_verification_reports ( + id,candidate_id,policy_version,validator_version,input_sha256, + report_sha256,result + ) VALUES ($1,$2,'external-policy-v1','external-validator-v1',$3,$4, + 'passed')`, + [sharedReportId, sharedCandidateId, "5".repeat(64), "6".repeat(64)], + ); + await pool.query( + `INSERT INTO external_classification_events ( + id,candidate_id,previous_classification,next_classification, + actor_kind,actor_id,reason_code,report_id + ) VALUES + ($1,$3,NULL,'discovered','synchronization','legacy-fixture', + 'CANDIDATE_DISCOVERED',NULL), + ($2,$3,'discovered','verified','verifier','legacy-fixture', + 'AUTOMATIC_VERIFICATION_PASSED',$4)`, + [sharedDiscoveredId, sharedVerifiedId, sharedCandidateId, sharedReportId], + ); + await pool.query( + `INSERT INTO external_current_classifications ( + candidate_id,classification,latest_event_id + ) VALUES ($1,'verified',$2)`, + [sharedCandidateId, sharedVerifiedId], + ); + await pool.query( + `INSERT INTO external_snapshot_skill_observations ( + snapshot_id,skill_identity_id,revision_id,result,candidate_id, + observed_content_identity_sha256 + ) VALUES ($1,$2,$3,'reused',$4,$5)`, + [ + sharedSnapshotId, + fixture.identityId, + fixture.revisionId, + sharedCandidateId, + fixture.contentIdentitySha256, + ], + ); + + const quarantinedSnapshotId = randomUUID(); + const quarantinedCandidateId = randomUUID(); + const quarantinedReportId = randomUUID(); + const quarantinedDiscoveredId = randomUUID(); + const quarantinedEventId = randomUUID(); + await pool.query( + `INSERT INTO external_source_snapshots ( + id,source_id,commit_sha,tree_sha,manifest_version,revision_count, + adapter_kind,candidate_count,quarantine_count,resource_count, + dependency_count,decoded_bytes,validation_input_sha256, + advisory_chain_head_sha256,origin_github_repository_id,origin_owner, + origin_repository + ) VALUES ($1,$2,$3,$4,'nested-v1',0,'nested-skill',1,1,0,0,0,$5,$6, + 9001,'fixture-owner','fixture-repository')`, + [ + quarantinedSnapshotId, + fixture.sourceId, + "7".repeat(40), + "8".repeat(40), + "9".repeat(64), + "0".repeat(64), + ], + ); + await pool.query( + `INSERT INTO external_import_candidates ( + id,snapshot_id,adapter_kind,normalized_skill_root,normalized_name, + display_name,description,skill_document_path,source_path_sha256 + ) VALUES ($1,$2,'nested-skill','skills/quarantined', + 'quarantined-skill','quarantined-skill', + 'Sanitized quarantined fixture.', + 'skills/quarantined/SKILL.md',$3)`, + [quarantinedCandidateId, quarantinedSnapshotId, "a".repeat(64)], + ); + await pool.query( + `INSERT INTO external_verification_reports ( + id,candidate_id,policy_version,validator_version,input_sha256, + report_sha256,result + ) VALUES ($1,$2,'external-policy-v1','external-validator-v1',$3,$4, + 'failed')`, + [ + quarantinedReportId, + quarantinedCandidateId, + "b".repeat(64), + "c".repeat(64), + ], + ); + await pool.query( + `INSERT INTO external_classification_events ( + id,candidate_id,previous_classification,next_classification, + actor_kind,actor_id,reason_code,report_id + ) VALUES + ($1,$3,NULL,'discovered','synchronization','legacy-fixture', + 'CANDIDATE_DISCOVERED',NULL), + ($2,$3,'discovered','quarantined','verifier','legacy-fixture', + 'AUTOMATIC_VERIFICATION_FAILED',$4)`, + [ + quarantinedDiscoveredId, + quarantinedEventId, + quarantinedCandidateId, + quarantinedReportId, + ], + ); + await pool.query( + `INSERT INTO external_current_classifications ( + candidate_id,classification,latest_event_id + ) VALUES ($1,'quarantined',$2)`, + [quarantinedCandidateId, quarantinedEventId], + ); + await pool.query( + `INSERT INTO external_snapshot_skill_observations ( + snapshot_id,skill_identity_id,revision_id,result + ) VALUES ($1,$2,NULL,'missing')`, + [quarantinedSnapshotId, fixture.identityId], + ); + return { quarantinedSnapshotId, sharedSnapshotId }; +} + +async function readFixtureEvidence( + pool: Pool, + installationAccountId = randomUUID(), +) { + return readDatabaseEvidence({ + dockerExecutable: "/usr/bin/docker", + dockerArgs: [], + databaseName: "skillwire", + databaseUser: "skillwire", + environment: {}, + signal: new AbortController().signal, + installationAccountId, + run: async (options: CommandOptions) => { + const query = options.args.at(-1); + if (query === undefined) throw new Error("Missing evidence query"); + const result = await pool.query<{ evidence: unknown }>( + `SELECT (${query})::json AS evidence`, + ); + return { + code: 0, + stdout: `${JSON.stringify(result.rows[0]?.evidence)}\n`, + stderr: "", + durationMilliseconds: 0, + }; + }, + }); +} + +describe("legacy imported snapshot byte totals", () => { + it.each([ + ["without reconciliation evidence", false], + ["with fabricated reconciliation evidence", true], + ] as const)( + "rejects a pre-011 writer %s", + async (_scenario, fabricateEvidence) => { + const database = await createTestDatabase(); + const client = await database.pool.connect(); + try { + await runMigrations(database.pool); + await client.query("BEGIN"); + const fixture = await insertLegacyVerifiedSnapshot(client); + if (fabricateEvidence) { + await client.query( + `INSERT INTO external_snapshot_byte_total_reconciliations ( + snapshot_id,prior_decoded_bytes,legacy_payload_decoded_bytes, + reconciled_decoded_bytes,prior_representation + ) VALUES ($1,$2,$2,$3,'legacy-payload')`, + [ + fixture.snapshotId, + fixture.legacyByteTotal, + fixture.canonicalByteTotal, + ], + ); + } + + await expect(client.query("COMMIT")).rejects.toThrow( + /snapshot byte-total projection is invalid/i, + ); + await client.query("ROLLBACK").catch(() => undefined); + } finally { + client.release(); + await database.close(); + } + }, + 120_000, + ); + + it.each([ + [ + "malformed content hash", + { corruptContentHash: true }, + /content objects/i, + ], + [ + "canonical revision hash drift", + { corruptBundleHash: true }, + /canonical revisions/i, + ], + ["arbitrary byte total", { storedByteTotal: 42 }, /unsupported totals/i], + ] as const)( + "fails closed and rolls migration 011 back for %s", + async (_name, options, message) => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot( + database.pool, + options, + ); + if ("storedByteTotal" in options) { + const evidence = await readFixtureEvidence(database.pool); + expect(evidence.catalog).toMatchObject({ + legacySnapshotByteTotals: 0, + invalidSnapshotByteTotals: 1, + }); + } + + await expect(runMigrations(database.pool)).rejects.toThrow(message); + const state = await database.pool.query<{ + audit_table: string | null; + immutable_trigger_enabled: string; + migration_010: boolean; + migration_011: boolean; + stored_decoded_bytes: string; + }>( + `SELECT + to_regclass('public.external_snapshot_byte_total_reconciliations')::text + AS audit_table, + (SELECT tgenabled FROM pg_trigger + WHERE tgname='external_snapshots_immutable') + AS immutable_trigger_enabled, + EXISTS (SELECT 1 FROM schema_migrations WHERE version='010') + AS migration_010, + EXISTS (SELECT 1 FROM schema_migrations WHERE version='011') + AS migration_011, + (SELECT decoded_bytes::text FROM external_source_snapshots + WHERE id=$1) AS stored_decoded_bytes`, + [fixture.snapshotId], + ); + expect(state.rows).toEqual([ + { + audit_table: null, + immutable_trigger_enabled: "O", + migration_010: true, + migration_011: false, + stored_decoded_bytes: String( + "storedByteTotal" in options + ? options.storedByteTotal + : fixture.legacyByteTotal, + ), + }, + ]); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, + 120_000, + ); + + it("rejects malformed null-revision observations before reconciliation", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot(database.pool); + const { quarantinedSnapshotId } = + await insertSharedAndQuarantinedSnapshots(database.pool, fixture); + await database.pool.query( + `ALTER TABLE external_snapshot_skill_observations + DISABLE TRIGGER external_observations_immutable`, + ); + await database.pool.query( + `UPDATE external_snapshot_skill_observations + SET observed_content_identity_sha256=$2 + WHERE snapshot_id=$1 AND revision_id IS NULL`, + [quarantinedSnapshotId, "f".repeat(64)], + ); + await database.pool.query( + `ALTER TABLE external_snapshot_skill_observations + ENABLE TRIGGER external_observations_immutable`, + ); + + const evidence = await readFixtureEvidence(database.pool); + expect(evidence.catalog.invalidSnapshotObjectGraph).toBe(1); + await expect(runMigrations(database.pool)).rejects.toThrow( + /malformed observation attribution/i, + ); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); + + it("rejects a missing authoritative content object before reconciliation", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot(database.pool); + await database.pool.query( + `ALTER TABLE external_skill_revisions + DISABLE TRIGGER external_revisions_immutable`, + ); + await database.pool.query( + `ALTER TABLE external_skill_revisions + DROP CONSTRAINT external_skill_revisions_instructions_sha256_fkey`, + ); + await database.pool.query( + `UPDATE external_skill_revisions + SET instructions_sha256=$2 + WHERE id=$1`, + [fixture.revisionId, "0".repeat(64)], + ); + await database.pool.query( + `ALTER TABLE external_skill_revisions + ENABLE TRIGGER external_revisions_immutable`, + ); + + await expect(runMigrations(database.pool)).rejects.toThrow( + /malformed catalog objects/i, + ); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); + + it("rejects duplicate per-snapshot revision accounting", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot(database.pool); + await database.pool.query( + `ALTER TABLE external_snapshot_skill_observations + DISABLE TRIGGER external_observations_immutable`, + ); + await database.pool.query( + `ALTER TABLE external_snapshot_skill_observations + DROP CONSTRAINT external_snapshot_skill_observations_pkey`, + ); + await database.pool.query( + `INSERT INTO external_snapshot_skill_observations ( + snapshot_id,skill_identity_id,revision_id,result,candidate_id, + observed_content_identity_sha256 + ) SELECT snapshot_id,skill_identity_id,revision_id,result,candidate_id, + observed_content_identity_sha256 + FROM external_snapshot_skill_observations + WHERE snapshot_id=$1`, + [fixture.snapshotId], + ); + await database.pool.query( + `ALTER TABLE external_snapshot_skill_observations + ENABLE TRIGGER external_observations_immutable`, + ); + + await expect(runMigrations(database.pool)).rejects.toThrow( + /duplicate revision accounting/i, + ); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); + + it("reconciles verified, quarantined, missing, and shared-revision snapshots idempotently", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot(database.pool); + const snapshots = await insertSharedAndQuarantinedSnapshots( + database.pool, + fixture, + ); + + await runMigrations(database.pool); + await runMigrations(database.pool); + await expect( + database.pool.query( + "TRUNCATE external_snapshot_byte_total_reconciliations", + ), + ).rejects.toThrow(/immutable/i); + + const result = await database.pool.query<{ + audit_count: string; + canonical_snapshots: string; + legacy_reconciliations: string; + migration_count: string; + }>( + `SELECT + (SELECT count(*)::text + FROM external_snapshot_byte_total_reconciliations) AS audit_count, + (SELECT count(*)::text + FROM external_source_snapshots snapshot + WHERE snapshot.decoded_bytes=( + SELECT COALESCE(sum(octet_length(revision.canonical_bytes)),0) + FROM external_snapshot_skill_observations observation + JOIN external_skill_revisions revision + ON revision.id=observation.revision_id + WHERE observation.snapshot_id=snapshot.id + )) AS canonical_snapshots, + (SELECT count(*)::text + FROM external_snapshot_byte_total_reconciliations + WHERE prior_representation='legacy-payload') + AS legacy_reconciliations, + (SELECT count(*)::text FROM schema_migrations WHERE version='011') + AS migration_count`, + ); + expect(result.rows).toEqual([ + { + audit_count: "3", + canonical_snapshots: "3", + legacy_reconciliations: "2", + migration_count: "1", + }, + ]); + expect(snapshots.sharedSnapshotId).not.toBe( + snapshots.quarantinedSnapshotId, + ); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); + + it("distinguishes exact legacy totals from corruption before and after reconciliation", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + await insertLegacyVerifiedSnapshot(database.pool); + const accountId = randomUUID(); + await database.pool.query( + `INSERT INTO accounts (id,status) VALUES ($1,'active')`, + [accountId], + ); + await database.pool.query( + `INSERT INTO api_keys ( + id,account_id,public_id,secret_digest + ) VALUES ($1,$2,$3,$4)`, + [randomUUID(), accountId, "fixturepublicid1", Buffer.alloc(32, 1)], + ); + + const before = await readFixtureEvidence(database.pool, accountId); + expect(before.catalog).toMatchObject({ + legacySnapshotByteTotals: 1, + invalidSnapshotByteTotals: 0, + invalidSnapshotByteOverflows: 0, + invalidSnapshotObjectGraph: 0, + invalidRevisionHashes: 0, + invalidSnapshotReconciliations: 0, + }); + const expected009 = await expectedMigrationInventory( + legacyMigrations, + "009", + ); + expect(() => + assessRestoredDatabaseEvidence( + { ...before, currentDatabase: "skillwire" }, + { + expectedMigrations: expected009, + installationAccountId: accountId, + expectedActiveApiKeys: 1, + expectedDatabase: "skillwire", + expectedState: databaseStateExpectation(before), + }, + ), + ).not.toThrow(); + + await runMigrations(database.pool); + const after = await readFixtureEvidence(database.pool, accountId); + expect(after.catalog).toMatchObject({ + legacySnapshotByteTotals: 0, + invalidSnapshotByteTotals: 0, + invalidSnapshotByteOverflows: 0, + invalidSnapshotObjectGraph: 0, + invalidRevisionHashes: 0, + invalidSnapshotReconciliations: 0, + }); + const expected011 = await expectedMigrationInventory( + join(process.cwd(), "migrations"), + "011", + ); + expect(() => + assessRestoredDatabaseEvidence( + { ...after, currentDatabase: "skillwire" }, + { + expectedMigrations: expected011, + installationAccountId: accountId, + expectedActiveApiKeys: 1, + expectedDatabase: "skillwire", + expectedState: forwardMigrationStateExpectation(before, after), + }, + ), + ).not.toThrow(); + const sourceConstraint = before.constraints[0]; + expect(sourceConstraint).toBeDefined(); + expect(() => + forwardMigrationStateExpectation(before, { + ...after, + constraints: after.constraints.map((constraint) => + constraint.constraintName === sourceConstraint?.constraintName + ? { ...constraint, validated: false } + : constraint, + ), + }), + ).toThrow(/changed pre-existing constraint/i); + await database.pool.query( + `ALTER TABLE external_snapshot_byte_total_reconciliations + DISABLE TRIGGER external_snapshot_byte_total_reconciliations_immutable`, + ); + await database.pool.query( + "DELETE FROM external_snapshot_byte_total_reconciliations", + ); + await database.pool.query( + `ALTER TABLE external_snapshot_byte_total_reconciliations + ENABLE TRIGGER external_snapshot_byte_total_reconciliations_immutable`, + ); + const corrupted = await readFixtureEvidence(database.pool, accountId); + expect(corrupted.catalog.invalidSnapshotReconciliations).toBe(1); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); + + it("reconciles an exact pre-v0.2.0 payload total to canonical bytes", async () => { + const database = await createTestDatabase(); + const legacyMigrations = await copyMigrationsThrough009(); + try { + await runMigrations(database.pool, legacyMigrations); + const fixture = await insertLegacyVerifiedSnapshot(database.pool); + + await runMigrations(database.pool); + + const result = await database.pool.query<{ + latest_migration: string; + prior_decoded_bytes: string; + reconciled_decoded_bytes: string; + stored_decoded_bytes: string; + }>( + `SELECT + (SELECT max(version) FROM schema_migrations) AS latest_migration, + reconciliation.prior_decoded_bytes::text, + reconciliation.reconciled_decoded_bytes::text, + snapshot.decoded_bytes::text AS stored_decoded_bytes + FROM external_source_snapshots snapshot + JOIN external_snapshot_byte_total_reconciliations reconciliation + ON reconciliation.snapshot_id=snapshot.id + WHERE snapshot.id=$1`, + [fixture.snapshotId], + ); + expect(result.rows).toEqual([ + { + latest_migration: "011", + prior_decoded_bytes: String(fixture.legacyByteTotal), + reconciled_decoded_bytes: String(fixture.canonicalByteTotal), + stored_decoded_bytes: String(fixture.canonicalByteTotal), + }, + ]); + } finally { + await database.close(); + await rm(legacyMigrations, { recursive: true, force: true }); + } + }, 120_000); +}); diff --git a/tests/integration/postgres/migrations.test.ts b/tests/integration/postgres/migrations.test.ts index 5bd1cbc..8961b05 100644 --- a/tests/integration/postgres/migrations.test.ts +++ b/tests/integration/postgres/migrations.test.ts @@ -54,6 +54,7 @@ describe("versioned PostgreSQL migrations", () => { "008", "009", "010", + "011", ]); }); @@ -97,7 +98,7 @@ describe("versioned PostgreSQL migrations", () => { ); }); - it("times out behind a forgotten writer and recovers from the untouched pre-010 state", async () => { + it("times out behind a forgotten writer and recovers through the latest migration", async () => { const legacy = await createTestDatabase(); const target = await copyMigrations(/^00[1-9]_.*\.sql$/); const blocker = await legacy.pool.connect(); @@ -141,7 +142,7 @@ describe("versioned PostgreSQL migrations", () => { "SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1", ) ).rows, - ).toEqual([{ version: "010" }]); + ).toEqual([{ version: "011" }]); } finally { await blocker.query("ROLLBACK").catch(() => undefined); blocker.release(); diff --git a/tests/unit/onboarding/restored-database-validation.test.ts b/tests/unit/onboarding/restored-database-validation.test.ts index 4d3d3a0..7f9cc00 100644 --- a/tests/unit/onboarding/restored-database-validation.test.ts +++ b/tests/unit/onboarding/restored-database-validation.test.ts @@ -51,6 +51,7 @@ function trigger( functionName: string, events: RestoredDatabaseEvidence["triggers"][number]["events"], timing: RestoredDatabaseEvidence["triggers"][number]["timing"] = "BEFORE", + level: RestoredDatabaseEvidence["triggers"][number]["level"] = "ROW", ): RestoredDatabaseEvidence["triggers"][number] { const functionBodySha256 = { reject_external_history_mutation: @@ -67,6 +68,8 @@ function trigger( "e26c0466292fb76c9f4cce6af78ea7f617617d7a5f117fd20e15f3bfbbfdde8c", validate_external_revision_classification_transition: "01461630956a69c1ead4bfd7bf1df621e217a352052123fcfe79e401dea840b8", + validate_external_snapshot_byte_total_projection: + "6c866c631670b92a1ea5082ad0440ce207c0a53274073871acdcf45de5d18877", }[functionName]; if (functionBodySha256 === undefined) throw new Error("Test trigger function is not release-bound"); @@ -80,7 +83,13 @@ function trigger( functionDefinition: `CREATE FUNCTION ${functionName}() RETURNS trigger LANGUAGE plpgsql AS 'fixture'`, functionBodySha256, timing, - level: "ROW", + level, + deferrable: + triggerName === "external_snapshot_finalization_required" || + triggerName === "external_snapshot_byte_total_projection_valid", + initiallyDeferred: + triggerName === "external_snapshot_finalization_required" || + triggerName === "external_snapshot_byte_total_projection_valid", events, enabled: "origin", definition: `CREATE TRIGGER ${triggerName} ${timing} ${events.join(" OR ")} ON ${tableName} FOR EACH ROW EXECUTE FUNCTION ${functionName}()`, @@ -191,6 +200,27 @@ const REQUIRED_TRIGGERS: RestoredDatabaseEvidence["triggers"] = [ "reject_external_history_mutation", ["DELETE", "UPDATE"], ), + trigger( + "external_snapshot_byte_total_projection_valid", + "external_source_snapshots", + "validate_external_snapshot_byte_total_projection", + ["INSERT", "UPDATE"], + "AFTER", + ), + trigger( + "external_snapshot_byte_total_reconciliations_immutable", + "external_snapshot_byte_total_reconciliations", + "reject_external_history_mutation", + ["DELETE", "UPDATE"], + ), + trigger( + "external_snapshot_byte_total_reconciliations_truncate_rejected", + "external_snapshot_byte_total_reconciliations", + "reject_external_history_mutation", + ["TRUNCATE"], + "BEFORE", + "STATEMENT", + ), ]; function evidence(options: { @@ -225,6 +255,12 @@ function evidence(options: { dependencyCount: 1, contentObjectCount: 5, identitySha256: "c".repeat(64), + legacySnapshotByteTotals: 0, + invalidSnapshotByteTotals: 0, + invalidSnapshotByteOverflows: 0, + invalidSnapshotObjectGraph: 0, + invalidRevisionHashes: 0, + invalidSnapshotReconciliations: 0, invalidSnapshotCounts: 0, invalidPublishedPointers: 0, invalidContentLengths: 0, @@ -246,6 +282,50 @@ function evidence(options: { } describe("production restored-database validation", () => { + it("accepts only the exact legacy byte-total representation before migration 011", () => { + const accountId = randomUUID(); + const checksums = Array.from({ length: 10 }, (_, index) => + (index + 1).toString(16).padStart(64, "0"), + ); + const legacy = evidence({ + accountId, + checksums, + triggers: REQUIRED_TRIGGERS.slice(0, -3), + }); + legacy.catalog.legacySnapshotByteTotals = 1; + + expect( + assessRestoredDatabaseEvidence(legacy, { + expectedMigrations: legacy.migrations, + installationAccountId: accountId, + expectedActiveApiKeys: 2, + expectedDatabase: "postgres", + expectedState: databaseStateExpectation(legacy), + }), + ).toMatchObject({ + latestMigration: "010", + catalogValid: true, + }); + + const arbitraryMismatch: RestoredDatabaseEvidence = { + ...legacy, + catalog: { + ...legacy.catalog, + legacySnapshotByteTotals: 0, + invalidSnapshotByteTotals: 1, + }, + }; + expect(() => + assessRestoredDatabaseEvidence(arbitraryMismatch, { + expectedMigrations: arbitraryMismatch.migrations, + installationAccountId: accountId, + expectedActiveApiKeys: 2, + expectedDatabase: "postgres", + expectedState: databaseStateExpectation(legacy), + }), + ).toThrow(/restore validation/i); + }); + it("requires the complete immutable migration inventory and checksums", async () => { const fixture = await createOnboardingEnvironment(); try { @@ -359,6 +439,27 @@ describe("production restored-database validation", () => { catalog: { ...value.catalog, invalidContentLengths: 1 }, }), ], + [ + "snapshot byte overflow", + (value: RestoredDatabaseEvidence) => ({ + ...value, + catalog: { ...value.catalog, invalidSnapshotByteOverflows: 1 }, + }), + ], + [ + "duplicate snapshot accounting", + (value: RestoredDatabaseEvidence) => ({ + ...value, + catalog: { ...value.catalog, invalidSnapshotObjectGraph: 1 }, + }), + ], + [ + "canonical revision hash", + (value: RestoredDatabaseEvidence) => ({ + ...value, + catalog: { ...value.catalog, invalidRevisionHashes: 1 }, + }), + ], [ "advisory integrity", (value: RestoredDatabaseEvidence) => ({ @@ -436,7 +537,7 @@ describe("production restored-database validation", () => { const valid = evidence({ accountId, checksums, - triggers: REQUIRED_TRIGGERS, + triggers: REQUIRED_TRIGGERS.slice(0, -3), }); expect( @@ -454,6 +555,64 @@ describe("production restored-database validation", () => { }); }); + it("requires the append-only byte-total reconciliation ledger after migration 011", () => { + const accountId = randomUUID(); + const checksums = Array.from({ length: 11 }, (_, index) => + (index + 1).toString(16).padStart(64, "0"), + ); + const valid = evidence({ + accountId, + checksums, + triggers: REQUIRED_TRIGGERS, + }); + + expect( + assessRestoredDatabaseEvidence(valid, { + expectedMigrations: valid.migrations, + installationAccountId: accountId, + expectedActiveApiKeys: 2, + expectedDatabase: "postgres", + expectedState: databaseStateExpectation(valid), + }), + ).toMatchObject({ latestMigration: "011", catalogValid: true }); + + const missingLedgerTrigger: RestoredDatabaseEvidence = { + ...valid, + triggers: valid.triggers.filter( + ({ triggerName }) => + triggerName !== + "external_snapshot_byte_total_reconciliations_truncate_rejected", + ), + }; + expect(() => + assessRestoredDatabaseEvidence(missingLedgerTrigger, { + expectedMigrations: missingLedgerTrigger.migrations, + installationAccountId: accountId, + expectedActiveApiKeys: 2, + expectedDatabase: "postgres", + expectedState: databaseStateExpectation(missingLedgerTrigger), + }), + ).toThrow(/restore validation/i); + + const nonDeferredProjection: RestoredDatabaseEvidence = { + ...valid, + triggers: valid.triggers.map((entry) => + entry.triggerName === "external_snapshot_byte_total_projection_valid" + ? { ...entry, deferrable: false, initiallyDeferred: false } + : entry, + ), + }; + expect(() => + assessRestoredDatabaseEvidence(nonDeferredProjection, { + expectedMigrations: nonDeferredProjection.migrations, + installationAccountId: accountId, + expectedActiveApiKeys: 2, + expectedDatabase: "postgres", + expectedState: databaseStateExpectation(nonDeferredProjection), + }), + ).toThrow(/restore validation/i); + }); + it("rejects the superseded classification-trigger function body after migration 010", () => { const accountId = randomUUID(); const checksums = Array.from({ length: 10 }, (_, index) => @@ -462,7 +621,7 @@ describe("production restored-database validation", () => { const valid = evidence({ accountId, checksums, - triggers: REQUIRED_TRIGGERS, + triggers: REQUIRED_TRIGGERS.slice(0, -3), }); const superseded: RestoredDatabaseEvidence = { ...valid, @@ -555,7 +714,7 @@ describe("production restored-database validation", () => { const valid = evidence({ accountId, checksums, - triggers: REQUIRED_TRIGGERS, + triggers: REQUIRED_TRIGGERS.slice(0, -3), }); const corrupted = corrupt(valid); @@ -578,7 +737,7 @@ describe("production restored-database validation", () => { const valid = evidence({ accountId, checksums, - triggers: REQUIRED_TRIGGERS, + triggers: REQUIRED_TRIGGERS.slice(0, -3), }); const drifted: RestoredDatabaseEvidence = { ...valid, @@ -704,6 +863,8 @@ describe("production restored-database validation", () => { expect(query).toContain("pg_get_functiondef"); expect(query).toContain("functionBodySha256"); expect(query).toContain("tgenabled"); + expect(query).toContain("tgdeferrable"); + expect(query).toContain("tginitdeferred"); expect(query).toContain("proname"); expect(query).not.toMatch(/AS\s+(?:invariants|catalog|advisory)_valid/i); });