mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-16 12:58:00 +00:00
dab5d9e756
# Which Problems Are Solved If many events are written to the same aggregate id it can happen that zitadel [starts to retry the push transaction](48ffc902cc/internal/eventstore/eventstore.go (L101)
) because [the locking behaviour](48ffc902cc/internal/eventstore/v3/sequence.go (L25)
) during push does compute the wrong sequence because newly committed events are not visible to the transaction. These events impact the current sequence. In cases with high command traffic on a single aggregate id this can have severe impact on general performance of zitadel. Because many connections of the `eventstore pusher` database pool are blocked by each other. # How the Problems Are Solved To improve the performance this locking mechanism was removed and the business logic of push is moved to sql functions which reduce network traffic and can be analyzed by the database before the actual push. For clients of the eventstore framework nothing changed. # Additional Changes - after a connection is established prefetches the newly added database types - `eventstore.BaseEvent` now returns the correct revision of the event # Additional Context - part of https://github.com/zitadel/zitadel/issues/8931 --------- Co-authored-by: Tim Möhlmann <tim+github@zitadel.com> Co-authored-by: Livio Spring <livio.a@gmail.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Elio Bischof <elio@zitadel.com> Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com> Co-authored-by: Miguel Cabrerizo <30386061+doncicuto@users.noreply.github.com> Co-authored-by: Joakim Lodén <Loddan@users.noreply.github.com> Co-authored-by: Yxnt <Yxnt@users.noreply.github.com> Co-authored-by: Stefan Benz <stefan@caos.ch> Co-authored-by: Harsha Reddy <harsha.reddy@klaviyo.com> Co-authored-by: Zach H <zhirschtritt@gmail.com>
225 lines
7.4 KiB
Go
225 lines
7.4 KiB
Go
package mock
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"go.uber.org/mock/gomock"
|
|
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
"github.com/zitadel/zitadel/internal/eventstore"
|
|
"github.com/zitadel/zitadel/internal/eventstore/repository"
|
|
)
|
|
|
|
type MockRepository struct {
|
|
*MockPusher
|
|
*MockQuerier
|
|
}
|
|
|
|
func NewRepo(t *testing.T) *MockRepository {
|
|
controller := gomock.NewController(t)
|
|
return &MockRepository{
|
|
MockPusher: NewMockPusher(controller),
|
|
MockQuerier: NewMockQuerier(controller),
|
|
}
|
|
}
|
|
|
|
func (m *MockRepository) ExpectFilterNoEventsNoError() *MockRepository {
|
|
m.MockQuerier.ctrl.T.Helper()
|
|
|
|
m.MockQuerier.EXPECT().FilterToReducer(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectFilterEvents(events ...eventstore.Event) *MockRepository {
|
|
m.MockQuerier.ctrl.T.Helper()
|
|
|
|
m.MockQuerier.EXPECT().FilterToReducer(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
|
|
func(_ context.Context, _ *eventstore.SearchQueryBuilder, reduce eventstore.Reducer) error {
|
|
for _, event := range events {
|
|
if err := reduce(event); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
},
|
|
)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectFilterEventsError(err error) *MockRepository {
|
|
m.MockQuerier.ctrl.T.Helper()
|
|
|
|
m.MockQuerier.EXPECT().FilterToReducer(gomock.Any(), gomock.Any(), gomock.Any()).Return(err)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectInstanceIDs(hasFilters []*repository.Filter, instanceIDs ...string) *MockRepository {
|
|
m.MockQuerier.ctrl.T.Helper()
|
|
|
|
matcher := gomock.Any()
|
|
if len(hasFilters) > 0 {
|
|
matcher = &filterQueryMatcher{SubQueries: [][]*repository.Filter{hasFilters}}
|
|
}
|
|
m.MockQuerier.EXPECT().InstanceIDs(gomock.Any(), matcher).Return(instanceIDs, nil)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectInstanceIDsError(err error) *MockRepository {
|
|
m.MockQuerier.ctrl.T.Helper()
|
|
|
|
m.MockQuerier.EXPECT().InstanceIDs(gomock.Any(), gomock.Any()).Return(nil, err)
|
|
return m
|
|
}
|
|
|
|
// ExpectPush checks if the expectedCommands are send to the Push method.
|
|
// The call will sleep at least the amount of passed duration.
|
|
func (m *MockRepository) ExpectPush(expectedCommands []eventstore.Command, sleep time.Duration) *MockRepository {
|
|
m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
|
|
func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) {
|
|
m.MockPusher.ctrl.T.Helper()
|
|
|
|
time.Sleep(sleep)
|
|
|
|
if len(expectedCommands) != len(commands) {
|
|
return nil, fmt.Errorf("unexpected amount of commands: want %d, got %d", len(expectedCommands), len(commands))
|
|
}
|
|
for i, expectedCommand := range expectedCommands {
|
|
if !assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Aggregate(), commands[i].Aggregate()) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.Aggregate [%d]: expected: %#v got: %#v", i, expectedCommand.Aggregate(), commands[i].Aggregate())
|
|
}
|
|
if !assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Creator(), commands[i].Creator()) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.Creator [%d]: expected: %#v got: %#v", i, expectedCommand.Creator(), commands[i].Creator())
|
|
}
|
|
if !assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Type(), commands[i].Type()) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.Type [%d]: expected: %#v got: %#v", i, expectedCommand.Type(), commands[i].Type())
|
|
}
|
|
if !assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Revision(), commands[i].Revision()) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.Revision [%d]: expected: %#v got: %#v", i, expectedCommand.Revision(), commands[i].Revision())
|
|
}
|
|
|
|
var expectedPayload []byte
|
|
expectedPayload, ok := expectedCommand.Payload().([]byte)
|
|
if !ok {
|
|
expectedPayload, _ = json.Marshal(expectedCommand.Payload())
|
|
}
|
|
if string(expectedPayload) == "" {
|
|
expectedPayload = []byte("null")
|
|
}
|
|
gotPayload, _ := json.Marshal(commands[i].Payload())
|
|
|
|
if !assert.Equal(m.MockPusher.ctrl.T, expectedPayload, gotPayload) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.Payload [%d]: expected: %#v got: %#v", i, expectedCommand.Payload(), commands[i].Payload())
|
|
}
|
|
if !assert.ElementsMatch(m.MockPusher.ctrl.T, expectedCommand.UniqueConstraints(), commands[i].UniqueConstraints()) {
|
|
m.MockPusher.ctrl.T.Errorf("invalid command.UniqueConstraints [%d]: expected: %#v got: %#v", i, expectedCommand.UniqueConstraints(), commands[i].UniqueConstraints())
|
|
}
|
|
}
|
|
events := make([]eventstore.Event, len(commands))
|
|
for i, command := range commands {
|
|
events[i] = &mockEvent{
|
|
Command: command,
|
|
}
|
|
}
|
|
return events, nil
|
|
},
|
|
)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectPushFailed(err error, expectedCommands []eventstore.Command) *MockRepository {
|
|
m.MockPusher.ctrl.T.Helper()
|
|
|
|
m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
|
|
func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) {
|
|
if len(expectedCommands) != len(commands) {
|
|
return nil, fmt.Errorf("unexpected amount of commands: want %d, got %d", len(expectedCommands), len(commands))
|
|
}
|
|
for i, expectedCommand := range expectedCommands {
|
|
assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Aggregate(), commands[i].Aggregate())
|
|
assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Creator(), commands[i].Creator())
|
|
assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Type(), commands[i].Type())
|
|
assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Revision(), commands[i].Revision())
|
|
assert.Equal(m.MockPusher.ctrl.T, expectedCommand.Payload(), commands[i].Payload())
|
|
assert.ElementsMatch(m.MockPusher.ctrl.T, expectedCommand.UniqueConstraints(), commands[i].UniqueConstraints())
|
|
}
|
|
|
|
return nil, err
|
|
},
|
|
)
|
|
return m
|
|
}
|
|
|
|
type mockEvent struct {
|
|
eventstore.Command
|
|
sequence uint64
|
|
createdAt time.Time
|
|
}
|
|
|
|
// DataAsBytes implements eventstore.Event
|
|
func (e *mockEvent) DataAsBytes() []byte {
|
|
if e.Payload() == nil {
|
|
return nil
|
|
}
|
|
payload, err := json.Marshal(e.Payload())
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return payload
|
|
}
|
|
|
|
func (e *mockEvent) Unmarshal(ptr any) error {
|
|
if e.Payload() == nil {
|
|
return nil
|
|
}
|
|
payload, err := json.Marshal(e.Payload())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return json.Unmarshal(payload, ptr)
|
|
}
|
|
|
|
func (e *mockEvent) Sequence() uint64 {
|
|
return e.sequence
|
|
}
|
|
|
|
func (e *mockEvent) Position() float64 {
|
|
return 0
|
|
}
|
|
|
|
func (e *mockEvent) CreatedAt() time.Time {
|
|
return e.createdAt
|
|
}
|
|
|
|
func (m *MockRepository) ExpectRandomPush(expectedCommands []eventstore.Command) *MockRepository {
|
|
m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
|
|
func(ctx context.Context, _ database.ContextQueryExecuter, commands ...eventstore.Command) ([]eventstore.Event, error) {
|
|
assert.Len(m.MockPusher.ctrl.T, commands, len(expectedCommands))
|
|
|
|
events := make([]eventstore.Event, len(commands))
|
|
for i, command := range commands {
|
|
events[i] = &mockEvent{
|
|
Command: command,
|
|
}
|
|
}
|
|
|
|
return events, nil
|
|
},
|
|
)
|
|
return m
|
|
}
|
|
|
|
func (m *MockRepository) ExpectRandomPushFailed(err error, expectedEvents []eventstore.Command) *MockRepository {
|
|
m.MockPusher.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
|
|
func(ctx context.Context, _ database.ContextQueryExecuter, events ...eventstore.Command) ([]eventstore.Event, error) {
|
|
assert.Len(m.MockPusher.ctrl.T, events, len(expectedEvents))
|
|
return nil, err
|
|
},
|
|
)
|
|
return m
|
|
}
|