Add advisory event when JetStream messages are removed from a stream

[AnasAhmedTMF]

Summary

Add a new JetStream advisory that fires whenever a message is removed from a stream, regardless of the reason. This would cover all deletion scenarios:

• MaxAge — message expired due to age limit
• MaxBytes — message evicted to enforce storage limits
• MaxMsgs / MaxMsgsPerSubject — message removed due to message count limits
• InterestPolicy — message deleted after all consumers acknowledged (or no consumers exist)
• Explicit purge/delete — administrative removal

Motivation

Currently, when a message is removed from a stream there is no notification mechanism. This makes it difficult to:

1. Audit — track what was removed and why, especially for compliance-sensitive workloads
2. Dead-letter processing — react to messages that expired before being consumed (e.g., retry, alert, persist elsewhere)
3. Observability — monitor message lifecycle without polling stream info
4. Data pipelines — forward expiring messages to cold storage or external systems before they're lost

The existing advisory system covers stream/consumer create/delete/update but not message-level lifecycle events.

Proposed Design

Advisory Subject

$JS.EVENT.ADVISORY.STREAM.MSG_REMOVED.<stream>

Subscribable via wildcard: $JS.EVENT.ADVISORY.STREAM.MSG_REMOVED.>

Payload

{
  "type": "io.nats.jetstream.advisory.v1.stream_msg_removed",
  "id": "<nuid>",
  "timestamp": "<RFC3339>",
  "stream": "<stream-name>",
  "seq": 123,
  "subject": "<original-msg-subject>",
  "reason": "max_age | max_bytes | max_msgs | interest | purge | delete",
  "data": "<base64-encoded-payload>",
  "hdr": "<base64-encoded-headers>"
}

Configuration

Since including the full payload could be expensive at high throughput, this could be opt-in at the stream config level:

{
  "msg_delete_advisory": true,
  "msg_delete_advisory_include_payload": true
}

Or only emit the advisory with metadata (seq, subject, reason) by default, with payload inclusion as an option.

Considerations

• Performance — At high message removal rates, emitting advisories with payloads could add overhead. Making it opt-in and allowing metadata-only mode mitigates this.
• Clustering — Advisory should be emitted by the stream leader only.
• Consistency — The advisory should fire before the message data is removed from the store, so the payload is still available.

Use Cases

1. Dead-letter queue pattern without a dedicated DLQ consumer
2. Audit trail for regulated environments
3. Automatic archival of expiring messages to object storage
4. Alerting when messages are dropped due to limits (capacity planning signal)

[ripienaar]

These limits (age etc) are not exceptional circumstances and especially the count ones would constantly spam advisories.

Explicit purge and delete have advisories in the api already and max retries also - these are exceptional circumstances.

I can’t imagine adding advisories for regular normal day to day (or even second to second) things it’s too expensive computationally to do.

We also can’t put the payload as when this was previously suggested we had a lot of feedback that advisories end up in all sorts of places (ops tooling) and so would come with a huge risk of exfilling sensitive data.