feat(eventstore): add spooler (#78)

* add spooler to eventstore

* Update internal/eventstore/spooler/spooler.go

Co-Authored-By: Silvan <silvan.reusser@gmail.com>

* Update internal/eventstore/spooler/spooler_test.go

Co-Authored-By: Silvan <silvan.reusser@gmail.com>

* remove comments

* fix race condition in test

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

internal/eventstore/query/handler.go

@@ -0,0 +1,11 @@
+package query
+
+import (
+	"github.com/caos/zitadel/internal/eventstore/models"
+)
+
+type Handler interface {
+	ViewModel() string
+	EventQuery() (*models.SearchQuery, error)
+	Process(*models.Event) error
+}

internal/eventstore/spooler/config.go

@@ -0,0 +1,29 @@
+package spooler
+
+import (
+	"github.com/caos/logging"
+	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/sony/sonyflake"
+	"strconv"
+)
+
+type Config struct {
+	Eventstore      eventstore.Eventstore
+	Locker          Locker
+	ViewHandlers    []Handler
+	ConcurrentTasks int
+}
+
+func (c *Config) New() *Spooler {
+	lockID, err := sonyflake.NewSonyflake(sonyflake.Settings{}).NextID()
+	logging.Log("SPOOL-bdO56").OnError(err).Panic("unable to generate lockID")
+
+	return &Spooler{
+		handlers:        c.ViewHandlers,
+		lockID:          strconv.FormatUint(lockID, 10),
+		eventstore:      c.Eventstore,
+		locker:          c.Locker,
+		queue:           make(chan *spooledHandler),
+		concurrentTasks: c.ConcurrentTasks,
+	}
+}

internal/eventstore/spooler/mock.go

@@ -0,0 +1,3 @@
+package spooler
+
+//go:generate mockgen -source spooler.go -destination ./mock/spooler.go -package mock

internal/eventstore/spooler/mock/spooler.go

@@ -0,0 +1,129 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: spooler.go
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	models "github.com/caos/zitadel/internal/eventstore/models"
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+	time "time"
+)
+
+// MockHandler is a mock of Handler interface
+type MockHandler struct {
+	ctrl     *gomock.Controller
+	recorder *MockHandlerMockRecorder
+}
+
+// MockHandlerMockRecorder is the mock recorder for MockHandler
+type MockHandlerMockRecorder struct {
+	mock *MockHandler
+}
+
+// NewMockHandler creates a new mock instance
+func NewMockHandler(ctrl *gomock.Controller) *MockHandler {
+	mock := &MockHandler{ctrl: ctrl}
+	mock.recorder = &MockHandlerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockHandler) EXPECT() *MockHandlerMockRecorder {
+	return m.recorder
+}
+
+// ViewModel mocks base method
+func (m *MockHandler) ViewModel() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ViewModel")
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// ViewModel indicates an expected call of ViewModel
+func (mr *MockHandlerMockRecorder) ViewModel() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ViewModel", reflect.TypeOf((*MockHandler)(nil).ViewModel))
+}
+
+// EventQuery mocks base method
+func (m *MockHandler) EventQuery() (*models.SearchQuery, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "EventQuery")
+	ret0, _ := ret[0].(*models.SearchQuery)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// EventQuery indicates an expected call of EventQuery
+func (mr *MockHandlerMockRecorder) EventQuery() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EventQuery", reflect.TypeOf((*MockHandler)(nil).EventQuery))
+}
+
+// Process mocks base method
+func (m *MockHandler) Process(arg0 *models.Event) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Process", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Process indicates an expected call of Process
+func (mr *MockHandlerMockRecorder) Process(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Process", reflect.TypeOf((*MockHandler)(nil).Process), arg0)
+}
+
+// MinimumCycleDuration mocks base method
+func (m *MockHandler) MinimumCycleDuration() time.Duration {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MinimumCycleDuration")
+	ret0, _ := ret[0].(time.Duration)
+	return ret0
+}
+
+// MinimumCycleDuration indicates an expected call of MinimumCycleDuration
+func (mr *MockHandlerMockRecorder) MinimumCycleDuration() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MinimumCycleDuration", reflect.TypeOf((*MockHandler)(nil).MinimumCycleDuration))
+}
+
+// MockLocker is a mock of Locker interface
+type MockLocker struct {
+	ctrl     *gomock.Controller
+	recorder *MockLockerMockRecorder
+}
+
+// MockLockerMockRecorder is the mock recorder for MockLocker
+type MockLockerMockRecorder struct {
+	mock *MockLocker
+}
+
+// NewMockLocker creates a new mock instance
+func NewMockLocker(ctrl *gomock.Controller) *MockLocker {
+	mock := &MockLocker{ctrl: ctrl}
+	mock.recorder = &MockLockerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockLocker) EXPECT() *MockLockerMockRecorder {
+	return m.recorder
+}
+
+// Renew mocks base method
+func (m *MockLocker) Renew(lockerID, viewModel string, waitTime time.Duration) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Renew", lockerID, viewModel, waitTime)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Renew indicates an expected call of Renew
+func (mr *MockLockerMockRecorder) Renew(lockerID, viewModel, waitTime interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Renew", reflect.TypeOf((*MockLocker)(nil).Renew), lockerID, viewModel, waitTime)
+}

internal/eventstore/spooler/spooler.go

@@ -0,0 +1,141 @@
+package spooler
+
+import (
+	"context"
+	"github.com/caos/logging"
+	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/query"
+	"time"
+)
+
+type Spooler struct {
+	handlers        []Handler
+	locker          Locker
+	lockID          string
+	eventstore      eventstore.Eventstore
+	concurrentTasks int
+	queue           chan *spooledHandler
+}
+
+type Handler interface {
+	query.Handler
+	MinimumCycleDuration() time.Duration
+}
+
+type Locker interface {
+	Renew(lockerID, viewModel string, waitTime time.Duration) error
+}
+
+type spooledHandler struct {
+	Handler
+	locker     Locker
+	lockID     string
+	queuedAt   time.Time
+	eventstore eventstore.Eventstore
+}
+
+func (s *Spooler) Start() {
+	defer logging.LogWithFields("SPOOL-N0V1g", "lockerID", s.lockID, "workers", s.concurrentTasks).Info("spooler started")
+	if s.concurrentTasks < 1 {
+		return
+	}
+	for i := 0; i < s.concurrentTasks; i++ {
+		go func() {
+			for handler := range s.queue {
+				go func(handler *spooledHandler, queue chan<- *spooledHandler) {
+					time.Sleep(handler.MinimumCycleDuration() - time.Since(handler.queuedAt))
+					handler.queuedAt = time.Now()
+					queue <- handler
+				}(handler, s.queue)
+
+				handler.load()
+			}
+		}()
+	}
+	for _, handler := range s.handlers {
+		handler := &spooledHandler{handler, s.locker, s.lockID, time.Now(), s.eventstore}
+		s.queue <- handler
+	}
+}
+
+func (s *spooledHandler) load() {
+	errs := make(chan error)
+	ctx, cancel := context.WithCancel(context.Background())
+	go s.awaitError(cancel, errs)
+	hasLocked := s.lock(ctx, errs)
+
+	defer close(errs)
+
+	if <-hasLocked {
+		events, err := s.query(ctx)
+		if err != nil {
+			errs <- err
+		} else {
+			errs <- s.process(ctx, events)
+		}
+	}
+	<-ctx.Done()
+}
+
+func (s *spooledHandler) awaitError(cancel func(), errs chan error) {
+	select {
+	case err := <-errs:
+		cancel()
+		logging.Log("SPOOL-K2lst").OnError(err).WithField("view", s.ViewModel()).Debug("load canceled")
+	}
+}
+
+func (s *spooledHandler) process(ctx context.Context, events []*models.Event) error {
+	for _, event := range events {
+		select {
+		case <-ctx.Done():
+			logging.Log("SPOOL-FTKwH").WithField("view", s.ViewModel()).Debug("context canceled")
+			return nil
+		default:
+			if err := s.Process(event); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (s *spooledHandler) query(ctx context.Context) ([]*models.Event, error) {
+	query, err := s.EventQuery()
+	if err != nil {
+		return nil, err
+	}
+	return s.eventstore.FilterEvents(ctx, query)
+}
+
+func (s *spooledHandler) lock(ctx context.Context, errs chan<- error) chan bool {
+	renewTimer := time.After(0)
+	renewDuration := s.MinimumCycleDuration() - 50*time.Millisecond
+	locked := make(chan bool, 1)
+
+	go func(locked chan bool) {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-renewTimer:
+				err := s.locker.Renew(s.lockID, s.ViewModel(), s.MinimumCycleDuration()*2)
+				if err == nil {
+					locked <- true
+					renewTimer = time.After(renewDuration)
+					continue
+				}
+
+				if ctx.Err() == nil {
+					errs <- err
+				}
+
+				locked <- false
+				return
+			}
+		}
+	}(locked)
+
+	return locked
+}

internal/eventstore/spooler/spooler_test.go

@@ -0,0 +1,332 @@
+package spooler
+
+import (
+	"context"
+	"fmt"
+	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/spooler/mock"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+)
+
+type testHandler struct {
+	cycleDuration time.Duration
+	processSleep  time.Duration
+	processError  error
+	queryError    error
+	viewModel     string
+}
+
+func (h *testHandler) ViewModel() string {
+	return h.viewModel
+}
+func (h *testHandler) EventQuery() (*models.SearchQuery, error) {
+	return nil, h.queryError
+}
+func (h *testHandler) Process(*models.Event) error {
+	<-time.After(h.processSleep)
+	return h.processError
+}
+func (h *testHandler) MinimumCycleDuration() time.Duration { return h.cycleDuration }
+
+type eventstoreStub struct {
+	events []*models.Event
+	err    error
+}
+
+func (es *eventstoreStub) Health(ctx context.Context) error {
+	return nil
+}
+
+func (es *eventstoreStub) AggregateCreator() *models.AggregateCreator {
+	return nil
+}
+
+func (es *eventstoreStub) FilterEvents(ctx context.Context, in *models.SearchQuery) ([]*models.Event, error) {
+	if es.err != nil {
+		return nil, es.err
+	}
+	return es.events, nil
+}
+func (es *eventstoreStub) PushAggregates(ctx context.Context, in ...*models.Aggregate) error {
+	return nil
+}
+
+func TestSpooler_process(t *testing.T) {
+	type fields struct {
+		currentHandler Handler
+	}
+	type args struct {
+		timeout time.Duration
+		events  []*models.Event
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "process all events",
+			fields: fields{
+				currentHandler: &testHandler{},
+			},
+			args: args{
+				timeout: 0,
+				events:  []*models.Event{&models.Event{}, &models.Event{}},
+			},
+			wantErr: false,
+		},
+		{
+			name: "deadline exeeded",
+			fields: fields{
+				currentHandler: &testHandler{processSleep: 501 * time.Millisecond},
+			},
+			args: args{
+				timeout: 1 * time.Second,
+				events:  []*models.Event{&models.Event{}, &models.Event{}, &models.Event{}, &models.Event{}},
+			},
+			wantErr: false,
+		},
+		{
+			name: "process error",
+			fields: fields{
+				currentHandler: &testHandler{processSleep: 1 * time.Second, processError: fmt.Errorf("i am an error")},
+			},
+			args: args{
+				events: []*models.Event{&models.Event{}, &models.Event{}},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &spooledHandler{
+				Handler: tt.fields.currentHandler,
+			}
+
+			ctx := context.Background()
+			var start time.Time
+			if tt.args.timeout > 0 {
+				ctx, _ = context.WithTimeout(ctx, tt.args.timeout)
+				start = time.Now()
+			}
+
+			if err := s.process(ctx, tt.args.events); (err != nil) != tt.wantErr {
+				t.Errorf("Spooler.process() error = %v, wantErr %v", err, tt.wantErr)
+			}
+
+			elapsed := time.Since(start).Round(1 * time.Second)
+			if tt.args.timeout != 0 && elapsed != tt.args.timeout {
+				t.Errorf("wrong timeout wanted %v elapsed %v since %v", tt.args.timeout, elapsed, time.Since(start))
+			}
+		})
+	}
+}
+
+func TestSpooler_awaitError(t *testing.T) {
+	type fields struct {
+		currentHandler Handler
+		err            error
+		canceled       bool
+	}
+	tests := []struct {
+		name   string
+		fields fields
+	}{
+		{
+			"no error",
+			fields{
+				err:            nil,
+				currentHandler: &testHandler{processSleep: 500 * time.Millisecond},
+				canceled:       false,
+			},
+		},
+		{
+			"with error",
+			fields{
+				err:            fmt.Errorf("hodor"),
+				currentHandler: &testHandler{processSleep: 500 * time.Millisecond},
+				canceled:       false,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &spooledHandler{
+				Handler: tt.fields.currentHandler,
+			}
+			errs := make(chan error)
+			ctx, cancel := context.WithCancel(context.Background())
+
+			go s.awaitError(cancel, errs)
+			errs <- tt.fields.err
+
+			if ctx.Err() == nil {
+				t.Error("cancel function was not called")
+			}
+		})
+	}
+}
+
+// TestSpooler_load checks if load terminates
+func TestSpooler_load(t *testing.T) {
+	type fields struct {
+		currentHandler Handler
+		locker         *testLocker
+		lockID         string
+		eventstore     eventstore.Eventstore
+	}
+	tests := []struct {
+		name   string
+		fields fields
+	}{
+		{
+			"lock exists",
+			fields{
+				currentHandler: &testHandler{processSleep: 500 * time.Millisecond, viewModel: "testView", cycleDuration: 1 * time.Second},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, fmt.Errorf("lock already exists"), 2000*time.Millisecond),
+			},
+		},
+		{
+			"lock fails",
+			fields{
+				currentHandler: &testHandler{processSleep: 100 * time.Millisecond, viewModel: "testView", cycleDuration: 1 * time.Second},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, fmt.Errorf("fail"), 2000*time.Millisecond),
+				eventstore:     &eventstoreStub{events: []*models.Event{&models.Event{}}},
+			},
+		},
+		{
+			"query fails",
+			fields{
+				currentHandler: &testHandler{processSleep: 100 * time.Millisecond, viewModel: "testView", queryError: fmt.Errorf("query fail"), cycleDuration: 1 * time.Second},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, nil, 2000*time.Millisecond),
+				eventstore:     &eventstoreStub{err: fmt.Errorf("fail")},
+			},
+		},
+		{
+			"process event fails",
+			fields{
+				currentHandler: &testHandler{processError: fmt.Errorf("oups"), processSleep: 100 * time.Millisecond, viewModel: "testView", cycleDuration: 500 * time.Millisecond},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, nil, 1000*time.Millisecond),
+				eventstore:     &eventstoreStub{events: []*models.Event{&models.Event{}}},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			defer tt.fields.locker.finish()
+			s := &spooledHandler{
+				Handler:    tt.fields.currentHandler,
+				locker:     tt.fields.locker.mock,
+				lockID:     tt.fields.lockID,
+				eventstore: tt.fields.eventstore,
+			}
+			s.load()
+		})
+	}
+}
+
+func TestSpooler_lock(t *testing.T) {
+	type fields struct {
+		currentHandler Handler
+		locker         *testLocker
+		lockID         string
+		expectsErr     bool
+	}
+	type args struct {
+		deadline time.Time
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+	}{
+		{
+			"renew correct",
+			fields{
+				currentHandler: &testHandler{cycleDuration: 1 * time.Second, viewModel: "testView"},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, nil, 2000*time.Millisecond),
+				expectsErr:     false,
+			},
+			args{
+				deadline: time.Now().Add(1 * time.Second),
+			},
+		},
+		{
+			"renew fails",
+			fields{
+				currentHandler: &testHandler{cycleDuration: 900 * time.Millisecond, viewModel: "testView"},
+				lockID:         "testID",
+				locker:         newTestLocker(t, "testID", "testView").expectRenew(t, fmt.Errorf("renew failed"), 1800*time.Millisecond),
+				expectsErr:     true,
+			},
+			args{
+				deadline: time.Now().Add(5 * time.Second),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			defer tt.fields.locker.finish()
+			s := &spooledHandler{
+				Handler: tt.fields.currentHandler,
+				locker:  tt.fields.locker.mock,
+				lockID:  tt.fields.lockID,
+			}
+
+			errs := make(chan error, 1)
+			ctx, _ := context.WithDeadline(context.Background(), tt.args.deadline)
+
+			locked := s.lock(ctx, errs)
+
+			if tt.fields.expectsErr {
+				err := <-errs
+				if err == nil {
+					t.Error("No error in error queue")
+				}
+			} else {
+				lock := <-locked
+				if !lock {
+					t.Error("lock should be true")
+				}
+			}
+		})
+	}
+}
+
+type testLocker struct {
+	mock     *mock.MockLocker
+	lockerID string
+	viewName string
+	ctrl     *gomock.Controller
+}
+
+func newTestLocker(t *testing.T, lockerID, viewName string) *testLocker {
+	ctrl := gomock.NewController(t)
+	return &testLocker{mock.NewMockLocker(ctrl), lockerID, viewName, ctrl}
+}
+
+func (l *testLocker) expectRenew(t *testing.T, err error, waitTime time.Duration) *testLocker {
+	l.mock.EXPECT().Renew(l.lockerID, l.viewName, gomock.Any()).DoAndReturn(
+		func(_, _ string, gotten time.Duration) error {
+			if waitTime-gotten != 0 {
+				t.Errorf("expected waittime %v got %v", waitTime, gotten)
+			}
+			return err
+		}).Times(1)
+
+	return l
+}
+
+func (l *testLocker) finish() {
+	l.ctrl.Finish()
+}