fixes

Author: Chetan11-dev

botasaurus-controls/package.json

@@ -25,6 +25,7 @@
     "build": "rimraf dist && tsc && rimraf dist/tsconfig.tsbuildinfo &&  python3 fix.py",
     "prepublishOnly": "npm run build",
     "build-mv-starter": "npm run build && mv -f ~/Documents/grow/botasaurus/botasaurus-controls/dist/* ~/Documents/botasaurus-desktop-starter/node_modules/botasaurus-controls/dist/",
+    "build-mv-maps": "npm run build && mv -f ~/Documents/grow/botasaurus/botasaurus-controls/dist/* ~/Documents/google-maps-extractor-desktop/node_modules/botasaurus-controls/dist/",
     "lint": "eslint src test",
     "clean-install": "rm -rf dist/ node_modules/ package-lock.json yarn.lock .next/* && npm install",
     "upload": "python3 increment_version.py && rm -rf dist/ && npm publish",

js/botasaurus-js/src/playwright.ts

@@ -31,13 +31,14 @@ import {
 import { FormatType } from "./formats";
 import { createPlaywrightChrome, PlaywrightChrome } from "./page";
 import { getBotasaurusStorage } from "./botasaurus-storage"
-import { determineMaxLimit, drainQueue, getItemRepr, removeItemFromSeenItemsSet } from "./task"
+import { determineMaxLimit, drainQueue, getItemRepr, removeItemFromSeenItemsSet, TaskRunOptions } from "./task"
 
-type PlaywrightRunOptions<I = any> = {
+
+
+type PlaywrightRunOptions<I = any> = TaskRunOptions<I> & {
     page: Page;
     context: BrowserContext;
-    data: I;
-    metadata: any;
+    
 };
 
 type PlaywrightOptions<I> = {
@@ -146,6 +147,10 @@ function createPlaywright<I>(
             : performPlaywright.__name__;
         // @ts-ignore
         const returnDontCacheAsIs = combined.returnDontCacheAsIs;
+        // @ts-ignore
+        const isAborted = combined.isAborted ?? (() => false);
+        // @ts-ignore 
+        const pushData = combined.pushData ?? (() => {});
         const fn_name = performPlaywright.__name__;
 
         if (cache) {
@@ -192,7 +197,7 @@ function createPlaywright<I>(
                     // ...
                 }
 
-                result = await run({ data, metadata, ...driver });
+                result = await run({ data, metadata, isAborted, pushData, ...driver });
                 if (result === undefined) {
                     result = null;
                 }

js/botasaurus-js/src/task.ts

@@ -17,9 +17,11 @@ import {
 import { flatten } from './list-utils'
 import { FormatType } from './formats'
 
-type TaskRunOptions<I=any> = {
+export type TaskRunOptions<I=any> = {
     data: I
     metadata: any
+    isAborted: () => boolean
+    pushData: (data: Record<string, any> | Record<string, any>[]) => Promise<void> 
 }
 
 type TaskOptions<I> = {
@@ -55,6 +57,10 @@ function createTask<I>(options: TaskOptions<I>, is_async_queue: boolean) {
         performTask.__name__ = isNotEmptyString(name) ? name!.trim() : performTask.__name__
         // @ts-ignore
         const returnDontCacheAsIs = combined.returnDontCacheAsIs
+        // @ts-ignore
+        const isAborted = combined.isAborted ?? (() => false)
+        // @ts-ignore 
+        const pushData = combined.pushData ?? (() => {})
         const fn_name = performTask.__name__
 
         if (cache) {
@@ -80,10 +86,9 @@ function createTask<I>(options: TaskOptions<I>, is_async_queue: boolean) {
                 path = _getCachePath(fn_name, data)
             }
 
-
             let result: any
             try {
-                result = await run({ data, metadata })
+                result = await run({ data, metadata, isAborted, pushData })
                 if (result === undefined) {
                     result = null
                 }

js/botasaurus-server-js/package-lock.json

@@ -220,6 +220,7 @@
   },
   "dependencies": {
     "@aws-sdk/client-s3": "^3.943.0",
+    "@aws-sdk/lib-storage": "^3.943.0",
     "@seald-io/nedb": "latest",
     "async-mutex": "^0.5.0",
     "botasaurus": "latest",

js/botasaurus-server-js/package.json

@@ -35,7 +35,7 @@ import { isDontCache } from "botasaurus/dontcache";
 import { JsonHTTPResponseWithMessage } from "./errors";
 import { cleanDataInPlace } from "botasaurus/output";
 import { db, removeDuplicatesByKey } from "./models";
-import { DEFAULT_TASK_TIMEOUT, MasterExecutor, TaskCompletionPayload, TaskFailurePayload } from "./master-executor";
+import { DEFAULT_TASK_TIMEOUT, MasterExecutor, TaskCompletionPayload, TaskFailurePayload, PushDataChunkPayload, PushDataCompletePayload } from "./master-executor";
 import { WorkerExecutor } from "./worker-executor";
 
 /**
@@ -403,6 +403,33 @@ function registerMasterRoutes(app: FastifyInstance) {
         const executor = getExecutor() as MasterExecutor;
         return executor.handleWorkerShutdown(taskIds);
     });
+
+    // Check abortion status for multiple tasks (used by workers)
+    app.post('/k8s/check-abortion-status', async (request: FastifyRequest<{ Body: { taskIds: number[] } }>, _) => {
+        const taskIds = request.body?.taskIds || [];
+        
+        const executor = getExecutor() as MasterExecutor;
+        const results = await executor.getTasksAbortionResults(taskIds);
+        return results;
+    });
+
+    // PushData chunk endpoint - receives data chunks from workers
+    app.post('/k8s/push-data-chunk', async (request: FastifyRequest<{ Body: PushDataChunkPayload }>, _) => {
+        const payload = request.body;
+        validateTaskIdInPayload(payload.taskId);
+
+        const executor = getExecutor() as MasterExecutor;
+        return executor.handlePushDataChunk(payload);
+    });
+
+    // PushData complete endpoint - finalizes task after all chunks sent
+    app.post('/k8s/push-data-complete', async (request: FastifyRequest<{ Body: PushDataCompletePayload }>, _) => {
+        const payload = request.body;
+        validateTaskIdInPayload(payload.taskId);
+
+        const executor = getExecutor() as MasterExecutor;
+        return executor.handlePushDataComplete(payload);
+    });
 }
 
 /**

js/botasaurus-server-js/src/api-config.ts

@@ -2,6 +2,7 @@ import { Mutex } from 'async-mutex';
 import { getPendingTasks, TaskExecutor, TaskPriority } from './task-executor';
 import { db, Task, TaskStatus } from './models';
 import { Server } from './server';
+import { TaskResults } from './task-results';
 export const DEFAULT_TASK_TIMEOUT = 8 * 60 * 60
 /**
  * Payload for task completion from worker
@@ -34,6 +35,31 @@ export interface TaskFailurePayload {
     } | null;
 }
 
+/**
+ * Payload for pushData chunk from worker
+ */
+export interface PushDataChunkPayload {
+    taskId: number;
+    chunk: any[];
+}
+
+/**
+ * Payload for pushData completion from worker
+ */
+export interface PushDataCompletePayload {
+    taskId: number;
+    itemCount: number;
+    isDontCache: boolean;
+    scraperName: string;
+    taskData: any;
+    parentTaskId?: number | null;
+    capacity?: {
+        scraperType?: string;
+        scraperName?: string;
+        maxTasks: number;
+    } | null;
+}
+
 
 /**
  * Stale task recovery interval: 60 seconds
@@ -233,4 +259,41 @@ export class MasterExecutor extends TaskExecutor {
         console.log(`[Master] Released ${releasedCount}/${inProgressTaskIds.length} tasks from shutting down worker`);
         return { releasedCount };
     }
+
+    /**
+     * Handle pushData chunk from worker.
+     * Appends chunk to the task's result file.
+     */
+    async handlePushDataChunk(payload: PushDataChunkPayload) {
+        const { taskId, chunk } = payload;
+
+        try {
+            if (chunk && chunk.length > 0) {
+                await TaskResults.appendAllTask(taskId, chunk);
+            }
+            return {};
+        } catch (error) {
+            console.error('[Master] Error handling pushData chunk:', error);
+            throw error;
+        }
+    }
+
+    /**
+     * Handle pushData completion from worker.
+     * Finalizes the task (caching, status update, parent update).
+     * Piggyback pattern: also return new tasks for worker's available capacity.
+     */
+    async handlePushDataComplete(payload: PushDataCompletePayload): Promise<{ nextTasks: any[] }> {
+        const { taskId, itemCount, isDontCache, scraperName, taskData, parentTaskId, capacity } = payload;
+        const taskFilePath = TaskResults.generateTaskFilePath(taskId);
+
+        try {
+            await this.reportTaskSuccessWithPushData(taskId, taskFilePath, itemCount, isDontCache, scraperName, taskData, parentTaskId as any, null as any)
+        } catch (error) {
+            console.error('[Master] Error handling pushData complete:', error);
+        }
+
+        // Piggyback: Acquire next tasks if capacity is provided
+        return this.acquireNextTasks(capacity);
+    }
 }

js/botasaurus-server-js/src/master-executor.ts

@@ -186,7 +186,7 @@ export async function appendNdJson(data: any[], taskPath: string) {
   return taskPath
 }
 
-export async function readNdJsonCallback(taskPath: string, onData: (item: any, index: number) => undefined | false | Promise<undefined | false>, limit: number | null | undefined = null): Promise<{processedItems: number, hasExited: boolean}> {
+export async function readNdJsonCallback(taskPath: string, onData: (item: any, index: number) => void | false | Promise<void | false>, limit: number | null | undefined = null): Promise<{processedItems: number, hasExited: boolean}> {
   const fileStream = fs.createReadStream(taskPath, { encoding: 'utf-8' });
   let lineNumber = 0;
   let processedItems = 0;
@@ -204,13 +204,22 @@ export async function readNdJsonCallback(taskPath: string, onData: (item: any, i
       if (trimmedLine !== '') {
         try {
           const item = JSON.parse(trimmedLine);
-          const result = await onData(item, processedItems);
+          let result;
+          try {
+            result = await onData(item, processedItems);
+          } catch (error: any) {
+            error.isOnDataError = true;
+            throw error;
+          }
           processedItems++;
           if (result === false) {
             hasExited = true
             break;
           }
-        } catch (error) {
+        } catch (error: any) {
+          if (error.isOnDataError) {
+            throw error;
+          }
           // Handle potential malformed JSON
           const splitLines = trimmedLine.split('}{');
 
@@ -241,7 +250,7 @@ export async function readNdJsonCallback(taskPath: string, onData: (item: any, i
         }
 
         // @ts-ignore
-         isLimitReached = isNotNullish(limit) && processedItems >= limit
+        isLimitReached = isNotNullish(limit) && processedItems >= limit
 
         if (isLimitReached) {
           break;
@@ -279,7 +288,7 @@ function fixNdjsonFilename(filename: string): string {
  */
 export async function readNdjson<T = any>(
   filename: string,
-  onData: (item: T, index: number) => undefined | false,
+  onData: (item: T, index: number) => void | false,
   limit?: number | null
 ): Promise<number>{
   filename = fixNdjsonFilename(filename)