For the complete documentation index, see llms.txt. This page is also available as Markdown.

PostgreSQL Message Broker

Reference · Applies to Brighter V10

Brighter supports for using PostgreSQL as a message broker, enabling pub/sub messaging patterns using your existing PostgreSQL infrastructure.

PostgreSQL Message Broker Overview

The PostgreSQL message broker uses a table-based queue approach where messages are stored in a PostgreSQL table and retrieved by consumers. This provides a lightweight messaging solution that leverages your existing PostgreSQL database without requiring additional message broker infrastructure.

How the PostgreSQL Broker Works

  1. Producer: Inserts messages into a queue store table

  2. Consumer: Retrieves messages from the queue store table based on visibility timeout

  3. Acknowledgement: Deletes processed messages from the table

  4. Reject/Requeue: Deletes or updates messages based on processing outcome

The system uses a visibility timeout mechanism (similar to AWS SQS) where messages become invisible to other consumers once retrieved, preventing duplicate processing.


PostgreSQL Message Broker Configuration

NuGet Package

Install the PostgreSQL messaging gateway package:

Database Table

Create the queue store table in your PostgreSQL database:

Index Requirements: The index on (queue, visible_timeout) is critical for performance.


Producer Configuration

Basic Producer Setup

Publishing Messages


Consumer Configuration

Basic Consumer Setup

Consuming Messages


PostgreSQL Message Broker Configuration Options

Three of the four settings below default to null, and a null is not "unset": the schema, the queue store table and the payload format each fall back to the same setting on the relational database configuration the connection carries, and the schema falls back once more to public.

PostgresPublication Options

PostgresPublication takes its options as properties and adds these three to the base publication options, which carry Topic.

Option
Type
Default
Description

SchemaName

string?

null

The schema the queue store table lives in.

QueueStoreTable

string?

null

The table messages are written to.

BinaryMessagePayload

bool?

null

Whether the payload column is written as JSONB rather than JSON.

PostgresSubscription Options

PostgresSubscription takes its options as constructor arguments, so the option is the parameter you type. The seventeen it shares with Subscription behave the same way here; the other seven are PostgreSQL's own.

Option
Type
Default
Description

subscriptionName

SubscriptionName

none

Names the subscription for diagnostics; read back as Name.

channelName

ChannelName

none

Names the queue this subscription reads.

routingKey

RoutingKey

none

The routing key messages are written under.

dataType

Type?

none

The request type messages on this queue are translated into; read back as RequestType.

getRequestType

Func<Message, Type>?

derives the type from dataType

Determines the request type from the message rather than from the queue.

bufferSize

int

1

Messages read from the queue at once and held in the channel.

noOfPerformers

int

1

Threads reading this queue, each with its own message pump.

timeOut

TimeSpan?

300 ms

How long a read waits before treating the queue as empty.

requeueCount

int

-1

Times a message is requeued before it is treated as a poison pill; -1 is unlimited.

requeueDelay

TimeSpan?

0 ms

How long delivery of a requeued message is delayed.

unacceptableMessageLimit

int

0

Unacceptable messages before the channel stops; 0 disables the limit.

unacceptableMessageLimitWindow

TimeSpan?

null

The window the unacceptable-message count resets at the end of.

messagePumpType

MessagePumpType

none

Selects the Reactor or Proactor concurrency model.

channelFactory

IAmAChannelFactory?

null

Creates the channel; falls back to DefaultChannelFactory when null.

makeChannels

OnMissingChannel

Create

Whether Brighter creates the queue store table, validates it, or assumes it.

emptyChannelDelay

TimeSpan?

500 ms

How long the pump pauses after a read that found no message.

channelFailureDelay

TimeSpan?

1000 ms

How long the pump pauses after a channel failure.

schemaName

string?

null

The schema the queue store table lives in.

queueStoreTable

string?

null

The table messages are read from.

visibleTimeout

TimeSpan?

30000 ms

How long a read message stays invisible to other consumers.

tableWithLargeMessage

bool

false

Whether payloads are read as streams to support large messages.

binaryMessagePayload

bool?

null

Whether the payload column is read as JSONB rather than JSON.

deadLetterRoutingKey

RoutingKey?

null

The routing key messages are dead-lettered to.

invalidMessageRoutingKey

RoutingKey?

null

The routing key unacceptable messages are routed to.

The request type parameter is dataType here rather than requestType, which is what every other transport in this documentation calls it, and it is read back as RequestType.

The generic form PostgresSubscription<T>, which every example above uses, takes the same options and supplies four defaults the table cannot: dataType is T, and subscriptionName, channelName and routingKey are T's full name. It leaves messagePumpType required, so state Reactor or Proactor on every subscription.

PostgresMessagingGatewayConnection Options

The connection wraps the relational database configuration rather than adding settings of its own.

Option
Type
Default
Description

configuration

RelationalDatabaseConfiguration

none

The connection string, schema and table names for the queue store; read back as Configuration.

The eight options on that configuration are documented once, at Relational Database Configuration Reference, because seventeen Brighter components share them.


Message Visibility

The PostgreSQL message broker uses a visibility timeout mechanism to prevent duplicate processing:

How Message Visibility Works

  1. Message Published: visible_timeout set to CURRENT_TIMESTAMP

  2. Message Retrieved: Consumer reads messages where visible_timeout <= CURRENT_TIMESTAMP

  3. Processing: Message becomes invisible to other consumers (timeout not updated)

  4. Acknowledged: Message deleted from table

  5. Timeout Expires: If not acknowledged, message becomes visible again

Visibility Timeout Example

Recommendation: Set visibility timeout to 2-3x your expected processing time to account for retries and delays.


Scheduled Messages

PostgreSQL message broker supports message scheduling using the visibility timeout:

How it works: The visible_timeout is set to CURRENT_TIMESTAMP + delay, making the message invisible until the scheduled time.


Transactional Messaging

A key advantage of PostgreSQL as a message broker is transactional messaging with your business data:

Using the Outbox Pattern

See Outbox Pattern and PostgreSQL Outbox for more details.


PostgreSQL Message Broker Monitoring and Observability

Query Queue Depth

Query In-Flight Messages

Find Stuck Messages

OpenTelemetry Integration

PostgreSQL message broker operations are automatically traced when OpenTelemetry is configured:


PostgreSQL Message Broker Best Practices

1. Use JSONB for Production

2. Set Appropriate Visibility Timeout

3. Use Connection Pooling

4. Monitor Queue Depth

Set up alerts for queue depth:

5. Index Your Queue Table

6. Regular Cleanup

Implement cleanup for old messages (if not using auto-vacuum):

7. Use Claim Check for Large Messages

For messages > 100KB, use the Claim Check pattern:

8. Separate Queue Tables for High Volume

For high-volume queues, use dedicated tables:


PostgreSQL Message Broker Troubleshooting

Messages Not Being Consumed

Problem: Messages remain in the queue but are not processed.

Solutions:

  1. Check visibility timeout hasn't expired:

  2. Verify consumer is running and subscriptions match queue names

  3. Check database connection pooling isn't exhausted

  4. Review logs for consumer exceptions

High Database Load

Problem: PostgreSQL CPU/disk usage is high.

Solutions:

  1. Verify index exists on (queue, visible_timeout)

  2. Use JSONB instead of JSON for better performance

  3. Reduce BufferSize if retrieving too many messages at once

  4. Consider partitioning the queue table for high volume

  5. Use connection pooling to reduce connection overhead

Messages Processed Multiple Times

Problem: Same message processed by multiple consumers.

Solutions:

  1. Increase visibleTimeout to allow more processing time

  2. Implement Inbox pattern for idempotency

  3. Check for long-running handlers that exceed visibility timeout

  4. Verify only one consumer process per subscription

Slow Message Retrieval

Problem: Consumer polls are slow.

Solutions:

  1. Add index: CREATE INDEX ON brighter_messages(queue, visible_timeout)

  2. Use JSONB instead of JSON

  3. Increase timeOut to reduce polling frequency

  4. Consider using bufferSize > 1 to retrieve multiple messages per poll


Further Reading

Last updated

Was this helpful?