fix(kvs): write record metadata sidecars for input files (#1416)

Author: janbuchar

src/commands/run.ts

@@ -22,12 +22,12 @@ import {
 	SUPPORTED_NODEJS_VERSION,
 } from '../lib/consts.js';
 import { execWithLog } from '../lib/exec.js';
-import { deleteFile } from '../lib/files.js';
 import { useActorConfig } from '../lib/hooks/useActorConfig.js';
 import { ProjectLanguage, useCwdProject } from '../lib/hooks/useCwdProject.js';
 import { useModuleVersion } from '../lib/hooks/useModuleVersion.js';
 import { CRAWLEE_INPUT_KEY_ENV, resolveInputKey, TEMP_INPUT_KEY_PREFIX } from '../lib/input-key.js';
 import { getAjvValidator, getDefaultsFromInputSchema, readInputSchema } from '../lib/input_schema.js';
+import { deleteKvsRecord, writeKvsRecord } from '../lib/kvs-metadata.js';
 import { error, info, warning } from '../lib/outputs.js';
 import { replaceSecretsValue } from '../lib/secrets.js';
 import {
@@ -39,6 +39,7 @@ import {
 	getLocalUserInfo,
 	isNodeVersionSupported,
 	isPythonVersionSupported,
+	type LocalInput,
 	purgeDefaultDataset,
 	purgeDefaultKeyValueStore,
 	purgeDefaultQueue,
@@ -50,13 +51,27 @@ interface TempInputResult {
 }
 
 interface OverwrittenInputResult {
-	existingInput: ReturnType<typeof getLocalInput>;
+	existingInput: LocalInput | undefined;
+	inputKey: string;
 	inputFilePath: string;
 	writtenAt: number;
 }
 
 type ValidateAndStoreInputResult = TempInputResult | OverwrittenInputResult;
 
+/**
+ * Write the input as a record, so a storage client resolves the bare key to `<key>.json`
+ * through the sidecar instead of probing extensions.
+ */
+const writeInputRecord = async (storePath: string, key: string, input: Record<string, unknown>) =>
+	writeKvsRecord({
+		storePath,
+		key,
+		fileName: `${key}.json`,
+		contentType: 'application/json; charset=utf-8',
+		body: JSON.stringify(input, null, 2),
+	});
+
 enum RunType {
 	DirectFile = 0,
 	Module = 1,
@@ -456,8 +471,8 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 		} finally {
 			if (storedInputResults) {
 				if ('tempInputKey' in storedInputResults) {
-					// Temp input file: just delete it, user's INPUT.json was never touched
-					await deleteFile(storedInputResults.tempInputFilePath);
+					// Temp input record: just delete it, user's INPUT.json was never touched
+					await deleteKvsRecord(storedInputResults.tempInputFilePath, storedInputResults.tempInputKey);
 				} else if (storedInputResults.existingInput) {
 					// Check if the input file was modified since we modified it. If it was, we abort the re-overwrite and warn the user
 					const stats = await stat(storedInputResults.inputFilePath);
@@ -478,7 +493,7 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 					await writeFile(storedInputResults.inputFilePath, storedInputResults.existingInput.body);
 				} else {
 					// No file -> we made it -> we delete it
-					await deleteFile(storedInputResults.inputFilePath);
+					await deleteKvsRecord(storedInputResults.inputFilePath, storedInputResults.inputKey);
 				}
 			}
 		}
@@ -511,7 +526,7 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 			const tempInputKey = `${TEMP_INPUT_KEY_PREFIX}${resolvedInputKey}`;
 			const tempInputFilePath = join(localStorePath, `${tempInputKey}.json`);
 
-			await writeFile(tempInputFilePath, JSON.stringify(inputOverride.input, null, 2));
+			await writeInputRecord(localStorePath, tempInputKey, inputOverride.input);
 
 			return {
 				tempInputKey,
@@ -570,7 +585,7 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 			const tempInputFilePath = join(localStorePath, `${tempInputKey}.json`);
 
 			await mkdir(localStorePath, { recursive: true });
-			await writeFile(tempInputFilePath, JSON.stringify(fullInputOverride, null, 2));
+			await writeInputRecord(localStorePath, tempInputKey, fullInputOverride);
 
 			return {
 				tempInputKey,
@@ -581,10 +596,11 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 		if (!existingInput) {
 			await mkdir(localStorePath, { recursive: true });
 			// No input -> use defaults for this run
-			await writeFile(inputFilePath, JSON.stringify(defaults, null, 2));
+			await writeInputRecord(localStorePath, resolvedInputKey, defaults);
 
 			return {
 				existingInput,
+				inputKey: resolvedInputKey,
 				inputFilePath,
 				writtenAt: Date.now(),
 			};
@@ -616,7 +632,7 @@ export class RunCommand extends ApifyCommand<typeof RunCommand> {
 			const tempInputKey = `${TEMP_INPUT_KEY_PREFIX}${resolvedInputKey}`;
 			const tempInputFilePath = join(localStorePath, `${tempInputKey}.json`);
 
-			await writeFile(tempInputFilePath, JSON.stringify(fullInput, null, 2));
+			await writeInputRecord(localStorePath, tempInputKey, fullInput);
 
 			return {
 				tempInputKey,

src/lib/input_schema.ts

@@ -1,4 +1,4 @@
-import { existsSync, writeFileSync } from 'node:fs';
+import { existsSync } from 'node:fs';
 import { join } from 'node:path';
 
 import type { Ajv, ErrorObject } from 'ajv';
@@ -12,6 +12,7 @@ import {
 } from '@apify/json_schemas';
 
 import { ACTOR_SPECIFICATION_FOLDER, LOCAL_CONFIG_PATH } from './consts.js';
+import { writeKvsRecord } from './kvs-metadata.js';
 import { info, warning } from './outputs.js';
 import { Ajv2019, getJsonFileContent, getLocalConfig, getLocalKeyValueStorePath } from './utils.js';
 
@@ -273,9 +274,13 @@ export const createPrefilledInputFileFromInputSchema = async (actorFolderDir: st
 			}`,
 		});
 	} finally {
-		const keyValueStorePath = getLocalKeyValueStorePath();
-		const inputJsonPath = join(actorFolderDir, keyValueStorePath, `${KEY_VALUE_STORE_KEYS.INPUT}.json`);
-		writeFileSync(inputJsonPath, JSON.stringify(inputFile, null, '\t'));
+		await writeKvsRecord({
+			storePath: join(actorFolderDir, getLocalKeyValueStorePath()),
+			key: KEY_VALUE_STORE_KEYS.INPUT,
+			fileName: `${KEY_VALUE_STORE_KEYS.INPUT}.json`,
+			contentType: 'application/json; charset=utf-8',
+			body: JSON.stringify(inputFile, null, '\t'),
+		});
 	}
 };
 

src/lib/kvs-metadata.ts

@@ -0,0 +1,97 @@
+import { readFileSync } from 'node:fs';
+import { rm, writeFile } from 'node:fs/promises';
+import { basename, dirname, join } from 'node:path';
+
+/** Name of a key-value store's own metadata file, as written by crawlee's storage clients. */
+const STORE_METADATA_FILE_NAME = '__metadata__.json';
+
+/** Suffix of a key-value store record's metadata sidecar, as written by crawlee's storage clients. */
+const RECORD_METADATA_SUFFIX = `.${STORE_METADATA_FILE_NAME}`;
+
+export interface KvsRecordMetadata {
+	key: string;
+	contentType: string;
+	/** Name of the value file on disk, when it is not the encoded key. */
+	filename?: string;
+}
+
+/**
+ * Percent-encode a record key into its on-disk name, the way crawlee's storage clients do
+ * (Python's `quote(key, safe='')`, which leaves `-._~` alone).
+ */
+export function encodeRecordKey(key: string) {
+	return encodeURIComponent(key).replaceAll(
+		/[!'()*]/g,
+		(character) => `%${character.charCodeAt(0).toString(16).toUpperCase()}`,
+	);
+}
+
+export const recordMetadataFileName = (key: string) => `${encodeRecordKey(key)}${RECORD_METADATA_SUFFIX}`;
+
+/**
+ * Write a record's value file plus the sidecar binding `key` to it.
+ *
+ * The binding is what lets a reader open `INPUT.json` under the bare `INPUT` key instead of
+ * probing extensions. `size` is deliberately omitted: readers fall back to the value file's
+ * length, so an Actor overwriting the value cannot leave a stale size behind.
+ */
+export async function writeKvsRecord({
+	storePath,
+	key,
+	fileName,
+	contentType,
+	body,
+}: {
+	storePath: string;
+	key: string;
+	fileName: string;
+	contentType: string;
+	body: string | Buffer;
+}) {
+	const metadata: KvsRecordMetadata = { key, contentType };
+
+	if (fileName !== encodeRecordKey(key)) {
+		metadata.filename = fileName;
+	}
+
+	await Promise.all([
+		writeFile(join(storePath, fileName), body),
+		writeFile(join(storePath, recordMetadataFileName(key)), JSON.stringify(metadata, null, 2)),
+	]);
+}
+
+/** Read a record's metadata sidecar, or undefined when there is no usable one. */
+export function readKvsRecordMetadata(storePath: string, key: string): KvsRecordMetadata | undefined {
+	let metadata: KvsRecordMetadata;
+
+	try {
+		metadata = JSON.parse(readFileSync(join(storePath, recordMetadataFileName(key)), 'utf8'));
+	} catch {
+		return undefined;
+	}
+
+	if (typeof metadata?.contentType !== 'string') {
+		return undefined;
+	}
+
+	const { filename } = metadata;
+
+	// A sidecar may only bind its key to a plain file in the store itself.
+	const bindsToStoreFile =
+		filename === undefined ||
+		(typeof filename === 'string' &&
+			!['', '.', '..'].includes(filename) &&
+			basename(filename) === filename &&
+			filename !== STORE_METADATA_FILE_NAME &&
+			!filename.endsWith(RECORD_METADATA_SUFFIX));
+
+	return bindsToStoreFile ? metadata : undefined;
+}
+
+/** Delete a record's value file and its metadata sidecar. Missing files are not an error. */
+export async function deleteKvsRecord(valueFilePath: string, key: string) {
+	await Promise.all([
+		rm(valueFilePath, { force: true }),
+		rm(join(dirname(valueFilePath), recordMetadataFileName(key)), { force: true }),
+	]);
+}

src/lib/utils.ts

@@ -45,6 +45,7 @@ import { ensureMigrated, getBackend, getProxyPassword, getToken, setProxyPasswor
 import { deleteFile, ensureApifyDirectory, ensureFolderExistsSync, rimrafPromised } from './files.js';
 import { useCLIMetadata } from './hooks/useCLIMetadata.js';
 import { inputFileRegExp, TEMP_INPUT_KEY_PREFIX } from './input-key.js';
+import { encodeRecordKey, readKvsRecordMetadata, recordMetadataFileName } from './kvs-metadata.js';
 import type { AuthJSON } from './types.js';
 import { cliDebugPrint } from './utils/cliDebugPrint.js';
 
@@ -523,18 +524,36 @@ export const createActZip = async (zipName: string, pathsToZip: string[], cwd: s
 	await archive.finalize();
 };
 
+export interface LocalInput {
+	body: Buffer;
+	/** Null when the input file is a bare, sidecar-less file with no telling extension. */
+	contentType: string | null;
+	fileName: string;
+}
+
 /**
  * Get Actor input from local store
  */
-export const getLocalInput = (cwd: string, inputKey?: string) => {
-	const defaultLocalStorePath = getLocalKeyValueStorePath();
-
-	const storePath = resolve(cwd, defaultLocalStorePath);
+export const getLocalInput = (cwd: string, inputKey?: string): LocalInput | undefined => {
+	const key = inputKey ?? KEY_VALUE_STORE_KEYS.INPUT;
+	const storePath = resolve(cwd, getLocalKeyValueStorePath());
 
 	if (!existsSync(storePath)) return;
 
+	// A tracked record: the sidecar names the file the value lives in, so a key like `INPUT`
+	// resolves to `INPUT.json` without guessing extensions.
+	const metadata = readKvsRecordMetadata(storePath, key);
+
+	if (metadata) {
+		const fileName = metadata.filename ?? encodeRecordKey(key);
+
+		if (existsSync(join(storePath, fileName))) {
+			return { body: readFileSync(join(storePath, fileName)), contentType: metadata.contentType, fileName };
+		}
+	}
+
 	const files = readdirSync(storePath);
-	const inputName = files.find((file) => !!file.match(inputFileRegExp(inputKey ?? 'INPUT')));
+	const inputName = files.find((file) => !!file.match(inputFileRegExp(key)));
 
 	// No input file
 	if (!inputName) return;
@@ -553,20 +572,29 @@ export const purgeDefaultDataset = async () => {
 };
 
 /**
- * Deletes every record from the default key-value store, except the files
- * matching the given input keys. Defaults to preserving `INPUT.*`.
+ * Deletes every record from the default key-value store, except the ones
+ * belonging to the given input keys. Defaults to preserving `INPUT.*`.
  */
 export const purgeDefaultKeyValueStore = async (...inputKeys: string[]) => {
 	const defaultKeyValueStorePath = resolve(process.cwd(), getLocalKeyValueStorePath());
 	if (!existsSync(defaultKeyValueStorePath)) {
 		return;
 	}
 	const filesToDelete = readdirSync(defaultKeyValueStorePath);
-	const preserveRegExps = (inputKeys.length > 0 ? inputKeys : ['INPUT']).map(inputFileRegExp);
+	const keys = inputKeys.length > 0 ? inputKeys : [KEY_VALUE_STORE_KEYS.INPUT];
+	const preserveRegExps = keys.map(inputFileRegExp);
+	// The value file may be bound to a name the regexps don't cover, and the sidecar itself
+	// never matches them, yet dropping it would strip the input of its metadata.
+	const preserveNames = new Set(
+		keys.flatMap((key) => [
+			recordMetadataFileName(key),
+			readKvsRecordMetadata(defaultKeyValueStorePath, key)?.filename,
+		]),
+	);
 
 	const deletePromises: Promise<void>[] = [];
 	filesToDelete.forEach((file) => {
-		if (!preserveRegExps.some((re) => re.test(file))) {
+		if (!preserveNames.has(file) && !preserveRegExps.some((re) => re.test(file))) {
 			deletePromises.push(deleteFile(join(defaultKeyValueStorePath, file)));
 		}
 	});
@@ -645,8 +673,8 @@ export const getNpmCmd = (): string => {
 };
 
 /**
- * Returns true if the local storage holds nothing but the input file, either
- * the user's own `<inputKey>.*` or the temporary copy the CLI writes next to it.
+ * Returns true if the local storage holds nothing but the input record, either
+ * the user's own `<inputKey>` / `<inputKey>.*` or the temporary copy the CLI writes next to it.
  */
 export const checkIfStorageIsEmpty = async (inputKey?: string) => {
 	const key = inputKey || KEY_VALUE_STORE_KEYS.INPUT;
@@ -655,7 +683,14 @@ export const checkIfStorageIsEmpty = async (inputKey?: string) => {
 	const keyValueStoreDir = getLocalKeyValueStorePath().replaceAll('\\', '/');
 
 	const filesWithoutInput = await glob(
-		[`${storageDir}/**`, `!${keyValueStoreDir}/${key}.*`, `!${keyValueStoreDir}/${TEMP_INPUT_KEY_PREFIX}${key}.*`],
+		[
+			`${storageDir}/**`,
+			// `<key>.*` also covers the record's `<key>.__metadata__.json` sidecar.
+			...[key, `${TEMP_INPUT_KEY_PREFIX}${key}`].flatMap((inputName) => [
+				`!${keyValueStoreDir}/${inputName}`,
+				`!${keyValueStoreDir}/${inputName}.*`,
+			]),
+		],
 		{ cwd: process.cwd() },
 	);
 

test/local/commands/init.test.ts

@@ -44,6 +44,15 @@ describe('apify init', () => {
 		expect(
 			JSON.parse(readFileSync(joinPath(getLocalKeyValueStorePath(), `${KEY_VALUE_STORE_KEYS.INPUT}.json`), 'utf8')),
 		).toStrictEqual({});
+		expect(
+			JSON.parse(
+				readFileSync(joinPath(getLocalKeyValueStorePath(), `${KEY_VALUE_STORE_KEYS.INPUT}.__metadata__.json`), 'utf8'),
+			),
+		).toStrictEqual({
+			key: KEY_VALUE_STORE_KEYS.INPUT,
+			contentType: 'application/json; charset=utf-8',
+			filename: `${KEY_VALUE_STORE_KEYS.INPUT}.json`,
+		});
 	});
 
 	it('correctly creates structure with prefilled INPUT.json', async () => {