Beamscope.Chunking.Pipeline (Beamscope v0.1.0)

Copy Markdown View Source

Walks a repo and chunks every file concurrently via Task.Supervisor.async_stream_nolink, merging the per-file chunk lists into one combined list. Each file's parse (:epp/Code.string_to_quoted) is a pure, self-contained operation with no shared state, so files are trivially parallelizable — only the final merge step needs to wait for every task. Using the supervised, nolink variant (rather than plain Task.async_stream) means a file that raises an exception (not just one that times out) is isolated to that file's :errors entry instead of crashing the caller — which, for callers like Beamscope.Search.Store, is a singleton GenServer caching state for every repo ever queried.

File discovery mirrors chunker.py's walk_repo/CODE_EXTENSIONS/ SKIP_DIRS so the file set stays comparable to the original pipeline.

Emits :telemetry events so callers can observe progress without the core pipeline doing any IO itself:

  • [:beamscope, :chunk_repo, :start] — measurements: %{file_count}; metadata: %{repo_path}

  • [:beamscope, :chunk_repo, :file, :start | :stop | :exception] — standard :telemetry.span/3 events per file; :stop measurements include :duration (native time units) and metadata includes %{file_path, chunk_count}

  • [:beamscope, :chunk_repo, :stop] — measurements: %{duration, total_chunks, total_errors}; metadata: %{repo_path}

See Beamscope.Chunking.ProgressReporter for an opt-in handler that prints a live progress line and summary from these events.

Summary

Functions

chunk_repo(repo_path, opts \\ [])

Chunks every discoverable file under repo_path concurrently, merging all per-file chunk lists into one combined list.

walk_repo(repo_path)

Functions

chunk_repo(repo_path, opts \\ [])

@spec chunk_repo(
  String.t(),
  keyword()
) :: %{chunks: [map()], errors: [{String.t(), term()}]}

Chunks every discoverable file under repo_path concurrently, merging all per-file chunk lists into one combined list.

Options:

  • :max_concurrency — files parsed in parallel (default: System.schedulers_online(); pass 1 to force sequential processing for comparison)
  • :timeout — per-file timeout in ms (default 30_000, matching chunker.py's SUBPROCESS_TIMEOUT_SECONDS)
  • :max_chunk_lines, :max_chunk_chars, :window_lines, :overlap_lines — forwarded to Support.split_if_oversized/2 and TextChunker.chunk/2 (module-level chunk-size controls)
  • :include_dirs — forwarded to ErlangChunker.chunk_file/2, merged with auto-discovered _build/*/lib dirs unless :auto_discover_includes is false
  • :auto_discover_includes — when true (default), scans repo_path for _build/*/lib/<dep>/{include,ebin} and wires them in for -include_lib resolution, mirroring chunker.py's find_include_dirs/find_ebin_dirs (see IncludePaths)

Returns %{chunks: [map()], errors: [{path, reason}]} — a file that times out or crashes is recorded in :errors rather than failing the whole run.

walk_repo(repo_path)

@spec walk_repo(String.t()) :: [String.t()]