Goravel queues and jobs: retries and failed jobs

Part 16 of the Goravel series. The original on DevChalk has the colour-matched code, the figures and the full diffs. Code for this part: krishankumar01/goravel-taskflow/tree/part-16.

This part adds a CSV export of all tasks. A big export takes a while, so the request doesn't build it. It only queues a job and returns. The job builds the file in the background, and the client downloads it when it's done.

Start from the Events & Listeners code:

git clone https://github.com/krishankumar01/goravel-taskflow
cd goravel-taskflow
git checkout part-15
cp .env.example .env
go run . artisan key:generate
go run . artisan jwt:secret
docker compose up -d
go run . artisan migrate:fresh --seed
docker compose exec postgres psql -U goravel -d taskflow -c "CREATE DATABASE taskflow_test"
go install github.com/air-verse/air@latest
air

A job, not an event

An event announces what happened, and whoever listens decides what to do about it. That fit Events & Listeners, where the notification was one reaction to assigning a task, and more could be added later without touching the handler. For an export, the handler already knows the one thing that has to happen, and the export isn't done until the file is built. So it names that work directly by sending a job. The job runs on the same queue and worker as the queued listener.

The exports table

Each export is a row with its owner, its status and the file's path. The path stays out of the JSON:

app/enums/export_status.go

package enums

type ExportStatus string

const (
    ExportPending ExportStatus = "pending"
    ExportDone    ExportStatus = "done"
)

app/models/export.go

package models

import (
    "github.com/goravel/framework/database/orm"

    "taskflow/app/enums"
)

type Export struct {
    orm.Model
    UserID uint               `gorm:"not null" json:"user_id"`
    Status enums.ExportStatus `gorm:"type:enum('pending','done');default:'pending';not null" json:"status"`
    Path   string             `gorm:"size:255" json:"-"`
}

make:migration builds the migration from a model, but it only finds models listed in Models in bootstrap/app.go, so Export goes there first:

bootstrap/app.go

 // ... 42 lines not shown
        WithCallback(func() {
            facades.Schema().Extend(schema.Extension{
-               Models: []any{models.User{}, models.Task{}, models.Attachment{}, models.Notification{}},
+               Models: []any{models.User{}, models.Task{}, models.Attachment{}, models.Notification{}, models.Export{}},
            })

            // ... 34 lines not shown

$ go run . artisan make:migration create_exports_table -m Export

make:migration builds the columns from the model, and one more line adds the foreign key to users:

database/migrations/20261026231209_create_exports_table.go

// ... 16 lines not shown
func (r *M20261026231209CreateExportsTable) Up() error {
    if !facades.Schema().HasTable("exports") {
        return facades.Schema().Create("exports", func(table schema.Blueprint) {
            table.TimestampTz("created_at").Nullable()
            table.TimestampTz("updated_at").Nullable()
            table.BigIncrements("id")
            table.UnsignedBigInteger("user_id")
            table.Enum("status", []any{"pending", "done"}).Default("pending")
            table.String("path", 255)

            table.Foreign("user_id").References("id").On("users")
        })
    }
// ... 8 lines not shown

$ go run . artisan migrate

The job

make:job creates the job and registers it:

$ go run . artisan make:job ExportTasks
 SUCCESS  Job created successfully
 SUCCESS  Job registered successfully

WithEvents registers each listener with its event. A job has no event, so make:job registers it in a new bootstrap/jobs.go:

bootstrap/jobs.go

package bootstrap

import (
    "github.com/goravel/framework/contracts/queue"

    "taskflow/app/jobs"
)

func Jobs() []queue.Job {
    return []queue.Job{
        &jobs.ExportTasks{},
    }
}

It also adds WithJobs to bootstrap/app.go. Without it, the worker can't find the job:

bootstrap/app.go

 // ... 26 lines not shown
 func Boot() contractsfoundation.Application {
    return foundation.Setup().
+       WithJobs(Jobs).
        WithRules(Rules).
        WithSeeders(Seeders).
        WithMigrations(Migrations).
        // ... 50 lines not shown

The job takes the export's id, writes the CSV to the disk from File Storage and marks the export done:

app/jobs/export_tasks.go

// ... 16 lines not shown

type ExportTasks struct {
}

func NewExportTasks(exportID uint) (*ExportTasks, []queue.Arg) {
    return &ExportTasks{}, []queue.Arg{
        {Type: "uint", Value: exportID},
    }
}

// Signature The name and signature of the job.
func (receiver *ExportTasks) Signature() string {
    return "export_tasks"
}

// Handle Execute the job.
func (receiver *ExportTasks) Handle(args ...any) error {
    exportID, ok := args[0].(uint)
    if !ok {
        return errors.New("export tasks: the argument must be the export id")
    }

    var export models.Export
    err := facades.Orm().Query().FindOrFail(&export, exportID)
    if errors.Is(err, frameworkerrors.OrmRecordNotFound) {
        return nil
    }
    if err != nil {
        return err
    }

    var tasks []models.Task
    if err := facades.Orm().Query().With("Assignee").OrderBy("id").Find(&tasks); err != nil {
        return err
    }

    rows := [][]string{{"id", "title", "status", "assignee", "due_at"}}
    for _, task := range tasks {
        assignee, dueAt := "", ""
        if task.Assignee != nil {
            assignee = task.Assignee.Name
        }
        if task.DueAt != nil {
            dueAt = task.DueAt.ToDateTimeString()
        }
        rows = append(rows, []string{strconv.Itoa(int(task.ID)), task.Title, string(task.Status), assignee, dueAt})
    }

    var file bytes.Buffer
    if err := csv.NewWriter(&file).WriteAll(rows); err != nil {
        return err
    }

    path := fmt.Sprintf("exports/%d.csv", export.ID)
    if err := facades.Storage().Put(path, file.String()); err != nil {
        return err
    }

    export.Status = enums.ExportDone
    export.Path = path

    return facades.Orm().Query().Save(&export)
}

• NewExportTasks builds the job and its args together, as NewTaskAssigned does for the event in Events & Listeners. Every call then passes the export's id, and always as a uint.
• Signature is the job's name, export_tasks. The worker finds the job by it and prints it.
• Handle reads the export's id from its args and checks its type. Args have to be plain values, as with the queued listener, so the job gets the id and loads the export itself. If the export is gone, an error would only put a job that can never succeed in failed_jobs, so it returns nil.
• It writes one CSV row per task to exports/<id>.csv and marks the export done.
• Returning an error tells the worker the job failed.

Starting an export

POST /exports saves a pending export and sends the job. It then answers 202 Accepted, which means the request was accepted but the work isn't finished. The client reads the status from GET /exports/{id}, and once it's done, GET /exports/{id}/download sends the file:

app/http/controllers/export_controller.go

// ... 28 lines not shown
    export := models.Export{
        UserID: authuser.From(ctx).ID,
        Status: enums.ExportPending,
    }
    if err := facades.Orm().WithContext(ctx).Query().Create(&export); err != nil {
        facades.Log().Error(err)
        return ctx.Response().Json(500, map[string]any{
            "error": "Could not start the export.",
        })
    }

    err := facades.Queue().Job(jobs.NewExportTasks(export.ID)).Dispatch()
    if err != nil {
        facades.Log().Error(err)
        return ctx.Response().Json(500, map[string]any{
            "error": "Could not start the export.",
        })
    }

    return ctx.Response().Json(202, export)
}

// ... 20 lines not shown
func (r *ExportController) Download(ctx http.Context) http.Response {
    var export models.Export
    // ... 14 lines not shown
    if export.Status != enums.ExportDone {
        return ctx.Response().Json(409, map[string]any{
            "error": "The export is not ready yet.",
        })
    }

    content, err := facades.Storage().GetBytes(export.Path)
    if err != nil {
        facades.Log().Error(err)
        return ctx.Response().Json(500, map[string]any{
            "error": "Could not load the export.",
        })
    }

    disposition := mime.FormatMediaType("attachment", map[string]string{"filename": fmt.Sprintf("tasks-%d.csv", export.ID)})

    return ctx.Response().
        Header("Content-Disposition", disposition).
        Data(200, "text/csv", content)
}

• NewExportTasks gives Job the job and its one arg, the export's id, and Dispatch puts the job on the queue.
• Show and Download only look among the user's own exports, so an export id that belongs to someone else gives 404.
• Download answers 409 Conflict until the export is done.

routes/api.go

 // ... 15 lines not shown
    attachmentController := controllers.NewAttachmentController()
    notificationController := controllers.NewNotificationController()
+   exportController := controllers.NewExportController()

    facades.Route().Prefix("api/v1").Group(func(router route.Router) {
        router.Post("register", authController.Register)
        // ... 18 lines not shown
            router.Post("notifications/{id}/read", notificationController.Read)

+           router.Post("exports", exportController.Store)
+           router.Get("exports/{id}", exportController.Show)
+           router.Get("exports/{id}/download", exportController.Download)
+
            router.Get("users", userController.Index)
            router.Get("users/{id}", userController.Show)
        })
    })
 }

See it work

The run starts from the same setup as Gates & Policies:

1. Reset the database with migrate:fresh --seed, which adds Ana as the manager, plus three members and seven tasks.
2. Delete storage/app/exports, so files from an earlier run can't stop the failure later on.
3. Register Ben. He gets id 5.
4. Log in as Ana and as Ben, and save each token in a variable, ANA and BEN.

The export is pending when the request returns, and done a moment later:

$ curl -X POST http://localhost:3000/api/v1/exports -H "Authorization: Bearer $ANA"
{…,"id":1,"user_id":1,"status":"pending"}

$ sleep 2
$ curl http://localhost:3000/api/v1/exports/1 -H "Authorization: Bearer $ANA"
{…,"id":1,"user_id":1,"status":"done"}

$ curl -D - http://localhost:3000/api/v1/exports/1/download -H "Authorization: Bearer $ANA"
HTTP/1.1 200 OK
Content-Disposition: attachment; filename=tasks-1.csv
Content-Length: 466
Content-Type: text/csv
Date: …

id,title,status,assignee,due_at
1,Review the pull request,done,Francesco Bradley,2026-11-04 09:00:00
2,Review the pull request,pending,Francesco Bradley,2026-11-25 09:00:00
…

$ curl http://localhost:3000/api/v1/exports/1 -H "Authorization: Bearer $BEN"
{"error":"The export does not exist."}

The seeder makes up names and dates, so the CSV rows differ after each reset.

The worker runs inside the app, so its lines show up next to the server's own output. It prints one line when a job starts and one when it ends, each with the job's signature:

  2026-10-27 00:09:53.215 export_tasks ............................... RUNNING
  2026-10-27 00:09:53.222 export_tasks .......................... 6.121ms DONE

When the job fails

The job can't create a file in a read-only folder, so chmod 555 on storage/app/exports makes it fail. POST /exports still answers 202, because it only queues the job, but the export stays pending and Download refuses the file:

$ chmod 555 storage/app/exports

$ curl -X POST http://localhost:3000/api/v1/exports -H "Authorization: Bearer $ANA"
{…,"id":2,"user_id":1,"status":"pending"}

$ curl http://localhost:3000/api/v1/exports/2 -H "Authorization: Bearer $ANA"
{…,"id":2,"user_id":1,"status":"pending"}

$ curl http://localhost:3000/api/v1/exports/2/download -H "Authorization: Bearer $ANA"
{"error":"The export is not ready yet."}

The worker tries the job once and gives up:

  2026-10-27 00:09:55.231 export_tasks ............................... RUNNING
  2026-10-27 00:09:55.235 export_tasks .......................... 3.839ms FAIL

The worker saves a failed job in the failed_jobs table, with its args and the error. queue:failed lists those jobs but not the error, so psql reads it from the table:

$ go run . artisan queue:failed
  2026-10-27 00:09:57 00950ba2-2c5b-43dd-9cd8-d757ce5d198b ..... redis@default

$ docker compose exec postgres psql -U goravel taskflow -At -c "select exception from failed_jobs"
open …/storage/app/exports/2.csv: permission denied

Nothing marks the export as failed, so it stays pending. Once the folder is writable again, queue:retry takes the job's uuid, the one queue:failed listed, and puts that job back on the queue. The worker then runs it and finishes the export:

$ chmod 755 storage/app/exports

$ go run . artisan queue:retry 00950ba2-2c5b-43dd-9cd8-d757ce5d198b
  INFO    Pushing failed queue jobs back onto the queue

  00950ba2-2c5b-43dd-9cd8-d757ce5d198b ........................... 2.91ms DONE

$ curl http://localhost:3000/api/v1/exports/2 -H "Authorization: Bearer $ANA"
{…,"id":2,"user_id":1,"status":"done"}

queue:retry also takes several uuids at once, and with none it retries every failed job.

Trying again by itself

Some failures go away after a few seconds, like a database that's restarting, so ShouldRetry lets the job try again:

app/jobs/export_tasks.go

 package jobs

 import (
    "bytes"
    "encoding/csv"
    "errors"
    "fmt"
    "strconv"
+   "time"

    "github.com/goravel/framework/contracts/queue"
    // ... 67 lines not shown
    return facades.Orm().Query().Save(&export)
 }
+
+// ShouldRetry gives a failed export two more tries, five seconds apart.
+func (receiver *ExportTasks) ShouldRetry(err error, attempt int) (bool, time.Duration) {
+   return attempt < 3, 5 * time.Second
+}

After each failed try, the worker calls ShouldRetry with the error and the try's number, starting at 1. ShouldRetry returns whether to try again and how long to wait. Here attempt < 3, 5 * time.Second means another try after tries 1 and 2, each five seconds later. After the third failed try, the worker saves the job in failed_jobs.

The worker runs one job at a time, and it sleeps through each five-second wait itself. So a job queued behind a failing export has to wait for all three tries. The block below makes the folder read-only again, starts an export, then assigns a task, which queues a notification:

$ chmod 555 storage/app/exports

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:3000/api/v1/exports -H "Authorization: Bearer $ANA"
202

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:3000/api/v1/tasks \
  -H "Authorization: Bearer $ANA" \
  -H "Content-Type: application/json" \
  -d '{"title":"Fix the login page","description":"The error message overlaps the button.","assigned_to":5}'
201

$ sleep 2
$ curl http://localhost:3000/api/v1/notifications -H "Authorization: Bearer $BEN"
{"data":[],"meta":{"page":1,"per_page":15,"total":0}}

$ sleep 10
$ curl http://localhost:3000/api/v1/notifications -H "Authorization: Bearer $BEN"
{"data":[{…,"id":1,"user_id":5,"task_id":8,"message":"Ana assigned you \"Fix the login page\"","read_at":null}],"meta":{"page":1,"per_page":15,"total":1}}

The notification only ran after the export gave up, as the worker's log shows:

  2026-10-27 00:10:06.816 export_tasks ............................... RUNNING
  2026-10-27 00:10:16.839 export_tasks .......................... 6.905ms FAIL
  2026-10-27 00:10:16.842 notify_assignee ............................ RUNNING
  2026-10-27 00:10:16.847 notify_assignee ........................ 4.77ms DONE

The figure steps through both cases, an export that works and this one:

Interactive figure on DevChalk: An export that works, then one that fails three times while a notification waits behind it.

An export that works: the client sends POST /exports, the API saves a pending export in Postgres, puts an export_tasks job on Redis and answers 202. The worker takes the job, loads the tasks, writes the CSV to the disk and marks the export done. An export that fails: the API answers 202 as before, but the worker can't write the file, waits five seconds and tries again, three times, then saves the job in failed_jobs. The notification job waits on Redis the whole time and runs only after the export gives up.

Two jobs at once

concurrent is a key on a connection in config/queue.go. It sets how many jobs the worker runs at the same time, and it defaults to 1. The redis connection doesn't set it, so it runs one job at a time until it's set to 2:

config/queue.go

 // ... 21 lines not shown
            "database": map[string]any{
                "driver":     "database",
                "connection": config.Env("DB_CONNECTION"),
                "queue":      "default",
                "concurrent": 1,
            },
            "redis": map[string]any{
                "driver":     "custom",
                "connection": "default",
                "queue":      "default",
+               "concurrent": 2,
                "via": func() (queue.Driver, error) {
                    return redisfacades.Queue("redis") // The `redis` value is the key of `connections`
                },
            // ... 13 lines not shown

With the same two requests, the notification is there after 2 seconds. In the log, it starts at once, while the export is still retrying:

  2026-10-27 00:10:25.057 export_tasks ............................... RUNNING
  2026-10-27 00:10:25.057 notify_assignee ............................ RUNNING
  2026-10-27 00:10:25.074 notify_assignee ...................... 17.074ms DONE
  2026-10-27 00:10:35.08 export_tasks ........................... 6.471ms FAIL

Make the folder writable again with chmod 755 storage/app/exports.

A queue of its own for exports:

Exports could also go on a queue of their own with OnQueue:

facades.Queue().Job(jobs.NewExportTasks(export.ID)).OnQueue("exports").Dispatch()

The worker that app.Start() runs reads one queue, the one named by the default connection's queue key. That's default in TaskFlow, as the start-up line says: Processing jobs from [redis] connection and [default] queue. A job on exports needs a second worker for that queue, which a runner added in WithRunners can start. With its own worker, a slow export never holds up the jobs on default.

Sending a job later with Delay:

Delay sends a job later:

facades.Queue().Job(jobs.NewExportTasks(export.ID)).Delay(time.Now().Add(time.Hour)).Dispatch()

On redis, the job waits until then. On sync, which the tests use, Dispatch itself sleeps until then, so a test with that delay would take an hour. An export should start as soon as it's asked for, so TaskFlow doesn't use Delay.

Tests

The tests are in tests/feature/export_test.go. They use the sync connection from Events & Listeners, so the job runs inside POST /exports. They check that:

• the export is done
• the download is the CSV
• other users get 404
• a pending export can't be downloaded

$ go test ./tests/... --env=.env.testing
?       taskflow/tests  [no test files]
ok      taskflow/tests/feature  1.542s
ok      taskflow/tests/unit 0.398s

Exports now run in the background and retry when they fail. An export that fails every try still shows as pending.

---

Previous: 15. Events & Listeners
Next: 17. Mail & Templates
Series: Goravel