perf(pipeline): optimize parallel scan, cache lookups, and ast block extraction (ticket-017)

packages/redup-fast-hash/src/simhash.rs

@@ -142,24 +142,17 @@ pub fn extract_fuzzy_tokens(cleaned: &str) -> Vec<String> {
         }
 
         // Multi-char operators: ===, !==, ==, !=, <=, >=, =>, ++, --, &&, ||, ??
-        if i + 2 < len {
-            let op3 = &cleaned[i..i + 3];
-            if op3 == "===" || op3 == "!==" {
-                tokens.push(op3.to_string());
-                i += 3;
-                continue;
-            }
+        let rem = &cleaned[i..];
+        if rem.starts_with("===") || rem.starts_with("!==") {
+            tokens.push(rem[..3].to_string());
+            i += 3;
+            continue;
         }
-        if i + 1 < len {
-            let op2 = &cleaned[i..i + 2];
-            if matches!(
-                op2,
-                "==" | "!=" | "<=" | ">=" | "=>" | "++" | "--" | "&&" | "||" | "??"
-            ) {
-                tokens.push(op2.to_string());
-                i += 2;
-                continue;
-            }
+        let op2 = ["==", "!=", "<=", ">=", "=>", "++", "--", "&&", "||", "??"].iter().find(|&&op| rem.starts_with(op));
+        if let Some(&op) = op2 {
+            tokens.push(op.to_string());
+            i += 2;
+            continue;
         }
 
         // Single non-whitespace character

project/TICKETS.md

@@ -19,4 +19,5 @@ New identities are reserved exclusively by project/new-ticket.sh.
 | **ticket-013** | [`README.md`](./ticket-013/README.md) | - | - | - | - | - |
 | **ticket-014** | [`README.md`](./ticket-014/README.md) | - | - | - | - | - |
 | **ticket-015** | [`README.md`](./ticket-015/README.md) | - | - | - | - | - |
+| **ticket-016** | [`README.md`](./ticket-016/README.md) | - | - | - | - | - |
 <!-- AUTO:TICKET_INDEX:END -->

project/ticket-017/README.md

@@ -0,0 +1,29 @@
+# Ticket 017: Pipeline optimization for parallel scanning, cache lookups, and AST block extraction
+
+- **ID**: ticket-017
+- **Owner**: agent:antigravity
+- **Status**: IN_PROGRESS
+- **Workflow state**: EDIT
+- **Created**: 2026-10-13
+
+## Goal and scope
+
+Optimize reDUP scan and duplicate finding bottlenecks:
+1. Enable parallel scanning by default in `scan_command` and `analyze_optimized` across CPU cores.
+2. Optimize AST function block extraction using a single-pass `ast.NodeVisitor` instead of multiple full tree traversals.
+3. Cache SQLite block hashes at the file level in `build_hash_index_with_cache` to eliminate redundant queries and SHA256 hashes per block.
+4. Skip redundant cache re-writes for unchanged files in `find_duplicates_phase_lazy`.
+5. Fix UTF-8 multi-byte operator boundary slicing panic in `redup_fast_hash::extract_fuzzy_tokens`.
+
+## Acceptance criteria
+
+- [x] AC-01: Default `parallel` to True in CLI scan command and auto-parallelize in `analyze_optimized`.
+- [x] AC-02: Single-pass AST extraction for Python function blocks without repeated whole-tree traversals.
+- [x] AC-03: File-level hash cache lookup in `build_hash_index_with_cache` avoiding per-block DB round-trips.
+- [x] AC-04: Skip re-storing unchanged file hashes into SQLite in `find_duplicates_phase_lazy`.
+- [x] AC-05: Fix UTF-8 multi-byte operator slicing in `redup_fast_hash` to prevent thread panics on unicode.
+- [x] AC-06: Pass test suite and governance checks.
+
+## Tracking boundary
+
+This directory contains the minimal reviewed intent.

project/ticket-017/intent.json

@@ -0,0 +1,24 @@
+{
+  "schema": "new-project.intent/v3",
+  "ticket": "ticket-017",
+  "summary": "perf(pipeline): optimize parallel scan, cache lookups, and ast block extraction",
+  "workstream": "application",
+  "classification": {
+    "kind": "SERVICE",
+    "priority": "P2",
+    "origin": "health"
+  },
+  "allowedPaths": [
+    "project/ticket-017/**",
+    "TODO.md",
+    "project/TICKETS.md",
+    "packages/**",
+    "src/**",
+    "tests/**"
+  ],
+  "forbiddenPaths": ["project/ticket-*/user-*.md"],
+  "stacks": [],
+  "dependsOn": [],
+  "conflictsWith": [],
+  "integrationTicket": null
+}

src/redup/cli_app/scan_commands.py

@@ -119,7 +119,7 @@ def scan_command(
         help="Only analyze function-level duplicates (default: enabled for speed).",
     ),
     parallel: bool = typer.Option(
-        False, "--parallel", help="Use parallel processing for faster analysis."
+        True, "--parallel/--no-parallel", help="Use parallel processing for faster analysis."
     ),
     max_workers: Any = typer.Option(
         None, "--max-workers", help="Maximum number of parallel workers."

src/redup/core/cache.py

@@ -271,6 +271,7 @@ def build_hash_index_with_cache(
     index = HashIndex()
     block_hash_cache: dict[Path, dict[tuple[int, int], tuple[str, str]]] = {}
     file_contents: dict[Path, str | None] = {}
+    file_cached_hashes: dict[Path, dict[tuple[int, int], tuple[str, str]] | None] = {}
 
     for block in blocks:
         if block.line_count < min_lines:
@@ -285,20 +286,26 @@ def build_hash_index_with_cache(
         if cache is not None:
             if file_path not in file_contents:
                 try:
-                    file_contents[file_path] = file_path.read_text(encoding="utf-8", errors="replace")
+                    content = file_path.read_text(encoding="utf-8", errors="replace")
+                    file_contents[file_path] = content
+                    file_cached_hashes[file_path] = cache.get_cached_block_hashes(file_path, content)
                 except OSError:
                     file_contents[file_path] = None
+                    file_cached_hashes[file_path] = None
             file_content = file_contents[file_path]
 
-        # Get hashes with cache
-        exact_hash, structural_hash = hash_block_with_cache(
-            block.text,
-            file_path,
-            block.line_start,
-            block.line_end,
-            cache,
-            file_content=file_content,
-        )
+        cached_hashes = file_cached_hashes.get(file_path)
+        if cached_hashes and block_key in cached_hashes:
+            exact_hash, structural_hash = cached_hashes[block_key]
+        else:
+            exact_hash, structural_hash = hash_block_with_cache(
+                block.text,
+                file_path,
+                block.line_start,
+                block.line_end,
+                cache,
+                file_content=file_content,
+            )
 
         # Store in per-file cache for later storage
         if file_path not in block_hash_cache:
@@ -316,3 +323,4 @@ def build_hash_index_with_cache(
         index.structural[structural_hash].append(hb)
 
     return index, block_hash_cache
+

src/redup/core/pipeline/__init__.py

@@ -159,25 +159,17 @@ def handle_interrupt(signum, frame):
 
     try:
         # Phase 1: Optimized scanning with memory cache
-        if config.parallel_workers and config.parallel_workers > 1:
-            # Explicit worker count specified
-            strategy = ScanStrategy(
-                parallel=True,
-                max_workers=config.parallel_workers,
-                memory_cache=use_memory_cache,
-                max_cache_mb=max_cache_mb if use_memory_cache else 256,
-            )
-            scanned_files, stats = scan_project(
-                config, strategy, function_level_only=function_level_only
-            )
-        elif config.parallel_workers is None and getattr(config, "_parallel_enabled", False):
-            # Auto-detect CPU count when parallel_workers is None (default behavior)
+        parallel_disabled = (
+            getattr(config, "_parallel_enabled", None) is False
+            or (config.parallel_workers is not None and config.parallel_workers <= 1)
+        )
+        if not parallel_disabled:
             import multiprocessing as mp
 
-            auto_workers = mp.cpu_count()
+            workers = config.parallel_workers or mp.cpu_count() or 4
             strategy = ScanStrategy(
                 parallel=True,
-                max_workers=auto_workers,
+                max_workers=workers,
                 memory_cache=use_memory_cache,
                 max_cache_mb=max_cache_mb if use_memory_cache else 256,
             )

src/redup/core/pipeline/duplicate_finder.py

@@ -540,12 +540,13 @@ def find_duplicates_phase_lazy(
             project_root=config.root,
         )
 
-        # Store cache data for incremental scans
+        # Store cache data for incremental scans (skip unchanged files)
         for file_path, hashes in block_hash_cache.items():
             if file_path.exists():
                 try:
-                    content = file_path.read_text()
-                    cache.store_file_hashes(file_path, content, hashes)
+                    content = file_path.read_text(encoding="utf-8", errors="replace")
+                    if not cache.is_file_unchanged(file_path, content):
+                        cache.store_file_hashes(file_path, content, hashes)
                 except OSError:
                     pass
         cache.flush()

src/redup/core/scanner/__init__.py

@@ -42,33 +42,68 @@ def _extract_function_blocks_python(source: str, filepath: str) -> list[CodeBloc
         return blocks
 
     lines = source.splitlines()
-    parent_map: dict[int, ast.ClassDef] = {}
 
-    for node in ast.walk(tree):
-        if isinstance(node, ast.ClassDef):
+    class _Visitor(ast.NodeVisitor):
+        def __init__(self):
+            self.current_class: str | None = None
+
+        def visit_ClassDef(self, node: ast.ClassDef):
+            prev = self.current_class
+            self.current_class = node.name
             for child in node.body:
                 if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
-                    parent_map[id(child)] = node
-
-    for node in ast.walk(tree):
-        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
-            start = node.lineno
-            end = node.end_lineno or start
-            class_node = parent_map.get(id(node))
-            blocks.append(
-                CodeBlock(
-                    file=filepath,
-                    line_start=start,
-                    line_end=end,
-                    text="\n".join(lines[start - 1 : end]),
-                    function_name=node.name,
-                    class_name=class_node.name if class_node else None,
+                    start = child.lineno
+                    end = child.end_lineno or start
+                    blocks.append(
+                        CodeBlock(
+                            file=filepath,
+                            line_start=start,
+                            line_end=end,
+                            text="\n".join(lines[start - 1 : end]),
+                            function_name=child.name,
+                            class_name=node.name,
+                        )
+                    )
+                self.visit(child)
+            self.current_class = prev
+
+        def visit_FunctionDef(self, node: ast.FunctionDef):
+            if self.current_class is None:
+                start = node.lineno
+                end = node.end_lineno or start
+                blocks.append(
+                    CodeBlock(
+                        file=filepath,
+                        line_start=start,
+                        line_end=end,
+                        text="\n".join(lines[start - 1 : end]),
+                        function_name=node.name,
+                        class_name=None,
+                    )
                 )
-            )
+            self.generic_visit(node)
+
+        def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef):
+            if self.current_class is None:
+                start = node.lineno
+                end = node.end_lineno or start
+                blocks.append(
+                    CodeBlock(
+                        file=filepath,
+                        line_start=start,
+                        line_end=end,
+                        text="\n".join(lines[start - 1 : end]),
+                        function_name=node.name,
+                        class_name=None,
+                    )
+                )
+            self.generic_visit(node)
 
+    _Visitor().visit(tree)
     return blocks
 
 
+
 def _extract_sliding_blocks(source: str, filepath: str, min_lines: int) -> list[CodeBlock]:
     """Extract simple sliding-window blocks for non-function-level scans."""
     if min_lines <= 0: