From 1890e28f7931b8f6d6efc16c933a25f9c0b86641 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tim=20M=C3=B6hlmann?= Date: Fri, 23 Feb 2024 10:29:10 +0200 Subject: [PATCH] fix(eventstore): retry push on primary key sequence collision (#7420) * fix(eventstore): retry push on primary key sequence collision * MaxRetries config option and unit test --- cmd/defaults.yaml | 2 + internal/eventstore/config.go | 1 + internal/eventstore/eventstore.go | 24 ++- internal/eventstore/eventstore_test.go | 226 ++++++++++++++++++++++++- 4 files changed, 244 insertions(+), 9 deletions(-) diff --git a/cmd/defaults.yaml b/cmd/defaults.yaml index f8f6274fa..fa940bf56 100644 --- a/cmd/defaults.yaml +++ b/cmd/defaults.yaml @@ -546,6 +546,8 @@ Quotas: Eventstore: # Sets the maximum duration of transactions pushing events PushTimeout: 15s #ZITADEL_EVENTSTORE_PUSHTIMEOUT + # Maximum amount of push retries in case of primary key violation on the sequence + MaxRetries: 5 #ZITADEL_EVENTSTORE_MAXRETRIES DefaultInstance: InstanceName: ZITADEL # ZITADEL_DEFAULTINSTANCE_INSTANCENAME diff --git a/internal/eventstore/config.go b/internal/eventstore/config.go index 96e788f50..fd3955667 100644 --- a/internal/eventstore/config.go +++ b/internal/eventstore/config.go @@ -6,6 +6,7 @@ import ( type Config struct { PushTimeout time.Duration + MaxRetries uint32 Pusher Pusher Querier Querier diff --git a/internal/eventstore/eventstore.go b/internal/eventstore/eventstore.go index 7aeaa53ad..17b570ab6 100644 --- a/internal/eventstore/eventstore.go +++ b/internal/eventstore/eventstore.go @@ -2,10 +2,14 @@ package eventstore import ( "context" + "errors" "sort" "sync" "time" + "github.com/jackc/pgconn" + "github.com/zitadel/logging" + "github.com/zitadel/zitadel/internal/api/authz" ) @@ -13,6 +17,7 @@ import ( // and filters the stored events type Eventstore struct { PushTimeout time.Duration + maxRetries int pusher Pusher querier Querier @@ -52,6 +57,7 @@ type eventTypeInterceptors struct { func NewEventstore(config *Config) *Eventstore { return &Eventstore{ PushTimeout: config.PushTimeout, + maxRetries: int(config.MaxRetries), pusher: config.Pusher, querier: config.Querier, @@ -77,7 +83,23 @@ func (es *Eventstore) Push(ctx context.Context, cmds ...Command) ([]Event, error ctx, cancel = context.WithTimeout(ctx, es.PushTimeout) defer cancel() } - events, err := es.pusher.Push(ctx, cmds...) + var ( + events []Event + err error + ) + + // Retry when there is a collision of the sequence as part of the primary key. + // "duplicate key value violates unique constraint \"events2_pkey\" (SQLSTATE 23505)" + // https://github.com/zitadel/zitadel/issues/7202 +retry: + for i := 0; i <= es.maxRetries; i++ { + events, err = es.pusher.Push(ctx, cmds...) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.ConstraintName != "events2_pkey" || pgErr.SQLState() != "23505" { + break retry + } + logging.WithError(err).Info("eventstore push retry") + } if err != nil { return nil, err } diff --git a/internal/eventstore/eventstore_test.go b/internal/eventstore/eventstore_test.go index bbbb383bb..938251409 100644 --- a/internal/eventstore/eventstore_test.go +++ b/internal/eventstore/eventstore_test.go @@ -8,6 +8,8 @@ import ( "testing" "time" + "github.com/jackc/pgconn" + "github.com/zitadel/zitadel/internal/api/authz" "github.com/zitadel/zitadel/internal/api/service" "github.com/zitadel/zitadel/internal/zerrors" @@ -329,7 +331,7 @@ func Test_eventData(t *testing.T) { type testPusher struct { events []Event - err error + errs []error t *testing.T } @@ -339,8 +341,9 @@ func (repo *testPusher) Health(ctx context.Context) error { } func (repo *testPusher) Push(ctx context.Context, commands ...Command) (events []Event, err error) { - if repo.err != nil { - return nil, repo.err + if len(repo.errs) != 0 { + err, repo.errs = repo.errs[0], repo.errs[1:] + return nil, err } if len(repo.events) != len(commands) { @@ -439,6 +442,7 @@ func TestEventstore_Push(t *testing.T) { events []Command } type fields struct { + maxRetries int pusher *testPusher eventMapper map[EventType]func(Event) (Event, error) } @@ -465,6 +469,51 @@ func TestEventstore_Push(t *testing.T) { }, }, fields: fields{ + maxRetries: 1, + pusher: &testPusher{ + t: t, + events: []Event{ + &BaseEvent{ + Agg: &Aggregate{ + ID: "1", + Type: "test.aggregate", + ResourceOwner: "caos", + InstanceID: "zitadel", + }, + Data: []byte(nil), + User: "editorUser", + EventType: "test.event", + }, + }, + }, + eventMapper: map[EventType]func(Event) (Event, error){ + "test.event": func(e Event) (Event, error) { + return &testEvent{ + BaseEvent: BaseEvent{ + Agg: &Aggregate{ + Type: e.Aggregate().Type, + }, + }, + }, nil + }, + }, + }, + }, + { + name: "one aggregate one event, retry disabled", + args: args{ + events: []Command{ + newTestEvent( + "1", + "", + func() interface{} { + return []byte(nil) + }, + false), + }, + }, + fields: fields{ + maxRetries: 0, pusher: &testPusher{ t: t, events: []Event{ @@ -515,6 +564,7 @@ func TestEventstore_Push(t *testing.T) { }, }, fields: fields{ + maxRetries: 1, pusher: &testPusher{ t: t, events: []Event{ @@ -586,6 +636,7 @@ func TestEventstore_Push(t *testing.T) { }, }, fields: fields{ + maxRetries: 1, pusher: &testPusher{ t: t, events: combineEventLists( @@ -658,9 +709,10 @@ func TestEventstore_Push(t *testing.T) { }, }, fields: fields{ + maxRetries: 1, pusher: &testPusher{ - t: t, - err: zerrors.ThrowInternal(nil, "V2-qaa4S", "test err"), + t: t, + errs: []error{zerrors.ThrowInternal(nil, "V2-qaa4S", "test err")}, }, }, res: res{ @@ -681,21 +733,179 @@ func TestEventstore_Push(t *testing.T) { }, }, fields: fields{ + maxRetries: 1, pusher: &testPusher{ - t: t, - err: zerrors.ThrowInternal(nil, "V2-qaa4S", "test err"), + t: t, + errs: []error{zerrors.ThrowInternal(nil, "V2-qaa4S", "test err")}, }, }, res: res{ wantErr: true, }, }, + { + name: "retry succeeds", + args: args{ + events: []Command{ + newTestEvent( + "1", + "", + func() interface{} { + return []byte(nil) + }, + false), + }, + }, + fields: fields{ + maxRetries: 1, + pusher: &testPusher{ + t: t, + events: []Event{ + &BaseEvent{ + Agg: &Aggregate{ + ID: "1", + Type: "test.aggregate", + ResourceOwner: "caos", + InstanceID: "zitadel", + }, + Data: []byte(nil), + User: "editorUser", + EventType: "test.event", + }, + }, + errs: []error{ + zerrors.ThrowInternal(&pgconn.PgError{ + ConstraintName: "events2_pkey", + Code: "23505", + }, "foo-err", "Errors.Internal"), + }, + }, + eventMapper: map[EventType]func(Event) (Event, error){ + "test.event": func(e Event) (Event, error) { + return &testEvent{ + BaseEvent: BaseEvent{ + Agg: &Aggregate{ + Type: e.Aggregate().Type, + }, + }, + }, nil + }, + }, + }, + }, + { + name: "retry fails", + args: args{ + events: []Command{ + newTestEvent( + "1", + "", + func() interface{} { + return []byte(nil) + }, + false), + }, + }, + fields: fields{ + maxRetries: 1, + pusher: &testPusher{ + t: t, + events: []Event{ + &BaseEvent{ + Agg: &Aggregate{ + ID: "1", + Type: "test.aggregate", + ResourceOwner: "caos", + InstanceID: "zitadel", + }, + Data: []byte(nil), + User: "editorUser", + EventType: "test.event", + }, + }, + errs: []error{ + zerrors.ThrowInternal(&pgconn.PgError{ + ConstraintName: "events2_pkey", + Code: "23505", + }, "foo-err", "Errors.Internal"), + zerrors.ThrowInternal(&pgconn.PgError{ + ConstraintName: "events2_pkey", + Code: "23505", + }, "foo-err", "Errors.Internal"), + }, + }, + eventMapper: map[EventType]func(Event) (Event, error){ + "test.event": func(e Event) (Event, error) { + return &testEvent{ + BaseEvent: BaseEvent{ + Agg: &Aggregate{ + Type: e.Aggregate().Type, + }, + }, + }, nil + }, + }, + }, + res: res{wantErr: true}, + }, + { + name: "retry disabled", + args: args{ + events: []Command{ + newTestEvent( + "1", + "", + func() interface{} { + return []byte(nil) + }, + false), + }, + }, + fields: fields{ + maxRetries: 0, + pusher: &testPusher{ + t: t, + events: []Event{ + &BaseEvent{ + Agg: &Aggregate{ + ID: "1", + Type: "test.aggregate", + ResourceOwner: "caos", + InstanceID: "zitadel", + }, + Data: []byte(nil), + User: "editorUser", + EventType: "test.event", + }, + }, + errs: []error{ + zerrors.ThrowInternal(&pgconn.PgError{ + ConstraintName: "events2_pkey", + Code: "23505", + }, "foo-err", "Errors.Internal"), + }, + }, + eventMapper: map[EventType]func(Event) (Event, error){ + "test.event": func(e Event) (Event, error) { + return &testEvent{ + BaseEvent: BaseEvent{ + Agg: &Aggregate{ + Type: e.Aggregate().Type, + }, + }, + }, nil + }, + }, + }, + res: res{wantErr: true}, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { eventInterceptors = map[EventType]eventTypeInterceptors{} es := &Eventstore{ - pusher: tt.fields.pusher, + maxRetries: tt.fields.maxRetries, + pusher: tt.fields.pusher, } for eventType, mapper := range tt.fields.eventMapper { RegisterFilterEventMapper("test", eventType, mapper)