diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e70ff6547..5b67f8255 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/apps/cli/src/commands/archive.ts b/apps/cli/src/commands/archive.ts index ecc4804de..2448d04c8 100644 --- a/apps/cli/src/commands/archive.ts +++ b/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, diff --git a/apps/cli/src/commands/server-args.ts b/apps/cli/src/commands/server-args.ts index 52c653b13..a5aca9ade 100644 --- a/apps/cli/src/commands/server-args.ts +++ b/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"] : []), ] } diff --git a/apps/cli/src/commands/server.ts b/apps/cli/src/commands/server.ts index 07be5622e..f8c777932 100644 --- a/apps/cli/src/commands/server.ts +++ b/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 => 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}` })), diff --git a/apps/cli/src/server/archives/export.ts b/apps/cli/src/server/archives/export.ts index ccf53d8d1..2f58ab7a7 100644 --- a/apps/cli/src/server/archives/export.ts +++ b/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 = new Set(["cityhash64-multiset-v3"]) +export const KNOWN_COMPLEX_DIGEST_ALGORITHMS: ReadonlySet = 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 => { +export const captureSourceSchema = ( + db: Pick, + signal: ArchiveSignal, +): ReadonlyArray => { 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, sliceFrom: string): string => { +export const multisetDigestSql = (sourceSchema: ReadonlyArray, 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, + sourceSchema: ReadonlyArray, + 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 } /** diff --git a/apps/cli/src/server/archives/generation.ts b/apps/cli/src/server/archives/generation.ts index ddededd89..7cec9e08f 100644 --- a/apps/cli/src/server/archives/generation.ts +++ b/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//` temporary output is removed. export interface ArchiveGenerationFaults { + /** Pause after all pre-lock checks but before maintenance-lock acquisition. */ + readonly beforeMaintenanceLock?: () => void | Promise readonly afterPinAcquired?: () => void | Promise readonly afterScratchRestored?: () => void | Promise readonly afterBuildingCreated?: () => void | Promise @@ -336,7 +338,14 @@ export const createArchiveGeneration = async ( faults: ArchiveGenerationFaults = {}, loadedTuningConfig: LoadedTuningConfig | null = null, ): Promise => { - 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)) diff --git a/apps/cli/src/server/archives/listing.ts b/apps/cli/src/server/archives/listing.ts index 9da929719..f34873673 100644 --- a/apps/cli/src/server/archives/listing.ts +++ b/apps/cli/src/server/archives/listing.ts @@ -188,21 +188,11 @@ export interface ArchiveVerification { readonly verifiedBytes: number } -/** Explicitly verify every selected active shard against its manifest SHA-256. - * Listing errors fail the operation before any partial success is reported. */ -export const verifyActiveGenerations = async ( +const verifyGenerationSummaries = async ( archiveDir: string, - signal?: ArchiveSignalName, + active: ReadonlyArray, + signals: ReadonlyArray, ): Promise => { - const listing = listActiveGenerations(archiveDir) - const relevantErrors = signal ? listing.errors.filter((error) => error.signal === signal) : listing.errors - if (relevantErrors.length > 0) { - const detail = relevantErrors - .map((error) => `${error.signal}/${error.rangeStart || "(root)"}: ${error.error}`) - .join("; ") - throw new Error(`refusing archive integrity verification: ${detail}`) - } - const active = signal ? listing.active.filter((summary) => summary.signal === signal) : listing.active let shardCount = 0 let verifiedBytes = 0 for (const summary of active) { @@ -242,13 +232,58 @@ export const verifyActiveGenerations = async ( } return { archiveDir, - signals: signal ? [signal] : listing.signals, + signals, generationCount: active.length, shardCount, verifiedBytes, } } +/** Explicitly verify every selected active shard against its manifest SHA-256. + * Listing errors fail the operation before any partial success is reported. */ +export const verifyActiveGenerations = async ( + archiveDir: string, + signal?: ArchiveSignalName, +): Promise => { + const listing = listActiveGenerations(archiveDir) + const relevantErrors = signal ? listing.errors.filter((error) => error.signal === signal) : listing.errors + if (relevantErrors.length > 0) { + const detail = relevantErrors + .map((error) => `${error.signal}/${error.rangeStart || "(root)"}: ${error.error}`) + .join("; ") + throw new Error(`refusing archive integrity verification: ${detail}`) + } + const active = signal ? listing.active.filter((summary) => summary.signal === signal) : listing.active + return verifyGenerationSummaries(archiveDir, active, signal ? [signal] : listing.signals) +} + +/** Verify exactly one active generation for a signal/day without re-hashing + * unrelated archive history. Used immediately before destructive live + * retirement. */ +export const verifyActiveGeneration = async ( + archiveDir: string, + signal: ArchiveSignalName, + rangeDate: string, +): Promise => { + const listing = listActiveGenerations(archiveDir) + const relevantErrors = listing.errors.filter( + (error) => error.signal === signal && (error.rangeStart === "" || error.rangeStart === rangeDate), + ) + if (relevantErrors.length > 0) { + const detail = relevantErrors + .map((error) => `${error.signal}/${error.rangeStart || "(root)"}: ${error.error}`) + .join("; ") + throw new Error(`refusing archive integrity verification: ${detail}`) + } + const active = listing.active.filter( + (summary) => summary.signal === signal && summary.rangeStart === rangeDate, + ) + if (active.length !== 1) { + throw new Error(`archive date ${rangeDate} lacks one active ${signal} generation`) + } + return verifyGenerationSummaries(archiveDir, active, [signal]) +} + /** * Resolve the active Parquet shard paths for one signal across all sealed * ranges, excluding superseded generations. This is the machine-readable output diff --git a/apps/cli/src/server/archives/manifest.ts b/apps/cli/src/server/archives/manifest.ts index a69622925..cb61b70c1 100644 --- a/apps/cli/src/server/archives/manifest.ts +++ b/apps/cli/src/server/archives/manifest.ts @@ -319,10 +319,6 @@ const parseShardRecord = ( ) } const bytes = requiredCount(value, "bytes") - const complexDigest = requiredString(value, "complexDigest") - if (!/^[0-9]+$/.test(complexDigest)) { - throw new Error(`invalid archive shard complexDigest (must be a numeric digest): ${complexDigest}`) - } const complexDigestAlgorithm = requiredString(value, "complexDigestAlgorithm") if (!KNOWN_COMPLEX_DIGEST_ALGORITHMS.has(complexDigestAlgorithm)) { throw new Error( @@ -330,6 +326,11 @@ const parseShardRecord = ( `(known: ${[...KNOWN_COMPLEX_DIGEST_ALGORITHMS].join(", ")}); the manifest is preserved as-is`, ) } + const complexDigest = requiredString(value, "complexDigest") + const digestPattern = + complexDigestAlgorithm === "cityhash64-bounded-multiset-v4" ? /^\d+:\d+:\d+:\d+$/ : /^\d+$/ + if (!digestPattern.test(complexDigest)) + throw new Error(`invalid archive shard complexDigest for ${complexDigestAlgorithm}: ${complexDigest}`) return { name, rowCount, diff --git a/apps/cli/src/server/archives/paths.ts b/apps/cli/src/server/archives/paths.ts index f3198b858..9c027eb57 100644 --- a/apps/cli/src/server/archives/paths.ts +++ b/apps/cli/src/server/archives/paths.ts @@ -56,6 +56,17 @@ export const nextMidnightUtc = (rangeDate: string): string => { return date.toISOString() } +/** Require a UTC day to have ended, optionally with an additional lateness lag. */ +export const validateSealedRangeDate = (value: string, sealingLagHours = 0, now = Date.now()): string => { + const rangeDate = validateRangeDate(value) + if (!Number.isSafeInteger(sealingLagHours) || sealingLagHours < 0) + throw new Error(`sealing lag hours must be a non-negative integer: ${sealingLagHours}`) + const eligibleAt = Date.parse(nextMidnightUtc(rangeDate)) + sealingLagHours * 60 * 60 * 1000 + if (now < eligibleAt) + throw new Error(`UTC day ${rangeDate} is not sealed until ${new Date(eligibleAt).toISOString()}`) + return rangeDate +} + export const newArchiveGenerationId = (): string => validateArchiveId(randomUUID(), "archive generation") export const archiveRoot = (archiveDir: string): string => resolve(archiveDir) diff --git a/apps/cli/src/server/archives/retention.ts b/apps/cli/src/server/archives/retention.ts new file mode 100644 index 000000000..d2cbcb9ec --- /dev/null +++ b/apps/cli/src/server/archives/retention.ts @@ -0,0 +1,650 @@ +import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto" +import { existsSync, lstatSync, readFileSync } from "node:fs" +import { mkdir, rm } from "node:fs/promises" +import { dirname, join, resolve } from "node:path" +import type { Chdb } from "../chdb" +import { durableJson, durableRemove, durableRename, durableWrite, syncDirectory } from "../durable-files" +import { withMaintenanceLock } from "../checkpoints" +import { computeMultisetDigest, captureSourceSchema } from "./export" +import { assertCatalogExact, listActiveGenerations, rebuildCatalog, verifyActiveGeneration } from "./listing" +import { readArchiveGenerationManifest } from "./manifest" +import { ARCHIVE_SIGNALS, type ArchiveSignalName } from "./signals" +import { + archiveRoot, + assertNoSymlink, + generationManifestPath, + rangeRoot, + shardsRoot, + validateArchiveId, + validateRangeDate, + validateSealedRangeDate, +} from "./paths" + +const LEDGER_FORMAT_VERSION = 1 +const EXPIRATION_FORMAT_VERSION = 3 +const TOKEN_BYTES = 32 +const SHA256 = /^[0-9a-f]{64}$/ + +export interface RetiredSignalEvidence { + readonly signal: ArchiveSignalName + readonly generationId: string + readonly manifestSha256: string + readonly archivedRowCount: number + readonly contentDigest: string +} + +export interface RetiredDayRecord { + readonly rangeDate: string + readonly retiredAt: string + readonly signals: ReadonlyArray +} + +export interface RetiredDayLedger { + readonly formatVersion: 1 + readonly retiredDays: ReadonlyArray +} + +interface ExpireOperation { + readonly formatVersion: 3 + readonly operationId: string + readonly rangeDate: string + readonly completedSignals: ReadonlyArray + readonly removingSignal: ArchiveSignalName | null + readonly archivedGenerations: Readonly> +} + +type Db = Pick + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value) + +const exactKeys = (value: Record, keys: readonly string[], label: string): void => { + const expected = new Set(keys) + for (const key of keys) if (!Object.hasOwn(value, key)) throw new Error(`${label} is missing ${key}`) + for (const key of Object.keys(value)) + if (!expected.has(key)) throw new Error(`${label} has unknown field ${key}`) +} + +const requiredString = (value: Record, key: string, label: string): string => { + const field = value[key] + if (typeof field !== "string" || field.length === 0) throw new Error(`${label}.${key} must be a string`) + return field +} + +const requiredCount = (value: Record, key: string, label: string): number => { + const field = value[key] + if (typeof field !== "number" || !Number.isSafeInteger(field) || field < 0) + throw new Error(`${label}.${key} must be a non-negative safe integer`) + return field +} + +const parseSignalEvidence = (value: unknown): RetiredSignalEvidence => { + if (!isRecord(value)) throw new Error("retired-day signal evidence must be a record") + exactKeys( + value, + ["signal", "generationId", "manifestSha256", "archivedRowCount", "contentDigest"], + "retired-day signal evidence", + ) + const signal = requiredString(value, "signal", "retired-day signal evidence") + if (!ARCHIVE_SIGNALS.some((candidate) => candidate.name === signal)) + throw new Error(`unknown retired-day signal: ${signal}`) + const manifestSha256 = requiredString(value, "manifestSha256", "retired-day signal evidence") + if (!SHA256.test(manifestSha256)) throw new Error("retired-day manifestSha256 must be lowercase SHA-256") + const contentDigest = requiredString(value, "contentDigest", "retired-day signal evidence") + if (!/^\d+:\d+:\d+:\d+$/.test(contentDigest)) + throw new Error("retired-day contentDigest must be a canonical bounded multiset digest") + return { + signal: signal as ArchiveSignalName, + generationId: validateArchiveId( + requiredString(value, "generationId", "retired-day signal evidence"), + "retired-day generation", + ), + manifestSha256, + archivedRowCount: requiredCount(value, "archivedRowCount", "retired-day signal evidence"), + contentDigest, + } +} + +const parseDayRecord = (value: unknown): RetiredDayRecord => { + if (!isRecord(value)) throw new Error("retired-day record must be a record") + exactKeys(value, ["rangeDate", "retiredAt", "signals"], "retired-day record") + const signalsRaw = value.signals + if (!Array.isArray(signalsRaw)) throw new Error("retired-day signals must be an array") + const signals = signalsRaw.map(parseSignalEvidence) + const names = signals.map((signal) => signal.signal) + if (names.length !== ARCHIVE_SIGNALS.length || new Set(names).size !== ARCHIVE_SIGNALS.length) + throw new Error("retired-day record must contain each raw signal exactly once") + for (let index = 0; index < ARCHIVE_SIGNALS.length; index++) + if (names[index] !== ARCHIVE_SIGNALS[index]!.name) + throw new Error("retired-day signals are not in canonical order") + const retiredAt = requiredString(value, "retiredAt", "retired-day record") + const retiredMillis = Date.parse(retiredAt) + if (Number.isNaN(retiredMillis) || new Date(retiredMillis).toISOString() !== retiredAt) + throw new Error("retired-day retiredAt must be canonical ISO-8601") + return { + rangeDate: validateRangeDate(requiredString(value, "rangeDate", "retired-day record")), + retiredAt, + signals, + } +} + +export const parseRetiredDayLedger = (value: unknown): RetiredDayLedger => { + if (!isRecord(value)) throw new Error("retired-day ledger must be a record") + exactKeys(value, ["formatVersion", "retiredDays"], "retired-day ledger") + if (value.formatVersion !== LEDGER_FORMAT_VERSION) + throw new Error(`unsupported retired-day ledger formatVersion: ${String(value.formatVersion)}`) + if (!Array.isArray(value.retiredDays)) throw new Error("retired-day ledger retiredDays must be an array") + const retiredDays = value.retiredDays.map(parseDayRecord) + const dates = retiredDays.map((day) => day.rangeDate) + if (dates.some((date, index) => index > 0 && dates[index - 1]! >= date)) + throw new Error("retired-day ledger dates must be unique and sorted") + return { formatVersion: LEDGER_FORMAT_VERSION, retiredDays } +} + +export const retiredDayLedgerPath = (dataDir: string): string => `${resolve(dataDir)}.retired-days.json` +export const maintenanceTokenPath = (dataDir: string): string => `${resolve(dataDir)}.maintenance-token` + +const readRealFile = (path: string, label: string): string => { + const stat = lstatSync(path) + if (stat.isSymbolicLink() || !stat.isFile()) throw new Error(`${label} is not a real file: ${path}`) + return readFileSync(path, "utf8") +} + +export const readRetiredDayLedger = (dataDir: string): RetiredDayLedger => { + const path = retiredDayLedgerPath(dataDir) + if (!existsSync(path)) return { formatVersion: LEDGER_FORMAT_VERSION, retiredDays: [] } + return parseRetiredDayLedger(JSON.parse(readRealFile(path, "retired-day ledger")) as unknown) +} + +export const ensureMaintenanceToken = async (dataDir: string): Promise => { + const path = maintenanceTokenPath(dataDir) + if (!existsSync(path)) await durableWrite(path, `${randomBytes(TOKEN_BYTES).toString("hex")}\n`) + const token = readRealFile(path, "maintenance token").trim() + if (!/^[0-9a-f]{64}$/.test(token)) throw new Error("maintenance token is malformed") + return token +} + +export const maintenanceTokenMatches = (expected: string, supplied: string | null): boolean => { + if (supplied === null) return false + const left = Buffer.from(expected) + const right = Buffer.from(supplied) + return left.length === right.length && timingSafeEqual(left, right) +} + +const eventDateKey: Readonly> = { + traces: "start_time", + logs: "timestamp", + metrics_sum: "timestamp", + metrics_gauge: "timestamp", + metrics_histogram: "timestamp", + metrics_exponential_histogram: "timestamp", +} + +export class RetiredDayAuthority { + readonly #dataDir: string + #ledger: RetiredDayLedger + #dates: Set + + constructor(dataDir: string) { + this.#dataDir = dataDir + this.#ledger = readRetiredDayLedger(dataDir) + this.#dates = new Set(this.#ledger.retiredDays.map((day) => day.rangeDate)) + } + + isRetired(rangeDate: string): boolean { + return this.#dates.has(rangeDate) + } + + hasRetiredDays(): boolean { + return this.#dates.size > 0 + } + + record(rangeDate: string): RetiredDayRecord | undefined { + return this.#ledger.retiredDays.find((day) => day.rangeDate === rangeDate) + } + + assertBatchAllowed(datasource: string, ndjson: string): void { + const filtered = this.filterBatch(datasource, ndjson) + if (filtered.rejected > 0) + throw new Error(`telemetry batch contains ${filtered.rejected} row(s) from a retired UTC day`) + } + + filterBatch( + datasource: string, + ndjson: string, + ): { readonly ndjson: string; readonly accepted: number; readonly rejected: number } { + const key = eventDateKey[datasource] + if (!key) throw new Error(`unknown raw telemetry datasource: ${datasource}`) + const accepted: string[] = [] + let rejected = 0 + for (const line of ndjson.split("\n")) { + if (line.length === 0) continue + const row = JSON.parse(line) as Record + const timestamp = row[key] + if (typeof timestamp !== "string" || !/^\d{4}-\d{2}-\d{2} /.test(timestamp)) + throw new Error(`encoded ${datasource} row has no canonical UTC timestamp`) + const rangeDate = timestamp.slice(0, 10) + if (this.isRetired(rangeDate)) rejected++ + else accepted.push(line) + } + return { ndjson: accepted.join("\n"), accepted: accepted.length, rejected } + } + + async commit(record: RetiredDayRecord): Promise { + const parsed = parseDayRecord(record) + const existing = this.record(parsed.rangeDate) + if (existing) { + if (JSON.stringify(existing) !== JSON.stringify(parsed)) + throw new Error(`retired-day evidence changed for ${parsed.rangeDate}`) + return + } + const retiredDays = [...this.#ledger.retiredDays, parsed].sort((a, b) => + a.rangeDate.localeCompare(b.rangeDate), + ) + const next = parseRetiredDayLedger({ formatVersion: LEDGER_FORMAT_VERSION, retiredDays }) + await durableJson(retiredDayLedgerPath(this.#dataDir), next) + this.#ledger = next + this.#dates.add(parsed.rangeDate) + } + + replay(db: Db): void { + removeRetiredDays(db, [...this.#dates]) + } + + replayDay(db: Db, rangeDate: string): void { + if (!this.isRetired(rangeDate)) throw new Error(`UTC day ${rangeDate} is not retired`) + removeUtcDay(db, rangeDate) + } +} + +const parseCount = (text: string, label: string): number => { + const line = text.split("\n").find((candidate) => candidate.trim().length > 0) + const row = line ? (JSON.parse(line) as Record) : {} + const count = Number(row.count ?? row["count()"] ?? 0) + if (!Number.isSafeInteger(count) || count < 0) throw new Error(`invalid ${label} count`) + return count +} + +const liveCount = (db: Db, table: string, column: string, rangeDate: string): number => + parseCount( + db.query( + `SELECT count() AS count FROM ${table} WHERE toDate(${column}, 'UTC') = '${rangeDate}'`, + "JSONEachRow", + ), + `${table}/${rangeDate}`, + ) + +const removeUtcDay = (db: Db, rangeDate: string): void => { + for (const signal of ARCHIVE_SIGNALS) { + db.exec( + `ALTER TABLE ${signal.name} DELETE WHERE toDate(${signal.eventTimeColumn}, 'UTC') = '${rangeDate}' SETTINGS mutations_sync = 2`, + ) + if (liveCount(db, signal.name, signal.eventTimeColumn, rangeDate) !== 0) + throw new Error(`retired UTC day remains in ${signal.name}/${rangeDate}`) + } +} + +const sqlDates = (dates: readonly string[]): string => dates.map((date) => `'${date}'`).join(", ") + +/** Startup repair is bounded by table count, not ledger length. */ +const removeRetiredDays = (db: Db, dates: readonly string[]): void => { + if (dates.length === 0) return + const allDates = sqlDates(dates) + for (const signal of ARCHIVE_SIGNALS) { + const rows = db.query( + `SELECT DISTINCT toString(toDate(${signal.eventTimeColumn}, 'UTC')) AS rangeDate FROM ${signal.name} WHERE toDate(${signal.eventTimeColumn}, 'UTC') IN (${allDates})`, + "JSONEachRow", + ) + const present = rows + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => (JSON.parse(line) as { rangeDate?: unknown }).rangeDate) + .filter((date): date is string => typeof date === "string" && dates.includes(date)) + if (present.length === 0) continue + db.exec( + `ALTER TABLE ${signal.name} DELETE WHERE toDate(${signal.eventTimeColumn}, 'UTC') IN (${sqlDates(present)}) SETTINGS mutations_sync = 2`, + ) + const remaining = parseCount( + db.query( + `SELECT count() AS count FROM ${signal.name} WHERE toDate(${signal.eventTimeColumn}, 'UTC') IN (${sqlDates(present)})`, + "JSONEachRow", + ), + `${signal.name}/retired-days`, + ) + if (remaining !== 0) throw new Error(`retired UTC days remain in ${signal.name}`) + } +} + +export const assertSealedUtcDay = (range: string, sealingLagHours = 24, now = Date.now()): string => { + return validateSealedRangeDate(range, sealingLagHours, now) +} + +const sha256File = (path: string): string => createHash("sha256").update(readFileSync(path)).digest("hex") +const safeSqlPath = (path: string): string => { + if (path.includes("'") || path.includes("\\")) throw new Error(`unsafe archive path for chDB: ${path}`) + return path +} + +const completeArchiveDay = (archiveDir: string, rangeDate: string) => { + const listing = listActiveGenerations(archiveDir) + const errors = listing.errors.filter((error) => error.rangeStart === rangeDate || error.rangeStart === "") + if (errors.length > 0) + throw new Error( + `archive date ${rangeDate} is malformed: ${errors.map((error) => error.error).join("; ")}`, + ) + return ARCHIVE_SIGNALS.map((signal) => { + const matches = listing.active.filter( + (generation) => generation.signal === signal.name && generation.rangeStart === rangeDate, + ) + if (matches.length !== 1) + throw new Error(`archive date ${rangeDate} lacks one active ${signal.name} generation`) + return { signal, summary: matches[0]! } + }) +} + +/** + * Execute the destructive boundary inside the live server while ingest and + * local queries are quiesced by the caller. The durable ledger is committed + * before any DELETE. A crash after that boundary is repaired by startup replay; + * a crash before it leaves all live data intact. + */ +export const retireLiveDayInServer = async (options: { + readonly db: Db + readonly authority: RetiredDayAuthority + readonly archiveDir: string + readonly rangeDate: string + readonly sealingLagHours?: number + readonly now?: number +}): Promise => { + const rangeDate = assertSealedUtcDay( + options.rangeDate, + options.sealingLagHours ?? 24, + options.now ?? Date.now(), + ) + const committed = options.authority.record(rangeDate) + if (committed) { + options.authority.replayDay(options.db, rangeDate) + return committed + } + + const complete = completeArchiveDay(options.archiveDir, rangeDate) + const evidence: RetiredSignalEvidence[] = [] + for (const { signal, summary } of complete) { + await verifyActiveGeneration(options.archiveDir, signal.name, rangeDate) + const manifestPath = generationManifestPath( + options.archiveDir, + signal.name, + rangeDate, + summary.generationId, + ) + const manifest = readArchiveGenerationManifest( + options.archiveDir, + signal.name, + rangeDate, + summary.generationId, + ) + const count = liveCount(options.db, signal.name, signal.eventTimeColumn, rangeDate) + if (count !== manifest.archivedRowCount) + throw new Error( + `refusing live retirement: ${signal.name}/${rangeDate} live=${count} archive=${manifest.archivedRowCount}`, + ) + const schema = captureSourceSchema(options.db, signal) + const liveDigest = computeMultisetDigest( + options.db, + schema, + `${signal.name} WHERE toDate(${signal.eventTimeColumn}, 'UTC') = '${rangeDate}'`, + ) + let archiveDigest = liveDigest + if (manifest.archivedRowCount > 0) { + const glob = safeSqlPath( + join( + shardsRoot(options.archiveDir, signal.name, rangeDate, summary.generationId), + "*.parquet", + ), + ) + archiveDigest = computeMultisetDigest(options.db, schema, `file('${glob}', Parquet)`) + } + if (liveDigest !== archiveDigest) + throw new Error(`refusing live retirement: ${signal.name}/${rangeDate} content digest mismatch`) + evidence.push({ + signal: signal.name, + generationId: summary.generationId, + manifestSha256: sha256File(manifestPath), + archivedRowCount: manifest.archivedRowCount, + contentDigest: archiveDigest, + }) + } + + // Re-read identities and re-hash every shard immediately before the durable + // commit. No destructive step can resume from stale verification evidence. + for (const item of evidence) { + await verifyActiveGeneration(options.archiveDir, item.signal, rangeDate) + const current = completeArchiveDay(options.archiveDir, rangeDate).find( + (candidate) => candidate.signal.name === item.signal, + )! + if (current.summary.generationId !== item.generationId) + throw new Error( + `archive generation changed before retirement commit: ${item.signal}/${rangeDate}`, + ) + const actualManifestSha = sha256File( + generationManifestPath(options.archiveDir, item.signal, rangeDate, item.generationId), + ) + if (actualManifestSha !== item.manifestSha256) + throw new Error(`archive manifest changed before retirement commit: ${item.signal}/${rangeDate}`) + } + + const record: RetiredDayRecord = { + rangeDate, + retiredAt: new Date(options.now ?? Date.now()).toISOString(), + signals: evidence, + } + await options.authority.commit(record) + removeUtcDay(options.db, rangeDate) + return record +} + +const expirationRecord = (archiveDir: string): string => + join(archiveRoot(archiveDir), ".retention", "expire.json") +const expirationTomb = (archiveDir: string, date: string, signal: string): string => + join(archiveRoot(archiveDir), ".retention", "expired", date, signal) + +const parseExpireOperation = (value: unknown): ExpireOperation => { + if (!isRecord(value)) throw new Error("archive expiration journal must be a record") + exactKeys( + value, + [ + "formatVersion", + "operationId", + "rangeDate", + "completedSignals", + "removingSignal", + "archivedGenerations", + ], + "archive expiration journal", + ) + if (value.formatVersion !== EXPIRATION_FORMAT_VERSION) + throw new Error(`unsupported archive expiration journal version: ${String(value.formatVersion)}`) + if (!Array.isArray(value.completedSignals)) throw new Error("completedSignals must be an array") + const completedSignals = value.completedSignals.map((signal) => { + if (typeof signal !== "string" || !ARCHIVE_SIGNALS.some((candidate) => candidate.name === signal)) + throw new Error(`invalid completed signal: ${String(signal)}`) + return signal as ArchiveSignalName + }) + if (new Set(completedSignals).size !== completedSignals.length) + throw new Error("completedSignals contains duplicates") + for (let index = 0; index < completedSignals.length; index++) + if (completedSignals[index] !== ARCHIVE_SIGNALS[index]!.name) + throw new Error("completedSignals must be a canonical completed prefix") + const removingSignal = value.removingSignal + if ( + removingSignal !== null && + (typeof removingSignal !== "string" || + !ARCHIVE_SIGNALS.some((candidate) => candidate.name === removingSignal)) + ) + throw new Error(`invalid removingSignal: ${String(removingSignal)}`) + const expectedRemoving = ARCHIVE_SIGNALS[completedSignals.length]?.name ?? null + if (removingSignal !== null && removingSignal !== expectedRemoving) + throw new Error("removingSignal must be the next canonical signal") + if (!isRecord(value.archivedGenerations)) throw new Error("archivedGenerations must be a record") + exactKeys( + value.archivedGenerations, + ARCHIVE_SIGNALS.map((signal) => signal.name), + "archivedGenerations", + ) + const archivedGenerations = Object.fromEntries( + ARCHIVE_SIGNALS.map((signal) => [ + signal.name, + validateArchiveId( + requiredString( + value.archivedGenerations as Record, + signal.name, + "archivedGenerations", + ), + "archive expiration generation", + ), + ]), + ) as Record + return { + formatVersion: EXPIRATION_FORMAT_VERSION, + operationId: validateArchiveId( + requiredString(value, "operationId", "archive expiration journal"), + "archive expiration operation", + ), + rangeDate: validateRangeDate(requiredString(value, "rangeDate", "archive expiration journal")), + completedSignals, + removingSignal: removingSignal as ArchiveSignalName | null, + archivedGenerations, + } +} + +const readExpireOperation = (path: string): ExpireOperation | null => { + if (!existsSync(path)) return null + return parseExpireOperation(JSON.parse(readRealFile(path, "archive expiration journal")) as unknown) +} + +/** Expire one already-retired archived UTC day. Resumes safely after interruption. */ +export const expireArchiveDay = async (options: { + readonly dataDir: string + readonly archiveDir: string + readonly scratchRoot: string + readonly rangeDate: string +}): Promise => { + const rangeDate = validateRangeDate(options.rangeDate) + await withMaintenanceLock(options.dataDir, randomUUID(), async () => { + const { reconcileArchiveGeneration } = await import("./generation") + await reconcileArchiveGeneration(options.dataDir, options.archiveDir, options.scratchRoot) + const retired = readRetiredDayLedger(options.dataDir).retiredDays.find( + (day) => day.rangeDate === rangeDate, + ) + if (!retired) throw new Error(`refusing archive expiration: UTC day ${rangeDate} is not retired`) + const path = expirationRecord(options.archiveDir) + let op = readExpireOperation(path) + if (!op) { + const complete = completeArchiveDay(options.archiveDir, rangeDate) + for (const signal of ARCHIVE_SIGNALS) assertCatalogExact(options.archiveDir, signal.name) + const archivedGenerations = Object.fromEntries( + complete.map(({ signal, summary }) => [signal.name, summary.generationId]), + ) as Record + for (const item of retired.signals) + if (archivedGenerations[item.signal] !== item.generationId) + throw new Error( + `active archive generation differs from retired authority: ${item.signal}/${rangeDate}`, + ) + op = { + formatVersion: EXPIRATION_FORMAT_VERSION, + operationId: randomUUID(), + rangeDate, + completedSignals: [], + removingSignal: null, + archivedGenerations, + } + await durableJson(path, op) + } else if (op.rangeDate !== rangeDate) { + throw new Error(`unfinished archive expiration for ${op.rangeDate}`) + } + for (const signal of ARCHIVE_SIGNALS) { + if (op.completedSignals.includes(signal.name)) continue + if (op.removingSignal !== null && op.removingSignal !== signal.name) + throw new Error(`archive expiration journal is out of canonical order at ${signal.name}`) + const source = rangeRoot(options.archiveDir, signal.name, rangeDate) + const tomb = expirationTomb(options.archiveDir, rangeDate, signal.name) + await assertNoSymlink(archiveRoot(options.archiveDir), source, "archive expiration source") + await assertNoSymlink(archiveRoot(options.archiveDir), tomb, "archive expiration tombstone") + if (op.removingSignal === null && !existsSync(source)) + throw new Error( + `archive expiration source vanished before durable removal intent: ${signal.name}/${rangeDate}`, + ) + if (existsSync(source)) { + const listing = listActiveGenerations(options.archiveDir) + const relevantErrors = listing.errors.filter( + (error) => + error.signal === signal.name && + (error.rangeStart === rangeDate || error.rangeStart === ""), + ) + if (relevantErrors.length > 0) + throw new Error(`archive became malformed during expiration: ${signal.name}/${rangeDate}`) + const current = listing.active.filter( + (candidate) => candidate.signal === signal.name && candidate.rangeStart === rangeDate, + ) + if (current.length !== 1 || current[0]!.generationId !== op.archivedGenerations[signal.name]) + throw new Error( + `archive generation changed after expiration intent: ${signal.name}/${rangeDate}`, + ) + await verifyActiveGeneration(options.archiveDir, signal.name, rangeDate) + } + if (existsSync(source) && existsSync(tomb)) + throw new Error( + `archive expiration has both source and tombstone: ${signal.name}/${rangeDate}`, + ) + if (op.removingSignal === null) { + op = { ...op, removingSignal: signal.name } + await durableJson(path, op) + } + if (existsSync(source) && !existsSync(tomb)) { + await mkdir(dirname(tomb), { recursive: true, mode: 0o700 }) + await durableRename(source, tomb) + } + if (existsSync(tomb)) { + await rm(tomb, { recursive: true, force: true }) + await syncDirectory(dirname(tomb)) + } + await rebuildCatalog(options.archiveDir, signal.name) + op = { + ...op, + completedSignals: [...op.completedSignals, signal.name], + removingSignal: null, + } + await durableJson(path, op) + } + await durableRemove(path) + }) +} + +export const retireLiveDay = async (options: { + readonly dataDir: string + readonly archiveDir: string + readonly scratchRoot: string + readonly rangeDate: string + readonly port: number + readonly sealingLagHours?: number + readonly request?: typeof fetch +}): Promise => { + const rangeDate = assertSealedUtcDay(options.rangeDate, options.sealingLagHours ?? 24) + await withMaintenanceLock(options.dataDir, randomUUID(), async () => { + const { reconcileArchiveGeneration } = await import("./generation") + await reconcileArchiveGeneration(options.dataDir, options.archiveDir, options.scratchRoot) + const token = readRealFile(maintenanceTokenPath(options.dataDir), "maintenance token").trim() + const request = options.request ?? fetch + const response = await request(`http://127.0.0.1:${options.port}/local/retention/retire`, { + method: "POST", + headers: { "content-type": "application/json", "x-maple-maintenance-token": token }, + body: JSON.stringify({ + archiveDir: resolve(options.archiveDir), + rangeDate, + sealingLagHours: options.sealingLagHours ?? 24, + }), + }) + if (!response.ok) + throw new Error((await response.text().catch(() => "")) || `HTTP ${response.status}`) + }) +} diff --git a/apps/cli/src/server/chdb.ts b/apps/cli/src/server/chdb.ts index f7209a187..416094c37 100644 --- a/apps/cli/src/server/chdb.ts +++ b/apps/cli/src/server/chdb.ts @@ -12,8 +12,10 @@ import { CString, dlopen, FFIType, type Pointer, ptr, read, toArrayBuffer } from "bun:ffi" import { Effect, Schema, type Scope } from "effect" import { existsSync } from "node:fs" +import { lstatSync, readFileSync } from "node:fs" import { homedir } from "node:os" -import { dirname, join } from "node:path" +import { dirname, join, resolve } from "node:path" +import { durableJson } from "./durable-files" import { markStoreClosed, markStoreOpen, storeHasData } from "./store-version" /** A chDB failure — locating libchdb, opening the connection, or bootstrapping @@ -85,6 +87,114 @@ export interface ChdbOptions { readonly configFile?: string /** Apply the Maple schema after connect. Defaults to true. */ readonly bootstrapSchema?: boolean + /** Loaded persistent floor; not a transient launch-only setting. */ + readonly rawTelemetryRetentionDays?: number +} + +const RAW_TELEMETRY_TTL_COLUMNS = [ + ["logs", "TimestampTime"], + ["traces", "Timestamp"], + ["metrics_sum", "TimeUnix"], + ["metrics_gauge", "TimeUnix"], + ["metrics_histogram", "TimeUnix"], + ["metrics_exponential_histogram", "TimeUnix"], +] as const + +export const MINIMUM_RAW_TELEMETRY_RETENTION_DAYS = 90 +export const MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS = 3_650 + +interface RawTelemetryRetentionConfig { + readonly formatVersion: 1 + readonly minimumDays: number +} + +export const rawTelemetryRetentionConfigPath = (dataDir: string): string => + `${resolve(dataDir)}.raw-telemetry-retention.json` + +const parseRawTelemetryRetentionDays = (value: unknown): number => { + if (typeof value !== "object" || value === null || Array.isArray(value)) + throw new Error("raw telemetry retention config must be a record") + const record = value as Record + if (Object.keys(record).sort().join(",") !== "formatVersion,minimumDays" || record.formatVersion !== 1) + throw new Error("unsupported or malformed raw telemetry retention config") + const days = record.minimumDays + if ( + typeof days !== "number" || + !Number.isSafeInteger(days) || + days < MINIMUM_RAW_TELEMETRY_RETENTION_DAYS || + days > MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS + ) + throw new Error( + `raw telemetry retention minimum must be an integer from ${MINIMUM_RAW_TELEMETRY_RETENTION_DAYS} through ${MAXIMUM_RAW_TELEMETRY_RETENTION_DAYS} days`, + ) + return days +} + +export const readRawTelemetryRetentionDays = (dataDir: string): number | undefined => { + const path = rawTelemetryRetentionConfigPath(dataDir) + if (!existsSync(path)) return undefined + const stat = lstatSync(path) + if (stat.isSymbolicLink() || !stat.isFile()) + throw new Error(`raw telemetry retention config is not a real file: ${path}`) + return parseRawTelemetryRetentionDays(JSON.parse(readFileSync(path, "utf8")) as unknown) +} + +export const configureRawTelemetryRetentionDays = async ( + dataDir: string, + minimumDays: number, +): Promise => { + const days = parseRawTelemetryRetentionDays({ formatVersion: 1, minimumDays }) + const existing = readRawTelemetryRetentionDays(dataDir) + if (existing !== undefined && days < existing) + throw new Error( + `refusing to shorten persistent raw telemetry retention from ${existing} to ${days} days`, + ) + const config: RawTelemetryRetentionConfig = { formatVersion: 1, minimumDays: days } + await durableJson(rawTelemetryRetentionConfigPath(dataDir), config) +} + +export const rawTelemetryTtlStatements = (days: number): ReadonlyArray => { + const validated = parseRawTelemetryRetentionDays({ formatVersion: 1, minimumDays: days }) + return RAW_TELEMETRY_TTL_COLUMNS.map( + ([table, column]) => `ALTER TABLE ${table} MODIFY TTL toDate(${column}) + INTERVAL ${validated} DAY`, + ) +} + +const existingTtlDays = (createTableQuery: string, table: string): number => { + const match = /\bTTL\s+toDate\([^)]*\)\s*\+\s*(?:toIntervalDay\((\d+)\)|INTERVAL\s+(\d+)\s+DAY)/i.exec( + createTableQuery, + ) + const value = Number(match?.[1] ?? match?.[2]) + if (!Number.isSafeInteger(value) || value < 1) + throw new Error(`cannot determine existing raw telemetry TTL for ${table}`) + return value +} + +/** Apply a floor without shortening a higher TTL already present in the schema. */ +export const applyRawTelemetryRetentionFloor = (db: Pick, days: number): void => { + const validated = parseRawTelemetryRetentionDays({ formatVersion: 1, minimumDays: days }) + const names = RAW_TELEMETRY_TTL_COLUMNS.map(([table]) => `'${table}'`).join(", ") + const rows = db + .query( + `SELECT name, create_table_query FROM system.tables WHERE database = 'default' AND name IN (${names}) ORDER BY name`, + "JSONEachRow", + ) + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line) as { name?: unknown; create_table_query?: unknown }) + const definitions = new Map( + rows.map((row) => { + if (typeof row.name !== "string" || typeof row.create_table_query !== "string") + throw new Error("invalid system.tables TTL metadata") + return [row.name, row.create_table_query] as const + }), + ) + for (const [table, column] of RAW_TELEMETRY_TTL_COLUMNS) { + const definition = definitions.get(table) + if (!definition) throw new Error(`raw telemetry table is missing: ${table}`) + if (existingTtlDays(definition, table) >= validated) continue + db.exec(`ALTER TABLE ${table} MODIFY TTL toDate(${column}) + INTERVAL ${validated} DAY`) + } } /** Build the embedded ClickHouse argv. Keep table metadata loading and restore @@ -149,7 +259,14 @@ export class Chdb { ) const db = new Chdb(sym, connPtrPtr, conn) - if (options.bootstrapSchema !== false) db.#bootstrap(options.schemaSql) + // Partition expressions, ingest conversions, and retention predicates must + // never inherit a host-specific timezone. + db.exec("SET session_timezone = 'UTC'") + if (options.bootstrapSchema !== false) { + db.#bootstrap(options.schemaSql) + if (options.rawTelemetryRetentionDays !== undefined) + applyRawTelemetryRetentionFloor(db, options.rawTelemetryRetentionDays) + } return db } diff --git a/apps/cli/src/server/checkpoints.ts b/apps/cli/src/server/checkpoints.ts index c1434cad6..848e1a64a 100644 --- a/apps/cli/src/server/checkpoints.ts +++ b/apps/cli/src/server/checkpoints.ts @@ -434,52 +434,58 @@ const localQueryError = (status: number, detail: string, cause = detail): LocalQ cause, }) +export const postLoopbackLocalQuery = async (port: number, sql: string): Promise => { + const response = await fetch(`http://127.0.0.1:${port}/local/query`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ sql }), + }) + if (!response.ok) { + const detail = await response.text().catch(() => "") + throw localQueryError(response.status, detail) + } + return response.json() +} + export const checkpointQueryUrl = (host: string, port: number): string => `${serverUrl(host, port)}/local/query` -const postLocalQuery = ( +const postCheckpointBackup = ( host: string, port: number, - sql: string, + dataDir: string, + checkpointId: CheckpointId, ): Effect.Effect => { - const url = checkpointQueryUrl(host, port) + const url = `${serverUrl(host, port)}/local/checkpoint/backup` return Effect.gen(function* () { + const token = yield* Effect.try({ + try: () => readFileSync(`${resolve(dataDir)}.maintenance-token`, "utf8").trim(), + catch: (error) => localQueryError(0, `failed to read maintenance token: ${errorMessage(error)}`), + }) const client = yield* HttpClient.HttpClient const request = HttpClientRequest.post(url).pipe( - HttpClientRequest.bodyText(JSON.stringify({ sql }), "application/json"), + HttpClientRequest.setHeader("x-maple-maintenance-token", token), + HttpClientRequest.bodyText(JSON.stringify({ checkpointId }), "application/json"), ) const response = yield* client .execute(request) .pipe(Effect.mapError((error) => localQueryError(0, errorMessage(error), errorCause(error)))) - yield* Effect.annotateCurrentSpan("http.response.status_code", response.status) - const text = yield* response.text.pipe( + const responseText = yield* response.text.pipe( Effect.mapError((error) => localQueryError(response.status, errorMessage(error), errorCause(error)), ), ) - if (response.status < 200 || response.status >= 300) { - const detail = text - return yield* localQueryError(response.status, detail) - } + if (response.status < 200 || response.status >= 300) + return yield* localQueryError(response.status, responseText) return yield* Effect.try({ - try: () => JSON.parse(text) as unknown, + try: () => JSON.parse(responseText) as unknown, catch: (error) => localQueryError(response.status, errorMessage(error), errorCause(error)), }) }).pipe( Effect.timeout("30 seconds"), Effect.catchTag("TimeoutError", () => - Effect.fail(localQueryError(0, "local checkpoint query timed out after 30 seconds")), + Effect.fail(localQueryError(0, "local checkpoint backup timed out after 30 seconds")), ), - Effect.withSpan("CheckpointService.postLocalQuery", { - kind: "client", - attributes: { - "peer.service": "maple-local", - "http.request.method": "POST", - "server.address": host, - "server.port": port, - "url.full": url, - }, - }), ) } @@ -1498,11 +1504,7 @@ export const createCheckpoint = Effect.fn("CheckpointService.create")(function* }, catch: createError, }) - yield* postLocalQuery( - options.host, - options.port, - `BACKUP DATABASE default TO Disk('default', '${snapshotBackupSqlPath(checkpointId)}')`, - ).pipe( + yield* postCheckpointBackup(options.host, options.port, options.dataDir, checkpointId).pipe( Effect.mapError((error) => createError( isMissingBackupConfigurationError(error) diff --git a/apps/cli/src/server/otlp/proto.ts b/apps/cli/src/server/otlp/proto.ts index 1887b5225..0a945aed9 100644 --- a/apps/cli/src/server/otlp/proto.ts +++ b/apps/cli/src/server/otlp/proto.ts @@ -363,6 +363,22 @@ message ExportLogsServiceRequest { message ExportMetricsServiceRequest { repeated ResourceMetrics resource_metrics = 1; } + +message ExportTracePartialSuccess { + int64 rejected_spans = 1; + string error_message = 2; +} +message ExportTraceServiceResponse { ExportTracePartialSuccess partial_success = 1; } +message ExportLogsPartialSuccess { + int64 rejected_log_records = 1; + string error_message = 2; +} +message ExportLogsServiceResponse { ExportLogsPartialSuccess partial_success = 1; } +message ExportMetricsPartialSuccess { + int64 rejected_data_points = 1; + string error_message = 2; +} +message ExportMetricsServiceResponse { ExportMetricsPartialSuccess partial_success = 1; } ` /** @@ -375,6 +391,11 @@ const otlpRoot = protobuf.parse(PROTO_SRC, { keepCase: false }).root const ExportTraceServiceRequest = otlpRoot.lookupType("ExportTraceServiceRequest") const ExportLogsServiceRequest = otlpRoot.lookupType("ExportLogsServiceRequest") const ExportMetricsServiceRequest = otlpRoot.lookupType("ExportMetricsServiceRequest") +const responseTypes = { + traces: otlpRoot.lookupType("ExportTraceServiceResponse"), + logs: otlpRoot.lookupType("ExportLogsServiceResponse"), + metrics: otlpRoot.lookupType("ExportMetricsServiceResponse"), +} as const /** * Normalize a decoded protobuf message into the same plain-object shape the @@ -420,3 +441,19 @@ export function encodeMetricsRequest(obj: unknown): Uint8Array { const message = ExportMetricsServiceRequest.fromObject(obj as Record) return ExportMetricsServiceRequest.encode(message).finish() } + +export function encodeExportResponse( + signal: "traces" | "logs" | "metrics", + rejected: number, + errorMessage: string, +): Uint8Array { + const type = responseTypes[signal] + const rejectedField = + signal === "traces" + ? { rejectedSpans: rejected } + : signal === "logs" + ? { rejectedLogRecords: rejected } + : { rejectedDataPoints: rejected } + const object = rejected > 0 ? { partialSuccess: { ...rejectedField, errorMessage } } : {} + return type.encode(type.fromObject(object)).finish() +} diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 1c59d4dec..e692fad24 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -7,12 +7,30 @@ import * as ManagedRuntime from "effect/ManagedRuntime" import { gunzipSync } from "node:zlib" import { TelemetryLayer } from "../core/telemetry" import { isLoopbackHostname } from "../lib/local-address" -import { acquireChdb, type Chdb, type ChdbError } from "./chdb" +import { + acquireChdb, + type Chdb, + ChdbError, + configureRawTelemetryRetentionDays, + readRawTelemetryRetentionDays, + rawTelemetryTtlStatements, +} from "./chdb" import { buildInsertStatements } from "./inserts" import { encodeLogs, encodeMetrics, encodeTraces, type EncodedBatch, OtlpFieldError } from "./otlp/encode" -import { decodeLogsRequest, decodeMetricsRequest, decodeTraceRequest } from "./otlp/proto" +import { + decodeLogsRequest, + decodeMetricsRequest, + decodeTraceRequest, + encodeExportResponse, +} from "./otlp/proto" import schemaSql from "./schema/local-schema.sql" with { type: "text" } import { schemaFingerprint } from "./store-version" +import { + ensureMaintenanceToken, + maintenanceTokenMatches, + RetiredDayAuthority, + retireLiveDayInServer, +} from "./archives/retention" /** Fingerprint of the schema this build bootstraps stores with. Stamped into the * store marker so `maple start` can rebuild a store left on an older schema. */ @@ -34,6 +52,7 @@ export interface ServerOptions { readonly port: number readonly dataDir: string readonly configFile?: string + readonly minimumRawTelemetryRetentionDays?: number /** Serves the bundled SPA; omit to disable the UI (API-only). */ readonly assets?: AssetResolver } @@ -149,7 +168,12 @@ interface IngestResult { readonly requestBytes: number } -async function ingest(db: Chdb, signal: Signal, req: Request): Promise { +async function ingest( + db: Chdb, + authority: RetiredDayAuthority, + signal: Signal, + req: Request, +): Promise { const raw = new Uint8Array(await req.arrayBuffer()) const requestBytes = raw.length const contentType = req.headers.get("content-type") ?? "" @@ -178,6 +202,12 @@ async function ingest(db: Chdb, signal: Signal, req: Request): Promise { + const filtered = authority.filterBatch(batch.datasource, batch.ndjson) + rejected += filtered.rejected + return { ...batch, ndjson: filtered.ndjson, rowCount: filtered.accepted } + }) let accepted = 0 for (const batch of batches) { if (batch.rowCount === 0) continue @@ -194,7 +224,28 @@ async function ingest(db: Chdb, signal: Signal, req: Request): Promise 0 ? "telemetry from permanently retired UTC days was rejected" : "" + if (contentType.includes("json")) { + const rejectedField = + signal === "traces" + ? { rejectedSpans: rejected } + : signal === "logs" + ? { rejectedLogRecords: rejected } + : { rejectedDataPoints: rejected } + return { + response: json(rejected > 0 ? { partialSuccess: { ...rejectedField, errorMessage } } : {}), + accepted, + requestBytes, + } + } + return { + response: new Response(encodeExportResponse(signal, rejected, errorMessage), { + status: 200, + headers: { "content-type": "application/x-protobuf" }, + }), + accepted, + requestBytes, + } } /** @@ -226,7 +277,7 @@ interface QueryResult { readonly sql: string | undefined } -async function handleQuery(db: Chdb, req: Request): Promise { +async function handleQuery(db: Chdb, authority: RetiredDayAuthority, req: Request): Promise { let sql: string try { const body = (await req.json()) as { sql?: unknown } @@ -238,6 +289,15 @@ async function handleQuery(db: Chdb, req: Request): Promise { } let out: string const started = performance.now() + const readOnly = /^\s*(?:SELECT|WITH|SHOW|DESCRIBE|DESC|EXPLAIN|EXISTS)\b/i.test(sql) + if (!readOnly && authority.hasRetiredDays()) { + return { + response: text("local SQL writes are disabled after the first UTC day is retired", 405), + rowCount: 0, + durationMs: Math.round(performance.now() - started), + sql, + } + } try { out = db.query(forceJsonEachRow(sql)) } catch (error) { @@ -319,12 +379,18 @@ const recoverResponse = (self: Effect.Effect): Effect. /** OTLP-ingest request as a `Server`-kind span, mirroring the Rust gateway * (`apps/ingest`): `maple.signal`, item count, request size, HTTP semconv. */ -const ingestSpan = (runSpan: SpanRunner, db: Chdb, signal: Signal, req: Request): Promise => +const ingestSpan = ( + runSpan: SpanRunner, + db: Chdb, + authority: RetiredDayAuthority, + signal: Signal, + req: Request, +): Promise => runSpan( recoverResponse( Effect.gen(function* () { const { response, accepted, requestBytes } = yield* Effect.promise(() => - ingest(db, signal, req), + ingest(db, authority, signal, req), ) yield* Effect.annotateCurrentSpan({ "http.request.body.size": requestBytes, @@ -346,12 +412,17 @@ const ingestSpan = (runSpan: SpanRunner, db: Chdb, signal: Signal, req: Request) ) /** `/local/query` request as a `Server`-kind span with the canonical DB attrs. */ -const querySpan = (runSpan: SpanRunner, db: Chdb, req: Request): Promise => +const querySpan = ( + runSpan: SpanRunner, + db: Chdb, + authority: RetiredDayAuthority, + req: Request, +): Promise => runSpan( recoverResponse( Effect.gen(function* () { const { response, rowCount, durationMs, sql } = yield* Effect.promise(() => - handleQuery(db, req), + handleQuery(db, authority, req), ) yield* Effect.annotateCurrentSpan({ "db.system.name": "clickhouse", @@ -370,11 +441,132 @@ const querySpan = (runSpan: SpanRunner, db: Chdb, req: Request): Promise void> = [] + + enter(): (() => void) | null { + if (this.#closed) return null + this.#active++ + let released = false + return () => { + if (released) return + released = true + this.#active-- + if (this.#active === 0) { + for (const resolve of this.#drained.splice(0)) resolve() + } + } + } + + async exclusive(work: () => Promise): Promise { + if (this.#closed) throw new Error("another server maintenance operation is active") + this.#closed = true + try { + if (this.#active > 0) await new Promise((resolve) => this.#drained.push(resolve)) + return await work() + } finally { + this.#closed = false + } + } +} + +const admitted = async (gate: RequestQuiescenceGate, work: () => Promise): Promise => { + const leave = gate.enter() + if (!leave) return text("server maintenance in progress", 503) + try { + return await work() + } finally { + leave() + } +} + +const handleRetirement = async ( + db: Chdb, + authority: RetiredDayAuthority, + gate: RequestQuiescenceGate, + token: string, + req: Request, +): Promise => { + if (!maintenanceTokenMatches(token, req.headers.get("x-maple-maintenance-token"))) + return text("maintenance authorization required", 403) + let body: unknown + try { + body = await req.json() + } catch { + return text("invalid JSON body", 400) + } + if (typeof body !== "object" || body === null || Array.isArray(body)) return text("invalid body", 400) + const record = body as Record + const keys = Object.keys(record).sort().join(",") + if (keys !== "archiveDir,rangeDate,sealingLagHours") return text("invalid retirement fields", 400) + if ( + typeof record.archiveDir !== "string" || + typeof record.rangeDate !== "string" || + typeof record.sealingLagHours !== "number" + ) + return text("invalid retirement values", 400) + try { + const retired = await gate.exclusive(() => + retireLiveDayInServer({ + db, + authority, + archiveDir: record.archiveDir as string, + rangeDate: record.rangeDate as string, + sealingLagHours: record.sealingLagHours as number, + }), + ) + return json(retired) + } catch (error) { + return text(`retirement failed: ${error instanceof Error ? error.message : String(error)}`, 409) + } +} + +const CHECKPOINT_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i + +/** Typed, authenticated replacement for sending BACKUP through /local/query. */ +const handleCheckpointBackup = async (db: Chdb, token: string, req: Request): Promise => { + if (!maintenanceTokenMatches(token, req.headers.get("x-maple-maintenance-token"))) + return text("maintenance authorization required", 403) + let body: unknown + try { + body = await req.json() + } catch { + return text("invalid JSON body", 400) + } + if (typeof body !== "object" || body === null || Array.isArray(body)) return text("invalid body", 400) + const record = body as Record + if (Object.keys(record).sort().join(",") !== "checkpointId" || typeof record.checkpointId !== "string") + return text("invalid checkpoint fields", 400) + if (!CHECKPOINT_ID.test(record.checkpointId)) return text("invalid checkpoint ID", 400) + try { + db.exec( + `BACKUP DATABASE default TO Disk('default', 'backups/snapshots/${record.checkpointId.toLowerCase()}/backup')`, + ) + return json({ checkpointId: record.checkpointId.toLowerCase() }) + } catch (error) { + return text( + `checkpoint backup failed: ${error instanceof Error ? error.message : String(error)}`, + 400, + ) + } +} + /** The `Bun.serve` fetch handler, closed over the chDB connection. Each ingest * and query request is run through `runSpan` so it leaves a trace; `/health` * and `OPTIONS` are skipped (loop-prevention convention — no health-check noise). */ const makeFetch = - (db: Chdb, options: ServerOptions, runSpan: SpanRunner) => + ( + db: Chdb, + options: ServerOptions, + runSpan: SpanRunner, + authority: RetiredDayAuthority, + gate: RequestQuiescenceGate, + maintenanceToken: string, + ) => async (req: Request): Promise => { const url = new URL(req.url) const origin = req.headers.get("origin") @@ -386,10 +578,18 @@ const makeFetch = if (req.method === "OPTIONS") return new Response(null, { status: 204, headers: corsHeaders }) if (url.pathname === "/health") return respond(text("OK")) if (req.method === "POST") { - if (url.pathname === "/v1/traces") return respond(await ingestSpan(runSpan, db, "traces", req)) - if (url.pathname === "/v1/logs") return respond(await ingestSpan(runSpan, db, "logs", req)) - if (url.pathname === "/v1/metrics") return respond(await ingestSpan(runSpan, db, "metrics", req)) - if (url.pathname === "/local/query") return respond(await querySpan(runSpan, db, req)) + if (url.pathname === "/v1/traces") + return respond(await admitted(gate, () => ingestSpan(runSpan, db, authority, "traces", req))) + if (url.pathname === "/v1/logs") + return respond(await admitted(gate, () => ingestSpan(runSpan, db, authority, "logs", req))) + if (url.pathname === "/v1/metrics") + return respond(await admitted(gate, () => ingestSpan(runSpan, db, authority, "metrics", req))) + if (url.pathname === "/local/query") + return respond(await admitted(gate, () => querySpan(runSpan, db, authority, req))) + if (url.pathname === "/local/checkpoint/backup") + return respond(await admitted(gate, () => handleCheckpointBackup(db, maintenanceToken, req))) + if (url.pathname === "/local/retention/retire") + return respond(await handleRetirement(db, authority, gate, maintenanceToken, req)) } if (req.method === "GET" && options.assets) return respond(serveAsset(options.assets, url.pathname)) return respond(text("not found", 404)) @@ -404,11 +604,60 @@ export const startServer = ( options: ServerOptions, ): Effect.Effect<{ readonly port: number }, ChdbError | ServerBindError, Scope.Scope> => Effect.gen(function* () { + const retention = yield* Effect.try({ + try: () => { + const existing = readRawTelemetryRetentionDays(options.dataDir) + const requested = options.minimumRawTelemetryRetentionDays + if (requested !== undefined) rawTelemetryTtlStatements(requested) + if (existing !== undefined && requested !== undefined && requested < existing) + throw new Error( + `refusing to shorten persistent raw telemetry retention from ${existing} to ${requested} days`, + ) + return { existing, requested, effective: requested ?? existing } + }, + catch: (error) => + new ChdbError({ + message: `failed to load persistent raw telemetry retention: ${error instanceof Error ? error.message : String(error)}`, + }), + }) const db = yield* acquireChdb({ dataDir: options.dataDir, schemaSql, configFile: options.configFile, + rawTelemetryRetentionDays: retention.effective, + }) + // The ALTER statements have now been accepted by the running database. + // Only after that validation succeeds does a requested value become the + // durable configuration used by subsequent launches. + if (retention.requested !== undefined) + yield* Effect.tryPromise({ + try: () => configureRawTelemetryRetentionDays(options.dataDir, retention.requested!), + catch: (error) => + new ChdbError({ + message: `failed to persist raw telemetry retention: ${error instanceof Error ? error.message : String(error)}`, + }), + }) + const authority = yield* Effect.try({ + try: () => { + const loaded = new RetiredDayAuthority(options.dataDir) + // Checkpoint restore may have resurrected retired rows. Replay before + // the listener is bound, so no restored representation is observable. + loaded.replay(db) + return loaded + }, + catch: (error) => + new ChdbError({ + message: `failed to enforce retired-day authority: ${error instanceof Error ? error.message : String(error)}`, + }), + }) + const maintenanceToken = yield* Effect.tryPromise({ + try: () => ensureMaintenanceToken(options.dataDir), + catch: (error) => + new ChdbError({ + message: `failed to load maintenance token: ${error instanceof Error ? error.message : String(error)}`, + }), }) + const gate = new RequestQuiescenceGate() // A dedicated runtime carrying the OTel tracer for per-request spans: the // Bun.serve handler runs outside Effect, so each request's span effect is // run through this runtime. Disposed on scope close, which flushes any @@ -424,7 +673,7 @@ export const startServer = ( Bun.serve({ port: options.port, hostname: options.hostname, - fetch: makeFetch(db, options, runSpan), + fetch: makeFetch(db, options, runSpan, authority, gate, maintenanceToken), }), catch: (error) => new ServerBindError({ diff --git a/apps/cli/test/archive-export-round5.test.ts b/apps/cli/test/archive-export-round5.test.ts index 924edae6b..d4c126c5e 100644 --- a/apps/cli/test/archive-export-round5.test.ts +++ b/apps/cli/test/archive-export-round5.test.ts @@ -1,10 +1,11 @@ import { describe, it } from "@effect/vitest" -import { deepStrictEqual, throws } from "node:assert" +import { deepStrictEqual, strictEqual, throws } from "node:assert" import { randomUUID } from "node:crypto" import { normalizeType, planHourShards, COMPLEX_DIGEST_ALGORITHM, + multisetDigestSql, type ExportSettings, } from "../src/server/archives/export" import { parseArchiveGenerationManifest } from "../src/server/archives/manifest" @@ -23,6 +24,15 @@ const settings = (overrides: Partial = {}): ExportSettings => ({ ...overrides, }) +describe("bounded multiset digest", () => { + it("uses fixed-size aggregate state instead of materializing every row", () => { + const sql = multisetDigestSql([{ name: "Body", type: "String" }], "logs") + strictEqual(sql.includes("groupArray"), false) + strictEqual(sql.includes("sumWithOverflow"), true) + strictEqual(sql.includes("groupBitXor"), true) + }) +}) + describe("schema normalizeType (measured chDB Parquet mapping)", () => { it("does NOT collapse parameterized array element types", () => { if (normalizeType("Array(UInt64)") === normalizeType("Array(String)")) { @@ -156,7 +166,7 @@ const manifestWith = ( sha256: "a".repeat(64), bytes: 4096, columns: ["Timestamp"], - complexDigest: "123456789", + complexDigest: "1:2:3:4", complexDigestAlgorithm: COMPLEX_DIGEST_ALGORITHM, ...shardOverrides, }, diff --git a/apps/cli/test/archive-generation.test.ts b/apps/cli/test/archive-generation.test.ts index 5512ef928..cc0e50718 100644 --- a/apps/cli/test/archive-generation.test.ts +++ b/apps/cli/test/archive-generation.test.ts @@ -44,8 +44,11 @@ import { checkpointRoot, checkpointSnapshotDir, checkpointStatePath, + withMaintenanceLock, } from "../src/server/checkpoints" import { assertCatalogExact, rebuildCatalog } from "../src/server/archives/listing" +import { RetiredDayAuthority } from "../src/server/archives/retention" +import { ARCHIVE_SIGNALS } from "../src/server/archives/signals" // Filesystem-level tests for generation promotion, supersession, and catalog // append. These exercise the durable state machine without a restored chDB; the @@ -300,6 +303,61 @@ describe("archive generation promotion", () => { }) }) + it("rechecks retired-day authority after acquiring the maintenance lock", async () => { + await withArchive(async (archiveDir) => { + const parent = join(archiveDir, "..") + const dataDir = join(parent, "data") + const scratchRoot = join(parent, "scratch") + mkdirSync(dataDir, { recursive: true }) + mkdirSync(scratchRoot, { recursive: true }) + let reachedPreLock!: () => void + let resumeCreate!: () => void + const preLock = new Promise((resolve) => (reachedPreLock = resolve)) + const resume = new Promise((resolve) => (resumeCreate = resolve)) + const creating = createArchiveGeneration( + dataDir, + archiveDir, + "traces", + "2026-06-01", + { + writerThreads: 1, + rowGroupRows: 1, + maxShardRows: 1, + maxShardBytes: 1, + targetChunkBytes: 1, + minFreeSpaceReserve: 0, + archiveDir, + scratchRoot, + }, + "current", + { + beforeMaintenanceLock: async () => { + reachedPreLock() + await resume + }, + }, + ) + await preLock + await withMaintenanceLock(dataDir, randomUUID(), () => + new RetiredDayAuthority(dataDir).commit({ + rangeDate: "2026-06-01", + retiredAt: "2026-06-03T00:00:00.000Z", + signals: ARCHIVE_SIGNALS.map((signal) => ({ + signal: signal.name, + generationId: randomUUID(), + manifestSha256: "a".repeat(64), + archivedRowCount: 1, + contentDigest: "1:2:3:4", + })), + }), + ) + resumeCreate() + await rejects(creating, /permanently retired/) + strictEqual(listActiveOperationIds(archiveDir).length, 0) + strictEqual(existsSync(activePointerPath(archiveDir, "traces", "2026-06-01")), false) + }) + }) + it("makes the complete manifest durable in building before publishing the generation", async () => { await withArchive(async (archiveDir) => { const generationId = randomUUID() diff --git a/apps/cli/test/archive-retention.test.ts b/apps/cli/test/archive-retention.test.ts new file mode 100644 index 000000000..1c9c30c33 --- /dev/null +++ b/apps/cli/test/archive-retention.test.ts @@ -0,0 +1,402 @@ +import { describe, it } from "@effect/vitest" +import { rejects, strictEqual, throws } from "node:assert" +import { createHash, randomUUID } from "node:crypto" +import { existsSync, mkdirSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { + assertSealedUtcDay, + expireArchiveDay, + parseRetiredDayLedger, + readRetiredDayLedger, + retiredDayLedgerPath, + RetiredDayAuthority, + retireLiveDayInServer, +} from "../src/server/archives/retention" +import { ARCHIVE_SIGNALS, type ArchiveSignalName } from "../src/server/archives/signals" +import { + activePointerPath, + generationManifestPath, + nextMidnightUtc, + rangeRoot, + shardsRoot, +} from "../src/server/archives/paths" +import { rebuildCatalog } from "../src/server/archives/listing" +import type { ArchiveGenerationManifest } from "../src/server/archives/manifest" +import { CHDB_VERSION, MAPLE_VERSION } from "../src/version" +import { RequestQuiescenceGate, SCHEMA_FINGERPRINT } from "../src/server/serve" + +const sha256 = (value: string): string => createHash("sha256").update(value).digest("hex") + +const seed = (archiveDir: string, signal: ArchiveSignalName, date: string): string => { + const generationId = randomUUID() + const shardDir = shardsRoot(archiveDir, signal, date, generationId) + mkdirSync(shardDir, { recursive: true }) + const shard = join(shardDir, "00.parquet") + writeFileSync(shard, "PAR1") + const eventTimeColumn = ARCHIVE_SIGNALS.find((candidate) => candidate.name === signal)!.eventTimeColumn + const manifest: ArchiveGenerationManifest = { + formatVersion: 3, + generationId, + signal, + rangeStart: date, + rangeEndExclusive: nextMidnightUtc(date), + checkpointId: randomUUID(), + checkpointManifestFingerprint: "checkpoint:test:4", + createdAt: new Date().toISOString(), + mapleVersion: MAPLE_VERSION, + chdbVersion: CHDB_VERSION, + schemaFingerprint: SCHEMA_FINGERPRINT, + sourceRowCount: 1, + archivedRowCount: 1, + tuning: { + writerThreads: 1, + rowGroupRows: 1, + maxShardRows: 1, + maxShardBytes: 1024, + targetChunkBytes: 1024, + minFreeSpaceReserve: 1, + }, + tuningConfig: null, + shards: [ + { + name: "00.parquet", + rowCount: 1, + minEventTimeUnixNano: `${BigInt(Date.parse(`${date}T00:00:00.000Z`)) * 1_000_000n}`, + maxEventTimeUnixNano: `${BigInt(Date.parse(`${date}T00:30:00.000Z`)) * 1_000_000n}`, + sha256: sha256("PAR1"), + bytes: statSync(shard).size, + columns: [eventTimeColumn], + complexDigest: "1", + complexDigestAlgorithm: "cityhash64-multiset-v3", + }, + ], + } + writeFileSync( + generationManifestPath(archiveDir, signal, date, generationId), + `${JSON.stringify(manifest)}\n`, + ) + writeFileSync( + activePointerPath(archiveDir, signal, date), + `${JSON.stringify({ formatVersion: 1, generationId, signal, rangeStart: date, selectedAt: new Date().toISOString() })}\n`, + ) + return generationId +} + +const fixture = async ( + run: (root: string, dataDir: string, archiveDir: string, scratchRoot: string) => Promise, +): Promise => { + const root = mkdtempSync(join(tmpdir(), "maple-retention-")) + const dataDir = join(root, "data") + const archiveDir = join(root, "archive") + const scratchRoot = join(root, "scratch") + mkdirSync(dataDir) + mkdirSync(archiveDir) + mkdirSync(scratchRoot) + try { + await run(root, dataDir, archiveDir, scratchRoot) + } finally { + rmSync(root, { recursive: true, force: true }) + } +} + +const seedCompleteDay = async ( + archiveDir: string, + date: string, +): Promise> => { + const generations = Object.fromEntries( + ARCHIVE_SIGNALS.map((signal) => [signal.name, seed(archiveDir, signal.name, date)]), + ) as Record + for (const signal of ARCHIVE_SIGNALS) await rebuildCatalog(archiveDir, signal.name) + return generations +} + +const mockDb = (options: { + readonly date: string + readonly liveDigest?: string + readonly archiveDigest?: string + readonly corruptDuringDigest?: () => void +}) => { + const counts = new Map(ARCHIVE_SIGNALS.map((signal) => [signal.name, 1])) + const execs: string[] = [] + let digestQueries = 0 + return { + counts, + execs, + db: { + query(sql: string): string { + const described = ARCHIVE_SIGNALS.find( + (signal) => sql === `DESCRIBE ${signal.name} FORMAT JSONEachRow`, + ) + if (described) + return `${JSON.stringify({ name: described.eventTimeColumn, type: "DateTime64(9)" })}\n` + const counted = ARCHIVE_SIGNALS.find((signal) => sql.includes(`FROM ${signal.name} WHERE`)) + if (sql.startsWith("SELECT DISTINCT") && counted) + return counts.get(counted.name) === 0 + ? "" + : `${JSON.stringify({ rangeDate: options.date })}\n` + if (sql.startsWith("SELECT count()") && counted) + return `${JSON.stringify({ count: counts.get(counted.name) })}\n` + if (sql.startsWith("SELECT concat")) { + digestQueries++ + if (digestQueries === 1) options.corruptDuringDigest?.() + return `${JSON.stringify({ d: sql.includes("file('") ? (options.archiveDigest ?? "1:2:3:4") : (options.liveDigest ?? "1:2:3:4") })}\n` + } + throw new Error(`unexpected query: ${sql}`) + }, + exec(sql: string): void { + execs.push(sql) + const signal = ARCHIVE_SIGNALS.find((candidate) => sql.includes(`TABLE ${candidate.name} `)) + if (!signal) throw new Error(`unexpected exec: ${sql}`) + counts.set(signal.name, 0) + }, + }, + } +} + +const writeRetiredLedger = ( + dataDir: string, + date: string, + generations: Record, +): void => { + writeFileSync( + retiredDayLedgerPath(dataDir), + `${JSON.stringify({ + formatVersion: 1, + retiredDays: [ + { + rangeDate: date, + retiredAt: "2026-01-03T00:00:00.000Z", + signals: ARCHIVE_SIGNALS.map((signal) => ({ + signal: signal.name, + generationId: generations[signal.name], + manifestSha256: "a".repeat(64), + archivedRowCount: 1, + contentDigest: "1:2:3:4", + })), + }, + ], + })}\n`, + ) +} + +describe("durable retired-day authority", () => { + it("rejects malformed or non-canonical ledgers", () => { + throws( + () => parseRetiredDayLedger({ formatVersion: 1, retiredDays: [], extra: true }), + /unknown field extra/, + ) + }) + + it("rejects late OTLP rows before insertion", async () => + fixture(async (_root, dataDir) => { + const date = "2026-01-01" + const generations = Object.fromEntries( + ARCHIVE_SIGNALS.map((signal) => [signal.name, randomUUID()]), + ) as Record + writeRetiredLedger(dataDir, date, generations) + const authority = new RetiredDayAuthority(dataDir) + await rejects( + async () => + authority.assertBatchAllowed( + "logs", + `${JSON.stringify({ timestamp: `${date} 12:00:00.000000000` })}`, + ), + /retired UTC day/, + ) + })) + + it("filters retired rows while preserving current rows in a mixed batch", async () => + fixture(async (_root, dataDir) => { + const date = "2026-01-01" + const generations = Object.fromEntries( + ARCHIVE_SIGNALS.map((signal) => [signal.name, randomUUID()]), + ) as Record + writeRetiredLedger(dataDir, date, generations) + const filtered = new RetiredDayAuthority(dataDir).filterBatch( + "logs", + [ + JSON.stringify({ timestamp: `${date} 12:00:00.000000000`, body: "late" }), + JSON.stringify({ timestamp: "2026-01-04 12:00:00.000000000", body: "current" }), + ].join("\n"), + ) + strictEqual(filtered.rejected, 1) + strictEqual(filtered.accepted, 1) + strictEqual(JSON.parse(filtered.ndjson).body, "current") + })) + + it("replays retirement after checkpoint restoration before serving", async () => + fixture(async (_root, dataDir) => { + const date = "2026-01-01" + const generations = Object.fromEntries( + ARCHIVE_SIGNALS.map((signal) => [signal.name, randomUUID()]), + ) as Record + writeRetiredLedger(dataDir, date, generations) + const mocked = mockDb({ date }) + new RetiredDayAuthority(dataDir).replay(mocked.db) + for (const signal of ARCHIVE_SIGNALS) strictEqual(mocked.counts.get(signal.name), 0) + strictEqual( + mocked.execs.every((sql) => sql.includes("toDate(") && sql.includes("'UTC'")), + true, + ) + })) +}) + +describe("server-coordinated live retirement", () => { + it("commits authority before exact UTC deletion", async () => + fixture(async (_root, dataDir, archiveDir) => { + await seedCompleteDay(archiveDir, "2026-01-01") + const mocked = mockDb({ date: "2026-01-01" }) + await retireLiveDayInServer({ + db: mocked.db, + authority: new RetiredDayAuthority(dataDir), + archiveDir, + rangeDate: "2026-01-01", + now: Date.parse("2026-01-03T00:00:00.000Z"), + }) + strictEqual(readRetiredDayLedger(dataDir).retiredDays.length, 1) + strictEqual(existsSync(join(dataDir, "retention")), false) + strictEqual(mocked.execs.length, ARCHIVE_SIGNALS.length) + strictEqual( + mocked.execs.every((sql) => sql.includes("DELETE WHERE") && !sql.includes("DROP PARTITION")), + true, + ) + })) + + it("refuses equal-count, different-content data before deletion", async () => + fixture(async (_root, dataDir, archiveDir) => { + await seedCompleteDay(archiveDir, "2026-01-01") + const mocked = mockDb({ + date: "2026-01-01", + liveDigest: "1:2:3:41", + archiveDigest: "1:2:3:42", + }) + await rejects( + retireLiveDayInServer({ + db: mocked.db, + authority: new RetiredDayAuthority(dataDir), + archiveDir, + rangeDate: "2026-01-01", + now: Date.parse("2026-01-03T00:00:00.000Z"), + }), + /content digest mismatch/, + ) + strictEqual(mocked.execs.length, 0) + strictEqual(existsSync(retiredDayLedgerPath(dataDir)), false) + })) + + it("re-hashes shards after evidence collection and before commit", async () => + fixture(async (_root, dataDir, archiveDir) => { + const generations = await seedCompleteDay(archiveDir, "2026-01-01") + const shard = join(shardsRoot(archiveDir, "logs", "2026-01-01", generations.logs), "00.parquet") + const mocked = mockDb({ + date: "2026-01-01", + corruptDuringDigest: () => writeFileSync(shard, "NOPE"), + }) + await rejects( + retireLiveDayInServer({ + db: mocked.db, + authority: new RetiredDayAuthority(dataDir), + archiveDir, + rangeDate: "2026-01-01", + now: Date.parse("2026-01-03T00:00:00.000Z"), + }), + /SHA-256 mismatch/, + ) + strictEqual(mocked.execs.length, 0) + })) + + it("enforces an explicit UTC sealing lag", () => { + strictEqual(assertSealedUtcDay("2026-01-01", 24, Date.parse("2026-01-03T00:00:00Z")), "2026-01-01") + throws(() => assertSealedUtcDay("2026-01-01", 24, Date.parse("2026-01-02T23:59:59Z")), /not sealed/) + }) +}) + +describe("request quiescence", () => { + it("closes admission and drains already accepted work", async () => { + const gate = new RequestQuiescenceGate() + const leave = gate.enter()! + let ran = false + const exclusive = gate.exclusive(async () => { + ran = true + }) + strictEqual(gate.enter(), null) + strictEqual(ran, false) + leave() + await exclusive + strictEqual(ran, true) + strictEqual(typeof gate.enter(), "function") + }) +}) + +describe("archive expiration", () => { + it("expires only an already-retired complete day", async () => + fixture(async (_root, dataDir, archiveDir, scratchRoot) => { + const date = "2026-01-01" + const generations = await seedCompleteDay(archiveDir, date) + writeRetiredLedger(dataDir, date, generations) + await expireArchiveDay({ dataDir, archiveDir, scratchRoot, rangeDate: date }) + for (const signal of ARCHIVE_SIGNALS) + strictEqual(existsSync(rangeRoot(archiveDir, signal.name, date)), false) + })) + + it("refuses to delete the only archive copy of a live day", async () => + fixture(async (_root, dataDir, archiveDir, scratchRoot) => { + const date = "2026-01-01" + await seedCompleteDay(archiveDir, date) + await rejects( + expireArchiveDay({ dataDir, archiveDir, scratchRoot, rangeDate: date }), + /is not retired/, + ) + strictEqual(existsSync(rangeRoot(archiveDir, "logs", date)), true) + })) + + it("resumes absence only after a durable per-signal removal intent", async () => + fixture(async (_root, dataDir, archiveDir, scratchRoot) => { + const date = "2026-01-01" + const generations = await seedCompleteDay(archiveDir, date) + writeRetiredLedger(dataDir, date, generations) + const journalDir = join(archiveDir, ".retention") + mkdirSync(journalDir, { recursive: true }) + writeFileSync( + join(journalDir, "expire.json"), + `${JSON.stringify({ + formatVersion: 3, + operationId: randomUUID(), + rangeDate: date, + completedSignals: [], + removingSignal: "logs", + archivedGenerations: generations, + })}\n`, + ) + rmSync(rangeRoot(archiveDir, "logs", date), { recursive: true }) + await expireArchiveDay({ dataDir, archiveDir, scratchRoot, rangeDate: date }) + for (const signal of ARCHIVE_SIGNALS) + strictEqual(existsSync(rangeRoot(archiveDir, signal.name, date)), false) + })) + + it("rejects altered progress that skips a canonical signal", async () => + fixture(async (_root, dataDir, archiveDir, scratchRoot) => { + const date = "2026-01-01" + const generations = await seedCompleteDay(archiveDir, date) + writeRetiredLedger(dataDir, date, generations) + const journalDir = join(archiveDir, ".retention") + mkdirSync(journalDir, { recursive: true }) + writeFileSync( + join(journalDir, "expire.json"), + `${JSON.stringify({ + formatVersion: 3, + operationId: randomUUID(), + rangeDate: date, + completedSignals: ["traces"], + removingSignal: null, + archivedGenerations: generations, + })}\n`, + ) + await rejects( + expireArchiveDay({ dataDir, archiveDir, scratchRoot, rangeDate: date }), + /canonical completed prefix/, + ) + strictEqual(existsSync(rangeRoot(archiveDir, "logs", date)), true) + })) +}) diff --git a/apps/cli/test/chdb-args.test.ts b/apps/cli/test/chdb-args.test.ts index ed9fca70b..463bc5a10 100644 --- a/apps/cli/test/chdb-args.test.ts +++ b/apps/cli/test/chdb-args.test.ts @@ -1,6 +1,15 @@ import { describe, it } from "@effect/vitest" -import { deepStrictEqual } from "node:assert" -import { chdbArgv } from "../src/server/chdb" +import { deepStrictEqual, rejects, strictEqual, throws } from "node:assert" +import { mkdtempSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { + applyRawTelemetryRetentionFloor, + chdbArgv, + configureRawTelemetryRetentionDays, + rawTelemetryTtlStatements, + readRawTelemetryRetentionDays, +} from "../src/server/chdb" describe("embedded chDB arguments", () => { it("waits for metadata and serializes table loading and restore work", () => { @@ -28,3 +37,60 @@ describe("embedded chDB arguments", () => { ]) }) }) + +describe("persistent raw telemetry retention floor", () => { + it("requires at least the longest built-in TTL", () => { + throws(() => rawTelemetryTtlStatements(89), /90 through 3650 days/) + throws(() => rawTelemetryTtlStatements(3651), /90 through 3650 days/) + strictEqual(rawTelemetryTtlStatements(120).length, 6) + strictEqual( + rawTelemetryTtlStatements(120)[0], + "ALTER TABLE logs MODIFY TTL toDate(TimestampTime) + INTERVAL 120 DAY", + ) + }) + + it("extends lower table TTLs without shortening higher schema TTLs", () => { + const executed: string[] = [] + const tableRows = [ + ["logs", "TimestampTime", 180], + ["traces", "Timestamp", 30], + ["metrics_sum", "TimeUnix", 90], + ["metrics_gauge", "TimeUnix", 90], + ["metrics_histogram", "TimeUnix", 90], + ["metrics_exponential_histogram", "TimeUnix", 90], + ].map(([name, column, days]) => + JSON.stringify({ + name, + create_table_query: `CREATE TABLE ${name} (...) TTL toDate(${column}) + toIntervalDay(${days})`, + }), + ) + applyRawTelemetryRetentionFloor( + { + query: () => `${tableRows.join("\n")}\n`, + exec: (sql) => executed.push(sql), + }, + 120, + ) + strictEqual(executed.length, 5) + strictEqual( + executed.some((sql) => sql.includes("TABLE logs ")), + false, + ) + }) + + it("persists across launches and refuses a later shortening", async () => { + const root = mkdtempSync(join(tmpdir(), "maple-retention-config-")) + const dataDir = join(root, "data") + try { + strictEqual(readRawTelemetryRetentionDays(dataDir), undefined) + await configureRawTelemetryRetentionDays(dataDir, 120) + strictEqual(readRawTelemetryRetentionDays(dataDir), 120) + await configureRawTelemetryRetentionDays(dataDir, 180) + strictEqual(readRawTelemetryRetentionDays(dataDir), 180) + await rejects(configureRawTelemetryRetentionDays(dataDir, 120), /refusing to shorten/) + strictEqual(readRawTelemetryRetentionDays(dataDir), 180) + } finally { + rmSync(root, { recursive: true, force: true }) + } + }) +}) diff --git a/apps/cli/test/native-archive-calibrate-crash-probe.sh b/apps/cli/test/native-archive-calibrate-crash-probe.sh index cd4a02308..460736355 100755 --- a/apps/cli/test/native-archive-calibrate-crash-probe.sh +++ b/apps/cli/test/native-archive-calibrate-crash-probe.sh @@ -25,7 +25,7 @@ SCRATCH="$ROOT/scratch" CONFIG="$ROOT/backups.xml" MARKER="$ROOT/marker" SERVER_PID="" -RANGE_DATE="$(date -u +%Y-%m-%d)" +RANGE_DATE="$(date -u -d '1 day ago' +%F 2>/dev/null || date -u -v-1d +%F)" cleanup() { if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then diff --git a/apps/cli/test/native-archive-calibrate-probe.sh b/apps/cli/test/native-archive-calibrate-probe.sh index 0c5a54dc3..57a9773d6 100755 --- a/apps/cli/test/native-archive-calibrate-probe.sh +++ b/apps/cli/test/native-archive-calibrate-probe.sh @@ -32,7 +32,7 @@ ARCHIVE="$ROOT/archive" SCRATCH="$ROOT/scratch" CONFIG="$ROOT/backups.xml" SERVER_PID="" -RANGE_DATE="$(date -u +%Y-%m-%d)" +RANGE_DATE="$(date -u -d '1 day ago' +%F 2>/dev/null || date -u -v-1d +%F)" cleanup() { if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then diff --git a/apps/cli/test/native-archive-crash-recovery-probe.sh b/apps/cli/test/native-archive-crash-recovery-probe.sh index 26e3e9185..e04b3f9ee 100755 --- a/apps/cli/test/native-archive-crash-recovery-probe.sh +++ b/apps/cli/test/native-archive-crash-recovery-probe.sh @@ -45,7 +45,7 @@ declare -a FAILURES=() # ---- store + checkpoint setup (one per boundary, so each is independent) ---- ROOT="" SERVER_PID="" -RANGE_DATE="$(date -u +%Y-%m-%d)" +RANGE_DATE="$(date -u -d '1 day ago' +%F 2>/dev/null || date -u -v-1d +%F)" SIGNAL="traces" cleanup() { diff --git a/apps/cli/test/native-archive-smoke.sh b/apps/cli/test/native-archive-smoke.sh index b2282bc9f..1f99f335b 100755 --- a/apps/cli/test/native-archive-smoke.sh +++ b/apps/cli/test/native-archive-smoke.sh @@ -18,8 +18,8 @@ SCRATCH="$ROOT/scratch" CONFIG="$ROOT/backups.xml" SERVER_PID="" -# Seal the UTC day containing "now" so the ingested markers fall inside it. -RANGE_DATE="$(date -u +%Y-%m-%d)" +# Use an already sealed UTC day; archive creation rejects the current day. +RANGE_DATE="$(date -u -d '1 day ago' +%F 2>/dev/null || date -u -v-1d +%F)" cleanup() { if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then diff --git a/apps/cli/test/native-retention-probe.sh b/apps/cli/test/native-retention-probe.sh new file mode 100755 index 000000000..4aa6b49bf --- /dev/null +++ b/apps/cli/test/native-retention-probe.sh @@ -0,0 +1,212 @@ +#!/usr/bin/env bash +# Native proof for durable, server-coordinated live retirement. +set -euo pipefail + +BUNDLE_DIR="${1:?usage: native-retention-probe.sh [port]}" +MAPLE="$BUNDLE_DIR/maple" +PORT="${2:-45271}" +ROOT="$(realpath "$(mktemp -d "${TMPDIR:-/tmp}/maple-native-retention.XXXXXX")")" +DATA="$ROOT/data" +ARCHIVE="$ROOT/archive" +SCRATCH="$ROOT/scratch" +CONFIG="$ROOT/backups.xml" +SERVER_PID="" +SIGNALS=(logs traces metrics_sum metrics_gauge metrics_histogram metrics_exponential_histogram) + +if date -u -v-3d +%F >/dev/null 2>&1; then + RANGE_DATE="$(date -u -v-3d +%F)" +else + RANGE_DATE="$(date -u -d '3 days ago' +%F)" +fi + +cleanup() { + if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then + kill "$SERVER_PID" 2>/dev/null || true + wait "$SERVER_PID" 2>/dev/null || true + fi + if [[ "${KEEP_ROOT:-0}" == "1" ]]; then + echo "preserved retention root: $ROOT" >&2 + else + rm -rf "$ROOT" + fi +} +trap cleanup EXIT + +fail() { + echo "FAIL: $*" >&2 + exit 1 +} + +query() { + local sql="$1" + curl --fail-with-body -sS "http://127.0.0.1:$PORT/local/query" \ + -H 'content-type: application/json' \ + --data "$(jq -nc --arg sql "$sql" '{sql:$sql}')" +} + +wait_health() { + for _ in $(seq 1 300); do + if curl -fsS "http://127.0.0.1:$PORT/health" >/dev/null 2>&1; then return; fi + sleep 0.1 + done + fail "server did not become healthy: $(tail -100 "$ROOT/server.log" 2>/dev/null)" +} + +start_server() { + TZ=America/New_York "$MAPLE" start \ + --port "$PORT" \ + --data-dir "$DATA" \ + --chdb-config-file "$CONFIG" \ + ${INITIAL_RETENTION_CONFIG:+--minimum-raw-telemetry-retention-days 120} \ + --on-dirty-store fail \ + --offline >"$ROOT/server.log" 2>&1 & + SERVER_PID=$! + wait_health + [[ "$(query "SELECT timezone() AS timezone" | jq -r '.[0].timezone')" == "UTC" ]] \ + || fail "live chDB session is not pinned to UTC" + INITIAL_RETENTION_CONFIG="" +} + +stop_server() { + "$MAPLE" stop --data-dir "$DATA" >/dev/null + wait "$SERVER_PID" 2>/dev/null || true + SERVER_PID="" +} + +insert_markers() { + local ts="${RANGE_DATE}T12:00:00" + query "INSERT INTO logs (OrgId, Timestamp, TimestampTime, TraceId, SpanId, TraceFlags, SeverityText, SeverityNumber, ServiceName, Body) SELECT 'local', toDateTime64('${ts}.000000000', 9, 'UTC'), toDateTime('${ts}', 'UTC'), 'trace-retention', 'span-retention', 1, 'INFO', 9, 'retention-probe', 'marker'" >/dev/null + # More than the default 500k shard bound proves the day-wide retirement + # digest stays fixed-memory while comparing a multi-shard archive. + query "INSERT INTO logs (OrgId, Timestamp, TimestampTime, TraceId, SpanId, TraceFlags, SeverityText, SeverityNumber, ServiceName, Body) SELECT 'local', toDateTime64('${ts}.100000000', 9, 'UTC'), toDateTime('${ts}', 'UTC'), concat('scale-trace-', toString(number)), concat('scale-span-', toString(number)), 1, 'INFO', 9, 'retention-scale', concat('scale-', toString(number)) FROM numbers(500000)" >/dev/null + query "INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, TraceState, SpanName, SpanKind, ServiceName, StatusCode, StatusMessage) SELECT 'local', toDateTime64('${ts}.000000000', 9, 'UTC'), 'trace-retention', 'span-retention', '', '', 'marker', 'Server', 'retention-probe', 'Ok', ''" >/dev/null + query "INSERT INTO metrics_sum (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Value, AggregationTemporality, IsMonotonic) SELECT 'local', 'retention-probe', 'sum', toDateTime64('${ts}.000000000', 9, 'UTC'), toDateTime64('${ts}.000000000', 9, 'UTC'), 1, 2, true" >/dev/null + query "INSERT INTO metrics_gauge (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Value) SELECT 'local', 'retention-probe', 'gauge', toDateTime64('${ts}.000000000', 9, 'UTC'), toDateTime64('${ts}.000000000', 9, 'UTC'), 1" >/dev/null + query "INSERT INTO metrics_histogram (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Count, Sum, BucketCounts, ExplicitBounds, AggregationTemporality) SELECT 'local', 'retention-probe', 'histogram', toDateTime64('${ts}.000000000', 9, 'UTC'), toDateTime64('${ts}.000000000', 9, 'UTC'), 1, 1, [1], [1.0], 2" >/dev/null + query "INSERT INTO metrics_exponential_histogram (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Count, Sum, Scale, ZeroCount, PositiveOffset, PositiveBucketCounts, NegativeOffset, NegativeBucketCounts, AggregationTemporality) SELECT 'local', 'retention-probe', 'exponential', toDateTime64('${ts}.000000000', 9, 'UTC'), toDateTime64('${ts}.000000000', 9, 'UTC'), 1, 1, 0, 0, 0, [1], 0, [], 2" >/dev/null +} + +assert_live_count() { + local expected="$1" signal column count signal_expected + for signal in "${SIGNALS[@]}"; do + case "$signal" in + logs) column="TimestampTime" ;; + traces) column="Timestamp" ;; + *) column="TimeUnix" ;; + esac + count="$(query "SELECT count() AS count FROM $signal WHERE toDate($column, 'UTC') = '$RANGE_DATE'" | jq -r '.[0].count | tonumber')" + signal_expected="$expected" + [[ "$expected" == "1" && "$signal" == "logs" ]] && signal_expected="500001" + [[ "$count" == "$signal_expected" ]] || fail "$signal live count: expected $signal_expected, got $count" + done +} + +printf '%s\n' \ + '' \ + ' ' \ + ' default' \ + ' backups' \ + ' ' \ + '' >"$CONFIG" +chmod 600 "$CONFIG" + +echo "native retention probe root: $ROOT (range: $RANGE_DATE; host TZ: America/New_York)" +INITIAL_RETENTION_CONFIG=1 +start_server +[[ -f "${DATA}.raw-telemetry-retention.json" ]] || fail "persistent TTL floor was not recorded" +TTL_SQL="$(query "SELECT create_table_query FROM system.tables WHERE database = 'default' AND name = 'logs'" | jq -r '.[0].create_table_query')" +grep -Eq "120 DAY|toIntervalDay\(120\)" <<<"$TTL_SQL" \ + || fail "persistent 120-day TTL floor was not applied" +insert_markers +assert_live_count 1 + +"$MAPLE" checkpoint --port "$PORT" --data-dir "$DATA" >"$ROOT/checkpoint.out" 2>&1 \ + || fail "checkpoint failed: $(cat "$ROOT/checkpoint.out")" +CHECKPOINT_ID="$(jq -r '.current' "$DATA/backups/state.json")" +[[ "$CHECKPOINT_ID" =~ ^[0-9a-f-]{36}$ ]] || fail "invalid checkpoint ID: $CHECKPOINT_ID" + +for signal in "${SIGNALS[@]}"; do + "$MAPLE" archive create "$RANGE_DATE" "$signal" \ + --data-dir "$DATA" --archive-dir "$ARCHIVE" --scratch-root "$SCRATCH" \ + --checkpoint-id "$CHECKPOINT_ID" >"$ROOT/archive-$signal.out" 2>&1 \ + || fail "archive create failed for $signal: $(cat "$ROOT/archive-$signal.out")" +done +LOGS_GENERATION="$(jq -r '.generationId' "$ARCHIVE/logs/$RANGE_DATE/active.json")" +[[ "$(jq '.shards | length' "$ARCHIVE/logs/$RANGE_DATE/generations/$LOGS_GENERATION/manifest.json")" -gt 1 ]] \ + || fail "scale archive did not produce multiple logs shards" + +# Replace one archived live row with a different row while preserving its count. +# Count-only retirement would delete it; the canonical digest must refuse before +# the retired-day ledger or any live deletion exists. +query "ALTER TABLE logs DELETE WHERE toDate(TimestampTime, 'UTC') = '$RANGE_DATE' AND Body = 'marker' SETTINGS mutations_sync = 2" >/dev/null +query "INSERT INTO logs (OrgId, Timestamp, TimestampTime, ServiceName, Body) SELECT 'local', toDateTime64('${RANGE_DATE}T12:00:01.000000000', 9, 'UTC'), toDateTime('${RANGE_DATE}T12:00:01', 'UTC'), 'retention-probe', 'different-same-count-row'" >/dev/null +if "$MAPLE" archive retire-live "$RANGE_DATE" --port "$PORT" \ + --data-dir "$DATA" --archive-dir "$ARCHIVE" --scratch-root "$SCRATCH" \ + --sealing-lag-hours 24 --apply >"$ROOT/mismatch-retire.out" 2>&1; then + fail "same-count/different-content retirement unexpectedly succeeded" +fi +grep -q "content digest mismatch" "$ROOT/mismatch-retire.out" \ + || fail "retirement did not report digest mismatch: $(cat "$ROOT/mismatch-retire.out")" +[[ ! -e "${DATA}.retired-days.json" ]] || fail "failed retirement committed the retired-day ledger" +assert_live_count 1 + +# Restore the exact checkpoint used for the archives before the successful run. +stop_server +"$MAPLE" restore --yes --data-dir "$DATA" --checkpoint-id "$CHECKPOINT_ID" >"$ROOT/pre-retire-restore.out" 2>&1 \ + || fail "pre-retirement restore failed: $(cat "$ROOT/pre-retire-restore.out")" +start_server + +"$MAPLE" archive retire-live "$RANGE_DATE" --port "$PORT" \ + --data-dir "$DATA" --archive-dir "$ARCHIVE" --scratch-root "$SCRATCH" \ + --sealing-lag-hours 24 --apply >"$ROOT/retire.out" 2>&1 \ + || fail "retirement failed: $(cat "$ROOT/retire.out")" + +[[ -f "${DATA}.retired-days.json" ]] || fail "external retired-day ledger was not committed" +[[ ! -e "$DATA/retention" ]] || fail "retirement state leaked into replaceable chDB directory" +assert_live_count 0 + +# A retired day cannot be re-exported into a partial/empty generation that +# supersedes the complete generation recorded by the durable authority. +if "$MAPLE" archive create "$RANGE_DATE" logs \ + --data-dir "$DATA" --archive-dir "$ARCHIVE" --scratch-root "$SCRATCH" \ + --checkpoint-id "$CHECKPOINT_ID" >"$ROOT/late-rearchive.out" 2>&1; then + fail "archive create unexpectedly superseded a retired day" +fi +grep -q "permanently retired" "$ROOT/late-rearchive.out" \ + || fail "retired-day rearchive did not fail for the expected reason: $(cat "$ROOT/late-rearchive.out")" + +# Direct SQL mutation is rejected before execution, so neither raw data nor +# insert-trigger materialized-view targets receive a late contribution. +SQL_STATUS="$(curl -sS -o "$ROOT/late-sql.out" -w '%{http_code}' "http://127.0.0.1:$PORT/local/query" \ + -H 'content-type: application/json' \ + --data "$(jq -nc --arg sql "INSERT INTO logs (OrgId, Timestamp, TimestampTime, ServiceName, Body) SELECT 'local', toDateTime64('${RANGE_DATE}T12:00:02.000000000', 9, 'UTC'), toDateTime('${RANGE_DATE}T12:00:02', 'UTC'), 'late-sql', 'late-sql'" '{sql:$sql}')")" +[[ "$SQL_STATUS" == "405" ]] || fail "late local SQL returned $SQL_STATUS instead of 405" +[[ "$(query "SELECT count() AS count FROM logs WHERE toDate(TimestampTime, 'UTC') = '$RANGE_DATE'" | jq -r '.[0].count | tonumber')" == "0" ]] \ + || fail "local SQL recreated a retired day" +[[ "$(query "SELECT count() AS count FROM logs_aggregates_hourly WHERE ServiceName = 'late-sql'" | jq -r '.[0].count | tonumber')" == "0" ]] \ + || fail "late SQL polluted logs_aggregates_hourly" +[[ "$(query "SELECT count() AS count FROM service_usage WHERE ServiceName = 'late-sql'" | jq -r '.[0].count | tonumber')" == "0" ]] \ + || fail "late SQL polluted service_usage" + +# A mixed OTLP request partially accepts its current row and reports only the +# retired row as rejected, per OTLP partial-success semantics. +NANOS="$(python3 -c 'import datetime,sys; print(int(datetime.datetime.fromisoformat(sys.argv[1]+"T12:00:03+00:00").timestamp()*1_000_000_000))' "$RANGE_DATE")" +CURRENT_NANOS="$(python3 -c 'import time; print(time.time_ns())')" +OTLP="$(jq -nc --arg old "$NANOS" --arg current "$CURRENT_NANOS" '{resourceLogs:[{resource:{attributes:[{key:"service.name",value:{stringValue:"retention-mixed"}}]},scopeLogs:[{logRecords:[{timeUnixNano:$old,body:{stringValue:"late-otlp"}},{timeUnixNano:$current,body:{stringValue:"current-otlp"}}]}]}]}')" +STATUS="$(curl -sS -o "$ROOT/late-otlp.out" -w '%{http_code}' "http://127.0.0.1:$PORT/v1/logs" -H 'content-type: application/json' --data "$OTLP")" +[[ "$STATUS" == "200" ]] || fail "mixed OTLP returned $STATUS instead of 200: $(cat "$ROOT/late-otlp.out")" +[[ "$(jq -r '.partialSuccess.rejectedLogRecords' "$ROOT/late-otlp.out")" == "1" ]] \ + || fail "mixed OTLP did not report one rejected log record" +[[ "$(query "SELECT count() AS count FROM logs WHERE ServiceName = 'retention-mixed' AND Body = 'current-otlp'" | jq -r '.[0].count | tonumber')" == "1" ]] \ + || fail "mixed OTLP dropped the valid current log record" + +# Restore the pre-retirement checkpoint, then prove startup removes resurrected +# rows before the listener becomes healthy. +stop_server +"$MAPLE" restore --yes --data-dir "$DATA" --checkpoint-id "$CHECKPOINT_ID" >"$ROOT/restore.out" 2>&1 \ + || fail "restore failed: $(cat "$ROOT/restore.out")" +start_server +assert_live_count 0 + +stop_server +echo "PASS native retention: UTC digest retirement, late rejection, SQL enforcement, and restore replay" diff --git a/apps/cli/test/server-args.test.ts b/apps/cli/test/server-args.test.ts index 73ef0f5b7..b86f1d937 100644 --- a/apps/cli/test/server-args.test.ts +++ b/apps/cli/test/server-args.test.ts @@ -76,6 +76,7 @@ describe("buildDetachedChildArgs", () => { offline: true, chdbConfigFile: "/tmp/backup config.xml", onDirtyStore: policy, + minimumRawTelemetryRetentionDays: 120, }) deepStrictEqual(args, [ "/repo/apps/cli/src/bin.ts", @@ -92,6 +93,8 @@ describe("buildDetachedChildArgs", () => { policy, "--chdb-config-file", "/tmp/backup config.xml", + "--minimum-raw-telemetry-retention-days", + "120", "--offline", ]) strictEqual(args.filter((arg) => arg === "--on-dirty-store").length, 1) @@ -111,6 +114,7 @@ describe("buildDetachedChildArgs", () => { offline: false, chdbConfigFile: undefined, onDirtyStore: "fail", + minimumRawTelemetryRetentionDays: undefined, }), [ "start", diff --git a/docs/local-telemetry-archives.md b/docs/local-telemetry-archives.md index 8b7cc91c8..b737a5424 100644 --- a/docs/local-telemetry-archives.md +++ b/docs/local-telemetry-archives.md @@ -136,8 +136,8 @@ released after the generation is durable. Calibration pins use the purpose ## Commands -`maple archive` has six operator-facing subcommands (`create`, `list`, -`rebuild`, `reconcile`, `gc`, `calibrate`) plus the internal +`maple archive` has eight operator-facing subcommands (`create`, `list`, +`rebuild`, `reconcile`, `gc`, `calibrate`, `retire-live`, `expire`) plus the internal `calibrate-session` and `calibrate-run` commands used by calibration and its fault probes. There are no short flags anywhere in this command tree. Root flags fall back to `~/.maple` defaults when omitted. @@ -208,6 +208,44 @@ Rebuild a signal's `catalog.jsonl` from the authoritative generation manifests, recovering from a truncated or missing catalog without rescanning Parquet bytes. `` is positional. +### `maple archive retire-live --apply` + +Remove one sealed UTC day from all six live raw telemetry tables. The CLI first +reconciles interrupted archive work, then asks the running server to close +ingest/query admission, drain accepted requests, re-hash the frozen active +generation, and compare a canonical day-wide content digest—not only counts. + +The server durably commits the retired day and its archive evidence to +`.retired-days.json` before deleting any live row. That authority is +outside the replaceable chDB directory: startup repairs only retired dates that +actually reappeared before binding after a checkpoint restore. OTLP ingestion +filters retired rows while accepting current rows from the same request and +returns an OTLP partial-success response. Once any day is retired, arbitrary +SQL writes through `/local/query` are refused before execution, preventing +insert-trigger materialized views from retaining a late contribution. +A crash before the ledger commit leaves the live day intact; a crash after it is +repaired by replay. The default `--sealing-lag-hours 24` can be increased but +never bypassed accidentally. The command refuses to run without `--apply`. + +The command does not hard-code an active-data window. Scheduling policy chooses +which eligible day to retire (for example, the deployment may configure 60, +90, or 120 days). If the policy relies on database TTL as a backstop, configure +it independently with `maple start --minimum-raw-telemetry-retention-days N`. +That setting is durable beside the data directory, survives reset/restore, and +can only be increased; omitting it on later launches preserves the configured +value. Maple validates the live table TTLs before persisting the request and +never shortens a higher schema TTL. `N` must be between 90 and 3650 days, and operators should keep it strictly above +their active rotation window so TTL cannot delete a day before archival. + +### `maple archive expire --apply` + +Delete one complete, already-retired archived UTC day across all six signals. +Maple first reconciles interrupted archive work, freezes the generation IDs, +re-hashes each generation before its destructive step, tombstone-renames each +signal range before removal, rebuilds each catalog, and journals progress at +`/.retention/expire.json`. A rerun resumes only the exact frozen day +and generations. The command refuses to run without `--apply`. + ### `maple archive reconcile` Reconcile an interrupted `create` or `gc` operation to its intended state @@ -700,6 +738,8 @@ idempotently confirms an already-absent target. | **Supersession** | Same as late telemetry: the newest generation becomes active; superseded ones remain on disk until `archive gc` reclaims them. | | **Interrupted create** | Reconciles automatically on the next `create`, or via `archive reconcile`. Pre-publication output moves to retained quarantine; post-publication repairs pointer and catalog. | | **Interrupted GC** | Resumes the frozen target set; a half-removed tombstone is finished, an already-absent target is confirmed. Out-of-order mutation fails closed. | +| **Interrupted live retirement** | Before ledger commit, all live rows remain. After commit, startup replays the authoritative retired day before binding and finishes exact UTC deletion. | +| **Interrupted archive expiration** | Resumes the frozen day and generation IDs; a tombstoned range is removed and its catalog rebuilt before progress advances. | | **Interrupted calibration** | The derived-pin and owned-dir reconciliation releases the pin and removes the sample; the record is preserved until cleanup is proven. | | **Insufficient memory budget** | Calibration reports `low` confidence (or no recommendation) rather than presenting synthetic precision. | | **Failed calibration** | No config is written; temporary calibration output is cleaned up. Existing configuration is unchanged. |