fix: Pop is incorrect while concurrent

[hwbrzzl]

Greptile Overview

Greptile Summary

• Adds a per-queue in-process mutex around Database.Pop() using a sync.Map of *sync.Mutex keyed by queue name.
• Intended to address concurrency issues when reserving jobs from the database-backed queue.
• Change affects the database queue driver’s reservation path but does not change the underlying DB transaction/row locking behavior across multiple worker processes/instances.

Confidence Score: 2/5

• This PR has correctness and operational risks that should be addressed before merging.
• The new mutex only coordinates Pop() within a single process, so it cannot fix cross-process/cluster concurrency issues; additionally, the per-queue mutex map grows without bound when queue names vary, creating a real memory growth risk in long-lived services.
• queue/driver_database.go

Important Files Changed

Filename	Overview
queue/driver_database.go	Adds a per-queue in-memory mutex around Pop() using sync.Map to serialize pops by queue; this introduces cross-process concurrency gaps and an unbounded per-queue mutex map.

Sequence Diagram

sequenceDiagram
    participant W1 as Worker (proc A)
    participant DB as Database driver
    participant SQL as DB
    participant W2 as Worker (proc B)

    W1->>DB: Pop(queue)
    DB->>DB: Lock per-queue mutex (proc A only)
    DB->>SQL: BEGIN
    DB->>SQL: SELECT ... FOR UPDATE (first available job)
    SQL-->>DB: job row
    DB->>SQL: UPDATE job reserved_at, attempts
    DB->>SQL: COMMIT
    DB-->>W1: ReservedJob

    W2->>DB: Pop(queue)
    Note over W2,DB: No shared mutex across processes/instances
    DB->>SQL: BEGIN
    DB->>SQL: SELECT ... FOR UPDATE (next available job)
    SQL-->>DB: different job row
    DB->>SQL: UPDATE job reserved_at, attempts
    DB->>SQL: COMMIT
    DB-->>W2: ReservedJob

    Note over W1,W2: In-memory mutex does not prevent
    Note over W1,W2: multiple concurrent reservations globally.

Loading

[greptile-apps]

_{1 file reviewed, 2 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Additional Comments (2)

queue/driver_database.go
In-memory lock doesn't ensure correctness.
This per-queue sync.Mutex only serializes Pop() calls within the same Go process; multiple workers/processes (or multiple app instances) will still concurrently SELECT ... FOR UPDATE different rows and reserve multiple jobs from the same queue. If the intended fix is to prevent more than one concurrent reservation per queue globally, this needs a DB-level constraint/locking strategy (e.g., queue-level lock table/advisory lock) rather than an in-process mutex.

Prompt To Fix With AI

This is a comment left during a code review.
Path: queue/driver_database.go
Line: 55:57

Comment:
**In-memory lock doesn't ensure** correctness.
This per-queue `sync.Mutex` only serializes `Pop()` calls within the same Go process; multiple workers/processes (or multiple app instances) will still concurrently `SELECT ... FOR UPDATE` different rows and reserve multiple jobs from the same queue. If the intended fix is to prevent more than one concurrent reservation per queue globally, this needs a DB-level constraint/locking strategy (e.g., queue-level lock table/advisory lock) rather than an in-process mutex.

How can I resolve this? If you propose a fix, please make it concise.

---

queue/driver_database.go
Unbounded mutex growth.
queueMutexes stores one *sync.Mutex per distinct queue string and never evicts entries, so a system that uses dynamic/tenant-based queue names will leak memory over time. If queue names aren’t strictly bounded, consider a bounded/evicting structure or avoid storing per-queue locks altogether.

Prompt To Fix With AI

This is a comment left during a code review.
Path: queue/driver_database.go
Line: 34:34

Comment:
**Unbounded mutex growth**.
`queueMutexes` stores one `*sync.Mutex` per distinct queue string and never evicts entries, so a system that uses dynamic/tenant-based queue names will leak memory over time. If queue names aren’t strictly bounded, consider a bounded/evicting structure or avoid storing per-queue locks altogether.

How can I resolve this? If you propose a fix, please make it concise.

[codecov]

Codecov Report

❌ Patch coverage is 65.21739% with 8 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (v1.17.x@e53059d). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
queue/application.go	28.57%	5 Missing ⚠️
queue/driver_database.go	75.00%	1 Missing and 1 partial ⚠️
queue/service_provider.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             v1.17.x    #1372   +/-   ##
==========================================
  Coverage           ?   68.71%           
==========================================
  Files              ?      281           
  Lines              ?    17768           
  Branches           ?        0           
==========================================
  Hits               ?    12209           
  Misses             ?     5055           
  Partials           ?      504           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.