From 59ff9d2342d6bb63b04aa2b62582b99c312c223a Mon Sep 17 00:00:00 2001
From: Jake Smith <jake.smith@lexisnexisrisk.com>
Date: Tue, 18 Jun 2024 18:37:55 +0100
Subject: [PATCH] HPCC-31807 Remove stale job queue client entries

Signed-off-by: Jake Smith <jake.smith@lexisnexisrisk.com>
---
 common/workunit/wujobq.cpp | 12 +++++++++---
 dali/base/dasds.cpp        | 13 +++++++++++++
 2 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/common/workunit/wujobq.cpp b/common/workunit/wujobq.cpp
index 0a5bd0576a1..f7b5d9c286d 100644
--- a/common/workunit/wujobq.cpp
+++ b/common/workunit/wujobq.cpp
@@ -1122,11 +1122,17 @@ class CJobQueue: public CJobQueueBase, implements IJobQueue
     void disconnect()   // signal no longer wil be dequeing (optional - done automatically on release)
     {
         Cconnlockblock block(this,true);
-        if (connected) {
+        if (connected)
+        {
             dounsubscribe();
-            ForEachQueue(qd) {
+            ForEachQueue(qd)
+            {
                 IPropertyTree *croot = queryClientRootSession(*qd);
-                croot->setPropInt64("@connected",croot->getPropInt64("@connected",0)-1);
+                unsigned newConnectedCount = croot->getPropInt64("@connected",0)-1;
+                if (newConnectedCount)
+                    croot->setPropInt64("@connected", newConnectedCount);
+                else
+                    qd->root->removeTree(croot);
             }
             connected = false;
         }
diff --git a/dali/base/dasds.cpp b/dali/base/dasds.cpp
index 9478cbe2665..7257522fc03 100644
--- a/dali/base/dasds.cpp
+++ b/dali/base/dasds.cpp
@@ -4938,6 +4938,18 @@ void initializeInternals(IPropertyTree *root)
     root->addPropTree("Status/Servers",createPTree());
 }
 
+void clearStaleMeteData(IPropertyTree *root)
+{
+    // JobQueues
+    // Remove all Client entries from all queues. By definition they are stale (they should normally be removed when the client disconnects)
+    Owned<IPropertyTreeIterator> jobQueues = root->getElements("JobQueues/Queue");
+    ForEach(*jobQueues)
+    {
+        IPropertyTree &queue = jobQueues->query();
+        while (queue.removeProp("Client"));
+    }
+}
+
 IPropertyTree *loadStore(const char *storeFilename, unsigned edition, IPTreeMaker *iMaker, unsigned crcValidation, bool logErrorsOnly=false, const bool *abort=NULL)
 {
     CHECKEDCRITICALBLOCK(loadStoreCrit, fakeCritTimeout);
@@ -6543,6 +6555,7 @@ void CCovenSDSManager::loadStore(const char *storeName, const bool *abort)
     }
     Owned<IRemoteConnection> conn = connect("/", 0, RTM_INTERNAL, INFINITE);
     initializeInternals(conn->queryRoot());
+    clearStaleMeteData(conn->queryRoot());
     conn.clear();
     initializeStorageGroups(oldEnvironment);
 }