feat: implement crawlee v4 RequestQueueClient over apify-client

package.json

@@ -77,9 +77,9 @@
         "@apify/log": "^2.4.3",
         "@apify/timeout": "^0.3.0",
         "@apify/utilities": "^2.13.0",
-        "@crawlee/core": "^4.0.0-beta.61",
-        "@crawlee/types": "^4.0.0-beta.61",
-        "@crawlee/utils": "^4.0.0-beta.61",
+        "@crawlee/core": "^4.0.0-beta.71",
+        "@crawlee/types": "^4.0.0-beta.71",
+        "@crawlee/utils": "^4.0.0-beta.71",
         "apify-client": "^2.23.4",
         "semver": "^7.5.4",
         "tslib": "^2.6.2",
@@ -90,15 +90,14 @@
         "@apify/oxlint-config": "^0.2.5",
         "@apify/tsconfig": "^0.1.2",
         "@commitlint/config-conventional": "^21.0.0",
-        "@crawlee/memory-storage": "^4.0.0-beta.61",
         "@playwright/browser-chromium": "^1.60.0",
         "@types/content-type": "^1.1.8",
         "@types/node": "^24.0.0",
         "@types/semver": "^7.5.8",
         "@types/tough-cookie": "^4.0.5",
         "@types/ws": "^8.5.12",
         "commitlint": "^21.0.0",
-        "crawlee": "^4.0.0-beta.61",
+        "crawlee": "^4.0.0-beta.71",
         "globby": "^16.0.0",
         "husky": "^9.1.7",
         "lint-staged": "^17.0.0",

src/actor.ts

@@ -9,7 +9,7 @@ import type {
     UseStateOptions,
 } from '@crawlee/core';
 import { Dataset, EventType, purgeDefaultStorages, RequestQueue, serviceLocator } from '@crawlee/core';
-import type { Awaitable, Constructor, Dictionary, SetStatusMessageOptions, StorageClient } from '@crawlee/types';
+import type { Awaitable, Constructor, Dictionary, StorageBackend } from '@crawlee/types';
 import { sleep, snakeCaseToCamelCase } from '@crawlee/utils';
 import type {
     ActorCallOptions,
@@ -54,7 +54,7 @@ import { openStorage } from './storage.js';
 import { checkCrawleeVersion, getSystemInfo, isNonEmptyObject, printOutdatedSdkWarning, validate } from './utils.js';
 
 export interface InitOptions {
-    storage?: StorageClient;
+    storage?: StorageBackend;
     /**
      * Whether to automatically handle platform shutdown signals.
      * When enabled, `Actor.exit()` is called on `aborting` events and `Actor.reboot()` on `migrating` events.
@@ -100,6 +100,13 @@ export interface ExitOptions {
 
 export interface MainOptions extends ExitOptions, InitOptions {}
 
+export interface SetStatusMessageOptions {
+    /** If `true`, the status message is treated as final and won't be overwritten by the platform. */
+    isStatusMessageTerminal?: boolean;
+    /** Log level used when the status message is also logged locally. Defaults to `INFO`. */
+    level?: 'DEBUG' | 'INFO' | 'WARNING' | 'ERROR';
+}
+
 /**
  * Parsed representation of the Apify environment variables.
  * This object is returned by the {@apilink Actor.getEnv} function.
@@ -587,12 +594,12 @@ export class Actor<Data extends Dictionary = Dictionary> {
         serviceLocator.setConfiguration(this.config);
 
         if (this.isAtHome()) {
-            serviceLocator.setStorageClient(
+            serviceLocator.setStorageBackend(
                 new ApifyStorageClient(this.apifyClient, this.config, () => this.chargingManager),
             );
             serviceLocator.setEventManager(this.eventManager);
         } else if (options.storage) {
-            serviceLocator.setStorageClient(options.storage);
+            serviceLocator.setStorageBackend(options.storage);
         }
 
         // Init the event manager the config uses
@@ -656,7 +663,7 @@ export class Actor<Data extends Dictionary = Dictionary> {
 
         this._ensureActorInit('exit');
 
-        const client = serviceLocator.getStorageClient();
+        const client = serviceLocator.getStorageBackend();
         const events = serviceLocator.getEventManager();
 
         // Remove graceful shutdown handlers to prevent them from interfering with exit
@@ -1031,27 +1038,14 @@ export class Actor<Data extends Dictionary = Dictionary> {
                 break;
         }
 
-        const client = serviceLocator.getStorageClient();
-
-        // just to be sure, this should be fast
-        await addTimeoutToPromise(
-            async () =>
-                client.setStatusMessage!(statusMessage, {
-                    isStatusMessageTerminal,
-                    level,
-                }),
-            1000,
-            'Setting status message timed out after 1s',
-        ).catch((e) => log.warning(e.message));
-
         const runId = this.config.actorRunId!;
 
         if (runId) {
             // just to be sure, this should be fast
             const run = await addTimeoutToPromise(
-                async () => this.apifyClient.run(runId).get(),
+                async () => this.apifyClient.run(runId).update({ statusMessage, isStatusMessageTerminal }),
                 1000,
-                'Getting the current run timed out after 1s',
+                'Setting status message timed out after 1s',
             ).catch((e) => log.warning(e.message));
 
             if (run) {
@@ -1332,9 +1326,6 @@ export class Actor<Data extends Dictionary = Dictionary> {
 
         const queue = await this._openStorage(RequestQueue, queueIdOrName, options);
 
-        // eslint-disable-next-line dot-notation
-        queue['initialCount'] = (await queue.client.getMetadata())?.totalRequestCount ?? 0;
-
         return queue;
     }
 
@@ -2209,7 +2200,7 @@ export class Actor<Data extends Dictionary = Dictionary> {
     }
 
     private usesPushDataInterception(dataset: Dataset): boolean {
-        return Boolean((dataset.client as any)[USES_PUSH_DATA_INTERCEPTION]);
+        return Boolean((dataset.backend as any)[USES_PUSH_DATA_INTERCEPTION]);
     }
 
     private async pushDataViaInterceptedClient(

src/apify_request_queue_client.ts

@@ -0,0 +1,190 @@
+import { createHash } from 'node:crypto';
+
+import type {
+    BatchAddRequestsResult,
+    QueueOperationInfo,
+    RequestQueueBackend,
+    RequestQueueInfo,
+    RequestQueueOperationOptions,
+    RequestSchema,
+    UpdateRequestSchema,
+} from '@crawlee/types';
+import type { RequestQueueClient as ApifyRequestQueueResourceClient } from 'apify-client';
+
+/** Apify request IDs are the first 15 chars of a base64url-ish SHA-256 of the unique key. */
+const REQUEST_ID_LENGTH = 15;
+
+/** Default lock duration for requests fetched via {@link ApifyRequestQueueClient.fetchNextRequest}. */
+const DEFAULT_REQUEST_LOCK_SECS = 3 * 60;
+
+/** How many head requests to lock per `listAndLockHead` round-trip. */
+const HEAD_LOCK_BATCH_SIZE = 25;
+
+/**
+ * Derives a request id from its unique key, exactly as the Apify platform does
+ * (`sha256(uniqueKey)` → base64 → strip `+`/`/`/`=` → first 15 chars). Lets us
+ * address a request by unique key without an extra round-trip.
+ */
+function uniqueKeyToRequestId(uniqueKey: string): string {
+    const hash = createHash('sha256').update(uniqueKey).digest('base64').replace(/[+/=]/g, '');
+    return hash.slice(0, REQUEST_ID_LENGTH);
+}
+
+/**
+ * Implements Crawlee v4's stateful, pull-based {@link RequestQueueBackend} interface
+ * on top of `apify-client`'s REST request-queue API.
+ *
+ * `fetchNextRequest` locks requests server-side (`listAndLockHead`) so the queue can
+ * be safely shared across multiple consumers; `markRequestAsHandled` / `reclaimRequest`
+ * update the request and release its lock. Modeled on the Apify Python SDK's
+ * request-queue client.
+ *
+ * @internal
+ */
+export class ApifyRequestQueueClient implements RequestQueueBackend {
+    /** Locked request ids waiting to be handed out by {@link fetchNextRequest}, in queue order. */
+    private readonly queueHead: string[] = [];
+
+    /** Lock duration applied to fetched requests; adjustable via {@link setExpectedRequestProcessingTimeSecs}. */
+    private lockSecs = DEFAULT_REQUEST_LOCK_SECS;
+
+    constructor(private readonly client: ApifyRequestQueueResourceClient) {}
+
+    async getMetadata(): Promise<RequestQueueInfo> {
+        const metadata = await this.client.get();
+        if (!metadata) {
+            throw new Error('Request queue not found or has been deleted.');
+        }
+        return metadata as unknown as RequestQueueInfo;
+    }
+
+    async drop(): Promise<void> {
+        await this.client.delete();
+    }
+
+    async purge(): Promise<void> {
+        // The Apify platform has no "empty but keep" endpoint, and a run's queue is
+        // already fresh, so purge is a no-op here (use drop() to remove it entirely).
+    }
+
+    async setExpectedRequestProcessingTimeSecs(secs: number): Promise<void> {
+        this.lockSecs = secs;
+    }
+
+    async addBatchOfRequests(
+        requests: RequestSchema[],
+        options: RequestQueueOperationOptions = {},
+    ): Promise<BatchAddRequestsResult> {
+        const { forefront } = options;
+        // The API assigns ids itself; strip any incoming id so its strict input validation passes.
+        const apiRequests = requests.map((request) => {
+            const { id: _id, ...rest } = request;
+            return rest;
+        });
+
+        const result = await this.client.batchAddRequests(
+            apiRequests as Parameters<ApifyRequestQueueResourceClient['batchAddRequests']>[0],
+            { forefront },
+        );
+
+        // A forefront insert changes the head order; drop our cached head so the next
+        // fetch re-reads (and locks) the new front of the queue.
+        if (forefront) {
+            this.queueHead.length = 0;
+        }
+        return result as unknown as BatchAddRequestsResult;
+    }
+
+    async getRequest(uniqueKey: string): Promise<UpdateRequestSchema | undefined> {
+        const request = await this.client.getRequest(uniqueKeyToRequestId(uniqueKey));
+        return (request as UpdateRequestSchema | undefined) ?? undefined;
+    }
+
+    async fetchNextRequest(): Promise<UpdateRequestSchema | undefined> {
+        await this.ensureHeadIsNonEmpty();
+
+        while (this.queueHead.length > 0) {
+            const requestId = this.queueHead.shift()!;
+            // Head items lack userData/payload, so hydrate the full request.
+            const request = await this.client.getRequest(requestId);
+            // Skip requests that vanished or were already handled (e.g. by another consumer).
+            if (!request || (request as { handledAt?: string }).handledAt) {
+                continue;
+            }
+            return request as UpdateRequestSchema;
+        }
+        return undefined;
+    }
+
+    async markRequestAsHandled(request: UpdateRequestSchema): Promise<QueueOperationInfo | undefined> {
+        const result = await this.client.updateRequest({
+            ...this.toApiRequest(request),
+            handledAt: request.handledAt ?? new Date().toISOString(),
+        });
+        // Marking handled takes the request out of the head; its lock becomes moot.
+        return this.toQueueOperationInfo(result);
+    }
+
+    async reclaimRequest(
+        request: UpdateRequestSchema,
+        options: RequestQueueOperationOptions = {},
+    ): Promise<QueueOperationInfo | undefined> {
+        const { forefront } = options;
+        const result = await this.client.updateRequest(this.toApiRequest(request), { forefront });
+
+        // Release the server-side lock so the request is fetchable again immediately,
+        // rather than only once the lock expires.
+        await this.client.deleteRequestLock(request.id, { forefront }).catch(() => {});
+        if (forefront) {
+            this.queueHead.length = 0;
+        }
+        return this.toQueueOperationInfo(result);
+    }
+
+    async isEmpty(): Promise<boolean> {
+        if (this.queueHead.length > 0) {
+            return false;
+        }
+        const head = await this.client.listHead({ limit: 1 });
+        return head.items.length === 0;
+    }
+
+    async isFinished(): Promise<boolean> {
+        if (this.queueHead.length > 0) {
+            return false;
+        }
+        // `pendingRequestCount` counts every not-yet-handled request, including ones
+        // currently locked/in-progress — so 0 means there is genuinely nothing left.
+        const metadata = await this.client.get();
+        return (metadata?.pendingRequestCount ?? 0) === 0;
+    }
+
+    private async ensureHeadIsNonEmpty(): Promise<void> {
+        if (this.queueHead.length > 0) {
+            return;
+        }
+        const head = await this.client.listAndLockHead({ limit: HEAD_LOCK_BATCH_SIZE, lockSecs: this.lockSecs });
+        for (const item of head.items) {
+            this.queueHead.push(item.id);
+        }
+    }
+
+    /** Crawlee's `UpdateRequestSchema` is structurally the apify-client request shape. */
+    private toApiRequest(
+        request: UpdateRequestSchema,
+    ): Parameters<ApifyRequestQueueResourceClient['updateRequest']>[0] {
+        return request as unknown as Parameters<ApifyRequestQueueResourceClient['updateRequest']>[0];
+    }
+
+    private toQueueOperationInfo(result: {
+        requestId: string;
+        wasAlreadyPresent: boolean;
+        wasAlreadyHandled: boolean;
+    }): QueueOperationInfo {
+        return {
+            requestId: result.requestId,
+            wasAlreadyPresent: result.wasAlreadyPresent,
+            wasAlreadyHandled: result.wasAlreadyHandled,
+        };
+    }
+}