Delayed / deferred message delivery #258

Context

There's no way to send or publish a message for delivery later. Transports only delay retries:

• SQS: visibility timeouts
• RabbitMQ: TTL bucket queues <queue>-retry-<n>ms
• in-memory: setTimeout

The original proposal doesn't hold up on every transport:

• SQS: bus-sqs sends and publishes through SNS, which has no per-message delay, so DelaySeconds can't be used. Holding a message on the receiving side is capped at 12 h per receive, and each receive counts towards maxReceiveCount.
• RabbitMQ: TTL buckets only expire at the head of the queue, so a long delay can arrive up to 2x late. The delayed-message plugin isn't in the CI image and doesn't replicate.
• Phase 4: Kafka and Redis Streams (Additional transports: Kafka, Azure Service Bus, Redis Streams #267) have no native delay at all.

This is a prerequisite for workflow timeouts (#259) and the durable outbox (#260), which reuses its store and dispatcher.

Decisions

Answered by the maintainer:

1. One mechanism for every transport. Delayed messages are stored in the persistence, and each bus runs a dispatcher that sends them when they're due. There are no native transport paths.
2. Warn when the store isn't durable. A bus whose persistence doesn't survive a restart (such as the default InMemoryPersistence) logs a warning that scheduled messages are lost on restart, and says to configure a durable persistence with withPersistence() (e.g. bus-postgres). The warning is logged once per bus, on the first delayed send.

Proposed:

3. API. send(cmd, { deliverAfter: ms }) or send(cmd, { deliverAt: Date }), and the same for publish, on the bus and on ctx. The two are a mutually exclusive union on the options parameter, which extends Partial<MessageAttributes>.
4. Store. Shared with Durable transactional outbox #260: rows of { id, kind: 'send' | 'publish', message (plain JSON), attributes, headers, dueAt, leaseUntil }. id is the messageId from Core message id, sent time and endpoint name #322.
5. When middleware runs. Outgoing middleware (Incoming and outgoing pipeline behaviours (middleware) #263) runs once, when the message is scheduled. The dispatcher hands the stored message to the transport.
6. Durability. Persistences declare it with an optional readonly durable?: boolean. InMemoryPersistence sets false.

Non-goals

• Native transport delays (SQS DelaySeconds, the RabbitMQ delayed-message plugin, TTL buckets).
• Cancelling a scheduled message.
• Recurring or cron schedules.
• Precision finer than the dispatcher's poll interval.

Approach

• Outgoing pipeline: when dueAt is in the future, the last outgoing step writes a row instead of calling the transport. Inside a handler this happens when the outbox flushes.
• New optional Persistence methods:
  • storeOutgoingMessages(messages)
  • claimDueOutgoingMessages(now, limit, leaseMs)
  • deleteOutgoingMessages(ids)
• Implementations:
  • InMemoryPersistence
  • bus-postgres: an outgoing_messages table in the configured schema, claimed with FOR UPDATE SKIP LOCKED
  • bus-mongodb: an outgoingmessages collection with an index on dueAt, claimed with a findOneAndUpdate lease
• Missing support: a delayed send on a persistence without these methods throws DelayedDeliveryNotSupported, which names the persistence and the fix.
• Dispatcher:
  • Runs on every started, receiving bus. It polls every 1 s (a constant), plus an in-process timer for the earliest dueAt this process scheduled.
  • Claims rows with a lease, so several instances on one store send each row once. It stops on stop().
  • Send-only buses and Receiver buses (Lambda) only write rows; another running bus on the same store dispatches them. The docs say so.
• Durability warning: logged at warn on the first delayed send when persistence.durable === false.

Acceptance criteria

• deliverAfter/deliverAt on bus.send/publish and ctx.send/publish, type-checked as mutually exclusive
• A delayed message keeps its correlationId, sticky attributes and headers, and isn't delivered before dueAt
• Several bus instances on one store send each due message once (lease test)
• InMemoryPersistence, bus-postgres and bus-mongodb implement the store; a shared scheduledMessageTests(persistence) suite in bus-test covers them
• A persistence without the store methods gets DelayedDeliveryNotSupported with the fix in its help
• A bus with a non-durable persistence (the default InMemoryPersistence) logs one warning on its first delayed send, saying scheduled messages are lost on restart and how to configure a durable persistence; a test covers it
• Docs page /guide/delayed-delivery, including the send-only, Lambda and durability limits; changeset