fix(registry): exempt proven first releases from holdback (#3078)

* fix(registry): exempt proven first releases from holdback

* fix(admin): preserve registry policy evidence

* test(admin): cover release policy fallback

Author: ascorbic

.changeset/calm-registry-first-releases.md

@@ -0,0 +1,10 @@
+---
+"@emdash-cms/registry-lexicons": minor
+"@emdash-cms/registry-client": minor
+"@emdash-cms/admin": patch
+"emdash": patch
+---
+
+Adds a fail-closed first-release exemption to the plugin registry's optional minimum release age policy. A package's first release can install immediately only when the aggregator reports exactly one retained release and confirms that it continuously observed the package's release history.
+
+Existing packages, backfilled packages, and packages with missing or incomplete history remain subject to the configured holdback. Deleted releases still count, and explicit publisher or package exemptions continue to work.

apps/aggregator/migrations/0006_release_history.sql

@@ -0,0 +1,29 @@
+-- Retain package-level release history evidence separately from the current
+-- profile/release projections. Existing rows came from an initial backfill or
+-- an earlier deployment whose cursor continuity cannot be proven, so they
+-- start incomplete and remain subject to the configured release-age holdback.
+CREATE TABLE IF NOT EXISTS package_release_history (
+	did TEXT NOT NULL,
+	package TEXT NOT NULL,
+	release_history_complete INTEGER NOT NULL CHECK (release_history_complete IN (0, 1)),
+	first_observed_at TEXT NOT NULL,
+	first_observed_source TEXT NOT NULL CHECK (
+		first_observed_source IN ('jetstream', 'backfill', 'unknown')
+	),
+	PRIMARY KEY (did, package)
+);
+
+INSERT OR IGNORE INTO package_release_history (
+	did,
+	package,
+	release_history_complete,
+	first_observed_at,
+	first_observed_source
+)
+SELECT
+	did,
+	slug,
+	0,
+	COALESCE(indexed_at, verified_at),
+	'unknown'
+FROM packages;

apps/aggregator/src/backfill.ts

@@ -395,6 +395,7 @@ async function paginateAndEnqueue(opts: PaginateOpts): Promise<number> {
 					rkey: parsed.rkey,
 					operation: "create",
 					cid: record.cid,
+					source: "backfill",
 				},
 			});
 		}

apps/aggregator/src/env.ts

@@ -13,6 +13,13 @@ export interface RecordsJob {
 	rkey: string;
 	operation: "create" | "update" | "delete";
 	cid: string;
+	/**
+	 * Identifies whether the aggregator observed this operation from its live,
+	 * cursor-backed stream or reconstructed current state through backfill.
+	 * Missing values come from an older producer during a rolling deployment
+	 * and must be treated as incomplete history.
+	 */
+	source?: "jetstream" | "backfill";
 	/**
 	 * The Jetstream-supplied (unverified) record bytes. Compared against the
 	 * verified PDS copy after fetch as a Jetstream-correctness signal; the

apps/aggregator/src/jetstream-ingestor.ts

@@ -239,6 +239,7 @@ export class JetstreamIngestor {
 			rkey: event.commit.rkey,
 			operation: event.commit.operation,
 			cid: event.commit.operation === "delete" ? "" : event.commit.cid,
+			source: "jetstream",
 			...(event.commit.operation !== "delete" ? { jetstreamRecord: event.commit.record } : {}),
 		};
 

apps/aggregator/src/records-consumer.ts

@@ -477,11 +477,25 @@ export async function ingestPackageProfile(
 			   updated_at = excluded.updated_at`,
 		)
 		.bind(job.did, slug, verified.cid, nowIso);
+	const retainReleaseHistory = db
+		.prepare(
+			`INSERT INTO package_release_history
+			   (did, package, release_history_complete, first_observed_at, first_observed_source)
+			 VALUES (?, ?, ?, ?, ?)
+			 ON CONFLICT(did, package) DO NOTHING`,
+		)
+		.bind(
+			job.did,
+			slug,
+			job.source === "jetstream" && job.operation === "create" ? 1 : 0,
+			nowIso,
+			job.source ?? "unknown",
+		);
 
 	// The revision must exist before the current pointer moves. D1 batches are
 	// transactional, so a failure leaves both the old pointer and old mutable
 	// compatibility row intact.
-	await db.batch([retainRevision, updateCurrentPackage, moveCurrentPointer]);
+	await db.batch([retainRevision, updateCurrentPackage, retainReleaseHistory, moveCurrentPointer]);
 }
 
 export async function ingestPackageRelease(
@@ -618,10 +632,21 @@ export async function ingestPackageRelease(
 	// roll back together and the message retries to a clean state. Without
 	// the batch, an insert-success / refresh-failure could leave
 	// `packages.latest_version` permanently stale.
-	const batchResults = await db.batch([
-		insertStmt,
-		refreshPackageLatestStmt(db, job.did, record.package),
-	]);
+	const batchStatements = [insertStmt, refreshPackageLatestStmt(db, job.did, record.package)];
+	if (job.source !== "jetstream") {
+		// A release first encountered outside the cursor-backed stream proves
+		// that the aggregator cannot claim continuous history for this package.
+		batchStatements.push(
+			db
+				.prepare(
+					`UPDATE package_release_history
+					 SET release_history_complete = 0
+					 WHERE did = ? AND package = ?`,
+				)
+				.bind(job.did, record.package),
+		);
+	}
+	const batchResults = await db.batch(batchStatements);
 	const insertResult = batchResults[0];
 	if (!insertResult) {
 		// Defensive: D1.batch() guarantees one result per statement; if it
@@ -1053,14 +1078,29 @@ async function writeDeadLetter(
 	// envelope of operation+cid so the row is still inspectable.
 	const payload = JSON.stringify(job.jetstreamRecord ?? { operation: job.operation, cid: job.cid });
 	const payloadBytes = new TextEncoder().encode(payload);
-	await db
+	const retainDeadLetter = db
 		.prepare(
 			`INSERT INTO dead_letters
 			   (did, collection, rkey, reason, detail, payload, received_at)
 			 VALUES (?, ?, ?, ?, ?, ?, ?)`,
 		)
-		.bind(job.did, job.collection, job.rkey, reason, detail, payloadBytes, now.toISOString())
-		.run();
+		.bind(job.did, job.collection, job.rkey, reason, detail, payloadBytes, now.toISOString());
+	const releaseIdentity =
+		job.collection === NSID.packageRelease ? parseReleaseRkey(job.rkey) : null;
+	if (!releaseIdentity) {
+		await retainDeadLetter.run();
+		return;
+	}
+	const markHistoryIncomplete = db
+		.prepare(
+			`INSERT INTO package_release_history
+			   (did, package, release_history_complete, first_observed_at, first_observed_source)
+			 VALUES (?, ?, 0, ?, ?)
+			 ON CONFLICT(did, package) DO UPDATE SET
+			   release_history_complete = 0`,
+		)
+		.bind(job.did, releaseIdentity.pkg, now.toISOString(), job.source ?? "unknown");
+	await db.batch([retainDeadLetter, markHistoryIncomplete]);
 }
 
 // ─── Production wiring ─────────────────────────────────────────────────────

apps/aggregator/src/routes/xrpc/listing-query.ts

@@ -13,6 +13,19 @@ import {
 } from "../../listing-policy.js";
 import { type PackageRow, packageColumns } from "./views.js";
 
+const RELEASE_HISTORY_COLUMNS_SQL = `
+	(SELECT COUNT(*)
+	   FROM releases release_history
+	  WHERE release_history.did = p.did
+	    AND release_history.package = p.slug) AS historical_release_count,
+	COALESCE(
+		(SELECT history.release_history_complete
+		   FROM package_release_history history
+		  WHERE history.did = p.did AND history.package = p.slug),
+		0
+	) AS release_history_complete
+`;
+
 export type PackageLookupResult =
 	| { state: "visible"; row: PackageRow }
 	| { state: "unavailable" }
@@ -28,7 +41,7 @@ export async function lookupPackage(
 	if (policy.mode === "projection") {
 		const row = await session
 			.prepare(
-				`SELECT ${packageColumns("p.")}, p.labels_json
+				`SELECT ${packageColumns("p.")}, p.labels_json, ${RELEASE_HISTORY_COLUMNS_SQL}
 				 FROM public_projection_state projection_state
 				 ${ACTIVE_PROJECTION_JOINS_SQL}
 				 JOIN public_packages p ON p.generation = projection_state.active_generation
@@ -52,7 +65,7 @@ export async function lookupPackage(
 
 	const row = await session
 		.prepare(
-			`SELECT ${packageColumns("p.")}
+			`SELECT ${packageColumns("p.")}, ${RELEASE_HISTORY_COLUMNS_SQL}
 			 FROM packages p
 			 WHERE p.did = ? AND p.slug = ?
 			   AND ${ACTIVE_PROFILE_SQL}

apps/aggregator/src/routes/xrpc/views.ts

@@ -46,6 +46,8 @@ export interface PackageRow {
 	verified_at: string;
 	indexed_at: string | null;
 	labels_json?: string;
+	historical_release_count?: number;
+	release_history_complete?: number;
 }
 
 /** Subset of columns from `releases` we read for `releaseView`. */
@@ -141,6 +143,12 @@ export function packageView(row: PackageRow): AggregatorDefs.PackageView {
 	if (row.latest_version !== null) {
 		view.latestVersion = row.latest_version;
 	}
+	if (row.historical_release_count !== undefined) {
+		view.historicalReleaseCount = row.historical_release_count;
+	}
+	if (row.release_history_complete !== undefined) {
+		view.releaseHistoryComplete = row.release_history_complete === 1;
+	}
 	return view;
 }
 

apps/aggregator/test/backfill.test.ts

@@ -208,6 +208,7 @@ describe("processBackfillJob", () => {
 			rkey: "demo",
 			operation: "create",
 			cid: "bafyc1",
+			source: "backfill",
 		});
 		// jetstreamRecord intentionally not set on backfill jobs — the
 		// consumer's DLQ payload field would otherwise mislabel

apps/aggregator/test/jetstream-ingestor.test.ts

@@ -126,6 +126,7 @@ describe("JetstreamIngestor", () => {
 			rkey: "p",
 			operation: "create",
 			cid: "bafyrecord",
+			source: "jetstream",
 			jetstreamRecord: { slug: "p", license: "MIT" },
 		});
 		expect(h.ingestor.currentCursor).toBe(event.time_us);
@@ -314,6 +315,7 @@ describe("JetstreamIngestor", () => {
 			rkey: "p",
 			operation: "delete",
 			cid: "",
+			source: "jetstream",
 		});
 		expect(h.queue.jobs[0]?.jetstreamRecord).toBeUndefined();