feat(sds): send and receive sync messages

Author: adklempner

packages/sds/src/sds.spec.ts

@@ -471,4 +471,80 @@ describe("MessageChannel", function () {
       expect(possiblyAcknowledged.length).to.equal(messagesA.length);
     });
   });
+
+  describe("Sync messages", () => {
+    beforeEach(() => {
+      channelA = new MessageChannel(channelId);
+      channelB = new MessageChannel(channelId);
+    });
+
+    it("should be sent with empty content", () => {
+      channelA.sendSyncMessage((message) => {
+        expect(message.content?.length).to.equal(0);
+        return true;
+      });
+    });
+
+    it("should not be added to outgoing buffer, bloom filter, or local log", () => {
+      channelA.sendSyncMessage();
+
+      const outgoingBuffer = (channelA as any).outgoingBuffer as Message[];
+      expect(outgoingBuffer.length).to.equal(0);
+
+      const bloomFilter = getBloomFilter(channelA);
+      expect(
+        bloomFilter.lookup(MessageChannel.getMessageId(new Uint8Array()))
+      ).to.equal(false);
+
+      const localLog = (channelA as any).messageIdLog as {
+        timestamp: number;
+        messageId: string;
+      }[];
+      expect(localLog.length).to.equal(0);
+    });
+
+    it("should be delivered but not added to local log or bloom filter", () => {
+      const timestampBefore = (channelB as any).lamportTimestamp;
+      let expectedTimestamp: number | undefined;
+      channelA.sendSyncMessage((message) => {
+        expectedTimestamp = message.lamportTimestamp;
+        channelB.receiveMessage(message);
+        return true;
+      });
+      const timestampAfter = (channelB as any).lamportTimestamp;
+      expect(timestampAfter).to.equal(expectedTimestamp);
+      expect(timestampAfter).to.be.greaterThan(timestampBefore);
+
+      const localLog = (channelB as any).messageIdLog as {
+        timestamp: number;
+        messageId: string;
+      }[];
+      expect(localLog.length).to.equal(0);
+
+      const bloomFilter = getBloomFilter(channelB);
+      expect(
+        bloomFilter.lookup(MessageChannel.getMessageId(new Uint8Array()))
+      ).to.equal(false);
+    });
+
+    it("should update ack status of messages in outgoing buffer", () => {
+      messagesA.forEach((m) => {
+        channelA.sendMessage(utf8ToBytes(m), (message) => {
+          channelB.receiveMessage(message);
+          return true;
+        });
+      });
+
+      channelB.sendSyncMessage((message) => {
+        channelA.receiveMessage(message);
+        return true;
+      });
+
+      const causalHistorySize = (channelA as any).causalHistorySize;
+      const outgoingBuffer = (channelA as any).outgoingBuffer as Message[];
+      expect(outgoingBuffer.length).to.equal(
+        messagesA.length - causalHistorySize
+      );
+    });
+  });
 });

packages/sds/src/sds.ts

@@ -112,8 +112,10 @@ export class MessageChannel {
   public receiveMessage(message: Message): void {
     // review ack status
     this.reviewAckStatus(message);
-    // add to bloom filter
-    this.filter.insert(message.messageId);
+    // add to bloom filter (skip for messages with empty content)
+    if (message.content?.length && message.content.length > 0) {
+      this.filter.insert(message.messageId);
+    }
     // verify causal history
     const dependenciesMet = message.causalHistory.every((messageId) =>
       this.messageIdLog.some(
@@ -203,13 +205,51 @@ export class MessageChannel {
     );
   }
 
+  /**
+   * Send a sync message to the SDS channel.
+   *
+   * Increments the lamport timestamp, constructs a `Message` object
+   * with an empty load. Skips outgoing buffer, filter, and local log.
+   *
+   * See https://rfc.vac.dev/vac/raw/sds/#send-sync-message
+   *
+   * @param callback - A callback function that returns a boolean indicating whether the message was sent successfully.
+   */
+  public sendSyncMessage(callback?: (message: Message) => boolean): boolean {
+    this.lamportTimestamp++;
+
+    const emptyMessage = new Uint8Array();
+
+    const message: Message = {
+      messageId: MessageChannel.getMessageId(emptyMessage),
+      channelId: this.channelId,
+      lamportTimestamp: this.lamportTimestamp,
+      causalHistory: this.messageIdLog
+        .slice(-this.causalHistorySize)
+        .map(({ messageId }) => messageId),
+      bloomFilter: this.filter.toBytes(),
+      content: emptyMessage
+    };
+
+    if (callback) {
+      return callback(message);
+    }
+    return false;
+  }
+
   // See https://rfc.vac.dev/vac/raw/sds/#deliver-message
   private deliverMessage(message: Message): void {
     const messageLamportTimestamp = message.lamportTimestamp ?? 0;
     if (messageLamportTimestamp > this.lamportTimestamp) {
       this.lamportTimestamp = messageLamportTimestamp;
     }
 
+    if (message.content?.length === 0) {
+      // Messages with empty content are sync messages.
+      // They are not added to the local log or bloom filter.
+      return;
+    }
+
     // The participant MUST insert the message ID into its local log,
     // based on Lamport timestamp.
     // If one or more message IDs with the same Lamport timestamp already exists,
@@ -227,6 +267,8 @@ export class MessageChannel {
     });
   }
 
+  // For each received message (including sync messages), inspect the causal history and bloom filter
+  // to determine the acknowledgement status of messages in the outgoing buffer.
   // See https://rfc.vac.dev/vac/raw/sds/#review-ack-status
   private reviewAckStatus(receivedMessage: Message): void {
     // the participant MUST mark all messages in the received causal_history as acknowledged.