From a34a50814026ee74abe8b0bc177538b8e6b8543a Mon Sep 17 00:00:00 2001 From: wangbo Date: Fri, 31 Jul 2026 08:40:54 +0800 Subject: [PATCH] =?UTF-8?q?perf(worker):=20=E5=85=AC=E5=B9=B3=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E8=B7=A8=E8=8A=82=E7=82=B9=E5=BC=82=E6=AD=A5=E5=87=86?= =?UTF-8?q?=E5=85=A5=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 持续突发提交会通过 pg_try_advisory_xact_lock 反复抢占全局容量锁，使 Worker 补槽器在首批之后饥饿。\n\n跨进程锁改为 PostgreSQL 公平等待；进程内锁仍保证每个 API 或 Worker 最多一条连接参与等待，30 秒 lock_timeout 会回到原有重试路径，60 秒空闲事务超时继续处理失主会话。\n\n验证：临时 PostgreSQL 18 上 64 并发锁竞争峰值连接不超过 2；Go 全量测试、go vet、gofmt 通过。 --- apps/api/internal/store/admission_lock.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/apps/api/internal/store/admission_lock.go b/apps/api/internal/store/admission_lock.go index 970411c..ad9d9c5 100644 --- a/apps/api/internal/store/admission_lock.go +++ b/apps/api/internal/store/admission_lock.go @@ -115,18 +115,21 @@ func admissionOperationLockKeys(input TaskAdmissionInput) []string { } func tryAdmissionTransactionLock(ctx context.Context, tx pgx.Tx, key string) error { - var locked bool - if err := tx.QueryRow( + // The process-local lock limits this wait to one PostgreSQL connection per + // API or Worker process. Let PostgreSQL queue those process representatives + // fairly: repeated pg_try_advisory_xact_lock calls let a sustained submit + // burst starve the Worker dispatcher that must release queue pressure. + // lock_timeout bounds a lost-owner wait and converts it back into the + // existing transaction retry path. + _, err := tx.Exec( ctx, - `SELECT pg_try_advisory_xact_lock(hashtextextended($1, 0))`, + `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, key, - ).Scan(&locked); err != nil { - return err - } - if !locked { + ) + if IsPostgresLockTimeout(err) { return errAdmissionLockBusy } - return nil + return err } func retryAdmissionOperation[T any](