perf: cache inspector views to avoid redundant JSON parsing

When source matching fails the "last match" optimization and falls back
to matchAll, the same raw bytes would be parsed multiple times by the
same inspector. Now views are cached per inspector during a single
Process call.

This reduces JSON parsing from O(sources * groups) to O(inspectors) in
the worst case when messages switch between sources.

Author: bjaus

router.go

@@ -221,32 +221,68 @@ func (r *Router) Process(ctx context.Context, raw []byte) error {
 	return err
 }
 
+// viewCache caches parsed views per inspector to avoid re-parsing the same
+// raw bytes multiple times during source matching.
+type viewCache struct {
+	raw   []byte
+	views map[Inspector]viewResult
+}
+
+type viewResult struct {
+	view View
+	ok   bool
+}
+
+func newViewCache(raw []byte) *viewCache {
+	return &viewCache{
+		raw:   raw,
+		views: make(map[Inspector]viewResult),
+	}
+}
+
+// get returns a cached view or parses and caches it.
+func (c *viewCache) get(insp Inspector) (View, bool) {
+	if result, ok := c.views[insp]; ok {
+		return result.view, result.ok
+	}
+
+	view, err := insp.Inspect(c.raw)
+	if err != nil {
+		c.views[insp] = viewResult{ok: false}
+		return nil, false
+	}
+
+	c.views[insp] = viewResult{view: view, ok: true}
+	return view, true
+}
+
 // match finds a source whose discriminator matches the raw message.
 // Uses adaptive ordering to try the last successful source first.
 func (r *Router) match(raw []byte) Source {
+	cache := newViewCache(raw)
+
 	// Try last successful source first (fast path)
 	if v := r.lastMatch.Load(); v != nil {
 		if lastMatch := v.(string); lastMatch != "" {
-			if src := r.trySource(raw, lastMatch); src != nil {
+			if src := r.trySource(cache, lastMatch); src != nil {
 				return src
 			}
 		}
 	}
 
 	// Full search through all groups
-	src := r.matchAll(raw)
+	src := r.matchAll(cache)
 	if src != nil {
 		r.lastMatch.Store(src.Name())
 	}
 	return src
 }
 
 // trySource attempts to match a specific source by name.
-func (r *Router) trySource(raw []byte, name string) Source {
+func (r *Router) trySource(cache *viewCache, name string) Source {
 	// Check default sources
 	if len(r.defaultSources) > 0 {
-		view, err := r.defaultInspector.Inspect(raw)
-		if err == nil {
+		if view, ok := cache.get(r.defaultInspector); ok {
 			for _, src := range r.defaultSources {
 				if src.Name() == name && src.Discriminator().Match(view) {
 					return src
@@ -257,8 +293,8 @@ func (r *Router) trySource(raw []byte, name string) Source {
 
 	// Check custom groups
 	for _, g := range r.groups {
-		view, err := g.inspector.Inspect(raw)
-		if err != nil {
+		view, ok := cache.get(g.inspector)
+		if !ok {
 			continue
 		}
 		for _, src := range g.sources {
@@ -272,11 +308,10 @@ func (r *Router) trySource(raw []byte, name string) Source {
 }
 
 // matchAll searches all groups for a matching source.
-func (r *Router) matchAll(raw []byte) Source {
+func (r *Router) matchAll(cache *viewCache) Source {
 	// Try default group first
 	if len(r.defaultSources) > 0 {
-		view, err := r.defaultInspector.Inspect(raw)
-		if err == nil {
+		if view, ok := cache.get(r.defaultInspector); ok {
 			for _, src := range r.defaultSources {
 				if src.Discriminator().Match(view) {
 					return src
@@ -287,8 +322,8 @@ func (r *Router) matchAll(raw []byte) Source {
 
 	// Try custom groups in order
 	for _, g := range r.groups {
-		view, err := g.inspector.Inspect(raw)
-		if err != nil {
+		view, ok := cache.get(g.inspector)
+		if !ok {
 			continue
 		}
 		for _, src := range g.sources {

router_test.go

@@ -1137,3 +1137,264 @@ func TestRouter_TrySourceInspectorFailsInCustomGroup(t *testing.T) {
 
 	assert.Error(t, err)
 }
+
+// countingInspector tracks how many times Inspect is called.
+type countingInspector struct {
+	count int
+}
+
+func (c *countingInspector) Inspect(raw []byte) (View, error) {
+	c.count++
+	return JSONInspector().Inspect(raw)
+}
+
+func (c *countingInspector) reset() {
+	c.count = 0
+}
+
+type ViewCachingSuite struct {
+	suite.Suite
+}
+
+func TestViewCachingSuite(t *testing.T) {
+	suite.Run(t, new(ViewCachingSuite))
+}
+
+func (s *ViewCachingSuite) TestInspectorCalledOnceWhenTrySourceSucceeds() {
+	inspector := &countingInspector{}
+
+	r := New(WithInspector(inspector))
+
+	source := SourceFunc("test-source", HasFields("type"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil || env.Type == "" {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+	r.AddSource(source)
+	Register(r, "test", &testHandler{})
+
+	// First message - no lastMatch, goes through matchAll
+	msg := []byte(`{"type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg)
+	s.Require().NoError(err)
+
+	// Inspector should be called exactly once
+	s.Assert().Equal(1, inspector.count)
+}
+
+func (s *ViewCachingSuite) TestInspectorCalledOnceWhenTrySourceFailsAndMatchAllSucceeds() {
+	inspector := &countingInspector{}
+
+	r := New(WithInspector(inspector))
+
+	// Source A - matches "a" field
+	sourceA := SourceFunc("source-a", HasFields("a"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			A       bool            `json:"a"`
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil || !env.A {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+
+	// Source B - matches "b" field
+	sourceB := SourceFunc("source-b", HasFields("b"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			B       bool            `json:"b"`
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil || !env.B {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+
+	r.AddSource(sourceA)
+	r.AddSource(sourceB)
+	Register(r, "test", &testHandler{})
+
+	// Prime with source-a
+	msg1 := []byte(`{"a": true, "type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg1)
+	s.Require().NoError(err)
+	inspector.reset()
+
+	// Now send message that matches source-b
+	// trySource will try "source-a" first (lastMatch), fail
+	// matchAll will find "source-b"
+	// With caching, inspector should only be called ONCE
+	msg2 := []byte(`{"b": true, "type": "test", "payload": {}}`)
+	err = r.Process(context.Background(), msg2)
+
+	s.Require().NoError(err)
+	s.Assert().Equal(1, inspector.count, "inspector should only be called once due to view caching")
+}
+
+func (s *ViewCachingSuite) TestInspectorCalledOncePerGroupWithMultipleGroups() {
+	defaultInspector := &countingInspector{}
+	group1Inspector := &countingInspector{}
+	group2Inspector := &countingInspector{}
+
+	r := New(WithInspector(defaultInspector))
+
+	// Default source - won't match
+	defaultSource := SourceFunc("default", HasFields("default_field"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	})
+	r.AddSource(defaultSource)
+
+	// Group 1 source - won't match
+	group1Source := SourceFunc("group1", HasFields("group1_field"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	})
+	r.AddGroup(group1Inspector, group1Source)
+
+	// Group 2 source - will match
+	group2Source := SourceFunc("group2", HasFields("group2_field"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+	r.AddGroup(group2Inspector, group2Source)
+
+	Register(r, "test", &testHandler{})
+
+	msg := []byte(`{"group2_field": true, "type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg)
+
+	s.Require().NoError(err)
+	s.Assert().Equal(1, defaultInspector.count, "default inspector should be called once")
+	s.Assert().Equal(1, group1Inspector.count, "group1 inspector should be called once")
+	s.Assert().Equal(1, group2Inspector.count, "group2 inspector should be called once")
+}
+
+func (s *ViewCachingSuite) TestSameInspectorSharedAcrossGroupsCalledOnce() {
+	sharedInspector := &countingInspector{}
+
+	r := New(WithInspector(sharedInspector))
+
+	// Default source - won't match
+	defaultSource := SourceFunc("default", HasFields("default_field"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	})
+	r.AddSource(defaultSource)
+
+	// Custom group using the SAME inspector
+	customSource := SourceFunc("custom", HasFields("custom_field"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+	r.AddGroup(sharedInspector, customSource)
+
+	Register(r, "test", &testHandler{})
+
+	msg := []byte(`{"custom_field": true, "type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg)
+
+	s.Require().NoError(err)
+	// Same inspector used for default group and custom group - should only be called once
+	s.Assert().Equal(1, sharedInspector.count, "shared inspector should only be called once")
+}
+
+func (s *ViewCachingSuite) TestViewCacheHandlesInspectorError() {
+	failingInspector := &mockInspector{err: ErrInvalidJSON}
+	workingInspector := &countingInspector{}
+
+	r := New(WithInspector(failingInspector))
+
+	// Default source with failing inspector
+	defaultSource := SourceFunc("default", HasFields("x"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	})
+	r.AddSource(defaultSource)
+
+	// Custom group with working inspector
+	customSource := SourceFunc("custom", HasFields("type"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+	r.AddGroup(workingInspector, customSource)
+
+	Register(r, "test", &testHandler{})
+
+	msg := []byte(`{"type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg)
+
+	s.Require().NoError(err)
+	s.Assert().Equal(1, workingInspector.count)
+}
+
+func (s *ViewCachingSuite) TestViewCacheCachesFailureResult() {
+	// Inspector that fails but counts calls
+	failingInspector := &failingCountingInspector{}
+	workingInspector := &countingInspector{}
+
+	r := New(WithInspector(failingInspector))
+
+	// Add multiple sources to default group to force multiple discriminator checks
+	r.AddSource(SourceFunc("src1", HasFields("a"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	}))
+	r.AddSource(SourceFunc("src2", HasFields("b"), func(raw []byte) (Parsed, bool) {
+		return Parsed{}, false
+	}))
+
+	// Custom group with working inspector
+	customSource := SourceFunc("custom", HasFields("type"), func(raw []byte) (Parsed, bool) {
+		var env struct {
+			Type    string          `json:"type"`
+			Payload json.RawMessage `json:"payload"`
+		}
+		if json.Unmarshal(raw, &env) != nil {
+			return Parsed{}, false
+		}
+		return Parsed{Key: env.Type, Payload: env.Payload}, true
+	})
+	r.AddGroup(workingInspector, customSource)
+
+	Register(r, "test", &testHandler{})
+
+	msg := []byte(`{"type": "test", "payload": {}}`)
+	err := r.Process(context.Background(), msg)
+
+	s.Require().NoError(err)
+	// Failing inspector should only be called once, even with multiple sources
+	s.Assert().Equal(1, failingInspector.count, "failing inspector should only be called once (result cached)")
+	s.Assert().Equal(1, workingInspector.count)
+}
+
+// failingCountingInspector always fails but counts calls.
+type failingCountingInspector struct {
+	count int
+}
+
+func (f *failingCountingInspector) Inspect(raw []byte) (View, error) {
+	f.count++
+	return nil, ErrInvalidJSON
+}