From fbe7b961384a2b42895f75a4275d592261fbe19c Mon Sep 17 00:00:00 2001
From: Silvan <27845747+adlerhurst@users.noreply.github.com>
Date: Wed, 9 Sep 2026 11:08:50 +0200
Subject: [PATCH] fix(eventstore): prevent projections from skipping events
(#12703)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
# Which Problems Are Solved
Handler/v2 projections (`users14` and every other event projection) can
overlook committed events under load:
- `eventstore.commands_to_events` stamped `position` with `NOW()`
(transaction start). A slow create TX could commit after a faster TX
that started later, so the projector stored a later position and never
saw the earlier one.
- Resume used `OFFSET` on a **filtered** row count. That could skip
`in_tx_order = 1` at the next position (for example `user.human.added`).
- Two transactions in the same microsecond both start at `in_tx_order =
1`. A cursor of `in_tx_order > last` alone would drop the other TX’s
first event.
This does not repair rows that are already missing from projections.
# How the Problems Are Solved
- Setup step 77 replaces `eventstore.commands_to_events` so `created_at
= statement_timestamp()` and `position = EXTRACT(EPOCH FROM
clock_timestamp())` at INSERT time. Initialise `08_events_table.sql`
matches that for new databases. The function is `VOLATILE` to match
`clock_timestamp()`.
- The projector still drains in-flight writers with `pg_advisory_lock` /
`pg_advisory_unlock` in one statement, then caps with `position <=
EXTRACT(EPOCH FROM now())` (projector TX start). The exclusive lock is
**not** held across the event `SELECT`. The cap must stay `now()`, not
`clock_timestamp()`, or later inserts become a high-water mark.
- Resume is a lexicographic `EventSortKey` on `(position, in_tx_order,
instance_id, aggregate_type, aggregate_id, sequence)` via
`AfterEventSortKey`. SQL `OFFSET` is no longer used for projection
catch-up.
- Handler and FieldHandler persist the last event’s `in_tx_order` in
`current_states.filter_offset` (existing column, new meaning). Step 77
backfills matching rows from `events2`; unmatched rows keep their
previous value.
# Additional Changes
- `eventstore.Event` exposes `InTxOrder() uint32` (BaseEvent, repository
Event, v3 event, v1 models).
- v3 push SELECT/RETURNING now includes `in_tx_order` so newly written
events carry a real ordinal.
- `WithMinPosition` stays an inclusive `PositionAtLeast` floor and does
not smash the sort-key cursor.
- `SearchQuery.PositionAfter(decimal)` is unchanged (used by
access-token lookup). Do not confuse it with
`SearchQueryBuilder.AfterEventSortKey`.
- `commands_to_events` in setup 77 and initialise 08 is `VOLATILE
PARALLEL SAFE`. Shipped 40/64/70 stay `STABLE` because they still stamp
with `NOW()`.
- Drop unused `last_owner TEXT` DECLARE from setup 77 and initialise
`08_events_table.sql` (`i INTEGER` remains for the enforce-owner loop).
Shipped 40/64/70 keep the leftover.
- This PR’s migration is numbered 77 because `main` already has setup 76
(`76_users14_login_equality_indexes`). First-loop order is s70, then
main’s 76, then this 77.
# Additional Context
**Reviewer map**
| Start here | Why |
| --- | --- |
| `cmd/setup/77.sql` + `cmd/setup/77.go` | Position stamp + atomic
function replace/backfill after shipped `s70` and main’s s76 |
| `internal/eventstore/event_sort_key.go` + `search_query.go` |
Canonical cursor and `AfterEventSortKey` |
| `internal/eventstore/handler/v2/{state.go,handler.go}` |
Persist/resume cursor; `eventQuery` |
| `internal/eventstore/repository/sql/{query.go,postgres.go}` | Tuple
`>` filter, 6-column `ORDER BY`, await cap |
| `internal/eventstore/v3/push.go` / `push.sql` | Scan `in_tx_order` on
write |
Mechanical / low-risk: handler and query test fixtures, Event interface
stubs (`event_base.go`, repository Event, mocks, v1 models).
**Do not look for** a `70.sql` change — shipped migrations stay as on
`main`. Setup 40/64 still use `NOW()`; step 77 overwrites the function
afterward. Fresh `FirstInstance` (runs before `s70`) can still stamp
`NOW()` until 77; runtime pushes after 77 use `clock_timestamp()`.
Main’s `cmd/setup/76.go` is the users14 login-equality indexes, not this
stamp.
**Rollout**
1. Run setup so 77 replaces `commands_to_events` and backfills
`filter_offset`.
2. Restart ZITADEL so projectors load the new resume path.
3. Existing missing `users14` rows are not repaired by this PR.
**Tests**
- `go test ./internal/eventstore/... ./cmd/setup/`
- Sort-key SQL fixtures in `query_sort_key_test.go` (no `OFFSET`;
aggregate-id queries keep 6-column `ORDER BY`)
- `cmd/setup/77_test.go` asserts the embed has no unused `last_owner`
and still uses `clock_timestamp()` / `VOLATILE PARALLEL SAFE`
Related: #8509, #10480 (remove OFFSET), #10560 (advisory locks), #10710
(stable ORDER BY).
Out of scope: repairing already-missing projection rows, holding the
exclusive lock during SELECT, rewriting historical setup 40/64 SQL.
---------
Co-authored-by: Cursor Agent
Co-authored-by: Silvan
---
benchmark/Makefile | 4 +
benchmark/src/use_cases/create_humans.ts | 24 +++
cmd/initialise/sql/08_events_table.sql | 7 +-
cmd/setup/77.go | 53 +++++++
cmd/setup/77.sql | 99 ++++++++++++
cmd/setup/77_current_states.sql | 8 +
cmd/setup/77_test.go | 14 ++
cmd/setup/config.go | 1 +
cmd/setup/setup.go | 2 +
internal/eventstore/event.go | 2 +
internal/eventstore/event_base.go | 7 +
internal/eventstore/event_sort_key.go | 28 ++++
.../eventstore/handler/v2/field_handler.go | 57 +------
internal/eventstore/handler/v2/handler.go | 61 ++------
.../eventstore/handler/v2/handler_test.go | 135 ++++++++++++++++
internal/eventstore/handler/v2/state.go | 44 +++---
internal/eventstore/handler/v2/state_get.sql | 1 +
internal/eventstore/handler/v2/state_test.go | 27 ++--
internal/eventstore/handler/v2/statement.go | 26 ++--
internal/eventstore/repository/event.go | 7 +
.../repository/mock/repository.mock.impl.go | 4 +
.../eventstore/repository/search_query.go | 2 +
.../eventstore/repository/sql/postgres.go | 11 +-
internal/eventstore/repository/sql/query.go | 18 ++-
.../repository/sql/query_sort_key_test.go | 144 ++++++++++++++++++
.../eventstore/repository/sql/query_test.go | 14 +-
internal/eventstore/search_query.go | 11 ++
internal/eventstore/v1/models/event.go | 5 +
internal/eventstore/v3/event.go | 6 +
internal/eventstore/v3/push.go | 4 +-
internal/eventstore/v3/push.sql | 2 +-
internal/eventstore/v3/push_without_func.go | 2 +-
32 files changed, 670 insertions(+), 160 deletions(-)
create mode 100644 benchmark/src/use_cases/create_humans.ts
create mode 100644 cmd/setup/77.go
create mode 100644 cmd/setup/77.sql
create mode 100644 cmd/setup/77_current_states.sql
create mode 100644 cmd/setup/77_test.go
create mode 100644 internal/eventstore/event_sort_key.go
create mode 100644 internal/eventstore/handler/v2/handler_test.go
create mode 100644 internal/eventstore/repository/sql/query_sort_key_test.go
diff --git a/benchmark/Makefile b/benchmark/Makefile
index 7e7974649..2589ea5b8 100644
--- a/benchmark/Makefile
+++ b/benchmark/Makefile
@@ -78,6 +78,10 @@ users_by_metadata_value: ensure_modules bundle
users_by_login_name: ensure_modules bundle
${K6} run --summary-trend-stats "min,avg,max,p(50),p(95),p(99)" dist/users_by_login_name.js --vus ${VUS} --duration ${DURATION} --out csv=output/users_by_login_name_${DATE}.csv
+.PHONY: create_humans
+create_humans: ensure_modules ensure_key_pair bundle
+ ${K6} run --summary-trend-stats "min,avg,max,p(50),p(95),p(99)" dist/create_humans.js --vus ${VUS} --duration ${DURATION}
+
.PHONY: lint
lint:
npm i
diff --git a/benchmark/src/use_cases/create_humans.ts b/benchmark/src/use_cases/create_humans.ts
new file mode 100644
index 000000000..33fbb46e4
--- /dev/null
+++ b/benchmark/src/use_cases/create_humans.ts
@@ -0,0 +1,24 @@
+import { Config } from '../config';
+import { loginByUsernamePassword } from '../login_ui';
+import { createOrg, Org } from '../org';
+import { createHuman, User } from '../user';
+
+type SetupData = {
+ tokens: { accessToken?: string };
+ org: Org;
+};
+
+export async function setup(): Promise {
+ const tokens = loginByUsernamePassword(Config.admin as User);
+ console.info('setup: admin signed in');
+
+ const org = await createOrg(tokens.accessToken!);
+ console.info(`setup: org (${org.organizationId}) created`);
+
+ return { tokens, org };
+}
+
+export default async function (data: SetupData) {
+ await createHuman(`zitizen3-${ __VU }-${ __ITER }`, data.org, data.tokens.accessToken!);
+}
+
diff --git a/cmd/initialise/sql/08_events_table.sql b/cmd/initialise/sql/08_events_table.sql
index c22394838..2a41cea1e 100644
--- a/cmd/initialise/sql/08_events_table.sql
+++ b/cmd/initialise/sql/08_events_table.sql
@@ -53,13 +53,12 @@ END $$;
CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[])
RETURNS SETOF eventstore.events2
LANGUAGE 'plpgsql'
- STABLE PARALLEL SAFE
+ VOLATILE PARALLEL SAFE
ROWS 10
AS $$
DECLARE
_latest_events eventstore.latest_command[];
command eventstore.command2;
- last_owner TEXT;
i INTEGER;
BEGIN
@@ -132,14 +131,14 @@ BEGIN
, c.command_type AS event_type
, COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.in_tx_order) AS sequence
, c.revision
- , NOW() AS created_at
+ , statement_timestamp() AS created_at
, c.payload
, c.creator
, CASE WHEN c.enforce_owner
THEN c.owner
ELSE COALESCE(e.owner, c.owner)
END AS owner
- , EXTRACT(EPOCH FROM NOW()) AS position
+ , EXTRACT(EPOCH FROM clock_timestamp()) AS position
, c.in_tx_order
FROM (
SELECT c.*, CAST(ROW_NUMBER() OVER () AS INTEGER) AS in_tx_order
diff --git a/cmd/setup/77.go b/cmd/setup/77.go
new file mode 100644
index 000000000..25d9e51d5
--- /dev/null
+++ b/cmd/setup/77.go
@@ -0,0 +1,53 @@
+package setup
+
+import (
+ "context"
+ _ "embed"
+ "fmt"
+
+ "github.com/zitadel/zitadel/backend/v3/instrumentation/logging"
+ "github.com/zitadel/zitadel/internal/database"
+ "github.com/zitadel/zitadel/internal/eventstore"
+)
+
+var (
+ //go:embed 77.sql
+ stampEventPositionAtInsert string
+ //go:embed 77_current_states.sql
+ backfillCurrentStatesInTxOrder string
+)
+
+type StampEventPositionAtInsert struct {
+ dbClient *database.DB
+}
+
+func (mig *StampEventPositionAtInsert) Execute(ctx context.Context, _ eventstore.Event) error {
+ inTxOrderType, err := inTxOrderType(ctx, mig.dbClient)
+ if err != nil {
+ return err
+ }
+
+ tx, err := mig.dbClient.BeginTx(ctx, nil)
+ if err != nil {
+ return err
+ }
+ stmt := fmt.Sprintf(stampEventPositionAtInsert, inTxOrderType)
+ _, err = tx.ExecContext(ctx, stmt)
+ if err == nil {
+ _, err = tx.ExecContext(ctx, backfillCurrentStatesInTxOrder)
+ }
+ if err = database.CloseTransaction(tx, err); err != nil {
+ return err
+ }
+
+ // close idle connections to prevent them from using the old prepared statement
+ for _, conn := range mig.dbClient.Pool.AcquireAllIdle(ctx) {
+ logging.OnError(ctx, conn.Conn().Close(ctx)).Debug("failed to close idle connection")
+ conn.Release()
+ }
+ return nil
+}
+
+func (mig *StampEventPositionAtInsert) String() string {
+ return "77_eventstore_position_clock_timestamp"
+}
diff --git a/cmd/setup/77.sql b/cmd/setup/77.sql
new file mode 100644
index 000000000..733ab348a
--- /dev/null
+++ b/cmd/setup/77.sql
@@ -0,0 +1,99 @@
+CREATE OR REPLACE FUNCTION eventstore.commands_to_events(commands eventstore.command2[])
+ RETURNS SETOF eventstore.events2
+ LANGUAGE 'plpgsql'
+ VOLATILE PARALLEL SAFE
+ ROWS 10
+AS $$
+DECLARE
+ _latest_events eventstore.latest_command[];
+ command eventstore.command2;
+ i INTEGER;
+BEGIN
+
+ SELECT array_agg(res) INTO _latest_events
+ FROM (
+ SELECT
+ ROW(
+ c.instance_id,
+ c.aggregate_type,
+ c.aggregate_id,
+ COALESCE(e.owner, c.owner),
+ COALESCE(e.sequence, 0)
+ )::eventstore.latest_command AS res
+ FROM (
+ SELECT DISTINCT ON (c.instance_id, c.aggregate_type, c.aggregate_id)
+ c.instance_id,
+ c.aggregate_type,
+ c.aggregate_id,
+ c.owner,
+ c.ordinality
+ FROM (
+ SELECT *
+ FROM UNNEST(commands) WITH ORDINALITY AS c
+ ) AS c
+ ORDER BY
+ c.instance_id,
+ c.aggregate_type,
+ c.aggregate_id,
+ c.ordinality
+ ) AS c
+ LEFT JOIN LATERAL (
+ SELECT
+ e.owner,
+ e.sequence
+ FROM eventstore.events2 AS e
+ WHERE e.instance_id = c.instance_id
+ AND e.aggregate_type = c.aggregate_type
+ AND e.aggregate_id = c.aggregate_id
+ ORDER BY e.sequence DESC
+ LIMIT 1
+ ) AS e ON TRUE
+ ) AS latest;
+
+ FOR command IN SELECT * FROM UNNEST(commands) LOOP
+ IF NOT command.enforce_owner THEN
+ SELECT owner INTO command.owner
+ FROM UNNEST(_latest_events) AS e
+ WHERE e.instance_id = command.instance_id
+ AND e.aggregate_type = command.aggregate_type
+ AND e.aggregate_id = command.aggregate_id;
+ ELSE
+ FOR i IN 1..array_length(_latest_events, 1) LOOP
+ IF _latest_events[i].instance_id = command.instance_id
+ AND _latest_events[i].aggregate_type = command.aggregate_type
+ AND _latest_events[i].aggregate_id = command.aggregate_id
+ THEN
+ _latest_events[i].owner := command.owner;
+ EXIT;
+ END IF;
+ END LOOP;
+ END IF;
+
+ END LOOP;
+
+
+ RETURN QUERY SELECT
+ c.instance_id
+ , c.aggregate_type
+ , c.aggregate_id
+ , c.command_type AS event_type
+ , COALESCE(e.sequence, 0) + ROW_NUMBER() OVER (PARTITION BY c.instance_id, c.aggregate_type, c.aggregate_id ORDER BY c.ordinality) AS sequence
+ , c.revision
+ , statement_timestamp() AS created_at
+ , c.payload
+ , c.creator
+ , CASE WHEN c.enforce_owner
+ THEN c.owner
+ ELSE COALESCE(e.owner, c.owner)
+ END AS owner
+ , EXTRACT(EPOCH FROM clock_timestamp()) AS position
+ , c.ordinality::%s AS in_tx_order
+ FROM UNNEST(commands) WITH ORDINALITY AS c
+ LEFT JOIN unnest(_latest_events) AS e
+ ON c.instance_id = e.instance_id
+ AND c.aggregate_type = e.aggregate_type
+ AND c.aggregate_id = e.aggregate_id
+ ORDER BY
+ in_tx_order;
+END;
+$$;
diff --git a/cmd/setup/77_current_states.sql b/cmd/setup/77_current_states.sql
new file mode 100644
index 000000000..bd5b59df7
--- /dev/null
+++ b/cmd/setup/77_current_states.sql
@@ -0,0 +1,8 @@
+-- filter_offset stores events2.in_tx_order for sort-key resume, not a row OFFSET.
+UPDATE projections.current_states cs
+SET filter_offset = e.in_tx_order
+FROM eventstore.events2 e
+WHERE cs.instance_id = e.instance_id
+ AND cs.aggregate_id = e.aggregate_id
+ AND cs.aggregate_type = e.aggregate_type
+ AND cs."sequence" = e.sequence;
diff --git a/cmd/setup/77_test.go b/cmd/setup/77_test.go
new file mode 100644
index 000000000..451576f23
--- /dev/null
+++ b/cmd/setup/77_test.go
@@ -0,0 +1,14 @@
+package setup
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestStampEventPositionAtInsertSQL(t *testing.T) {
+ assert.NotContains(t, stampEventPositionAtInsert, "last_owner")
+ assert.Contains(t, stampEventPositionAtInsert, "i INTEGER")
+ assert.Contains(t, stampEventPositionAtInsert, "clock_timestamp()")
+ assert.Contains(t, stampEventPositionAtInsert, "VOLATILE PARALLEL SAFE")
+}
diff --git a/cmd/setup/config.go b/cmd/setup/config.go
index 0cfc5c15f..07a7d100c 100644
--- a/cmd/setup/config.go
+++ b/cmd/setup/config.go
@@ -196,6 +196,7 @@ type Steps struct {
s74Apps7OIDCConfigsAddRegistrationToken *Apps7OIDCConfigsAddRegistrationToken
s75Apps7OIDCConfigsAddAppLinkConfig *Apps7OIDCConfigsAddAppLinkConfig
s76Users14LoginEqualityIndexes *Users14LoginEqualityIndexes
+ s77StampEventPositionAtInsert *StampEventPositionAtInsert
RelationalTables *TransactionalTables
}
diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go
index 4013b6071..b3a82ec9a 100644
--- a/cmd/setup/setup.go
+++ b/cmd/setup/setup.go
@@ -259,6 +259,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
steps.s74Apps7OIDCConfigsAddRegistrationToken = &Apps7OIDCConfigsAddRegistrationToken{dbClient: dbClient}
steps.s75Apps7OIDCConfigsAddAppLinkConfig = &Apps7OIDCConfigsAddAppLinkConfig{dbClient: dbClient}
steps.s76Users14LoginEqualityIndexes = &Users14LoginEqualityIndexes{dbClient: dbClient}
+ steps.s77StampEventPositionAtInsert = &StampEventPositionAtInsert{dbClient: dbClient}
err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil, nil)
if err != nil {
@@ -318,6 +319,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
steps.s69CacheTablesLogged,
steps.s70AddEventStoreCommandEnforceOwner,
steps.s76Users14LoginEqualityIndexes,
+ steps.s77StampEventPositionAtInsert,
} {
setupErr = executeMigration(ctx, eventstoreClient, step, "migration failed")
if setupErr != nil {
diff --git a/internal/eventstore/event.go b/internal/eventstore/event.go
index 35392c2cb..b6e028537 100644
--- a/internal/eventstore/event.go
+++ b/internal/eventstore/event.go
@@ -61,6 +61,8 @@ type Event interface {
CreatedAt() time.Time
// Position is the global position of the event
Position() decimal.Decimal
+ // InTxOrder is the 1-based ordinal of the event inside its push transaction
+ InTxOrder() uint32
// Unmarshal parses the payload and stores the result
// in the value pointed to by ptr. If ptr is nil or not a pointer,
diff --git a/internal/eventstore/event_base.go b/internal/eventstore/event_base.go
index 3c4b5a709..5dbf9adfc 100644
--- a/internal/eventstore/event_base.go
+++ b/internal/eventstore/event_base.go
@@ -27,6 +27,7 @@ type BaseEvent struct {
Seq uint64
Pos decimal.Decimal
+ InTx uint32
Creation time.Time
previousAggregateSequence uint64
previousAggregateTypeSequence uint64
@@ -68,6 +69,11 @@ func (e *BaseEvent) Sequence() uint64 {
return e.Seq
}
+// InTxOrder implements Event
+func (e *BaseEvent) InTxOrder() uint32 {
+ return e.InTx
+}
+
// CreationDate is the time, the event is inserted into the eventstore
func (e *BaseEvent) CreationDate() time.Time {
return e.Creation
@@ -116,6 +122,7 @@ func BaseEventFromRepo(event Event) *BaseEvent {
User: event.Creator(),
Data: event.DataAsBytes(),
Pos: event.Position(),
+ InTx: event.InTxOrder(),
}
}
diff --git a/internal/eventstore/event_sort_key.go b/internal/eventstore/event_sort_key.go
new file mode 100644
index 000000000..87a8db3c0
--- /dev/null
+++ b/internal/eventstore/event_sort_key.go
@@ -0,0 +1,28 @@
+package eventstore
+
+import "github.com/shopspring/decimal"
+
+// EventSortKey is the events2 cursor: position, in_tx_order, instance_id, aggregate_type, aggregate_id, sequence.
+type EventSortKey struct {
+ Position decimal.Decimal
+ InTxOrder uint32
+ InstanceID string
+ AggregateType AggregateType
+ AggregateID string
+ Sequence uint64
+}
+
+func EventSortKeyFromEvent(e Event) EventSortKey {
+ return EventSortKey{
+ Position: e.Position(),
+ InTxOrder: e.InTxOrder(),
+ InstanceID: e.Aggregate().InstanceID,
+ AggregateType: e.Aggregate().Type,
+ AggregateID: e.Aggregate().ID,
+ Sequence: e.Sequence(),
+ }
+}
+
+func (k EventSortKey) IsZero() bool {
+ return k.Position.IsZero()
+}
diff --git a/internal/eventstore/handler/v2/field_handler.go b/internal/eventstore/handler/v2/field_handler.go
index 957a9c0dd..f8a2f6d2a 100644
--- a/internal/eventstore/handler/v2/field_handler.go
+++ b/internal/eventstore/handler/v2/field_handler.go
@@ -141,17 +141,11 @@ func (h *FieldHandler) processEvents(ctx context.Context, config *triggerConfig)
if err != nil {
return additionalIteration, err
}
- // stop execution if currentState.eventTimestamp >= config.maxCreatedAt
- if !config.maxPosition.IsZero() && currentState.position.GreaterThanOrEqual(config.maxPosition) {
+ if !config.maxPosition.IsZero() && currentState.cursor.Position.GreaterThanOrEqual(config.maxPosition) {
return false, nil
}
- if config.minPosition.GreaterThan(decimal.NewFromInt(0)) {
- currentState.position = config.minPosition
- currentState.offset = 0
- }
-
- events, additionalIteration, err := h.fetchEvents(ctx, tx, currentState)
+ events, additionalIteration, err := h.fetchEvents(ctx, tx, currentState, config.minPosition)
if err != nil {
return additionalIteration, err
}
@@ -170,59 +164,20 @@ func (h *FieldHandler) processEvents(ctx context.Context, config *triggerConfig)
return additionalIteration, err
}
-func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState *state) (_ []eventstore.FillFieldsEvent, additionalIteration bool, err error) {
- events, err := h.es.Filter(ctx, h.eventQuery(currentState).SetTx(tx))
+func (h *FieldHandler) fetchEvents(ctx context.Context, tx *sql.Tx, currentState *state, minPosition decimal.Decimal) (_ []eventstore.FillFieldsEvent, additionalIteration bool, err error) {
+ events, err := h.es.Filter(ctx, h.eventQuery(currentState, minPosition).SetTx(tx))
if err != nil || len(events) == 0 {
logging.OnError(ctx, err).Debug("filter eventstore failed")
return nil, false, err
}
- eventAmount := len(events)
- idx, offset := skipPreviouslyReducedEvents(events, currentState)
-
- if currentState.position.Equal(events[len(events)-1].Position()) {
- offset += currentState.offset
- }
- currentState.position = events[len(events)-1].Position()
- currentState.offset = offset
- currentState.aggregateID = events[len(events)-1].Aggregate().ID
- currentState.aggregateType = events[len(events)-1].Aggregate().Type
- currentState.sequence = events[len(events)-1].Sequence()
- currentState.eventTimestamp = events[len(events)-1].CreatedAt()
-
- if idx+1 == len(events) {
- return nil, false, nil
- }
- events = events[idx+1:]
-
- additionalIteration = eventAmount == int(h.bulkLimit)
+ currentState.applyEvent(events[len(events)-1])
+ additionalIteration = len(events) == int(h.bulkLimit)
fillFieldsEvents := make([]eventstore.FillFieldsEvent, len(events))
- highestPosition := events[len(events)-1].Position()
for i, event := range events {
- if event.Position().Equal(highestPosition) {
- offset++
- }
fillFieldsEvents[i] = event.(eventstore.FillFieldsEvent)
}
return fillFieldsEvents, additionalIteration, nil
}
-
-func skipPreviouslyReducedEvents(events []eventstore.Event, currentState *state) (index int, offset uint32) {
- var position decimal.Decimal
- for i, event := range events {
- if !event.Position().Equal(position) {
- offset = 0
- position = event.Position()
- }
- offset++
- if event.Position().Equal(currentState.position) &&
- event.Aggregate().ID == currentState.aggregateID &&
- event.Aggregate().Type == currentState.aggregateType &&
- event.Sequence() == currentState.sequence {
- return i, offset
- }
- }
- return -1, 0
-}
diff --git a/internal/eventstore/handler/v2/handler.go b/internal/eventstore/handler/v2/handler.go
index 6fcb7f782..2c423cbf9 100644
--- a/internal/eventstore/handler/v2/handler.go
+++ b/internal/eventstore/handler/v2/handler.go
@@ -583,18 +583,12 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
if err != nil {
return additionalIteration, err
}
- // stop execution if currentState.position >= config.maxPosition
- if !config.maxPosition.IsZero() && currentState.position.GreaterThanOrEqual(config.maxPosition) {
+ if !config.maxPosition.IsZero() && currentState.cursor.Position.GreaterThanOrEqual(config.maxPosition) {
return false, nil
}
- if config.minPosition.GreaterThan(decimal.NewFromInt(0)) {
- currentState.position = config.minPosition
- currentState.offset = 0
- }
-
var statements []*Statement
- statements, additionalIteration, err = h.generateStatements(ctx, tx, currentState)
+ statements, additionalIteration, err = h.generateStatements(ctx, tx, currentState, config.minPosition)
if err != nil {
return additionalIteration, err
}
@@ -607,7 +601,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
h.metrics.ProjectionEventsProcessed(ctx, h.ProjectionName(), int64(len(statements)), err == nil)
- if err == nil && currentState.aggregateID != "" && len(statements) > 0 {
+ if err == nil && currentState.cursor.AggregateID != "" && len(statements) > 0 {
// Don't update projection timing or latency unless we successfully processed events
h.metrics.ProjectionUpdateTiming(ctx, h.ProjectionName(), float64(time.Since(start).Seconds()))
h.metrics.ProjectionStateLatency(ctx, h.ProjectionName(), time.Since(currentState.eventTimestamp).Seconds())
@@ -627,12 +621,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
return false, err
}
- currentState.position = statements[lastProcessedIndex].Position
- currentState.offset = statements[lastProcessedIndex].offset
- currentState.aggregateID = statements[lastProcessedIndex].Aggregate.ID
- currentState.aggregateType = statements[lastProcessedIndex].Aggregate.Type
- currentState.sequence = statements[lastProcessedIndex].Sequence
- currentState.eventTimestamp = statements[lastProcessedIndex].CreationDate
+ currentState.applyStatement(statements[lastProcessedIndex])
setStateErr := h.setState(ctx, tx, currentState)
if setStateErr != nil {
@@ -642,7 +631,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
return additionalIteration, err
}
-func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentState *state) (_ []*Statement, additionalIteration bool, err error) {
+func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentState *state, minPosition decimal.Decimal) (_ []*Statement, additionalIteration bool, err error) {
if h.triggerWithoutEvents != nil {
stmt, err := h.triggerWithoutEvents(pseudo.NewScheduledEvent(ctx, time.Now(), currentState.instanceID))
if err != nil {
@@ -651,31 +640,18 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta
return []*Statement{stmt}, false, nil
}
- events, err := h.es.Filter(ctx, h.eventQuery(currentState).SetTx(tx))
+ events, err := h.es.Filter(ctx, h.eventQuery(currentState, minPosition).SetTx(tx))
if err != nil {
logging.WithError(ctx, err).Debug("filter eventstore failed")
return nil, false, err
}
eventAmount := len(events)
- statements, err := h.eventsToStatements(ctx, tx, events, currentState)
+ statements, err := h.eventsToStatements(ctx, tx, events)
if err != nil || len(statements) == 0 {
return nil, false, err
}
- idx := skipPreviouslyReducedStatements(statements, currentState)
- if idx+1 == len(statements) {
- currentState.position = statements[len(statements)-1].Position
- currentState.offset = statements[len(statements)-1].offset
- currentState.aggregateID = statements[len(statements)-1].Aggregate.ID
- currentState.aggregateType = statements[len(statements)-1].Aggregate.Type
- currentState.sequence = statements[len(statements)-1].Sequence
- currentState.eventTimestamp = statements[len(statements)-1].CreationDate
-
- return nil, false, nil
- }
- statements = statements[idx+1:]
-
additionalIteration = eventAmount == int(h.bulkLimit)
if len(statements) < len(events) {
// retry immediately if statements failed
@@ -685,18 +661,6 @@ func (h *Handler) generateStatements(ctx context.Context, tx *sql.Tx, currentSta
return statements, additionalIteration, nil
}
-func skipPreviouslyReducedStatements(statements []*Statement, currentState *state) int {
- for i, statement := range statements {
- if statement.Position.Equal(currentState.position) &&
- statement.Aggregate.ID == currentState.aggregateID &&
- statement.Aggregate.Type == currentState.aggregateType &&
- statement.Sequence == currentState.sequence {
- return i
- }
- }
- return -1
-}
-
func (h *Handler) executeStatements(ctx context.Context, tx *sql.Tx, statements []*Statement) (lastProcessedIndex int, err error) {
lastProcessedIndex = -1
@@ -743,18 +707,17 @@ func (h *Handler) executeStatement(ctx context.Context, tx *sql.Tx, statement *S
return nil
}
-func (h *Handler) eventQuery(currentState *state) *eventstore.SearchQueryBuilder {
+func (h *Handler) eventQuery(currentState *state, minPosition decimal.Decimal) *eventstore.SearchQueryBuilder {
builder := eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
AwaitOpenTransactions().
Limit(uint64(h.bulkLimit)).
OrderAsc().
InstanceID(currentState.instanceID)
- if currentState.position.GreaterThan(decimal.Decimal{}) {
- builder = builder.PositionAtLeast(currentState.position)
- if currentState.offset > 0 {
- builder = builder.Offset(currentState.offset)
- }
+ if minPosition.GreaterThan(decimal.NewFromInt(0)) {
+ builder = builder.PositionAtLeast(minPosition)
+ } else if !currentState.cursor.IsZero() {
+ builder = builder.AfterEventSortKey(currentState.cursor)
}
if h.queryGlobal {
diff --git a/internal/eventstore/handler/v2/handler_test.go b/internal/eventstore/handler/v2/handler_test.go
new file mode 100644
index 000000000..50dbd0066
--- /dev/null
+++ b/internal/eventstore/handler/v2/handler_test.go
@@ -0,0 +1,135 @@
+package handler
+
+import (
+ "context"
+ "testing"
+
+ "github.com/shopspring/decimal"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/zitadel/zitadel/internal/eventstore"
+)
+
+func TestHandler_eventQuery(t *testing.T) {
+ h := &Handler{
+ bulkLimit: 200,
+ eventTypes: map[eventstore.AggregateType][]eventstore.EventType{
+ "user": {"user.human.added", "user.human.password.changed"},
+ },
+ }
+
+ t.Run("no cursor on empty position", func(t *testing.T) {
+ builder := h.eventQuery(&state{instanceID: "inst"}, decimal.Decimal{})
+ assert.Equal(t, uint32(0), builder.GetOffset())
+ assert.Nil(t, builder.GetEventSortKeyAfter())
+ assert.True(t, builder.GetPositionAtLeast().IsZero())
+ })
+
+ t.Run("resumes after event sort key", func(t *testing.T) {
+ cursor := eventstore.EventSortKey{
+ Position: decimal.NewFromFloat(1788336088.993079),
+ InTxOrder: 5,
+ AggregateType: "user",
+ AggregateID: "388963124960608862",
+ Sequence: 5,
+ }
+ builder := h.eventQuery(&state{
+ instanceID: "inst",
+ cursor: cursor,
+ }, decimal.Decimal{})
+ assert.Equal(t, uint32(0), builder.GetOffset())
+ assert.True(t, builder.GetPositionAtLeast().IsZero())
+ require.NotNil(t, builder.GetEventSortKeyAfter())
+ assert.Equal(t, cursor, *builder.GetEventSortKeyAfter())
+ })
+
+ t.Run("min position is an inclusive floor not a smashed cursor", func(t *testing.T) {
+ cursor := eventstore.EventSortKey{
+ Position: decimal.NewFromInt(1),
+ InTxOrder: 5,
+ AggregateType: "user",
+ AggregateID: "agg-a",
+ Sequence: 5,
+ }
+ minPosition := decimal.NewFromInt(10)
+ builder := h.eventQuery(&state{
+ instanceID: "inst",
+ cursor: cursor,
+ }, minPosition)
+ assert.Nil(t, builder.GetEventSortKeyAfter())
+ assert.True(t, builder.GetPositionAtLeast().Equal(minPosition))
+ assert.Equal(t, uint32(0), builder.GetOffset())
+ })
+}
+
+func TestHandler_eventsToStatements_inTxOrder(t *testing.T) {
+ h := &Handler{
+ projection: &projection{name: "users14"},
+ }
+
+ pos1 := decimal.NewFromFloat(1)
+ pos2 := decimal.NewFromFloat(1.5)
+ events := []eventstore.Event{
+ &eventstore.BaseEvent{
+ Agg: &eventstore.Aggregate{ID: "agg-a", Type: "user"},
+ EventType: "user.human.added",
+ Seq: 1,
+ Pos: pos1,
+ InTx: 1,
+ },
+ &eventstore.BaseEvent{
+ Agg: &eventstore.Aggregate{ID: "agg-a", Type: "user"},
+ EventType: "user.human.email.verified",
+ Seq: 2,
+ Pos: pos1,
+ InTx: 2,
+ },
+ &eventstore.BaseEvent{
+ Agg: &eventstore.Aggregate{ID: "agg-b", Type: "user"},
+ EventType: "user.human.password.changed",
+ Seq: 10,
+ Pos: pos2,
+ InTx: 1,
+ },
+ }
+
+ statements, err := h.eventsToStatements(context.Background(), nil, events)
+ require.NoError(t, err)
+ require.Len(t, statements, 3)
+ assert.Equal(t, uint32(1), statements[0].inTxOrder)
+ assert.Equal(t, uint32(2), statements[1].inTxOrder)
+ assert.Equal(t, uint32(1), statements[2].inTxOrder)
+ assert.Equal(t, "agg-b", statements[2].Aggregate.ID)
+}
+
+func TestHandler_eventsToStatements_samePositionInTxOrder(t *testing.T) {
+ h := &Handler{
+ projection: &projection{name: "users14"},
+ }
+ pos := decimal.NewFromInt(1)
+ events := []eventstore.Event{
+ &eventstore.BaseEvent{
+ Agg: &eventstore.Aggregate{ID: "agg-a", Type: "user"},
+ EventType: "user.human.added",
+ Seq: 1,
+ Pos: pos,
+ InTx: 1,
+ },
+ &eventstore.BaseEvent{
+ Agg: &eventstore.Aggregate{ID: "agg-b", Type: "user"},
+ EventType: "user.human.password.changed",
+ Seq: 1,
+ Pos: pos,
+ InTx: 1,
+ },
+ }
+
+ statements, err := h.eventsToStatements(context.Background(), nil, events)
+ require.NoError(t, err)
+ require.Len(t, statements, 2)
+ assert.Equal(t, "agg-a", statements[0].Aggregate.ID)
+ assert.Equal(t, "agg-b", statements[1].Aggregate.ID)
+ assert.Equal(t, uint32(1), statements[0].inTxOrder)
+ assert.Equal(t, uint32(1), statements[1].inTxOrder)
+}
diff --git a/internal/eventstore/handler/v2/state.go b/internal/eventstore/handler/v2/state.go
index 5de452685..056c5924d 100644
--- a/internal/eventstore/handler/v2/state.go
+++ b/internal/eventstore/handler/v2/state.go
@@ -17,12 +17,8 @@ import (
type state struct {
instanceID string
- position decimal.Decimal
eventTimestamp time.Time
- aggregateType eventstore.AggregateType
- aggregateID string
- sequence uint64
- offset uint32
+ cursor eventstore.EventSortKey
}
var (
@@ -43,7 +39,7 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx) (currentState *s
sequence = new(sql.NullInt64)
timestamp = new(sql.NullTime)
position = new(decimal.NullDecimal)
- offset = new(sql.NullInt64)
+ inTxOrder = new(sql.NullInt64)
)
row := tx.QueryRow(currentStateStmt, currentState.instanceID, h.projection.Name())
@@ -53,20 +49,22 @@ func (h *Handler) currentState(ctx context.Context, tx *sql.Tx) (currentState *s
sequence,
timestamp,
position,
- offset,
+ inTxOrder,
)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
logging.WithError(ctx, err).Debug("unable to query current state")
return nil, err
}
- currentState.aggregateID = aggregateID.String
- currentState.aggregateType = eventstore.AggregateType(aggregateType.String)
- currentState.sequence = uint64(sequence.Int64)
currentState.eventTimestamp = timestamp.Time
- currentState.position = position.Decimal
- // psql does not provide unsigned numbers so we work around it
- currentState.offset = uint32(offset.Int64)
+ currentState.cursor = eventstore.EventSortKey{
+ Position: position.Decimal,
+ InTxOrder: uint32(inTxOrder.Int64),
+ InstanceID: currentState.instanceID,
+ AggregateType: eventstore.AggregateType(aggregateType.String),
+ AggregateID: aggregateID.String,
+ Sequence: uint64(sequence.Int64),
+ }
return currentState, nil
}
@@ -74,12 +72,12 @@ func (h *Handler) setState(ctx context.Context, tx *sql.Tx, updatedState *state)
res, err := tx.Exec(updateStateStmt,
h.projection.Name(),
updatedState.instanceID,
- updatedState.aggregateID,
- updatedState.aggregateType,
- updatedState.sequence,
+ updatedState.cursor.AggregateID,
+ updatedState.cursor.AggregateType,
+ updatedState.cursor.Sequence,
updatedState.eventTimestamp,
- updatedState.position,
- updatedState.offset,
+ updatedState.cursor.Position,
+ updatedState.cursor.InTxOrder,
)
if err != nil {
err = zerrors.ThrowInternal(err, "V2-WF23g2", "unable to update state")
@@ -93,3 +91,13 @@ func (h *Handler) setState(ctx context.Context, tx *sql.Tx, updatedState *state)
}
return nil
}
+
+func (s *state) applyStatement(stmt *Statement) {
+ s.cursor = stmt.eventSortKey()
+ s.eventTimestamp = stmt.CreationDate
+}
+
+func (s *state) applyEvent(event eventstore.Event) {
+ s.cursor = eventstore.EventSortKeyFromEvent(event)
+ s.eventTimestamp = event.CreatedAt()
+}
diff --git a/internal/eventstore/handler/v2/state_get.sql b/internal/eventstore/handler/v2/state_get.sql
index a324dbbab..689202fb9 100644
--- a/internal/eventstore/handler/v2/state_get.sql
+++ b/internal/eventstore/handler/v2/state_get.sql
@@ -1,3 +1,4 @@
+-- filter_offset stores events2.in_tx_order (reused column; not a row OFFSET).
SELECT
aggregate_id
, aggregate_type
diff --git a/internal/eventstore/handler/v2/state_test.go b/internal/eventstore/handler/v2/state_test.go
index f793f133f..6537c3946 100644
--- a/internal/eventstore/handler/v2/state_test.go
+++ b/internal/eventstore/handler/v2/state_test.go
@@ -50,7 +50,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
- position: decimal.NewFromInt(42),
+ cursor: eventstore.EventSortKey{Position: decimal.NewFromInt(42)},
},
},
isErr: func(t *testing.T, err error) {
@@ -76,7 +76,7 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
- position: decimal.NewFromInt(42),
+ cursor: eventstore.EventSortKey{Position: decimal.NewFromInt(42)},
},
},
isErr: func(t *testing.T, err error) {
@@ -112,10 +112,12 @@ func TestHandler_updateLastUpdated(t *testing.T) {
updatedState: &state{
instanceID: "instance",
eventTimestamp: time.Now(),
- position: decimal.NewFromInt(42),
- aggregateType: "aggregate type",
- aggregateID: "aggregate id",
- sequence: 42,
+ cursor: eventstore.EventSortKey{
+ Position: decimal.NewFromInt(42),
+ AggregateType: "aggregate type",
+ AggregateID: "aggregate id",
+ Sequence: 42,
+ },
},
},
},
@@ -261,11 +263,14 @@ func TestHandler_currentState(t *testing.T) {
currentState: &state{
instanceID: "instance",
eventTimestamp: testTime,
- position: decimal.NewFromInt(42),
- aggregateType: "aggregate type",
- aggregateID: "aggregate id",
- sequence: 42,
- offset: 10,
+ cursor: eventstore.EventSortKey{
+ Position: decimal.NewFromInt(42),
+ InTxOrder: 10,
+ InstanceID: "instance",
+ AggregateType: "aggregate type",
+ AggregateID: "aggregate id",
+ Sequence: 42,
+ },
},
},
},
diff --git a/internal/eventstore/handler/v2/statement.go b/internal/eventstore/handler/v2/statement.go
index 3aeb043c6..f960aef58 100644
--- a/internal/eventstore/handler/v2/statement.go
+++ b/internal/eventstore/handler/v2/statement.go
@@ -39,11 +39,9 @@ func (s *executionError) Unwrap() error {
return s.parent
}
-func (h *Handler) eventsToStatements(ctx context.Context, tx *sql.Tx, events []eventstore.Event, currentState *state) (statements []*Statement, err error) {
+func (h *Handler) eventsToStatements(ctx context.Context, tx *sql.Tx, events []eventstore.Event) (statements []*Statement, err error) {
statements = make([]*Statement, 0, len(events))
- previousPosition := currentState.position
- offset := currentState.offset
for _, event := range events {
statement, err := h.reduce(event)
if err != nil {
@@ -53,14 +51,6 @@ func (h *Handler) eventsToStatements(ctx context.Context, tx *sql.Tx, events []e
}
return statements, &executionError{err}
}
- offset++
- if !previousPosition.Equal(event.Position()) {
- // offset is 1 because we want to skip this event
- offset = 1
- }
- statement.offset = offset
- statement.Position = event.Position()
- previousPosition = event.Position()
statements = append(statements, statement)
}
return statements, nil
@@ -87,11 +77,22 @@ type Statement struct {
Position decimal.Decimal
CreationDate time.Time
- offset uint32
+ inTxOrder uint32
Execute Exec
}
+func (s *Statement) eventSortKey() eventstore.EventSortKey {
+ return eventstore.EventSortKey{
+ Position: s.Position,
+ InTxOrder: s.inTxOrder,
+ InstanceID: s.Aggregate.InstanceID,
+ AggregateType: s.Aggregate.Type,
+ AggregateID: s.Aggregate.ID,
+ Sequence: s.Sequence,
+ }
+}
+
type Exec func(ctx context.Context, ex Executer, projectionName string) error
func WithTableSuffix(name string) func(*execConfig) {
@@ -112,6 +113,7 @@ func NewStatement(event eventstore.Event, e Exec) *Statement {
Sequence: event.Sequence(),
Position: event.Position(),
CreationDate: event.CreatedAt(),
+ inTxOrder: event.InTxOrder(),
Execute: e,
}
}
diff --git a/internal/eventstore/repository/event.go b/internal/eventstore/repository/event.go
index 12a0201b7..fbf41c751 100644
--- a/internal/eventstore/repository/event.go
+++ b/internal/eventstore/repository/event.go
@@ -24,6 +24,8 @@ type Event struct {
Seq uint64
// Pos is the global sequence of the event multiple events can have the same sequence
Pos decimal.Decimal
+ // InTx is the 1-based ordinal of the event inside its push transaction
+ InTx uint32
// CreationDate is the time the event is created
// it's used for human readability.
@@ -102,6 +104,11 @@ func (e *Event) Position() decimal.Decimal {
return e.Pos
}
+// InTxOrder implements [eventstore.Event]
+func (e *Event) InTxOrder() uint32 {
+ return e.InTx
+}
+
// CreatedAt implements [eventstore.Event]
func (e *Event) CreatedAt() time.Time {
return e.CreationDate
diff --git a/internal/eventstore/repository/mock/repository.mock.impl.go b/internal/eventstore/repository/mock/repository.mock.impl.go
index 313f7ee5e..b58790f41 100644
--- a/internal/eventstore/repository/mock/repository.mock.impl.go
+++ b/internal/eventstore/repository/mock/repository.mock.impl.go
@@ -202,6 +202,10 @@ func (e *mockEvent) Position() decimal.Decimal {
return decimal.Decimal{}
}
+func (e *mockEvent) InTxOrder() uint32 {
+ return 0
+}
+
func (e *mockEvent) CreatedAt() time.Time {
return e.createdAt
}
diff --git a/internal/eventstore/repository/search_query.go b/internal/eventstore/repository/search_query.go
index e1f848b14..5a605d6b7 100644
--- a/internal/eventstore/repository/search_query.go
+++ b/internal/eventstore/repository/search_query.go
@@ -31,6 +31,7 @@ type SearchQuery struct {
CreatedAfter *Filter
CreatedBefore *Filter
ExcludeAggregateIDs []*Filter
+ EventSortKeyAfter *eventstore.EventSortKey
}
// Filter represents all fields needed to compare a field of an event with a value
@@ -132,6 +133,7 @@ func QueryFromBuilder(builder *eventstore.SearchQueryBuilder) (*SearchQuery, err
Tx: builder.GetTx(),
AwaitOpenTransactions: builder.GetAwaitOpenTransactions(),
SubQueries: make([][]*Filter, len(builder.GetQueries())),
+ EventSortKeyAfter: builder.GetEventSortKeyAfter(),
}
for _, f := range []func(builder *eventstore.SearchQueryBuilder, query *SearchQuery) *Filter{
diff --git a/internal/eventstore/repository/sql/postgres.go b/internal/eventstore/repository/sql/postgres.go
index 0bd9910dc..8671ea783 100644
--- a/internal/eventstore/repository/sql/postgres.go
+++ b/internal/eventstore/repository/sql/postgres.go
@@ -15,7 +15,9 @@ import (
"github.com/zitadel/zitadel/internal/telemetry/tracing"
)
-// awaitOpenTransactions ensures event ordering, so we don't events younger that open transactions
+// awaitOpenTransactions drains in-flight writers, then caps at projector TX start (now()),
+// not clock_timestamp(). A wall-clock cap at SELECT time would include rows inserted after
+// BEGIN and recreate position overtake.
var (
awaitOpenTransactionsV1 = ` AND created_at <= now()`
awaitOpenTransactionsV2 = ` AND "position" <= EXTRACT(EPOCH FROM now())`
@@ -38,6 +40,8 @@ func NewPostgres(client *database.DB) *Postgres {
func (db *Postgres) Health(ctx context.Context) error { return db.Ping() }
+const eventSortKeySQL = `"position", in_tx_order, instance_id, aggregate_type, aggregate_id, "sequence"`
+
// FilterToReducer finds all events matching the given search query and passes them to the reduce function.
func (psql *Postgres) FilterToReducer(ctx context.Context, searchQuery *eventstore.SearchQueryBuilder, reduce eventstore.Reducer) (err error) {
ctx, span := tracing.NewSpan(ctx)
@@ -91,9 +95,9 @@ func (db *Postgres) orderByEventSequence(desc, shouldOrderBySequence, useV1 bool
}
if desc {
- return ` ORDER BY "position" DESC, in_tx_order DESC, instance_id, aggregate_type, aggregate_id`
+ return ` ORDER BY (` + eventSortKeySQL + `) DESC`
}
- return ` ORDER BY "position", in_tx_order, instance_id, aggregate_type, aggregate_id`
+ return ` ORDER BY ` + eventSortKeySQL
}
func (db *Postgres) eventQuery(useV1 bool) string {
@@ -123,6 +127,7 @@ func (db *Postgres) eventQuery(useV1 bool) string {
", aggregate_type" +
", aggregate_id" +
", revision" +
+ ", in_tx_order" +
" FROM eventstore.events2"
}
diff --git a/internal/eventstore/repository/sql/query.go b/internal/eventstore/repository/sql/query.go
index 956396913..64c670885 100644
--- a/internal/eventstore/repository/sql/query.go
+++ b/internal/eventstore/repository/sql/query.go
@@ -103,7 +103,7 @@ func query(ctx context.Context, criteria querier, searchQuery *eventstore.Search
// if there is only one subquery we can optimize the query ordering by sequence
var shouldOrderBySequence bool
- if len(q.SubQueries) == 1 {
+ if q.EventSortKeyAfter == nil && len(q.SubQueries) == 1 {
for _, filter := range q.SubQueries[0] {
if filter.Field == repository.FieldAggregateID {
shouldOrderBySequence = filter.Operation == repository.OperationEquals
@@ -226,6 +226,7 @@ func eventsScanner(useV1 bool) func(scanner scan, dest interface{}) (err error)
&event.AggregateType,
&event.AggregateID,
&revision,
+ &event.InTx,
)
event.Version = eventstore.Version("v" + strconv.Itoa(int(revision)))
}
@@ -279,6 +280,21 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
args = append(args, additionalArgs...)
}
+ if query.EventSortKeyAfter != nil && !useV1 {
+ if clauses != "" {
+ clauses += " AND "
+ }
+ clauses += `(` + eventSortKeySQL + `) > (?, ?, ?, ?, ?, ?)`
+ args = append(args,
+ query.EventSortKeyAfter.Position,
+ query.EventSortKeyAfter.InTxOrder,
+ query.EventSortKeyAfter.InstanceID,
+ query.EventSortKeyAfter.AggregateType,
+ query.EventSortKeyAfter.AggregateID,
+ query.EventSortKeyAfter.Sequence,
+ )
+ }
+
excludeAggregateIDs := query.ExcludeAggregateIDs
if len(excludeAggregateIDs) > 0 {
excludeAggregateIDs = append(excludeAggregateIDs, query.InstanceID, query.InstanceIDs, query.Position, query.CreatedAfter, query.CreatedBefore)
diff --git a/internal/eventstore/repository/sql/query_sort_key_test.go b/internal/eventstore/repository/sql/query_sort_key_test.go
new file mode 100644
index 000000000..ed7c53df4
--- /dev/null
+++ b/internal/eventstore/repository/sql/query_sort_key_test.go
@@ -0,0 +1,144 @@
+package sql
+
+import (
+ "context"
+ "database/sql/driver"
+ "regexp"
+ "strings"
+ "testing"
+
+ "github.com/shopspring/decimal"
+
+ "github.com/zitadel/zitadel/internal/database"
+ "github.com/zitadel/zitadel/internal/eventstore"
+ "github.com/zitadel/zitadel/internal/eventstore/repository"
+)
+
+func Test_query_event_sort_key(t *testing.T) {
+ cursor := eventstore.EventSortKey{
+ Position: decimal.NewFromFloat(1.5),
+ InTxOrder: 5,
+ InstanceID: "instanceID",
+ AggregateType: "user",
+ AggregateID: "agg-a",
+ Sequence: 5,
+ }
+ tuple := `(` + eventSortKeySQL + `) > (`
+
+ tests := []struct {
+ name string
+ query *eventstore.SearchQueryBuilder
+ sql string
+ args []driver.Value
+ awaitLock bool
+ }{
+ {
+ name: "after sort key with await cap",
+ query: eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+ InstanceID("instanceID").
+ OrderAsc().
+ AwaitOpenTransactions().
+ Limit(200).
+ AfterEventSortKey(cursor).
+ AddQuery().
+ AggregateTypes("user").
+ EventTypes("user.human.added").
+ Builder(),
+ sql: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND (` + eventSortKeySQL + `) > ($4, $5, $6, $7, $8, $9) AND "position" <= EXTRACT(EPOCH FROM now()) ORDER BY ` + eventSortKeySQL + ` LIMIT $10`,
+ args: []driver.Value{
+ "instanceID",
+ eventstore.AggregateType("user"),
+ eventstore.EventType("user.human.added"),
+ cursor.Position,
+ cursor.InTxOrder,
+ cursor.InstanceID,
+ cursor.AggregateType,
+ cursor.AggregateID,
+ cursor.Sequence,
+ uint64(200),
+ },
+ awaitLock: true,
+ },
+ {
+ name: "after sort key without await",
+ query: eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+ InstanceID("instanceID").
+ OrderAsc().
+ Limit(200).
+ AfterEventSortKey(cursor).
+ AddQuery().
+ AggregateTypes("user").
+ Builder(),
+ sql: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND (` + eventSortKeySQL + `) > ($3, $4, $5, $6, $7, $8) ORDER BY ` + eventSortKeySQL + ` LIMIT $9`,
+ args: []driver.Value{
+ "instanceID",
+ eventstore.AggregateType("user"),
+ cursor.Position,
+ cursor.InTxOrder,
+ cursor.InstanceID,
+ cursor.AggregateType,
+ cursor.AggregateID,
+ cursor.Sequence,
+ uint64(200),
+ },
+ },
+ {
+ name: "after sort key with aggregate id still uses sort-key order",
+ query: eventstore.NewSearchQueryBuilder(eventstore.ColumnsEvent).
+ InstanceID("instanceID").
+ OrderAsc().
+ Limit(200).
+ AfterEventSortKey(cursor).
+ AddQuery().
+ AggregateTypes("user").
+ AggregateIDs("agg-a").
+ Builder(),
+ sql: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND aggregate_id = $3 AND (` + eventSortKeySQL + `) > ($4, $5, $6, $7, $8, $9) ORDER BY ` + eventSortKeySQL + ` LIMIT $10`,
+ args: []driver.Value{
+ "instanceID",
+ eventstore.AggregateType("user"),
+ "agg-a",
+ cursor.Position,
+ cursor.InTxOrder,
+ cursor.InstanceID,
+ cursor.AggregateType,
+ cursor.AggregateID,
+ cursor.Sequence,
+ uint64(200),
+ },
+ },
+ }
+
+ client := NewPostgres(&database.DB{Database: new(testDB)})
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if strings.Contains(tt.sql, "OFFSET") {
+ t.Fatal("sort-key resume must not use OFFSET")
+ }
+ if strings.Contains(tt.sql, `ORDER BY "sequence"`) {
+ t.Fatal("sort-key resume must not order by sequence")
+ }
+ if !strings.Contains(tt.sql, tuple) {
+ t.Fatalf("expected sort-key tuple %s in SQL", tuple)
+ }
+
+ mock := newMockClient(t)
+ if tt.awaitLock {
+ mock = mock.expectExec(regexp.QuoteMeta(
+ `select pg_advisory_lock('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1)), pg_advisory_unlock('eventstore.events2'::REGCLASS::OID::INTEGER, hashtext($1))`),
+ []driver.Value{"instanceID"},
+ )
+ }
+ mock = mock.expectQuery(regexp.QuoteMeta(tt.sql), tt.args)
+ client.DB.DB = mock.client
+
+ err := query(context.Background(), client, tt.query, &[]*repository.Event{}, false)
+ if err != nil {
+ t.Fatalf("query() error = %v", err)
+ }
+ if err := mock.mock.ExpectationsWereMet(); err != nil {
+ t.Errorf("not all expectations met: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/eventstore/repository/sql/query_test.go b/internal/eventstore/repository/sql/query_test.go
index 16e2ad66f..c00eb7f78 100644
--- a/internal/eventstore/repository/sql/query_test.go
+++ b/internal/eventstore/repository/sql/query_test.go
@@ -181,13 +181,13 @@ func Test_prepareColumns(t *testing.T) {
}),
},
res: res{
- query: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2`,
+ query: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2`,
expected: []eventstore.Event{
- &repository.Event{AggregateID: "hodor", AggregateType: "user", Seq: 5, Pos: decimal.NewFromInt(42), Data: nil, Version: "v1"},
+ &repository.Event{AggregateID: "hodor", AggregateType: "user", Seq: 5, Pos: decimal.NewFromInt(42), Data: nil, Version: "v1", InTx: 3},
},
},
fields: fields{
- dbRow: []interface{}{time.Time{}, eventstore.EventType(""), uint64(5), decimal.NewNullDecimal(decimal.NewFromInt(42)), sql.RawBytes(nil), "", sql.NullString{}, "", eventstore.AggregateType("user"), "hodor", uint8(1)},
+ dbRow: []interface{}{time.Time{}, eventstore.EventType(""), uint64(5), decimal.NewNullDecimal(decimal.NewFromInt(42)), sql.RawBytes(nil), "", sql.NullString{}, "", eventstore.AggregateType("user"), "hodor", uint8(1), uint32(3)},
},
},
{
@@ -200,13 +200,13 @@ func Test_prepareColumns(t *testing.T) {
}),
},
res: res{
- query: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2`,
+ query: `SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2`,
expected: []eventstore.Event{
&repository.Event{AggregateID: "hodor", AggregateType: "user", Seq: 5, Pos: decimal.Decimal{}, Data: nil, Version: "v1"},
},
},
fields: fields{
- dbRow: []interface{}{time.Time{}, eventstore.EventType(""), uint64(5), decimal.NullDecimal{}, sql.RawBytes(nil), "", sql.NullString{}, "", eventstore.AggregateType("user"), "hodor", uint8(1)},
+ dbRow: []interface{}{time.Time{}, eventstore.EventType(""), uint64(5), decimal.NullDecimal{}, sql.RawBytes(nil), "", sql.NullString{}, "", eventstore.AggregateType("user"), "hodor", uint8(1), uint32(0)},
},
},
{
@@ -904,7 +904,7 @@ func Test_query_events_mocked(t *testing.T) {
},
fields: fields{
mock: newMockClient(t).expectQuery(
- regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND "position" >= $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" >= $8) ORDER BY "position" DESC, in_tx_order DESC, instance_id, aggregate_type, aggregate_id LIMIT $9`),
+ regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND "position" >= $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND "position" >= $8) ORDER BY (`+eventSortKeySQL+`) DESC LIMIT $9`),
[]driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), decimal.NewFromFloat(123.456), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", decimal.NewFromFloat(123.456), uint64(5)},
),
},
@@ -933,7 +933,7 @@ func Test_query_events_mocked(t *testing.T) {
},
fields: fields{
mock: newMockClient(t).expectQuery(
- regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND created_at > $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND created_at > $8) ORDER BY "position" DESC, in_tx_order DESC, instance_id, aggregate_type, aggregate_id LIMIT $9`),
+ regexp.QuoteMeta(`SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision, in_tx_order FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND event_type = $3 AND created_at > $4 AND aggregate_id NOT IN (SELECT aggregate_id FROM eventstore.events2 WHERE aggregate_type = $5 AND event_type = ANY($6) AND instance_id = $7 AND created_at > $8) ORDER BY (`+eventSortKeySQL+`) DESC LIMIT $9`),
[]driver.Value{"instanceID", eventstore.AggregateType("notify"), eventstore.EventType("notify.foo.bar"), time.Unix(123, 456), eventstore.AggregateType("notify"), []eventstore.EventType{"notification.failed", "notification.success"}, "instanceID", time.Unix(123, 456), uint64(5)},
),
},
diff --git a/internal/eventstore/search_query.go b/internal/eventstore/search_query.go
index d431c4dcb..2fe16dcff 100644
--- a/internal/eventstore/search_query.go
+++ b/internal/eventstore/search_query.go
@@ -26,6 +26,7 @@ type SearchQueryBuilder struct {
excludeAggregateIDs *ExclusionQuery
tx *sql.Tx
positionAtLeast decimal.Decimal
+ eventSortKeyAfter *EventSortKey
awaitOpenTransactions bool
creationDateAfter time.Time
creationDateBefore time.Time
@@ -80,6 +81,10 @@ func (b SearchQueryBuilder) GetPositionAtLeast() decimal.Decimal {
return b.positionAtLeast
}
+func (b SearchQueryBuilder) GetEventSortKeyAfter() *EventSortKey {
+ return b.eventSortKeyAfter
+}
+
func (b SearchQueryBuilder) GetAwaitOpenTransactions() bool {
return b.awaitOpenTransactions
}
@@ -286,6 +291,12 @@ func (builder *SearchQueryBuilder) PositionAtLeast(position decimal.Decimal) *Se
return builder
}
+// AfterEventSortKey resumes after the given event using a lexicographic row comparison.
+func (builder *SearchQueryBuilder) AfterEventSortKey(key EventSortKey) *SearchQueryBuilder {
+ builder.eventSortKeyAfter = &key
+ return builder
+}
+
// AwaitOpenTransactions filters for events which are older than the oldest transaction of the database
func (builder *SearchQueryBuilder) AwaitOpenTransactions() *SearchQueryBuilder {
builder.awaitOpenTransactions = true
diff --git a/internal/eventstore/v1/models/event.go b/internal/eventstore/v1/models/event.go
index ab2b60887..0c4f2142b 100644
--- a/internal/eventstore/v1/models/event.go
+++ b/internal/eventstore/v1/models/event.go
@@ -86,6 +86,11 @@ func (e *Event) Position() decimal.Decimal {
return e.Pos
}
+// InTxOrder implements [eventstore.Event]. The v1 events table has no in_tx_order column.
+func (e *Event) InTxOrder() uint32 {
+ return 0
+}
+
// Type implements [eventstore.action]
func (e *Event) Type() eventstore.EventType {
return e.Typ
diff --git a/internal/eventstore/v3/event.go b/internal/eventstore/v3/event.go
index 38e2247bd..1f04cf44f 100644
--- a/internal/eventstore/v3/event.go
+++ b/internal/eventstore/v3/event.go
@@ -45,6 +45,7 @@ type event struct {
createdAt time.Time
sequence uint64
position decimal.Decimal
+ inTxOrder uint32
}
// TODO: remove on v3
@@ -160,6 +161,11 @@ func (e *event) Position() decimal.Decimal {
return e.position
}
+// InTxOrder implements [eventstore.Event]
+func (e *event) InTxOrder() uint32 {
+ return e.inTxOrder
+}
+
// Unmarshal implements [eventstore.Event]
func (e *event) Unmarshal(ptr any) error {
if len(e.command.Payload) == 0 {
diff --git a/internal/eventstore/v3/push.go b/internal/eventstore/v3/push.go
index ffae4246e..980679719 100644
--- a/internal/eventstore/v3/push.go
+++ b/internal/eventstore/v3/push.go
@@ -85,14 +85,14 @@ func writeEvents(ctx context.Context, tx database.Transaction, commands []events
commandMapping = "eventstore.command"
}
- rows, err := tx.Query(ctx, fmt.Sprintf(`select owner, created_at, "sequence", position from eventstore.push($1::%s[])`, commandMapping), cmds)
+ rows, err := tx.Query(ctx, fmt.Sprintf(`select owner, created_at, "sequence", position, in_tx_order from eventstore.push($1::%s[])`, commandMapping), cmds)
if err != nil {
return nil, err
}
defer rows.Close()
for i := 0; rows.Next(); i++ {
- err = rows.Scan(&events[i].(*event).command.Owner, &events[i].(*event).createdAt, &events[i].(*event).sequence, &events[i].(*event).position)
+ err = rows.Scan(&events[i].(*event).command.Owner, &events[i].(*event).createdAt, &events[i].(*event).sequence, &events[i].(*event).position, &events[i].(*event).inTxOrder)
if err != nil {
logging.WithError(err).Warn("failed to scan events")
return nil, err
diff --git a/internal/eventstore/v3/push.sql b/internal/eventstore/v3/push.sql
index 025094895..671cb78f5 100644
--- a/internal/eventstore/v3/push.sql
+++ b/internal/eventstore/v3/push.sql
@@ -15,4 +15,4 @@ INSERT INTO eventstore.events2 (
, in_tx_order
) VALUES
%s
-RETURNING created_at, "position";
\ No newline at end of file
+RETURNING created_at, "position", in_tx_order;
\ No newline at end of file
diff --git a/internal/eventstore/v3/push_without_func.go b/internal/eventstore/v3/push_without_func.go
index 3386e4f9b..2488ace53 100644
--- a/internal/eventstore/v3/push_without_func.go
+++ b/internal/eventstore/v3/push_without_func.go
@@ -82,7 +82,7 @@ func (es *Eventstore) writeEventsOld(ctx context.Context, tx database.Transactio
defer rows.Close()
for i := 0; rows.Next(); i++ {
- err = rows.Scan(&events[i].(*event).createdAt, &events[i].(*event).position)
+ err = rows.Scan(&events[i].(*event).createdAt, &events[i].(*event).position, &events[i].(*event).inTxOrder)
if err != nil {
logging.WithError(err).Warn("failed to scan events")
return nil, err
