fix(spooler): correct workers (#508)

* one concurrent task

* disable spooler

* fix: improve concurrency in spooler

* fix: dont block lock

* fix: break if lock failed

* fix: check if handler is working

* fix: worker id

* fix: test

* fix: use limit for spoolers configured in startup.yaml

* fix test

* fix: factory

* fix(key): only reduce if not expired

* fix(searchQueryFactory): check for string-slice in aggregateID

* fix(migrations): combine migrations

* fix: allow saving multiple objects in one request

* fix(eventstore): logging

* fix(eventstore): rethink insert i locks table

* fix: ignore failed tests for the moment

* fix: tuubel

* fix: for tests in io

* fix: ignore tests for io

* fix: rename concurrent tasks to workers

* fix: incomment tests and remove some tests

* fix: refert changes for io

* refactor(eventstore): combine types of sql in one file

* refactor(eventstore): logs, TODO's, tests

* fix(eventstore): sql package

* test(eventstore): add tests for search query factory

* chore: logs

* fix(spooler): optimize lock query
chore(migrations): rename locks.object_type to view_name
chore(migrations): refactor migrations

* test: incomment tests

* fix: rename PrepareSaves to PrepareBulkSave

* chore: go dependencies

* fix(migrations): add id in events table

* refactor(lock): less magic numbers

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

internal/eventstore/spooler/config.go

@@ -1,28 +1,34 @@
 package spooler
 
 import (
+	"os"
+
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/eventstore"
+	"github.com/caos/zitadel/internal/eventstore/query"
 	"github.com/caos/zitadel/internal/id"
 )
 
 type Config struct {
-	Eventstore      eventstore.Eventstore
-	Locker          Locker
-	ViewHandlers    []Handler
-	ConcurrentTasks int
+	Eventstore        eventstore.Eventstore
+	Locker            Locker
+	ViewHandlers      []query.Handler
+	ConcurrentWorkers int
 }
 
 func (c *Config) New() *Spooler {
-	lockID, err := id.SonyFlakeGenerator.Next()
-	logging.Log("SPOOL-bdO56").OnError(err).Panic("unable to generate lockID")
+	lockID, err := os.Hostname()
+	if err != nil || lockID == "" {
+		lockID, err = id.SonyFlakeGenerator.Next()
+		logging.Log("SPOOL-bdO56").OnError(err).Panic("unable to generate lockID")
+	}
 
 	return &Spooler{
-		handlers:        c.ViewHandlers,
-		lockID:          lockID,
-		eventstore:      c.Eventstore,
-		locker:          c.Locker,
-		queue:           make(chan *spooledHandler),
-		concurrentTasks: c.ConcurrentTasks,
+		handlers:   c.ViewHandlers,
+		lockID:     lockID,
+		eventstore: c.Eventstore,
+		locker:     c.Locker,
+		queue:      make(chan *spooledHandler),
+		workers:    c.ConcurrentWorkers,
 	}
 }

internal/eventstore/spooler/spooler.go

@@ -2,6 +2,7 @@ package spooler
 
 import (
 	"context"
+	"strconv"
 
 	"github.com/caos/logging"
 	"github.com/caos/zitadel/internal/eventstore"
@@ -13,17 +14,12 @@ import (
 )
 
 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
+	handlers   []query.Handler
+	locker     Locker
+	lockID     string
+	eventstore eventstore.Eventstore
+	workers    int
+	queue      chan *spooledHandler
 }
 
 type Locker interface {
@@ -31,69 +27,78 @@ type Locker interface {
 }
 
 type spooledHandler struct {
-	Handler
+	query.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 {
+	defer logging.LogWithFields("SPOOL-N0V1g", "lockerID", s.lockID, "workers", s.workers).Info("spooler started")
+	if s.workers < 1 {
 		return
 	}
-	for i := 0; i < s.concurrentTasks; i++ {
-		go func() {
-			for handler := range s.queue {
+
+	for i := 0; i < s.workers; i++ {
+		go func(workerIdx int) {
+			workerID := s.lockID + "--" + strconv.Itoa(workerIdx)
+			for task := 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)
+				}(task, s.queue)
 
-				handler.load()
+				task.load(workerID)
 			}
-		}()
+		}(i)
 	}
 	for _, handler := range s.handlers {
-		handler := &spooledHandler{handler, s.locker, s.lockID, time.Now(), s.eventstore}
+		handler := &spooledHandler{Handler: handler, locker: s.locker, queuedAt: time.Now(), eventstore: s.eventstore}
 		s.queue <- handler
 	}
 }
 
-func (s *spooledHandler) load() {
+func (s *spooledHandler) load(workerID string) {
 	errs := make(chan error)
-	ctx, cancel := context.WithCancel(context.Background())
-	go s.awaitError(cancel, errs)
-	hasLocked := s.lock(ctx, errs)
-
 	defer close(errs)
+	ctx, cancel := context.WithCancel(context.Background())
+	go s.awaitError(cancel, errs, workerID)
+	hasLocked := s.lock(ctx, errs, workerID)
 
 	if <-hasLocked {
+		go func() {
+			for l := range hasLocked {
+				if !l {
+					// we only need to break. An error is already written by the lock-routine to the errs channel
+					break
+				}
+			}
+		}()
 		events, err := s.query(ctx)
 		if err != nil {
 			errs <- err
 		} else {
-			errs <- s.process(ctx, events)
+			errs <- s.process(ctx, events, workerID)
+			logging.Log("SPOOL-0pV8o").WithField("view", s.ViewModel()).WithField("worker", workerID).Debug("process done")
 		}
 	}
 	<-ctx.Done()
 }
 
-func (s *spooledHandler) awaitError(cancel func(), errs chan error) {
+func (s *spooledHandler) awaitError(cancel func(), errs chan error, workerID string) {
 	select {
 	case err := <-errs:
 		cancel()
-		logging.Log("SPOOL-K2lst").OnError(err).WithField("view", s.ViewModel()).Debug("load canceled")
+		logging.Log("SPOOL-K2lst").OnError(err).WithField("view", s.ViewModel()).WithField("worker", workerID).Debug("load canceled")
 	}
 }
 
-func (s *spooledHandler) process(ctx context.Context, events []*models.Event) error {
+func (s *spooledHandler) process(ctx context.Context, events []*models.Event, workerID string) error {
 	for _, event := range events {
 		select {
 		case <-ctx.Done():
-			logging.Log("SPOOL-FTKwH").WithField("view", s.ViewModel()).Debug("context canceled")
+			logging.LogWithFields("SPOOL-FTKwH", "view", s.ViewModel(), "worker", workerID).Debug("context canceled")
 			return nil
 		default:
 			if err := s.Reduce(event); err != nil {
@@ -129,13 +134,27 @@ func (s *spooledHandler) query(ctx context.Context) ([]*models.Event, error) {
 	if err != nil {
 		return nil, err
 	}
+	factory := models.FactoryFromSearchQuery(query)
+	sequence, err := s.eventstore.LatestSequence(ctx, factory)
+	logging.Log("SPOOL-7SciK").OnError(err).Debug("unable to query latest sequence")
+	var processedSequence uint64
+	for _, filter := range query.Filters {
+		if filter.GetField() == models.Field_LatestSequence {
+			processedSequence = filter.GetValue().(uint64)
+		}
+	}
+	if sequence != 0 && processedSequence == sequence {
+		return nil, nil
+	}
+
+	query.Limit = s.QueryLimit()
 	return s.eventstore.FilterEvents(ctx, query)
 }
 
-func (s *spooledHandler) lock(ctx context.Context, errs chan<- error) chan bool {
+func (s *spooledHandler) lock(ctx context.Context, errs chan<- error, workerID string) chan bool {
 	renewTimer := time.After(0)
-	renewDuration := s.MinimumCycleDuration() - 50*time.Millisecond
-	locked := make(chan bool, 1)
+	renewDuration := s.MinimumCycleDuration()
+	locked := make(chan bool)
 
 	go func(locked chan bool) {
 		for {
@@ -143,7 +162,9 @@ func (s *spooledHandler) lock(ctx context.Context, errs chan<- error) chan bool
 			case <-ctx.Done():
 				return
 			case <-renewTimer:
-				err := s.locker.Renew(s.lockID, s.ViewModel(), s.MinimumCycleDuration()*2)
+				logging.Log("SPOOL-K2lst").WithField("view", s.ViewModel()).WithField("worker", workerID).Debug("renew")
+				err := s.locker.Renew(workerID, s.ViewModel(), s.MinimumCycleDuration()*2)
+				logging.Log("SPOOL-K2lst").WithField("view", s.ViewModel()).WithField("worker", workerID).WithError(err).Debug("renew done")
 				if err == nil {
 					locked <- true
 					renewTimer = time.After(renewDuration)

internal/eventstore/spooler/spooler_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/caos/zitadel/internal/errors"
 	"github.com/caos/zitadel/internal/eventstore"
 	"github.com/caos/zitadel/internal/eventstore/models"
+	"github.com/caos/zitadel/internal/eventstore/query"
 	"github.com/caos/zitadel/internal/eventstore/spooler/mock"
 	"github.com/caos/zitadel/internal/view/repository"
 	"github.com/golang/mock/gomock"
@@ -20,13 +21,17 @@ type testHandler struct {
 	processError  error
 	queryError    error
 	viewModel     string
+	bulkLimit     uint64
 }
 
 func (h *testHandler) ViewModel() string {
 	return h.viewModel
 }
 func (h *testHandler) EventQuery() (*models.SearchQuery, error) {
-	return nil, h.queryError
+	if h.queryError != nil {
+		return nil, h.queryError
+	}
+	return &models.SearchQuery{}, nil
 }
 func (h *testHandler) Reduce(*models.Event) error {
 	<-time.After(h.processSleep)
@@ -35,7 +40,12 @@ func (h *testHandler) Reduce(*models.Event) error {
 func (h *testHandler) OnError(event *models.Event, err error) error {
 	return err
 }
-func (h *testHandler) MinimumCycleDuration() time.Duration { return h.cycleDuration }
+func (h *testHandler) MinimumCycleDuration() time.Duration {
+	return h.cycleDuration
+}
+func (h *testHandler) QueryLimit() uint64 {
+	return h.bulkLimit
+}
 
 type eventstoreStub struct {
 	events []*models.Event
@@ -60,9 +70,13 @@ func (es *eventstoreStub) PushAggregates(ctx context.Context, in ...*models.Aggr
 	return nil
 }
 
+func (es *eventstoreStub) LatestSequence(ctx context.Context, in *models.SearchQueryFactory) (uint64, error) {
+	return 0, nil
+}
+
 func TestSpooler_process(t *testing.T) {
 	type fields struct {
-		currentHandler Handler
+		currentHandler query.Handler
 	}
 	type args struct {
 		timeout time.Duration
@@ -81,7 +95,7 @@ func TestSpooler_process(t *testing.T) {
 			},
 			args: args{
 				timeout: 0,
-				events:  []*models.Event{&models.Event{}, &models.Event{}},
+				events:  []*models.Event{{}, {}},
 			},
 			wantErr: false,
 		},
@@ -92,7 +106,7 @@ func TestSpooler_process(t *testing.T) {
 			},
 			args: args{
 				timeout: 1 * time.Second,
-				events:  []*models.Event{&models.Event{}, &models.Event{}, &models.Event{}, &models.Event{}},
+				events:  []*models.Event{{}, {}, {}, {}},
 			},
 			wantErr: false,
 		},
@@ -102,7 +116,7 @@ func TestSpooler_process(t *testing.T) {
 				currentHandler: &testHandler{processSleep: 1 * time.Second, processError: fmt.Errorf("i am an error")},
 			},
 			args: args{
-				events: []*models.Event{&models.Event{}, &models.Event{}},
+				events: []*models.Event{{}, {}},
 			},
 			wantErr: true,
 		},
@@ -120,7 +134,7 @@ func TestSpooler_process(t *testing.T) {
 				start = time.Now()
 			}
 
-			if err := s.process(ctx, tt.args.events); (err != nil) != tt.wantErr {
+			if err := s.process(ctx, tt.args.events, "test"); (err != nil) != tt.wantErr {
 				t.Errorf("Spooler.process() error = %v, wantErr %v", err, tt.wantErr)
 			}
 
@@ -134,7 +148,7 @@ func TestSpooler_process(t *testing.T) {
 
 func TestSpooler_awaitError(t *testing.T) {
 	type fields struct {
-		currentHandler Handler
+		currentHandler query.Handler
 		err            error
 		canceled       bool
 	}
@@ -167,7 +181,7 @@ func TestSpooler_awaitError(t *testing.T) {
 			errs := make(chan error)
 			ctx, cancel := context.WithCancel(context.Background())
 
-			go s.awaitError(cancel, errs)
+			go s.awaitError(cancel, errs, "test")
 			errs <- tt.fields.err
 
 			if ctx.Err() == nil {
@@ -180,9 +194,8 @@ func TestSpooler_awaitError(t *testing.T) {
 // TestSpooler_load checks if load terminates
 func TestSpooler_load(t *testing.T) {
 	type fields struct {
-		currentHandler Handler
+		currentHandler query.Handler
 		locker         *testLocker
-		lockID         string
 		eventstore     eventstore.Eventstore
 	}
 	tests := []struct {
@@ -193,7 +206,6 @@ func TestSpooler_load(t *testing.T) {
 			"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),
 			},
 		},
@@ -201,16 +213,14 @@ func TestSpooler_load(t *testing.T) {
 			"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{}}},
+				eventstore:     &eventstoreStub{events: []*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")},
 			},
@@ -219,9 +229,8 @@ func TestSpooler_load(t *testing.T) {
 			"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{}}},
+				eventstore:     &eventstoreStub{events: []*models.Event{{}}},
 			},
 		},
 	}
@@ -231,19 +240,17 @@ func TestSpooler_load(t *testing.T) {
 			s := &spooledHandler{
 				Handler:    tt.fields.currentHandler,
 				locker:     tt.fields.locker.mock,
-				lockID:     tt.fields.lockID,
 				eventstore: tt.fields.eventstore,
 			}
-			s.load()
+			s.load("test-worker")
 		})
 	}
 }
 
 func TestSpooler_lock(t *testing.T) {
 	type fields struct {
-		currentHandler Handler
+		currentHandler query.Handler
 		locker         *testLocker
-		lockID         string
 		expectsErr     bool
 	}
 	type args struct {
@@ -258,7 +265,6 @@ func TestSpooler_lock(t *testing.T) {
 			"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,
 			},
@@ -270,7 +276,6 @@ func TestSpooler_lock(t *testing.T) {
 			"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,
 			},
@@ -285,13 +290,12 @@ func TestSpooler_lock(t *testing.T) {
 			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)
+			locked := s.lock(ctx, errs, "test-worker")
 
 			if tt.fields.expectsErr {
 				err := <-errs
@@ -321,7 +325,7 @@ func newTestLocker(t *testing.T, lockerID, viewName string) *testLocker {
 }
 
 func (l *testLocker) expectRenew(t *testing.T, err error, waitTime time.Duration) *testLocker {
-	l.mock.EXPECT().Renew(l.lockerID, l.viewName, gomock.Any()).DoAndReturn(
+	l.mock.EXPECT().Renew(gomock.Any(), l.viewName, gomock.Any()).DoAndReturn(
 		func(_, _ string, gotten time.Duration) error {
 			if waitTime-gotten != 0 {
 				t.Errorf("expected waittime %v got %v", waitTime, gotten)