Merge pull request #4822 from Countly/SER-1069-fail-and-infinity-bug

[SER-1069] [push] Push: Some messages fail and show infinity on detail

Author: coskunaydinoglu

plugins/push/api/jobs/util/resultor.js

@@ -323,7 +323,7 @@ class Resultor extends DoFinish {
 
                 if (this.data.isSending(m.id)) {
                     this.log.d('message %s is still in processing (%d out of %d)', m.id, m.result.processed, m.result.total);
-                    return m.save();
+                    return syncTotalProp(this.db, m).then(() => m.save());
                 }
                 this.log.d('message %s is done processing', m.id);
                 let state, status, error;
@@ -392,11 +392,11 @@ class Resultor extends DoFinish {
                     if (error) {
                         m.result.error = error;
                     }
-                    return m.save();
+                    return syncTotalProp(this.db, m).then(() => m.save());
                 }
                 else {
                     this.log.d('message %s is in processing (%d out of %d)', m.id, m.result.processed, m.result.total);
-                    return m.save();
+                    return syncTotalProp(this.db, m).then(() => m.save());
                 }
             }).concat(Object.keys(this.noMessage).map(mid => {
                 this.log.e('Message %s doesn\'t exist, ignoring result %j', mid, this.noMessage[mid]);
@@ -822,4 +822,35 @@ class Resultor extends DoFinish {
     // }
 }
 
-module.exports = { Resultor };
+module.exports = { Resultor };
+/**
+ * This function is required to be called before making any updates to the "message".
+ * It is implemented because of the race condition between the Job and the Runner.
+ * Sometimes Job starts creating "push" records (queue items) but the Runner
+ * starts consuming before all of the push records created. Then the race condition
+ * occurs and the "total" property gets overwritten by the Runner.
+ * @param {MongoClient} db mongo client
+ * @param {Message} message message model instance
+ */
+async function syncTotalProp(db, message) {
+    const saved = await db.collection("messages").findOne({ _id: message._id });
+
+    // fields to sync:
+    // m.result.total
+    // m.result.subs.a.total
+    // m.result.subs.a.subs.en.total
+
+    if (saved?.result?.total !== undefined) {
+        message.result.total = saved.result.total;
+    }
+    if (typeof saved?.result?.subs === "object") {
+        for (let key in saved.result.subs) {
+            message.result.subs[key].total = saved.result.subs[key].total;
+            if (typeof saved.result.subs[key]?.subs === "object") {
+                for (let key2 in saved.result.subs[key].subs) {
+                    message.result.subs[key].subs[key2].total = saved.result.subs[key].subs[key2].total;
+                }
+            }
+        }
+    }
+}