fix: prevent lost iterations in CircularKey concurrent worker race

Two workers sharing a DocumentGenerator could both detect end-of-pass
(updateItr >= update_e) before either called resetUpdate(), causing
double-decrements of the plain-int iterations counter in CircularKey.

Each race event silently dropped one full pass, producing fewer total
mutations than expected (observed: 54204 vs 60000 ep_dcp_items_sent
in CDC history retention tests).

Add double-checked locking (synchronized on this.keys) in
has_next_read(), has_next_update(), and has_next_expiry() so only one
worker triggers checkIterations() + reset per pass-end event.

The fast unsynchronized path is preserved for normal in-pass execution,
keeping full worker concurrency for deduplication-sensitive tests.

Author: ShivamPaliwal1

src/main/java/utils/docgen/DocumentGenerator.java

@@ -211,13 +211,17 @@ public boolean has_next_read() {
         if (this.ws.dr.readItr.get() < this.ws.dr.read_e)
             return true;
         if (this.keyInstance.getSimpleName().equals(CircularKey.class.getSimpleName())) {
-            try {
-                if ((boolean)this.iterationsMethod.invoke(this.keys)) {
-                    this.resetRead();
+            synchronized (this.keys) {
+                if (this.ws.dr.readItr.get() < this.ws.dr.read_e)
                     return true;
+                try {
+                    if ((boolean)this.iterationsMethod.invoke(this.keys)) {
+                        this.resetRead();
+                        return true;
+                    }
+                } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
+                    e1.printStackTrace();
                 }
-            } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
-                e1.printStackTrace();
             }
         }
         return false;
@@ -226,21 +230,25 @@ public boolean has_next_read() {
     public boolean has_next_update() {
         if (this.ws.dr.updateItr.get() < this.ws.dr.update_e)
             return true;
-        if (this.keyInstance.getSimpleName().equals(CircularKey.class.getSimpleName())) {
-            try {
-                if ((boolean)this.iterationsMethod.invoke(this.keys)) {
-                    this.resetUpdate();
-                    this.ws.mutated += 1;
-                    return true;
+        synchronized (this.keys) {
+            if (this.ws.dr.updateItr.get() < this.ws.dr.update_e)
+                return true;
+            if (this.keyInstance.getSimpleName().equals(CircularKey.class.getSimpleName())) {
+                try {
+                    if ((boolean)this.iterationsMethod.invoke(this.keys)) {
+                        this.resetUpdate();
+                        this.ws.mutated += 1;
+                        return true;
+                    }
+                } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
+                    e1.printStackTrace();
                 }
-            } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
-                e1.printStackTrace();
             }
-        }
-        if (TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())-startTime<ws.mutation_timeout) {
-            this.resetUpdate();
-            this.ws.mutated += 1;
-            return true;
+            if (TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())-startTime<ws.mutation_timeout) {
+                this.resetUpdate();
+                this.ws.mutated += 1;
+                return true;
+            }
         }
         return false;
     }
@@ -249,13 +257,17 @@ public boolean has_next_expiry() {
         if (this.ws.dr.expiryItr.get() < this.ws.dr.expiry_e)
             return true;
         if (this.keyInstance.getSimpleName().equals(CircularKey.class.getSimpleName())) {
-            try {
-                if ((boolean)this.iterationsMethod.invoke(this.keys, null)) {
-                    this.resetExpiry();
+            synchronized (this.keys) {
+                if (this.ws.dr.expiryItr.get() < this.ws.dr.expiry_e)
                     return true;
+                try {
+                    if ((boolean)this.iterationsMethod.invoke(this.keys)) {
+                        this.resetExpiry();
+                        return true;
+                    }
+                } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
+                    e1.printStackTrace();
                 }
-            } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e1) {
-                e1.printStackTrace();
             }
         }
         return false;
@@ -311,7 +323,7 @@ public boolean has_next_subdoc_lookup() {
     abstract Tuple2<String, Object> next();
 
     void resetRead() {
-        this.ws.dr.readItr =  new AtomicLong(this.ws.dr.read_s);
+        this.ws.dr.readItr.set(this.ws.dr.read_s);
     }
 
     void resetExpiry() {