mirror of
https://github.com/zitadel/zitadel.git
synced 2025-12-23 03:06:46 +00:00
a32208118b
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
(cherry picked from commit 136363deda)
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
|
|
}
|