feat: add DriverWithReceive interface for batch-consuming queue drivers

[h2zi]

Adds an optional DriverWithReceive interface to support high-throughput drivers (e.g., Kafka) that benefit from blocking batch consumption rather than single-job polling.

Contract

// contracts/queue/driver.go
type DriverWithReceive interface {
    Receive(ctx context.Context, queue string, count int) ([]ReservedJob, error)
}

Workers detect this interface at runtime and dispatch to runWithReceive instead of runWithPop.

Worker changes (queue/worker.go)

• runWithReceive: blocking batch loop with 5s per-call timeout derived from shutdownCtx; exponential backoff (100ms–3.2s) on errors/empty batches; context.Canceled skips sleep and re-checks shutdown immediately
• runWithPop: goroutine-local backoff state (eliminates data race on currentDelay); time.Sleep replaced with select on shutdownCtx.Done() for prompt shutdown
• Shutdown: shutdownCtx/shutdownCancel reinitialized in Run() to support restart; jobWg tracks the receive loop so Shutdown() waits before closing failedJobChan
• run(): failed-job processor goroutine started once, shared by both paths; processReservedJob extracted as shared helper

Tests (queue/worker_test.go)

New Test_runWithReceive suite covering: empty batch, receive error, single job, batch jobs, and shutdown-waits-for-in-flight-jobs.

[Copilot]

Pull request overview

This PR adds an optional batch-receive capability to the queue driver contract and updates the queue worker to use it when available (targeting high-throughput drivers like Kafka).

Changes:

• Introduces DriverWithReceive in the queue driver contracts to support blocking batch consumption via Receive(ctx, queue, count).
• Refactors Worker.run() to choose between a Receive-based loop and the existing Pop-based loop.
• Extracts shared reserved-job handling into processReservedJob for reuse across both consumption strategies.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

File	Description
queue/worker.go	Adds Receive-based consumption loop, selects between Receive and Pop, and factors out shared job-processing logic.
contracts/queue/driver.go	Defines the optional DriverWithReceive interface to enable batch blocking receives.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[hwbrzzl]

Thanks, a few questions.

[codecov]

Codecov Report

❌ Patch coverage is 80.21978% with 18 lines in your changes missing coverage. Please review.
✅ Project coverage is 68.78%. Comparing base (55ee67d) to head (4838a69).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
queue/worker.go	80.21%	12 Missing and 6 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1435      +/-   ##
==========================================
+ Coverage   68.76%   68.78%   +0.02%     
==========================================
  Files         361      361              
  Lines       27776    27829      +53     
==========================================
+ Hits        19099    19143      +44     
- Misses       7831     7839       +8     
- Partials      846      847       +1     

☔ 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.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 5 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[hwbrzzl]

@copilot summarize this PR

[Copilot]

@copilot summarize this PR

This PR adds a DriverWithReceive interface to the queue driver contracts, enabling high-throughput drivers (e.g., Kafka) to use blocking batch consumption instead of the existing single-job Pop polling.

Key changes:

• contracts/queue/driver.go: New optional DriverWithReceive interface with Receive(ctx, queue, count) ([]ReservedJob, error) for batch/blocking job retrieval.

• queue/worker.go:

  • run() detects if the driver implements DriverWithReceive and dispatches to runWithReceive or the existing runWithPop.
  • runWithReceive: blocking batch loop with a 5s per-call timeout, exponential backoff (100ms–3.2s) on errors/empty batches, interruptible sleeps via shutdownCtx, and jobWg tracking so Shutdown() waits cleanly.
  • runWithPop: goroutine-local backoff state (no data race), interruptible sleep on shutdownCtx.Done().
  • shutdownCtx/shutdownCancel reinitialized in Run() to support multiple start/stop cycles.
  • Failed-job processor goroutine moved to run() (shared by both paths).

• queue/worker_test.go: New Test_runWithReceive suite covering: no messages, receive errors, single messages, batch messages, and shutdown-waits-for-receive-loop scenarios.

• mocks/queue/: Regenerated mocks for DriverWithReceive and ReservedJob.

[hwbrzzl]

Great, LGTM