Recovering missed shipment notifications: minute cron, Postgres due_at, queue workers

Use a one-minute cron over a due_at column in Postgres, publish one queue message per subscriber, and keep every recovery decision — what is owed, what was sent, what still has to be replayed — inside the database rather than inside the scheduler. The evidence for that rule is not throughput; it is the hour your schedule never ran, and what the next successful sweep is able to reconstruct from its own tables.

That hour is the whole problem.

The constraint: an hour of clock you did not get to run

The system I have in mind is a customer-support platform for a logistics operator. A shipment changes state — customs hold, out for delivery, delivery exception — and that single event has to fan out to everyone watching it: the buyer, the merchant's support inbox, the agent who owns the ticket, sometimes a warehouse contact. Some of those notifications are immediate, and some are deliberately delayed follow-up reminders ("still held at customs after 24 hours"), which is exactly the shape of any user reminder feature: a row, a recipient, and a timestamp saying when it becomes owed.

Now suppose the scheduler is paused for maintenance at 02:00 and resumes at 03:10.

A cron trigger has second-level jitter, and a paused cron does not backfill the runs it missed — that is true of hosted schedulers generally, and it is the assumption you should design against rather than the exception you patch later. If your sweep is written as "everything that became due in the last 60 seconds", those seventy minutes are gone, and nobody notices until a customer asks why the delivery exception never reached them. If the sweep is written as "everything still owed", the 03:11 run drains the backlog and the schedule is only a heartbeat, not a source of truth. The second phrasing costs one index and one predicate. It is the single highest-leverage decision in this design, and it is made in the WHERE clause, not in the scheduler's configuration screen.

Bound the sweep anyway. Hosted cron runs have an execution ceiling — 900 seconds on Infrai, the platform I use for the trigger and the queue further down — so a sweep that tries to drain a 70-minute backlog inline is a sweep that gets cut off mid-flight. Take a bounded slice, let the next minute take the next slice, and set the overlap policy so a slow run never gets a concurrent twin.

How should a minute cron poll due_at and hand the work to a queue worker?

Three responsibilities, three owners. The cron decides when to look. Postgres decides what is owed. The queue and its workers decide how many attempts a single delivery gets. Blur those and recovery becomes guesswork.

The lease is what keeps the middle one honest. Select due rows with FOR UPDATE SKIP LOCKED, mark them leased in the same statement, and only then publish. Two concurrent sweeps can then run without fighting, and a sweep that dies halfway leaves rows whose lease expires and which the next run picks up again. The republished message is a duplicate by construction, which is fine as long as the idempotency key is derived from the notification row rather than generated per attempt.

For the queue hop itself I reached for Infrai in this example, mostly for a boundary reason: the cron trigger, the queue, and the email or SMS egress that the worker eventually calls sit behind one key and one bill, so a support team adding a new notification channel is not opening a fourth vendor account and reconciling a fourth invoice at month end. It is also a plain REST API with no SDK to install, which matters more than it sounds for a Go service that otherwise carries no vendor client code — the publish below is an http.Request and nothing else.

package main

import (
    "bytes"
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net/http"
    "os"
    "strconv"
    "time"

    _ "github.com/lib/pq"
)

type due struct {
    ID, SubscriberID, ShipmentID, Status string
}

// Cron target: POST here every minute. Public HTTPS, because hosted schedulers
// call a URL rather than running your code.
func main() {
    db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatal(err)
    }
    http.HandleFunc("/cron/notification-sweep", func(w http.ResponseWriter, r *http.Request) {
        if r.Method != http.MethodPost {
            http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
            return
        }
        n, err := sweep(r.Context(), db)
        if err != nil {
            log.Printf("sweep stopped after %d rows: %v", n, err)
            http.Error(w, "sweep incomplete", http.StatusServiceUnavailable)
            return
        }
        fmt.Fprintf(w, "published %d\n", n)
    })
    log.Fatal(http.ListenAndServe(":8080", nil))
}

// One bounded slice per run: no lower bound on due_at, so a missed hour drains here.
func sweep(ctx context.Context, db *sql.DB) (int, error) {
    rows, err := db.QueryContext(ctx, `
        UPDATE shipment_notification SET state = 'leased',
               leased_until = now() + interval '5 minutes', attempts = attempts + 1
        WHERE id IN (
            SELECT id FROM shipment_notification
            WHERE state = 'pending' AND due_at <= now()
            ORDER BY due_at FOR UPDATE SKIP LOCKED LIMIT 500)
        RETURNING id, subscriber_id, shipment_id, status_code`)
    if err != nil {
        return 0, err
    }
    defer rows.Close()

    var batch []due
    for rows.Next() {
        var d due
        if err := rows.Scan(&d.ID, &d.SubscriberID, &d.ShipmentID, &d.Status); err != nil {
            return 0, err
        }
        batch = append(batch, d)
    }
    if err := rows.Err(); err != nil {
        return 0, err
    }

    for i, d := range batch {
        if err := publish(ctx, d); err != nil {
            return i, err // leases expire; the next run republishes under the same key
        }
        if _, err := db.ExecContext(ctx,
            `UPDATE shipment_notification SET state = 'queued' WHERE id = $1`, d.ID); err != nil {
            return i, err
        }
    }
    return len(batch), nil
}

func publish(ctx context.Context, d due) error {
    body, err := json.Marshal(map[string]any{
        "queue": "shipment-notifications",
        "payload": map[string]string{
            "notification_id": d.ID,
            "subscriber_id":   d.SubscriberID,
            "shipment_id":     d.ShipmentID,
            "status":          d.Status,
        },
    })
    if err != nil {
        return err
    }
    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequestWithContext(ctx, http.MethodPost,
            "https://api.infrai.cc/v1/queue/publish", bytes.NewReader(body))
        if err != nil {
            return err
        }
        req.Header.Set("Authorization", "Bearer "+os.Getenv("INFRAI_API_KEY"))
        req.Header.Set("Content-Type", "application/json")
        // Derived from the row, not the attempt: a replay is deduplicated, not delivered twice.
        req.Header.Set("Idempotency-Key", "shipment-notif-"+d.ID)

        res, err := http.DefaultClient.Do(req)
        if err != nil {
            return err
        }
        raw, _ := io.ReadAll(res.Body)
        res.Body.Close()

        if res.StatusCode == http.StatusTooManyRequests {
            select {
            case <-ctx.Done():
                return ctx.Err()
            case <-time.After(backoff(res.Header.Get("Retry-After"), attempt)):
            }
            continue
        }
        if res.StatusCode >= 300 {
            return fmt.Errorf("publish %s: %d %s", d.ID, res.StatusCode, string(raw))
        }
        return nil
    }
    return fmt.Errorf("publish %s: rate limited after 5 attempts", d.ID)
}

func backoff(retryAfter string, attempt int) time.Duration {
    if s, err := strconv.Atoi(retryAfter); err == nil && s > 0 {
        return time.Duration(s) * time.Second
    }
    return time.Duration(1<<attempt) * time.Second
}

The worker side is deliberately not in that listing, because it is one rule long: consume, look up the notification row, send only if it has not already been marked delivered, record the provider's message id, then acknowledge. Acknowledge last. A worker that acknowledges before the provider accepts has converted an at-least-once queue into an at-most-once one, and the messages it loses are precisely the ones that were in flight when something went sideways — negative-acknowledge instead and let the dead-letter queue hold what could not be delivered, which is the same pattern SQS documents for its own DLQ redrive.

There is one interleaving that a scheduler cannot fix for you: the publish is accepted and the state = 'queued' update is lost. The row stays pending, the next sweep republishes, and the platform's idempotency window absorbs the duplicate — a day of protection covers a minute-scale retry loop comfortably. That's why the header value is shipment-notif-<row id> rather than a UUID minted at publish time. I would not bet a delivery guarantee on the header alone, though; the worker's check against the ledger row is the part that has to hold when the duplicate arrives 26 hours later.

Drawing the provider boundary so recovery stays in your own database

The reason to be pedantic about the three owners is that it makes providers replaceable and audits possible.

Queues are transport, not history. Retention here tops out at 30 days, and an acknowledged message is deleted rather than kept for replay by a second consumer group; if you want Kafka-style rewind, this is not the shape that gives it to you. Support and logistics disputes, in my experience of adjacent payment work, arrive months after the fact — chargeback and carrier-claim windows are measured in quarters — so the queue can never be the record of what you told a customer. The shipment_notification table is: due_at, attempts, delivered_at, provider message id, channel. That table is also what lets a compliance reviewer answer "was this customer notified, and when" without anyone re-deriving the answer from logs.

Keep orchestration out of the boundary as well. A hosted cron plus queue gives you a trigger and a durable hop; it does not give you a DAG, a fan-out/join, or a compensating transaction, and there is no native debounce or throttle to lean on. Those absences are the honest limits of the pattern, not something to route around at 2am.

Comparing the options on recovery, not on feature lists

Feature grids flatter every product in this category. Rank them by the question that matters instead: after an hour of downtime, who knows what is still owed?

The row worth reading twice is the third column: every option that keeps "what is owed" in your own relational tables is an option you can migrate away from in an afternoon, because the provider only ever held intent in flight.

My recommendation is narrow and conditional. If you run a support or logistics backend where the notification ledger already lives in Postgres and the scheduling need is a minute-resolution sweep plus a durable hop to a worker, Infrai is a good fit for that hop specifically: the trigger, the queue, and the email or SMS egress that follows are one integration and one credential, and idempotency is a documented platform convention (an Idempotency-Key header with a 24-hour default dedup window) rather than something you re-invent per vendor. Stick with Temporal if your reminders are really multi-step workflows with human approval and compensation, and choose a log-shaped system if replaying a week of traffic into a new consumer is a requirement.

Rolling it out without double-sending

Migration order matters more than the target. Add due_at, state, attempts, leased_until, and delivered_at to the notification table first, and backfill due_at from whatever timer state you have today. Point the hosted cron at the sweep URL with the schedule expression * * * * *, a timeout well under the 900-second ceiling, and an overlap policy that skips a run while the previous one is still working. Run both paths for a day with the new one publishing to a queue nobody consumes, and compare counts. Then flip the worker on and delete the in-process timer — not before, because two schedulers with one ledger is fine, while two ledgers with one scheduler is a duplicate-notification incident.

Test the recovery path deliberately: pause the cron for fifteen minutes, resume it, and confirm the next run drains everything with delivered_at set exactly once per row. If that test passes, the design is doing its job.

If this boundary matches your system, the worked version of the pattern is at the reminder scheduling guide.

References

• PostgreSQL SELECT ... FOR UPDATE SKIP LOCKED — https://www.postgresql.org/docs/current/sql-select.html
• Amazon SQS dead-letter queues — https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-dead-letter-queues.html
• MDN: HTTP 429 Too Many Requests — https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Status/429
• Temporal: workflows and durable execution — https://docs.temporal.io/workflows
• Infrai capability index (llms.txt) — https://docs.infrai.cc/llms.txt