Add integrated, scaling, and per-record benchmarks; correct README perf claims

README.md

@@ -106,12 +106,19 @@ Measured on a single core (AMD Ryzen 7 3800X, `make bench`):
 | Path | Per record | Throughput | Allocations |
 |---|---|---|---|
 | Rules stage, first-rule hit | 46 ns | ~22M records/sec | 0 |
-| Rules stage, worst case (20-rule walk, regex-heavy) | 7-8 µs | ~130k records/sec | 0 |
+| Rules stage, 20-rule regex-heavy miss (benchmark ruleset) | 7-8 µs | ~130k records/sec | 0 |
 | Full pipeline (engine + rules, discard sink) | 500-550 ns | ~1.8M records/sec | 0 |
+| Integrated pipeline (text source -> rules -> JSONL sink) | ~600 ns | ~1.6M records/sec | 2 |
 
 Per-record cost depends on your ruleset: order rules by expected volume so the
-hot path exits early. The pipeline row isolates engine + rules behind a discard
-sink; the JSONL output path is benchmarked separately in `sink/jsonl`.
+hot path exits early. The miss row is the benchmark ruleset's cost, not a
+general bound: a pattern with no extractable required literal defeats the
+prefilter and every such rule pays its full regex on a miss
+(`BenchmarkClassifyRegexMissUnprefilterable`: ~101 µs at 20 rules on the same
+machine). The pipeline row isolates engine + rules behind a discard sink; the
+integrated row adds the source read and JSON encode every CLI run pays. The
+structured path (JSONL source -> field rules -> structured output) is
+benchmarked in `shared/pkg/engine` as well.
 
 The comparison that motivates the cascade: classifying 1M short log lines with
 a budget LLM API (~25 input + 5 output tokens each at $0.15/$0.60 per million

shared/pkg/engine/parallel_perrecord_bench_test.go

@@ -0,0 +1,48 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/ClassMesh/classmesh/shared/pkg/domain"
+	"github.com/ClassMesh/classmesh/shared/pkg/stage"
+)
+
+func BenchmarkEnginePerRecordAllocs(b *testing.B) {
+	payload := []byte(`2026-06-12T10:00:00Z WARN payment declined order=84712 user=991 amount=49.90`)
+	cases := []struct {
+		name  string
+		stage func() stage.Stage
+	}{
+		{"cheap-stage", func() stage.Stage { return &spinStage{} }},
+		{"cpu-heavy-stage", func() stage.Stage { return &spinStage{iters: 400} }},
+	}
+	for _, tc := range cases {
+		for _, workers := range []int{0, 1, 2, 4, 8, 16} {
+			b.Run(fmt.Sprintf("%s/workers=%d", tc.name, workers), func(b *testing.B) {
+				src := &benchSource{record: domain.Record{ID: "bench", Data: payload}, n: b.N}
+				e, err := New(Deps{
+					Source:  src,
+					Stages:  []stage.Stage{tc.stage()},
+					Sink:    discardSink{},
+					Logger:  discardLogger(),
+					Workers: workers,
+				})
+				if err != nil {
+					b.Fatal(err)
+				}
+				b.SetBytes(int64(len(payload)))
+				b.ReportAllocs()
+				b.ResetTimer()
+				stats, err := e.Run(context.Background())
+				if err != nil {
+					b.Fatal(err)
+				}
+				if stats.Processed != b.N {
+					b.Fatalf("processed = %d, want %d", stats.Processed, b.N)
+				}
+			})
+		}
+	}
+}

shared/pkg/engine/pipeline_bench_test.go

@@ -0,0 +1,164 @@
+package engine_test
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/ClassMesh/classmesh/shared/pkg/domain"
+	"github.com/ClassMesh/classmesh/shared/pkg/engine"
+	"github.com/ClassMesh/classmesh/shared/pkg/sink"
+	jsonlsink "github.com/ClassMesh/classmesh/shared/pkg/sink/jsonl"
+	"github.com/ClassMesh/classmesh/shared/pkg/source"
+	jsonlsrc "github.com/ClassMesh/classmesh/shared/pkg/source/jsonl"
+	"github.com/ClassMesh/classmesh/shared/pkg/source/textfile"
+	"github.com/ClassMesh/classmesh/shared/pkg/stage"
+	"github.com/ClassMesh/classmesh/shared/pkg/stage/rules"
+)
+
+// loopReader replays its data forever so a bounded source can draw an
+// arbitrary number of lines without holding them all in memory.
+type loopReader struct {
+	data []byte
+	off  int
+}
+
+func (r *loopReader) Read(p []byte) (int, error) {
+	if r.off == len(r.data) {
+		r.off = 0
+	}
+	n := copy(p, r.data[r.off:])
+	r.off += n
+	return n, nil
+}
+
+func lineReader(line string) *loopReader {
+	return &loopReader{data: bytes.Repeat([]byte(line+"\n"), 1024)}
+}
+
+// boundedSource caps an underlying source at n records so engine.Run drains
+// after exactly n records.
+type boundedSource struct {
+	inner source.Source
+	n, i  int
+}
+
+func newBounded(inner source.Source, n int) *boundedSource {
+	return &boundedSource{inner: inner, n: n}
+}
+
+func (s *boundedSource) Next(ctx context.Context) (domain.Record, error) {
+	if s.i >= s.n {
+		return domain.Record{}, source.ErrDrained
+	}
+	s.i++
+	return s.inner.Next(ctx)
+}
+
+func (s *boundedSource) Close() error { return s.inner.Close() }
+
+// realisticRules is the shape of a log-triage config: a health-check noise
+// rule the text sample hits, plus regex rules a record walks past on a miss.
+func realisticRules(tb testing.TB) *rules.Stage {
+	tb.Helper()
+	s, err := rules.New([]rules.Rule{
+		{Category: "noise", Contains: []string{"healthz", "readiness", "liveness"}},
+		{Category: "billing", Regex: []string{`payment (failed|declined)`}},
+		{Category: "auth", Contains: []string{"login failed"}, Regex: []string{`(?i)unauthorized`}},
+		{Category: "db", Regex: []string{`(connection refused|deadlock detected)`}},
+	})
+	if err != nil {
+		tb.Fatalf("rules.New() error = %v", err)
+	}
+	return s
+}
+
+func runEngine(b *testing.B, src source.Source, st stage.Stage, out sink.Sink) {
+	b.Helper()
+	e, err := engine.New(engine.Deps{Source: src, Stages: []stage.Stage{st}, Sink: out})
+	if err != nil {
+		b.Fatal(err)
+	}
+	stats, err := e.Run(context.Background())
+	if err != nil {
+		b.Fatal(err)
+	}
+	if stats.Processed != b.N {
+		b.Fatalf("processed = %d, want %d", stats.Processed, b.N)
+	}
+}
+
+// BenchmarkPipelineTextRulesJSONL is text source -> rules -> JSONL sink, the
+// single-tier CLI path (./classmesh run --rules ... file.txt) end to end. The
+// sample hits the first rule, so a classified record is encoded every time.
+func BenchmarkPipelineTextRulesJSONL(b *testing.B) {
+	line := `10.2.3.4 - - [12/Jun/2026:10:00:00] "GET /healthz HTTP/1.1" 200 2 "-" "kube-probe/1.29"`
+	src := newBounded(textfile.New(lineReader(line), "bench"), b.N)
+	sink := jsonlsink.New(io.Discard)
+	defer func() { _ = sink.Close() }()
+	st := realisticRules(b)
+
+	b.SetBytes(int64(len(line) + 1))
+	b.ReportAllocs()
+	b.ResetTimer()
+	runEngine(b, src, st, sink)
+}
+
+// BenchmarkPipelineJSONLFieldsJSONL is JSONL source -> field rules -> JSONL
+// sink, the structured path (./classmesh run --input jsonl ...) end to end:
+// decode into Fields, match on a field, re-encode with the decoded Fields.
+func BenchmarkPipelineJSONLFieldsJSONL(b *testing.B) {
+	line := `{"level":"error","http":{"status":503},"msg":"upstream timeout","user_id":"u1234"}`
+	gte := 500.0
+	st, err := rules.New([]rules.Rule{
+		{Category: "alert", Fields: []rules.FieldMatcher{{Path: "http.status", Gte: &gte}}},
+	})
+	if err != nil {
+		b.Fatal(err)
+	}
+	src := newBounded(jsonlsrc.New(lineReader(line), "bench"), b.N)
+	sink := jsonlsink.New(io.Discard)
+	defer func() { _ = sink.Close() }()
+
+	b.SetBytes(int64(len(line) + 1))
+	b.ReportAllocs()
+	b.ResetTimer()
+	runEngine(b, src, st, sink)
+}
+
+// BenchmarkPipelineSink contrasts the same text pipeline draining into
+// io.Discard against a real file. The file case measures buffered writes into
+// the OS page cache, not durable disk latency: nothing calls fsync.
+func BenchmarkPipelineSink(b *testing.B) {
+	line := `10.2.3.4 - - [12/Jun/2026:10:00:00] "GET /healthz HTTP/1.1" 200 2 "-" "kube-probe/1.29"`
+	cases := []struct {
+		name   string
+		writer func(b *testing.B) io.Writer
+	}{
+		{"discard", func(b *testing.B) io.Writer { return io.Discard }},
+		{"file", func(b *testing.B) io.Writer {
+			f, err := os.Create(filepath.Join(b.TempDir(), "out.jsonl"))
+			if err != nil {
+				b.Fatal(err)
+			}
+			b.Cleanup(func() { _ = f.Close() })
+			return f
+		}},
+	}
+	for _, tc := range cases {
+		b.Run(tc.name, func(b *testing.B) {
+			src := newBounded(textfile.New(lineReader(line), "bench"), b.N)
+			sink := jsonlsink.New(tc.writer(b))
+			defer func() { _ = sink.Close() }()
+			st := realisticRules(b)
+
+			b.SetBytes(int64(len(line) + 1))
+			b.ReportAllocs()
+			b.ResetTimer()
+			runEngine(b, src, st, sink)
+		})
+	}
+}

shared/pkg/stage/rules/rules_bench_test.go

@@ -2,10 +2,12 @@ package rules
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 
 	"github.com/ClassMesh/classmesh/shared/pkg/domain"
+	"github.com/ClassMesh/classmesh/shared/pkg/stage"
 )
 
 // benchStage builds a realistic 20-rule set: a mix of substring and regex
@@ -42,7 +44,7 @@ func benchClassify(b *testing.B, payload string) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		_, err := s.Classify(ctx, r)
-		if err != nil && err.Error() != "stage: unclassified" {
+		if err != nil && !errors.Is(err, stage.ErrUnclassified) {
 			b.Fatal(err)
 		}
 	}
@@ -95,7 +97,7 @@ func BenchmarkClassifyMissLiteral(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		_, err := s.Classify(ctx, r)
-		if err != nil && err.Error() != "stage: unclassified" {
+		if err != nil && !errors.Is(err, stage.ErrUnclassified) {
 			b.Fatal(err)
 		}
 	}

shared/pkg/stage/rules/scaling_bench_test.go

@@ -0,0 +1,86 @@
+package rules
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/ClassMesh/classmesh/shared/pkg/domain"
+	"github.com/ClassMesh/classmesh/shared/pkg/stage"
+)
+
+func runClassify(b *testing.B, s *Stage, payload string) {
+	b.Helper()
+	r := domain.Record{ID: "bench", Data: []byte(payload)}
+	ctx := context.Background()
+	b.SetBytes(int64(len(payload)))
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := s.Classify(ctx, r)
+		if err != nil && !errors.Is(err, stage.ErrUnclassified) {
+			b.Fatal(err)
+		}
+	}
+}
+
+// containsRuleset builds n substring rules whose markers never appear in the
+// benchmark payload, so a record walks the whole set before missing.
+func containsRuleset(b *testing.B, n int) *Stage {
+	b.Helper()
+	rs := make([]Rule, 0, n)
+	for i := 0; i < n; i++ {
+		rs = append(rs, Rule{
+			Category: fmt.Sprintf("svc-%d", i),
+			Contains: []string{fmt.Sprintf("service-%d-unique-marker", i)},
+		})
+	}
+	s, err := New(rs)
+	if err != nil {
+		b.Fatalf("New() error = %v", err)
+	}
+	return s
+}
+
+// BenchmarkRulesetScaling shows the linear cost of the miss walk as the
+// ruleset grows: order rules by expected volume so the hot path exits early.
+func BenchmarkRulesetScaling(b *testing.B) {
+	payload := `2026-06-12T10:00:00Z INFO order shipped id=84712 warehouse=tlv carrier=ups`
+	for _, n := range []int{20, 100, 1000} {
+		b.Run(fmt.Sprintf("rules=%d", n), func(b *testing.B) {
+			runClassify(b, containsRuleset(b, n), payload)
+		})
+	}
+}
+
+// BenchmarkClassifyMiddleRuleHit fills the gap between first- and last-rule
+// hit: the record matches rule 10 of the realistic 20-rule set.
+func BenchmarkClassifyMiddleRuleHit(b *testing.B) {
+	runClassify(b, benchStage(b), `2026-06-12T10:00:00Z INFO service-4-marker request completed duration=12ms`)
+}
+
+// unprefilterableRegexRuleset is 20 regex rules that all begin with an
+// alternation or wildcard, so no literal prefix can be extracted and the
+// prefilter cannot skip any of them: the true worst-case miss.
+func unprefilterableRegexRuleset(b *testing.B) *Stage {
+	b.Helper()
+	rs := make([]Rule, 0, 20)
+	for i := 0; i < 20; i++ {
+		rs = append(rs, Rule{
+			Category: fmt.Sprintf("re-%d", i),
+			Regex:    []string{fmt.Sprintf(`(?i)(alpha%d|beta%d|gamma%d)\d+`, i, i, i)},
+		})
+	}
+	s, err := New(rs)
+	if err != nil {
+		b.Fatalf("New() error = %v", err)
+	}
+	return s
+}
+
+// BenchmarkClassifyRegexMissUnprefilterable is the miss the literal prefilter
+// cannot help: every regex is run against the payload.
+func BenchmarkClassifyRegexMissUnprefilterable(b *testing.B) {
+	runClassify(b, unprefilterableRegexRuleset(b), `2026-06-12T10:00:00Z INFO order shipped id=84712 warehouse=tlv carrier=ups`)
+}