fix: throw an error instead of ignoring it when uploading batches

jest.config.cjs

@@ -13,6 +13,7 @@ module.exports = {
         'src/tests/dummy-connector/metadata-extraction.test.ts',
         'src/http/axios-client-internal.test.ts',
         'src/tests/event-data-size-limit/.*.test.ts',
+        'src/tests/upload-failure/upload-failure.slow.test.ts',
       ],
     },
     {
@@ -33,6 +34,7 @@ module.exports = {
         '<rootDir>/src/tests/dummy-connector/metadata-extraction.test.ts',
         '<rootDir>/src/http/axios-client-internal.test.ts',
         '<rootDir>/src/tests/event-data-size-limit/size-limit-1.test.ts',
+        '<rootDir>/src/tests/upload-failure/upload-failure.slow.test.ts',
       ],
     },
   ],

src/mock-server/mock-server.interfaces.ts

@@ -56,6 +56,15 @@ export interface RouteConfig {
   headers?: Record<string, string>;
   /** Optional retry configuration for simulating failures before success */
   retry?: RetryConfig;
+  /**
+   * Respond successfully for the first N requests, then return an error.
+   * Useful for partial-upload scenarios (first batch succeeds, second fails).
+   */
+  succeedThenFail?: {
+    successCount: number;
+    errorStatus?: number;
+    errorBody?: unknown;
+  };
   /** Optional delay in milliseconds before sending the response */
   delay?: number;
 }

src/mock-server/mock-server.ts

@@ -216,18 +216,51 @@ export class MockServer {
    * Configures a route to return a specific status code and optional response body.
    */
   public setRoute(config: RouteConfig): void {
-    const { path, method, status, body, bodyBuffer, retry, headers, delay } =
-      config;
+    const {
+      path,
+      method,
+      status,
+      body,
+      bodyBuffer,
+      retry,
+      succeedThenFail,
+      headers,
+      delay,
+    } = config;
     const key = this.getRouteKey(method, path);
 
-    if (retry) {
+    if (retry || succeedThenFail) {
       this.requestCounts.set(key, 0);
     }
 
     this.routeHandlers.set(key, (req: ParsedRequest, res: MockResponse) => {
       const sendResponse = (responseDelay?: number) => {
         const send = () => {
-          if (retry) {
+          if (succeedThenFail) {
+            const currentCount = this.requestCounts.get(key) || 0;
+            this.requestCounts.set(key, currentCount + 1);
+
+            if (currentCount < succeedThenFail.successCount) {
+              if (headers) {
+                res.set(headers);
+              }
+
+              if (bodyBuffer !== undefined) {
+                res.status(status).buffer(bodyBuffer);
+              } else if (body !== undefined) {
+                res.status(status).json(body);
+              } else {
+                this.defaultRouteHandler(req, res);
+              }
+            } else {
+              const errorStatus = succeedThenFail.errorStatus ?? 400;
+              if (succeedThenFail.errorBody !== undefined) {
+                res.status(errorStatus).json(succeedThenFail.errorBody);
+              } else {
+                res.status(errorStatus).send();
+              }
+            }
+          } else if (retry) {
             const currentCount = this.requestCounts.get(key) || 0;
             const failureCount = retry.failureCount ?? 4;
             const errorStatus = retry.errorStatus ?? 500;
@@ -303,6 +336,16 @@ export class MockServer {
     this.requests = [];
   }
 
+  /**
+   * Returns all POST requests to the platform callback URL.
+   */
+  public getCallbackRequests(): RequestInfo[] {
+    return this.requests.filter(
+      (req) =>
+        req.method.toUpperCase() === 'POST' && req.url.includes('callback_url')
+    );
+  }
+
   /**
    * Returns the most recent request or undefined if no requests exist.
    */

src/multithreading/process-task.test.ts

@@ -62,6 +62,7 @@ jest.mock('./worker-adapter/worker-adapter', () => ({
   WorkerAdapter: jest.fn().mockImplementation(() => ({
     isTimeout: false,
     hasWorkerEmitted: false,
+    emitError: jest.fn(),
   })),
 }));
 
@@ -150,7 +151,11 @@ describe(processTask.name, () => {
     // Arrange
     const event = makeEvent();
     setWorkerData({ event, initialState: {}, options: {} });
-    const mockAdapter = { isTimeout: false, hasWorkerEmitted: false };
+    const mockAdapter = {
+      isTimeout: false,
+      hasWorkerEmitted: false,
+      emitError: jest.fn(),
+    };
     (WorkerAdapter as jest.Mock).mockImplementation(() => mockAdapter);
     const task = jest.fn().mockResolvedValue(undefined);
     const onTimeout = jest.fn().mockResolvedValue(undefined);
@@ -200,4 +205,31 @@ describe(processTask.name, () => {
     expect(processExitSpy).toHaveBeenCalledWith(1);
     expect(onTimeout).not.toHaveBeenCalled();
   });
+
+  it('should emit a failure event through the adapter before falling back', async () => {
+    const event = makeEvent();
+    setWorkerData({ event, initialState: {}, options: {} });
+    const mockAdapter = {
+      isTimeout: false,
+      hasWorkerEmitted: false,
+      emitError: jest.fn().mockImplementation(() => {
+        mockAdapter.hasWorkerEmitted = true;
+      }),
+    };
+    (WorkerAdapter as jest.Mock).mockImplementation(() => mockAdapter);
+    const taskError = new Error('task upload failed');
+    const task = jest.fn().mockRejectedValue(taskError);
+    const onTimeout = jest.fn().mockResolvedValue(undefined);
+
+    processTask({ task, onTimeout });
+    await flush();
+
+    expect(mockAdapter.emitError).toHaveBeenCalledWith(taskError);
+    expect(mockParentPortPostMessage).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        subject: WorkerMessageSubject.WorkerMessageFailed,
+      })
+    );
+    expect(processExitSpy).toHaveBeenCalledWith(1);
+  });
 });

src/multithreading/process-task.ts

@@ -23,6 +23,8 @@ export function processTask<ConnectorState>({
 
   void (async () => {
     await runWithSdkLogContext(async () => {
+      let adapter: WorkerAdapter<ConnectorState> | undefined;
+
       try {
         const event = workerData.event;
 
@@ -44,27 +46,48 @@ export function processTask<ConnectorState>({
           options,
         });
 
-        const adapter = new WorkerAdapter<ConnectorState>({
+        const workerAdapter = new WorkerAdapter<ConnectorState>({
           event,
           adapterState,
           options,
         });
+        adapter = workerAdapter;
 
         parentPort?.on(WorkerEvent.WorkerMessage, (message) => {
           if (message.subject !== WorkerMessageSubject.WorkerMessageExit) {
             return;
           }
           console.log('Timeout received. Waiting for the task to finish.');
-          adapter.isTimeout = true;
+          workerAdapter.isTimeout = true;
         });
 
-        await runWithUserLogContext(async () => task({ adapter }));
-        if (adapter.isTimeout && !adapter.hasWorkerEmitted) {
-          await runWithUserLogContext(async () => onTimeout({ adapter }));
+        await runWithUserLogContext(async () =>
+          task({ adapter: workerAdapter })
+        );
+        if (workerAdapter.isTimeout && !workerAdapter.hasWorkerEmitted) {
+          await runWithUserLogContext(async () =>
+            onTimeout({ adapter: workerAdapter })
+          );
         }
         process.exit(0);
       } catch (error) {
-        runWithUserLogContext(() => {
+        await runWithUserLogContext(async () => {
+          if (adapter && !adapter.hasWorkerEmitted) {
+            try {
+              await adapter.emitError(error);
+            } catch (failureError) {
+              console.error(
+                'Error while emitting task failure.',
+                serializeError(failureError)
+              );
+            }
+          }
+
+          if (adapter?.hasWorkerEmitted) {
+            process.exit(1);
+            return;
+          }
+
           const errorMessage = `Error while processing task. ${serializeError(
             error
           )}`;

src/multithreading/worker-adapter/worker-adapter.emit.test.ts

@@ -146,12 +146,11 @@ describe(`${WorkerAdapter.name}.emit`, () => {
     expect(mockPostMessage).toHaveBeenCalledTimes(1);
   });
 
-  it('should correctly emit one event even if uploadAllRepos errors', async () => {
+  it('should log state size and current state before posting state', async () => {
     // Arrange
+    const logSpy = jest.spyOn(console, 'log').mockImplementation();
     adapter['adapterState'].postState = jest.fn().mockResolvedValue(undefined);
-    adapter.uploadAllRepos = jest
-      .fn()
-      .mockRejectedValue(new Error('uploadAllRepos error'));
+    adapter.uploadAllRepos = jest.fn().mockResolvedValue(undefined);
 
     // Act
     await adapter.emit(ExtractorEventType.MetadataExtractionError, {
@@ -160,7 +159,46 @@ describe(`${WorkerAdapter.name}.emit`, () => {
     });
 
     // Assert
-    expect(mockPostMessage).toHaveBeenCalledTimes(1);
+    const stateLogCall = logSpy.mock.calls.find(
+      ([message]) =>
+        typeof message === 'string' &&
+        message.includes('Saving ') &&
+        message.includes('Current state')
+    );
+    expect(stateLogCall).toBeDefined();
+    expect(stateLogCall?.[0]).toContain(
+      'KB state before emitting event with event type: METADATA_EXTRACTION_ERROR. Current state'
+    );
+    expect(stateLogCall?.[1]).toEqual(
+      expect.objectContaining({ attachments: { completed: false } })
+    );
+  });
+
+  it('should emit phase extraction error when uploadAllRepos fails', async () => {
+    const { emit: mockEmit } = require('../../common/control-protocol');
+    adapter['adapterState'].postState = jest.fn().mockResolvedValue(undefined);
+    adapter.uploadAllRepos = jest
+      .fn()
+      .mockRejectedValue(new Error('uploadAllRepos error'));
+
+    await adapter.emit(ExtractorEventType.DataExtractionDone);
+
+    expect(mockEmit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        eventType: ExtractorEventType.DataExtractionError,
+        data: expect.objectContaining({
+          error: expect.objectContaining({
+            message: expect.stringContaining('uploadAllRepos error'),
+          }),
+        }),
+      })
+    );
+    expect(mockPostMessage).toHaveBeenCalledWith(
+      expect.objectContaining({
+        subject: 'emit',
+        payload: { eventType: ExtractorEventType.DataExtractionError },
+      })
+    );
   });
 
   it('should include artifacts in data for extraction events', async () => {
@@ -493,6 +531,35 @@ describe(`${WorkerAdapter.name}.emit — ExternalSyncUnitExtractionDone legacy p
     >;
     expect(emittedData).not.toHaveProperty('external_sync_units');
   });
+
+  it('should emit an error when external sync unit upload fails', async () => {
+    const { adapter } = makeAdapter(EventType.StartExtractingExternalSyncUnits);
+    adapter['adapterState'].postState = jest.fn().mockResolvedValue(undefined);
+    adapter.uploadAllRepos = jest.fn().mockResolvedValue(undefined);
+    const pushMock = jest
+      .fn()
+      .mockRejectedValue(new Error('external sync unit upload failed'));
+    jest.spyOn(adapter, 'initializeRepos').mockImplementation(() => undefined);
+    jest.spyOn(adapter, 'getRepo').mockReturnValue({ push: pushMock } as never);
+
+    await adapter.emit(ExtractorEventType.ExternalSyncUnitExtractionDone, {
+      external_sync_units: [{ id: 'esu-1' }] as never,
+    });
+
+    const { emit: mockEmit } = require('../../common/control-protocol');
+    expect(mockEmit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        eventType: ExtractorEventType.ExternalSyncUnitExtractionError,
+        data: expect.objectContaining({
+          error: expect.objectContaining({
+            message: expect.stringContaining(
+              'external sync unit upload failed'
+            ),
+          }),
+        }),
+      })
+    );
+  });
 });
 
 describe('WorkerAdapter — workersOldest / workersNewest boundary updates', () => {