From 2a666df3c1099fb9a86b951069233058e1677a56 Mon Sep 17 00:00:00 2001
From: weichen <weichen@example.com>
Date: Sat, 19 Sep 2026 09:24:51 +0000
Subject: [PATCH] fix(events): hold aggregate gaps on production claim

PostgreSQL claim now uses the same next-version rule as MemoryBus, and checkpoints only move forward. A later version cannot be leased while an earlier one is still missing.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 internal/events/bus.go        |  5 ++---
 internal/events/dispatcher.go | 22 +++++++++++++++++++++-
 internal/events/pg_test.go    | 12 ++++++++++++
 3 files changed, 35 insertions(+), 4 deletions(-)

diff --git a/internal/events/bus.go b/internal/events/bus.go
index f9eba70f..ec699316 100644
--- a/internal/events/bus.go
+++ b/internal/events/bus.go
@@ -98,8 +98,7 @@ func (b *MemoryBus) Claim(consumer, worker string, n int, now time.Time) ([]*Del
 			continue
 		}
 		env := b.envelopes[id]
-		head := b.heads[env.AggregateID]
-		if env.AggregateVersion != 0 && env.AggregateVersion > head+1 {
+		if !VersionClaimable(b.heads[env.AggregateID], env.AggregateVersion) {
 			continue // gap: do not skip missing versions
 		}
 		token, err := randomToken()
@@ -147,7 +146,7 @@ func (b *MemoryBus) Apply(consumer string, claimed *Delivery, fail error) error
 		return fail
 	}
 	head := b.heads[env.AggregateID]
-	if env.AggregateVersion != 0 && env.AggregateVersion > head+1 {
+	if !VersionClaimable(head, env.AggregateVersion) {
 		d.Status = "pending"
 		d.LeaseOwner = ""
 		d.FencingToken = ""
diff --git a/internal/events/dispatcher.go b/internal/events/dispatcher.go
index 3939239b..a8b7c379 100644
--- a/internal/events/dispatcher.go
+++ b/internal/events/dispatcher.go
@@ -36,6 +36,16 @@ type Lease struct {
 	Envelope      Envelope
 }
 
+// VersionClaimable is the gap rule shared by MemoryBus and claimSQL.
+// Version 0 is unversioned and always claimable. A versioned event is claimable
+// only when it does not skip past the consumer checkpoint (head+1).
+func VersionClaimable(head, version int64) bool {
+	if version == 0 {
+		return true
+	}
+	return version <= head+1
+}
+
 const claimSQL = `
 WITH candidates AS (
   SELECT d.event_id, d.consumer
@@ -62,6 +72,16 @@ WITH candidates AS (
         AND ex.aggregate_type = e.aggregate_type
         AND ex.aggregate_id = e.aggregate_id
     )
+    AND (
+      e.aggregate_version = 0
+      OR e.aggregate_version <= 1 + COALESCE((
+        SELECT c.last_aggregate_version
+        FROM event_consumer_checkpoints c
+        WHERE c.consumer = d.consumer
+          AND c.aggregate_type = e.aggregate_type
+          AND c.aggregate_id = e.aggregate_id
+      ), 0)
+    )
   ORDER BY d.next_attempt_at, d.event_id
   FOR UPDATE OF d SKIP LOCKED
   LIMIT $2
@@ -230,7 +250,7 @@ func AckLease(ctx context.Context, tx pgx.Tx, l Lease) error {
 			) VALUES ($1,$2,$3,$4,$5,$6)
 			ON CONFLICT (consumer, aggregate_type, aggregate_id) DO UPDATE SET
 				last_event_id = EXCLUDED.last_event_id,
-				last_aggregate_version = EXCLUDED.last_aggregate_version,
+				last_aggregate_version = GREATEST(COALESCE(event_consumer_checkpoints.last_aggregate_version, 0), EXCLUDED.last_aggregate_version),
 				fencing_token = EXCLUDED.fencing_token,
 				updated_at = now()
 		`, l.Consumer, l.AggregateType, l.AggregateID, l.EventID, l.Envelope.AggregateVersion, l.Token); err != nil {
diff --git a/internal/events/pg_test.go b/internal/events/pg_test.go
index 25fee6ec..65d485b5 100644
--- a/internal/events/pg_test.go
+++ b/internal/events/pg_test.go
@@ -15,6 +15,18 @@ func TestClaimSQLUsesSkipLockedOnlyForClaim(t *testing.T) {
 	if strings.Count(claimSQL, "SKIP LOCKED") != 1 {
 		t.Fatal("SKIP LOCKED is only for claim")
 	}
+	if !strings.Contains(claimSQL, "event_consumer_checkpoints") || !strings.Contains(claimSQL, "e.aggregate_version = 0") {
+		t.Fatal("claim SQL must hold the same aggregate gap as MemoryBus")
+	}
+}
+
+func TestVersionClaimableMatchesMemoryBusGap(t *testing.T) {
+	if !VersionClaimable(0, 0) || !VersionClaimable(0, 1) || VersionClaimable(0, 2) {
+		t.Fatal("head 0 allows unversioned and v1 only")
+	}
+	if !VersionClaimable(1, 2) || VersionClaimable(1, 3) {
+		t.Fatal("head 1 allows v2 and rejects v3")
+	}
 }
 
 func TestSkipLockedAbsentFromInsertSQL(t *testing.T) {