feat(cli): add bounded telemetry retention

.github/workflows/ci.yml

@@ -250,6 +250,7 @@ jobs:
                   apps/cli/test/native-archive-calibrate-probe.sh
                   apps/cli/test/native-archive-gc-probe.sh
                   apps/cli/test/native-archive-merge-probe.sh
+                  apps/cli/test/native-retention-probe.sh
 
             # Syntax-check the POSIX installers against the actual /bin/sh they claim.
             - name: sh -n
@@ -345,6 +346,11 @@ jobs:
                       tmp: maple-native-archive-gc
                       keep_root: "1"
                       duckdb: true
+                    - probe: retention
+                      script: native-retention-probe.sh
+                      tmp: maple-native-retention
+                      keep_root: "1"
+                      duckdb: true
         steps:
             - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6
 

apps/cli/src/commands/archive.ts

@@ -23,6 +23,7 @@ import {
 	type LoadedTuningConfig,
 } from "../server/archives/config"
 import { ARCHIVE_SIGNALS, isArchiveSignalName, type ArchiveSignalName } from "../server/archives/signals"
+import { expireArchiveDay, readRetiredDayLedger, retireLiveDay } from "../server/archives/retention"
 import { validateRangeDate } from "../server/archives/paths"
 import {
 	type CalibrationBudget,
@@ -119,6 +120,21 @@ const dryRunFlag = Flag.boolean("dry-run").pipe(
 	Flag.withDefault(false),
 )
 
+const applyFlag = Flag.boolean("apply").pipe(
+	Flag.withDescription("Apply the destructive operation (omitting this flag is a non-mutating refusal)"),
+	Flag.withDefault(false),
+)
+
+const localPortFlag = Flag.integer("port").pipe(
+	Flag.withDescription("Private Maple local-query port"),
+	Flag.withDefault(4318),
+)
+
+const sealingLagHoursFlag = Flag.integer("sealing-lag-hours").pipe(
+	Flag.withDescription("Hours after UTC midnight before a completed day may be retired"),
+	Flag.withDefault(24),
+)
+
 const keepFlag = Flag.integer("keep").pipe(
 	Flag.withDescription(
 		"Newest superseded generations to retain per signal/range (default 1; 0 reclaims all superseded)",
@@ -260,6 +276,11 @@ export const archiveCreate = Command.make("create", {
 				})
 			}
 			const { dataDir, archiveDir, scratchRoot } = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot)
+			if (readRetiredDayLedger(dataDir).retiredDays.some((day) => day.rangeDate === rangeDate)) {
+				return yield* new ArchiveError({
+					message: `refusing archive create: UTC day ${rangeDate} is permanently retired`,
+				})
+			}
 			const checkpointId = Option.getOrUndefined(a.checkpointId)
 			// Resolve tuning. Precedence: explicit CLI tuning flags > config-file
 			// effective values > defaults. A --config document is loaded from one fd
@@ -584,6 +605,70 @@ export const archiveGc = Command.make("gc", {
 	),
 )
 
+export const archiveExpire = Command.make("expire", {
+	dataDir: dataDirFlag,
+	archiveDir: archiveDirFlag,
+	scratchRoot: scratchRootFlag,
+	rangeDate: rangeDateArgument,
+	apply: applyFlag,
+}).pipe(
+	Command.withDescription("Expire one complete active archived UTC day across all six signals"),
+	Command.withHandler(
+		Effect.fnUntraced(function* (a) {
+			if (!a.apply)
+				return yield* new ArchiveError({ message: "refusing archive expiration without --apply" })
+			const roots = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot)
+			yield* Effect.tryPromise({
+				try: () =>
+					expireArchiveDay({
+						dataDir: roots.dataDir,
+						archiveDir: roots.archiveDir,
+						scratchRoot: roots.scratchRoot,
+						rangeDate: a.rangeDate,
+					}),
+				catch: (error) =>
+					new ArchiveError({ message: error instanceof Error ? error.message : String(error) }),
+			})
+			yield* Effect.sync(() =>
+				process.stdout.write(`${green("✓")} expired archive day ${a.rangeDate}\n`),
+			)
+		}),
+	),
+)
+
+export const archiveRetireLive = Command.make("retire-live", {
+	dataDir: dataDirFlag,
+	archiveDir: archiveDirFlag,
+	scratchRoot: scratchRootFlag,
+	rangeDate: rangeDateArgument,
+	port: localPortFlag,
+	sealingLagHours: sealingLagHoursFlag,
+	apply: applyFlag,
+}).pipe(
+	Command.withDescription("Remove one UTC day from live raw tables after complete archive verification"),
+	Command.withHandler(
+		Effect.fnUntraced(function* (a) {
+			if (!a.apply)
+				return yield* new ArchiveError({ message: "refusing live retirement without --apply" })
+			const roots = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot)
+			yield* Effect.tryPromise({
+				try: () =>
+					retireLiveDay({
+						dataDir: roots.dataDir,
+						archiveDir: roots.archiveDir,
+						scratchRoot: roots.scratchRoot,
+						rangeDate: a.rangeDate,
+						port: a.port,
+						sealingLagHours: a.sealingLagHours,
+					}),
+				catch: (error) =>
+					new ArchiveError({ message: error instanceof Error ? error.message : String(error) }),
+			})
+			yield* Effect.sync(() => process.stdout.write(`${green("✓")} retired live day ${a.rangeDate}\n`))
+		}),
+	),
+)
+
 const formatBytes = (bytes: number): string => {
 	if (bytes < 1024) return `${bytes} B`
 	if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KiB`
@@ -1764,6 +1849,8 @@ export const archive = Command.make("archive").pipe(
 		archiveRebuild,
 		archiveReconcile,
 		archiveGc,
+		archiveExpire,
+		archiveRetireLive,
 		archiveCalibrate,
 		archiveCalibrateRun,
 		archiveCalibrateSession,

apps/cli/src/commands/server-args.ts

@@ -23,6 +23,7 @@ export interface DetachedChildArgs {
 	readonly offline: boolean
 	readonly chdbConfigFile: string | undefined
 	readonly onDirtyStore: DirtyStorePolicy
+	readonly minimumRawTelemetryRetentionDays: number | undefined
 }
 
 /** Build the foreground child argv without forwarding compiled-Bun virtual
@@ -43,6 +44,9 @@ export const buildDetachedChildArgs = (options: DetachedChildArgs): string[] =>
 		"--on-dirty-store",
 		options.onDirtyStore,
 		...(options.chdbConfigFile ? ["--chdb-config-file", options.chdbConfigFile] : []),
+		...(options.minimumRawTelemetryRetentionDays !== undefined
+			? ["--minimum-raw-telemetry-retention-days", String(options.minimumRawTelemetryRetentionDays)]
+			: []),
 		...(options.offline ? ["--offline"] : []),
 	]
 }

apps/cli/src/commands/server.ts

@@ -169,6 +169,14 @@ const chdbConfigFileFlag = Flag.optional(
 	),
 )
 
+const minimumRawTelemetryRetentionDaysFlag = Flag.optional(
+	Flag.integer("minimum-raw-telemetry-retention-days").pipe(
+		Flag.withDescription(
+			"Persist a monotonic raw-table retention floor (minimum 90 days; survives reset and restore)",
+		),
+	),
+)
+
 const backgroundFlag = Flag.boolean("background").pipe(
 	Flag.withAlias("d"),
 	Flag.withDescription("Run the server detached (logs to ~/.maple/maple.log); stop with `maple stop`"),
@@ -233,6 +241,7 @@ const startDetached = (
 	offline: boolean,
 	chdbConfigFile: string | undefined,
 	onDirtyStore: DirtyStorePolicy,
+	minimumRawTelemetryRetentionDays: number | undefined,
 ): Effect.Effect<void, ServerError> =>
 	Effect.gen(function* () {
 		const logPath = logFilePath(dataDir)
@@ -249,6 +258,7 @@ const startDetached = (
 			offline,
 			chdbConfigFile,
 			onDirtyStore,
+			minimumRawTelemetryRetentionDays,
 		})
 
 		const child = yield* Effect.try({
@@ -303,6 +313,7 @@ export const start = Command.make("start", {
 	port,
 	dataDir: dataDirFlag,
 	chdbConfigFile: chdbConfigFileFlag,
+	minimumRawTelemetryRetentionDays: minimumRawTelemetryRetentionDaysFlag,
 	background: backgroundFlag,
 	offline: offlineFlag,
 	reset: resetFlag,
@@ -424,6 +435,8 @@ export const start = Command.make("start", {
 				})
 			}
 
+			const requestedRetentionDays = Option.getOrUndefined(a.minimumRawTelemetryRetentionDays)
+
 			// Detached: spawn the same command without --background and exit.
 			if (a.background)
 				return yield* startDetached(
@@ -434,6 +447,7 @@ export const start = Command.make("start", {
 					a.offline,
 					Option.getOrUndefined(a.chdbConfigFile),
 					a.onDirtyStore,
+					requestedRetentionDays,
 				)
 
 			yield* Effect.sync(() =>
@@ -474,6 +488,7 @@ export const start = Command.make("start", {
 						port: a.port,
 						dataDir,
 						configFile: Option.getOrUndefined(a.chdbConfigFile),
+						minimumRawTelemetryRetentionDays: requestedRetentionDays,
 						assets,
 					}).pipe(
 						Effect.mapError((e) => new ServerError({ message: `failed to start: ${e.message}` })),

apps/cli/src/server/archives/export.ts

@@ -60,10 +60,13 @@ import { type ArchiveSignal } from "./signals"
  * an EXPLICIT isNull(c) flag as a separate hash argument, so NULL-ness is never
  * conflated with a value. An unknown value fails closed.
  */
-export const COMPLEX_DIGEST_ALGORITHM = "cityhash64-multiset-v3"
+export const COMPLEX_DIGEST_ALGORITHM = "cityhash64-bounded-multiset-v4"
 
 /** Digest algorithms this reader accepts in a manifest shard record. */
-export const KNOWN_COMPLEX_DIGEST_ALGORITHMS: ReadonlySet<string> = new Set(["cityhash64-multiset-v3"])
+export const KNOWN_COMPLEX_DIGEST_ALGORITHMS: ReadonlySet<string> = new Set([
+	"cityhash64-multiset-v3",
+	COMPLEX_DIGEST_ALGORITHM,
+])
 
 export interface ExportSettings {
 	readonly writerThreads: number
@@ -176,7 +179,10 @@ export interface SourceColumn {
  * shard's reopened schema is compared against this to prove the schema
  * round-tripped — not just that it has "some" columns.
  */
-export const captureSourceSchema = (db: Chdb, signal: ArchiveSignal): ReadonlyArray<SourceColumn> => {
+export const captureSourceSchema = (
+	db: Pick<Chdb, "query">,
+	signal: ArchiveSignal,
+): ReadonlyArray<SourceColumn> => {
 	const rows = readRows(db.query(`DESCRIBE ${signal.name} FORMAT JSONEachRow`, "JSONEachRow"))
 	const cols = rows.map((r) => ({ name: String(r.name), type: String(r.type) }))
 	if (cols.length === 0) throw new Error(`source table ${signal.name} has no columns`)
@@ -336,23 +342,36 @@ const normalizeValueForHash = (name: string, type: string): string => {
 }
 
 /**
- * The multiset complex-value digest of a slice: the sorted multiset of per-row
- * position-bound hashes, folded into one hash. Order-independent (sorted) so it
- * tolerates row-order differences between source and reopened Parquet, yet it
- * preserves row identity + multiplicity — so it detects:
+ * A fixed-memory multiset digest of a slice. Four independent commutative
+ * accumulators over position-bound row hashes make it order-independent while
+ * retaining row identity and multiplicity. Unlike the former sorted
+ * `groupArray`, memory usage does not grow with the number of rows. It detects:
  *   - a same-typed column swap (each affected row's hash changes),
  *   - cross-row value reassociation (a row's hash changes),
  *   - duplicate-one/drop-another (the multiset of row hashes changes),
  * all of which preserve count and time extrema and defeated the round-4
- * commutative per-column sum. Measured at maxShardRows (500k): 41ms, +15MiB RSS.
+ * commutative per-column sum.
  *
  * `sliceFrom` is the FROM clause (e.g. `traces` or `file('p', Parquet)`, with a
  * WHERE predicate already applied where needed). The sort is inside chDB; no
  * rows are materialized in JavaScript.
  */
-const multisetDigestSql = (sourceSchema: ReadonlyArray<SourceColumn>, sliceFrom: string): string => {
+export const multisetDigestSql = (sourceSchema: ReadonlyArray<SourceColumn>, sliceFrom: string): string => {
 	const args = perRowHashArgs(sourceSchema)
-	return `SELECT toString(cityHash64(groupArray(h))) AS d FROM (SELECT cityHash64(${args}) AS h FROM ${sliceFrom} ORDER BY h)`
+	return `SELECT concat(toString(count()), ':', toString(sumWithOverflow(h)), ':', toString(groupBitXor(h)), ':', toString(sumWithOverflow(cityHash64(h)))) AS d FROM (SELECT cityHash64(${args}) AS h FROM ${sliceFrom})`
+}
+
+/** Compute the canonical order-independent digest used by archive validation. */
+export const computeMultisetDigest = (
+	db: Pick<Chdb, "query">,
+	sourceSchema: ReadonlyArray<SourceColumn>,
+	sliceFrom: string,
+): string => {
+	const rows = readRows(db.query(multisetDigestSql(sourceSchema, sliceFrom), "JSONEachRow"))
+	const digest = rows[0]?.d
+	if (typeof digest !== "string" || !/^\d+:\d+:\d+:\d+$/.test(digest))
+		throw new Error(`invalid multiset digest result: ${String(digest)}`)
+	return digest
 }
 
 /**

apps/cli/src/server/archives/generation.ts

@@ -41,7 +41,7 @@ import {
 	newArchiveGenerationId,
 	nextMidnightUtc,
 	rangeRoot,
-	validateRangeDate,
+	validateSealedRangeDate,
 } from "./paths"
 import { type ArchiveSignal, archiveSignal } from "./signals"
 import { COMPLEX_DIGEST_ALGORITHM, exportSignalShards, type WrittenShard } from "./export"
@@ -87,6 +87,8 @@ import {
 // reported; only provably owned `building/<gen>/` temporary output is removed.
 
 export interface ArchiveGenerationFaults {
+	/** Pause after all pre-lock checks but before maintenance-lock acquisition. */
+	readonly beforeMaintenanceLock?: () => void | Promise<void>
 	readonly afterPinAcquired?: () => void | Promise<void>
 	readonly afterScratchRestored?: () => void | Promise<void>
 	readonly afterBuildingCreated?: () => void | Promise<void>
@@ -336,7 +338,14 @@ export const createArchiveGeneration = async (
 	faults: ArchiveGenerationFaults = {},
 	loadedTuningConfig: LoadedTuningConfig | null = null,
 ): Promise<ArchiveGenerationResult> => {
-	validateRangeDate(rangeDate)
+	validateSealedRangeDate(rangeDate)
+	// This invariant belongs at the mutation boundary, not only in the CLI:
+	// callers must never supersede durable retirement evidence with a new
+	// empty/partial archive generation.
+	const { readRetiredDayLedger } = await import("./retention")
+	if (readRetiredDayLedger(dataDir).retiredDays.some((day) => day.rangeDate === rangeDate)) {
+		throw new Error(`refusing archive create: UTC day ${rangeDate} is permanently retired`)
+	}
 	assertArchiveRootSeparate(archiveDir, dataDir)
 	if (resolve(archiveDir) !== resolve(tuning.archiveDir)) {
 		throw new Error(
@@ -356,10 +365,17 @@ export const createArchiveGeneration = async (
 	const pinPurpose = `archive:${generationId}`
 	const scratchSubdir = `archive-${operationId}`
 
+	await faults.beforeMaintenanceLock?.()
 	return withMaintenanceLock(dataDir, operationId, async () => {
 		// Step 1: reconcile any prior interrupted operation before allocating a
 		// new one. This is the crash-recovery entry point.
 		await reconcileArchiveGeneration(dataDir, archiveDir, tuning.scratchRoot, faults)
+		// The pre-lock check is only a fast refusal. Retirement uses this same
+		// maintenance lock, so this re-read is the authoritative check that closes
+		// the create-versus-retire TOCTOU window.
+		if (readRetiredDayLedger(dataDir).retiredDays.some((day) => day.rangeDate === rangeDate)) {
+			throw new Error(`refusing archive create: UTC day ${rangeDate} is permanently retired`)
+		}
 		// Step 2: resolve and validate the checkpoint so its immutable backup size
 		// can be included in scratch-volume capacity planning. This is read-only.
 		const resolved = await resolveCheckpoint(dataDir, parseCheckpointSelector(checkpointSelector))