|
| 1 | +# PostgreSQL transport provider for SlimMessageBus <!-- omit in toc --> |
| 2 | + |
| 3 | +Please read the [Introduction](intro.md) before reading this provider documentation. |
| 4 | + |
| 5 | +- [About](#about) |
| 6 | +- [Configuration](#configuration) |
| 7 | + - [Provider settings](#provider-settings) |
| 8 | + - [Queues, topics, and request/response](#queues-topics-and-requestresponse) |
| 9 | + - [Message id generation](#message-id-generation) |
| 10 | +- [How it works](#how-it-works) |
| 11 | + - [Polling and locking](#polling-and-locking) |
| 12 | + - [Retries and failed messages](#retries-and-failed-messages) |
| 13 | + - [Schema provisioning](#schema-provisioning) |
| 14 | +- [Testing locally](#testing-locally) |
| 15 | + |
| 16 | +## About |
| 17 | + |
| 18 | +The PostgreSQL transport provider allows a shared PostgreSQL database to act as the message broker for collaborating producers and consumers. |
| 19 | + |
| 20 | +This transport is useful for applications that already operate PostgreSQL and do not need a dedicated messaging broker yet. |
| 21 | + |
| 22 | +## Configuration |
| 23 | + |
| 24 | +Install the transport package: |
| 25 | + |
| 26 | +```bash |
| 27 | +dotnet add package SlimMessageBus.Host.PostgreSql |
| 28 | +``` |
| 29 | + |
| 30 | +The configuration is arranged via the `.WithProviderPostgreSql(cfg => {})` method on the message bus builder. |
| 31 | + |
| 32 | +```cs |
| 33 | +using SlimMessageBus.Host.PostgreSql; |
| 34 | + |
| 35 | +services.AddSlimMessageBus(mbb => |
| 36 | +{ |
| 37 | + mbb.WithProviderPostgreSql(cfg => |
| 38 | + { |
| 39 | + cfg.ConnectionString = "..."; |
| 40 | + cfg.DatabaseSchemaName = "smb"; |
| 41 | + cfg.DatabaseTableName = "messages"; |
| 42 | + cfg.PollDelay = TimeSpan.FromMilliseconds(250); |
| 43 | + cfg.PollBatchSize = 10; |
| 44 | + cfg.LockDuration = TimeSpan.FromSeconds(30); |
| 45 | + cfg.MaxDeliveryAttempts = 10; |
| 46 | + }); |
| 47 | + |
| 48 | + mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue")); |
| 49 | + mbb.Consume<PingMessage>(x => x.Queue("ping-queue")); |
| 50 | + |
| 51 | + mbb.Produce<OrderSubmitted>(x => x.DefaultTopic("orders").ToTopic()); |
| 52 | + mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "billing")); |
| 53 | + mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "shipping")); |
| 54 | + |
| 55 | + mbb.AddServicesFromAssemblyContaining<PingConsumer>(); |
| 56 | + mbb.AddJsonSerializer(); |
| 57 | +}); |
| 58 | +``` |
| 59 | + |
| 60 | +### Provider settings |
| 61 | + |
| 62 | +The most commonly configured settings are: |
| 63 | + |
| 64 | +- `ConnectionString` - required PostgreSQL connection string. |
| 65 | +- `DatabaseSchemaName` - schema containing the transport tables. Defaults to `public`. |
| 66 | +- `DatabaseTableName` - base message table name. The durable subscription table uses the same base name with a `_subscriptions` suffix. |
| 67 | +- `DatabaseMigrationsTableName` - table used to track transport schema migrations. |
| 68 | +- `CommandTimeout` - optional PostgreSQL command timeout. |
| 69 | +- `TransactionIsolationLevel` - isolation level used for schema provisioning. Defaults to `ReadCommitted`. |
| 70 | +- `PollDelay` - delay used when no message is available or after a transient polling error. |
| 71 | +- `PollBatchSize` - maximum number of messages locked by one polling operation. |
| 72 | +- `LockDuration` - how long a message lock is held before another consumer may pick it up. |
| 73 | +- `MaxDeliveryAttempts` - number of processing attempts before a message is marked aborted. |
| 74 | +- `NotifyOnPublish` - when enabled, producers call `pg_notify` after inserting messages. Polling remains the correctness mechanism. |
| 75 | +- `SchemaCreationRetry` and `OperationRetry` - retry settings for schema creation and regular database operations. |
| 76 | + |
| 77 | +PostgreSQL identifiers configured through `DatabaseSchemaName`, `DatabaseTableName`, and `DatabaseMigrationsTableName` are validated and quoted. Use letters, numbers, and underscores, and do not start identifiers with a number. |
| 78 | + |
| 79 | +### Queues, topics, and request/response |
| 80 | + |
| 81 | +Use `DefaultQueue()` and `Queue()` for competing-consumer queues: |
| 82 | + |
| 83 | +```cs |
| 84 | +mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue")); |
| 85 | +mbb.Consume<PingMessage>(x => x.Queue("ping-queue")); |
| 86 | +``` |
| 87 | + |
| 88 | +Use `DefaultTopic().ToTopic()` and `Topic(topic, subscriptionName)` for durable pub/sub: |
| 89 | + |
| 90 | +```cs |
| 91 | +mbb.Produce<OrderSubmitted>(x => x.DefaultTopic("orders").ToTopic()); |
| 92 | +mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "billing")); |
| 93 | +mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "shipping")); |
| 94 | +``` |
| 95 | + |
| 96 | +Request/response endpoints can also use PostgreSQL queues or topics: |
| 97 | + |
| 98 | +```cs |
| 99 | +mbb.Handle<PingRequest, PingResponse>(x => x.Queue("ping-handler")); |
| 100 | +mbb.ExpectRequestResponses(x => x.ReplyToQueue("replies")); |
| 101 | +``` |
| 102 | + |
| 103 | +### Message id generation |
| 104 | + |
| 105 | +The transport stores messages with two identifiers: |
| 106 | + |
| 107 | +- `sequence_id` - a `bigserial` physical key used for insert locality and ordered polling. |
| 108 | +- `id` - a logical `uuid` message id used by the transport when completing or failing messages. |
| 109 | + |
| 110 | +By default, PostgreSQL uses `PostgreSqlMessageIdGenerationMode.ClientGuidGenerator` with `PostgreSqlSequentialGuidGenerator`, which creates sequential-ish UUIDs client-side for better index locality than random UUIDs. |
| 111 | + |
| 112 | +You can change the id strategy: |
| 113 | + |
| 114 | +```cs |
| 115 | +mbb.WithProviderPostgreSql(cfg => |
| 116 | +{ |
| 117 | + cfg.ConnectionString = "..."; |
| 118 | + cfg.IdGeneration.Mode = PostgreSqlMessageIdGenerationMode.DatabaseRandomUuid; |
| 119 | +}); |
| 120 | +``` |
| 121 | + |
| 122 | +Available modes: |
| 123 | + |
| 124 | +- `ClientGuidGenerator` - client-side UUID generation. Defaults to `PostgreSqlSequentialGuidGenerator`, but `GuidGenerator` or `GuidGeneratorType` can be replaced. |
| 125 | +- `DatabaseRandomUuid` - uses PostgreSQL `gen_random_uuid()`. |
| 126 | + |
| 127 | +## How it works |
| 128 | + |
| 129 | +- A messages table stores exchanged messages. |
| 130 | +- A subscriptions table stores durable topic subscriptions configured by consumers. |
| 131 | +- Queue consumers compete for rows using `FOR UPDATE SKIP LOCKED`. |
| 132 | +- Topic publishes create one row per configured subscription. |
| 133 | +- Message rows use a `bigserial` physical key for insert locality and a logical `uuid` message id. |
| 134 | +- The default client-side id generator is sequential-ish for index locality. Random database ids can be selected through `cfg.IdGeneration`. |
| 135 | +- Producers optionally call `pg_notify` after inserting messages. Polling remains the correctness mechanism. |
| 136 | + |
| 137 | +### Polling and locking |
| 138 | + |
| 139 | +Consumers poll the shared message table in batches. PostgreSQL uses `FOR UPDATE SKIP LOCKED` so competing consumers can skip rows already locked by another instance. |
| 140 | + |
| 141 | +When a consumer locks a row, the transport stores the consumer instance id and lock expiration. If the process stops before completing the message, the row becomes visible again after `LockDuration`. |
| 142 | + |
| 143 | +`pg_notify` is used as a lightweight wake-up hint when `NotifyOnPublish` is enabled. It is not used as the durable delivery mechanism; message rows in the database remain the source of truth. |
| 144 | + |
| 145 | +### Retries and failed messages |
| 146 | + |
| 147 | +Successful processing marks the row as complete. Failed processing increments `delivery_attempt`, clears the lock, and makes the row available for another attempt. Once `MaxDeliveryAttempts` is reached, the row is marked aborted and will no longer be delivered. |
| 148 | + |
| 149 | +The transport retries transient PostgreSQL errors around schema provisioning and operations according to `SchemaCreationRetry` and `OperationRetry`. |
| 150 | + |
| 151 | +### Schema provisioning |
| 152 | + |
| 153 | +The provider provisions the required message, subscription, and migration tables during bus startup. All cooperating services should use the same database, schema, and table names. |
| 154 | + |
| 155 | +## Testing locally |
| 156 | + |
| 157 | +The integration tests use Testcontainers and require Docker to be running: |
| 158 | + |
| 159 | +```bash |
| 160 | +dotnet test src/Tests/SlimMessageBus.Host.PostgreSql.Test/SlimMessageBus.Host.PostgreSql.Test.csproj --filter "Category=Integration" |
| 161 | +``` |
| 162 | + |
| 163 | +The repository also contains `infrastructure.ps1` for standing up shared development infrastructure used by broader integration test runs. |
0 commit comments