Consuming Events

This guide covers how to consume events from the Notifications Service using both the REST API and gRPC API, including detailed information about durable and non-durable queues.

Overview

The Event Consumer Service allows clients to stream events in real-time. You can filter events by event type and resource ID, and choose between durable and non-durable consumption modes.

Durable vs Non-Durable Queues

One of the most important decisions when consuming events is whether to use a durable queue or a non-durable queue. Understanding the difference is critical for building reliable applications.

Non-Durable Queues

What they are:

• Temporary queues created automatically when you start consuming

• Only receive events that occur while you are connected

• Automatically deleted after disconnection (after a brief TTL)

• No setup required - just start streaming

When to use:

• Real-time notifications in UI applications

• Live dashboards and monitoring displays

• Event logging where missing some events is acceptable

• Prototyping and development

• Scenarios where events are only relevant “right now”

Advantages:

• Simple - no queue management needed

• Lightweight - no persistent storage

• Automatic cleanup

• Good for applications with many concurrent users (each gets their own temporary queue)

Disadvantages:

• Events occurring while disconnected are lost

• Not suitable for critical event processing

Example use cases:

✓ Showing "Someone just uploaded a file" notifications in a web UI
✓ Live activity feed in a dashboard
✓ Real-time log streaming during development
✗ Processing every file upload for thumbnail generation (use durable)
✗ Triggering critical workflows (use durable)
✗ Syncing data between services (use durable)

Durable Queues

What they are:

• Persistent queues that must be explicitly created before use

• Store events even when you’re disconnected

• Receive all events from the time the queue was created

• Must be explicitly deleted when no longer needed

When to use:

• Critical event processing where no events can be missed

• Microservices that need to process every event

• Batch processing workflows

• Systems that may have downtime for deployments

• Event-driven architectures where events trigger actions

Advantages:

• No events are lost - all events since queue creation are preserved

• Can disconnect and reconnect without losing messages

• Events are queued up during downtime

• Reliable for building event-driven systems

Disadvantages:

• Requires explicit queue lifecycle management (create/delete)

• Consumes storage resources

• Need to persist the queue_id in your application configuration

• Must remember to delete queues when done

Example use cases:

✓ Thumbnail generation service processing every file upload
✓ Notification delivery service sending emails for events
✓ Data synchronization between services
✓ Audit logging where every event must be recorded
✗ UI notifications (non-durable is simpler)
✗ Temporary development testing (non-durable is easier)

Decision Matrix

Scenario	Recommended Choice	Reason
UI real-time notifications	Non-durable	Users only care about what’s happening now
File processing pipeline	Durable	Must process every file
Live dashboard	Non-durable	Current state is what matters
Email notification service	Durable	Must send every notification
Development/testing	Non-durable	Easier, no cleanup needed
Production microservice	Durable	Can’t miss events during restarts
Webhook relay	Durable	Must deliver all webhooks
Activity feed	Non-durable	Only recent activity matters

Key Principle

If missing an event would break your application or cause data inconsistency, use a durable queue. If events are only relevant “in the moment,” use a non-durable queue.

Consuming with Non-Durable Queues

REST API

Endpoint: GET /api/v1beta/events/stream

The REST API uses Server-Sent Events (SSE) for streaming.

Basic Example (No Filtering):

# Consume all events of all types
curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream" \
  -H "Authorization: Bearer YOUR_TOKEN"

With Event Type Filtering:

# Create filter groups as JSON
FILTERS='[
  {
    "event_type": "storage.file.created",
    "filters": []
  }
]'

# URL encode and pass as query parameter
curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -G --data-urlencode "filter_groups=$FILTERS"

With Resource Filtering:

# Only receive events for files in /uploads/ directory and subdirectories
FILTERS='[
  {
    "event_type": "storage.file.created",
    "filters": [
      {
        "filter_type": "starts_with_greedy",
        "resource_id": "/uploads/"
      }
    ]
  }
]'

curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -G --data-urlencode "filter_groups=$FILTERS"

Response Format:

event: connected
id: my-queue-id
data: {"message":"Connection established","reconnect_token":"my-queue-id"}

event: event
data: {"event_type":"storage.file.created","principal_identity":"user@example.com","occurred_at":"2024-10-16T14:30:00Z","published_at":"2024-10-16T14:30:01Z","message":{"file_name":"doc.pdf"}}

event: event
data: {"event_type":"storage.file.created","principal_identity":"user2@example.com","occurred_at":"2024-10-16T14:31:00Z","published_at":"2024-10-16T14:31:01Z","message":{"file_name":"image.png"}}

The initial connected event includes the reconnect_token (which is the queue ID). Save this token for reconnection.

Python Example with SSE Client:

import sseclient
import requests
import json

url = "https://your-consumer-service.example.com/api/v1beta/events/stream"
headers = {
    "Authorization": f"Bearer {YOUR_TOKEN}",
    "Accept": "text/event-stream"
}

# Define filters
filters = [
    {
        "event_type": "storage.file.created",
        "filters": [
            {
                "filter_type": "starts_with_greedy",
                "resource_id": "/uploads/"
            }
        ]
    }
]

params = {"filter_groups": json.dumps(filters)}

# Stream events
response = requests.get(url, headers=headers, params=params, stream=True)
client = sseclient.SSEClient(response)

reconnect_token = None

for event in client.events():
    if event.event == "connected":
        connected_data = json.loads(event.data)
        reconnect_token = connected_data['reconnect_token']
        print(f"Connected, reconnect token: {reconnect_token}")
    elif event.event == "event":
        data = json.loads(event.data)
        print(f"Received: {data['event_type']}")
        print(f"Message: {data['message']}")

Reconnecting After Disconnection:

If your connection drops, use the reconnect_token to reconnect to the same queue and avoid missing events:

curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream?reconnect_token=abc123xyz" \
  -H "Authorization: Bearer YOUR_TOKEN"

Updating Filters Dynamically (REST):

You can update filters on a non-durable queue while it continues to exist. This is useful when you need to change what events you’re listening for without losing the queue. To update filters, make a new request with the reconnect_token, the new filter_groups, and the previous_filter_groups:

# Original filters we were using
PREV_FILTERS='[
  {
    "event_type": "storage.file.created",
    "filters": [
      {
        "filter_type": "starts_with_greedy",
        "resource_id": "/uploads/"
      }
    ]
  }
]'

# New filters we want to use
NEW_FILTERS='[
  {
    "event_type": "storage.file.created",
    "filters": [
      {
        "filter_type": "starts_with_greedy",
        "resource_id": "/uploads/"
      }
    ]
  },
  {
    "event_type": "storage.file.deleted",
    "filters": [
      {
        "filter_type": "starts_with_greedy",
        "resource_id": "/uploads/"
      }
    ]
  }
]'

# Reconnect with updated filters
curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -G --data-urlencode "reconnect_token=abc123xyz" \
  --data-urlencode "filter_groups=$NEW_FILTERS" \
  --data-urlencode "previous_filter_groups=$PREV_FILTERS"

Python Example with Filter Updates:

import sseclient
import requests
import json

def consume_with_filter_updates():
    url = "https://your-consumer-service.example.com/api/v1beta/events/stream"
    headers = {
        "Authorization": f"Bearer {YOUR_TOKEN}",
        "Accept": "text/event-stream"
    }
    
    # Initial filters
    current_filters = [
        {
            "event_type": "storage.file.created",
            "filters": [
                {"filter_type": "starts_with_greedy", "resource_id": "/uploads/"}
            ]
        }
    ]
    
    params = {"filter_groups": json.dumps(current_filters)}
    reconnect_token = None
    
    # Start consuming
    response = requests.get(url, headers=headers, params=params, stream=True)
    client = sseclient.SSEClient(response)
    
    for event in client.events():
        if event.event == "connected":
            connected_data = json.loads(event.data)
            reconnect_token = connected_data['reconnect_token']
        elif event.event == "event":
            data = json.loads(event.data)
            process_event(data)
            
            # Example: Update filters after some condition
            if should_update_filters():
                # Close current connection
                response.close()
                
                # Define new filters
                new_filters = [
                    {
                        "event_type": "storage.file.created",
                        "filters": [
                            {"filter_type": "starts_with_greedy", 
                             "resource_id": "/uploads/"}
                        ]
                    },
                    {
                        "event_type": "storage.file.deleted",
                        "filters": [
                            {"filter_type": "starts_with_greedy", 
                             "resource_id": "/uploads/"}
                        ]
                    }
                ]
                
                # Reconnect with updated filters
                params = {
                    "reconnect_token": reconnect_token,
                    "filter_groups": json.dumps(new_filters),
                    "previous_filter_groups": json.dumps(current_filters)
                }
                
                response = requests.get(url, headers=headers, params=params, 
                                       stream=True)
                client = sseclient.SSEClient(response)
                
                # Update our tracking of current filters
                current_filters = new_filters

Important Notes on Updating Filters (REST):

• When updating filters, you must provide both filter_groups (new filters) and previous_filter_groups (old filters)

• The reconnect_token identifies which queue to update

• The queue must already exist (you cannot update filters before making an initial consume request)

• If filter_groups equals previous_filter_groups, no update is performed

gRPC API

RPC: ConsumeNonDurableEvents (bidirectional streaming)

Python Example:

import grpc
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2_grpc

# Setup channel and stub
channel = grpc.secure_channel(
    'your-consumer-service.example.com:50052',
    grpc.ssl_channel_credentials()
)
stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)

# Define filters
filter_group = event_consumer_pb2.FilterGroup(
    event_type='storage.file.created'
)
resource_filter = event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
)
filter_group.filters.append(resource_filter)

# Create request stream
def request_stream():
    # Send initial request with filters
    request = event_consumer_pb2.ConsumeNonDurableEventsRequest()
    request.filter_groups.append(filter_group)
    yield request
    # Keep stream open (or send updated filters later)

# Start consuming
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]
responses = stub.ConsumeNonDurableEvents(request_stream(), metadata=metadata)

reconnect_token = None

try:
    for response in responses:
        # Save reconnect token for potential reconnection
        if response.reconnect_token:
            reconnect_token = response.reconnect_token
        
        # Process events
        for event in response.events:
            print(f"Event type: {event.event_type}")
            print(f"Occurred at: {event.occurred_at}")
            print(f"Message: {event.message}")
            print(f"Published by: {event.principal_identity}")
            print("---")
            
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.UNAUTHENTICATED:
        print("Token expired - reconnect with new token")
        # Reconnect with new token and reconnect_token

Reconnecting with Token:

def request_stream_reconnect(reconnect_token):
    request = event_consumer_pb2.ConsumeNonDurableEventsRequest(
        reconnect_token=reconnect_token
    )
    yield request

# Reconnect to same queue
responses = stub.ConsumeNonDurableEvents(
    request_stream_reconnect(reconnect_token), 
    metadata=new_metadata
)

Updating Filters Dynamically:

The bidirectional stream allows you to update filters while consuming. When sending an update request, you must include both the new filters in filter_groups and the previous filters in previous_filter_groups:

import threading
import queue

# Track current filters so we can send them as previous_filter_groups
current_filters = []

def request_stream_dynamic(filter_queue, initial_filters):
    global current_filters
    
    # Send initial filters
    initial_request = event_consumer_pb2.ConsumeNonDurableEventsRequest()
    for fg in initial_filters:
        initial_request.filter_groups.append(fg)
    current_filters = initial_filters
    yield initial_request
    
    # Wait for filter updates
    while True:
        new_filters = filter_queue.get()  # Blocks until new filters available
        if new_filters is None:  # Sentinel to stop
            break
            
        # Create update request with both new and previous filters
        update_request = event_consumer_pb2.ConsumeNonDurableEventsRequest()
        for fg in new_filters:
            update_request.filter_groups.append(fg)
        for fg in current_filters:
            update_request.previous_filter_groups.append(fg)
        
        # Update our tracking of current filters
        current_filters = new_filters
        yield update_request

# Initial filters
initial_filter_group = event_consumer_pb2.FilterGroup(
    event_type='storage.file.created'
)
resource_filter = event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
)
initial_filter_group.filters.append(resource_filter)
initial_filters = [initial_filter_group]

filter_queue = queue.Queue()
responses = stub.ConsumeNonDurableEvents(
    request_stream_dynamic(filter_queue, initial_filters), 
    metadata=metadata
)

# Later, update filters to add a new event type
new_filter_group_1 = event_consumer_pb2.FilterGroup(
    event_type='storage.file.created'
)
new_filter_group_1.filters.append(event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
))

new_filter_group_2 = event_consumer_pb2.FilterGroup(
    event_type='storage.file.deleted'
)
new_filter_group_2.filters.append(event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
))

# Send updated filters - the request_stream_dynamic function will
# automatically include the previous filters
filter_queue.put([new_filter_group_1, new_filter_group_2])

Important Notes on Updating Filters (gRPC):

• When updating filters, you must provide both filter_groups (new filters) and previous_filter_groups (old filters)

• If filter_groups equals previous_filter_groups, no update is performed

• The server uses previous_filter_groups to determine which bindings to remove when updating the queue

Consuming with Durable Queues

Durable queues require a three-step process: create, consume, delete.

Step 1: Create a Durable Queue

This is typically done once during application setup or deployment, not in your regular application code.

REST API

Endpoint: POST /api/v1beta/queues/durable

The request body MAY include an optional queue_id field. When you omit it, the server generates a queue_id for you. When you supply it, the server creates the durable queue with your client-specified ID.

Queue creation is ensure-exists (idempotent): if you supply a queue_id that already exists, the call succeeds against the existing queue rather than returning an error. The server signals which path the request took through the HTTP status code:

• 201 Created — the request created a new durable queue.

• 200 OK — the request resolved to a queue that already existed; nothing was created.

Branch on the status code to tell the two apart. In both cases the response body returns the queue_id.

Server-generated queue_id (omit queue_id from the body):

curl -X POST "https://your-consumer-service.example.com/api/v1beta/queues/durable" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "filter_groups": [
      {
        "event_type": "storage.file.created",
        "filters": [
          {
            "filter_type": "starts_with_greedy",
            "resource_id": "/uploads/"
          }
        ]
      }
    ]
  }'

Client-specified queue_id (include queue_id in the body):

curl -i -X POST "https://your-consumer-service.example.com/api/v1beta/queues/durable" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "queue_id": "thumbnail-service",
    "filter_groups": [
      {
        "event_type": "storage.file.created",
        "filters": [
          {
            "filter_type": "starts_with_greedy",
            "resource_id": "/uploads/"
          }
        ]
      }
    ]
  }'

The -i flag prints the status line so you can see whether the queue was created (201 Created) or already existed (200 OK). A second identical call with the same queue_id succeeds and returns 200 OK.

Response:

{
  "queue_id": "durable-queue-abc123xyz",
  "created": true
}

Important: Save the queue_id - you’ll need it for consuming and deleting!

gRPC API

RPC: CreateDurableQueue

import grpc
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2_grpc

# Setup channel and stub
channel = grpc.secure_channel(
    'your-consumer-service.example.com:50052',
    grpc.ssl_channel_credentials()
)
stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)

# Define filters
filter_group = event_consumer_pb2.FilterGroup(
    event_type='storage.file.created'
)
resource_filter = event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
)
filter_group.filters.append(resource_filter)

# Create durable queue
request = event_consumer_pb2.CreateDurableQueueRequest()
request.filter_groups.append(filter_group)
# Optionally supply a client-specified queue_id. Omit it to have the server
# generate one. Creation is ensure-exists (idempotent): supplying a queue_id
# that already exists succeeds against the existing queue instead of erroring.
request.queue_id = 'thumbnail-service'
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

response = stub.CreateDurableQueue(request, metadata=metadata)
queue_id = response.queue_id

# The created field mirrors the REST 201/200 distinction: it is True when this
# call created a new queue and False when an existing queue was returned unchanged.
if response.created:
    print(f"Created durable queue: {queue_id}")
else:
    print(f"Durable queue already existed: {queue_id}")
# IMPORTANT: Save this queue_id to your application config!

Step 2: Consume from Durable Queue

Now consume events using the queue_id. Your application can disconnect and reconnect using the same queue_id.

REST API

Endpoint: GET /api/v1beta/events/stream-durable

curl -N GET "https://your-consumer-service.example.com/api/v1beta/events/stream-durable?queue_id=durable-queue-abc123xyz" \
  -H "Authorization: Bearer YOUR_TOKEN"

Python Example:

Messages are acknowledged to the broker automatically by the server after a short delay. Clients do not need to acknowledge individual messages. If your application requires explicit client-side acknowledgment (e.g. to guarantee at-least-once processing), use the gRPC ConsumeDurableEventsWithAcks RPC instead.

import sseclient
import requests
import json

url = "https://your-consumer-service.example.com/api/v1beta/events/stream-durable"
headers = {
    "Authorization": f"Bearer {YOUR_TOKEN}",
    "Accept": "text/event-stream"
}
params = {"queue_id": queue_id}

response = requests.get(url, headers=headers, params=params, stream=True)
client = sseclient.SSEClient(response)

for event in client.events():
    if event.event == "event":
        data = json.loads(event.data)
        process_event(data)

Reconnecting: Just use the same queue_id — no reconnect token needed:

# After disconnection or application restart
response = requests.get(url, headers=headers, params=params, stream=True)
# Will continue from where you left off

gRPC API

There are two gRPC RPCs for consuming from durable queues:

• ConsumeDurableEventsWithAcks (recommended) — bidirectional streaming with explicit client acknowledgment. Unacknowledged messages are redelivered on reconnect up to a configurable maximum number of attempts, providing at-least-once delivery up to that redelivery bound. A message that exceeds the redelivery limit (or the queue’s configured message TTL or maximum length) is moved to a dead-letter queue rather than redelivered indefinitely. See Bounded Redelivery and the Dead-Letter Queue below.

• ConsumeDurableEvents (deprecated) — server streaming with automatic server-side acknowledgment after a short delay. Has a window for message loss on disconnect.

Using ConsumeDurableEventsWithAcks (recommended):

import grpc
import queue
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2_grpc

# Setup channel and stub
channel = grpc.secure_channel(
    'your-consumer-service.example.com:50052',
    grpc.ssl_channel_credentials()
)
stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

ack_queue = queue.Queue()

def request_stream(queue_id):
    # First request: identify the durable queue
    yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
        queue_id=queue_id
    )
    # Subsequent requests: send acknowledgments
    while True:
        tags = ack_queue.get()
        if tags is None:
            break
        yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
            ack_delivery_tags=tags
        )

try:
    responses = stub.ConsumeDurableEventsWithAcks(
        request_stream(queue_id), metadata=metadata
    )
    
    for response in responses:
        for event in response.events:
            print(f"Event type: {event.event_type}")
            print(f"Message: {event.message}")
            # Process event
        # Acknowledge after processing
        ack_queue.put([response.delivery_tag])
            
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.UNAUTHENTICATED:
        # Refresh token and reconnect with same queue_id
        print("Token expired, reconnecting...")

Using ConsumeDurableEvents (deprecated):

# Consume from durable queue (auto-ack, deprecated)
request = event_consumer_pb2.ConsumeDurableEventsRequest(
    queue_id=queue_id  # Use the saved queue_id
)
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

try:
    responses = stub.ConsumeDurableEvents(request, metadata=metadata)
    
    for response in responses:
        for event in response.events:
            print(f"Event type: {event.event_type}")
            print(f"Message: {event.message}")
            # Process event
            
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.UNAUTHENTICATED:
        # Refresh token and reconnect with same queue_id
        print("Token expired, reconnecting...")

Bounded Redelivery and the Dead-Letter Queue

When you consume from a durable queue with ConsumeDurableEventsWithAcks, the Event Consumer Service redelivers any message you do not acknowledge — for example, after a disconnect or application restart. Redelivery is bounded: the service redelivers an unacknowledged message up to a configurable maximum number of attempts. This gives you at-least-once delivery up to that redelivery bound, not redelivery forever.

A message leaves the durable queue and is moved to a dead-letter queue (DLQ) when any of the following happens:

• It exceeds the configured redelivery limit (for example, a consistently failing “poison” message that is never successfully acknowledged).

• It outlives the queue’s configured message TTL.

• It is evicted when the queue reaches its configured maximum length.

The practical implications for a consumer are:

• A poison message will not block the stream indefinitely. A message your handler can never process is redelivered only a finite number of times and is then dead-lettered, so it stops being redelivered and the rest of the stream continues to flow.

• At-least-once is qualified. Acknowledge each message promptly. If a message is never successfully acknowledged within the redelivery bound, it is dead-lettered rather than redelivered again, so design your handlers to make forward progress within the retry budget.

• The durable queue is bounded. A slow or stalled consumer cannot grow the queue without limit — once the queue reaches its configured maximum length, the oldest overflow messages are dead-lettered rather than retained indefinitely.

The concrete redelivery limit, message TTL, and maximum-length values are configured by the deployment and are not surfaced to consumers through the consuming RPCs. Operators with the appropriate permission can inspect, replay, discard, or leave dead-lettered messages through a dedicated management RPC — see Managing the Dead-Letter Queue (Operators) below.

Managing the Dead-Letter Queue (Operators)

The sections above describe how messages arrive in the dead-letter queue (DLQ). This section describes how an operator inspects and triages them.

Who this is for: This is an operator-facing capability, not a per-consumer one. It is gated by the event-consumer-service:inspect-and-replay-all-dlq permission. A caller that lacks this permission is denied with PERMISSION_DENIED.

gRPC-only: Dead-letter management is exposed only over gRPC via the ManageDeadLetteredEvents RPC. There is no REST/SSE equivalent.

The browse-and-act model

ManageDeadLetteredEvents is a bidirectional streaming RPC. The session works as a live browse-and-act loop:

1. The client opens the stream with an initial (empty) request.

2. The server streams dead-lettered messages to the client, one per dead_lettered_event response, each carrying triage metadata.

3. The client replies with per-message actions — ACTION_REPLAY, ACTION_DISCARD, or ACTION_LEAVE — keyed by the message’s session-scoped delivery_tag.

4. For each action the operator sends, the server streams back an action_outcome response reporting what happened, correlated by the same delivery_tag.

The dead-letter queue is global rather than scoped to a single durable queue, so a single management session surfaces dead-lettered messages from across the deployment.

No datastore required. This bidirectional browse-and-act stream is the permanent baseline DLQ-management surface. It works against the message broker alone and requires no additional datastore. It is the forward-compatible floor for dead-letter management: future additive unary store-backed RPCs (for example, listing dead-lettered events with a server-side cursor, filtering them, replaying a specific event by id, or reporting aggregate statistics) may be added later without changing this RPC. Those additive capabilities are not available today — do not depend on them.

Triage metadata

Each dead_lettered_event carries metadata to support triage:

• reason — why the message was dead-lettered: REASON_DELIVERY_LIMIT (exceeded its redelivery limit), REASON_EXPIRED (outlived the queue’s configured message TTL), or REASON_MAXLEN (evicted when the queue reached its configured maximum length).

• origin_queue — the name of the durable queue the message was dead-lettered from.

• delivery_count — the number of times the broker attempted delivery before dead-lettering.

• dead_lettered_at — the timestamp when the message was dead-lettered.

• event — the dead-lettered event itself, carrying its event_type, principal_identity, and message payload.

• resource_id — the resource id associated with the dead-lettered event.

• delivery_tag — the session-scoped tag used to act on this message (see the caveat below).

Actions: ACTION_REPLAY, ACTION_DISCARD, ACTION_LEAVE

The operator replies with an action per delivery_tag:

• ACTION_REPLAY — republishes the message to its origin queue. Replay is confirm-gated and no-loss: the dead-letter copy is removed only after the publish to the origin queue is confirmed. A replay that cannot be confirmed never drops the message — the dead-letter copy is left in place and the server reports an OUTCOME_REPLAY_FAILED outcome.

• ACTION_DISCARD — acknowledges the message and drops it from the dead-letter queue.

• ACTION_LEAVE — leaves the message in place in the dead-letter queue for later triage.

Action-outcome reports

The server-to-client stream carries two kinds of message, modeled as a response oneof payload with two arms: a dead_lettered_event (a browse item, described above) and an action_outcome (a report). Every action the operator sends produces exactly one correlated action_outcome, keyed by delivery_tag, so the operator can observe the result of every action rather than relying on server-side logs.

An action_outcome carries:

• delivery_tag — the session-scoped tag identifying which dead-lettered message the outcome pertains to (matches the tag the operator sent).

• outcome — one of OUTCOME_REPLAYED, OUTCOME_DISCARDED, OUTCOME_LEFT, OUTCOME_ORIGIN_QUEUE_DELETED, or OUTCOME_REPLAY_FAILED.

• detail — a human-readable note with additional context (for example, the name of the missing origin queue for an OUTCOME_ORIGIN_QUEUE_DELETED outcome).

The outcome values are:

• OUTCOME_REPLAYED — the message was successfully republished to its origin queue and the dead-letter copy was removed after the publish confirm.

• OUTCOME_DISCARDED — the message was acknowledged and dropped from the dead-letter queue.

• OUTCOME_LEFT — the message was left in place in the dead-letter queue for later triage.

• OUTCOME_ORIGIN_QUEUE_DELETED — an ACTION_REPLAY could not proceed because the origin queue no longer exists (see below). This outcome is no-loss.

• OUTCOME_REPLAY_FAILED — an ACTION_REPLAY could not be confirmed. This outcome is no-loss: the dead-letter copy is left in place exactly as for an ACTION_LEAVE, so the message is never dropped on a failed replay.

Origin-queue-deleted handling. A ACTION_REPLAY of a message whose origin queue has been deleted does not lose the message. The replay cannot proceed because there is no viable target, so the server leaves the dead-letter copy in place (exactly the ACTION_LEAVE semantics) and reports an OUTCOME_ORIGIN_QUEUE_DELETED outcome on the stream — the condition is reported to the operator, not silently swallowed. To obtain a viable replay target, re-create the origin queue and replay the message again.

Tenant scope. ACTION_REPLAY returns a message to its exact origin queue, and the origin queue name embeds the hashed principal id, so a replayed message goes back to its original tenant — replay does not cross tenants. The event-consumer-service:inspect-and-replay-all-dlq permission is coarse: it grants cross-tenant dead-letter visibility and action by design; per-principal scoping is deferred. When a replayed message is subsequently re-consumed, it still runs the existing per-event authorization check.

Session delivery tags are live-only. A delivery_tag is valid only within the live stream that produced it. Tags are not persisted and must not be reused across reconnects. If the stream drops, re-open a new management session; the server re-presents the still-dead-lettered messages with fresh tags.

gRPC Python Example

import grpc
import queue
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2_grpc

# Setup channel and stub
channel = grpc.secure_channel(
    'your-consumer-service.example.com:50052',
    grpc.ssl_channel_credentials()
)
stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)
# The caller's token must carry the
# event-consumer-service:inspect-and-replay-all-dlq permission.
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

action_queue = queue.Queue()

def request_stream():
    # First request opens the stream and carries no actions.
    yield event_consumer_pb2.ManageDeadLetteredEventsRequest()
    # Subsequent requests carry per-delivery-tag actions.
    while True:
        entries = action_queue.get()
        if entries is None:
            break
        request = event_consumer_pb2.ManageDeadLetteredEventsRequest()
        request.actions.extend(entries)
        yield request

responses = stub.ManageDeadLetteredEvents(request_stream(), metadata=metadata)

Request = event_consumer_pb2.ManageDeadLetteredEventsRequest
Response = event_consumer_pb2.ManageDeadLetteredEventsResponse

for response in responses:
    # The response oneof has two arms: dead_lettered_event and action_outcome.
    arm = response.WhichOneof('payload')

    if arm == 'dead_lettered_event':
        dle = response.dead_lettered_event
        print(f"Dead-lettered: tag={dle.delivery_tag} "
              f"reason={Response.DeadLetteredEvent.Reason.Name(dle.reason)} "
              f"origin_queue={dle.origin_queue} "
              f"event_type={dle.event.event_type} "
              f"resource_id={dle.resource_id}")

        # Decide what to do and reply, keyed by the session delivery tag.
        entry = Request.ActionEntry(
            delivery_tag=dle.delivery_tag,
            action=Request.ACTION_REPLAY,
        )
        action_queue.put([entry])

    elif arm == 'action_outcome':
        outcome = response.action_outcome
        name = Response.ActionOutcome.Outcome.Name(outcome.outcome)
        print(f"Outcome for tag={outcome.delivery_tag}: {name} "
              f"({outcome.detail})")

        if outcome.outcome == Response.ActionOutcome.Outcome.OUTCOME_ORIGIN_QUEUE_DELETED:
            # The message was NOT lost: the dead-letter copy is left in place.
            # Re-create the origin queue, then replay again.
            print("Origin queue is gone; re-create it before replaying.")

Step 3: Delete Durable Queue

When your application no longer needs the queue (e.g., during decommissioning), delete it to free resources.

REST API

Endpoint: DELETE /api/v1beta/queues/durable

curl -X DELETE "https://your-consumer-service.example.com/api/v1beta/queues/durable?queue_id=durable-queue-abc123xyz" \
  -H "Authorization: Bearer YOUR_TOKEN"

Response:

{
  "success": true
}

gRPC API

RPC: DeleteDurableQueue

request = event_consumer_pb2.DeleteDurableQueueRequest(
    queue_id=queue_id
)
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

response = stub.DeleteDurableQueue(request, metadata=metadata)
print("Queue deleted successfully")

Updating Durable Queue Filters

You can update filters on an existing durable queue without deleting it:

REST API

Endpoint: PATCH /api/v1beta/queues/durable

curl -X PATCH "https://your-consumer-service.example.com/api/v1beta/queues/durable" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "queue_id": "durable-queue-abc123xyz",
    "current_filter_groups": [
      {
        "event_type": "storage.file.created",
        "filters": []
      }
    ],
    "new_filter_groups": [
      {
        "event_type": "storage.file.created",
        "filters": []
      },
      {
        "event_type": "storage.file.deleted",
        "filters": []
      }
    ]
  }'

gRPC API

RPC: UpdateDurableQueue

request = event_consumer_pb2.UpdateDurableQueueRequest(
    queue_id=queue_id
)
for fg in current_filters:
    request.current_filter_groups.append(fg)
for fg in new_filters:
    request.new_filter_groups.append(fg)
metadata = [('authorization', f'Bearer {YOUR_TOKEN}')]

response = stub.UpdateDurableQueue(request, metadata=metadata)

Resource Filtering

Filter Types

Three types of resource filters are available:

1. EQ (Exact Match)

Matches only exact resource_id:

{
    "filter_type": "eq",
    "resource_id": "/projects/project1/file.txt"
}

# Matches: /projects/project1/file.txt
# No match: /projects/project1/file2.txt
# No match: /projects/project1/sub/file.txt

2. STARTS_WITH_LAZY (Shallow Prefix Match)

Matches resources at the same level or one level deep:

{
        "filter_type": "starts_with_lazy",
    "resource_id": "/projects/project1/"
}

# Matches: /projects/project1/file.txt
# No match: /projects/project1/subfolder/file.txt (too deep)
# No match: /projects/project10/file.txt (not a prefix match)

3. STARTS_WITH_GREEDY (Deep Prefix Match)

Matches resources at any depth under the prefix:

{
            "filter_type": "starts_with_greedy",
    "resource_id": "/projects/project1/"
}

# Matches: /projects/project1/file.txt
# Matches: /projects/project1/subfolder/file.txt
# Matches: /projects/project1/a/b/c/d/file.txt
# No match: /projects/project10/file.txt

Empty Resource ID

Special behavior for resource_id="":

# STARTS_WITH_GREEDY with empty resource_id: receive ALL events of this type
{
            "filter_type": "starts_with_greedy",
    "resource_id": ""
}

# STARTS_WITH_LAZY with empty resource_id: receive events with no resource 
# or single-level resources
{
        "filter_type": "starts_with_lazy",
    "resource_id": ""
}

# EQ with empty resource_id: ERROR (not allowed)
{
    "filter_type": "eq",
    "resource_id": ""
}

Multiple Filters (OR Logic)

Multiple filters within a FilterGroup use OR logic:

# Receive events from /uploads/ OR /shared/
{
    "event_type": "storage.file.created",
    "filters": [
        {
            "filter_type": "starts_with_greedy",
            "resource_id": "/uploads/"
        },
        {
            "filter_type": "starts_with_greedy",
            "resource_id": "/shared/"
        }
    ]
}

Multiple Event Types

Multiple FilterGroups let you subscribe to multiple event types:

[
    {
        "event_type": "storage.file.created",
        "filters": [
            {"filter_type": "starts_with_greedy", "resource_id": "/uploads/"}
        ]
    },
    {
        "event_type": "storage.file.deleted",
        "filters": [
            {"filter_type": "starts_with_greedy", "resource_id": "/uploads/"}
        ]
    }
]

Error Handling

Authorization Errors During Streaming

Important: Authorization errors can occur at any time during streaming, not just at connection time.

For example:

• Your authentication token expires while streaming

• Permissions change while you’re connected

When this happens, the stream will terminate with a 401 (UNAUTHENTICATED) or 403 (PERMISSION_DENIED) error.

Handling in REST/SSE:

import time

def consume_with_auto_reconnect(url, get_token_func, queue_id=None):
    """
    Consume events with automatic reconnection on auth errors.
    Messages are acknowledged automatically by the server.
    """
    while True:
        try:
            token = get_token_func()  # Get fresh token
            headers = {
                "Authorization": f"Bearer {token}",
                "Accept": "text/event-stream"
            }
            
            params = {}
            if queue_id:
                params["queue_id"] = queue_id
            
            response = requests.get(url, headers=headers, params=params, stream=True)
            
            if response.status_code == 401 or response.status_code == 403:
                print("Auth error, refreshing token...")
                time.sleep(1)
                continue
            
            client = sseclient.SSEClient(response)

            for event in client.events():
                if event.event == "event":
                    data = json.loads(event.data)
                    process_event(data)
                    
        except requests.exceptions.RequestException as e:
            print(f"Connection error: {e}, reconnecting...")
            time.sleep(5)

Handling in gRPC:

import queue

def consume_with_retry(stub, queue_id, get_token_func):
    """
    Consume with automatic reconnection on auth errors.
    Uses ConsumeDurableEventsWithAcks for explicit acknowledgment.
    """
    while True:
        try:
            token = get_token_func()
            metadata = [('authorization', f'Bearer {token}')]
            ack_queue = queue.Queue()

            def request_stream():
                yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
                    queue_id=queue_id
                )
                while True:
                    tags = ack_queue.get()
                    if tags is None:
                        break
                    yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
                        ack_delivery_tags=tags
                    )

            responses = stub.ConsumeDurableEventsWithAcks(
                request_stream(), metadata=metadata
            )
            
            for response in responses:
                for event in response.events:
                    process_event(event)
                ack_queue.put([response.delivery_tag])
                    
        except grpc.RpcError as e:
            if e.code() in [grpc.StatusCode.UNAUTHENTICATED, 
                          grpc.StatusCode.PERMISSION_DENIED]:
                print("Auth error, refreshing token and reconnecting...")
                time.sleep(1)
            elif e.code() == grpc.StatusCode.RESOURCE_EXHAUSTED:
                print("Service overloaded, backing off...")
                time.sleep(5)
            elif e.code() == grpc.StatusCode.UNAVAILABLE:
                print("Connection lost, reconnecting...")
                time.sleep(1)
            else:
                print(f"Error: {e.code()}: {e.details()}")
                time.sleep(5)

Other Common Errors

HTTP Status / gRPC Code	Meaning	Action
400 / INVALID_ARGUMENT	Invalid filter_groups or queue_id	Check your request format
404 / NOT_FOUND	Queue not found (durable)	Verify queue_id is correct
503 / RESOURCE_EXHAUSTED	Service temporarily overloaded	Retry with exponential backoff
— / UNAVAILABLE	Connection lost (e.g. load balancer idle timeout)	Reconnect automatically

Best Practices

1. Choose the Right Queue Type

See the decision matrix above.

2. Handle Disconnections Gracefully

# Non-durable: Save reconnect_token
reconnect_token = None
for response in responses:
    if response.reconnect_token:
        reconnect_token = response.reconnect_token

# Use it to reconnect
request = ConsumeNonDurableEventsRequest(reconnect_token=reconnect_token)

# Durable: Just use the same queue_id
request = ConsumeDurableEventsWithAcksRequest(queue_id=saved_queue_id)

3. Process Events Idempotently

Events may be delivered more than once (rare but possible) — for durable queues, an unacknowledged message is redelivered up to the configured redelivery limit before it is dead-lettered. Design your event handlers to be idempotent so a redelivered message is safe to process again:

def process_event(event):
    event_id = get_event_id(event)
    
    # Check if already processed
    if redis.exists(f"processed:{event_id}"):
        print(f"Event {event_id} already processed, skipping")
        return
    
    # Process event
    do_work(event)
    
    # Mark as processed
    redis.set(f"processed:{event_id}", "1", ex=86400)  # 24h TTL

4. Don’t Block the Event Stream

Process events quickly or hand them off to a worker:

import threading
import queue

event_queue = queue.Queue()

def event_processor():
    """Worker thread to process events."""
    while True:
        event = event_queue.get()
        if event is None:
            break
        process_event_slowly(event)

# Start worker
worker = threading.Thread(target=event_processor)
worker.start()

# Consume events quickly
for response in responses:
    for event in response.events:
        event_queue.put(event)  # Don't block the stream

5. Monitor Queue Depth (Durable Queues)

If you’re not consuming fast enough, the durable queue will grow. The queue is bounded — it does not grow without limit. Once it reaches its configured maximum length, overflow messages are dead-lettered rather than retained (see Bounded Redelivery and the Dead-Letter Queue), so a sustained backlog can lead to dropped (dead-lettered) messages. Monitor queue depth and scale your consumers before the backlog approaches the maximum length.

6. Delete Durable Queues When Done

Always delete durable queues when your application is decommissioned:

# In your application shutdown handler
def cleanup():
    stub.DeleteDurableQueue(
        DeleteDurableQueueRequest(queue_id=queue_id),
        metadata=metadata
    )

7. Use Appropriate Filters

Don’t receive more events than you need:

# Bad - receiving all events then filtering in code
bad_filter_group = FilterGroup(event_type="storage.file.created")
# No filters - receiving everything
for response in responses:
    for event in response.events:
        if event.message['file_name'].endswith('.pdf'):
            process(event)  # Wasting bandwidth

# Better - filter by resource if possible
better_filter_group = FilterGroup(
    event_type="storage.file.created"
)
resource_filter = ResourceFilter(
    filter_type=FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id="/pdfs/"  # Only get events from /pdfs/
)
better_filter_group.filters.append(resource_filter)

Complete Example: Thumbnail Generation Service

Here’s a complete example of a service that generates thumbnails for uploaded images using a durable queue with explicit acknowledgment:

import grpc
import logging
import queue
import time
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2
from nvidia.omniverse.notifications.consumer.v1beta import event_consumer_pb2_grpc

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ThumbnailService:
    def __init__(self, consumer_endpoint, auth_token):
        self.consumer_endpoint = consumer_endpoint
        self.auth_token = auth_token
        self.queue_id = None
        
    def setup(self):
        """Create durable queue - call once during deployment."""
        channel = grpc.secure_channel(
            self.consumer_endpoint,
            grpc.ssl_channel_credentials()
        )
        stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)
        
        # Create queue for image upload events
        filter_group = event_consumer_pb2.FilterGroup(
            event_type='storage.file.created'
        )
        resource_filter = event_consumer_pb2.ResourceFilter(
            filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
            resource_id='/images/'
        )
        filter_group.filters.append(resource_filter)
        
        request = event_consumer_pb2.CreateDurableQueueRequest()
        request.filter_groups.append(filter_group)
        metadata = [('authorization', f'Bearer {self.auth_token}')]
        
        response = stub.CreateDurableQueue(request, metadata=metadata)
        self.queue_id = response.queue_id
        
        logger.info(f"Created durable queue: {self.queue_id}")
        # SAVE THIS QUEUE_ID TO YOUR CONFIG!

    def _request_stream(self, ack_queue):
        """Generate requests for the bidirectional stream."""
        # First request: identify the durable queue
        yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
            queue_id=self.queue_id
        )
        # Subsequent requests: send acknowledgments
        while True:
            tags = ack_queue.get()
            if tags is None:
                break
            yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
                ack_delivery_tags=tags
            )

    def run(self, queue_id):
        """Run the service - consumes events and generates thumbnails."""
        self.queue_id = queue_id
        
        while True:
            try:
                channel = grpc.secure_channel(
                    self.consumer_endpoint,
                    grpc.ssl_channel_credentials()
                )
                stub = event_consumer_pb2_grpc.EventConsumerServiceStub(channel)
                metadata = [('authorization', f'Bearer {self.auth_token}')]
                ack_queue = queue.Queue()
                
                logger.info("Starting to consume events...")
                responses = stub.ConsumeDurableEventsWithAcks(
                    self._request_stream(ack_queue), metadata=metadata
                )
                
                for response in responses:
                    for event in response.events:
                        self.process_event(event)
                    # Acknowledge after processing
                    ack_queue.put([response.delivery_tag])
                        
            except grpc.RpcError as e:
                logger.error(f"gRPC error: {e.code()}: {e.details()}")
                if e.code() in [grpc.StatusCode.UNAUTHENTICATED, 
                              grpc.StatusCode.PERMISSION_DENIED]:
                    # Refresh token and reconnect
                    self.auth_token = self.get_fresh_token()
                    logger.info("Refreshed auth token, reconnecting...")
                else:
                    logger.error("Connection failed, retrying in 5s...")
                    time.sleep(5)
                    
    def process_event(self, event):
        """Process a single file created event."""
        try:
            file_path = event.message.get('file_path', '')
            logger.info(f"Processing file: {file_path}")
            
            # Generate thumbnail
            self.generate_thumbnail(file_path)
            
            logger.info(f"Thumbnail generated for: {file_path}")
            
        except Exception as e:
            logger.error(f"Error processing event: {e}")
            # Don't raise - continue processing other events
            
    def generate_thumbnail(self, file_path):
        """Generate thumbnail for the given file."""
        # Implementation here
        pass
        
    def get_fresh_token(self):
        """Get a fresh auth token."""
        # Implementation here
        pass

# Usage
if __name__ == "__main__":
    service = ThumbnailService(
        consumer_endpoint="your-consumer-service.example.com:50052",
        auth_token="YOUR_TOKEN"
    )
    
    # One-time setup (or load from config)
    # service.setup()
    
    # Run service with saved queue_id
    service.run(queue_id="saved-queue-id-from-config")

Next Steps

• Understand Permissions & Authorization for consuming events

• Review the API Reference for complete API documentation

• Learn about Publishing Events