Skip to content

Commit 75837b7

Browse files
zaruszzarusz
committed
Document SQL and PostgreSQL transports
Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>
1 parent 02a9f17 commit 75837b7

3 files changed

Lines changed: 241 additions & 20 deletions

File tree

‎docs/README.md‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,12 @@
88
- [Azure ServiceBus](provider_azure_servicebus.md)
99
- [Hybrid](provider_hybrid.md)
1010
- [MQTT](provider_mqtt.md)
11-
- [Memory](provider_memory.md)
12-
- [NATS](provider_nats.md)
13-
- [RabbitMQ](provider_rabbitmq.md)
14-
- [Redis](provider_redis.md)
15-
- [SQL](provider_sql.md)
11+
- [Memory](provider_memory.md)
12+
- [NATS](provider_nats.md)
13+
- [PostgreSQL](provider_postgresql.md)
14+
- [RabbitMQ](provider_rabbitmq.md)
15+
- [Redis](provider_redis.md)
16+
- [SQL](provider_sql.md)
1617
- Plugins
1718
- [Serialization](serialization.md)
1819
- [Transactional Outbox](plugin_outbox.md)

‎docs/provider_postgresql.md‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,14 @@ Please read the [Introduction](intro.md) before reading this provider documentat
44

55
- [About](#about)
66
- [Configuration](#configuration)
7+
- [Provider settings](#provider-settings)
8+
- [Queues, topics, and request/response](#queues-topics-and-requestresponse)
9+
- [Message id generation](#message-id-generation)
710
- [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)
815

916
## About
1017

@@ -14,9 +21,17 @@ This transport is useful for applications that already operate PostgreSQL and do
1421

1522
## Configuration
1623

24+
Install the transport package:
25+
26+
```bash
27+
dotnet add package SlimMessageBus.Host.PostgreSql
28+
```
29+
1730
The configuration is arranged via the `.WithProviderPostgreSql(cfg => {})` method on the message bus builder.
1831

1932
```cs
33+
using SlimMessageBus.Host.PostgreSql;
34+
2035
services.AddSlimMessageBus(mbb =>
2136
{
2237
mbb.WithProviderPostgreSql(cfg =>
@@ -26,6 +41,8 @@ services.AddSlimMessageBus(mbb =>
2641
cfg.DatabaseTableName = "messages";
2742
cfg.PollDelay = TimeSpan.FromMilliseconds(250);
2843
cfg.PollBatchSize = 10;
44+
cfg.LockDuration = TimeSpan.FromSeconds(30);
45+
cfg.MaxDeliveryAttempts = 10;
2946
});
3047

3148
mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue"));
@@ -40,6 +57,73 @@ services.AddSlimMessageBus(mbb =>
4057
});
4158
```
4259

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+
43127
## How it works
44128

45129
- A messages table stores exchanged messages.
@@ -49,3 +133,31 @@ services.AddSlimMessageBus(mbb =>
49133
- Message rows use a `bigserial` physical key for insert locality and a logical `uuid` message id.
50134
- The default client-side id generator is sequential-ish for index locality. Random database ids can be selected through `cfg.IdGeneration`.
51135
- 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.

‎docs/provider_sql.md‎

Lines changed: 123 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,20 @@
22

33
Please read the [Introduction](intro.md) before reading this provider documentation.
44

5-
- [About](#about)
6-
- [SQL Compatibility](#sql-compatibility)
7-
- [Configuration](#configuration)
8-
- [How it works](#how-it-works)
9-
10-
## About
11-
5+
- [About](#about)
6+
- [SQL Compatibility](#sql-compatibility)
7+
- [Configuration](#configuration)
8+
- [Provider settings](#provider-settings)
9+
- [Queues, topics, and request/response](#queues-topics-and-requestresponse)
10+
- [Message id generation](#message-id-generation)
11+
- [How it works](#how-it-works)
12+
- [Polling and locking](#polling-and-locking)
13+
- [Retries and failed messages](#retries-and-failed-messages)
14+
- [Schema provisioning](#schema-provisioning)
15+
- [Testing locally](#testing-locally)
16+
17+
## About
18+
1219
The SQL transport provider allows to leverage a single shared SQL database instance as a messaging broker for all the collaborating producers and consumers.
1320

1421
This transport might be optimal for simpler applications that do not have a dedicated messaging infrastructure available, do not have high throughput needs, or want to target a simplistic deployment model.
@@ -18,12 +25,20 @@ When the application grows over time, and given that SMB is an abstraction, the
1825
## SQL Compatibility
1926

2027
This transport targets SQL Server / Azure SQL (T-SQL).
21-
22-
## Configuration
23-
28+
29+
## Configuration
30+
31+
Install the transport package:
32+
33+
```bash
34+
dotnet add package SlimMessageBus.Host.Sql
35+
```
36+
2437
The configuration is arranged via the `.WithProviderSql(cfg => {})` method on the message bus builder.
2538

2639
```cs
40+
using SlimMessageBus.Host.Sql;
41+
2742
services.AddSlimMessageBus(mbb =>
2843
{
2944
mbb.WithProviderSql(cfg =>
@@ -33,6 +48,8 @@ services.AddSlimMessageBus(mbb =>
3348
cfg.DatabaseTableName = "Messages";
3449
cfg.PollDelay = TimeSpan.FromMilliseconds(250);
3550
cfg.PollBatchSize = 10;
51+
cfg.LockDuration = TimeSpan.FromSeconds(30);
52+
cfg.MaxDeliveryAttempts = 10;
3653
});
3754

3855
mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue"));
@@ -46,13 +63,78 @@ services.AddSlimMessageBus(mbb =>
4663
mbb.AddJsonSerializer();
4764
});
4865
```
49-
50-
## How it works
51-
52-
The same SQL database instance is required for all the producers and consumers to collaborate.
66+
67+
### Provider settings
68+
69+
The most commonly configured settings are:
70+
71+
- `ConnectionString` - required SQL Server or Azure SQL connection string.
72+
- `DatabaseSchemaName` - schema containing the transport tables. Defaults to `dbo`.
73+
- `DatabaseTableName` - base message table name, for example `Messages`. The durable subscription table uses the same base name with a `Subscriptions` suffix.
74+
- `DatabaseMigrationsTableName` - table used to track transport schema migrations.
75+
- `CommandTimeout` - optional SQL command timeout.
76+
- `TransactionIsolationLevel` - isolation level used by transport transactions. Defaults to `ReadCommitted`.
77+
- `PollDelay` - delay used when no message is available or after a transient polling error.
78+
- `PollBatchSize` - maximum number of messages locked by one polling operation.
79+
- `LockDuration` - how long a message lock is held before another consumer may pick it up.
80+
- `MaxDeliveryAttempts` - number of processing attempts before a message is marked aborted.
81+
- `SchemaCreationRetry` and `OperationRetry` - retry settings for schema creation and regular database operations.
82+
83+
### Queues, topics, and request/response
84+
85+
Use `DefaultQueue()` and `Queue()` for competing-consumer queues:
86+
87+
```cs
88+
mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue"));
89+
mbb.Consume<PingMessage>(x => x.Queue("ping-queue"));
90+
```
91+
92+
Use `DefaultTopic().ToTopic()` and `Topic(topic, subscriptionName)` for durable pub/sub:
93+
94+
```cs
95+
mbb.Produce<OrderSubmitted>(x => x.DefaultTopic("orders").ToTopic());
96+
mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "billing"));
97+
mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "shipping"));
98+
```
99+
100+
Request/response endpoints can also use SQL queues or topics:
101+
102+
```cs
103+
mbb.Handle<PingRequest, PingResponse>(x => x.Queue("ping-handler"));
104+
mbb.ExpectRequestResponses(x => x.ReplyToQueue("replies"));
105+
```
106+
107+
### Message id generation
108+
109+
The transport stores messages with two identifiers:
110+
111+
- `SequenceId` - a clustered `bigint identity` physical key used for insert locality and ordered polling.
112+
- `Id` - a logical `uniqueidentifier` message id used by the transport when completing or failing messages.
113+
114+
By default, SQL uses `SqlMessageIdGenerationMode.ClientGuidGenerator` with `SqlSequentialGuidGenerator`, which creates sequential-ish GUIDs client-side for better index locality than random GUIDs.
115+
116+
You can change the id strategy:
117+
118+
```cs
119+
mbb.WithProviderSql(cfg =>
120+
{
121+
cfg.ConnectionString = "...";
122+
cfg.IdGeneration.Mode = SqlMessageIdGenerationMode.DatabaseGeneratedSequentialGuid;
123+
});
124+
```
125+
126+
Available modes:
127+
128+
- `ClientGuidGenerator` - client-side GUID generation. Defaults to `SqlSequentialGuidGenerator`, but `GuidGenerator` or `GuidGeneratorType` can be replaced.
129+
- `DatabaseGeneratedGuid` - uses `NEWID()`.
130+
- `DatabaseGeneratedSequentialGuid` - uses the table default, which is intended for SQL Server sequential GUID generation.
131+
132+
## How it works
133+
134+
The same SQL database instance is required for all the producers and consumers to collaborate.
53135
Therefore ensure all of the service instances point to the same database cluster.
54136

55-
- A messages table is used to store exchanged messages (by default table is called `Messages`).
137+
- A messages table is used to store exchanged messages.
56138
- A subscriptions table is used to store durable topic subscriptions configured by consumers.
57139
- Producers send messages to the messages table.
58140
- There are two types of entities (queues, and topics for pub/sub).
@@ -66,3 +148,29 @@ Therefore ensure all of the service instances point to the same database cluster
66148
- In the future we might consider:
67149
- Table per each entity, so that we can minimize table locking.
68150
- Sessions to ensure order of processing within the same message session ID - similar to how Azure Service Bus feature or Apache Kafka topic-partition works.
151+
152+
### Polling and locking
153+
154+
Consumers poll the shared message table in batches. SQL Server locking hints (`ROWLOCK`, `UPDLOCK`, `READPAST`) are used so competing consumers can skip rows already locked by another instance.
155+
156+
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`.
157+
158+
### Retries and failed messages
159+
160+
Successful processing marks the row as complete. Failed processing increments `DeliveryAttempt`, 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.
161+
162+
The transport retries transient SQL errors around schema provisioning and operations according to `SchemaCreationRetry` and `OperationRetry`.
163+
164+
### Schema provisioning
165+
166+
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.
167+
168+
## Testing locally
169+
170+
The integration tests use Testcontainers and require Docker to be running:
171+
172+
```bash
173+
dotnet test src/Tests/SlimMessageBus.Host.Sql.Test/SlimMessageBus.Host.Sql.Test.csproj --filter "Category=Integration"
174+
```
175+
176+
The repository also contains `infrastructure.ps1` for standing up shared development infrastructure used by broader integration test runs.

0 commit comments

Comments
 (0)