mirror of
https://github.com/zitadel/zitadel.git
synced 2025-12-25 21:36:26 +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
56 lines
1.6 KiB
PL/PgSQL
56 lines
1.6 KiB
PL/PgSQL
CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command[])
|
|
RETURNS SETOF eventstore.events2
|
|
LANGUAGE 'plpgsql'
|
|
STABLE PARALLEL SAFE
|
|
ROWS 10
|
|
AS $$
|
|
DECLARE
|
|
"aggregate" RECORD;
|
|
current_sequence BIGINT;
|
|
current_owner TEXT;
|
|
created_at TIMESTAMPTZ;
|
|
BEGIN
|
|
created_at := statement_timestamp();
|
|
FOR "aggregate" IN
|
|
SELECT DISTINCT
|
|
instance_id
|
|
, aggregate_type
|
|
, aggregate_id
|
|
FROM UNNEST(commands)
|
|
LOOP
|
|
SELECT
|
|
*
|
|
INTO
|
|
current_sequence
|
|
, current_owner
|
|
FROM eventstore.latest_aggregate_state(
|
|
"aggregate".instance_id
|
|
, "aggregate".aggregate_type
|
|
, "aggregate".aggregate_id
|
|
);
|
|
|
|
RETURN QUERY
|
|
SELECT
|
|
c.instance_id
|
|
, c.aggregate_type
|
|
, c.aggregate_id
|
|
, c.command_type -- AS event_type
|
|
, COALESCE(current_sequence, 0) + ROW_NUMBER() OVER () -- AS sequence
|
|
, c.revision
|
|
, created_at
|
|
, c.payload
|
|
, c.creator
|
|
, COALESCE(current_owner, c.owner) -- AS owner
|
|
, EXTRACT(EPOCH FROM created_at) -- AS position
|
|
, c.ordinality::%s -- AS in_tx_order
|
|
FROM
|
|
UNNEST(commands) WITH ORDINALITY AS c
|
|
WHERE
|
|
c.instance_id = aggregate.instance_id
|
|
AND c.aggregate_type = aggregate.aggregate_type
|
|
AND c.aggregate_id = aggregate.aggregate_id;
|
|
END LOOP;
|
|
RETURN;
|
|
END;
|
|
$$;
|