TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-01-12 14:43:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: ensure setup steps are done in order and one at a time (#4749)

Browse Source 
* fix: ensure setup steps are done in order and one at a time

* fix err handling
This commit is contained in:
Livio Spring 2022-11-23 16:31:59 +01:00 committed by GitHub
parent c64a29a0a0
commit f86d057a92
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
internal/migration/migration.go
View File

@ -2,6 +2,8 @@ package migration
import (
"context"
errs "errors"
"time"
"github.com/zitadel/logging"
@ -18,6 +20,10 @@ const (
aggregateID = "SYSTEM"
)
var (
errMigrationAlreadyStarted = errs.New("already started")
)
type Migration interface {
String() string
Execute(context.Context) error
@ -32,7 +38,7 @@ type RepeatableMigration interface {
func Migrate(ctx context.Context, es *eventstore.Eventstore, migration Migration) (err error) {
logging.Infof("verify migration %s", migration.String())
if should, err := shouldExec(ctx, es, migration); !should || err != nil {
if should, err := checkExec(ctx, es, migration); !should || err != nil {
return err
}
@ -52,6 +58,30 @@ func Migrate(ctx context.Context, es *eventstore.Eventstore, migration Migration
return pushErr
}
// checkExec ensures that only one setup step is done concurrently
// if a setup step is already started, it calls shouldExec after some time again
func checkExec(ctx context.Context, es *eventstore.Eventstore, migration Migration) (bool, error) {
timer := time.NewTimer(0)
for {
select {
case <-ctx.Done():
return false, errors.ThrowInternal(nil, "MIGR-as3f7", "Errors.Internal")
case <-timer.C:
should, err := shouldExec(ctx, es, migration)
if err != nil {
if !errs.Is(err, errMigrationAlreadyStarted) {
return false, err
}
logging.WithFields("migration step", migration.String()).
Warn("migration already started, will check again in 5 seconds")
timer.Reset(5 * time.Second)
break
}
return should, nil
}
}
}
func shouldExec(ctx context.Context, es *eventstore.Eventstore, migration Migration) (should bool, err error) {
events, err := es.Filter(ctx, eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
OrderAsc().
@ -90,7 +120,7 @@ func shouldExec(ctx context.Context, es *eventstore.Eventstore, migration Migrat
}
if isStarted {
return false, nil
return false, errMigrationAlreadyStarted
}
repeatable, ok := migration.(RepeatableMigration)
if !ok {