perf(task): offload completion notification fan-out to Asynq worker
POST /api/task-completions/ was spending ~1.5-1.75s synchronously on
APNs push + SMTP email + B2 image fetches inside sendTaskCompletedNotification.
Per-user loop made it scale linearly with residence membership; one image
attached + one residence user is the 1.75s baseline observed in the live
honeydue-eli5-overview Grafana panel.
Replace the inline call (and the fire-and-forget goroutine in QuickComplete,
which violated the project's "no goroutines in handlers" rule) with an
Asynq job:
- new task type notification:task_completed (worker/scheduler.go)
- new payload {task_id, completion_id} — IDs only, worker re-reads
canonical state from Postgres so concurrent edits between enqueue
and dequeue are reflected
- new HandleTaskCompletedNotification on jobs.Handler delegates to
TaskService.SendTaskCompletedNotificationByID
- new dispatchTaskCompletedNotification in task_service.go picks
between enqueue (preferred) and inline (fallback) when Redis is
unreachable or the enqueuer isn't wired (tests / local dev)
Other changes required to wire it up:
- widen worker.NewTaskClient signature to accept asynq.RedisClientOpt
so the file-mounted Redis password (audit HIGH-1) can be supplied;
no prior callers, no breakage
- extend worker.Enqueuer interface with EnqueueTaskCompletedNotification
- add TaskEnqueuer field to router.Dependencies; wire from cmd/api/main.go
with the standard typed-nil interface guard
- wire a worker-side TaskService in cmd/worker/main.go so the handler
can use the shared SendTaskCompletedNotificationByID implementation
(storage service shared with the existing upload-cleanup wiring)
Expected impact on POST /api/task-completions/ p50:
~1.75s -> ~120-170ms (DB + tx + Asynq enqueue only)
Notifications still deliver; they just go via the worker instead of in
the request path. MaxRetry=3; "row not found" returns nil so a deleted
task/completion doesn't churn the retry loop.
All 31 test packages pass. No DB migrations.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -29,6 +29,16 @@ var (
|
||||
ErrCompletionNotFound = errors.New("task completion not found")
|
||||
)
|
||||
|
||||
// TaskCompletedNotificationEnqueuer is the narrow contract TaskService needs
|
||||
// from the Asynq client to offload completion-notification fan-out (push +
|
||||
// email + B2 image fetches) into the background worker. Implemented by
|
||||
// internal/worker.TaskClient; nil when the api is started without a worker
|
||||
// queue (tests / local dev), in which case CreateCompletion falls back to
|
||||
// the inline synchronous path.
|
||||
type TaskCompletedNotificationEnqueuer interface {
|
||||
EnqueueTaskCompletedNotification(taskID, completionID uint) error
|
||||
}
|
||||
|
||||
// TaskService handles task business logic
|
||||
type TaskService struct {
|
||||
taskRepo *repositories.TaskRepository
|
||||
@@ -39,6 +49,7 @@ type TaskService struct {
|
||||
storageService *StorageService
|
||||
uploadService *UploadService // optional — only set when S3 storage is configured
|
||||
cache *CacheService
|
||||
notifEnqueuer TaskCompletedNotificationEnqueuer
|
||||
}
|
||||
|
||||
// SetUploadService wires the presigned-URL upload service so CreateCompletion
|
||||
@@ -82,6 +93,14 @@ func (s *TaskService) SetStorageService(ss *StorageService) {
|
||||
s.storageService = ss
|
||||
}
|
||||
|
||||
// SetTaskCompletedNotificationEnqueuer wires the Asynq enqueuer. When set,
|
||||
// CreateCompletion/QuickComplete schedule the notification fan-out as a
|
||||
// background job instead of running it inline (saving ~1-1.5s on the
|
||||
// user-facing response). When nil, the inline path runs.
|
||||
func (s *TaskService) SetTaskCompletedNotificationEnqueuer(e TaskCompletedNotificationEnqueuer) {
|
||||
s.notifEnqueuer = e
|
||||
}
|
||||
|
||||
// getSummaryForUser returns an empty summary placeholder.
|
||||
// DEPRECATED: Summary calculation has been removed from CRUD responses for performance.
|
||||
// Clients should calculate summary from kanban data instead (which already includes all tasks).
|
||||
@@ -769,8 +788,11 @@ func (s *TaskService) CreateCompletion(ctx context.Context, req *requests.Create
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Send notification to residence owner and other users
|
||||
s.sendTaskCompletedNotification(ctx, task, completion)
|
||||
// Dispatch notification fan-out to the Asynq worker so the api response
|
||||
// returns without waiting on APNs + SMTP + B2 image fetches (which cost
|
||||
// ~1-1.5s end-to-end). Falls back to the inline path when no enqueuer is
|
||||
// wired (tests / local dev) or when Redis is unreachable.
|
||||
s.dispatchTaskCompletedNotification(ctx, task, completion)
|
||||
|
||||
// Return completion with updated task (includes kanban_column for UI update)
|
||||
resp := responses.NewTaskCompletionWithTaskResponseWithTime(completion, task, 30, now)
|
||||
@@ -876,19 +898,71 @@ func (s *TaskService) QuickComplete(ctx context.Context, taskID uint, userID uin
|
||||
}
|
||||
log.Info().Uint("task_id", task.ID).Msg("QuickComplete: Task updated successfully")
|
||||
|
||||
// Send notification (fire and forget with panic recovery)
|
||||
go func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Error().Interface("panic", r).Uint("task_id", task.ID).Msg("Panic in quick-complete notification goroutine")
|
||||
}
|
||||
}()
|
||||
s.sendTaskCompletedNotification(ctx, task, completion)
|
||||
}()
|
||||
// Dispatch notification fan-out to the Asynq worker. Replaces the previous
|
||||
// fire-and-forget goroutine — which violated the project rule against
|
||||
// spawning goroutines in request handlers and was unbounded under load.
|
||||
// Falls back to inline send when no enqueuer is wired.
|
||||
s.dispatchTaskCompletedNotification(ctx, task, completion)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// dispatchTaskCompletedNotification routes the notification fan-out to either
|
||||
// the Asynq worker (preferred — keeps the api request path fast) or runs it
|
||||
// inline as a fallback. The fallback covers:
|
||||
// - tests / local dev where no enqueuer is wired (notifEnqueuer == nil)
|
||||
// - Redis outages where Enqueue returns an error
|
||||
//
|
||||
// Inline fallback is intentionally synchronous (no goroutines) per the
|
||||
// project's rule against unbounded goroutine spawning in handlers. The
|
||||
// caller is expected to be in a request goroutine and accepting the cost.
|
||||
func (s *TaskService) dispatchTaskCompletedNotification(ctx context.Context, task *models.Task, completion *models.TaskCompletion) {
|
||||
if s.notifEnqueuer != nil {
|
||||
if err := s.notifEnqueuer.EnqueueTaskCompletedNotification(task.ID, completion.ID); err == nil {
|
||||
return
|
||||
} else {
|
||||
log.Warn().
|
||||
Err(err).
|
||||
Uint("task_id", task.ID).
|
||||
Uint("completion_id", completion.ID).
|
||||
Msg("Failed to enqueue completion notification; falling back to inline send")
|
||||
}
|
||||
}
|
||||
s.sendTaskCompletedNotification(ctx, task, completion)
|
||||
}
|
||||
|
||||
// SendTaskCompletedNotificationByID is the public entry point used by the
|
||||
// Asynq worker. It re-reads the canonical Task + TaskCompletion rows from
|
||||
// Postgres (cheap with Neon ~10ms away) so the worker reflects any concurrent
|
||||
// edits between enqueue and dequeue, then delegates to the shared
|
||||
// sendTaskCompletedNotification implementation.
|
||||
//
|
||||
// Returns nil for "row not found" cases (task or completion was deleted before
|
||||
// the job ran) so Asynq's retry loop doesn't churn on a permanent miss. All
|
||||
// other errors propagate to Asynq for retry per the queue's MaxRetry setting.
|
||||
func (s *TaskService) SendTaskCompletedNotificationByID(ctx context.Context, taskID, completionID uint) error {
|
||||
task, err := s.taskRepo.WithContext(ctx).FindByID(taskID)
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
log.Warn().Uint("task_id", taskID).Msg("task not found for completion notification (likely deleted between enqueue and dequeue)")
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
completion, err := s.taskRepo.WithContext(ctx).FindCompletionByID(completionID)
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
log.Warn().Uint("completion_id", completionID).Msg("completion not found for notification (likely deleted between enqueue and dequeue)")
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
s.sendTaskCompletedNotification(ctx, task, completion)
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendTaskCompletedNotification sends notifications when a task is completed
|
||||
func (s *TaskService) sendTaskCompletedNotification(ctx context.Context, task *models.Task, completion *models.TaskCompletion) {
|
||||
// Get all users with access to this residence
|
||||
|
||||
Reference in New Issue
Block a user