mirror of
https://github.com/zitadel/zitadel.git
synced 2025-12-26 16:56:36 +00:00
136363deda
A timing issue (a race condition) was identified in our event processing system. Under specific circumstances, it was possible for the system to skip processing certain events, leading to potential data inconsistencies. ## Which problems are solved The system tracks its progress through the event log using timestamps. The issue occurred because we were using the timestamp from the start of a database transaction. If a query to read new events began after the transaction started but before the new event was committed, the query would not see the new event and would fail to process it. ## How the problems are solved The fix is to change which timestamp is used for tracking. We now use the precise timestamp of when the event is actually written to the database. This ensures that the event's timestamp is always correctly ordered, closing the timing gap and preventing the race condition. This change enhances the reliability and integrity of our event processing pipeline. It guarantees that all events are processed in the correct order and eliminates the risk of skipped events, ensuring data is always consistent across the system. ## Additional information original fix: https://github.com/zitadel/zitadel/pull/10560
45 lines
1.1 KiB
Go
45 lines
1.1 KiB
Go
package setup
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
_ "embed"
|
|
"fmt"
|
|
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
"github.com/zitadel/zitadel/internal/eventstore"
|
|
)
|
|
|
|
var (
|
|
//go:embed 64.sql
|
|
changePushPosition string
|
|
)
|
|
|
|
type ChangePushPosition struct {
|
|
dbClient *database.DB
|
|
}
|
|
|
|
func (mig *ChangePushPosition) Execute(ctx context.Context, _ eventstore.Event) error {
|
|
inTxOrderType, err := mig.inTxOrderType(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stmt := fmt.Sprintf(changePushPosition, inTxOrderType)
|
|
_, err = mig.dbClient.ExecContext(ctx, stmt)
|
|
return err
|
|
}
|
|
|
|
func (mig *ChangePushPosition) String() string {
|
|
return "64_change_push_position"
|
|
}
|
|
|
|
func (mig *ChangePushPosition) inTxOrderType(ctx context.Context) (typeName string, err error) {
|
|
err = mig.dbClient.QueryRowContext(ctx, func(row *sql.Row) error {
|
|
return row.Scan(&typeName)
|
|
}, `SELECT data_type FROM information_schema.columns WHERE table_schema = 'eventstore' AND table_name = 'events2' AND column_name = 'in_tx_order'`)
|
|
if err != nil {
|
|
return "", fmt.Errorf("get in_tx_order_type: %w", err)
|
|
}
|
|
return typeName, nil
|
|
}
|