Queues & Background Jobs

// swift-tools-version:6.3
import PackageDescription

let package = Package(
    name: "Bookshelf",
    platforms: [.macOS(.v13)],
    dependencies: [
        .package(url: "https://github.com/vapor/vapor.git", from: "4.122.2"),
        .package(url: "https://github.com/vapor/queues.git", from: "1.18.0"),
        .package(url: "https://github.com/vapor/queues-redis-driver.git", from: "1.1.2"),
    ],
    targets: [
        .executableTarget(name: "App", dependencies: [
            .product(name: "Vapor", package: "vapor"),
            .product(name: "Queues", package: "queues"),
            .product(name: "QueuesRedisDriver", package: "queues-redis-driver"),
        ]),
        .testTarget(name: "AppTests", dependencies: [
            .target(name: "App"),
            .product(name: "VaporTesting", package: "vapor"),
            .product(name: "XCTQueues", package: "queues"),
        ]),
    ]
)

import Queues
import QueuesRedisDriver
import Vapor

func configure(_ app: Application, retryBaseSeconds: Int = 30) async throws {
    // 1. Driver. Tests choose their own driver, so the production one is skipped there.
    if app.environment != .testing {
        let url = Environment.get("REDIS_URL") ?? "redis://127.0.0.1:6379"
        try app.queues.use(.redis(url: url))
    }

    // 2. Tuning.
    app.queues.configuration.refreshInterval = .seconds(1)
    app.queues.configuration.persistenceKey = "bookshelf"
    app.queues.configuration.workerCount = 2

    // 3. Every job type must be registered, in the web process and in the worker process.
    app.queues.add(WelcomeEmailJob(retryBaseSeconds: retryBaseSeconds))
    app.queues.add(LoanReminderJob())

    // 4. Monitoring.
    app.queues.add(JobAuditDelegate(logger: app.logger, stats: app.jobStats))

    // 5. Optional in-process workers, for small deployments and local development.
    if Environment.get("QUEUES_IN_PROCESS") == "true" {
        try app.queues.startInProcessJobs(on: .emails)
        try app.queues.startInProcessJobs(on: .reminders)
    }

    try routes(app)
}

extension Application {
    private struct JobStatsKey: StorageKey { typealias Value = JobStats }

    var jobStats: JobStats {
        if let existing = storage[JobStatsKey.self] { return existing }
        let stats = JobStats()
        storage[JobStatsKey.self] = stats
        return stats
    }
}

import Queues
import QueuesRedisDriver
import Vapor

/// The same driver with an explicit pool, a connection timeout and TLS taken from a `rediss://` URL.
func makeRedisQueuesProvider() throws -> Application.Queues.Provider {
    let url = Environment.get("REDIS_URL") ?? "redis://127.0.0.1:6379"
    let configuration = try RedisConfiguration(
        url: url,
        pool: .init(
            maximumConnectionCount: .maximumActiveConnections(8),
            minimumConnectionCount: 1,
            connectionRetryTimeout: .seconds(5)))
    return .redis(configuration)
}

import Foundation
import Queues
import Vapor

struct WelcomeEmail: Codable, Sendable {
    let memberID: UUID
    let email: String
    let name: String
}

struct WelcomeEmailJob: AsyncJob {
    typealias Payload = WelcomeEmail

    /// A stable name: renaming the Swift type must not orphan jobs already waiting in Redis.
    static var name: String { "welcome-email" }

    /// Seconds before the first retry; each further retry doubles it (capped at 15 minutes).
    var retryBaseSeconds = 30

    func dequeue(_ context: QueueContext, _ payload: WelcomeEmail) async throws {
        let key = "welcome-\(payload.memberID)"
        let ledger = context.application.sentLedger
        guard await !ledger.contains(key) else {
            context.logger.info("Welcome email already sent, skipping", metadata: ["key": "\(key)"])
            return
        }
        try await context.application.emailSender.send(OutgoingEmail(
            to: payload.email,
            subject: "Welcome to Bookshelf, \(payload.name)",
            body: "Your account is ready.",
            idempotencyKey: key))
        await ledger.record(key)
    }

    func error(_ context: QueueContext, _ error: any Error, _ payload: WelcomeEmail) async throws {
        // Called once, after the last attempt failed. Park it where a human or a repair job will look.
        context.logger.error("Welcome email gave up", metadata: [
            "member": "\(payload.memberID)", "error": "\(error)",
        ])
    }

    func nextRetryIn(attempt: Int) -> Int {
        guard retryBaseSeconds > 0 else { return 0 }
        return min(retryBaseSeconds << (attempt - 1), 15 * 60)
    }
}

struct LoanReminder: Codable, Sendable {
    let loanID: UUID
    let email: String
    let title: String
}

struct LoanReminderJob: AsyncJob {
    typealias Payload = LoanReminder

    func dequeue(_ context: QueueContext, _ payload: LoanReminder) async throws {
        try await context.application.emailSender.send(OutgoingEmail(
            to: payload.email,
            subject: "\"\(payload.title)\" is due tomorrow",
            body: "Please return it or renew the loan.",
            idempotencyKey: "reminder-\(payload.loanID)"))
    }
}

extension QueueName {
    /// Everything that sends mail: its own queue, so a backlog never delays the others.
    static let emails = QueueName(string: "emails")
    /// Reminders are processed strictly one at a time.
    static let reminders = QueueName(string: "reminders", workerCount: 1)
}

error: type 'ResizeJob.Payload' (aka 'ResizePayload') does not conform to the 'Sendable' protocol
error: type 'ResizePayload' does not conform to protocol 'Decodable'
error: type 'PlainJob' does not conform to protocol 'Job'

func dispatch<J: Job>(
    _ job: J.Type,
    _ payload: J.Payload,
    maxRetryCount: Int = 0,
    delayUntil: Date? = nil,
    id: JobIdentifier = .init()
) async throws

import Queues
import Vapor

struct RegisterMember: Content {
    let email: String
    let name: String
}

struct LoanRequest: Content {
    let email: String
    let title: String
    let dueDate: Date
}

func routes(_ app: Application) throws {
    app.post("members") { req async throws -> HTTPStatus in
        let input = try req.content.decode(RegisterMember.self)
        let memberID = UUID()
        // ... save the member with Fluent first; dispatch only after the row is committed.
        try await req.queues(.emails).dispatch(
            WelcomeEmailJob.self,
            WelcomeEmail(memberID: memberID, email: input.email, name: input.name),
            maxRetryCount: 5,
            id: JobIdentifier(string: "welcome-\(memberID)"))
        return .accepted
    }

    app.post("loans") { req async throws -> HTTPStatus in
        let input = try req.content.decode(LoanRequest.self)
        let reminderAt = input.dueDate.addingTimeInterval(-24 * 60 * 60)
        try await req.queues(.reminders).dispatch(
            LoanReminderJob.self,
            LoanReminder(loanID: UUID(), email: input.email, title: input.title),
            maxRetryCount: 3,
            delayUntil: reminderAt)
        return .accepted
    }
}

extension QueueName {
    /// Everything that sends mail: its own queue, so a backlog never delays the others.
    static let emails = QueueName(string: "emails")
    /// Reminders are processed strictly one at a time.
    static let reminders = QueueName(string: "reminders", workerCount: 1)
}

import Queues
import Vapor

/// Counters kept by the delegate (an actor, because delegates are called from several event loops).
actor JobStats {
    private(set) var dispatched = 0
    private(set) var dequeued = 0
    private(set) var succeeded = 0
    private(set) var failed = 0

    func didDispatch() { dispatched += 1 }
    func didDequeue() { dequeued += 1 }
    func didSucceed() { succeeded += 1 }
    func didFail() { failed += 1 }
}

struct JobAuditDelegate: AsyncJobEventDelegate {
    let logger: Logger
    let stats: JobStats

    func dispatched(job: JobEventData) async throws {
        await stats.didDispatch()
        logger.info("job dispatched", metadata: [
            "job": "\(job.jobName)", "id": "\(job.id)", "queue": "\(job.queueName)",
        ])
    }

    func didDequeue(jobId: String) async throws {
        await stats.didDequeue()
    }

    func success(jobId: String) async throws {
        await stats.didSucceed()
        logger.info("job succeeded", metadata: ["id": "\(jobId)"])
    }

    func error(jobId: String, error: any Error) async throws {
        await stats.didFail()
        logger.error("job failed for good", metadata: ["id": "\(jobId)", "error": "\(error)"])
    }
}

import Foundation
import Queues
import Testing
import VaporTesting
import XCTQueues
@testable import App

/// Fake provider: fails the first `failures` calls, records the rest.
actor RecordingEmailSender: EmailSender {
    private(set) var sent: [OutgoingEmail] = []
    private(set) var attempts = 0
    private var failuresLeft: Int

    struct Outage: Error {}

    init(failures: Int = 0) { failuresLeft = failures }

    func send(_ message: OutgoingEmail) async throws {
        attempts += 1
        if failuresLeft > 0 {
            failuresLeft -= 1
            throw Outage()
        }
        sent.append(message)
    }
}

func withApp(
    sender: RecordingEmailSender = RecordingEmailSender(),
    retryBaseSeconds: Int = 0,
    _ body: (Application) async throws -> Void
) async throws {
    let app = try await Application.make(.testing)
    do {
        app.queues.use(.asyncTest)
        app.emailSender = sender
        try await configure(app, retryBaseSeconds: retryBaseSeconds)
        try await body(app)
    } catch {
        try await app.asyncShutdown()
        throw error
    }
    try await app.asyncShutdown()
}

@Suite struct QueueTests {
    @Test func registrationDispatchesAJobWithoutSendingMail() async throws {
        let sender = RecordingEmailSender()
        try await withApp(sender: sender) { app in
            try await app.testing().test(.POST, "members", beforeRequest: { req in
                try req.content.encode(RegisterMember(email: "ada@example.com", name: "Ada"))
            }, afterResponse: { res async in
                #expect(res.status == .accepted)
            })

            // The handler only enqueued the job.
            #expect(app.queues.asyncTest.contains(WelcomeEmailJob.self))
            let payload = try #require(app.queues.asyncTest.first(WelcomeEmailJob.self))
            #expect(payload.email == "ada@example.com")
            #expect(await sender.sent.isEmpty)

            // Drain the queue the way a worker does.
            try await app.queues.queue.worker.run()
            #expect(await sender.sent.count == 1)
            #expect(app.queues.asyncTest.queue.isEmpty)
        }
    }

    @Test func aTransientFailureIsRetried() async throws {
        let sender = RecordingEmailSender(failures: 2)
        try await withApp(sender: sender) { app in
            let payload = WelcomeEmail(memberID: UUID(), email: "ada@example.com", name: "Ada")
            try await app.queues.queue.dispatch(WelcomeEmailJob.self, payload, maxRetryCount: 3)
            try await app.queues.queue.worker.run()

            #expect(await sender.attempts == 3)
            #expect(await sender.sent.count == 1)
            #expect(await app.jobStats.dequeued == 3)
            #expect(await app.jobStats.succeeded == 1)
            #expect(await app.jobStats.failed == 0)
        }
    }

    @Test func delayedJobsAtTheFrontBlockReadyJobs() async throws {
        let sender = RecordingEmailSender()
        try await withApp(sender: sender) { app in
            let later = Date(timeIntervalSinceNow: 3600)
            for n in 1...3 {
                try await app.queues.queue.dispatch(
                    LoanReminderJob.self,
                    LoanReminder(loanID: UUID(), email: "d\(n)@example.com", title: "Later"),
                    delayUntil: later)
            }
            try await app.queues.queue.dispatch(
                LoanReminderJob.self,
                LoanReminder(loanID: UUID(), email: "ready@example.com", title: "Now"))

            var passes = 0
            while await sender.sent.isEmpty {
                passes += 1
                try await app.queues.queue.worker.run()
            }
            print("the ready job ran on pass \(passes)")
            #expect(passes == 4)
        }
    }
}

✔ Test backoffParksTheJobUntilItsRetryTime() passed
✔ Test aRepeatedJobDoesNotSendTwice() passed
✔ Test dispatchingTwiceWithTheSameIdIsNotDeduplicated() passed
✔ Test delayedJobsAtTheFrontBlockReadyJobs() passed
✔ Test aTransientFailureIsRetried() passed
✔ Test maxRetryCountOfTwoMeansThreeAttemptsThenError() passed
✔ Test registrationDispatchesAJobWithoutSendingMail() passed
✔ Test anInProcessWorkerDrainsTheQueue() passed
✔ Test run with 8 tests in 1 suite passed after 0.064 seconds.

x-shared_environment: &shared_environment
  LOG_LEVEL: ${LOG_LEVEL:-info}
  REDIS_URL: redis://redis:6379

services:
  app:
    image: bookshelf:latest
    environment:
      <<: *shared_environment
    ports:
      - '8080:8080'
    command: ["serve", "--env", "production", "--hostname", "0.0.0.0", "--port", "8080"]
  worker-emails:
    image: bookshelf:latest
    environment:
      <<: *shared_environment
    command: ["queues", "--queue", "emails"]
  worker-reminders:
    image: bookshelf:latest
    environment:
      <<: *shared_environment
    command: ["queues", "--queue", "reminders"]
  redis:
    image: redis:7
    command: ["redis-server", "--appendonly", "yes"]