Repository: igorkamyshev/farfetched (ts)

Issue:
Unable to use `onAbort` more that one time
Hello, thank you for the library!

According to the [documentation](https://farfetched.pages.dev/api/utils/on_abort.html), for the actual interruption of a request, we should pass the abort controller's signal to a lower level in the handler or effect properties. However, if we don't use createJsonQuery and attempt to apply onAbort to our effects or handlers more than once, the code simply stops working with an onAbort call is not allowed error.

Easy example:
```
const someFx = createRequest({
    endpoint: '/',
    method: 'GET'
})

const queryA = createQuery({
    async handler() {
        const abortController = new AbortController()

        console.log('called')

        onAbort(() => abortController.abort())
        
        console.log('executed')

        return someFx({ __: { signal: abortController.signal } })
    }
})

const queryB = createQuery({
    async handler() {
        const abortController = new AbortController()

        console.log('called')

        onAbort(() => abortController.abort())

        console.log('fails here')

        return someFx({ __: { signal: abortController.signal } })
    }
})
```

In this example, onAbort in queryA works well, but onAbort in queryB fails.

Pull request description:
Fix race in validation of `obAbort` call in case of starting many operations by single `sample`
fixes #449

Patch:
diff --git a/.changeset/strong-starfishes-pretend.md b/.changeset/strong-starfishes-pretend.md
new file mode 100644
index 0000000..ed8bacb
--- /dev/null
+++ b/.changeset/strong-starfishes-pretend.md
@@ -0,0 +1,5 @@
+---
+'@farfetched/core': patch
+---
+
+Fix race in validation of `obAbort` call in case of starting many operations by single `sample`
diff --git a/packages/core/package.json b/packages/core/package.json
index cb4786b..bedab4c 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -41,7 +41,7 @@
   "size-limit": [
     {
       "path": "./dist/core.js",
-      "limit": "15.9 kB"
+      "limit": "15.95 kB"
     }
   ]
 }
diff --git a/packages/core/src/remote_operation/create_remote_operation.ts b/packages/core/src/remote_operation/create_remote_operation.ts
index bb9f999..34560ad 100644
--- a/packages/core/src/remote_operation/create_remote_operation.ts
+++ b/packages/core/src/remote_operation/create_remote_operation.ts
@@ -29,7 +29,6 @@ import { type RemoteOperation } from './type';
 import { get } from '../libs/lohyphen';
 import { isAbortError } from '../errors/guards';
 import { getCallObjectEvent } from './with_call_object';
-import { allowCancelSetting, disallowCancelSetting } from './on_abort';
 
 export function createRemoteOperation<
   Params,
@@ -96,11 +95,8 @@ export function createRemoteOperation<
       { result: unknown; stale: boolean } | null,
       unknown
     >(async ({ params }) => {
-      allowCancelSetting();
-      const promise = executeFx(params);
-      disallowCancelSetting();
+      const result = await executeFx(params);
 
-      const result = await promise;
       return { result, stale: false };
     }),
   };
diff --git a/packages/core/src/remote_operation/with_call_object.ts b/packages/core/src/remote_operation/with_call_object.ts
index 9549329..c48b898 100644
--- a/packages/core/src/remote_operation/with_call_object.ts
+++ b/packages/core/src/remote_operation/with_call_object.ts
@@ -9,7 +9,11 @@ import {
 
 import { Defer, createDefer } from '../libs/lohyphen';
 import { abortError } from '../errors/create_error';
-import { occupyCurrentCancelCallback } from './on_abort';
+import {
+  occupyCurrentCancelCallback,
+  allowCancelSetting,
+  disallowCancelSetting,
+} from './on_abort';
 
 export type CallObject = {
   id: string;
@@ -84,7 +88,8 @@ function createPatchedHandler(
      * Normal flow of the handler,
      * its result is reflected to the outside world
      */
-    const result = h(...p);
+    const { result, abortCallback } = runWithExposedAbort(h, ...p);
+
     if (result instanceof Promise) {
       /**
        * Async handlers are patched to emit call object,
@@ -92,7 +97,7 @@ function createPatchedHandler(
        */
 
       const def = createDefer();
-      const callObj = createCallObject(def);
+      const callObj = createCallObject(def, abortCallback);
       calledEvent(callObj);
       result.then(def.resolve, def.reject);
 
@@ -100,9 +105,9 @@ function createPatchedHandler(
     } else {
       /**
        * It is not possible to control sync handlers at all, because their execution is instant
-       * So call object is emitted as "finished" right away
+       * So call object is not provided with defer and thus emitted as "finished" right away
        */
-      const callObj = createCallObject();
+      const callObj = createCallObject(undefined, abortCallback);
       calledEvent(callObj);
 
       return result;
@@ -112,7 +117,7 @@ function createPatchedHandler(
   return ffMagicHandler;
 }
 
-function createCallObject(def?: Defer<unknown, unknown>) {
+function createCallObject(def?: Defer<unknown, unknown>, onAbort?: any) {
   let callStatus: CallObject['status'] = def ? 'pending' : 'finished';
 
   function finish() {
@@ -124,13 +129,11 @@ function createCallObject(def?: Defer<unknown, unknown>) {
     def.promise.then(finish, finish);
   }
 
-  const cancelCallback = occupyCurrentCancelCallback();
-
   const callObj: CallObject = {
     id: getCallId(),
     status: callStatus,
     abort: (error: unknown = abortError()) => {
-      cancelCallback?.();
+      onAbort?.();
       if (callStatus === 'finished') {
         /**
          * It is not possible to abort already finished call,
@@ -162,3 +165,39 @@ function getCallId(): string {
 
   return id;
 }
+
+/**
+ * Runs handler with exposed abort control - `onAbort` hooks works here
+ *
+ * @param h - original handler
+ * @param p - parameters to pass to the handler
+ * @returns {result: unknown, abortCallback: (() => void) | null}
+ */
+function runWithExposedAbort(
+  h: (...args: any[]) => any,
+  ...p: any[]
+): {
+  result: unknown;
+  abortCallback: (() => void) | null;
+} {
+  /**
+   * Hacky way to work around effector's mechanic to sync flush internal queue on amy imperative call,
+   * which in turn messes with `onAbort` hook mechanics, because it relies upon JS callstack here
+   *
+   * By calling dummy event we are forcing the flush of the queue early, before the handler is called.
+   * This essentially isolates handler execution from any other effector nodes in the queue.
+   */
+  flushQueue();
+
+  allowCancelSetting();
+  const result = h(...p);
+  const abortCallback = occupyCurrentCancelCallback();
+  disallowCancelSetting();
+
+  return {
+    result,
+    abortCallback,
+  };
+}
+
+const flushQueue = createEvent();


Test patch:
diff --git a/packages/core/src/attach/__tests__/attach.query.test.ts b/packages/core/src/attach/__tests__/attach.query.test.ts
index 1661ce6..9cf3196 100644
--- a/packages/core/src/attach/__tests__/attach.query.test.ts
+++ b/packages/core/src/attach/__tests__/attach.query.test.ts
@@ -222,7 +222,7 @@ describe('attach for query', () => {
 
     expect(originalHandler).toBeCalledTimes(2);
 
-    expect(originalHandler.mock.calls[0][0].url).toBe('https://api.salo.com/1');
-    expect(originalHandler.mock.calls[1][0].url).toBe('https://api.salo.com/2');
+    expect(originalHandler.mock.calls[0][0].url).toBe('https://api.salo.com/2');
+    expect(originalHandler.mock.calls[1][0].url).toBe('https://api.salo.com/1');
   });
 });
diff --git a/packages/core/src/concurrency/__tests__/concurrency.test.ts b/packages/core/src/concurrency/__tests__/concurrency.test.ts
index ae1291d..af5e815 100644
--- a/packages/core/src/concurrency/__tests__/concurrency.test.ts
+++ b/packages/core/src/concurrency/__tests__/concurrency.test.ts
@@ -1,4 +1,4 @@
-import { allSettled, createEvent, createWatch, fork } from 'effector';
+import { allSettled, createEvent, createWatch, fork, sample } from 'effector';
 import { describe, test, expect, vi } from 'vitest';
 import { setTimeout } from 'timers/promises';
 
@@ -6,6 +6,9 @@ import { createDefer } from '../../libs/lohyphen';
 import { createQuery } from '../../query/create_query';
 import { onAbort } from '../../remote_operation/on_abort';
 import { concurrency } from '../concurrency';
+import { createJsonQuery } from '../../query/create_json_query';
+import { unknownContract } from '../../contract/unknown_contract';
+import { fetchFx } from '../../fetch/fetch';
 
 describe('concurrency', async () => {
   test('cancel all pending requests on abort event', async () => {
@@ -114,4 +117,366 @@ describe('concurrency', async () => {
 
     expect(scope.getState(q.$pending)).toBeFalsy();
   });
+
+  describe('issue #449', () => {
+    test('two createQuery, two onAbort, abortAll', async () => {
+      const q1Aborted = vi.fn();
+      const q2Aborted = vi.fn();
+
+      const q1 = createQuery({
+        /* Will not be resolved */
+        handler: async () => {
+          const defer = createDefer<any>();
+
+          onAbort(() => defer.reject());
+
+          return defer.promise;
+        },
+      });
+
+      const q2 = createQuery({
+        /* Will be resolved immediately */
+        handler: async () => {
+          const defer = createDefer<string>();
+
+          onAbort(() => defer.reject());
+
+          await setTimeout(1);
+
+          defer.resolve('Data');
+
+          return defer.promise;
+        },
+      });
+
+      const stop = createEvent();
+
+      concurrency(q1, { abortAll: stop });
+
+      const scope = fork();
+
+      createWatch({ unit: q1.aborted, scope, fn: q1Aborted });
+      createWatch({ unit: q2.aborted, scope, fn: q2Aborted });
+
+      allSettled(q1.start, { scope });
+      allSettled(q2.start, { scope });
+
+      await allSettled(stop, { scope });
+
+      expect(q1Aborted).toHaveBeenCalledTimes(1);
+      expect(q2Aborted).not.toHaveBeenCalled();
+
+      expect(scope.getState(q1.$pending)).toBeFalsy();
+      expect(scope.getState(q2.$pending)).toBeFalsy();
+
+      expect(scope.getState(q1.$error)).toEqual(null);
+      expect(scope.getState(q2.$error)).toEqual(null);
+
+      expect(scope.getState(q1.$data)).toEqual(null);
+      expect(scope.getState(q2.$data)).toEqual('Data');
+    });
+
+    test('two createJsonQuery, abortAll', async () => {
+      const q1Aborted = vi.fn();
+      const q2Aborted = vi.fn();
+
+      const q1 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q1', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      const q2 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q2', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      const stop = createEvent();
+
+      concurrency(q1, { abortAll: stop });
+
+      const scope = fork({
+        handlers: [
+          [
+            fetchFx,
+            async (r: Request) =>
+              setTimeout(10).then(
+                () => new Response(`{ "data": "Data from ${r.url}"}`)
+              ),
+          ],
+        ],
+      });
+
+      createWatch({ unit: q1.aborted, scope, fn: q1Aborted });
+      createWatch({ unit: q2.aborted, scope, fn: q2Aborted });
+
+      allSettled(q1.start, { scope });
+      allSettled(q2.start, { scope });
+
+      await allSettled(stop, { scope });
+
+      expect(q1Aborted).toHaveBeenCalledTimes(1);
+      expect(q2Aborted).not.toHaveBeenCalled();
+
+      expect(scope.getState(q1.$pending)).toBeFalsy();
+      expect(scope.getState(q2.$pending)).toBeFalsy();
+
+      expect(scope.getState(q1.$error)).toEqual(null);
+      expect(scope.getState(q2.$error)).toEqual(null);
+
+      expect(scope.getState(q1.$data)).toEqual(null);
+      expect(scope.getState(q2.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q2",
+        }
+      `);
+    });
+
+    test('two createJsonQuery, TAKE_LATEST', async () => {
+      const q1Aborted = vi.fn();
+      const q2Aborted = vi.fn();
+
+      const q1FinishedSuccessfully = vi.fn();
+
+      const q1 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q1', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      const q2 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q2', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q1, { strategy: 'TAKE_LATEST' });
+
+      const scope = fork({
+        handlers: [
+          [
+            fetchFx,
+            async (r: Request) => {
+              await setTimeout(10);
+
+              if (r.signal.aborted) {
+                throw new Error('Request aborted, WTF');
+              }
+
+              return new Response(`{ "data": "Data from ${r.url}"}`);
+            },
+          ],
+        ],
+      });
+
+      createWatch({ unit: q1.aborted, scope, fn: q1Aborted });
+      createWatch({ unit: q2.aborted, scope, fn: q2Aborted });
+
+      createWatch({
+        unit: q1.finished.success,
+        scope,
+        fn: q1FinishedSuccessfully,
+      });
+
+      allSettled(q1.start, { scope });
+      allSettled(q1.start, { scope });
+
+      allSettled(q2.start, { scope });
+
+      await allSettled(scope);
+
+      expect(q1Aborted).toHaveBeenCalledTimes(1);
+      expect(q2Aborted).not.toHaveBeenCalled();
+
+      expect(q1FinishedSuccessfully).toHaveBeenCalledTimes(1);
+
+      expect(scope.getState(q1.$pending)).toBeFalsy();
+      expect(scope.getState(q2.$pending)).toBeFalsy();
+
+      expect(scope.getState(q1.$error)).toEqual(null);
+      expect(scope.getState(q2.$error)).toEqual(null);
+
+      expect(scope.getState(q1.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q1",
+        }
+      `);
+      expect(scope.getState(q2.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q2",
+        }
+      `);
+    });
+
+    test('two createJsonQuery, TAKE_LATEST, start by sample', async () => {
+      const q1Aborted = vi.fn();
+      const q2Aborted = vi.fn();
+
+      const q1 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q1', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q1, { strategy: 'TAKE_LATEST' });
+
+      const q2 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q2', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q1, { strategy: 'TAKE_LATEST' });
+
+      const start = createEvent();
+
+      sample({
+        clock: start,
+        target: [q1.start, q2.start],
+      });
+
+      const scope = fork({
+        handlers: [
+          [
+            fetchFx,
+            async (r: Request) => {
+              await setTimeout(10);
+
+              if (r.signal.aborted) {
+                throw new Error('Request aborted, WTF');
+              }
+
+              return new Response(`{ "data": "Data from ${r.url}"}`);
+            },
+          ],
+        ],
+      });
+
+      createWatch({ unit: q1.aborted, scope, fn: q1Aborted });
+      createWatch({ unit: q2.aborted, scope, fn: q2Aborted });
+
+      allSettled(start, { scope });
+
+      await allSettled(scope);
+
+      expect(q1Aborted).not.toHaveBeenCalled();
+      expect(q2Aborted).not.toHaveBeenCalled();
+
+      expect(scope.getState(q1.$pending)).toBeFalsy();
+      expect(scope.getState(q2.$pending)).toBeFalsy();
+
+      expect(scope.getState(q1.$error)).toEqual(null);
+      expect(scope.getState(q2.$error)).toEqual(null);
+
+      expect(scope.getState(q1.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q1",
+        }
+      `);
+      expect(scope.getState(q2.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q2",
+        }
+      `);
+    });
+
+    test('4 parallel createJsonQuery, TAKE_LATEST, start by sample', async () => {
+      const q1Aborted = vi.fn();
+      const q2Aborted = vi.fn();
+      const q3Aborted = vi.fn();
+      const q4Aborted = vi.fn();
+
+      const q1 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q1', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q1, { strategy: 'TAKE_LATEST' });
+
+      const q2 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q2', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q1, { strategy: 'TAKE_LATEST' });
+
+      const q3 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q3', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q3, { strategy: 'TAKE_LATEST' });
+
+      const q4 = createJsonQuery({
+        request: { url: 'https://api.salo.com/q4', method: 'GET' },
+        response: { contract: unknownContract },
+      });
+
+      concurrency(q4, { strategy: 'TAKE_LATEST' });
+
+      const start = createEvent();
+
+      sample({
+        clock: start,
+        target: [q1.start, q2.start, q3.start, q4.start],
+      });
+
+      const scope = fork({
+        handlers: [
+          [
+            fetchFx,
+            async (r: Request) => {
+              await setTimeout(10);
+
+              if (r.signal.aborted) {
+                throw new Error('Request aborted, WTF');
+              }
+
+              return new Response(`{ "data": "Data from ${r.url}"}`);
+            },
+          ],
+        ],
+      });
+
+      createWatch({ unit: q1.aborted, scope, fn: q1Aborted });
+      createWatch({ unit: q2.aborted, scope, fn: q2Aborted });
+      createWatch({ unit: q3.aborted, scope, fn: q3Aborted });
+      createWatch({ unit: q4.aborted, scope, fn: q4Aborted });
+
+      allSettled(start, { scope });
+
+      await allSettled(scope);
+
+      expect(q1Aborted).not.toHaveBeenCalled();
+      expect(q2Aborted).not.toHaveBeenCalled();
+      expect(q3Aborted).not.toHaveBeenCalled();
+      expect(q4Aborted).not.toHaveBeenCalled();
+
+      expect(scope.getState(q1.$pending)).toBeFalsy();
+      expect(scope.getState(q2.$pending)).toBeFalsy();
+      expect(scope.getState(q3.$pending)).toBeFalsy();
+      expect(scope.getState(q4.$pending)).toBeFalsy();
+
+      expect(scope.getState(q1.$error)).toEqual(null);
+      expect(scope.getState(q2.$error)).toEqual(null);
+      expect(scope.getState(q3.$error)).toEqual(null);
+      expect(scope.getState(q4.$error)).toEqual(null);
+
+      expect(scope.getState(q1.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q1",
+        }
+      `);
+      expect(scope.getState(q2.$data)).toMatchInlineSnapshot(`
+        {
+          "data": "Data from https://api.salo.com/q2",
+        }
+      `);
+      expect(scope.getState(q3.$data)).toMatchInlineSnapshot(`
+      {
+        "data": "Data from https://api.salo.com/q3",
+      }
+    `);
+      expect(scope.getState(q4.$data)).toMatchInlineSnapshot(`
+      {
+        "data": "Data from https://api.salo.com/q4",
+      }
+    `);
+    });
+  });
 });