Messaging and jobs

How the Colden kit moves work off the request: Wolverine messages on PostgreSQL queues published with the data that causes them, the worker's handlers, the recurring jobs Hangfire schedules, failures and replay, and how to add a message or a job.

Work that does not have to finish inside a request goes through a message: sending an email, applying a Stripe event, cleaning up old rows. The kit uses Wolverine with its queues and its message store in the kit's own PostgreSQL database, so there is no broker to run, and Hangfire only to keep the clock: each recurring job publishes one message, and the message's handler does the work. The packages are described in Colden.Messaging and Colden.Jobs.

How a message travels

1. A slice in the api changes data through ColdenOutbox, publishes the message and saves once. The message is written to the outbox in the same transaction as the data, so a message exists only when the change committed, and every committed change has its message.
2. The outbox moves the message to its queue's table. The message carries its organization (the envelope's tenant), the request's correlation id and the person who acted.
3. The worker listens on every queue (Messaging:Listen). Before the handler runs, the message's organization becomes the tenant of the handler's scope, so the handler's database context sees that organization's rows only.
4. The handler does the work. A handler may publish further messages the same way, through the outbox of its own transaction.

A message may arrive more than once (a worker that stops after the work and before the acknowledgment), so every handler is idempotent: it checks the state it would create before it creates it.

Messages are records in packages/core/Colden.Core/Messages/: events named in the past tense (InvitationCreated, InvoicePaid) and commands in the imperative (SendOutgoingEmail, ReconcileBilling), each with its organization, the time it happened and the correlation id. A contract changes only by adding fields, because messages written by the previous release may still wait in a queue during a deployment.

The queues

Each queue is a table in the wolverine schema (wolverine_queue_<name>). Messaging:Queues:<queue>:Concurrency (4) sets how many of its messages a worker processes at once; more workers process more in parallel.

The recurring jobs

The worker runs the Hangfire server (Jobs:Server:Enabled), registers every job at startup and removes the recurring jobs no definition declares any more. Schedules are UTC.

Jobs:Recurring:<id>:Cron replaces a job's schedule without a code change, for example Jobs:Recurring:billing-reconciliation:Cron set to 0 2 * * * in a host's appsettings.json or its environment.

Failures and replay

• A failing handler is retried Messaging:Retry:Attempts (3) times, the first after Messaging:Retry:BaseDelayMilliseconds (1 second) and each next one five times later. Then the message moves to the error queue, wolverine.wolverine_dead_letters, with its exception.
• A handler that knows a retry cannot help throws PermanentException, and the message goes to the error queue at once.
• Super-admins see every queue's waiting, scheduled and failed messages with the latest failures in the system health screen (/admin/system-health, GET /v1/admin/system/queues). Replay puts a queue's failed messages (of one type, when chosen) back on the queue once the cause is fixed; discard deletes them. Both are logged as warnings with the count (events 7301 and 7302).
• A failed Hangfire job only means the message could not be published; its run and the next run are in the same screen and in the job dashboard.
• The worker reports the counts as metrics every Diagnostics:ReportIntervalSeconds (30), so an alert can watch the error queue (observability).

Adding a message and its handler

1. Add the record to packages/core/Colden.Core/Messages/<Area>/, implementing IColdenEvent or IColdenCommand, with TenantId, OccurredAt and CorrelationId.
2. Route it to a queue in the host that publishes it: services.AddColdenMessageRoute<TheMessage>(ColdenQueues.<Queue>). Pick the queue by the work's kind, so a slow kind cannot delay another.
3. Publish it through ColdenOutbox in the slice, and save once.
4. Add the handler to apps/worker/Colden.Worker/Handlers/<Area>/<Name>Handler.cs: a class with a Handle (or HandleAsync) method that takes the message and the services it needs. Make it idempotent. A platform-wide handler (no organization on the message) takes TenantContext and opens AllowCrossTenant() around the queries that read across organizations.
5. Test the slice's publication (the message is in the outbox after the save, and absent after a failed save) and the handler (its effect, and the same message handled twice changes nothing) with the worker's host factory.
6. Add the message to the queue table above and to the messaging section of the architecture file.

Adding a recurring job

public sealed class ArchiveOldProjectsJob : IRecurringJobDefinition
{
    public string Id => "archive-old-projects";

    public string Cron => "30 4 * * *"; // 04:30 UTC every day

    public object CreateMessage(DateTimeOffset now) => new ArchiveOldProjects(TenantId: null, now, CorrelationId: $"job:{Id}:{now:O}");
}

Put the definition in apps/worker/Colden.Worker/Jobs/, route its message to a queue in the worker's Program.cs, and write the handler as above; the job publishes, the handler works. Keep the job's id stable: a renamed id is a new job, and the old one is removed at the next start.

Running it

• Storage. The message store and the queue tables are created by the api binary's setup-messaging command before the hosts start: a Helm hook after the migration in a cluster, a one-off service in Compose. In Development a host builds them itself (Messaging:AutoBuildStorage). Hangfire creates its own schema, hangfire, on first use.
• Health. Both hosts are not ready while the message store is missing (the wolverine check), and the worker while Hangfire's storage does not answer (the hangfire check).
• Dashboard. The job dashboard is on the worker's management port at /hangfire, behind Jobs:DashboardPassword; it has no ingress, so reach it with a port-forward in a cluster.
• Scaling. More worker replicas share the queues; each message is handled by one of them, and Hangfire runs each scheduled job once across them.

When PostgreSQL queues are not enough

The queues share the database with the data, which keeps the outbox transactional and the operations simple. When the measured message rate or the queue tables' load on the database become the bottleneck, Wolverine's RabbitMQ transport replaces the PostgreSQL transport for the queues without changing a message or a handler; the outbox stays in PostgreSQL. Measure first, and record the decision with its numbers.