Merge pull request #74 from powersync-ja/fix-mutex

Fix hanging streams

Author: rkistner

.changeset/dull-foxes-explode.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-rsocket-router': patch
+'@powersync/service-core': patch
+---
+
+Fix hanging streams

packages/rsocket-router/src/router/ReactiveSocketRouter.ts

@@ -81,16 +81,19 @@ export class ReactiveSocketRouter<C> {
             // RequestStream is currently the only supported connection type
             requestStream: (payload, initialN, responder) => {
               const observer = new SocketRouterObserver();
+              const abortController = new AbortController();
 
               // TODO: Consider limiting the number of active streams per connection to prevent abuse
-              handleReactiveStream(context, { payload, initialN, responder }, observer, params).catch((ex) => {
-                logger.error(ex);
-                responder.onError(ex);
-                responder.onComplete();
-              });
+              handleReactiveStream(context, { payload, initialN, responder }, observer, abortController, params).catch(
+                (ex) => {
+                  logger.error(ex);
+                  responder.onError(ex);
+                  responder.onComplete();
+                }
+              );
               return {
                 cancel: () => {
-                  observer.triggerCancel();
+                  abortController.abort();
                 },
                 onExtension: () => observer.triggerExtension(),
                 request: (n) => observer.triggerRequest(n)
@@ -118,6 +121,7 @@ export async function handleReactiveStream<Context>(
     responder: SocketResponder;
   },
   observer: SocketRouterObserver,
+  abortController: AbortController,
   params: CommonParams<Context>
 ) {
   const { payload, responder, initialN } = request;
@@ -154,7 +158,13 @@ export async function handleReactiveStream<Context>(
   }
 
   if (authorize) {
-    const isAuthorized = await authorize({ params: requestPayload, context, observer, responder });
+    const isAuthorized = await authorize({
+      params: requestPayload,
+      context,
+      observer,
+      signal: abortController.signal,
+      responder
+    });
     if (!isAuthorized.authorized) {
       return exitWithError(new errors.AuthorizationError(isAuthorized.errors));
     }
@@ -165,6 +175,7 @@ export async function handleReactiveStream<Context>(
       params: requestPayload,
       context,
       observer,
+      signal: abortController.signal,
       responder,
       initialN
     });

packages/rsocket-router/src/router/SocketRouterListener.ts

@@ -1,16 +1,11 @@
 import { BaseObserver } from '../utils/BaseObserver.js';
 
 export interface SocketRouterListener {
-  cancel: () => void;
   onExtension: () => void;
   request: (n: number) => void;
 }
 
 export class SocketRouterObserver extends BaseObserver<SocketRouterListener> {
-  triggerCancel() {
-    this.iterateListeners((l) => l.cancel?.());
-  }
-
   triggerExtension() {
     this.iterateListeners((l) => l.onExtension?.());
   }

packages/rsocket-router/src/router/types.ts

@@ -25,6 +25,7 @@ export type SocketResponder = OnTerminalSubscriber & OnNextSubscriber & OnExtens
 export type CommonStreamPayload = {
   observer: SocketRouterObserver;
   responder: SocketResponder;
+  signal: AbortSignal;
 };
 
 export type ReactiveStreamPayload<O> = CommonStreamPayload & {

packages/rsocket-router/tests/src/requests.test.ts

@@ -23,6 +23,7 @@ async function handleRoute(path: string, endpoints: ReactiveEndpoint[], responde
       responder
     },
     createMockObserver(),
+    new AbortController(),
     {
       contextProvider: async () => ({}),
       endpoints,

packages/service-core/src/routes/endpoints/socket-route.ts

@@ -11,9 +11,26 @@ import { SyncRoutes } from './sync-stream.js';
 export const syncStreamReactive: SocketRouteGenerator = (router) =>
   router.reactiveStream<util.StreamingSyncRequest, any>(SyncRoutes.STREAM, {
     validator: schema.createTsCodecValidator(util.StreamingSyncRequest, { allowAdditional: true }),
-    handler: async ({ context, params, responder, observer, initialN }) => {
+    handler: async ({ context, params, responder, observer, initialN, signal: upstreamSignal }) => {
       const { system } = context;
 
+      // Create our own controller that we can abort directly
+      const controller = new AbortController();
+      upstreamSignal.addEventListener('abort', () => {
+        controller.abort();
+      });
+      if (upstreamSignal.aborted) {
+        controller.abort();
+      }
+      const signal = controller.signal;
+
+      let requestedN = initialN;
+      const disposer = observer.registerListener({
+        request(n) {
+          requestedN += n;
+        }
+      });
+
       if (system.closed) {
         responder.onError(
           new errors.JourneyError({
@@ -26,8 +43,6 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
         return;
       }
 
-      const controller = new AbortController();
-
       const syncParams = new RequestParameters(context.token_payload!, params.parameters ?? {});
 
       const storage = system.storage;
@@ -45,18 +60,8 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
         return;
       }
 
-      let requestedN = initialN;
-      const disposer = observer.registerListener({
-        request(n) {
-          requestedN += n;
-        },
-        cancel: () => {
-          controller.abort();
-        }
-      });
-
       const removeStopHandler = system.addStopHandler(() => {
-        observer.triggerCancel();
+        controller.abort();
       });
 
       Metrics.getInstance().concurrent_connections.add(1);
@@ -75,8 +80,11 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
             keep_alive: false
           },
           tracker,
-          signal: controller.signal
+          signal
         })) {
+          if (signal.aborted) {
+            break;
+          }
           if (data == null) {
             // Empty value just to flush iterator memory
             continue;
@@ -93,22 +101,25 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
             tracker.addDataSynced(serialized.length);
           }
 
-          if (requestedN <= 0) {
+          if (requestedN <= 0 && !signal.aborted) {
             await new Promise<void>((resolve) => {
               const l = observer.registerListener({
                 request() {
                   if (requestedN > 0) {
                     // Management of updating the total requested items is done above
                     resolve();
                     l();
+                    signal.removeEventListener('abort', onAbort);
                   }
-                },
-                cancel: () => {
-                  // Don't wait here if the request is cancelled
-                  resolve();
-                  l();
                 }
               });
+              const onAbort = () => {
+                // Don't wait here if the request is cancelled
+                resolve();
+                l();
+                signal.removeEventListener('abort', onAbort);
+              };
+              signal.addEventListener('abort', onAbort);
             });
           }
         }

packages/service-core/src/sync/sync.ts

@@ -1,23 +1,34 @@
 import { JSONBig, JsonContainer } from '@powersync/service-jsonbig';
 import { RequestParameters } from '@powersync/service-sync-rules';
-import { Semaphore } from 'async-mutex';
+import { Semaphore, withTimeout } from 'async-mutex';
 import { AbortError } from 'ix/aborterror.js';
 
 import * as auth from '../auth/auth-index.js';
 import * as storage from '../storage/storage-index.js';
 import * as util from '../util/util-index.js';
 
 import { logger } from '@powersync/lib-services-framework';
-import { Metrics } from '../metrics/Metrics.js';
 import { mergeAsyncIterables } from './merge.js';
-import { TokenStreamOptions, tokenStream } from './util.js';
 import { RequestTracker } from './RequestTracker.js';
+import { TokenStreamOptions, tokenStream } from './util.js';
 
 /**
  * Maximum number of connections actively fetching data.
  */
 const MAX_ACTIVE_CONNECTIONS = 10;
-const syncSemaphore = new Semaphore(MAX_ACTIVE_CONNECTIONS);
+
+/**
+ * Maximum duration to wait for the mutex to become available.
+ *
+ * This gives an explicit error if there are mutex issues, rather than just hanging.
+ */
+const MUTEX_ACQUIRE_TIMEOUT = 30_000;
+
+const syncSemaphore = withTimeout(
+  new Semaphore(MAX_ACTIVE_CONNECTIONS),
+  MUTEX_ACQUIRE_TIMEOUT,
+  new Error(`Timeout while waiting for data`)
+);
 
 export interface SyncStreamParameters {
   storage: storage.BucketStorageFactory;