HPCC-12625 subscription notifying erroneously on some changes

If there were >1 subscriber on a parent node, and 1 of the
subscribers was a subscriber with a change below it, but
unrelated to another child subscriber, both were were being
notified, e.g.
/a/b (sub=true)
/a/b/c (sub=false)

change to /a/b/d

/a/b AND /a/b/c were being notified.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.com>

dali/base/dasds.cpp

@@ -810,6 +810,7 @@ public:
 };
 
 typedef IArrayOf<CSubscriberContainer> CSubscriberArray;
+typedef ICopyArrayOf<CSubscriberContainer> CSubscriberCopyArray;
 class CSubscriberContainerList : public CInterface, public CSubscriberArray
 {
 public:
@@ -8179,13 +8180,19 @@ public:
             head++;
             if ('\0' == *path)
             {
-                if (!wild && '/' != *(path-1) && '/' != *head) return false;
-                sub = true;
+                if ('\0' == *head) // absolute match
+                {
+                    sub = false;
+                    return true;
+                }
+                else if (!wild && '/' != *head) // e.g. change=/a/bc, subscriber=/a/b
+                    return false;
+                sub = true; // e.g. change=/a/b/c, subscriber=/a/b
                 return true;
             }
             else 
             {
-                if ('\0' == *head)
+                if ('\0' == *head) // e.g. change=/a/b, subscriber=/a/b/c - not matched yet, but returning true keeps it from being pruned
                 {
                     sub = false;
                     return true;
@@ -8203,7 +8210,7 @@ public:
         scan(*rootChanges, stack, pruned);
     }
 
-    bool prune(const char *xpath, bool &sub, CSubscriberArray &pruned)
+    bool prune(const char *xpath, bool &sub, CSubscriberCopyArray *matches, CSubscriberArray &pruned)
     {
         sub = false;
         ForEachItemInRev(s, subs)
@@ -8215,8 +8222,10 @@ public:
                 pruned.append(*LINK(&subscriber));
                 subs.remove(s);
             }
-            else
-                sub |= _sub;
+            else if (_sub)
+                sub = true;
+            else if (matches)
+                matches->append(subscriber);
         }
         return (subs.ordinality() > 0);
     }
@@ -8225,7 +8234,7 @@ public:
     void scanAll(PDState state, CBranchChange &changes, CPTStack &stack, CSubscriberArray &pruned)
     {
         bool sub;
-        if (prune(xpath.str(), sub, pruned))
+        if (prune(xpath.str(), sub, NULL, pruned))
         {
             MemoryBuffer notifyData;
             if (sub)
@@ -8275,9 +8284,7 @@ public:
                         CBranchChange &childChange = changes.children.item(c);
                         PushPop pp(stack, *childChange.tree);
                         size32_t parentLength = xpath.length();
-                        xpath.append(childChange.tree->queryName());
-                        if ('/' != xpath.charAt(xpath.length()-1))
-                            xpath.append('/');
+                        xpath.append('/').append(childChange.tree->queryName());
                         CSubscriberArray _pruned;
                         scanAll(state, childChange, stack, _pruned);
                         ForEachItemIn(i, _pruned) subs.append(*LINK(&_pruned.item(i)));
@@ -8292,17 +8299,18 @@ public:
 
     void scan(CBranchChange &changes, CPTStack &stack, CSubscriberArray &pruned)
     {
+        CSubscriberCopyArray matches;
         bool sub;
-        if (!prune(xpath.str(), sub, pruned))
+        if (!prune(xpath.str(), sub, &matches, pruned))
             return;
-    
+
         PushPop pp(stack, *changes.tree);
         if (PDS_Deleted == (changes.local & PDS_Deleted))
         {
             scanAll(changes.local, changes, stack, pruned);
             return;
         }
-        else if (sub) // xpath matched some subscribers, and/or below some, need to check for sub subscribers
+        else if (matches.ordinality()) // xpath matched some subscribers, and/or below some, need to check for sub subscribers
         {
             bool ret = false;
             // avoid notifying on PDS_Structure only, which signifies changes deeper down only
@@ -8310,9 +8318,9 @@ public:
             if (changes.state && changes.local && (changes.local != PDS_Structure))
             {
                 int lastSendValue = -1;
-                ForEachItemInRev(s, subs)
+                ForEachItemInRev(s, matches)
                 {
-                    CSubscriberContainer &subscriber = subs.item(s);
+                    CSubscriberContainer &subscriber = matches.item(s);
                     if (!subscriber.isUnsubscribed())
                     {
                         if (subscriber.qualify(stack, false))
@@ -8340,7 +8348,7 @@ public:
                         else
                             pruned.append(*LINK(&subscriber));
                     }
-                    subs.remove(s);
+                    subs.zap(subscriber);
                 }
             }
             else
@@ -8363,18 +8371,13 @@ public:
         ForEachItemIn(c, changes.children)
         {
             CBranchChange &childChanges = changes.children.item(c);
-
             size32_t parentLength = xpath.length();
-            xpath.append(childChanges.tree->queryName());
-            if ('/' != xpath.charAt(xpath.length()-1))
-                xpath.append('/');
-
+            xpath.append('/').append(childChanges.tree->queryName());
             CSubscriberArray pruned;
             scan(childChanges, stack, pruned);
             ForEachItemIn(i, pruned) subs.append(*LINK(&pruned.item(i)));
             if (0 == subs.ordinality())
                 break;
-
             xpath.setLength(parentLength);
         }
     }

dali/base/dasds.ipp

@@ -209,9 +209,9 @@ public:
             loop
             {
                 str.append(item(i).queryName());
-                str.append('/');
                 if (++i >= ordinality())
                     break;
+                str.append('/');
             }
         }
         return str;

testing/regress/ecl-test.json

@@ -22,7 +22,7 @@
             "thor",
             "roxie"
         ],
-        "timeout":"720",
+        "timeout":"72000000",
         "maxAttemptCount":"3",
         "defaultSetupClusters": [
             "all"