diff --git a/defects/elasticsearch/unit/test_elasticsearch_scale.py b/defects/elasticsearch/unit/test_elasticsearch_scale.py
new file mode 100644
index 000000000..11e4afc17
--- /dev/null
+++ b/defects/elasticsearch/unit/test_elasticsearch_scale.py
@@ -0,0 +1,146 @@
+"""
+Elasticsearch CWE-407 scaling benchmark — projected wall-clock savings.
+
+Simulates bulk ingest with append dedup (elasticsearch-0001) at production scale.
+Each document has an array field with k existing values and receives N new values
+with ~50% overlap (realistic for tag/label append pipelines).
+
+The cost per document:
+  Before: O(N×k) for the list.contains() calls (k grows as values append)
+  After:  O(N) for the set.add() calls (O(1) each)
+
+Total cost for D documents:
+  Before: D × O(N×k) ≈ D × N × k/2 comparisons (average)
+  After:  D × O(N) ≈ D × N hash lookups
+"""
+
+import time
+import sys
+
+
+def append_values_before(existing, new_values):
+    """Original: list.contains O(k) per value."""
+    result = list(existing)
+    for val in new_values:
+        if val not in result:
+            result.append(val)
+    return result
+
+
+def append_values_after(existing, new_values):
+    """Fixed: HashSet shadow O(1) per value."""
+    result = list(existing)
+    seen = set(result)
+    for val in new_values:
+        if val not in seen:
+            result.append(val)
+            seen.add(val)
+    return result
+
+
+def benchmark_bulk_ingest(num_docs, field_size, new_per_doc, overlap_pct=50):
+    """
+    Simulate bulk ingest of num_docs documents.
+    Each document has an array field with field_size existing values.
+    Each append adds new_per_doc values with overlap_pct% already present.
+    """
+    overlap = int(new_per_doc * overlap_pct / 100)
+    unique_new = new_per_doc - overlap
+
+    # Build representative document field and new values
+    existing = list(range(field_size))
+    new_values = list(range(field_size - overlap, field_size - overlap + new_per_doc))
+
+    # Time BEFORE (list.contains)
+    t0 = time.perf_counter()
+    for _ in range(num_docs):
+        append_values_before(existing, new_values)
+    t_before = time.perf_counter() - t0
+
+    # Time AFTER (set shadow)
+    t0 = time.perf_counter()
+    for _ in range(num_docs):
+        append_values_after(existing, new_values)
+    t_after = time.perf_counter() - t0
+
+    speedup = t_before / t_after if t_after > 1e-9 else float('inf')
+
+    return t_before, t_after, speedup
+
+
+def format_time(seconds):
+    if seconds < 0.001:
+        return f"{seconds*1_000_000:.0f}us"
+    elif seconds < 1:
+        return f"{seconds*1000:.1f}ms"
+    elif seconds < 60:
+        return f"{seconds:.2f}s"
+    elif seconds < 3600:
+        return f"{seconds/60:.1f}min"
+    else:
+        return f"{seconds/3600:.1f}hr"
+
+
+def main():
+    print("=" * 90)
+    print("Elasticsearch CWE-407 Scaling Benchmark — IngestDocument.appendValues()")
+    print("=" * 90)
+    print()
+
+    # Scenario 1: Small array fields (k=20, N=10) — tags, labels, categories
+    print("Scenario 1: Tag/label append (k=20 existing, N=10 new, 50% overlap)")
+    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
+    print("-" * 65)
+    for num_docs in [10_000, 100_000, 1_000_000]:
+        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 20, 10)
+        saved = t_b - t_a
+        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")
+
+    print()
+
+    # Scenario 2: Medium array fields (k=100, N=50) — enrichment pipelines
+    print("Scenario 2: Enrichment pipeline (k=100 existing, N=50 new, 50% overlap)")
+    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
+    print("-" * 65)
+    for num_docs in [10_000, 100_000, 1_000_000]:
+        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 100, 50)
+        saved = t_b - t_a
+        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")
+
+    print()
+
+    # Scenario 3: Large array fields (k=500, N=200) — IoT sensor tags, log enrichment
+    print("Scenario 3: Large field append (k=500 existing, N=200 new, 50% overlap)")
+    print(f"{'Documents':>15} {'Before':>12} {'After':>12} {'Speedup':>10} {'Saved':>12}")
+    print("-" * 65)
+    for num_docs in [10_000, 100_000, 1_000_000]:
+        t_b, t_a, spd = benchmark_bulk_ingest(num_docs, 500, 200)
+        saved = t_b - t_a
+        print(f"{num_docs:>15,} {format_time(t_b):>12} {format_time(t_a):>12} {spd:>9.0f}x {format_time(saved):>12}")
+
+    print()
+
+    # Scenario 4: Projected at 10M/100M scale (extrapolate from 10K measurement)
+    print("Scenario 4: Projected savings at production scale (k=100, N=50)")
+    print("(Extrapolated linearly from measured 10K document benchmark)")
+    print("-" * 65)
+    t_b_10k, t_a_10k, spd = benchmark_bulk_ingest(10_000, 100, 50)
+    for label, multiplier in [("10K", 1), ("100K", 10), ("1M", 100),
+                               ("10M", 1_000), ("100M", 10_000)]:
+        proj_before = t_b_10k * multiplier
+        proj_after = t_a_10k * multiplier
+        proj_saved = proj_before - proj_after
+        print(f"  {label:>6} docs: before={format_time(proj_before):>10}  "
+              f"after={format_time(proj_after):>10}  "
+              f"saved={format_time(proj_saved):>10}  ({spd:.0f}x)")
+
+    print()
+    print("=" * 90)
+    print(f"Conclusion: at k=100, N=50 (typical enrichment pipeline),")
+    print(f"  our patch saves {format_time(t_b_10k - t_a_10k)} per 10K documents ({spd:.0f}x faster).")
+    print(f"  At 100M documents: projected savings of {format_time((t_b_10k - t_a_10k) * 10_000)}.")
+    print("=" * 90)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/defects/elasticsearch/unit/test_elasticsearch_stacked.py b/defects/elasticsearch/unit/test_elasticsearch_stacked.py
new file mode 100644
index 000000000..b891a11de
--- /dev/null
+++ b/defects/elasticsearch/unit/test_elasticsearch_stacked.py
@@ -0,0 +1,241 @@
+"""
+Elasticsearch CWE-407 Stacked Benchmark — all 3 patches combined.
+
+Models a realistic Elasticsearch cluster workload that exercises all three
+patched code paths in a single pipeline run:
+
+1. IngestDocument.appendValues() — per-document, runs on every ingested doc
+2. XContentHelper.mergeDefaults() — per-index, runs on index creation/template apply
+3. SnapshotsService clone — per-snapshot, runs on repository clone operations
+
+Workload model (realistic production scenario):
+  - Bulk ingest D documents across I indices
+  - Each document has array field with k values, receives N new values (50% overlap)
+  - After ingest, create/update I index mappings (triggers mergeDefaults)
+  - After mapping, clone snapshot across S snapshots × I indices
+
+This benchmark measures the COMBINED savings from all 3 patches applied together.
+"""
+
+import time
+import sys
+
+
+# ---------------------------------------------------------------------------
+# Patch 1: IngestDocument.appendValues()
+# ---------------------------------------------------------------------------
+
+def ingest_append_before(existing, new_values):
+    result = list(existing)
+    for val in new_values:
+        if val not in result:
+            result.append(val)
+    return result
+
+def ingest_append_after(existing, new_values):
+    result = list(existing)
+    seen = set(result)
+    for val in new_values:
+        if val not in seen:
+            result.append(val)
+            seen.add(val)
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Patch 2: XContentHelper.mergeDefaults()
+# ---------------------------------------------------------------------------
+
+def merge_defaults_before(base_list, merge_list):
+    """Original: list.contains O(k) per item in merge loop."""
+    merged = list(merge_list)
+    for item in base_list:
+        if item not in merged:
+            merged.append(item)
+    return merged
+
+def merge_defaults_after(base_list, merge_list):
+    """Fixed: LinkedHashSet for O(1) membership."""
+    merged = list(merge_list)
+    seen = set(merged)
+    for item in base_list:
+        if item not in seen:
+            merged.append(item)
+            seen.add(item)
+    return merged
+
+
+# ---------------------------------------------------------------------------
+# Patch 3: SnapshotsService clone
+# ---------------------------------------------------------------------------
+
+def snapshot_clone_before(snapshot_ids, index_snapshots_list):
+    """Original: List.contains O(S) per index."""
+    cloneable = []
+    for idx_id in range(len(index_snapshots_list)):
+        if snapshot_ids[0] in index_snapshots_list[idx_id]:  # O(S) list scan
+            cloneable.append(idx_id)
+    return cloneable
+
+def snapshot_clone_after(snapshot_ids, index_snapshots_list):
+    """Fixed: HashSet wrapper for O(1) lookup."""
+    cloneable = []
+    for idx_id in range(len(index_snapshots_list)):
+        snap_set = set(index_snapshots_list[idx_id])  # O(S) once
+        if snapshot_ids[0] in snap_set:  # O(1)
+            cloneable.append(idx_id)
+    return cloneable
+
+
+# ---------------------------------------------------------------------------
+# Stacked benchmark
+# ---------------------------------------------------------------------------
+
+def format_time(seconds):
+    if seconds < 0.001:
+        return f"{seconds*1_000_000:.0f}us"
+    elif seconds < 1:
+        return f"{seconds*1000:.1f}ms"
+    elif seconds < 60:
+        return f"{seconds:.2f}s"
+    elif seconds < 3600:
+        return f"{seconds/60:.1f}min"
+    elif seconds < 86400:
+        return f"{seconds/3600:.1f}hr"
+    else:
+        return f"{seconds/86400:.1f} days"
+
+
+def run_stacked_benchmark(num_docs, num_indices, num_snapshots, field_k, field_n,
+                          merge_fields, label=""):
+    """
+    Run all 3 operations before and after patches.
+
+    Args:
+        num_docs: documents to ingest
+        num_indices: indices (mapping merges)
+        num_snapshots: snapshots per index (clone lookups)
+        field_k: existing array field size per document
+        field_n: new values appended per document
+        merge_fields: fields per mapping merge
+    """
+    overlap = field_n // 2
+    existing = list(range(field_k))
+    new_values = list(range(field_k - overlap, field_k - overlap + field_n))
+
+    base_mapping = [f"field_{i}" for i in range(merge_fields)]
+    merge_mapping = [f"field_{i}" for i in range(merge_fields // 2, merge_fields + merge_fields // 2)]
+
+    # Build snapshot index lists: each index has num_snapshots snapshot IDs
+    index_snap_lists = [list(range(num_snapshots)) for _ in range(num_indices)]
+    target_snap = [num_snapshots // 2]  # snapshot to clone
+
+    # --- BEFORE ---
+    t0 = time.perf_counter()
+
+    # 1. Ingest
+    for _ in range(num_docs):
+        ingest_append_before(existing, new_values)
+
+    # 2. Mapping merges
+    for _ in range(num_indices):
+        merge_defaults_before(base_mapping, merge_mapping)
+
+    # 3. Snapshot clone
+    for _ in range(10):  # simulate 10 clone operations
+        snapshot_clone_before(target_snap, index_snap_lists)
+
+    t_before = time.perf_counter() - t0
+
+    # --- AFTER ---
+    t0 = time.perf_counter()
+
+    # 1. Ingest
+    for _ in range(num_docs):
+        ingest_append_after(existing, new_values)
+
+    # 2. Mapping merges
+    for _ in range(num_indices):
+        merge_defaults_after(base_mapping, merge_mapping)
+
+    # 3. Snapshot clone
+    for _ in range(10):
+        snapshot_clone_after(target_snap, index_snap_lists)
+
+    t_after = time.perf_counter() - t0
+
+    speedup = t_before / t_after if t_after > 1e-9 else float('inf')
+    saved = t_before - t_after
+
+    return t_before, t_after, speedup, saved
+
+
+def main():
+    print("=" * 95)
+    print("Elasticsearch CWE-407 STACKED Benchmark — All 3 Patches Combined")
+    print("=" * 95)
+    print()
+    print("Workload: ingest D docs (k fields, N appends) + merge I index mappings + 10 snapshot clones")
+    print()
+
+    # --- Scenario A: Typical cluster (k=100 enrichment) ---
+    print("Scenario A: Enrichment pipeline (k=100, N=50, 200 merge fields, 100 snapshots)")
+    print(f"{'Docs':>10} {'Indices':>8} {'Snaps':>6}  {'Before':>12} {'After':>12} {'Speedup':>8} {'Saved':>12}")
+    print("-" * 80)
+
+    for docs, indices, snaps in [(10_000, 50, 100), (100_000, 200, 100), (1_000_000, 500, 100)]:
+        tb, ta, spd, saved = run_stacked_benchmark(docs, indices, snaps, 100, 50, 200, "A")
+        print(f"{docs:>10,} {indices:>8} {snaps:>6}  {format_time(tb):>12} {format_time(ta):>12} {spd:>7.0f}x {format_time(saved):>12}")
+
+    print()
+
+    # --- Scenario B: SIEM/IoT (k=500 large fields) ---
+    print("Scenario B: SIEM/IoT (k=500, N=200, 500 merge fields, 200 snapshots)")
+    print(f"{'Docs':>10} {'Indices':>8} {'Snaps':>6}  {'Before':>12} {'After':>12} {'Speedup':>8} {'Saved':>12}")
+    print("-" * 80)
+
+    for docs, indices, snaps in [(10_000, 100, 200), (100_000, 500, 200)]:
+        tb, ta, spd, saved = run_stacked_benchmark(docs, indices, snaps, 500, 200, 500, "B")
+        print(f"{docs:>10,} {indices:>8} {snaps:>6}  {format_time(tb):>12} {format_time(ta):>12} {spd:>7.0f}x {format_time(saved):>12}")
+
+    print()
+
+    # --- Multi-node projections ---
+    print("=" * 95)
+    print("Multi-node cluster projections (measured single-node × node count)")
+    print("=" * 95)
+    print()
+
+    # Measure base at 10K docs for projection
+    tb_100_10k, ta_100_10k, spd_100, _ = run_stacked_benchmark(10_000, 50, 100, 100, 50, 200)
+    tb_500_10k, ta_500_10k, spd_500, _ = run_stacked_benchmark(10_000, 100, 200, 500, 200, 500)
+
+    print("Enrichment pipeline (k=100):")
+    print(f"{'Cluster':>10} {'Nodes':>6} {'Docs':>10}  {'Before':>14} {'After':>14} {'Speedup':>8} {'Saved':>14}")
+    print("-" * 85)
+    for label, nodes, doc_mult in [("Small", 3, 1_000), ("Medium", 10, 10_000),
+                                    ("Large", 42, 10_000), ("XL", 42, 100_000)]:
+        proj_b = tb_100_10k * doc_mult * nodes
+        proj_a = ta_100_10k * doc_mult * nodes
+        print(f"{label:>10} {nodes:>6} {doc_mult*10_000:>10,}  {format_time(proj_b):>14} {format_time(proj_a):>14} {spd_100:>7.0f}x {format_time(proj_b - proj_a):>14}")
+
+    print()
+    print("SIEM/IoT pipeline (k=500):")
+    print(f"{'Cluster':>10} {'Nodes':>6} {'Docs':>10}  {'Before':>14} {'After':>14} {'Speedup':>8} {'Saved':>14}")
+    print("-" * 85)
+    for label, nodes, doc_mult in [("Small", 3, 1_000), ("Medium", 10, 10_000),
+                                    ("Large", 42, 10_000), ("XL", 42, 100_000)]:
+        proj_b = tb_500_10k * doc_mult * nodes
+        proj_a = ta_500_10k * doc_mult * nodes
+        print(f"{label:>10} {nodes:>6} {doc_mult*10_000:>10,}  {format_time(proj_b):>14} {format_time(proj_a):>14} {spd_500:>7.0f}x {format_time(proj_b - proj_a):>14}")
+
+    print()
+    print("=" * 95)
+    print("All three patches stacked: ingest + mapping merge + snapshot clone.")
+    print(f"Enrichment (k=100): {spd_100:.0f}x combined speedup per node.")
+    print(f"SIEM/IoT (k=500):   {spd_500:.0f}x combined speedup per node.")
+    print("=" * 95)
+
+
+if __name__ == "__main__":
+    main()