refactor!: Rewrite the FilesystemStorageClient to use apify/crawlee-storage

docs/guides/parallel-scraping/parallel-scraper.mjs

@@ -1,5 +1,6 @@
 import { fork } from 'node:child_process';
 
+import { FileSystemStorageClient } from '@crawlee/fs-storage';
 import { Configuration, Dataset, PlaywrightCrawler, log } from 'crawlee';
 
 import { router } from './routes.mjs';
@@ -76,13 +77,18 @@ if (!process.env.IN_WORKER_THREAD) {
     // Get the request queue
     const requestQueue = await getOrInitQueue(false);
 
-    // Disable the automatic purge on start and configure crawlee to store the worker-specific data in a separate directory
-    // (needs to be done AFTER the queue is initialized when running locally)
+    // Disable the automatic purge on start, so we don't lose the queue we prepared
     const config = new Configuration({
         purgeOnStart: false,
-        storageClientOptions: {
-            localDataDirectory: `./storage/worker-${process.env.WORKER_INDEX}`,
-        },
+    });
+
+    // Store the worker's own internal state (its default dataset, key-value store, etc.) in a separate
+    // directory so the workers don't collide with each other. This directory is private to a single
+    // worker, so we set `requestQueueAccess: 'single'` — the concurrency-safe locking only matters for
+    // the shared `shop-urls` queue, which gets its own storage client in `requestQueue.mjs`.
+    const storageClient = new FileSystemStorageClient({
+        localDataDirectory: `./storage/worker-${process.env.WORKER_INDEX}`,
+        requestQueueAccess: 'single',
     });
 
     workerLogger.debug('Setting up crawler.');
@@ -98,6 +104,10 @@ if (!process.env.IN_WORKER_THREAD) {
             // highlight-end
             // Let's also limit the crawler's concurrency, we don't want to overload a single process 🐌
             maxConcurrency: 5,
+            // Use the worker-specific, concurrency-safe storage client we created above
+            // highlight-start
+            storageClient,
+            // highlight-end
         },
         config,
     );

docs/guides/parallel-scraping/parallel-scraping.mdx

@@ -60,6 +60,16 @@ The first step in our conversion process will be creating a common file (let's c
 
 The exported function, `getOrInitQueue`, might seem like it does a lot. In essence, it just ensures the request queue is initialized, and if requested, ensures it starts off with an empty state.
 
+:::caution Make the shared queue concurrency-safe with `requestQueueAccess: 'shared'`
+
+Because every worker process opens this same `shop-urls` queue at the same time, it **must** use the concurrency-safe locking behavior of `FileSystemStorageClient`. That's why `getOrInitQueue` opens the queue with a storage client constructed with `requestQueueAccess: 'shared'`.
+
+By default, `FileSystemStorageClient` assumes it is the *sole* consumer of a queue (`requestQueueAccess: 'single'`). On open it immediately reclaims any requests left *in progress* — great for a single-process crawl recovering after a crash, but disastrous when workers run side by side: each worker would happily grab requests another worker is still processing, so the same URL gets scraped multiple times.
+
+Setting `requestQueueAccess: 'shared'` tells the client to treat an in-progress request as a potential live peer's lock and only reclaim it once the lock expires on the wall clock, so two workers never process the same request at once.
+
+:::
+
 ### Adapting our previous scraper to enqueue the product URLs to the new queue
 
 In the `src/routes.mjs` file of the scraper we previously built, we have a handler for the `CATEGORY` label. Let's adapt that handler to enqueue the product URLs to the new queue we created.
@@ -122,34 +132,44 @@ This will check how the script is executed as. If this value has _any_ value, it
 
 We use this to ensure the parent process stays alive until all the worker processes exit. Otherwise, the worker processes would just get spawned, and lose the ability to communicate with the parent. You might not need this depending on your use case (maybe you just need to spawn workers and let them process).
 
-#### What's with all those `Configuration` calls?
+#### What's with all the `Configuration` and storage client setup?
 
-There are three steps we want to do for the worker processes:
+There are two things we want to do for the worker processes:
 
-- get the queue that supports locking from the same location as the parent process
-- ensure the default storages do **not** get purged on start, as otherwise we'd lose the queue we prepared, and initialize a special storage for worker processes so they do not collide with each other
+- get the shared queue from the same location as the parent process (it already comes with the concurrency-safe storage client we set up in `requestQueue.mjs`)
+- ensure the default storages do **not** get purged on start, as otherwise we'd lose the queue we prepared, and give each worker its own private storage directory for its internal state so the workers don't collide with each other
 
 In order, that's what these lines do:
 
 ```javascript title="src/parallel-scraper.mjs"
-// Get the request queue from the parent process (step 1)
+import { FileSystemStorageClient } from '@crawlee/fs-storage';
+
+// Get the shared request queue from the parent process (step 1)
 const requestQueue = await getOrInitQueue(false);
 
-// Disable the automatic purge on start and configure crawlee to store the worker-specific data
-// in a separate directory (needs to be done AFTER the queue is initialized when running locally) (step 2)
-const config = new Configuration({
-    purgeOnStart: false,
-    storageClientOptions: {
-        localDataDirectory: `./storage/worker-${process.env.WORKER_INDEX}`,
-    },
+// Disable the automatic purge on start, so we don't lose the queue we prepared (step 2)
+const config = new Configuration({ purgeOnStart: false });
+
+// Store the worker's own internal state in a separate directory so workers don't collide (step 2,
+// cont.). This directory is private to a single worker, so we explicitly set
+// `requestQueueAccess: 'single'`.
+const storageClient = new FileSystemStorageClient({
+    localDataDirectory: `./storage/worker-${process.env.WORKER_INDEX}`,
+    requestQueueAccess: 'single',
 });
 ```
 
+:::note Why no `requestQueueAccess: 'shared'` here?
+
+Each worker's `./storage/worker-N` directory is private to that single worker — nothing else opens it — so the default `requestQueueAccess: 'single'` is exactly right. The concurrency-safe locking only matters for storage that is genuinely shared across processes, which is the `shop-urls` queue in `requestQueue.mjs`, not this per-worker internal state.
+
+:::
+
 #### Telling the crawler to use the worker configuration
 
 You might have noticed several lines highlighted in the code above. Those show how you provide the shared request queue to the crawler.
 
-You might have also noticed we passed in a second parameter to the constructor of the crawler, the `config` variable we created earlier. This is needed to ensure the crawler uses the worker-specific storages for internal states, and that they do not collide with each other.
+You might have also noticed we passed in the `config` and `storageClient` we created earlier to the crawler. These ensure the crawler uses the worker-specific storages for its own internal state (so the workers do not collide with each other), while still consuming the shared, concurrency-safe `shop-urls` queue we provided explicitly.
 
 #### Why do we use `process.send` instead of `context.pushData`?
 

docs/guides/parallel-scraping/shared.mjs

@@ -1,8 +1,19 @@
+import { FileSystemStorageClient } from '@crawlee/fs-storage';
 import { RequestQueue } from 'crawlee';
 
 // The request queue shared by all the parallel workers
 let queue;
 
+// The `shop-urls` queue is opened concurrently by every worker process, so it must use the
+// concurrency-safe locking behavior. With `requestQueueAccess: 'shared'`, a request another worker
+// is still processing is treated as a live peer's lock and is not handed out again until that lock
+// expires — so two workers never scrape the same URL at once. (We point at the `./storage`
+// location, which is where this shared queue lives.)
+const sharedStorageClient = new FileSystemStorageClient({
+    localDataDirectory: './storage',
+    requestQueueAccess: 'shared',
+});
+
 /**
  * @param {boolean} makeFresh Whether the queue should be cleared before returning it
  * @returns The queue
@@ -12,11 +23,11 @@ export async function getOrInitQueue(makeFresh = false) {
         return queue;
     }
 
-    queue = await RequestQueue.open('shop-urls');
+    queue = await RequestQueue.open('shop-urls', { storageClient: sharedStorageClient });
 
     if (makeFresh) {
         await queue.drop();
-        queue = await RequestQueue.open('shop-urls');
+        queue = await RequestQueue.open('shop-urls', { storageClient: sharedStorageClient });
     }
 
     return queue;

docs/public-api/crawlee-core.api.md

@@ -294,6 +294,7 @@ export const crawleeConfigFields: {
     disableBrowserSandbox: ConfigField<z.ZodDefault<z.ZodPipe<z.ZodTransform<unknown, unknown>, z.ZodBoolean>>>;
     logLevel: ConfigField<z.ZodOptional<z.ZodPipe<z.ZodTransform<{} | null | undefined, unknown>, z.ZodEnum<typeof LogLevel>>>>;
     persistStorage: ConfigField<z.ZodDefault<z.ZodPipe<z.ZodTransform<unknown, unknown>, z.ZodBoolean>>>;
+    storageDir: ConfigField<z.ZodDefault<z.ZodString>>;
     containerized: ConfigField<z.ZodOptional<z.ZodPipe<z.ZodTransform<unknown, unknown>, z.ZodBoolean>>>;
 };
 

docs/public-api/crawlee-fs-storage.api.md

@@ -6,11 +6,14 @@
 
 import type { CrawleeLogger } from '@crawlee/types';
 import type { Dictionary } from '@crawlee/types';
+import type { FileSystemDatasetClient } from '@crawlee/fs-storage-native';
+import type { FileSystemKeyValueStoreClient } from '@crawlee/fs-storage-native';
+import type { FileSystemRequestQueueClient } from '@crawlee/fs-storage-native';
 import type * as storage from '@crawlee/types';
 
-// @public (undocumented)
+// @public
 export class FileSystemStorageClient implements storage.StorageClient {
-    constructor(options?: FileSystemStorageOptions);
+    constructor(options: FileSystemStorageOptions);
     // (undocumented)
     createDatasetClient(options?: storage.CreateDatasetClientOptions): Promise<storage.DatasetClient>;
     // (undocumented)
@@ -32,6 +35,8 @@ export class FileSystemStorageClient implements storage.StorageClient {
     readonly logger?: CrawleeLogger;
     purge(): Promise<void>;
     // (undocumented)
+    readonly requestQueueAccess: 'single' | 'shared';
+    // (undocumented)
     readonly requestQueueCache: RequestQueueClient[];
     // (undocumented)
     readonly requestQueuesDirectory: string;
@@ -40,15 +45,13 @@ export class FileSystemStorageClient implements storage.StorageClient {
     // (undocumented)
     storageExists(id: string, type: 'Dataset' | 'KeyValueStore' | 'RequestQueue'): Promise<boolean>;
     teardown(): Promise<void>;
-    // (undocumented)
-    readonly writeMetadata: boolean;
 }
 
 // @public (undocumented)
 export interface FileSystemStorageOptions {
-    localDataDirectory?: string;
+    localDataDirectory: string;
     logger?: CrawleeLogger;
-    writeMetadata?: boolean;
+    requestQueueAccess?: 'single' | 'shared';
 }
 
 // (No @packageDocumentation comment for this package)

docs/upgrading/upgrading_v4.md

@@ -687,18 +687,41 @@ import { FileSystemStorageClient } from '@crawlee/fs-storage';
 import { MemoryStorageClient } from '@crawlee/core';
 
 // Persists to disk (the old default behavior):
-const storageClient = new FileSystemStorageClient();
+const storageClient = new FileSystemStorageClient({ localDataDirectory: './storage' });
 
 // Or keep everything in memory only (the old `persistStorage: false`):
 const inMemory = new MemoryStorageClient();
 ```
 
-The `localDataDirectory`, `persistStorage`, and `writeMetadata` options are still accepted by `MemoryStorageClient` for source compatibility, but they are ignored — in-memory storage has nowhere to write. `FileSystemStorageClient` honors `localDataDirectory` and `writeMetadata`; it always persists, so it has no `persistStorage` option.
+`MemoryStorageClient` no longer takes the `localDataDirectory`, `persistStorage`, or `writeMetadata` options — in-memory storage has nowhere to write, so they had no meaning. `FileSystemStorageClient` honors `localDataDirectory`; it always persists, so it has no `persistStorage` option, and the `writeMetadata` option has been removed there too (see [`writeMetadata` option removed](#writemetadata-option-removed)).
 
 ### No request lock expiry in `MemoryStorageClient`
 
 Because the in-memory queue lives entirely within a single process and is never shared with another consumer, `MemoryStorageClient`'s request queue no longer uses an expiring, cross-process lock. A fetched request simply stays *in progress* until it is handled or reclaimed; it never becomes fetchable again on its own after a timeout. `setExpectedRequestProcessingTimeSecs()` is therefore a no-op for in-memory storage. (Disk-backed `FileSystemStorageClient` keeps the lock-with-expiry behavior.)
 
+### `writeMetadata` option removed
+
+`FileSystemStorageClient` no longer accepts the `writeMetadata` option. The underlying file-system storage now always writes metadata files (`__metadata__.json` for each storage and a `<key>.__metadata__.json` sidecar for each key-value record), so the toggle no longer had any effect. Remove it from your storage client options:
+
+```diff
+ import { FileSystemStorageClient } from '@crawlee/fs-storage';
+
+ const storageClient = new FileSystemStorageClient({
+     localDataDirectory: './storage',
+-    writeMetadata: true,
+ });
+```
+
+`MemoryStorageClient` never accepted `writeMetadata` (it has no on-disk format to begin with), so there is nothing to change there.
+
+### Out-of-band key-value files (e.g. a hand-placed `INPUT.json`)
+
+`FileSystemStorageClient` only fully tracks records it wrote itself (those have a `<key>.__metadata__.json` sidecar). It still reads a value file placed in the store directory out-of-band — such as a hand-written or platform-provided `INPUT.json` — by probing the requested key plus the `.json` and `.txt` extensions. A few behaviors around these "bare" files changed in v4:
+
+- **Extensionless bare files report `application/octet-stream`.** In v3 a bare value file with no extension was read as `text/plain`. In v4 the client is a plain byte transport and only infers a content type from a real extension, so an extensionless file now comes back as `application/octet-stream`. Give the file a `.json` or `.txt` extension if you need a more specific type.
+- **Malformed bare files are no longer silently swallowed.** In v3 a bare `INPUT.json` containing invalid JSON was treated as a missing record (`getValue` returned `undefined`). In v4 the raw bytes are returned verbatim and parsing happens in the `KeyValueStore` frontend, so a malformed value now surfaces a parse error at read time instead of looking absent.
+- **Bare files are enumerated by `listKeys` under their actual on-disk name.** A bare `INPUT.json` (or `.txt`/`.bin`) shows up in `listKeys` as `INPUT.json` and reads back cleanly under that key via `getValue` / `recordExists` / `getPublicUrl`; the logical `INPUT` lookup keeps resolving the same file as well. An extensionless bare file is listed as `INPUT`. If both a tracked `INPUT` record and a bare `INPUT.json` exist, the tracked record wins and the bare variant is not listed. Everything `listKeys` needs is read from the filesystem index, so this no longer triggers the per-read O(n) directory scans the v3 fallback performed.
+
 ## Multiple crawler instances use separate default request queues
 
 In v3, every `BasicCrawler` (or subclass) that didn't receive an explicit `requestQueue` option would open the same default request queue. If you created two crawlers in the same process, they would silently share a queue — leading to request collisions and hard-to-debug deduplication issues.

packages/core/src/configuration.ts

@@ -82,6 +82,8 @@ export const crawleeConfigFields = {
     logLevel: field(logLevelSchema.optional(), 'CRAWLEE_LOG_LEVEL'),
     /** @default true */
     persistStorage: field(coerceBoolean.default(true), 'CRAWLEE_PERSIST_STORAGE'),
+    /** @default './storage' */
+    storageDir: field(z.string().default('./storage'), 'CRAWLEE_STORAGE_DIR'),
     containerized: field(coerceBoolean.optional(), 'CRAWLEE_CONTAINERIZED'),
 };
 
@@ -155,6 +157,7 @@ export interface Configuration extends ResolvedConfigValues {}
  * `persistStateIntervalMillis` | `CRAWLEE_PERSIST_STATE_INTERVAL_MILLIS` | `60_000`
  * `purgeOnStart` | `CRAWLEE_PURGE_ON_START` | `true`
  * `persistStorage` | `CRAWLEE_PERSIST_STORAGE` | `true`
+ * `storageDir` | `CRAWLEE_STORAGE_DIR` | `'./storage'`
  *
  * ## Advanced Configuration Options
  *

packages/core/src/service_locator.ts

@@ -220,6 +220,7 @@ export class ServiceLocator implements ServiceLocatorInterface {
             const config = this.getConfiguration();
             this.storageClient = config.persistStorage
                 ? new FileSystemStorageClient({
+                      localDataDirectory: config.storageDir,
                       logger: this.getLogger().child({ prefix: 'FileSystemStorageClient' }),
                   })
                 : new MemoryStorageClient({

packages/core/src/storages/key_value_store.ts

@@ -1,8 +1,4 @@
-import { readFile } from 'node:fs/promises';
-import { join } from 'node:path';
-
 import type { Dictionary, KeyValueStoreClient, KeyValueStoreItemData } from '@crawlee/types';
-import JSON5 from 'json5';
 import ow, { ArgumentError } from 'ow';
 
 import { KEY_VALUE_STORE_KEY_REGEX } from '@apify/consts';
@@ -855,8 +851,9 @@ export class KeyValueStore {
 
     /**
      * Gets the crawler input value from the default {@apilink KeyValueStore} associated with the current crawler run.
-     * By default, it will try to find root input files (either extension-less, `.json` or `.txt`),
-     * or alternatively read the input from the default {@apilink KeyValueStore}.
+     *
+     * The input is read from the default {@apilink KeyValueStore} under the configured input key
+     * (`CRAWLEE_INPUT_KEY`, default `INPUT`).
      *
      * Note that the `getInput()` function does not cache the value read from the key-value store.
      * If you need to use the input multiple times in your crawler,
@@ -874,32 +871,7 @@ export class KeyValueStore {
      */
     static async getInput<T = Dictionary | string | Buffer>(): Promise<T | null> {
         const store = await this.open();
-        const inputKey = store.config.inputKey;
-
-        const cwd = process.cwd();
-        const possibleExtensions = ['', '.json', '.txt'];
-
-        // Attempt to read input from root file instead of key-value store
-        for (const extension of possibleExtensions) {
-            const inputFile = join(cwd, `${inputKey}${extension}`);
-            let input: Buffer;
-
-            // Try getting the file from the file system
-            try {
-                input = await readFile(inputFile);
-            } catch {
-                continue;
-            }
-
-            // Attempt to parse as JSON, or return the input as is otherwise
-            try {
-                return JSON5.parse(input.toString()) as T;
-            } catch {
-                return input as unknown as T;
-            }
-        }
-
-        return store.getValue<T>(inputKey);
+        return store.getValue<T>(store.config.inputKey);
     }
 }
 

packages/fs-storage/package.json

@@ -42,15 +42,8 @@
         "access": "public"
     },
     "dependencies": {
+        "@crawlee/fs-storage-native": "0.1.5-beta.18",
         "@crawlee/types": "workspace:*",
-        "@crawlee/utils": "workspace:*",
-        "@sapphire/async-queue": "^1.5.5",
-        "@sapphire/shapeshift": "^4.0.0",
-        "content-type": "^1.0.5",
-        "fs-extra": "^11.3.0",
-        "json5": "^2.2.3",
-        "mime-types": "^3.0.1",
-        "proper-lockfile": "^4.1.2",
-        "tslib": "^2.8.1"
+        "@sapphire/shapeshift": "^4.0.0"
     }
 }