fix(eventstore): prevent allocation of filtered events (#6749)

* fix(eventstore): prevent allocation of filtered events

Directly reduce each event obtained from a sql.Rows scan,
so that we do not have to allocate all events in a slice.

* reinstate the mutex as RWMutex

* scan data directly

* add todos

* fix(writemodels): add reduce of parent

* test: remove comment

* update comments

---------

Co-authored-by: adlerhurst <silvan.reusser@gmail.com>

internal/eventstore/eventstore.go

@@ -12,7 +12,9 @@ import (
 // Eventstore abstracts all functions needed to store valid events
 // and filters the stored events
 type Eventstore struct {
-	interceptorMutex  sync.Mutex
+	// TODO: get rid of this mutex,
+	// or if we scale to >4vCPU use a sync.Map
+	interceptorMutex  sync.RWMutex
 	eventInterceptors map[EventType]eventTypeInterceptors
 	eventTypes        []string
 	aggregateTypes    []string
@@ -33,7 +35,6 @@ type eventTypeInterceptors struct {
 func NewEventstore(config *Config) *Eventstore {
 	return &Eventstore{
 		eventInterceptors: map[EventType]eventTypeInterceptors{},
-		interceptorMutex:  sync.Mutex{},
 		PushTimeout:       config.PushTimeout,
 
 		pusher:  config.Pusher,
@@ -83,28 +84,33 @@ func (es *Eventstore) AggregateTypes() []string {
 
 // Filter filters the stored events based on the searchQuery
 // and maps the events to the defined event structs
-func (es *Eventstore) Filter(ctx context.Context, queryFactory *SearchQueryBuilder) ([]Event, error) {
-	// make sure that the instance id is always set
-	if queryFactory.instanceID == nil && authz.GetInstance(ctx).InstanceID() != "" {
-		queryFactory.InstanceID(authz.GetInstance(ctx).InstanceID())
-	}
-
-	events, err := es.querier.Filter(ctx, queryFactory)
+//
+// Deprecated: Use [FilterToQueryReducer] instead to avoid allocations.
+func (es *Eventstore) Filter(ctx context.Context, searchQuery *SearchQueryBuilder) ([]Event, error) {
+	events := make([]Event, 0, searchQuery.GetLimit())
+	searchQuery.ensureInstanceID(ctx)
+	err := es.querier.FilterToReducer(ctx, searchQuery, func(event Event) error {
+		event, err := es.mapEvent(event)
+		if err != nil {
+			return err
+		}
+		events = append(events, event)
+		return nil
+	})
 	if err != nil {
 		return nil, err
 	}
-
-	return es.mapEvents(events)
+	return events, nil
 }
 
 func (es *Eventstore) mapEvents(events []Event) (mappedEvents []Event, err error) {
 	mappedEvents = make([]Event, len(events))
 
-	es.interceptorMutex.Lock()
-	defer es.interceptorMutex.Unlock()
+	es.interceptorMutex.RLock()
+	defer es.interceptorMutex.RUnlock()
 
 	for i, event := range events {
-		mappedEvents[i], err = es.mapEvent(event)
+		mappedEvents[i], err = es.mapEventLocked(event)
 		if err != nil {
 			return nil, err
 		}
@@ -114,6 +120,12 @@ func (es *Eventstore) mapEvents(events []Event) (mappedEvents []Event, err error
 }
 
 func (es *Eventstore) mapEvent(event Event) (Event, error) {
+	es.interceptorMutex.RLock()
+	defer es.interceptorMutex.RUnlock()
+	return es.mapEventLocked(event)
+}
+
+func (es *Eventstore) mapEventLocked(event Event) (Event, error) {
 	interceptors, ok := es.eventInterceptors[event.Type()]
 	if !ok || interceptors.eventMapper == nil {
 		return BaseEventFromRepo(event), nil
@@ -121,6 +133,14 @@ func (es *Eventstore) mapEvent(event Event) (Event, error) {
 	return interceptors.eventMapper(event)
 }
 
+// TODO: refactor so we can change to the following interface:
+/*
+type reducer interface {
+	// Reduce applies an event on the object.
+	Reduce(Event) error
+}
+*/
+
 type reducer interface {
 	//Reduce handles the events of the internal events list
 	// it only appends the newly added events
@@ -131,14 +151,15 @@ type reducer interface {
 
 // FilterToReducer filters the events based on the search query, appends all events to the reducer and calls it's reduce function
 func (es *Eventstore) FilterToReducer(ctx context.Context, searchQuery *SearchQueryBuilder, r reducer) error {
-	events, err := es.Filter(ctx, searchQuery)
-	if err != nil {
-		return err
-	}
-
-	r.AppendEvents(events...)
-
-	return r.Reduce()
+	searchQuery.ensureInstanceID(ctx)
+	return es.querier.FilterToReducer(ctx, searchQuery, func(event Event) error {
+		event, err := es.mapEvent(event)
+		if err != nil {
+			return err
+		}
+		r.AppendEvents(event)
+		return r.Reduce()
+	})
 }
 
 // LatestSequence filters the latest sequence for the given search query
@@ -180,13 +201,7 @@ type QueryReducer interface {
 // FilterToQueryReducer filters the events based on the search query of the query function,
 // appends all events to the reducer and calls it's reduce function
 func (es *Eventstore) FilterToQueryReducer(ctx context.Context, r QueryReducer) error {
-	events, err := es.Filter(ctx, r.Query())
-	if err != nil {
-		return err
-	}
-	r.AppendEvents(events...)
-
-	return r.Reduce()
+	return es.FilterToReducer(ctx, r.Query(), r)
 }
 
 // RegisterFilterEventMapper registers a function for mapping an eventstore event to an event
@@ -207,11 +222,13 @@ func (es *Eventstore) RegisterFilterEventMapper(aggregateType AggregateType, eve
 	return es
 }
 
+type Reducer func(event Event) error
+
 type Querier interface {
 	// Health checks if the connection to the storage is available
 	Health(ctx context.Context) error
-	// Filter returns all events matching the given search query
-	Filter(ctx context.Context, searchQuery *SearchQueryBuilder) (events []Event, err error)
+	// FilterToReducer calls r for every event returned from the storage
+	FilterToReducer(ctx context.Context, searchQuery *SearchQueryBuilder, r Reducer) error
 	// LatestSequence returns the latest sequence found by the search query
 	LatestSequence(ctx context.Context, queryFactory *SearchQueryBuilder) (float64, error)
 	// InstanceIDs returns the instance ids found by the search query