Multi-Pattern Log Scanner

Requirements and Signature Model

Use one explicit symbol contract. This design scans Unicode code points after NFC normalization and optional casefold(), applying the identical policy to signatures and input. Offsets are half-open positions in the normalized stream. If consumers require original byte spans, the decoder must preserve a provenance map because normalization and case folding can change length.

Each signature has a stable ID, non-empty pattern, severity, tags, and JSON metadata. Duplicate normalized patterns are allowed and produce separate records; duplicate IDs are rejected. Matches remain ordered, overlapping, nested, and co-terminal. Incremental UTF-8 decoding prevents split characters, while exact streaming NFC additionally requires retaining the unresolved final starter and combining marks or normalizing complete logical records.

Record	Field	Type	Required	Semantics	Validation
Signature	`id`	String	Yes	Stable external identifier	Non-empty and unique per dictionary
Signature	`pattern`	String	Yes	Text compiled into the automaton	Non-empty after normalization
Signature	`severity`	String	Yes	Routing or priority class	Controlled deployment value
Signature	`tags`	List of strings	No	Classification labels	Deduplicate while preserving order
Signature	`metadata`	JSON object	No	Owner, source, revision, response data	JSON-serializable
Dictionary	`normalization`	String	Yes	Unicode normalization policy	`NFC` here
Dictionary	`casefold`	Boolean	Yes	Unicode case policy	Identical at compile and scan time
Match	`signature_id`	String	Yes	Matched rule identifier	Copied from compiled signature
Match	`start`	Integer	Yes	Inclusive normalized offset	Non-negative
Match	`end`	Integer	Yes	Exclusive normalized offset	Greater than `start`
Match	`pattern`	String	Yes	Original authored signature	Retained for audit and display
Match	`severity`	String	Yes	Signature priority	Immutable for a dictionary version
Match	`tags`	List of strings	Yes	Signature classifications	Copied without mutation
Match	`metadata`	JSON object	Yes	Application-defined context	Copied or exposed read-only

Automaton Compilation

Insert normalized signatures into a trie, then traverse it breadth-first to assign failure links. A failure link points to the longest proper suffix represented by another state. Each node inherits terminal outputs from its failure target, so reaching she also reports he. Sparse transitions keep memory proportional to dictionary structure; scanning follows failure links on missing edges.

Serialization captures the normalization policy, signatures, pattern lengths, transitions, failure links, and outputs as a versioned unit. Validate every index when loading and never use a general object serializer for untrusted automata.

from __future__ import annotations
from collections import deque
from dataclasses import asdict, dataclass, field
import json
import unicodedata
from typing import Any


@dataclass(frozen=True)
class Signature:
    id: str
    pattern: str
    severity: str = "info"
    tags: tuple[str, ...] = ()
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class Match:
    signature_id: str
    start: int
    end: int
    pattern: str
    severity: str
    tags: tuple[str, ...]
    metadata: dict[str, Any]


@dataclass
class Node:
    next: dict[str, int] = field(default_factory=dict)
    fail: int = 0
    outputs: list[int] = field(default_factory=list)


class AhoCorasick:
    VERSION = 1

    def __init__(self, signatures: list[Signature], *, casefold: bool = False):
        if not signatures:
            raise ValueError("at least one signature is required")
        if len({item.id for item in signatures}) != len(signatures):
            raise ValueError("signature ids must be unique")
        self.signatures = signatures.copy()
        self.casefold = casefold
        self.nodes = [Node()]
        self.pattern_lengths: list[int] = []
        self._compile()

    def normalize(self, text: str) -> str:
        value = unicodedata.normalize("NFC", text)
        if self.casefold:
            value = unicodedata.normalize("NFC", value.casefold())
        return value

    def _compile(self) -> None:
        for output, signature in enumerate(self.signatures):
            if not signature.id:
                raise ValueError("signature id must be non-empty")
            pattern = self.normalize(signature.pattern)
            if not pattern:
                raise ValueError("normalized pattern must be non-empty")
            self.pattern_lengths.append(len(pattern))
            state = 0
            for symbol in pattern:
                if symbol not in self.nodes[state].next:
                    self.nodes[state].next[symbol] = len(self.nodes)
                    self.nodes.append(Node())
                state = self.nodes[state].next[symbol]
            self.nodes[state].outputs.append(output)

        queue: deque[int] = deque(self.nodes[0].next.values())
        while queue:
            state = queue.popleft()
            for symbol, child in self.nodes[state].next.items():
                queue.append(child)
                fallback = self.nodes[state].fail
                while fallback and symbol not in self.nodes[fallback].next:
                    fallback = self.nodes[fallback].fail
                self.nodes[child].fail = self.nodes[fallback].next.get(symbol, 0)
                self.nodes[child].outputs.extend(
                    self.nodes[self.nodes[child].fail].outputs
                )

    def scanner(self) -> Scanner:
        return Scanner(self)

    def to_json(self) -> str:
        return json.dumps({
            "version": self.VERSION,
            "normalization": "NFC",
            "casefold": self.casefold,
            "signatures": [{**asdict(s), "tags": list(s.tags)} for s in self.signatures],
            "pattern_lengths": self.pattern_lengths,
            "nodes": [asdict(node) for node in self.nodes],
        }, ensure_ascii=False, separators=(",", ":"))

    @classmethod
    def from_json(cls, encoded: str) -> AhoCorasick:
        data = json.loads(encoded)
        if data.get("version") != cls.VERSION or data.get("normalization") != "NFC":
            raise ValueError("unsupported automaton format")
        instance = cls.__new__(cls)
        instance.casefold = bool(data["casefold"])
        instance.signatures = [Signature(
            item["id"], item["pattern"], item["severity"],
            tuple(item["tags"]), item["metadata"],
        ) for item in data["signatures"]]
        instance.pattern_lengths = [int(x) for x in data["pattern_lengths"]]
        instance.nodes = [Node(
            {str(k): int(v) for k, v in item["next"].items()},
            int(item["fail"]), [int(x) for x in item["outputs"]],
        ) for item in data["nodes"]]
        count, outputs = len(instance.nodes), len(instance.signatures)
        if not count or len(instance.pattern_lengths) != outputs:
            raise ValueError("invalid dimensions")
        for node in instance.nodes:
            if not 0 <= node.fail < count:
                raise ValueError("invalid failure link")
            if any(not 0 <= target < count for target in node.next.values()):
                raise ValueError("invalid transition")
            if any(not 0 <= output < outputs for output in node.outputs):
                raise ValueError("invalid output")
        return instance


class Scanner:
    def __init__(self, automaton: AhoCorasick):
        self.automaton = automaton
        self.state = 0
        self.position = 0

    def feed_normalized(self, text: str) -> list[Match]:
        matches: list[Match] = []
        for symbol in text:
            while self.state and symbol not in self.automaton.nodes[self.state].next:
                self.state = self.automaton.nodes[self.state].fail
            self.state = self.automaton.nodes[self.state].next.get(symbol, 0)
            end = self.position + 1
            for index in self.automaton.nodes[self.state].outputs:
                signature = self.automaton.signatures[index]
                matches.append(Match(
                    signature.id, end - self.automaton.pattern_lengths[index], end,
                    signature.pattern, signature.severity, signature.tags,
                    signature.metadata.copy(),
                ))
            self.position = end
        return matches


if __name__ == "__main__":
    automaton = AhoCorasick([
        Signature("he", "he"), Signature("she", "she"),
        Signature("hers", "hers"), Signature("his", "his"),
        Signature("duplicate-he", "he"),
    ])
    scanner = automaton.scanner()
    matches = scanner.feed_normalized("us") + scanner.feed_normalized("hers")
    assert [(m.signature_id, m.start, m.end) for m in matches] == [
        ("she", 1, 4), ("he", 2, 4),
        ("duplicate-he", 2, 4), ("hers", 2, 6),
    ]
    assert AhoCorasick.from_json(automaton.to_json()).scanner().feed_normalized("ushers") == matches
    overlap = AhoCorasick([Signature("a", "a"), Signature("aa", "aa"), Signature("aaa", "aaa")])
    assert len(overlap.scanner().feed_normalized("aaaa")) == 9

Streaming Chunk Boundaries

Two states cross each chunk boundary: the incremental UTF-8 decoder retains incomplete bytes, and the scanner retains its automaton state and normalized position. Never decode chunks independently with replacement characters or reset the automaton. A pattern may begin in one chunk and finish in another.

NFC adds a third boundary concern because a starter in one chunk may compose with combining marks in the next. Scan complete normalized records or retain an unresolved normalization suffix. At end of stream, finalize the decoder, reject malformed trailing bytes according to policy, flush normalization carry, and then complete scanning.

Reporting, Backpressure, and Deduplication

Scanning $N$ symbols costs $O(N+Z)$ for $Z$ outputs. A repetitive stream and nested signatures can overwhelm a slower sink, so place a bounded queue after scanning. On saturation, explicitly pause reads, spill durable batches, aggregate counts, or drop declared low-priority records while recording loss—never grow an unbounded list.

Distinct IDs sharing a pattern remain distinct matches. For retry-safe delivery, use an idempotency key such as (stream_id, dictionary_version, signature_id, start, end). Checkpoint decoder bytes, normalization carry, automaton state, normalized position, and source cursor atomically.

Streaming scanner pipeline

flowchart LR
    A["Ordered byte chunks"] --> B["Incremental UTF-8 decoder"]
    B --> C["Normalization and offset map"]
    C --> D["Stateful Aho-Corasick scan"]
    D --> E{"Bounded match queue"}
    E -- "capacity" --> F["Batch and deduplicate"]
    F --> G["Idempotent sink"]
    E -- "high-water mark" --> H["Pause input or spill"]
    H --> E
    D --> I["Atomic checkpoint"]

Validation and Performance

Compare every streaming run with a batch scan using the same normalization contract. Split multibyte characters and every signature at every possible boundary; include suffix-related duplicates, malformed UTF-8 policies, combining marks, checkpoint restores, and alternating sink backpressure.

Measure compilation separately from scanning. Report pattern bytes, states, transitions, serialized size, compile time, decoded throughput, transitions per symbol, outputs per symbol, queue high-water mark, and sink throughput. Include both low-output corpora and output storms because reporting can dominate automaton work.

Load Test

Overlap Storm

Compile signatures a, aa, …, a repeated 32 times, all with distinct IDs. Stream 100,000 a symbols using chunk sizes cycling through 1, 7, 64, and 4096 bytes. Position $i$ emits $\min(i+1,32)$ records, for

$$ \frac{32\cdot33}{2}+32(100{,}000-32)=3{,}199{,}504 $$

total matches.

Use a queue bounded at 1,024 records and sink batches of 256. Verify batch and streaming triples are identical; chunk crossings lose no matches; the queue never exceeds capacity; reads pause at the high-water mark; checkpoint replay creates no duplicate sink records under idempotency keys; and metrics distinguish 100,000 consumed symbols from 3,199,504 emitted records. Repeat with reporting disabled and with count-only aggregation to isolate transition throughput from output cost.

Cookie Consent