AI feedback

Author: simolus3

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -16,10 +16,10 @@ import {
   doneResult,
   injectable,
   InjectableIterator,
-  notifyIterator,
   SimpleAsyncIterator,
   valueResult
 } from '../../../utils/stream_transform.js';
+import { asyncNotifier } from '../../../utils/async.js';
 import { StreamingSyncRequestParameterType } from './JsonValue.js';
 
 export enum LockType {
@@ -219,13 +219,12 @@ export abstract class AbstractStreamingSyncImplementation
   protected logger: ILogger;
   private activeStreams: SubscribedStream[];
   private connectionMayHaveChanged = false;
-  private crudUploadNotifier = notifyIterator();
+  private crudUploadNotifier = asyncNotifier();
 
   private notifyCompletedUploads?: () => void;
   private handleActiveStreamsChange?: () => void;
 
   syncStatus: SyncStatus;
-  triggerCrudUpload: () => void;
 
   constructor(options: AbstractStreamingSyncImplementationOptions) {
     super();
@@ -243,8 +242,10 @@ export abstract class AbstractStreamingSyncImplementation
       }
     });
     this.abortController = null;
+  }
 
-    this.triggerCrudUpload = () => this.crudUploadNotifier.notify();
+  triggerCrudUpload() {
+    this.crudUploadNotifier.notify();
   }
 
   async waitForReady() {}
@@ -321,7 +322,7 @@ export abstract class AbstractStreamingSyncImplementation
         this.delayRetry(signal, this.options.crudUploadThrottleMs!)
       ]);
 
-      await this.crudUploadNotifier.next();
+      await this.crudUploadNotifier.waitForNotification(signal);
     }
   }
 
@@ -483,7 +484,7 @@ The next upload iteration will be delayed.`);
       });
     });
 
-    this.crudUploadLoop(signal);
+    this.crudUploadLoop(signal).catch((ex) => this.logger.error('Error in crud upload loop', ex));
 
     /**
      * This loops runs until [retry] is false or the abort signal is set to aborted.

packages/common/src/utils/async.ts

@@ -18,3 +18,60 @@ export function throttleTrailing(func: () => void, wait: number) {
     }
   };
 }
+
+export interface AsyncNotifier {
+  /**
+   * @param signal Also resolve the promise once this signal completes.
+   * @returns A promise that resolves once {@link notify} is called after this promise was last resolved.
+   */
+  waitForNotification(signal: AbortSignal): Promise<void>;
+
+  /**
+   * Notifies any pending listener, or makes the next {@link waitForNotification} complete immediately if no listener
+   * is currently active.
+   */
+  notify(): void;
+}
+
+export function asyncNotifier(): AsyncNotifier {
+  let waitingConsumers: (() => void)[] = [];
+  let hasPendingNotification = false;
+
+  return {
+    notify() {
+      if (waitingConsumers.length > 0) {
+        waitingConsumers.splice(0, 1)[0]();
+      } else {
+        hasPendingNotification = true;
+      }
+    },
+    waitForNotification(signal: AbortSignal) {
+      return new Promise((resolve) => {
+        if (hasPendingNotification) {
+          resolve();
+          hasPendingNotification = false;
+        } else {
+          function complete() {
+            signal.removeEventListener('abort', onAbort);
+            resolve();
+          }
+
+          function onAbort() {
+            const i = waitingConsumers.indexOf(complete);
+            if (i > -1) {
+              waitingConsumers.splice(i, 1);
+            }
+            resolve();
+          }
+
+          waitingConsumers.push(complete);
+          if (signal.aborted) {
+            onAbort();
+          } else {
+            signal.addEventListener('abort', onAbort);
+          }
+        }
+      });
+    }
+  };
+}

packages/common/src/utils/stream_transform.ts

@@ -121,53 +121,6 @@ export function injectable<T>(source: SimpleAsyncIterator<T>): InjectableIterato
   };
 }
 
-export interface NotifyIterator extends SimpleAsyncIterator<undefined> {
-  notify(): void;
-}
-
-/**
- * An iterator dispatching notifications without associated data.
- *
- * Backpressure is handled by folding calls to `notify()`: The iterator emits an item once there's been at least one
- * `notify()` call since the last completed call to `next()`.
- */
-export function notifyIterator(): NotifyIterator {
-  // Possible states:
-  //   - Idle (null), no outstanding listener and no outstanding notification.
-  //   - Waiting (function), has an outstanding listener and no notification.
-  //   - Pending (true), no outstanding listener but an undelivered notification.
-  let state: null | ((complete: IteratorResult<undefined, any>) => void) | true = null;
-  const item = valueResult(undefined);
-
-  return {
-    notify() {
-      if (typeof state == 'function') {
-        // waiting -> idle
-        state(item);
-        state = null;
-      } else {
-        state = true; // idle -> pending or pending -> pending.
-      }
-    },
-    next() {
-      return new Promise((resolve) => {
-        if (typeof state == 'function') {
-          throw new Error('concurrent call to next');
-        }
-
-        if (state != null) {
-          // pending -> idle
-          resolve(item);
-          state = null;
-        } else {
-          // idle -> waiting
-          state = resolve;
-        }
-      });
-    }
-  };
-}
-
 /**
  * Splits a byte stream at line endings, emitting each line as a string.
  */

packages/common/tests/utils/async.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, test } from 'vitest';
+import { asyncNotifier } from '../../src/utils/async';
+
+describe('notifyIterator', () => {
+  const neverAbort = new AbortController().signal;
+
+  test('waits for event', async () => {
+    const notifier = asyncNotifier();
+    let didReceiveEvent = false;
+    notifier.waitForNotification(neverAbort).then((e) => {
+      didReceiveEvent = true;
+    });
+
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeFalsy();
+    notifier.notify();
+
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeTruthy();
+  });
+
+  test('merges events', async () => {
+    const notifier = asyncNotifier();
+    for (let i = 0; i < 1000; i++) {
+      notifier.notify();
+    }
+
+    let didReceiveEvent = false;
+    notifier.waitForNotification(neverAbort).then((e) => {
+      didReceiveEvent = true;
+    });
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeTruthy();
+
+    // The first iterator.next() should have consumed everything.
+    didReceiveEvent = false;
+    notifier.waitForNotification(neverAbort).then((e) => {
+      didReceiveEvent = true;
+    });
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeFalsy();
+  });
+
+  test('completes on abort', async () => {
+    const notifier = asyncNotifier();
+    await notifier.waitForNotification(AbortSignal.abort());
+  });
+
+  test('completes on abort later', async () => {
+    const notifier = asyncNotifier();
+    await notifier.waitForNotification(AbortSignal.timeout(10));
+  });
+});

packages/common/tests/utils/stream_transform.test.ts

@@ -4,7 +4,6 @@ import {
   extractBsonObjects,
   extractJsonLines,
   injectable,
-  notifyIterator,
   SimpleAsyncIterator,
   valueResult
 } from '../../src/utils/stream_transform';
@@ -121,48 +120,6 @@ describe('injectable', () => {
   });
 });
 
-describe('notifyIterator', () => {
-  test('waits for event', async () => {
-    const iterator = notifyIterator();
-    let didReceiveEvent = false;
-    iterator.next().then((e) => {
-      expect(e.done).toStrictEqual(false);
-      didReceiveEvent = true;
-    });
-
-    await Promise.resolve();
-    expect(didReceiveEvent).toBeFalsy();
-    iterator.notify();
-
-    await Promise.resolve();
-    expect(didReceiveEvent).toBeTruthy();
-  });
-
-  test('merges events', async () => {
-    const iterator = notifyIterator();
-    for (let i = 0; i < 1000; i++) {
-      iterator.notify();
-    }
-
-    let didReceiveEvent = false;
-    iterator.next().then((e) => {
-      expect(e.done).toStrictEqual(false);
-      didReceiveEvent = true;
-    });
-    await Promise.resolve();
-    expect(didReceiveEvent).toBeTruthy();
-
-    // The first iterator.next() should have consumed everything.
-    didReceiveEvent = false;
-    iterator.next().then((e) => {
-      expect(e.done).toStrictEqual(false);
-      didReceiveEvent = true;
-    });
-    await Promise.resolve();
-    expect(didReceiveEvent).toBeFalsy();
-  });
-});
-
 describe('bson objects', () => {
   test('empty stream', async () => {
     async function* source() {}