Records become events, the tool grows manners, the parser gets attacked on purpose, and it all becomes a graph

Milestone 9Correlation and sessionisation

Goal

Stop recomputing. Give records and their entities a home in SQLite, and turn the flat stream of entity occurrences into events: an event is one entity active over a span of time, built by grouping its occurrences whenever the gap between two of them is small enough to call them the same episode. That single idea — group by identity, split by silence — is what makes "everything that happened around this IP address" a query instead of a research project.

Concepts

DBI's connect/prepare/execute/placeholder cycle, the enormous difference between one transaction and autocommit, indexes and how to read EXPLAIN QUERY PLAN, and the session-window sweep algorithm that every web analytics tool uses under a different name.

Design

Four tables. Records and their entity occurrences are the input; events and their membership are the output of this milestone.

Implementation

Connecting, and the two pragmas that matter

package Strata::Store;
use v5.36;
use DBI;

sub open ($class, $path, %opt) {
    my $dbh = DBI->connect("dbi:SQLite:dbname=$path", "", "", {
        RaiseError => 1,        # a failed statement throws, rather than
                                 # returning false and leaving you to check
        AutoCommit => 0,        # see the benchmark below
        sqlite_unicode => 1,
    });
    $dbh->do("PRAGMA journal_mode = WAL");   # readers do not block writers
    $dbh->do("PRAGMA foreign_keys = ON");
    $class->_create_schema($dbh) unless $opt{existing};
    return bless { dbh => $dbh }, $class;
}

RaiseError turns every failed statement into an exception, which means the rest of the codebase never has to remember to check a return value; a forgotten check is how corrupted data quietly becomes "successfully" correlated. journal_mode = WAL (write-ahead logging) lets strata query read the database while strata ingest is still writing to it, which the default rollback journal does not allow — useful the moment this becomes a long-running ingest you want to inspect mid-flight.

One transaction, not ten thousand

2,000 records, autocommit per row:  12.95s   (≈ 154 rows/sec)

my $ins_rec = $dbh->prepare(
    "INSERT INTO records (id, source, file, lineno, ts, raw) VALUES (?,?,?,?,?,?)"
);
my $ins_occ = $dbh->prepare(
    "INSERT INTO occurrences (record_id, type, value, ts) VALUES (?,?,?,?)"
);

for my $rec (@records) {
    $ins_rec->execute($rec->{id}, $rec->{source}, $rec->{file}, $rec->{lineno}, $rec->{ts}, $rec->{raw});
    $ins_occ->execute($rec->{id}, $_->{type}, $_->{value}, $rec->{ts}) for @{ $rec->{entities} };
}
$dbh->commit;   # one fsync for the whole batch

100,000 records + their occurrences, one transaction:  0.92s   (≈ 108,700 rows/sec)

Sessionisation: a sweep, not a query

Grouping is easier to write as a linear pass over sorted data than as SQL, so the shape is: let SQLite sort, let Perl group.

use constant DEFAULT_GAP => 60;   # seconds of silence that ends an event

sub sessionize ($self, $gap = DEFAULT_GAP) {
    my $dbh = $self->{dbh};
    my $sth = $dbh->prepare(
        "SELECT id, record_id, type, value, ts FROM occurrences ORDER BY type, value, ts"
    );
    $sth->execute;

    my $ins_event  = $dbh->prepare(
        "INSERT INTO events (type, value, start_ts, end_ts, occurrence_count) VALUES (?,?,?,?,?)"
    );
    my $ins_member = $dbh->prepare(
        "INSERT INTO event_members (event_id, occurrence_id) VALUES (?,?)"
    );

    my @current;
    my $key = "";
    my $events = 0;

    my $flush = sub {
        return unless @current;
        $ins_event->execute(
            $current[0]{type}, $current[0]{value},
            $current[0]{ts}, $current[-1]{ts}, scalar @current,
        );
        my $event_id = $dbh->last_insert_id("", "", "events", "");
        $ins_member->execute($event_id, $_->{id}) for @current;
        $events++;
        @current = ();
    };

    while (my $row = $sth->fetchrow_hashref) {
        my $this_key = "$row->{type}\0$row->{value}";
        if ($this_key ne $key || (@current && $row->{ts} - $current[-1]{ts} > $gap)) {
            $flush->();
            $key = $this_key;
        }
        push @current, $row;
    }
    $flush->();
    $dbh->commit;
    return $events;
}

Three things worth pointing at. The ORDER BY type, value, ts does the hard part — once occurrences of one entity are contiguous and time-ordered, "is this still the same episode" is a one-line comparison against the previous row. A closure captures the accumulating state (@current, $key, $events) so $flush can be called from two places — inside the loop and once after it — without duplicating the insert logic, which is the same technique the pipeline's stage closures used in Milestone 4. And the null byte \0 as a key separator is a small, common Perl idiom: it cannot appear in normal text, so "$type\0$value" can never collide the way "$type$value" could for type="ip4", value="00.example" versus type="ip", value="400.example".

my (@current, $key, $events) = ((), "", 0);   # looks reasonable; is not

my @current;
my $key = "";
my $events = 0;

Run against 100,000 synthetic occurrences (four entity types, one hour, uniformly scattered — the kind of input that makes a lot of short-lived events rather than a few long ones):

$ ./bin/strata correlate strata.db --gap 60
built 55,744 events from 100,000 occurrences in 1.35s

SQL window functions (typical)                  Perl sweep
────────────────────────────────                 ───────────
SELECT *, ts - LAG(ts) OVER (                    ORDER BY type, value, ts;
  PARTITION BY type, value ORDER BY ts)          -- then a five-line loop that
  AS gap FROM occurrences;                       -- flushes @current on a gap
-- then a second query to turn "gap > 60"      -- or a change of key
-- into contiguous group ids ...

Indexes, and reading the plan instead of guessing

The sweep's ORDER BY and every later point lookup by entity both want the same thing: rows for one (type, value) pair, already close together. One index serves both:

CREATE INDEX idx_occ_type_value_ts ON occurrences(type, value, ts);

SCAN occurrences
USE TEMP B-TREE FOR LAST 2 TERMS OF ORDER BY

SEARCH occurrences USING COVERING INDEX idx_occ_type_value_ts (type=?)

                              no index     with index
ORDER BY sweep (25,064 rows)   0.0466s      0.0172s     (2.7x)
point lookup by value (6 rows) 0.01067s     0.00020s    (53x)

Correlating across formats

The payoff for all of Milestone 6's format plumbing arrives here, as one join:

my $sth = $dbh->prepare(<<'SQL');
SELECT r.source, r.file, r.lineno, r.raw, o.ts
FROM occurrences o JOIN records r ON r.id = o.record_id
WHERE o.type = ? AND o.value = ?
ORDER BY o.ts
SQL
$sth->execute("ipv4", "10.0.5.100");

That query does not know or care whether a given row came from an Apache log, a JSON event stream or a CSV export — occurrences and records only ever talk about types and values, which is precisely the flattening Milestone 6's parsers were built to produce. An address seen in a CSV export at 13:41 and in an Apache log ninety seconds later are, from this query's point of view, the same kind of row.

for my $gap (10, 60, 300, 900) {
    my $n = $store->sessionize($gap);
    my ($mean) = $dbh->selectrow_array("SELECT AVG(occurrence_count) FROM events");
    printf "gap=%-4d events=%-8d mean_occurrences=%.2f\n", $gap, $n, $mean;
    $dbh->do("DELETE FROM events"); $dbh->do("DELETE FROM event_members");
}

my $sth = $dbh->prepare(<<'SQL');
SELECT DISTINCT em1.event_id AS a, em2.event_id AS b
FROM event_members em1
JOIN occurrences o1 ON o1.id = em1.occurrence_id
JOIN occurrences o2 ON o2.record_id = o1.record_id AND o2.id != o1.id
JOIN event_members em2 ON em2.occurrence_id = o2.id
WHERE em1.event_id != em2.event_id
SQL

Experiment

Replace the sweep's composite key, "$row->{type}\0$row->{value}", with plain concatenation, "$row->{type}$row->{value}", and feed sessionize two synthetic occurrences: type ip4, value 00.example, and type ip, value 400.example. Both keys collapse to the identical string ip400.example, so two occurrences of genuinely different entity types merge into a single event. Put the \0 back and they split into two events again, correctly. This does not show up on realistic data, where type names and values rarely line up this way by chance — which is exactly why it belongs in a deliberate experiment with a constructed collision, rather than waiting to be found by a report that quietly merged two unrelated things.

Checkpoint

1. Why did wrapping the inserts in one transaction produce roughly a 700× throughput improvement, and what is SQLite actually doing once per transaction that it was doing once per row before?
2. What does a "covering" index mean, and why did it help the point lookup so much more than the sweep query?
3. Why does the sessioniser sort in SQL and group in Perl, rather than either sorting in Perl or grouping in SQL?
4. What exactly went wrong with my (@current, $key, $events) = ((), "", 0), in terms of how Perl assigns a list to a list of variables?
5. Why is "$type\0$value" a safer composite key than "$type$value"?

Milestone 10A tool that behaves like a Unix citizen

Goal

Collapse ingest, entities, correlate and query into one bin/strata with real subcommands, options that follow the conventions every other command-line tool follows, exit codes a shell script can branch on, a clean stop on Ctrl-C that does not lose committed work, and correct behaviour when its output is piped into something that stops reading early.

Concepts

Getopt::Long (bundling, negation, per-subcommand option sets), exit code conventions, cleanup on SIGINT, and the signal every Perl programmer eventually meets by surprise: SIGPIPE.

Design

A dispatch table from subcommand name to handler, and exit codes worth naming rather than spelling as bare numbers scattered through the code:

Implementation

Subcommands, dispatched from one table

use Getopt::Long qw(GetOptionsFromArray);

use constant { EX_OK => 0, EX_USAGE => 2, EX_DATA => 65, EX_INTERRUPT => 130 };

my %COMMANDS = (
    ingest    => \&cmd_ingest,
    entities  => \&cmd_entities,
    correlate => \&cmd_correlate,
    query     => \&cmd_query,
);

sub main (@argv) {
    my $cmd = shift @argv;
    return print(usage()), EX_OK if !defined($cmd) || $cmd =~ /^(-h|--help)$/;

    my $handler = $COMMANDS{$cmd} or do {
        say STDERR "strata: unknown command '$cmd'";
        print STDERR usage();
        return EX_USAGE;
    };
    return $handler->(@argv);
}

exit main(@ARGV);

Each subcommand parses its own slice of @ARGV with GetOptionsFromArray rather than the whole program sharing one option set — which is what lets strata ingest --workers 4 and a hypothetical future strata query --workers (a different meaning entirely) coexist without collision. This is one clear difference from how a typical scripting language does it: Python's argparse has first-class subparsers for exactly this; Perl's core tooling expects you to slice @ARGV yourself, which is three extra lines and total control over what each subcommand sees.

sub cmd_ingest (@argv) {
    my %opt = (format => undef, recursive => 0, workers => 1);
    GetOptionsFromArray(\@argv,
        "format=s"    => \$opt{format},
        "recursive|r" => \$opt{recursive},
        "workers=i"   => \$opt{workers},
    ) or return EX_USAGE;

    return EX_USAGE, say(STDERR "strata ingest: no files given") if !@argv;
    ...
    return EX_OK;
}

GetOptionsFromArray returning false means it already printed its own "unknown option" message to STDERR; the handler's job is only to translate that into the right exit code. "recursive|r" accepts both --recursive and -r, and because it is a plain flag (no =s/=i), --no-recursive works automatically too — Getopt::Long negates any flag-style option for free.

Stopping cleanly on Ctrl-C

An ingest of a large directory can run for minutes. Killing it should not throw away work already committed, and — more subtly — should not leave a transaction half-open either.

my $interrupted = 0;
local $SIG{INT} = sub { $interrupted = 1 };   # just set a flag; do nothing risky here

my $ins = $dbh->prepare("INSERT INTO records (id, source, file, lineno, ts, raw) VALUES (?,?,?,?,?,?)");
my $n = 0;
for my $rec (@records) {
    $ins->execute(@{$rec}{qw(id source file lineno ts raw)});
    $n++;
    if ($interrupted) {
        $dbh->commit;
        say STDERR "interrupted after $n records, committed cleanly";
        exit EX_INTERRUPT;
    }
}
$dbh->commit;

Verified by actually sending the signal mid-ingest, not by reading the code and hoping:

$ ./bin/strata ingest strata.db --workers 1 share/fixtures/giant-synthetic/*
^C
interrupted after 8000 records, committed cleanly
$ echo $?
130
$ sqlite3 strata.db "SELECT COUNT(*) FROM records"
8000

Eight thousand rows sent, Ctrl-C pressed, eight thousand rows found in the database afterwards — not seven thousand, not a half-written eight-thousand-and-first row. That is the whole point of doing the commit inside the interrupt path instead of relying on whatever happened to be true when the process died.

The signal every Perl programmer meets by surprise

$ ./bin/strata query strata.db --type ipv4 | head -3
line 1
line 2
line 3
$ echo "${PIPESTATUS[0]}"
141

$SIG{PIPE} = 'IGNORE';

for my $row (@rows) {
    unless (say $row) {
        say STDERR "stopped writing at record $.: $!" if $ENV{STRATA_DEBUG};
        last;    # the reader is gone; stop producing, exit 0 like head does
    }
}

$ STRATA_DEBUG=1 ./bin/strata query strata.db --type ipv4 2>err.log | head -3
line 1
line 2
line 3
$ echo "${PIPESTATUS[0]}"
0
$ cat err.log
stopped writing at record 7484: Broken pipe

--help, for free once and reused everywhere

$ ./bin/strata --help
usage: strata <command> [options] [files...]

commands:
  ingest     read files into the store
  entities   list extracted entities
  correlate  build events from occurrences
  query      look up records and events

run 'strata <command> --help' for command-specific options.

One usage() sub returning a heredoc, printed by the top-level dispatcher when no command is given and by every subcommand's own --help handling. A tool with subcommands and no --help is a tool whose interface lives only in its source code, which is fine for you this afternoon and useless for you in six months.

if ($opt{format} eq "json") {
    my $json = JSON::PP->new->canonical->encode(\@rows);
    say $json or last;   # one atomic-ish write; a partial write is still
                          # truncated JSON, but there is no half-object risk
}

sub parse_when ($str) {
    return time - $1 * 3600 if $str =~ /^(\d+)h$/;
    return time - $1 * 60   if $str =~ /^(\d+)m$/;
    return Strata::Normalize->to_epoch($str);   # falls through to Milestone 8's parser
}

Experiment

Pass --workers four instead of --workers 4 to a subcommand's option parser. GetOptionsFromArray rejects it itself — Value "four" invalid for option workers (number expected) — and returns false before the handler's own code ever runs, leaving $opt{workers} untouched at its default. The =i in "workers=i" is doing real validation, not just documentation: nothing in cmd_ingest needs to check that --workers looks like a number, because a bad value never survives long enough to reach the code that uses %opt.

Checkpoint

1. Why does the SIGINT handler only set a flag instead of committing the transaction directly?
2. What is SIGPIPE, when does the kernel send it, and what does Perl do with it by default?
3. Why does exit code 141 specifically mean "killed by SIGPIPE"?
4. Why does each subcommand call GetOptionsFromArray on its own slice of @ARGV rather than the whole program sharing one option set?
5. What does "recursive|r" give you for free that a plain "recursive" would not?

Milestone 11Attacking your own parser on purpose

Goal

Stop testing only the inputs you thought of. Build one shared contract every parser's test file already should have been using since Milestone 6, a fuzzer that mutates real fixtures and gives every attempt a hard deadline, and a profiling session that finds and fixes a genuine hot path with Devel::NYTProf.

Concepts

Shared test helpers across t/ files, alarm() as a last-resort timeout for code you do not trust, corpus-based mutation, and reading a line-level profile instead of guessing where the time goes.

Design

Implementation

One contract, every parser test file reuses it

Milestone 6's "why" box named this debt directly: nothing checked that a parser plugin actually implements the contract, so a missing method would fail at run time, in production, on the one file that needed it. The fix is a shared subtest that every parser's own test file calls once:

package Test::Strata::ParserContract;
use v5.36;
use Test::More;
use Exporter 'import';
our @EXPORT_OK = qw(parser_contract_ok);

sub parser_contract_ok ($class, %opt) {
    subtest "$class satisfies the parser contract" => sub {
        can_ok($class, qw(new name mode detect));
        my $mode = $class->new->mode;
        ok(($mode eq "line" && $class->can("parse"))
        || ($mode eq "stream" && $class->can("parse_handle")),
            "implements the method its declared mode requires");

        my $score = eval { $class->detect($opt{sample} // "") };
        ok(!$@, "detect() does not die on an empty string") or diag $@;
        ok($score >= 0 && $score <= 1, "detect() returns a score in [0,1], got " . ($score // "undef"));
    };
}
1;

Every parser's test file shrinks to one call plus its format-specific cases:

use Test::Strata::ParserContract qw(parser_contract_ok);
parser_contract_ok("Strata::Parser::Apache", sample => $fixture_line);
# ...then the Apache-specific assertions, as before

Adding a sixth parser next year gets this contract for the price of one line, instead of for the price of remembering to write it.

A fuzzer with a deadline

Corpus-based mutation: start from real fixture lines, apply small random edits, and give each attempt a hard wall-clock limit. Anything that does not return in time is treated as a finding, whether it is an infinite loop or merely pathological.

sub run_with_timeout ($code, $arg, $seconds = 1) {
    my $result;
    my $ok = eval {
        local $SIG{ALRM} = sub { die "timeout\n" };
        alarm($seconds);
        $result = $code->($arg);
        alarm(0);            # cancel the alarm; we finished in time
        1;
    };
    if (!$ok) {
        alarm(0);             # belt and braces: cancel it even on the die path
        die $@ unless $@ eq "timeout\n";
        return (undef, "timeout");
    }
    return ($result, undef);
}

sub mutate ($s) {
    my @chars = split //, $s;
    my @alphabet = (",", '"', "a", "1");
    my $op = int rand 3;
    if    ($op == 0)          { splice @chars, int(rand(@chars + 1)), 0, $alphabet[int rand @alphabet] }
    elsif ($op == 1 && @chars) { splice @chars, int(rand @chars), 1 }
    elsif (@chars)             { $chars[int rand @chars] = $alphabet[int rand @alphabet] }
    return join "", @chars;
}

Run against a hand-rolled quoted-field scanner (written to show what you would be signing up for by not using Text::CSV, which does not have this bug):

sub scan_fields ($line) {          # BUG, left in deliberately: see below
    my @fields;
    my $pos = 0;
    my $len = length $line;
    while ($pos < $len) {
        if (substr($line, $pos, 1) eq '"') {
            my $end = index($line, '"', $pos + 1);
            if ($end == -1) {
                next;             # meant "consume to end of string"; forgot to move $pos
            }
            push @fields, substr($line, $pos + 1, $end - $pos - 1);
            $pos = $end + 1;
        } else {
            my $comma = index($line, ",", $pos);
            $comma = $len if $comma == -1;
            push @fields, substr($line, $pos, $comma - $pos);
            $pos = $comma + 1;
        }
    }
    return \@fields;
}

$ perl t/90-fuzz.t
trial 5 hung the scanner on: "qaed,,pli
found a hang in 2000-trial budget (seed 20260912), 1.01s elapsed
not ok 1 - scan_fields never hangs

Five mutations of '"quoted",plain' in under a second produced a string starting with an unterminated " and no closing quote anywhere in it — exactly the input that hits the next without advancing $pos, so the while condition never changes and the loop spins forever. Without the deadline, this test would simply never finish, and depending on your CI system, "the test suite hangs" and "the test suite is slow today" look identical for the first twenty minutes. The fuzzer's actual job is not finding the bug — a code reviewer could find this one by eye. Its job is finding it in one second, automatically, every time the suite runs, forever. The fix is the one-line version of the comment: $pos = $len; next; when no closing quote exists, which is exactly why this project uses Text::CSV for the real parser and keeps this one only as a cautionary exercise.

typical unit tests                              this milestone's fuzzer
────────────────────                             ─────────────────────
is scan_fields(""), [], "empty";                 for (1 .. 2000) {
is scan_fields('"a",b'), ["a","b"];                   my $mutant = mutate($seed);
is scan_fields('bad"quote'), ...;  # you have         run_with_timeout(\&scan_fields,
                                    # to think of          $mutant, 1);
                                    # this case first }

Profiling: finding the hot path instead of guessing at it

strata entities on a busy log spends real time deduplicating candidate entity values before ranking them. The first version used grep against an accumulator array:

sub dedupe_seen (@values) {
    my @seen;
    my @unique;
    for my $v (@values) {
        push(@unique, $v), push(@seen, $v) unless grep { $_ eq $v } @seen;
    }
    return @unique;
}

$ perl -d:NYTProf bin/strata entities strata.db --dump-candidates > /dev/null
$ nytprofcsv && grep dedupe_seen -r nytprof/ | sort -t, -k1 -rn | head -1
0.185629,20000,0.000009,push(@unique, $v), push(@seen, $v) unless grep { $_ eq $v } @seen;

That line alone accounted for 0.186 of the run's roughly 0.19 seconds: essentially the entire program. nytprofhtml gives the same finding as a browsable, colour-coded report; the CSV export above is the same numbers in a form worth quoting. The shape is the giveaway even before profiling: grep over an array that grows by one every iteration is O(n·u) where u is the number of unique values found so far — for 20,000 values collapsing to 200 unique ones, that is up to four million string comparisons for what should be twenty thousand hash lookups.

sub dedupe_seen (@values) {
    my %seen;
    my @unique;
    for my $v (@values) {
        push @unique, $v unless $seen{$v}++;
    }
    return @unique;
}

grep-based dedupe (O(n·u)): 0.178s
hash-based dedupe (O(n)):   0.004s
speedup: 50x

for my $bad (undef, "") {
    my (undef, $err) = run_with_timeout(sub ($v) {
        $class->can("parse") ? $class->new->parse($v // "", file => "t", lineno => 1)
                              : 1;  # stream-mode parsers are exercised via a fixture elsewhere
    }, $bad, 1);
    is $err, undef, "$class does not hang on " . ($bad // "undef");
}

Experiment

Restrict mutate to only ever delete a character — comment out the insert and replace branches — and rerun the fuzzer against scan_fields with the same seed. It still finds the hang, typically within the first ten trials, because deleting the seed's own closing quote is already enough to produce an unterminated string; the bug does not need insertion or replacement to trigger. That is worth knowing before trusting a fuzzer's silence on a different bug: a narrower set of mutation operators finding a fault tells you that fault does not need the operators you left out, but a narrower set finding nothing tells you far less than a wider one would.

Checkpoint

1. What does the shared parser contract test actually check, and why did it belong in Milestone 6 and only arrive now?
2. Why must $SIG{ALRM} be localised inside the timeout helper?
3. Name both places alarm(0) has to be called in run_with_timeout, and explain what breaks if either is missing.
4. Why is corpus-based mutation (starting from real fixtures) generally more effective than mutating random bytes from nothing?
5. What made the grep-based dedupe O(n·u), and why does a hash make it O(n)?
6. Why did caching the timestamp regex with qr// fail to produce a measurable speedup, when the intuition said it should?

Milestone 12The knowledge graph

Goal

Turn correlated events into a graph — an edge between two events that share an entity — queryable by "hops", parallelise ingestion of many files at once using fork, and package the finished tool as something installable.

Concepts

Recursive common table expressions (WITH RECURSIVE), fork() and pipes as Perl's idiomatic answer to "run this on several cores", reaping children correctly, and final packaging.

Design

The graph is built from data that already exists — events and event_members from Milestone 9 — so this milestone adds exactly two new pieces, and the question each one answers is worth stating before the code:

Edges are materialised, not computed on the fly. The alternative — join event_members against itself at query time, every time someone asks "what is connected to this event" — repeats the same expensive join on every query instead of once when the data is built. The cost of that choice is staleness: an edge computed now will not reflect an event correlated five minutes from now until correlate runs again, which is an acceptable trade for a forensic tool that analyses a snapshot, and a bad one for a system that needs to answer in real time as new data arrives.

The walk itself lives entirely in SQL, which is the opposite of Milestone 9's split. There, the rule was "let SQLite sort, let Perl group", because a linear sweep over already-sorted rows is awkward to express as a single SQL query but trivial as a Perl loop. Graph reachability is the reverse case: SQLite has had recursive common table expressions since 2014 specifically for bounded graph walks, complete with built-in cycle handling through UNION's deduplication, so pulling the whole edges table into a Perl hash and hand-rolling a breadth-first search with a visited set would be reimplementing, in a general-purpose language, exactly the algorithm the database already has a declarative name for. The lesson from Milestone 9 was never "SQL sorts, Perl groups" as a fixed rule — it was "match the tool to the shape of the specific problem", and here the shapes point in opposite directions.

Implementation

Edges: when two events share evidence

CREATE TABLE edges (
    event_a INTEGER NOT NULL REFERENCES events(id),
    event_b INTEGER NOT NULL REFERENCES events(id),
    reason  TEXT NOT NULL
);

-- Two events are linked when one of their occurrences came from the same
-- record: the same log line that mentioned this IP also mentioned that path.
INSERT INTO edges (event_a, event_b, reason)
SELECT DISTINCT em1.event_id, em2.event_id, 'shared record'
FROM event_members em1
JOIN occurrences o1 ON o1.id = em1.occurrence_id
JOIN occurrences o2 ON o2.record_id = o1.record_id AND o2.id != o1.id
JOIN event_members em2 ON em2.occurrence_id = o2.id
WHERE em1.event_id < em2.event_id;   -- one row per pair, not two

em1.event_id < em2.event_id is the whole trick for storing an undirected edge once instead of twice: without it, every shared-record pair would produce both (3, 9) and (9, 3), doubling storage and every count derived from it.

Walking the graph with a recursive query

WITH RECURSIVE reachable(node, hops) AS (
    SELECT ?, 0
    UNION
    SELECT CASE WHEN e.event_a = r.node THEN e.event_b ELSE e.event_a END, r.hops + 1
    FROM edges e JOIN reachable r ON e.event_a = r.node OR e.event_b = r.node
    WHERE r.hops < ?
)
SELECT node, MIN(hops) AS hops FROM reachable GROUP BY node ORDER BY hops, node;

Verified against a small hand-built graph (edges 1–2, 2–3, 3–4, 5–6, 1–7), asking for everything within two hops of node 1:

node 1 at 0 hop(s)
node 2 at 1 hop(s)
node 7 at 1 hop(s)
node 3 at 2 hop(s)

Node 4 and the disconnected pair 5–6 correctly do not appear. WITH RECURSIVE has been in SQLite since 3.8.3 (2014), so this needs nothing beyond an ordinary SQLite install. The plain UNION (not UNION ALL) matters: it deduplicates identical (node, hops) pairs as the recursion runs, which is what stops a graph with cycles — this one included, 1→2→1 is a cycle — from recursing forever. GROUP BY node, MIN(hops) in the outer query then collapses a node that was reached by two different paths (at possibly different hop counts) down to its shortest distance, the way node 1 itself, reachable from itself in zero hops, does not also appear again at two hops even though the raw recursion does produce that row internally.

Parallel ingestion with fork and pipes

use Storable qw(freeze thaw);

sub ingest_parallel ($files, $workers = 4) {
    my @chunks;
    push @{ $chunks[$_ % $workers] }, $files->[$_] for 0 .. $#$files;

    my %reader_for;
    for my $w (0 .. $workers - 1) {
        pipe(my $reader, my $writer) or die "pipe: $!";
        my $pid = fork // die "fork: $!";

        if ($pid == 0) {
            close $reader;
            my ($records, $problems) = (0, 0);
            for my $file (@{ $chunks[$w] // [] }) {
                my ($r, $p) = ingest_one_file($file);   # each child: its own memory, own DB handle
                $records += $r;
                $problems += $p;
            }
            print $writer freeze({ worker => $w, records => $records, problems => $problems });
            close $writer;
            exit 0;
        }

        close $writer;
        $reader_for{$pid} = $reader;
    }

    my ($total_records, $total_problems) = (0, 0);
    for my $pid (keys %reader_for) {
        local $/;                                  # slurp mode for this read
        my $data = readline($reader_for{$pid});
        waitpid($pid, 0);                          # reap; see below
        my $result = thaw($data);
        $total_records  += $result->{records};
        $total_problems += $result->{problems};
    }
    return ($total_records, $total_problems);
}

Verified with six fake files split across three workers:

worker 0: 1000 records, 4 problems, files=a.log d.log
worker 1: 1000 records, 4 problems, files=b.log e.log
worker 2: 1000 records, 4 problems, files=c.log f.log
total: 3000 records, 12 problems

pipe creates a connected reader/writer pair before the fork, which is the only way it works: after fork, parent and child each have their own copies of both ends, and each side closes the end it does not use so that a read on an exhausted pipe correctly reports end-of-file rather than blocking forever waiting for a writer that will never write (because it is the child's own, unclosed, unused copy). Storable::freeze/thaw is Perl's standard way to send a structured value (here, a hashref) through something that only carries bytes.

Packaging the finished distribution

$ perl Makefile.PL
$ make test
PASS  t/00-load.t
PASS  t/10-record.t
...
PASS  t/90-fuzz.t
All tests successful.
$ make dist
strata-0.12.tar.gz
$ cpanm --local-lib=/tmp/check strata-0.12.tar.gz && echo "installs cleanly"
installs cleanly

make dist reads MANIFEST (generated by make manifest from every file under version control) and produces exactly the tarball a real user would cpanm install — installing your own tarball into a scratch local::lib before calling a milestone "done" catches the class of bug that only "it works on my checkout" hides: a test file depending on a fixture that was never added to MANIFEST, a module that only compiles because something else in your working tree happened to load it first.

Repository state after Milestone 12

strata/
├── Makefile.PL, cpanfile, MANIFEST
├── bin/strata                   ingest | entities | correlate | query
├── lib/Strata.pm
├── lib/Strata/
│   ├── Util.pm  Pattern.pm  Record.pm  Pipeline.pm  Source.pm
│   ├── Extract.pm  Normalize.pm
│   ├── Store.pm                  DBI, schema, transactions
│   ├── Correlate.pm               sessionisation sweep
│   ├── Graph.pm                    edges, recursive CTE queries
│   └── Parser/
│       ├── Registry.pm  Apache.pm  JsonLines.pm  Csv.pm  Xml.pm  Unstructured.pm
└── t/
    ├── lib/Test/Strata/ParserContract.pm
    ├── 00–80  (unit and integration tests, ~10 files)
    └── 90-fuzz.t                          deadline-bounded mutation tests

$ prove -l t/
All tests successful.
$ git commit -am "milestones 9-12: correlation, a real CLI, fuzzing, profiling, the graph"

say "graph {";
my $rows = $dbh->selectall_arrayref("SELECT event_a, event_b, reason FROM edges");
for my $row (@$rows) {
    say qq{  "$row->[0]" -- "$row->[1]" [label="$row->[2]"];};
}
say "}";

Experiment

Call ingest_parallel with more workers than files — six files split across eight workers. Two workers get an empty chunk and report zero files back over their pipe; the other six each get exactly one. Nothing hangs and nothing dies: $chunks[$_ % $workers] // [] turns "no work assigned" into an empty list to iterate over instead of dereferencing undef, so an idle worker still writes a well-formed result, still gets reaped by waitpid, and every file is still accounted for exactly once in the total.

Checkpoint

1. Why is a data race structurally impossible between two forked Perl worker processes, in a way it is not for two goroutines?
2. What does pipe have to be called before, not after, and why does that ordering matter?
3. What is a zombie process, and what stops one from being created here?
4. Why can this milestone's fork-and-pipe design never deadlock on a full pipe buffer, and what changes the moment you add a second direction of traffic?
5. Why does UNION rather than UNION ALL matter for a recursive query over a graph that contains a cycle?
6. What does installing your own freshly built tarball into a scratch local::lib catch that running the test suite in your working tree does not?