wave12: 501/237 — ClickHouse/Druid/Pinot + Ansible/OpenTofu/Pulumi + Celery/Camel + VictoriaMetrics/Ceph

defects/celery/patch/celery-0001-result-set-quadratic-membership.md

@@ -0,0 +1,76 @@
+# celery-0001 — O(N²) ResultSet Membership Test in update()/add()
+
+**Severity:** MEDIUM
+**Complexity:** O(N²) → O(N)
+**CWE:** CWE-407 (Algorithmic Complexity)
+
+## Affected File
+
+| File | Lines | Notes |
+|------|-------|-------|
+| `celery/result.py` | 597–598, 629–631 | ResultSet.add() and ResultSet.update() |
+
+## Defective Code
+
+### `celery/result.py` line 597–598 — `ResultSet.add()`
+
+```python
+def add(self, result):
+    if result not in self.results:   # O(N) list scan
+        self.results.append(result)
+```
+
+### `celery/result.py` line 629–631 — `ResultSet.update()`
+
+```python
+def update(self, results):
+    """Extend from iterable of results."""
+    self.results.extend(r for r in results if r not in self.results)
+    #                                          ^^^^^^^^^^^^^^^^^^^
+    #                                          O(N) scan per element → O(M*N)
+```
+
+`self.results` is a plain `list` (see line 586: `self.results = results`).
+Each `r not in self.results` scan is O(N). When `update()` merges M new results
+into a set of N existing results the total cost is O(M×N). For chord groups with
+thousands of tasks this becomes the bottleneck.
+
+## Root Cause
+
+`ResultSet.results` stores `AsyncResult` objects in a `list`. Deduplication uses
+linear scan (`not in list`) instead of a `set`/`dict` lookup.
+
+## Fix
+
+Maintain a parallel `set` for O(1) membership:
+
+```python
+def __init__(self, results, app=None, ready_barrier=None, **kwargs):
+    self._app = app
+    self.results = results
+    self._result_ids = {r.id for r in results}   # shadow set for O(1) lookup
+    ...
+
+def add(self, result):
+    if result.id not in self._result_ids:
+        self._result_ids.add(result.id)
+        self.results.append(result)
+        if self._on_full:
+            self._on_full.add(result)
+
+def update(self, results):
+    for r in results:
+        if r.id not in self._result_ids:
+            self._result_ids.add(r.id)
+            self.results.append(r)
+```
+
+Alternatively, change `results` to an `OrderedDict` keyed by `result.id`.
+
+## Impact
+
+- Affects `group()` / `chord()` workflows with large task counts
+- `update()` is O(M×N) → becomes the dominant cost for chord result collection
+- At N=M=1000: 1,000,000 comparisons vs 1,000 dict lookups (1000x overhead)
+- Backend chord unlock (`celery.chord_unlock`) iterates `header_result.results`
+  which triggers the O(N²) path when merging partial completions

defects/celery/unit/ResultSetAlgorithm.java

@@ -0,0 +1,150 @@
+package unit;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * celery-0001: ResultSet.update() / ResultSet.add()
+ *
+ * Defect: self.results is a plain list. The expression
+ *   'r not in self.results'  (Python)
+ * becomes an O(N) scan on each of M elements in update() → O(M*N) total.
+ *
+ * Fix: maintain a parallel dict keyed by result ID for O(1) membership.
+ *
+ * Op counting: each comparison during the list scan counts as one op.
+ * For hash lookup we count 1 op per lookup.
+ */
+public class ResultSetAlgorithm {
+
+    // --- Slow path: simulate list.contains() with explicit scan counting ---
+
+    static long slowUpdate(int existingCount, int newCount) {
+        long ops = 0;
+        List<String> results = new ArrayList<>();
+
+        for (int i = 0; i < existingCount; i++) {
+            results.add("result-" + i);
+        }
+
+        // simulate: self.results.extend(r for r in results if r not in self.results)
+        for (int i = existingCount; i < existingCount + newCount; i++) {
+            String r = "result-" + i;
+            // Scan the list: worst case is a full pass (item not found)
+            for (String existing : results) {
+                ops++; // each comparison costs 1 op
+                if (existing.equals(r)) break;
+            }
+            // Item not in list, add it
+            results.add(r);
+        }
+        return ops;
+    }
+
+    // --- Fast path: LinkedHashMap (id→result) simulation ---
+
+    static long fastUpdate(int existingCount, int newCount) {
+        long ops = 0;
+        Map<String, String> resultMap = new LinkedHashMap<>();
+
+        for (int i = 0; i < existingCount; i++) {
+            String r = "result-" + i;
+            resultMap.put(r, r);
+        }
+
+        for (int i = existingCount; i < existingCount + newCount; i++) {
+            String r = "result-" + i;
+            ops++; // O(1) hash lookup counted as 1 op
+            if (!resultMap.containsKey(r)) {
+                resultMap.put(r, r);
+            }
+        }
+        return ops;
+    }
+
+    // --- Slow path: add() pattern ---
+    // Simulates: if result not in self.results: self.results.append(result)
+    // Called N times for sequential task completions in a chord group.
+
+    static long slowAdd(int N) {
+        long ops = 0;
+        List<String> results = new ArrayList<>();
+
+        for (int i = 0; i < N; i++) {
+            String r = "result-" + i;
+            // Scan entire list for membership check
+            for (String existing : results) {
+                ops++;
+                if (existing.equals(r)) break;
+            }
+            // Not found → append (all N items are unique, so full scan each time)
+            results.add(r);
+        }
+        return ops;
+    }
+
+    // --- Fast path: add() pattern ---
+
+    static long fastAdd(int N) {
+        long ops = 0;
+        Map<String, String> resultMap = new LinkedHashMap<>();
+
+        for (int i = 0; i < N; i++) {
+            String r = "result-" + i;
+            ops++; // O(1) lookup
+            resultMap.putIfAbsent(r, r);
+        }
+        return ops;
+    }
+
+    static void runTest(String label, int N, long slowOps, long fastOps) {
+        double ratio = (double) slowOps / fastOps;
+        boolean pass = ratio >= 10.0;
+        System.out.printf("  %-14s N=%-5d slow=%,8d fast=%,5d ratio=%6.1fx  %s%n",
+                label, N, slowOps, fastOps, ratio, pass ? "PASS" : "FAIL");
+        if (!pass) {
+            throw new AssertionError("Ratio " + ratio + " < 10x threshold at N=" + N);
+        }
+    }
+
+    public static void main(String[] args) {
+        System.out.println("celery-0001: ResultSetAlgorithm");
+        System.out.println("  Defect: list.contains() per element in update()/add() → O(N²)");
+        System.out.println("  Fix:    dict/map key lookup → O(N)");
+        System.out.println();
+
+        int[] sizes = {200, 500, 1000};
+        int passed = 0;
+        int total = sizes.length * 2;
+
+        for (int N : sizes) {
+            // Test update(): merge N new results into N existing
+            long slowU = slowUpdate(N, N);
+            long fastU = fastUpdate(N, N);
+            try {
+                runTest("update", N, slowU, fastU);
+                passed++;
+            } catch (AssertionError e) {
+                System.out.println("  FAIL update: " + e.getMessage());
+            }
+
+            // Test add(): add N items one at a time (each unique)
+            long slowA = slowAdd(N);
+            long fastA = fastAdd(N);
+            try {
+                runTest("add", N, slowA, fastA);
+                passed++;
+            } catch (AssertionError e) {
+                System.out.println("  FAIL add: " + e.getMessage());
+            }
+        }
+
+        System.out.println();
+        System.out.println(passed + "/" + total + " PASS");
+        if (passed < total) {
+            System.exit(1);
+        }
+    }
+}