Merge remote-tracking branch 'origin/main' into client-id

Author: rkistner

.changeset/plenty-ads-matter.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-rsocket-router': patch
+---
+
+Added a log for current count of connected WebSocket connections.

.changeset/tricky-phones-tell.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-core': patch
+---
+
+Log stats on sync lock when reaching concurrency limit

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

@@ -197,7 +197,8 @@ async function* streamResponseInner(
       raw_data,
       binary_data,
       signal,
-      tracker
+      tracker,
+      user_id: syncParams.user_id
     });
 
     await new Promise((resolve) => setTimeout(resolve, 10));
@@ -214,6 +215,7 @@ interface BucketDataRequest {
   binary_data: boolean | undefined;
   tracker: RequestTracker;
   signal: AbortSignal;
+  user_id?: string;
 }
 
 async function* bucketDataInBatches(request: BucketDataRequest) {
@@ -262,8 +264,19 @@ async function* bucketDataBatch(request: BucketDataRequest): AsyncGenerator<Buck
   const checkpointOp = BigInt(checkpoint);
   let checkpointInvalidated = false;
 
-  const [_, release] = await syncSemaphore.acquire();
+  if (syncSemaphore.isLocked()) {
+    logger.info('Sync concurrency limit reached, waiting for lock', { user_id: request.user_id });
+  }
+  const [value, release] = await syncSemaphore.acquire();
   try {
+    if (value <= 3) {
+      // This can be noisy, so we only log when we get close to the
+      // concurrency limit.
+      logger.info(`Got sync lock. Slots available: ${value - 1}`, {
+        user_id: request.user_id,
+        sync_data_slots: value - 1
+      });
+    }
     // Optimization: Only fetch buckets for which the checksums have changed since the last checkpoint
     // For the first batch, this will be all buckets.
     const filteredBuckets = new Map(bucketsToFetch.map((bucket) => [bucket, dataBuckets.get(bucket)!]));
@@ -331,6 +344,13 @@ async function* bucketDataBatch(request: BucketDataRequest): AsyncGenerator<Buck
       }
     }
   } finally {
+    if (value <= 3) {
+      // This can be noisy, so we only log when we get close to the
+      // concurrency limit.
+      logger.info(`Releasing sync lock`, {
+        user_id: request.user_id
+      });
+    }
     release();
   }
 }