feat: simple backpressure for sync storage

Author: gdorsi

packages/cojson/src/PriorityBasedMessageQueue.ts

@@ -1,4 +1,5 @@
 import { Counter, ValueType, metrics } from "@opentelemetry/api";
+import { PeerState } from "./PeerState.js";
 import { CO_VALUE_PRIORITY, type CoValuePriority } from "./priority.js";
 import type { SyncMessage } from "./sync.js";
 
@@ -65,6 +66,10 @@ export class LinkedList<T> {
     this.meter?.pull();
     return value;
   }
+
+  isEmpty() {
+    return this.head === undefined;
+  }
 }
 
 class QueueMeter {
@@ -152,3 +157,79 @@ export class PriorityBasedMessageQueue {
     return this.queues[priority]?.shift();
   }
 }
+
+export class IncomingMessagesQueue {
+  private queues: [LinkedList<SyncMessage>, PeerState][];
+  private peerToQueue: WeakMap<PeerState, LinkedList<SyncMessage>>;
+  currentQueue = 0;
+
+  constructor() {
+    this.queues = [];
+    this.peerToQueue = new WeakMap();
+  }
+
+  public push(msg: SyncMessage, peer: PeerState) {
+    const queue = this.peerToQueue.get(peer);
+
+    if (!queue) {
+      const newQueue = new LinkedList<SyncMessage>();
+      this.peerToQueue.set(peer, newQueue);
+      this.queues.push([newQueue, peer]);
+      newQueue.push(msg);
+    } else {
+      queue.push(msg);
+    }
+  }
+
+  public pull() {
+    const entry = this.queues[this.currentQueue];
+
+    if (!entry) {
+      return undefined;
+    }
+
+    const [queue, peer] = entry;
+    const msg = queue.shift();
+
+    if (queue.isEmpty()) {
+      this.queues.splice(this.currentQueue, 1);
+      this.peerToQueue.delete(peer);
+    } else {
+      this.currentQueue++;
+    }
+
+    if (this.currentQueue >= this.queues.length) {
+      this.currentQueue = 0;
+    }
+
+    if (msg) {
+      return { msg, peer };
+    }
+
+    return undefined;
+  }
+
+  processing = false;
+
+  processQueue(
+    callback: (msg: SyncMessage, peer: PeerState, stop: () => void) => void,
+  ) {
+    this.processing = true;
+
+    let entry: { msg: SyncMessage; peer: PeerState } | undefined;
+
+    while ((entry = this.pull())) {
+      const { msg, peer } = entry;
+
+      callback(msg, peer, () => {
+        this.processing = false;
+      });
+
+      if (!this.processing) {
+        break;
+      }
+    }
+
+    this.processing = false;
+  }
+}

packages/cojson/src/sync.ts

@@ -1,5 +1,9 @@
 import { Histogram, ValueType, metrics } from "@opentelemetry/api";
 import { PeerState } from "./PeerState.js";
+import {
+  IncomingMessagesQueue,
+  PriorityBasedMessageQueue,
+} from "./PriorityBasedMessageQueue.js";
 import { SyncStateManager } from "./SyncStateManager.js";
 import { CoValueCore } from "./coValueCore/coValueCore.js";
 import { getDependedOnCoValuesFromRawData } from "./coValueCore/utils.js";
@@ -8,7 +12,7 @@ import { Signature } from "./crypto/crypto.js";
 import { RawCoID, SessionID } from "./ids.js";
 import { LocalNode } from "./localNode.js";
 import { logger } from "./logger.js";
-import { CoValuePriority } from "./priority.js";
+import { CO_VALUE_PRIORITY, CoValuePriority } from "./priority.js";
 import { accountOrAgentIDfromSessionID } from "./typeUtils/accountOrAgentIDfromSessionID.js";
 import { isAccountID } from "./typeUtils/isAccountID.js";
 
@@ -303,6 +307,30 @@ export class SyncManager {
     }
   }
 
+  messagesQueue = new IncomingMessagesQueue();
+  async pushMessage(incoming: SyncMessage, peer: PeerState) {
+    this.messagesQueue.push(incoming, peer);
+
+    if (this.messagesQueue.processing) {
+      return;
+    }
+
+    let lastTimer = performance.now();
+    let paused = false;
+    this.messagesQueue.processQueue((msg, peer, stop) => {
+      this.handleSyncMessage(msg, peer);
+
+      if (performance.now() - lastTimer > 50) {
+        paused = true;
+        stop();
+      }
+    });
+
+    if (paused) {
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+  }
+
   addPeer(peer: Peer) {
     const prevPeer = this.peers[peer.id];
 
@@ -327,7 +355,7 @@ export class SyncManager {
 
     peerState
       .processIncomingMessages((msg) => {
-        this.handleSyncMessage(msg, peerState);
+        this.pushMessage(msg, peerState);
       })
       .then(() => {
         if (peer.crashOnClose) {