Skip to content

Redundancy/flow

BranchesTags

Repository files navigation

flow

A declarative query engine for files — text, JSON, whatever lives on disk — with a DSL you can read in ten seconds instead of auditing, and that runs 5–200× faster than the Python or PowerShell an agent would have written instead.

When you ask a code agent "which components logged the most errors in the last hour", "find all the auth failures in these access logs", or "walk every shell script transitively sourced by deploy.sh and flag any rm -rf", the default answer is a Python or PowerShell script. Before running it you have to read the code and convince yourself it won't shell out, hit the network, or clobber a file. That review tax compounds with every question.

flow turns the same questions into a pipeline of composable stages. The agent writes a few lines you can glance at:

glob("logs", name_ts="%Y-%m-%d")
| read_lines ordered
| since @leading "1h ago"
| parse_json
| count by .level, .component
| sort by .state.count desc limit 5

That's the top 5 (level, component) pairs by error count in the last hour, across a directory of daily-rotated logs — one screenful, every stage visible. No JSON parse happens until the time window has dropped out-of-range lines; files whose filenames say they're older than the window are never opened.

What's in the box:

  • Parallel glob + mmap line reader. ignore::WalkBuilder for discovery, zero-copy Bytes slices for lines, chunked transport between stages, one worker pool per stage.
  • SIMD literal search. memchr::memmem (single pattern) or Aho-Corasick (a list) — drops non-matches before any VRL runs. context N gives you grep-style surround.
  • Byte-level time windows. since / until / between pull timestamps straight from raw bytes — no JSON parse, no VRL. Works with JSON fields, leading tokens, bracketed prefixes, or Apache CLF.
  • Grouped folds with typed state. fold by <keys> init { VRL } reduce { VRL } for arbitrary per-key accumulators; count and sort … limit N are sugar for the common shapes.
  • Recursive discover. Extract a file path from an event (VRL block or .field) and it feeds back into the reader — walk a dependency graph as one pipeline. Dedup + cycle-safe termination.
  • Opt-in skips for rotated logs. glob(..., name_ts="%Y-%m-%d") drops files outside the window before opening; read_lines ordered skips each file's out-of-window prefix.

A read-only sandbox, almost. The DSL cannot write files, spawn processes, open sockets, or evaluate arbitrary code outside the constrained VRL expression surface. What it can do is read: today that means any file the invoking process has permission to read, so an agent-written pipeline with a path like ../../etc/passwd would happily scan it. A workspace-scoping mode that confines reads (and discover follows) to the directory you launch from is planned; until then, auditing a flow script means checking the paths it names, nothing more.

Examples

Print matching lines, grep-style

A pipeline with no trailing fold emits each surviving event as path:line_number:content on stdout — the shape you'd expect from grep -Hn:

glob("logs")
| read_lines
| search "ERROR"

Output:

logs/app-000.log:412:1601479096 [ERROR] api - processed request 0 in 1007ms
logs/app-000.log:987:1601479104 [ERROR] db  - processed request 13 in 882ms
...

Narrow with a fast literal scan, then parse only the survivors

Most log lines aren't interesting. search is a SIMD byte scan (memchr::memmem, AVX2 on x86-64) that drops non-matches before any VRL machinery runs. You can chain it before a parse_json! to skip the expensive parse for every line that can't match — a shape grep or Select-String can't produce:

glob("logs")
| read_lines
| search "ERROR"
| parse_json
| count by .component

On 56 MB of JSON lines this runs in ~77 ms end-to-end — the 80 % of lines that don't contain ERROR never enter the JSON parser.

Group by two keys with a typed accumulator

glob("logs")
| read_lines
| parse_json
| count by .level, .component

init { .count = 0 } is evaluated once at compile time; its output Kind becomes the target type for reduce, so .count + 1 type-checks without any ?? 0 fallback noise. Every new group key is seeded by cloning the pre-computed initial state — no VRL at runtime.

Time-window filter before JSON parse

Every log query eventually wants "in the last hour" or "between these two dates". since / until pull the timestamp straight out of raw bytes — no JSON parse, no VRL — so out-of-window lines are dropped before any downstream work:

glob("logs")
| read_lines
| since .ts "1h ago"
| parse_json
| count by .component

For JSON logs since .ts "..." does a SIMD needle-scan for "ts": and parses the integer or ISO string that follows (~5 ns/event). Text logs use positional extractors: @leading for "timestamp is the first token" (unix int or ISO), @bracketed for "first [...] content", @clf for Apache / nginx Common Log Format.

Time values: relative ("1h ago", "24h ago", "7d ago", "now") or absolute ("2024-01-15", "2024-01-15T10:30:00Z"). Relative forms are resolved once at pipeline compile time.

Narrow windows on rotated logs: name_ts + ordered

The query above still opens every file in logs/ and scans every line. For the common case of daily-rotated logs where the filename carries the date (app-2024-01-15.log, access_log_20240115) and each file is internally time-ordered, two opt-in hints collapse that cost dramatically:

glob("logs", name_ts="%Y-%m-%d")
| read_lines ordered
| between @leading "2024-03-15T10:00:00Z" .. "2024-03-15T11:00:00Z"
| count
  • name_ts="..." parses a chrono strftime out of each filename. Files whose name-timestamp falls outside [since, until] are dropped before they're opened — except for the latest pre-window file, which is always kept because it may straddle the lower boundary. Fail-safe: files without a parseable timestamp always pass through, so a wrong strftime silently degrades to no-pruning rather than dropping events.
  • read_lines ordered tells the reader it can skip past each file's out-of-window prefix and stop reading the moment a line falls past until. Small files use a linear-skip scan; larger files (≥ 1 MB) use a byte-level binary search (log₂ probes + one final sweep). Trust-me: if you set ordered on a file that isn't actually time-sorted, events can be silently dropped. Only set it when you know the invariant holds — log rotation tools generally give you this for free.

Measured payoff on a synthetic year-of-daily-logs corpus (1 GB, 365 files, 1-hour window):

hints elapsed events
none (full scan) 227 ms 2084
both hints 52 ms 2084

Counts match exactly; the speedup is pure skipped I/O. The override env var FLOW_ORDERED_BINSEARCH_MIN lets you tune the linear↔binsearch cutover if your extractor cost differs markedly from integer-leading (default 1 MB).

Top 5 loudest components in the last hour

glob("logs")
| read_lines
| since .ts "1h ago"
| parse_json
| count by .component
| sort by .state.count desc limit 5

Output is five rows, loudest first. count is sugar for the idiomatic counter fold; sort is a barrier stage that runs after fold. limit 5 truncates after the sort.

Compute a max and a count per key in one pass

glob("logs")
| read_lines
| parse_json
| filter { int!(.latency_ms) > 1000 }
| fold by .component
    init { .count = 0; .max_ms = 0 }
    reduce {
      .count = .count + 1
      incoming = to_int(%latency_ms)
      if incoming > .max_ms { .max_ms = incoming }
    }

. is the mutable accumulator; % is the incoming event. One pass does JSON parse, latency filter, and a two-field accumulation per component. Running this against a 40 k-line JSON-lines corpus produces one row per component:

{ "_key": ["db"], "state": { "count": 6397, "max_ms": 4999 } }
{ "_key": ["api"], "state": { "count": 6408, "max_ms": 4999 } }
{ "_key": ["worker"], "state": { "count": 6372, "max_ms": 4999 } }
{ "_key": ["frontend"], "state": { "count": 6372, "max_ms": 4999 } }
{ "_key": ["auth"], "state": { "count": 6366, "max_ms": 4997 } }
Pipeline finished. Emitted 5 events.
Execution Complete in 64.7ms

Audit every script transitively sourced by a deploy entrypoint

Most search tools (grep, ripgrep, ag) operate strictly on the files you point at — they don't follow references between files. Without a programmable runtime, the natural version of "trace all includes and then check for X" is a recursive shell loop with manual dedup and cycle handling.

discover turns that into one pipeline stage. Extract a path from an event via VRL, and the path is fed back into the read stage. A shared dedup set prevents revisits (cycles terminate). Non-existent references are silently skipped. The pipeline closes automatically the moment no more work can arrive — no grace period, no heuristic.

The canonical example: a deploy entrypoint that sources several helpers, some of which source others in turn. You want to know whether any transitively-sourced script contains rm -rf — a real security-audit question.

Input. A four-script corpus where deploy.sh sources three helpers and logging.sh sources one of them transitively:

scripts/deploy.sh
  #!/bin/bash
  source scripts/lib/env.sh
  source scripts/lib/logging.sh
  source scripts/rollback.sh
  run_migration

scripts/lib/env.sh            → export APP_HOME=/opt/app
scripts/lib/logging.sh        → source scripts/lib/env.sh
                              → log_info() { echo "$1"; }
scripts/rollback.sh           → rm -rf /opt/app.old
                              → mv /opt/app /opt/app.old

Pipeline:

glob("scripts/deploy.sh")
| read_lines
    # .             = one line of file, as raw bytes
    # %path         = "scripts/deploy.sh" (or whatever file we're reading)
    # %line_number  = 1-based line number within the file
| discover(paths = {
    # VRL block runs per event; return the path to queue for reading,
    # or null to ignore. For "source scripts/lib/env.sh" the block
    # returns "scripts/lib/env.sh"; for any other line, null.
    parse_groks(., ["source %{data:path}"]).path ?? null
  })
    # events pass through unchanged; newly-discovered paths go back
    # to the read stage and deduped against already-seen files.
| search "rm -rf"
    # SIMD byte-scan against the raw line; emits each match grep-style.

Output. Every real file traversed once (env.sh deduped despite two references), cycles safe by construction, total ~8 ms for this graph:

scripts/rollback.sh:1:rm -rf /opt/app.old

Swap "source %{data:path}" for your toolchain's include syntax ("#include \"%{data:path}\"", "include::%{data:path}", "import %{data:module}", nginx "include %{data:path}") and the same pipeline walks that dependency graph.

Language

One pipeline, stages joined by |:

Stage What it does
glob("path") Parallel recursive file discovery via ignore::WalkBuilder. Optional name_ts="%Y-%m-%d" prunes files whose filename-stem timestamp is outside the pipeline's time window — fail-safe (unparseable names always pass through)
read_lines mmap-backed; emits one zero-copy event per line. Optional ordered promises per-file monotonic timestamps, letting the reader skip past out-of-window prefixes — trust-me (wrong claim silently drops events)
read_file Emits one event per whole-file payload
search "literal" or search ["a", "b", ...] SIMD substring filter. One literal uses memchr::memmem; a list uses Aho-Corasick single-pass multi-literal scan. Drops non-matches before any VRL
filter { VRL } Keep events where the VRL body evaluates to true
map { VRL } Replace the event body with the VRL result
parse_json Sugar for map { . = parse_json!(.) } — parse the current . as JSON
parse_frontmatter Sugar for map { . = parse_frontmatter!(.) } — split a leading ----fenced YAML block from its body. See below
discover(paths = <field-or-block>) Feed new file paths back into the pipeline. Value is a .field accessor, or an inline { VRL } block returning a path string (single), array of strings, or null to ignore
fold by <fields> init { VRL } reduce { VRL } Grouped aggregation with a typed initial state
count [by <fields>] Sugar for the most common fold shape — .count = .count + 1
sort by <path> [desc] [limit N] Order events by an integer-valued path; limit N truncates to top N
since <source> "<time>" Keep events whose timestamp ≥ <time>. Byte-level extraction — no JSON parse needed
until <source> "<time>" Keep events whose timestamp ≤ <time>
between <source> "<from>" .. "<to>" Sugar for since … | until … — one visual unit for a bounded interval

<source> is either .field (JSON needle scan), @leading (first token on the line), @bracketed (first [...] content), or @clf (Apache Common Log Format).

YAML

parse_yaml!(.) is the YAML analogue of parse_json!(.) — a general-purpose VRL function, useful for .yaml config files, Kubernetes manifests, CI configs, docker-compose.yml, anything beyond markdown:

glob("k8s") | read_file | filter { .name == "*.yaml" } | map { . = parse_yaml!(.) } | filter { .kind == "Deployment" }

Markdown frontmatter — Jekyll/Hugo/Zola/Astro/mkdocs style, a ----fenced YAML block at the top of a file — gets a dedicated function and stage sugar, parse_frontmatter, rather than being a parse_yaml + manual-split recipe:

glob("posts") | read_file | parse_frontmatter | filter { !bool!(.frontmatter.draft) } | count

parse_frontmatter!(.) returns { "frontmatter": <parsed yaml or null>, "body": "<remaining text>" }. Fence detection is line-based, not a blind string split, so it's precise about what counts as a fence:

  • The opening fence must be the file's first line, consisting of exactly --- (trailing whitespace tolerated, optional leading UTF-8 BOM).
  • The closing fence is the first subsequent line that is exactly --- or ... (both are valid YAML end-of-document markers) — a --- appearing mid-line, indented inside a YAML block scalar, or later in the body as a markdown horizontal rule, is never mistaken for a fence.
  • A file with no opening fence, or an opening fence with no matching close, yields {frontmatter: null, body: <whole file>} rather than an error — a mixed corpus of frontmatter and plain markdown files needs no pre-filtering stage.

Events travel between stages in chunks of 1024 across bounded channels, with parallel workers per stage. Set FLOW_TIMING=1 to get a per-stage events-in/out / work-time / send-wait / recv-wait breakdown on stderr.

VRL convention (same as Vector): . is the event body — the raw line bytes for a freshly read event, or whatever a prior map / parse_json set it to. % is file-level metadata: %path, %size, %line_number. Replacing . via map doesn't clobber % — file context is preserved across transforms. Both . and % accept arbitrary depth paths (.meta.region, %path).

All four runtime constants — chunk size, inter-stage channel depth, fold shard count, and the mmap-vs-buffer file-size threshold — are overridable via FLOW_CHUNK_SIZE / FLOW_CHANNEL_DEPTH / FLOW_FOLD_SHARDS / FLOW_TINY_FILE_THRESHOLD_BYTES. Defaults are validated in bench/results/TUNING.md; deviate only if a sweep shows a win on your workload.

Performance

Measured with hyperfine on Windows 11, corpora generated by the bundled datagen, all three runtimes print only a summary line (apples-to-apples).

Workload Corpus flow python powershell
count-error (substring + count) 56 MB / 1 M lines 127 ms 437 ms 738 ms
count-error 168 MB / 3 M lines 341 ms 1.07 s 1.76 s
search-error (SIMD stage) 168 MB / 3 M lines 187 ms 1.07 s 1.77 s
group-by-level (JSON parse + fold) 56 MB jsonl 252 ms 1.22 s 27.5 s
count-all (100 k tiny JSON files) ~14 MB 773 ms 5.24 s 2.37 s
time-window-1h (name_ts + ordered) 17 MB / 30 daily files 21 ms 229 ms 1.24 s
time-window-1h (name_ts + ordered) 68 MB / 60 daily files 22 ms 505 ms 4.56 s

The last row is where the two opt-in hints shine: flow's wall-clock barely moves as the corpus grows 4× (20.7 → 21.9 ms) because name_ts prunes 59 of 60 files before open and read_lines ordered skips most of the one survivor. Python and PowerShell scale linearly — they open every file.

See bench/results/BASELINE.md for the pre-optimization baseline and bench/results/ for current per-size tables.

Usage

cargo build --release

# Inline (positional — the LLM-friendly form)
./target/release/flow 'glob("logs") | read_lines | search "ERROR" | count'

# On stdin (auto-detected when stdin is a pipe; `-` also works explicitly)
printf 'glob("logs") | read_lines | count' | ./target/release/flow

# From a file
./target/release/flow --script-file pipeline.dsl

# Also still available
./target/release/flow --script 'glob("logs") | read_lines | count'

# With per-stage timing on stderr
FLOW_TIMING=1 ./target/release/flow --script-file pipeline.dsl

# As an MCP server on stdio (for Claude Code, Claude Desktop, etc.)
./target/release/flow --mcp

MCP server

flow --mcp starts a JSON-RPC 2.0 server on stdio speaking the Model Context Protocol (v2024-11-05). An LLM host can call flow directly without going through the Bash tool — no shell quoting, structured errors, structured events.

Two tools are exposed:

  • flow_run({script: string}) — runs a pipeline, returns emitted events as JSON (up to 1000 items / 1 MB, with a truncated flag). VRL / parse errors come back in error.message with macro attribution intact — the LLM never parses stderr.
  • flow_help() — returns the DSL cheat-sheet as text so the LLM can self-ground before writing its first script.

Example Claude Desktop config:

{
  "mcpServers": {
    "flow": {
      "command": "/absolute/path/to/flow",
      "args": ["--mcp"]
    }
  }
}

Sample happy-path response payload (from flow_run):

{
  "ok": true,
  "event_count": 4,
  "returned_count": 4,
  "truncated": false,
  "events": [
    { "kind": "raw", "path": "src/main.rs", "line_number": 86, "line": "fn main() -> anyhow::Result<()> {\n" }
  ]
}

Using flow as a Rust library

The flow crate builds both the CLI binary and a library (src/lib.rs) — a Rust program can embed the pipeline engine directly instead of shelling out or going through MCP:

[dependencies]
flow = { path = "../flow" }  # or a git dependency once published
let pipeline = flow::parse(r#"glob("logs") | read_lines | search "ERROR" | count"#)?;
flow::compile_and_run(pipeline)?;

compile_and_run prints to stdout exactly like the CLI. To capture events programmatically instead — the common case for embedding — use compile_and_run_with_sink with EventSink::Collect, the same mechanism the MCP server (src/mcp.rs) uses internally:

use flow::EventSink;
use std::sync::{atomic::AtomicBool, Arc, Mutex};

let pipeline = flow::parse(r#"glob("logs") | read_lines | parse_json | count by .component"#)?;
let buf = Arc::new(Mutex::new(Vec::new()));
let truncated = Arc::new(AtomicBool::new(false));
let emitted = Arc::new(std::sync::atomic::AtomicU64::new(0));
flow::compile_and_run_with_sink(pipeline, EventSink::Collect {
    buf: buf.clone(), cap_events: 1000, cap_bytes: 1_000_000, truncated, emitted,
})?;
let events = buf.lock().unwrap(); // Vec<serde_json::Value>

Scripts using include / macro directives need a preprocessor pass before parse — flow::preprocess::expand(script, origin, cwd) — which the CLI does automatically; library callers that skip it just get plain DSL with no macro/include support.

Building a Pipeline without parsing DSL text — Stage and Pipeline are plain public types, so you can always construct one by hand (Pipeline { stages: vec![Stage::Glob { path: "logs".into(), name_ts: None }, ...] }). PipelineBuilder wraps that in a fluent API with one method per stage, plus the same sugar the DSL parser lowers at parse time (count, parse_json, parse_frontmatter):

let pipeline = flow::PipelineBuilder::new()
    .glob("logs")
    .read_lines()
    .search("ERROR")
    .count()
    .build();
flow::compile_and_run(pipeline)?;

.stage(Stage::Whatever { .. }) is an escape hatch for any shape without a named method yet. A test (builder::tests::builder_matches_parser_sugar) pins each sugar method's output to exactly what parser::parse produces for the equivalent DSL script, so the two surfaces can't silently drift apart.

Reusable snippets with macro + lib.flow

Common VRL fragments — counters, seed states, parameterised filters — can be defined once and referenced by name. If a file called lib.flow exists in the current directory it's auto-loaded, so shared snippets are available without an include line.

# lib.flow — shared snippets
macro seed_count     = { .count = 0 }
macro bump_count     = { .count = .count + 1 }
macro at_least(n)    = { int!(.latency_ms) >= $n }

# then in any script run from this directory
glob("logs")
| read_lines
| parse_json
| filter &at_least(1000)
| fold by .component init &seed_count reduce &bump_count
| sort by .state.count desc limit 5
  • Invocation is bracket-free at stage-body position. filter &at_least(1000) — not filter { &at_least(1000) }.
  • macro is textual substitution, not a function. No lexical scope, no single-evaluation of args, no recursion.
  • include "path" pulls in another .flow file (relative to the including file). Idempotent — including the same file twice is a no-op.
  • Bad-VRL errors point at the macro. If a macro body has a VRL syntax error, the compile error names the offending macro and its declaration site (e.g. (via macro at_least at lib.flow:3)) — no grepping the library.

Compose at the pipe boundary, not inside a block

Macros are stage-body-only by design. AND-compose filters by chaining them; sequence transforms by chaining maps. Each pipe boundary is a fresh VRL block, so the macros never mix operator-precedence with each other or with inline VRL.

# lib.flow
macro live_only          = { bool!(.live) }
macro slow(ms)           = { int!(.latency_ms) > $ms }
macro us_traffic         = { starts_with(string!(.region), "us-") }
macro parse_ts           = { .ts_ms = to_unix_timestamp!(parse_timestamp!(.ts, "%Y-%m-%dT%H:%M:%S%.fZ")) }
macro drop_stack         = { del(.stack_trace) }

# Query: slow live US requests, normalised
glob("logs")
| read_lines
| parse_json
| filter &live_only        # ← AND-compose by chaining
| filter &us_traffic
| filter &slow(1000)
| map &parse_ts            # ← sequence transforms
| map &drop_stack
| count by .component

What doesn't work — and what to write instead:

  • Trying to AND two macros inside one filter body:

    # ✗ Nested braces — the preprocessor catches this and hints at the fix
filter { &live_only && &slow(1000) }

# ✓ Chain the filters instead
filter &live_only | filter &slow(1000)

  • Building an AND combinator as a macro:

    # ✗ Don't
macro and(a, b) = { $a && $b }
filter &and(bool!(.live), int!(.latency_ms) > 1000)

# ✓ The pipeline IS the AND combinator
filter &live_only | filter &slow(1000)

  • Trying to sequence two transforms inside one map body:

    # ✗ Nested braces again
map { &parse_ts; &drop_stack }

# ✓ Chain the maps
map &parse_ts | map &drop_stack

The pipeline stages are the composition operators. filter chains as AND. map chains as function composition. fold init / fold reduce each take a single macro because a fold is one operation with two named slots. If you find yourself wanting to compose macros inside one stage body, that's the signal to split the stage.

For OR-shaped filters, either write the disjunction inline (filter { .a || .b } — no macros needed) or build a single macro that captures the compound predicate: macro live_or_test = { bool!(.live) || bool!(.test_mode) }.

Output: structured aggregates (fold results) go to stdout; diagnostics go to stderr. Raw events are drained silently so that filter/map pipelines aren't bottlenecked by terminal I/O.

Generate benchmark corpora:

./target/release/datagen              # all nine corpora (~475 MB)
./target/release/datagen --size small # fast iteration (~10 MB)

Run the benchmark harness:

powershell -File bench/run.ps1
powershell -File bench/run.ps1 -Workload count-error

Status

Working: glob, read_lines, read_file, search, filter, map, fold (init / reduce). Chunked transport, zero-copy event plumbing, SIMD literal search, per-stage instrumentation. Ships as both a CLI binary and a Rust library (src/lib.rs) — see "Using flow as a Rust library" above.

Planned (TODO.md): stage fusion for adjacent pure-CPU stages, a no-VRL fast path for count-shaped accumulators, and the Stage::Discover capability for dynamically feeding new paths back into glob.

Primary target is Windows today; the runtime is portable Rust and the harness is PowerShell. A bash runner will follow when Linux testing starts.

About

A DSL based file search tool

Resources

Readme
Activity

Stars

0 stars

Watchers

1 watching

Forks

0 forks
Report repository

Releases 1

v0.1.0 Latest
Jul 4, 2026

Packages

 
 
 

Contributors

Languages