TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-12 06:07:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(eventstore): revert precise decimal (#8527) (#8679)

Browse Source
This commit is contained in:
Tim Möhlmann
2024-09-24 19:43:29 +03:00
committed by GitHub
parent eb97be6fdf
commit aeb379e7de
47 changed files with 215 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
internal/eventstore/handler/v2/handler.go
View File

@@ -4,13 +4,13 @@ import (
"context"
"database/sql"
"errors"
"math"
"math/rand"
"slices"
"sync"
"time"
"github.com/jackc/pgx/v5/pgconn"
"github.com/shopspring/decimal"
"github.com/zitadel/logging"
"github.com/zitadel/zitadel/internal/api/authz"
@@ -379,7 +379,7 @@ func (h *Handler) existingInstances(ctx context.Context) ([]string, error) {
type triggerConfig struct {
awaitRunning bool
maxPosition decimal.Decimal
maxPosition float64
}
type TriggerOpt func(conf *triggerConfig)
@@ -390,7 +390,7 @@ func WithAwaitRunning() TriggerOpt {
}
}
func WithMaxPosition(position decimal.Decimal) TriggerOpt {
func WithMaxPosition(position float64) TriggerOpt {
return func(conf *triggerConfig) {
conf.maxPosition = position
}
@@ -500,7 +500,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
return additionalIteration, err
}
// stop execution if currentState.eventTimestamp >= config.maxCreatedAt
if !config.maxPosition.Equal(decimal.Decimal{}) && currentState.position.GreaterThanOrEqual(config.maxPosition) {
if config.maxPosition != 0 && currentState.position >= config.maxPosition {
return false, nil
}
@@ -576,7 +576,7 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta
func skipPreviouslyReducedStatements(statements []*Statement, currentState *state) int {
for i, statement := range statements {
if statement.Position.Equal(currentState.position) &&
if statement.Position == currentState.position &&
statement.AggregateID == currentState.aggregateID &&
statement.AggregateType == currentState.aggregateType &&
statement.Sequence == currentState.sequence {
@@ -609,14 +609,14 @@ func (h *Handler) executeStatement(ctx context.Context, tx *sql.Tx, currentState
return nil
}
_, err = tx.ExecContext(ctx, "SAVEPOINT exec")
_, err = tx.Exec("SAVEPOINT exec")
if err != nil {
h.log().WithError(err).Debug("create savepoint failed")
return err
}
var shouldContinue bool
defer func() {
_, errSave := tx.ExecContext(ctx, "RELEASE SAVEPOINT exec")
_, errSave := tx.Exec("RELEASE SAVEPOINT exec")
if err == nil {
err = errSave
}
@@ -644,8 +644,9 @@ func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder
OrderAsc().
InstanceID(currentState.instanceID)
if currentState.position.GreaterThan(decimal.Decimal{}) {
builder = builder.PositionGreaterEqual(currentState.position)
if currentState.position > 0 {
// decrease position by 10 because builder.PositionAfter filters for position > and we need position >=
builder = builder.PositionAfter(math.Float64frombits(math.Float64bits(currentState.position) - 10))
if currentState.offset > 0 {
builder = builder.Offset(currentState.offset)
}