From 076e197dc9563a208183658ac7beb7ae9a969a0b Mon Sep 17 00:00:00 2001 From: Silvan <27845747+adlerhurst@users.noreply.github.com> Date: Mon, 26 May 2025 12:01:45 +0200 Subject: [PATCH] fix(handler): report error correctly (#9919) # Which Problems Are Solved 1. The projection handler reported no error if an error happened but updating the current state was successful. This can lead to skipped projections during setup as soon as the projection has an error but does not correctly report if to the caller. 2. Mirror projections skipped as soon as an error occures, this leads to unprojected projections. 3. Mirror checked position wrongly in some cases # How the Problems Are Solved 1. the error returned by the `Trigger` method will will only be set to the error of updating current states if there occured an error. 2. triggering projections checks for the error type returned and retries if the error had code `23505` 3. Corrected to use the `Equal` method # Additional Changes unify logging on mirror projections --- .../eventsourcing/handler/handler.go | 16 +++++++-- .../eventsourcing/handler/handler.go | 16 +++++++-- internal/database/database.go | 4 +++ .../eventstore/handler/v2/field_handler.go | 4 +-- internal/eventstore/handler/v2/handler.go | 5 ++- internal/eventstore/handler/v2/statement.go | 2 +- internal/query/projection/projection.go | 33 +++++++++++++++---- 7 files changed, 63 insertions(+), 17 deletions(-) diff --git a/internal/admin/repository/eventsourcing/handler/handler.go b/internal/admin/repository/eventsourcing/handler/handler.go index 6993006f98..e7dbb23921 100644 --- a/internal/admin/repository/eventsourcing/handler/handler.go +++ b/internal/admin/repository/eventsourcing/handler/handler.go @@ -2,9 +2,11 @@ package handler import ( "context" + "errors" "fmt" "time" + "github.com/jackc/pgx/v5/pgconn" "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/admin/repository/eventsourcing/view" @@ -58,9 +60,17 @@ func Start(ctx context.Context) { func ProjectInstance(ctx context.Context) error { for i, projection := range projections { logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("starting admin projection") - _, err := projection.Trigger(ctx) - if err != nil { - return err + for { + _, err := projection.Trigger(ctx) + if err == nil { + break + } + var pgErr *pgconn.PgError + errors.As(err, &pgErr) + if pgErr.Code != database.PgUniqueConstraintErrorCode { + return err + } + logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID()).WithError(err).Debug("admin projection failed because of unique constraint, retrying") } logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("admin projection done") } diff --git a/internal/auth/repository/eventsourcing/handler/handler.go b/internal/auth/repository/eventsourcing/handler/handler.go index c71dc04586..002674f2d6 100644 --- a/internal/auth/repository/eventsourcing/handler/handler.go +++ b/internal/auth/repository/eventsourcing/handler/handler.go @@ -2,9 +2,11 @@ package handler import ( "context" + "errors" "fmt" "time" + "github.com/jackc/pgx/v5/pgconn" "github.com/zitadel/logging" "github.com/zitadel/zitadel/internal/api/authz" @@ -72,9 +74,17 @@ func Projections() []*handler2.Handler { func ProjectInstance(ctx context.Context) error { for i, projection := range projections { logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("starting auth projection") - _, err := projection.Trigger(ctx) - if err != nil { - return err + for { + _, err := projection.Trigger(ctx) + if err == nil { + break + } + var pgErr *pgconn.PgError + errors.As(err, &pgErr) + if pgErr.Code != database.PgUniqueConstraintErrorCode { + return err + } + logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID()).WithError(err).Debug("auth projection failed because of unique constraint, retrying") } logging.WithFields("name", projection.ProjectionName(), "instance", authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("auth projection done") } diff --git a/internal/database/database.go b/internal/database/database.go index d6ccf2873c..630404568a 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -18,6 +18,10 @@ import ( "github.com/zitadel/zitadel/internal/zerrors" ) +const ( + PgUniqueConstraintErrorCode = "23505" +) + type Config struct { Dialects map[string]interface{} `mapstructure:",remain"` EventPushConnRatio float64 diff --git a/internal/eventstore/handler/v2/field_handler.go b/internal/eventstore/handler/v2/field_handler.go index c42bed7e3f..1a8bc912d5 100644 --- a/internal/eventstore/handler/v2/field_handler.go +++ b/internal/eventstore/handler/v2/field_handler.go @@ -182,7 +182,7 @@ func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState fillFieldsEvents := make([]eventstore.FillFieldsEvent, len(events)) highestPosition := events[len(events)-1].Position() for i, event := range events { - if event.Position() == highestPosition { + if event.Position().Equal(highestPosition) { offset++ } fillFieldsEvents[i] = event.(eventstore.FillFieldsEvent) @@ -199,7 +199,7 @@ func skipPreviouslyReducedEvents(events []eventstore.Event, currentState *state) position = event.Position() } offset++ - if event.Position() == currentState.position && + if event.Position().Equal(currentState.position) && event.Aggregate().ID == currentState.aggregateID && event.Aggregate().Type == currentState.aggregateType && event.Sequence() == currentState.sequence { diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go index feb12ac3e5..8f75475a0c 100644 --- a/internal/eventstore/handler/v2/handler.go +++ b/internal/eventstore/handler/v2/handler.go @@ -552,7 +552,10 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add currentState.aggregateType = statements[lastProcessedIndex].Aggregate.Type currentState.sequence = statements[lastProcessedIndex].Sequence currentState.eventTimestamp = statements[lastProcessedIndex].CreationDate - err = h.setState(tx, currentState) + setStateErr := h.setState(tx, currentState) + if setStateErr != nil { + err = setStateErr + } return additionalIteration, err } diff --git a/internal/eventstore/handler/v2/statement.go b/internal/eventstore/handler/v2/statement.go index 8fc686b6fc..9ae1bc01ea 100644 --- a/internal/eventstore/handler/v2/statement.go +++ b/internal/eventstore/handler/v2/statement.go @@ -53,7 +53,7 @@ func (h *Handler) eventsToStatements(tx *sql.Tx, events []eventstore.Event, curr return statements, err } offset++ - if previousPosition != event.Position() { + if !previousPosition.Equal(event.Position()) { // offset is 1 because we want to skip this event offset = 1 } diff --git a/internal/query/projection/projection.go b/internal/query/projection/projection.go index 34e437be85..0186275790 100644 --- a/internal/query/projection/projection.go +++ b/internal/query/projection/projection.go @@ -2,8 +2,10 @@ package projection import ( "context" + "errors" "fmt" + "github.com/jackc/pgx/v5/pgconn" "github.com/zitadel/logging" internal_authz "github.com/zitadel/zitadel/internal/api/authz" @@ -203,10 +205,19 @@ func Start(ctx context.Context) { func ProjectInstance(ctx context.Context) error { for i, projection := range projections { logging.WithFields("name", projection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("starting projection") - _, err := projection.Trigger(ctx) - if err != nil { - return err + for { + _, err := projection.Trigger(ctx) + if err == nil { + break + } + var pgErr *pgconn.PgError + errors.As(err, &pgErr) + if pgErr.Code != database.PgUniqueConstraintErrorCode { + return err + } + logging.WithFields("name", projection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID()).WithError(err).Debug("projection failed because of unique constraint, retrying") } + logging.WithFields("name", projection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(projections))).Info("projection done") } return nil } @@ -214,11 +225,19 @@ func ProjectInstance(ctx context.Context) error { func ProjectInstanceFields(ctx context.Context) error { for i, fieldProjection := range fields { logging.WithFields("name", fieldProjection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(fields))).Info("starting fields projection") - err := fieldProjection.Trigger(ctx) - if err != nil { - return err + for { + err := fieldProjection.Trigger(ctx) + if err == nil { + break + } + var pgErr *pgconn.PgError + errors.As(err, &pgErr) + if pgErr.Code != database.PgUniqueConstraintErrorCode { + return err + } + logging.WithFields("name", fieldProjection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID()).WithError(err).Debug("fields projection failed because of unique constraint, retrying") } - logging.WithFields("name", fieldProjection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID()).Info("fields projection done") + logging.WithFields("name", fieldProjection.ProjectionName(), "instance", internal_authz.GetInstance(ctx).InstanceID(), "index", fmt.Sprintf("%d/%d", i, len(fields))).Info("fields projection done") } return nil }