From f19f545e7f44e1419a5419e644c3020de9a803e2 Mon Sep 17 00:00:00 2001 From: Blake Watters Date: Wed, 1 Jul 2026 22:27:21 -0400 Subject: [PATCH 1/2] fix(msgqueue): prevent MessageQueueItem FK violation when an auto-deleted queue is reaped mid-publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On the Postgres message-queue backend, triggering a workflow can intermittently fail with `MessageQueueItem_queueId_fkey` (SQLSTATE 23503), surfacing as an opaque gRPC Internal / 500. It's a race: CleanupMessageQueue reaps an auto-deleted queue (idle >1h, refreshed only by a subscriber heartbeat) while a producer, whose 15s existence cache skips the re-bind, inserts against the now-missing parent via the >8000-byte pg_notify fallback. Two complementary fixes: - UpsertMessageQueue refreshes "lastActive" on conflict, so producing counts as activity and an actively-used queue is no longer reaped out from under itself. - New AddMessageEnsuringQueue upserts the parent in the same statement as the item insert, so a reap can't race between "ensure" and "insert". Routed for every autoDeleted queue (the reaper's exact predicate) — including exclusive ones (the dispatcher queue and controller consumer queues), threading the queue's real durable/autoDeleted/exclusive attributes so a recreated row keeps correct semantics. Adds testcontainers coverage for missing-parent create, reaped-queue recreate (non-exclusive and exclusive), and lastActive refresh on bind. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/msgqueue/postgres/exchange.go | 12 +- internal/msgqueue/postgres/msgqueue.go | 47 ++++--- pkg/repository/mq.go | 45 +++--- pkg/repository/mq_selfheal_test.go | 186 +++++++++++++++++++++++++ pkg/repository/sqlcv1/mq.sql | 37 ++++- pkg/repository/sqlcv1/mq.sql.go | 57 +++++++- 6 files changed, 345 insertions(+), 39 deletions(-) create mode 100644 pkg/repository/mq_selfheal_test.go diff --git a/internal/msgqueue/postgres/exchange.go b/internal/msgqueue/postgres/exchange.go index 88e91eb3e5..ef012d8f0c 100644 --- a/internal/msgqueue/postgres/exchange.go +++ b/internal/msgqueue/postgres/exchange.go @@ -25,17 +25,19 @@ func (p *PostgresMessageQueue) addTenantExchangeMessage(ctx context.Context, q m return err } - queueName := msgqueue.GetTenantExchangeName(tenantId) + tenantQueue := msgqueue.TenantEventConsumerQueue(tenantId) // if the queue name does not equal the tenant exchange name, publish the message to the queue - if queueName != q.Name() { - return p.pubNonDurableMessages(ctx, queueName, msg) + if tenantQueue.Name() != q.Name() { + return p.pubNonDurableMessages(ctx, tenantQueue, msg) } return nil } -func (p *PostgresMessageQueue) pubNonDurableMessages(ctx context.Context, queueName string, msg *msgqueue.Message) error { +func (p *PostgresMessageQueue) pubNonDurableMessages(ctx context.Context, queue msgqueue.Queue, msg *msgqueue.Message) error { + durable, autoDeleted, exclusive := bindAttrs(queue) + eg := errgroup.Group{} for _, payload := range msg.Payloads { @@ -48,7 +50,7 @@ func (p *PostgresMessageQueue) pubNonDurableMessages(ctx context.Context, queueN eg.Go(func() error { // Notify will automatically fall back to database storage if the // wrapped message exceeds pg_notify's 8KB limit - return p.repo.Notify(ctx, queueName, string(msgBytes)) + return p.repo.Notify(ctx, queue.Name(), string(msgBytes), durable, autoDeleted, exclusive) }) } else { p.l.Error().Ctx(ctx).Err(err).Msg("error marshalling message") diff --git a/internal/msgqueue/postgres/msgqueue.go b/internal/msgqueue/postgres/msgqueue.go index a756d8f637..eba0d14539 100644 --- a/internal/msgqueue/postgres/msgqueue.go +++ b/internal/msgqueue/postgres/msgqueue.go @@ -150,8 +150,15 @@ func (p *PostgresMessageQueue) addMessage(ctx context.Context, queue msgqueue.Qu return err } + durable, autoDeleted, exclusive := bindAttrs(queue) + if !queue.Durable() { - err = p.pubNonDurableMessages(ctx, queue.Name(), task) + err = p.pubNonDurableMessages(ctx, queue, task) + } else if autoDeleted { + // Any auto-deleted queue is reap-eligible (CleanupMessageQueue keys on + // autoDeleted alone), so self-heal the parent on insert — including the + // durable+exclusive dispatcher queue, which is expirable⇒autoDeleted. + err = p.repo.AddMessageEnsuringQueue(ctx, queue.Name(), msgBytes, durable, autoDeleted, exclusive) } else { err = p.repo.AddMessage(ctx, queue.Name(), msgBytes) } @@ -162,7 +169,7 @@ func (p *PostgresMessageQueue) addMessage(ctx context.Context, queue msgqueue.Qu } // notify the queue that a new message has been added - err = p.repo.Notify(ctx, queue.Name(), "") + err = p.repo.Notify(ctx, queue.Name(), "", durable, autoDeleted, exclusive) if err != nil { p.l.Error().Err(err).Msgf("error notifying queue %s", queue.Name()) @@ -342,12 +349,10 @@ func (p *PostgresMessageQueue) IsReady() bool { return true } -func (p *PostgresMessageQueue) upsertQueue(ctx context.Context, queue msgqueue.Queue) error { - if valid, exists := p.ttlCache.Get(queue.Name()); valid && exists { - return nil - } +func bindAttrs(queue msgqueue.Queue) (durable, autoDeleted, exclusive bool) { + durable = queue.Durable() - exclusive := queue.Exclusive() + exclusive = queue.Exclusive() // If the queue is a fanout exchange, then it is not exclusive. This is different from the RabbitMQ // implementation, where a fanout exchange will map to an exclusively bound queue which has a random @@ -356,14 +361,7 @@ func (p *PostgresMessageQueue) upsertQueue(ctx context.Context, queue msgqueue.Q exclusive = false } - var consumer *string - - if exclusive { - str := uuid.New().String() - consumer = &str - } - - autoDeleted := queue.AutoDeleted() + autoDeleted = queue.AutoDeleted() // FIXME: note that this differs from the RabbitMQ implementation, since we auto-delete Postgres MQs after // 1 hour of inactivity instead of immediately. So if the queue is expirable, we set it to autoDeleted and @@ -372,8 +370,25 @@ func (p *PostgresMessageQueue) upsertQueue(ctx context.Context, queue msgqueue.Q autoDeleted = true } + return durable, autoDeleted, exclusive +} + +func (p *PostgresMessageQueue) upsertQueue(ctx context.Context, queue msgqueue.Queue) error { + if valid, exists := p.ttlCache.Get(queue.Name()); valid && exists { + return nil + } + + durable, autoDeleted, exclusive := bindAttrs(queue) + + var consumer *string + + if exclusive { + str := uuid.New().String() + consumer = &str + } + // bind the queue - err := p.repo.BindQueue(ctx, queue.Name(), queue.Durable(), autoDeleted, exclusive, consumer) + err := p.repo.BindQueue(ctx, queue.Name(), durable, autoDeleted, exclusive, consumer) if err != nil { p.l.Error().Err(err).Msg("error binding queue") diff --git a/pkg/repository/mq.go b/pkg/repository/mq.go index 78b0fd9a56..aab0233304 100644 --- a/pkg/repository/mq.go +++ b/pkg/repository/mq.go @@ -5,13 +5,11 @@ import ( "encoding/json" "errors" "fmt" - "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" - "github.com/hatchet-dev/hatchet/pkg/repository/sqlchelpers" "github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1" "github.com/hatchet-dev/hatchet/pkg/telemetry" ) @@ -24,7 +22,7 @@ type PubSubMessage struct { type MessageQueueRepository interface { // PubSub Listen(ctx context.Context, name string, f func(ctx context.Context, notification *PubSubMessage) error) error - Notify(ctx context.Context, name string, payload string) error + Notify(ctx context.Context, name string, payload string, durable, autoDeleted, exclusive bool) error // Queues BindQueue(ctx context.Context, queue string, durable, autoDeleted, exclusive bool, exclusiveConsumer *string) error @@ -33,6 +31,7 @@ type MessageQueueRepository interface { // Messages AddMessage(ctx context.Context, queue string, payload []byte) error + AddMessageEnsuringQueue(ctx context.Context, queue string, payload []byte, durable, autoDeleted, exclusive bool) error ReadMessages(ctx context.Context, queue string, qos int) ([]*sqlcv1.ReadMessagesRow, error) AckMessage(ctx context.Context, id int64) error CleanupMessageQueueItems(ctx context.Context) error @@ -60,7 +59,7 @@ func (m *messageQueueRepository) Listen(ctx context.Context, name string, f func return m.m.listen(ctx, name, f) } -func (m *messageQueueRepository) Notify(ctx context.Context, name string, payload string) error { +func (m *messageQueueRepository) Notify(ctx context.Context, name string, payload string, durable, autoDeleted, exclusive bool) error { wrappedPayload, err := m.m.wrapMessage(name, payload) if err != nil { m.l.Error().Ctx(ctx).Err(err).Msg("error wrapping message") @@ -70,6 +69,17 @@ func (m *messageQueueRepository) Notify(ctx context.Context, name string, payloa // PostgreSQL's pg_notify has an 8000 byte limit // If the wrapped message exceeds this, fall back to database storage if len(wrappedPayload) > 8000 { + // An auto-deleted queue can be reaped by CleanupMessageQueue in the window + // between the producer's 15s existence-cache hit and this insert, so ensure + // the parent row exists in the same statement (FK can't fail). This covers + // EXCLUSIVE auto-deleted queues too — the dispatcher queue (expirable ⇒ + // autoDeleted, exclusive) and controller consumer queues — which the reaper + // deletes regardless of exclusivity; gating on `!exclusive` left them + // exposed to the exact 23503 this fix exists to kill. + if autoDeleted { + return m.AddMessageEnsuringQueue(ctx, name, []byte(payload), durable, autoDeleted, exclusive) + } + return m.AddMessage(ctx, name, []byte(payload)) } @@ -77,21 +87,24 @@ func (m *messageQueueRepository) Notify(ctx context.Context, name string, payloa } func (m *messageQueueRepository) AddMessage(ctx context.Context, queue string, payload []byte) error { - p := []sqlcv1.BulkAddMessageParams{} - - p = append(p, sqlcv1.BulkAddMessageParams{ - QueueId: pgtype.Text{ - String: queue, - Valid: true, - }, - Payload: payload, - ExpiresAt: sqlchelpers.TimestampFromTime(time.Now().UTC().Add(5 * time.Minute)), - ReadAfter: sqlchelpers.TimestampFromTime(time.Now().UTC()), + return m.queries.AddMessage(ctx, m.pool, sqlcv1.AddMessageParams{ + Queueid: queue, + Payload: payload, }) +} - _, err := m.queries.BulkAddMessage(ctx, m.pool, p) +func (m *messageQueueRepository) AddMessageEnsuringQueue(ctx context.Context, queue string, payload []byte, durable, autoDeleted, exclusive bool) error { + if !autoDeleted { + return m.AddMessage(ctx, queue, payload) + } - return err + return m.queries.AddMessageEnsuringQueue(ctx, m.pool, sqlcv1.AddMessageEnsuringQueueParams{ + Queueid: queue, + Payload: payload, + Durable: durable, + Autodeleted: autoDeleted, + Exclusive: exclusive, + }) } func (m *messageQueueRepository) BindQueue(ctx context.Context, queue string, durable, autoDeleted, exclusive bool, exclusiveConsumer *string) error { diff --git a/pkg/repository/mq_selfheal_test.go b/pkg/repository/mq_selfheal_test.go new file mode 100644 index 0000000000..8528a7f24b --- /dev/null +++ b/pkg/repository/mq_selfheal_test.go @@ -0,0 +1,186 @@ +//go:build !e2e && !load && !rampup && !integration + +package repository + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1" +) + +func TestAddMessageEnsuringQueueCreatesMissingQueue(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + q := sqlcv1.New() + const name = "mq-selfheal-missing-queue" + + err := q.AddMessageEnsuringQueue(ctx, pool, sqlcv1.AddMessageEnsuringQueueParams{ + Queueid: name, + Payload: []byte(`{"hello":"world"}`), + Durable: false, + Autodeleted: true, + Exclusive: false, + }) + require.NoError(t, err, "AddMessageEnsuringQueue must create the parent queue, not raise MessageQueueItem_queueId_fkey (23503)") + + var autoDeleted, exclusive bool + err = pool.QueryRow(ctx, `SELECT "autoDeleted", "exclusive" FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&autoDeleted, &exclusive) + require.NoError(t, err) + assert.True(t, autoDeleted, "the recreated queue must mirror the supplied bind attributes") + assert.False(t, exclusive, "the recreated queue must mirror the supplied exclusive flag") + + var items int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueueItem" WHERE "queueId" = $1`, name).Scan(&items) + require.NoError(t, err) + assert.Equal(t, 1, items) +} + +func TestAddMessageEnsuringQueueSurvivesQueueGC(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + q := sqlcv1.New() + const name = "mq-selfheal-gc-queue" + + _, err := q.UpsertMessageQueue(ctx, pool, sqlcv1.UpsertMessageQueueParams{ + Name: name, + Durable: false, + Autodeleted: true, + Exclusive: false, + }) + require.NoError(t, err) + + _, err = pool.Exec(ctx, + `UPDATE "MessageQueue" SET "lastActive" = NOW() - INTERVAL '2 hours' WHERE "name" = $1`, + name, + ) + require.NoError(t, err) + + require.NoError(t, q.CleanupMessageQueue(ctx, pool)) + + var reaped int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&reaped) + require.NoError(t, err) + require.Equal(t, 0, reaped, "queue should have been reaped, setting up the race") + + err = q.AddMessageEnsuringQueue(ctx, pool, sqlcv1.AddMessageEnsuringQueueParams{ + Queueid: name, + Payload: []byte(`{"hello":"world"}`), + Durable: false, + Autodeleted: true, + }) + require.NoError(t, err, "AddMessageEnsuringQueue must recreate a GC'd parent, not raise 23503") + + var exists int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&exists) + require.NoError(t, err) + assert.Equal(t, 1, exists, "the parent queue must be recreated by AddMessageEnsuringQueue") +} + +func TestAddMessageEnsuringQueueRecreatesReapedExclusiveQueue(t *testing.T) { + // Regression for the autoDeleted+exclusive gap: CleanupMessageQueue reaps on + // "autoDeleted" alone, so exclusive auto-deleted queues — the dispatcher queue + // (expirable ⇒ autoDeleted, exclusive) and controller consumer queues — are + // reap-eligible too. A message after a reap must recreate the parent with + // exclusive preserved, not raise 23503. (The `!exclusive` gate that this test + // guards against left exactly these queues exposed.) + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + q := sqlcv1.New() + const name = "mq-selfheal-exclusive-gc-queue" + + _, err := q.UpsertMessageQueue(ctx, pool, sqlcv1.UpsertMessageQueueParams{ + Name: name, + Durable: true, + Autodeleted: true, + Exclusive: true, + }) + require.NoError(t, err) + + _, err = pool.Exec(ctx, + `UPDATE "MessageQueue" SET "lastActive" = NOW() - INTERVAL '2 hours' WHERE "name" = $1`, + name, + ) + require.NoError(t, err) + + require.NoError(t, q.CleanupMessageQueue(ctx, pool)) + + var reaped int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&reaped) + require.NoError(t, err) + require.Equal(t, 0, reaped, "an exclusive auto-deleted queue must also be reaped — autoDeleted alone triggers cleanup") + + err = q.AddMessageEnsuringQueue(ctx, pool, sqlcv1.AddMessageEnsuringQueueParams{ + Queueid: name, + Payload: []byte(`{"hello":"world"}`), + Durable: true, + Autodeleted: true, + Exclusive: true, + }) + require.NoError(t, err, "AddMessageEnsuringQueue must recreate a reaped EXCLUSIVE queue, not raise 23503") + + var exclusive, consumerIsNull bool + err = pool.QueryRow(ctx, + `SELECT "exclusive", "exclusiveConsumerId" IS NULL FROM "MessageQueue" WHERE "name" = $1`, + name, + ).Scan(&exclusive, &consumerIsNull) + require.NoError(t, err) + assert.True(t, exclusive, "the recreated queue must preserve exclusive=true") + assert.True(t, consumerIsNull, "a self-healed exclusive queue has no consumer until the next BindQueue") + + var items int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueueItem" WHERE "queueId" = $1`, name).Scan(&items) + require.NoError(t, err) + assert.Equal(t, 1, items, "the message must be stored against the recreated queue") +} + +func TestBindRefreshesLastActive(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + q := sqlcv1.New() + const name = "mq-selfheal-bind-refresh-queue" + + bind := func() { + _, err := q.UpsertMessageQueue(ctx, pool, sqlcv1.UpsertMessageQueueParams{ + Name: name, + Durable: false, + Autodeleted: true, + Exclusive: false, + }) + require.NoError(t, err) + } + + bind() + + _, err := pool.Exec(ctx, + `UPDATE "MessageQueue" SET "lastActive" = NOW() - INTERVAL '2 hours' WHERE "name" = $1`, + name, + ) + require.NoError(t, err) + + bind() + + var lastActive time.Time + err = pool.QueryRow(ctx, `SELECT "lastActive" FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&lastActive) + require.NoError(t, err) + assert.WithinDuration(t, time.Now(), lastActive, time.Minute) + + require.NoError(t, q.CleanupMessageQueue(ctx, pool)) + + var exists int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&exists) + require.NoError(t, err) + assert.Equal(t, 1, exists, "a recently rebound queue must survive cleanup") +} diff --git a/pkg/repository/sqlcv1/mq.sql b/pkg/repository/sqlcv1/mq.sql index 7d65f12c69..a968cf7fbc 100644 --- a/pkg/repository/sqlcv1/mq.sql +++ b/pkg/repository/sqlcv1/mq.sql @@ -18,7 +18,8 @@ SET "durable" = @durable::boolean, "autoDeleted" = @autoDeleted::boolean, "exclusive" = @exclusive::boolean, - "exclusiveConsumerId" = CASE WHEN sqlc.narg('exclusiveConsumerId')::uuid IS NOT NULL THEN sqlc.narg('exclusiveConsumerId')::uuid ELSE NULL END + "exclusiveConsumerId" = CASE WHEN sqlc.narg('exclusiveConsumerId')::uuid IS NOT NULL THEN sqlc.narg('exclusiveConsumerId')::uuid ELSE NULL END, + "lastActive" = NOW() RETURNING *; -- name: UpdateMessageQueueActive :exec @@ -52,6 +53,40 @@ VALUES NOW() + INTERVAL '5 minutes' ); +-- name: AddMessageEnsuringQueue :exec +WITH ensure_queue AS ( + INSERT INTO "MessageQueue" ( + "name", + "lastActive", + "durable", + "autoDeleted", + "exclusive" + ) + VALUES ( + @queueId::text, + NOW(), + @durable::boolean, + @autoDeleted::boolean, + @exclusive::boolean + ) + ON CONFLICT ("name") DO UPDATE + SET "lastActive" = NOW() + RETURNING "name" +) +INSERT INTO + "MessageQueueItem" ( + "payload", + "queueId", + "readAfter", + "expiresAt" + ) +SELECT + @payload::jsonb, + "name", + NOW(), + NOW() + INTERVAL '5 minutes' +FROM ensure_queue; + -- name: BulkAddMessage :copyfrom INSERT INTO "MessageQueueItem" ( diff --git a/pkg/repository/sqlcv1/mq.sql.go b/pkg/repository/sqlcv1/mq.sql.go index 9f45ec7b07..4130d1b5a3 100644 --- a/pkg/repository/sqlcv1/mq.sql.go +++ b/pkg/repository/sqlcv1/mq.sql.go @@ -39,6 +39,60 @@ func (q *Queries) AddMessage(ctx context.Context, db DBTX, arg AddMessageParams) return err } +const addMessageEnsuringQueue = `-- name: AddMessageEnsuringQueue :exec +WITH ensure_queue AS ( + INSERT INTO "MessageQueue" ( + "name", + "lastActive", + "durable", + "autoDeleted", + "exclusive" + ) + VALUES ( + $2::text, + NOW(), + $3::boolean, + $4::boolean, + $5::boolean + ) + ON CONFLICT ("name") DO UPDATE + SET "lastActive" = NOW() + RETURNING "name" +) +INSERT INTO + "MessageQueueItem" ( + "payload", + "queueId", + "readAfter", + "expiresAt" + ) +SELECT + $1::jsonb, + "name", + NOW(), + NOW() + INTERVAL '5 minutes' +FROM ensure_queue +` + +type AddMessageEnsuringQueueParams struct { + Payload []byte `json:"payload"` + Queueid string `json:"queueid"` + Durable bool `json:"durable"` + Autodeleted bool `json:"autodeleted"` + Exclusive bool `json:"exclusive"` +} + +func (q *Queries) AddMessageEnsuringQueue(ctx context.Context, db DBTX, arg AddMessageEnsuringQueueParams) error { + _, err := db.Exec(ctx, addMessageEnsuringQueue, + arg.Payload, + arg.Queueid, + arg.Durable, + arg.Autodeleted, + arg.Exclusive, + ) + return err +} + const bulkAckMessages = `-- name: BulkAckMessages :exec DELETE FROM "MessageQueueItem" @@ -227,7 +281,8 @@ SET "durable" = $2::boolean, "autoDeleted" = $3::boolean, "exclusive" = $4::boolean, - "exclusiveConsumerId" = CASE WHEN $5::uuid IS NOT NULL THEN $5::uuid ELSE NULL END + "exclusiveConsumerId" = CASE WHEN $5::uuid IS NOT NULL THEN $5::uuid ELSE NULL END, + "lastActive" = NOW() RETURNING name, "lastActive", durable, "autoDeleted", exclusive, "exclusiveConsumerId" ` From 46d11907e329f87ad598e52302b0cbb9b741a5d8 Mon Sep 17 00:00:00 2001 From: Blake Watters Date: Wed, 8 Jul 2026 21:16:24 -0400 Subject: [PATCH 2/2] perf(msgqueue): publish optimistically, run the ensure-queue upsert only on an actual FK violation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unconditional ensure-on-every-publish CTE (ON CONFLICT DO UPDATE SET lastActive = NOW()) takes a row-exclusive tuple lock on the parent MessageQueue row per publish, so all concurrent publishers to the same auto-deleted queue (including every dispatcher-queue message) serialize on a handful of rows. Measured in production: 93% of the statement's execution time was lock wait, 85% of total instance DB time. The plain MessageQueueItem insert's FK check takes only a KEY SHARE lock, which does not conflict with other KEY SHARE locks (publishers never wait on publishers) but does conflict with DELETE (a concurrent reap and an in-flight insert serialize correctly). So: insert optimistically, and run the ensure-queue CTE only when the insert fails with 23503 — i.e. only when the parent was actually reaped. With the bind path refreshing lastActive behind the producer's 15s existence cache (earlier commit in this PR), an actively-published queue can never accrue the 1h of staleness the reaper requires, so the heal fires at most once per queue per reap; it logs at WARN because a sustained rate indicates out-of-band MessageQueue deletes. Co-Authored-By: Claude Fable 5 --- pkg/repository/mq.go | 43 +++++++++-- pkg/repository/mq_selfheal_test.go | 118 +++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 8 deletions(-) diff --git a/pkg/repository/mq.go b/pkg/repository/mq.go index aab0233304..a5cd5b2445 100644 --- a/pkg/repository/mq.go +++ b/pkg/repository/mq.go @@ -7,7 +7,9 @@ import ( "fmt" "github.com/google/uuid" + "github.com/jackc/pgerrcode" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgtype" "github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1" @@ -70,12 +72,11 @@ func (m *messageQueueRepository) Notify(ctx context.Context, name string, payloa // If the wrapped message exceeds this, fall back to database storage if len(wrappedPayload) > 8000 { // An auto-deleted queue can be reaped by CleanupMessageQueue in the window - // between the producer's 15s existence-cache hit and this insert, so ensure - // the parent row exists in the same statement (FK can't fail). This covers - // EXCLUSIVE auto-deleted queues too — the dispatcher queue (expirable ⇒ - // autoDeleted, exclusive) and controller consumer queues — which the reaper - // deletes regardless of exclusivity; gating on `!exclusive` left them - // exposed to the exact 23503 this fix exists to kill. + // between the producer's 15s existence-cache hit and this insert, so route + // through the self-healing path. This covers EXCLUSIVE auto-deleted queues + // too — the dispatcher queue (expirable ⇒ autoDeleted, exclusive) and + // controller consumer queues — which the reaper deletes regardless of + // exclusivity. if autoDeleted { return m.AddMessageEnsuringQueue(ctx, name, []byte(payload), durable, autoDeleted, exclusive) } @@ -94,10 +95,31 @@ func (m *messageQueueRepository) AddMessage(ctx context.Context, queue string, p } func (m *messageQueueRepository) AddMessageEnsuringQueue(ctx context.Context, queue string, payload []byte, durable, autoDeleted, exclusive bool) error { - if !autoDeleted { - return m.AddMessage(ctx, queue, payload) + // The plain insert's FK check takes only a KEY SHARE lock on the parent + // MessageQueue row, so concurrent publishers to the same queue do not + // serialize on it (an unconditional ON CONFLICT DO UPDATE here takes a + // row-exclusive tuple lock per publish, convoying every publisher on one + // row). It fails with a foreign-key violation exactly when the parent has + // been reaped, which is the only case that needs the self-heal. + err := m.AddMessage(ctx, queue, payload) + + if err == nil || !autoDeleted || !isForeignKeyViolation(err) { + return err } + // The parent queue was reaped by CleanupMessageQueue (auto-deleted queues + // are reap-eligible after 1h idle). Healing is expected at most once per + // queue per reap: the producer's queue-bind cache expires every 15s and + // each re-bind refreshes lastActive, so an actively-published queue can + // never accrue the 1h of staleness the reaper requires (15s ≪ 1h). A + // sustained heal rate therefore means something else is deleting + // MessageQueue rows out-of-band. + m.l.Warn().Ctx(ctx).Str("queue", queue).Msg("parent queue row missing on publish; self-healing by recreating it") + + // Recreate the queue and insert the item in a single atomic statement. + // The upsert sets lastActive = NOW(), so the healed queue is not + // reap-eligible for another hour and subsequent publishes take the plain + // insert above. return m.queries.AddMessageEnsuringQueue(ctx, m.pool, sqlcv1.AddMessageEnsuringQueueParams{ Queueid: queue, Payload: payload, @@ -107,6 +129,11 @@ func (m *messageQueueRepository) AddMessageEnsuringQueue(ctx context.Context, qu }) } +func isForeignKeyViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == pgerrcode.ForeignKeyViolation +} + func (m *messageQueueRepository) BindQueue(ctx context.Context, queue string, durable, autoDeleted, exclusive bool, exclusiveConsumer *string) error { // if exclusive, but no consumer, return error if exclusive && exclusiveConsumer == nil { diff --git a/pkg/repository/mq_selfheal_test.go b/pkg/repository/mq_selfheal_test.go index 8528a7f24b..aeaff76bdc 100644 --- a/pkg/repository/mq_selfheal_test.go +++ b/pkg/repository/mq_selfheal_test.go @@ -7,6 +7,8 @@ import ( "testing" "time" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -184,3 +186,119 @@ func TestBindRefreshesLastActive(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, exists, "a recently rebound queue must survive cleanup") } + +func mqRepoForTest(pool *pgxpool.Pool) *messageQueueRepository { + l := zerolog.Nop() + + return &messageQueueRepository{ + sharedRepository: &sharedRepository{ + pool: pool, + queries: sqlcv1.New(), + l: &l, + }, + } +} + +func TestAddMessageEnsuringQueueFastPathSkipsParentUpsert(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + repo := mqRepoForTest(pool) + q := sqlcv1.New() + const name = "mq-optimistic-fast-path-queue" + + _, err := q.UpsertMessageQueue(ctx, pool, sqlcv1.UpsertMessageQueueParams{ + Name: name, + Durable: false, + Autodeleted: true, + Exclusive: false, + }) + require.NoError(t, err) + + // Backdate lastActive so any parent-row write by the publish is detectable. + _, err = pool.Exec(ctx, + `UPDATE "MessageQueue" SET "lastActive" = NOW() - INTERVAL '30 minutes' WHERE "name" = $1`, + name, + ) + require.NoError(t, err) + + err = repo.AddMessageEnsuringQueue(ctx, name, []byte(`{"hello":"world"}`), false, true, false) + require.NoError(t, err) + + var lastActive time.Time + err = pool.QueryRow(ctx, `SELECT "lastActive" FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&lastActive) + require.NoError(t, err) + assert.Less(t, time.Since(lastActive), 45*time.Minute) + assert.Greater(t, time.Since(lastActive), 15*time.Minute, + "a publish to an existing queue must not rewrite the parent row (no ON CONFLICT DO UPDATE on the hot path)") + + var items int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueueItem" WHERE "queueId" = $1`, name).Scan(&items) + require.NoError(t, err) + assert.Equal(t, 1, items) +} + +func TestAddMessageEnsuringQueueHealsReapedParentOnFKViolation(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + repo := mqRepoForTest(pool) + q := sqlcv1.New() + const name = "mq-optimistic-heal-queue" + + _, err := q.UpsertMessageQueue(ctx, pool, sqlcv1.UpsertMessageQueueParams{ + Name: name, + Durable: true, + Autodeleted: true, + Exclusive: true, + }) + require.NoError(t, err) + + err = repo.AddMessageEnsuringQueue(ctx, name, []byte(`{"seq":1}`), true, true, true) + require.NoError(t, err) + + // Reap the queue mid-stream, exactly as CleanupMessageQueue does in the race. + _, err = pool.Exec(ctx, + `UPDATE "MessageQueue" SET "lastActive" = NOW() - INTERVAL '2 hours' WHERE "name" = $1`, + name, + ) + require.NoError(t, err) + require.NoError(t, q.CleanupMessageQueue(ctx, pool)) + + var reaped int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueue" WHERE "name" = $1`, name).Scan(&reaped) + require.NoError(t, err) + require.Equal(t, 0, reaped, "queue should have been reaped, setting up the race") + + err = repo.AddMessageEnsuringQueue(ctx, name, []byte(`{"seq":2}`), true, true, true) + require.NoError(t, err, "a publish against a reaped queue must self-heal, not surface 23503") + + var exclusive bool + var lastActive time.Time + err = pool.QueryRow(ctx, + `SELECT "exclusive", "lastActive" FROM "MessageQueue" WHERE "name" = $1`, name, + ).Scan(&exclusive, &lastActive) + require.NoError(t, err) + assert.True(t, exclusive, "the healed queue must preserve the supplied bind attributes") + assert.WithinDuration(t, time.Now(), lastActive, time.Minute, + "the heal must refresh lastActive so the queue is not immediately reap-eligible again") + + var items int + err = pool.QueryRow(ctx, `SELECT count(*) FROM "MessageQueueItem" WHERE "queueId" = $1`, name).Scan(&items) + require.NoError(t, err) + assert.Equal(t, 1, items, "the post-reap message must be stored against the recreated queue") +} + +func TestAddMessageEnsuringQueueNonAutoDeletedPropagatesFKViolation(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + repo := mqRepoForTest(pool) + + err := repo.AddMessageEnsuringQueue(ctx, "mq-optimistic-missing-static-queue", []byte(`{}`), true, false, false) + require.Error(t, err, "non-auto-deleted queues are never reaped, so a missing parent is a real bug and must not be masked by a self-heal") + assert.True(t, isForeignKeyViolation(err)) +}