fix tests

packages/cojson/src/PeerState.ts

@@ -207,6 +207,7 @@ export class PeerState {
     this.peer.crashOnClose = false;
     this.peer.outgoing.close();
     this.closed = true;
+    this.queue.clearMetrics();
     this.emitClose();
   }
 

packages/cojson/src/PriorityBasedMessageQueue.ts

@@ -65,6 +65,10 @@ export class LinkedList<T> {
     this.meter?.pull();
     return value;
   }
+
+  clearMetrics() {
+    this.meter?.clear(this.length);
+  }
 }
 
 class QueueMeter {
@@ -106,6 +110,10 @@ class QueueMeter {
   public push() {
     this.pushCounter.add(1, this.attrs);
   }
+
+  public clear(items: number) {
+    this.pullCounter.add(items, this.attrs);
+  }
 }
 
 function meteredList<T>(attrs?: Record<string, string | number>) {
@@ -151,4 +159,8 @@ export class PriorityBasedMessageQueue {
 
     return this.queues[priority]?.shift();
   }
+
+  public clearMetrics() {
+    this.queues.forEach((queue) => queue.clearMetrics());
+  }
 }

packages/cojson/src/tests/PriorityBasedMessageQueue.test.ts

@@ -19,6 +19,42 @@ describe("PriorityBasedMessageQueue", () => {
   });
 
   describe("meteredQueue", () => {
+    test("should corretly clear metrics", async () => {
+      const { queue, metricReader } = setup();
+      const message: SyncMessage = {
+        action: "load",
+        id: "co_ztest-id",
+        header: false,
+        sessions: {},
+      };
+
+      expect(
+        await metricReader.getMetricValue("jazz.messagequeue.pushed", {
+          priority: CO_VALUE_PRIORITY.MEDIUM,
+        }),
+      ).toBe(0);
+
+      void queue.push(message);
+      void queue.push(message);
+      void queue.push(message);
+      expect(
+        await metricReader.getMetricValue("jazz.messagequeue.pushed", {
+          priority: CO_VALUE_PRIORITY.MEDIUM,
+        }),
+      ).toBe(3);
+
+      queue.clearMetrics();
+      expect(
+        await metricReader.getMetricValue("jazz.messagequeue.pulled", {
+          priority: CO_VALUE_PRIORITY.MEDIUM,
+        }),
+      ).toBe(
+        await metricReader.getMetricValue("jazz.messagequeue.pushed", {
+          priority: CO_VALUE_PRIORITY.MEDIUM,
+        }),
+      );
+    });
+
     test("should corretly count pushes", async () => {
       const { queue, metricReader } = setup();
       const message: SyncMessage = {