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

@@ -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 = ""

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 {

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) {