stream-management: Implement requesting ACKs (#1054)

---

Co-authored-by: Stephen Paul Weber <singpolyma@singpolyma.net>

packages/client-core/src/bind2/bind2.test.js

@@ -66,7 +66,6 @@ test("with function resource returning string", async () => {
 test("with function resource throwing", async () => {
   const error = new Error("foo");
 
-
   function resource() {
     throw error;
   }
@@ -102,7 +101,6 @@ test("with function resource returning resolved promise", async () => {
 test("with function resource returning rejected promise", async () => {
   const error = new Error("foo");
 
-
   async function resource() {
     throw error;
   }

packages/stream-management/README.md

@@ -10,7 +10,50 @@ When the session is resumed the `online` event is not emitted as session resumpt
 However `entity.status` is set to `online`.
 If the session fails to resume, entity will fallback to regular session establishment in which case `online` event will be emitted.
 
-Automatically responds to acks but does not support requesting acks yet.
+- Automatically responds to acks.
+- Periodically request acks.
+- If server fails to respond, triggers a reconnect.
+
+## Events
+
+### resumed
+
+Indicates that the connection was resumed. When that happens the `online` event is not emitted but `xmpp.status` will be `online`.
+
+```js
+const xmpp = client(...);
+const {streamManagement} = xmpp;
+
+streamManagement.on('resumed', () => {
+  console.log("session resumed");
+});
+```
+
+### fail
+
+Indicates that a stanza failed to send to the server and will not be retried.
+
+```js
+const xmpp = client(...);
+const {streamManagement} = xmpp;
+
+streamManagement.on('fail', (stanza) => {
+  console.log("fail to send", stanza.toString());
+});
+```
+
+### ack
+
+Indicates that a stanza has been acknowledged by the server.
+
+```js
+const xmpp = client(...);
+const {streamManagement} = xmpp;
+
+streamManagement.on('ack', (stanza) => {
+  console.log("stanza acknowledge by the server", stanza.toString());
+});
+```
 
 ## References
 

packages/stream-management/index.js

@@ -1,6 +1,7 @@
 import XMPPError from "@xmpp/error";
-import { procedure } from "@xmpp/events";
+import { EventEmitter, procedure } from "@xmpp/events";
 import xml from "@xmpp/xml";
+import { datetime } from "@xmpp/time";
 
 // https://xmpp.org/extensions/xep-0198.html
 
@@ -45,24 +46,52 @@ export default function streamManagement({
   bind2,
   sasl2,
 }) {
-  const sm = {
+  let timeoutTimeout = null;
+  let requestAckTimeout = null;
+
+  const sm = new EventEmitter();
+  Object.assign(sm, {
     allowResume: true,
     preferredMaximum: null,
     enabled: false,
     id: "",
+    outbound_q: [],
     outbound: 0,
     inbound: 0,
     max: null,
-  };
+    timeout: 60_000,
+    requestAckInterval: 300_000,
+    debounceAckRequest: 100,
+  });
+
+  entity.on("disconnect", () => {
+    clearTimeout(timeoutTimeout);
+    clearTimeout(requestAckTimeout);
+  });
 
-  function resumed() {
+  async function resumed(resumed) {
     sm.enabled = true;
+    const oldOutbound = sm.outbound;
+    for (let i = 0; i < resumed.attrs.h - oldOutbound; i++) {
+      let item = sm.outbound_q.shift();
+      sm.outbound++;
+      sm.emit("ack", item.stanza);
+    }
+    let q = sm.outbound_q;
+    sm.outbound_q = [];
+    // This will trigger the middleware and re-add to the queue
+    await entity.sendMany(q.map((item) => queueToStanza({ entity, item })));
+    sm.emit("resumed");
     entity._ready(true);
   }
 
   function failed() {
     sm.enabled = false;
     sm.id = "";
+    let item;
+    while ((item = sm.outbound_q.shift())) {
+      sm.emit("fail", item.stanza);
+    }
     sm.outbound = 0;
   }
 
@@ -73,11 +102,20 @@ export default function streamManagement({
   }
 
   entity.on("online", () => {
+    if (sm.outbound_q.length > 0) {
+      throw new Error(
+        "Stream Management assertion failure, queue should be empty during online",
+      );
+    }
     sm.outbound = 0;
     sm.inbound = 0;
   });
 
   entity.on("offline", () => {
+    let item;
+    while ((item = sm.outbound_q.shift())) {
+      sm.emit("fail", item.stanza);
+    }
     sm.outbound = 0;
     sm.inbound = 0;
     sm.enabled = false;
@@ -86,14 +124,20 @@ export default function streamManagement({
 
   middleware.use((context, next) => {
     const { stanza } = context;
+    clearTimeout(timeoutTimeout);
     if (["presence", "message", "iq"].includes(stanza.name)) {
       sm.inbound += 1;
     } else if (stanza.is("r", NS)) {
       // > When an <r/> element ("request") is received, the recipient MUST acknowledge it by sending an <a/> element to the sender containing a value of 'h' that is equal to the number of stanzas handled by the recipient of the <r/> element.
       entity.send(xml("a", { xmlns: NS, h: sm.inbound })).catch(() => {});
     } else if (stanza.is("a", NS)) {
       // > When a party receives an <a/> element, it SHOULD keep a record of the 'h' value returned as the sequence number of the last handled outbound stanza for the current stream (and discard the previous value).
-      sm.outbound = stanza.attrs.h;
+      const oldOutbound = sm.outbound;
+      for (let i = 0; i < stanza.attrs.h - oldOutbound; i++) {
+        let item = sm.outbound_q.shift();
+        sm.outbound++;
+        sm.emit("ack", item.stanza);
+      }
     }
 
     return next();
@@ -105,6 +149,33 @@ export default function streamManagement({
   if (sasl2) {
     setupSasl2({ sasl2, sm, failed, resumed });
   }
+
+  function requestAck() {
+    clearTimeout(timeoutTimeout);
+    if (sm.timeout) {
+      timeoutTimeout = setTimeout(
+        () => entity.disconnect().catch(() => {}),
+        sm.timeout,
+      );
+    }
+    entity.send(xml("r", { xmlns: NS })).catch(() => {});
+    // Periodically send r to check the connection
+    // If a stanza goes out it will cancel this and set a sooner timer
+    requestAckTimeout = setTimeout(requestAck, sm.requestAckInterval);
+  }
+
+  middleware.filter((context, next) => {
+    if (!sm.enabled) return next();
+    const { stanza } = context;
+    if (!["presence", "message", "iq"].includes(stanza.name)) return next();
+
+    sm.outbound_q.push({ stanza, stamp: datetime() });
+    // Debounce requests so we send only one after a big run of stanza together
+    clearTimeout(requestAckTimeout);
+    requestAckTimeout = setTimeout(requestAck, sm.debounceAckRequest);
+    return next();
+  });
+
   if (streamFeatures) {
     setupStreamFeature({
       streamFeatures,
@@ -133,8 +204,8 @@ function setupStreamFeature({
     // Resuming
     if (sm.id) {
       try {
-        await resume(entity, sm);
-        resumed();
+        const element = await resume(entity, sm);
+        await resumed(element);
         return;
         // If resumption fails, continue with session establishment
       } catch {
@@ -149,6 +220,12 @@ function setupStreamFeature({
 
     const promiseEnable = enable(entity, sm);
 
+    if (sm.outbound_q.length > 0) {
+      throw new Error(
+        "Stream Management assertion failure, queue should be empty after enable",
+      );
+    }
+
     // > The counter for an entity's own sent stanzas is set to zero and started after sending either <enable/> or <enabled/>.
     sm.outbound = 0;
 
@@ -172,7 +249,7 @@ function setupSasl2({ sasl2, sm, failed, resumed }) {
     },
     (element) => {
       if (element.is("resumed")) {
-        resumed();
+        resumed(element);
       } else if (element.is(failed)) {
         // const error = StreamError.fromElement(element)
         failed();
@@ -198,3 +275,20 @@ function setupBind2({ bind2, sm, failed, enabled }) {
     },
   );
 }
+
+function queueToStanza({ entity, item }) {
+  const { stanza, stamp } = item;
+  if (
+    stanza.name === "message" &&
+    !stanza.getChild("delay", "urn:xmpp:delay")
+  ) {
+    stanza.append(
+      xml("delay", {
+        xmlns: "urn:xmpp:delay",
+        from: entity.jid.toString(),
+        stamp,
+      }),
+    );
+  }
+  return stanza;
+}

packages/stream-management/package.json

@@ -16,7 +16,8 @@
   "dependencies": {
     "@xmpp/error": "^0.14.0",
     "@xmpp/events": "^0.14.0",
-    "@xmpp/xml": "^0.14.0"
+    "@xmpp/xml": "^0.14.0",
+    "@xmpp/time": "^0.14.0"
   },
   "engines": {
     "node": ">= 20.10"