Stage 7 · Master
Phase 14 — Notification Service
Background Workers
Persist every send request as a claimable job and let a small worker pool process it with FOR UPDATE SKIP LOCKED, so two running replicas never send the same notification twice.
Sending Inline Couples a Caller's Latency to a Provider's Latency
If Payment called notification-service and waited for the SMS provider's HTTP response before returning, a slow provider would make invoice payment itself feel slow — and a provider outage would make payment fail outright. notification-service instead accepts a send request, persists it, and returns immediately; delivery happens asynchronously.
CREATE TYPE job_status AS ENUM ('pending', 'processing', 'sent', 'failed', 'dead');
CREATE TABLE notification_jobs (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
organization_id uuid NOT NULL,
channel text NOT NULL,
payload jsonb NOT NULL,
status job_status NOT NULL DEFAULT 'pending',
attempts int NOT NULL DEFAULT 0,
max_attempts int NOT NULL DEFAULT 5,
next_attempt_at timestamptz NOT NULL DEFAULT now(),
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX idx_notification_jobs_claimable ON notification_jobs (status, next_attempt_at);next_attempt_at doubles as both the initial eligibility time and the backoff schedule after a failed attempt — a claim query only ever needs one condition, not a separate retry-time column.
FOR UPDATE SKIP LOCKED Is the Whole Safety Mechanism
Running two worker replicas for throughput creates an obvious race: both could read the same pending job before either commits a status change. SELECT ... FOR UPDATE SKIP LOCKED lets each worker lock a batch of rows no other transaction currently holds, and simply skips rows another worker already claimed — no external coordinator, distributed lock, or message broker required for this.
package postgresadapter
import (
"context"
"github.com/jackc/pgx/v5/pgxpool"
)
type ClaimedJob struct {
ID string
OrganizationID string
Channel string
Payload []byte
Attempts int
MaxAttempts int
}
type JobRepository struct {
pool *pgxpool.Pool
}
func NewJobRepository(pool *pgxpool.Pool) *JobRepository {
return &JobRepository{pool: pool}
}
// Claim locks up to batchSize eligible jobs for this transaction only.
// Any job another worker's transaction already locked is silently skipped,
// not waited for — that is what makes two replicas safe to run together.
func (r *JobRepository) Claim(ctx context.Context, batchSize int) ([]ClaimedJob, func(commit bool) error, error) {
tx, err := r.pool.Begin(ctx)
if err != nil {
return nil, nil, err
}
rows, err := tx.Query(ctx, `
SELECT id, organization_id, channel, payload, attempts, max_attempts
FROM notification_jobs
WHERE status = 'pending' AND next_attempt_at <= now()
ORDER BY next_attempt_at
LIMIT $1
FOR UPDATE SKIP LOCKED`, batchSize,
)
if err != nil {
tx.Rollback(ctx)
return nil, nil, err
}
var claimed []ClaimedJob
for rows.Next() {
var j ClaimedJob
if err := rows.Scan(&j.ID, &j.OrganizationID, &j.Channel, &j.Payload, &j.Attempts, &j.MaxAttempts); err != nil {
rows.Close()
tx.Rollback(ctx)
return nil, nil, err
}
claimed = append(claimed, j)
}
rows.Close()
_, err = tx.Exec(ctx, `UPDATE notification_jobs SET status = 'processing' WHERE id = ANY($1)`,
jobIDs(claimed))
if err != nil {
tx.Rollback(ctx)
return nil, nil, err
}
finish := func(commit bool) error {
if commit {
return tx.Commit(ctx)
}
return tx.Rollback(ctx)
}
return claimed, finish, nil
}
func jobIDs(jobs []ClaimedJob) []string {
ids := make([]string, len(jobs))
for i, j := range jobs {
ids[i] = j.ID
}
return ids
}Claim marks rows 'processing' and commits before the worker attempts delivery — if the process crashes mid-delivery, the job is left in 'processing' rather than silently reappearing as 'pending' for a second worker to also pick up; a stuck-processing sweep is a natural follow-up, not covered here.
Every Attempt, Successful or Not, Leaves a Record
notification_deliveries is the evidence trail an operator or auditor needs later: which job, which attempt number, which provider message ID, and what error, if any. It is written on every attempt, not only on failure — a support agent investigating 'my resident says they never got the SMS' needs to see that it was, in fact, attempted and accepted by the provider.
package application
import (
"context"
"errors"
"time"
"github.com/hoa-platform/backend/services/notification-service/internal/domain"
)
type DeliveryRecorder interface {
Record(ctx context.Context, jobID string, attempt int, providerMessageID string, err error)
}
type JobStore interface {
MarkSent(ctx context.Context, jobID string) error
MarkFailed(ctx context.Context, jobID string, nextAttemptAt time.Time) error
MarkDead(ctx context.Context, jobID string) error
}
func backoff(attempt int) time.Duration {
// 1m, 2m, 4m, 8m, capped — deliberately short because most transient
// provider failures resolve within minutes, not hours.
step := time.Minute << attempt
if step > 30*time.Minute {
step = 30 * time.Minute
}
return step
}
func processOne(ctx context.Context, job ClaimedJob, deliver func(context.Context, ClaimedJob) (string, error), store JobStore, recorder DeliveryRecorder) {
providerMessageID, err := deliver(ctx, job)
recorder.Record(ctx, job.ID, job.Attempts+1, providerMessageID, err)
switch {
case err == nil:
_ = store.MarkSent(ctx, job.ID)
case errors.Is(err, domain.ErrPermanent):
_ = store.MarkDead(ctx, job.ID)
case job.Attempts+1 >= job.MaxAttempts:
_ = store.MarkDead(ctx, job.ID)
default:
_ = store.MarkFailed(ctx, job.ID, time.Now().Add(backoff(job.Attempts)))
}
}A permanent error skips the remaining retry budget entirely and goes straight to dead — retrying a malformed recipient five times would only delay the operator noticing it needs a human, not a resend.
A job marked dead stays in the table with its full attempt history intact. Once Kafka Integration is introduced later in the course, dead jobs become the natural source for a formal dead-letter topic — this table does not need to anticipate that migration today.
Prove Two Workers Cannot Claim the Same Job
Applied exercise
Add a stuck-processing sweep
A worker that crashes mid-delivery leaves a job stuck in 'processing' forever, since nothing currently reclaims it.
- Add a column or use existing timestamps to detect a job that has been 'processing' for longer than a reasonable delivery time.
- Write the SQL that resets such a job back to 'pending' so a healthy worker can claim it again.
- Decide whether this sweep runs as its own scheduled job or inline before every Claim call.
- Write a test proving a freshly-claimed, still-in-progress job is NOT reset by the sweep.
Deliverable
A migration or query change implementing the sweep, plus the test distinguishing stuck from in-progress jobs.
Completion checks
- The sweep only resets jobs stuck well past a reasonable delivery window, not every 'processing' row.
- A job actively being delivered by a healthy worker is left untouched.
What would happen if Claim used a plain SELECT ... FOR UPDATE without SKIP LOCKED, while two worker replicas ran concurrently?
Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.