feat(reliable-channel): Emit a "Synced" Status with message counts

Return a "synced" or "syncing" status on `ReliableChannel.status` that
let the developer know whether messages are missing, and if so, how many.

Author: fryorcraken

packages/sdk/src/reliable_channel/index.ts

@@ -1,2 +1,8 @@
 export { ReliableChannel, ReliableChannelOptions } from "./reliable_channel.js";
 export { ReliableChannelEvents, ReliableChannelEvent } from "./events.js";
+export {
+  StatusEvent,
+  StatusEvents,
+  StatusDetail,
+  SyncStatus
+} from "./sync_status.js";

packages/sdk/src/reliable_channel/reliable_channel.ts

@@ -33,6 +33,7 @@ import {
 import { ReliableChannelEvent, ReliableChannelEvents } from "./events.js";
 import { MissingMessageRetriever } from "./missing_message_retriever.js";
 import { RetryManager } from "./retry_manager.js";
+import { SyncStatus } from "./sync_status.js";
 
 const log = new Logger("sdk:reliable-channel");
 
@@ -234,8 +235,20 @@ export class ReliableChannel<
     }
 
     this._started = false;
+
+    this.syncStatus = new SyncStatus();
   }
 
+  /**
+   * Emit events when the channel is aware of missing message.
+   * Note that "syncd" may mean some messages are irretrievably lost.
+   * Check the emitted data for details.
+   *
+   * @emits [[StatusEvents]]
+   *
+   */
+  public syncStatus: SyncStatus;
+
   public get isStarted(): boolean {
     return this._started;
   }
@@ -709,6 +722,7 @@ export class ReliableChannel<
     this.addTrackedEventListener(
       MessageChannelEvent.InMessageReceived,
       (event) => {
+        this.syncStatus.onMessagesReceived(event.detail.messageId);
         // restart the timeout when a content message has been received
         if (isContentMessage(event.detail)) {
           // send a sync message faster to ack someone's else
@@ -730,6 +744,10 @@ export class ReliableChannel<
     this.addTrackedEventListener(
       MessageChannelEvent.InMessageMissing,
       (event) => {
+        this.syncStatus.onMessagesMissing(
+          ...event.detail.map((m) => m.messageId)
+        );
+
         for (const { messageId, retrievalHint } of event.detail) {
           if (retrievalHint && this.missingMessageRetriever) {
             this.missingMessageRetriever.addMissingMessage(
@@ -741,6 +759,13 @@ export class ReliableChannel<
       }
     );
 
+    this.messageChannel.addEventListener(
+      MessageChannelEvent.InMessageLost,
+      (event) => {
+        this.syncStatus.onMessagesLost(...event.detail.map((m) => m.messageId));
+      }
+    );
+
     if (this.queryOnConnect) {
       const queryListener = (event: any): void => {
         void this.processIncomingMessages(event.detail);

packages/sdk/src/reliable_channel/reliable_channel_sync_status.spec.ts

@@ -0,0 +1,224 @@
+import { TypedEventEmitter } from "@libp2p/interface";
+import { createDecoder, createEncoder } from "@waku/core";
+import {
+  AutoSharding,
+  IDecodedMessage,
+  IDecoder,
+  IEncoder
+} from "@waku/interfaces";
+import {
+  createRoutingInfo,
+  delay,
+  MockWakuEvents,
+  MockWakuNode
+} from "@waku/utils";
+import { utf8ToBytes } from "@waku/utils/bytes";
+import { expect } from "chai";
+import { beforeEach, describe } from "mocha";
+
+import { ReliableChannel, StatusDetail } from "./index.js";
+
+const TEST_CONTENT_TOPIC = "/my-tests/0/topic-name/proto";
+const TEST_NETWORK_CONFIG: AutoSharding = {
+  clusterId: 0,
+  numShardsInCluster: 1
+};
+const TEST_ROUTING_INFO = createRoutingInfo(TEST_NETWORK_CONFIG, {
+  contentTopic: TEST_CONTENT_TOPIC
+});
+
+describe("Status", () => {
+  let encoder: IEncoder;
+  let decoder: IDecoder<IDecodedMessage>;
+
+  beforeEach(async () => {
+    encoder = createEncoder({
+      contentTopic: TEST_CONTENT_TOPIC,
+      routingInfo: TEST_ROUTING_INFO
+    });
+    decoder = createDecoder(TEST_CONTENT_TOPIC, TEST_ROUTING_INFO);
+  });
+
+  it("Synced status is emitted when a message is received", async () => {
+    const commonEventEmitter = new TypedEventEmitter<MockWakuEvents>();
+    const mockWakuNodeAlice = new MockWakuNode(commonEventEmitter);
+    const mockWakuNodeBob = new MockWakuNode(commonEventEmitter);
+
+    const reliableChannelAlice = await ReliableChannel.create(
+      mockWakuNodeAlice,
+      "MyChannel",
+      "alice",
+      encoder,
+      decoder
+    );
+    const reliableChannelBob = await ReliableChannel.create(
+      mockWakuNodeBob,
+      "MyChannel",
+      "bob",
+      encoder,
+      decoder
+    );
+
+    let statusDetail: StatusDetail;
+    reliableChannelBob.syncStatus.addEventListener("synced", (event) => {
+      statusDetail = event.detail;
+    });
+
+    const message = utf8ToBytes("message in channel");
+
+    reliableChannelAlice.send(message);
+    while (!statusDetail!) {
+      await delay(50);
+    }
+
+    expect(statusDetail.received).to.eq(1);
+  });
+
+  it("Synced status is emitted when a missing message is received", async () => {
+    const commonEventEmitter = new TypedEventEmitter<MockWakuEvents>();
+    const mockWakuNodeAlice = new MockWakuNode(commonEventEmitter);
+    const mockWakuNodeBob = new MockWakuNode(commonEventEmitter);
+
+    const reliableChannelAlice = await ReliableChannel.create(
+      mockWakuNodeAlice,
+      "MyChannel",
+      "alice",
+      encoder,
+      decoder,
+      {
+        retryIntervalMs: 300 // shorter retry so that it resends message in test
+      }
+    );
+
+    // Send a message before Bob goes online so it's marked as missing
+    let messageSent = false;
+    reliableChannelAlice.addEventListener("message-sent", (_event) => {
+      messageSent = true;
+    });
+    reliableChannelAlice.send(utf8ToBytes("missing message"));
+    while (!messageSent) {
+      await delay(50);
+    }
+
+    const reliableChannelBob = await ReliableChannel.create(
+      mockWakuNodeBob,
+      "MyChannel",
+      "bob",
+      encoder,
+      decoder
+    );
+
+    let syncingStatusDetail: StatusDetail;
+    reliableChannelBob.syncStatus.addEventListener("syncing", (event) => {
+      syncingStatusDetail = event.detail;
+    });
+
+    let syncedStatusDetail: StatusDetail;
+    reliableChannelBob.syncStatus.addEventListener("synced", (event) => {
+      syncedStatusDetail = event.detail;
+    });
+
+    messageSent = false;
+    reliableChannelAlice.addEventListener("message-sent", (_event) => {
+      messageSent = true;
+    });
+    reliableChannelAlice.send(
+      utf8ToBytes("second message with missing message as dep")
+    );
+    while (!messageSent) {
+      await delay(50);
+    }
+
+    while (!syncingStatusDetail!) {
+      await delay(50);
+    }
+
+    expect(syncingStatusDetail.missing).to.eq(1);
+    expect(syncingStatusDetail.received).to.eq(1);
+
+    while (!syncedStatusDetail!) {
+      await delay(50);
+    }
+
+    expect(syncedStatusDetail.missing).to.eq(0);
+    expect(syncedStatusDetail.received).to.eq(2);
+  });
+
+  it("Synced status is emitted when a missing message is mark as lost", async () => {
+    const commonEventEmitter = new TypedEventEmitter<MockWakuEvents>();
+    const mockWakuNodeAlice = new MockWakuNode(commonEventEmitter);
+    const mockWakuNodeBob = new MockWakuNode(commonEventEmitter);
+
+    const reliableChannelAlice = await ReliableChannel.create(
+      mockWakuNodeAlice,
+      "MyChannel",
+      "alice",
+      encoder,
+      decoder,
+      {
+        syncMinIntervalMs: 0,
+        retryIntervalMs: 0 // Do not retry so we can lose the message
+      }
+    );
+
+    // Send a message before Bob goes online so it's marked as missing
+    let messageSent = false;
+    reliableChannelAlice.addEventListener("message-sent", (_event) => {
+      messageSent = true;
+    });
+    reliableChannelAlice.send(utf8ToBytes("missing message"));
+    while (!messageSent) {
+      await delay(50);
+    }
+
+    const reliableChannelBob = await ReliableChannel.create(
+      mockWakuNodeBob,
+      "MyChannel",
+      "bob",
+      encoder,
+      decoder,
+      {
+        retrieveFrequencyMs: 0,
+        syncMinIntervalMs: 0,
+        sweepInBufIntervalMs: 50, // frequently sweep incoming buffer to mark msg as lost
+        timeoutForLostMessagesMs: 200 // timeout within the test
+      }
+    );
+
+    let syncingStatusDetail: StatusDetail;
+    reliableChannelBob.syncStatus.addEventListener("syncing", (event) => {
+      syncingStatusDetail = event.detail;
+    });
+
+    messageSent = false;
+    reliableChannelAlice.addEventListener("message-sent", (_event) => {
+      messageSent = true;
+    });
+    reliableChannelAlice.send(
+      utf8ToBytes("second message with missing message as dep")
+    );
+    while (!messageSent) {
+      await delay(50);
+    }
+
+    while (!syncingStatusDetail!) {
+      await delay(50);
+    }
+
+    expect(syncingStatusDetail.missing).to.eq(1, "at first, one missing");
+    expect(syncingStatusDetail.received).to.eq(1, "at first, one received");
+    expect(syncingStatusDetail.lost).to.eq(0, "at first, no loss");
+
+    let syncedStatusDetail: StatusDetail;
+    reliableChannelBob.syncStatus.addEventListener("synced", (event) => {
+      syncedStatusDetail = event.detail;
+    });
+    while (!syncedStatusDetail!) {
+      await delay(50);
+    }
+
+    expect(syncedStatusDetail.missing).to.eq(0, "no more missing message");
+    expect(syncedStatusDetail.received).to.eq(1, "still one received message");
+    expect(syncedStatusDetail.lost).to.eq(1, "missing message is marked lost");
+  });
+});