Goravel Tutorial #16: Queues and jobs, retries and failed jobs

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.

CodeStart from the Events & Listeners: Notifications 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
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: Notifications, 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:

package enums

type ExportStatus string

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

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:

42 collapsed lines
package bootstrap

import (
  "context"
  "strings"

  "github.com/goravel/framework/auth/access"
  contractsaccess "github.com/goravel/framework/contracts/auth/access"
  "github.com/goravel/framework/contracts/database/schema"
  contractsevent "github.com/goravel/framework/contracts/event"
  contractsfoundation "github.com/goravel/framework/contracts/foundation"
  "github.com/goravel/framework/contracts/http"
  "github.com/goravel/framework/foundation"
  "github.com/goravel/framework/http/limit"

  "taskflow/app/enums"
  "taskflow/app/events"
  "taskflow/app/facades"
  "taskflow/app/http/authuser"
  "taskflow/app/listeners"
  "taskflow/app/models"
  "taskflow/app/policies"
  "taskflow/config"
  "taskflow/routes"
)

func Boot() contractsfoundation.Application {
  return foundation.Setup().
    WithRules(Rules).
    WithSeeders(Seeders).
    WithMigrations(Migrations).
    WithRouting(func() {
      routes.Api()
      routes.Web()
    }).
    WithEvents(func() map[contractsevent.Event][]contractsevent.Listener {
      return map[contractsevent.Event][]contractsevent.Listener{
        &events.TaskAssigned{}: {&listeners.NotifyAssignee{}},
      }
    }).
    WithProviders(Providers).
    WithConfig(config.Boot).
    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 collapsed lines
      facades.RateLimiter().For("login", func(ctx http.Context) http.Limit {
        email := strings.ToLower(strings.TrimSpace(ctx.Request().Input("email")))

        return limit.PerMinute(5).By(email).Response(func(ctx http.Context) {
          ctx.Response().Json(429, map[string]any{
            "error": "Too many login attempts. Try again in a minute.",
          }).Abort()
        })
      })

      facades.Gate().Before(func(ctx context.Context, ability string, arguments map[string]any) contractsaccess.Response {
        if user := authuser.From(ctx); user != nil && user.Role == enums.RoleManager {
          return access.NewAllowResponse()
        }

        return nil
      })

      facades.Gate().Define(policies.ViewUsers, func(ctx context.Context, arguments map[string]any) contractsaccess.Response {
        return access.NewDenyResponse("Only managers can see the users.")
      })

      taskPolicy := policies.NewTaskPolicy()
      facades.Gate().Define(policies.CreateTask, taskPolicy.Create)
      facades.Gate().Define(policies.UpdateTask, taskPolicy.Update)
      facades.Gate().Define(policies.DeleteTask, taskPolicy.Delete)

      attachmentPolicy := policies.NewAttachmentPolicy()
      facades.Gate().Define(policies.CreateAttachment, attachmentPolicy.Create)
      facades.Gate().Define(policies.ViewAttachment, attachmentPolicy.View)
      facades.Gate().Define(policies.DeleteAttachment, attachmentPolicy.Delete)
    }).
    Create()
}

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:

16 collapsed lines
package migrations

import (
  "github.com/goravel/framework/contracts/database/schema"

  "taskflow/app/facades"
)

type M20261026231209CreateExportsTable struct{}

// Signature The unique signature for the migration.
func (r *M20261026231209CreateExportsTable) Signature() string {
  return "20261026231209_create_exports_table"
}

// Up Run the migrations.
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 collapsed lines

  return nil
}

// Down Reverse the migrations.
func (r *M20261026231209CreateExportsTable) Down() error {
  return facades.Schema().DropIfExists("exports")
}

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:

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:

26 collapsed lines
package bootstrap

import (
  "context"
  "strings"

  "github.com/goravel/framework/auth/access"
  contractsaccess "github.com/goravel/framework/contracts/auth/access"
  "github.com/goravel/framework/contracts/database/schema"
  contractsevent "github.com/goravel/framework/contracts/event"
  contractsfoundation "github.com/goravel/framework/contracts/foundation"
  "github.com/goravel/framework/contracts/http"
  "github.com/goravel/framework/foundation"
  "github.com/goravel/framework/http/limit"

  "taskflow/app/enums"
  "taskflow/app/events"
  "taskflow/app/facades"
  "taskflow/app/http/authuser"
  "taskflow/app/listeners"
  "taskflow/app/models"
  "taskflow/app/policies"
  "taskflow/config"
  "taskflow/routes"
)

func Boot() contractsfoundation.Application {
  return foundation.Setup().
    WithJobs(Jobs).
    WithRules(Rules).
    WithSeeders(Seeders).
    WithMigrations(Migrations).
50 collapsed lines
    WithRouting(func() {
      routes.Api()
      routes.Web()
    }).
    WithEvents(func() map[contractsevent.Event][]contractsevent.Listener {
      return map[contractsevent.Event][]contractsevent.Listener{
        &events.TaskAssigned{}: {&listeners.NotifyAssignee{}},
      }
    }).
    WithProviders(Providers).
    WithConfig(config.Boot).
    WithCallback(func() {
      facades.Schema().Extend(schema.Extension{
        Models: []any{models.User{}, models.Task{}, models.Attachment{}, models.Notification{}, models.Export{}},
      })

      facades.RateLimiter().For("login", func(ctx http.Context) http.Limit {
        email := strings.ToLower(strings.TrimSpace(ctx.Request().Input("email")))

        return limit.PerMinute(5).By(email).Response(func(ctx http.Context) {
          ctx.Response().Json(429, map[string]any{
            "error": "Too many login attempts. Try again in a minute.",
          }).Abort()
        })
      })

      facades.Gate().Before(func(ctx context.Context, ability string, arguments map[string]any) contractsaccess.Response {
        if user := authuser.From(ctx); user != nil && user.Role == enums.RoleManager {
          return access.NewAllowResponse()
        }

        return nil
      })

      facades.Gate().Define(policies.ViewUsers, func(ctx context.Context, arguments map[string]any) contractsaccess.Response {
        return access.NewDenyResponse("Only managers can see the users.")
      })

      taskPolicy := policies.NewTaskPolicy()
      facades.Gate().Define(policies.CreateTask, taskPolicy.Create)
      facades.Gate().Define(policies.UpdateTask, taskPolicy.Update)
      facades.Gate().Define(policies.DeleteTask, taskPolicy.Delete)

      attachmentPolicy := policies.NewAttachmentPolicy()
      facades.Gate().Define(policies.CreateAttachment, attachmentPolicy.Create)
      facades.Gate().Define(policies.ViewAttachment, attachmentPolicy.View)
      facades.Gate().Define(policies.DeleteAttachment, attachmentPolicy.Delete)
    }).
    Create()
}

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

16 collapsed lines
package jobs

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

  "github.com/goravel/framework/contracts/queue"
  frameworkerrors "github.com/goravel/framework/errors"

  "taskflow/app/enums"
  "taskflow/app/facades"
  "taskflow/app/models"
)

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: Notifications. 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:

28 collapsed lines
package controllers

import (
  "errors"
  "fmt"
  "mime"

  "github.com/goravel/framework/contracts/http"
  frameworkerrors "github.com/goravel/framework/errors"

  "taskflow/app/enums"
  "taskflow/app/facades"
  "taskflow/app/http/authuser"
  "taskflow/app/jobs"
  "taskflow/app/models"
)

type ExportController struct {
  // Dependent services
}

func NewExportController() *ExportController {
  return &ExportController{
    // Inject services
  }
}

func (r *ExportController) Store(ctx http.Context) http.Response {
  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 collapsed lines
func (r *ExportController) Show(ctx http.Context) http.Response {
  var export models.Export
  err := facades.Orm().WithContext(ctx).Query().
    Where("user_id", authuser.From(ctx).ID).
    FindOrFail(&export, ctx.Request().RouteInt("id"))
  if errors.Is(err, frameworkerrors.OrmRecordNotFound) {
    return ctx.Response().Json(404, map[string]any{
      "error": "The export does not exist.",
    })
  }
  if err != nil {
    facades.Log().Error(err)
    return ctx.Response().Json(500, map[string]any{
      "error": "Could not load the export.",
    })
  }

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

func (r *ExportController) Download(ctx http.Context) http.Response {
  var export models.Export
14 collapsed lines
  err := facades.Orm().WithContext(ctx).Query().
    Where("user_id", authuser.From(ctx).ID).
    FindOrFail(&export, ctx.Request().RouteInt("id"))
  if errors.Is(err, frameworkerrors.OrmRecordNotFound) {
    return ctx.Response().Json(404, map[string]any{
      "error": "The export does not exist.",
    })
  }
  if err != nil {
    facades.Log().Error(err)
    return ctx.Response().Json(500, map[string]any{
      "error": "Could not load the export.",
    })
  }
  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.

15 collapsed lines
package routes

import (
  "taskflow/app/facades"
  "taskflow/app/http/controllers"
  "taskflow/app/http/middleware"

  "github.com/goravel/framework/contracts/route"
  httpmiddleware "github.com/goravel/framework/http/middleware"
)

func Api() {
  authController := controllers.NewAuthController()
  taskController := controllers.NewTaskController()
  userController := controllers.NewUserController()
  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 collapsed lines
    router.Middleware(httpmiddleware.Throttle("login")).Post("login", authController.Login)
    router.Post("refresh", authController.Refresh)

    router.Middleware(&middleware.Auth{}).Group(func(router route.Router) {
      router.Get("me", authController.Me)
      router.Post("logout", authController.Logout)

      router.Post("tasks", taskController.Store).Name("tasks.create")
      router.Get("tasks/{id}", taskController.Show)
      router.Get("tasks", taskController.Index)
      router.Patch("tasks/{id}", taskController.Update)
      router.Delete("tasks/{id}", taskController.Destroy)

      router.Post("tasks/{id}/attachments", attachmentController.Store)
      router.Get("tasks/{id}/attachments/{attachment}", attachmentController.Show)
      router.Delete("tasks/{id}/attachments/{attachment}", attachmentController.Destroy)

      router.Get("notifications", notificationController.Index)
      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:

Reset the database with migrate:fresh --seed, which adds Ana as the manager, plus three members and seven tasks.
Delete storage/app/exports, so files from an earlier run can’t stop the failure later on.
Register Ben. He gets id 5.
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:

package jobs

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

  "github.com/goravel/framework/contracts/queue"
67 collapsed lines
  frameworkerrors "github.com/goravel/framework/errors"

  "taskflow/app/enums"
  "taskflow/app/facades"
  "taskflow/app/models"
)

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)
}

// 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:

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:

21 collapsed lines
package config

import (
  "github.com/goravel/framework/contracts/queue"
  redisfacades "github.com/goravel/redis/facades"
  "taskflow/app/facades"
)

func init() {
  config := facades.Config()
  config.Add("queue", map[string]any{
    "default": config.Env("QUEUE_CONNECTION", "redis"),

    // Queue Connections
    //
    // Here you may configure the connection information for each server that is used by your application.
    // Drivers: "sync", "database", "custom"
    "connections": map[string]any{
      "sync": map[string]any{
        "driver": "sync",
      },
      "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 collapsed lines
      },
    },

    // Failed Queue Jobs
    //
    // These options configure the behavior of failed queue job logging so you
    // can control how and where failed jobs are stored.
    "failed": map[string]any{
      "database": config.Env("DB_CONNECTION"),
      "table":    "failed_jobs",
    },
  })
}

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.

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

MoreSending 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: Notifications, 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.