Skip to content

Commit e1c93cc

Browse files
danielmarbachdanielmarbach
authored
Non durable persistence updates (#8385)
* Non durable persistence updates * Additional lookup scenario --------- Co-authored-by: Daniel Marbach <danielmarbach@users.noreply.github.com>
1 parent ab2c6f5 commit e1c93cc

7 files changed

Lines changed: 305 additions & 2 deletions

File tree

‎Snippets/NonDurablePersistence/NonDurablePersistence_3/MultiEndpointHosting.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
namespace NonDurablePersistence_3
2+
{
3+
using Microsoft.Extensions.DependencyInjection;
4+
using NServiceBus;
5+
using NServiceBus.Persistence.NonDurable;
6+
7+
class MultiEndpointHosting
8+
{
9+
void Configure(IServiceCollection services)
10+
{
11+
#region NonDurableMultiEndpointStorage
12+
13+
services.AddSingleton<NonDurableStorage>();
14+
15+
#endregion
16+
}
17+
}
18+
}

‎Snippets/NonDurablePersistence/NonDurablePersistence_3/SagaFinderWithPersister.cs‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
namespace NonDurablePersistence_3
2+
{
3+
using System;
4+
using System.Collections.Concurrent;
5+
using System.Threading;
6+
using System.Threading.Tasks;
7+
using NServiceBus;
8+
using NServiceBus.Extensibility;
9+
using NServiceBus.Persistence;
10+
using NServiceBus.Sagas;
11+
12+
#region NonDurableSagaFinderWithPersister
13+
14+
class TaskIndex
15+
{
16+
public ConcurrentDictionary<Guid, Guid> ServerTaskIdToSagaId { get; } = new();
17+
}
18+
19+
class TaskSagaFinder(TaskIndex index, ISagaPersister persister)
20+
: ISagaFinder<TaskSagaData, ContinueTask>
21+
{
22+
public async Task<TaskSagaData> FindBy(ContinueTask message,
23+
ISynchronizedStorageSession storageSession, IReadOnlyContextBag context,
24+
CancellationToken cancellationToken = default)
25+
{
26+
if (!index.ServerTaskIdToSagaId.TryGetValue(message.ServerTaskId, out var sagaId))
27+
{
28+
return null;
29+
}
30+
31+
return await persister.Get<TaskSagaData>(sagaId, storageSession, (ContextBag)context, cancellationToken);
32+
}
33+
}
34+
35+
class ContinueTask : IMessage
36+
{
37+
public Guid ServerTaskId { get; set; }
38+
}
39+
40+
class TaskSagaData : ContainSagaData
41+
{
42+
public Guid ServerTaskId { get; set; }
43+
}
44+
45+
#endregion
46+
}

‎Snippets/NonDurablePersistence/NonDurablePersistence_3/SagaProjection.cs‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
namespace NonDurablePersistence_3
2+
{
3+
using System.Threading;
4+
using System.Threading.Tasks;
5+
using NServiceBus;
6+
using NServiceBus.Extensibility;
7+
using NServiceBus.Persistence;
8+
using NServiceBus.Persistence.NonDurable;
9+
using NServiceBus.Sagas;
10+
11+
#region NonDurableSagaProjection
12+
13+
class OrderSagaFinder :
14+
ISagaFinder<OrderSagaData, CompleteOrder>
15+
{
16+
public Task<OrderSagaData> FindBy(CompleteOrder message, ISynchronizedStorageSession session, IReadOnlyContextBag context, CancellationToken cancellationToken = default)
17+
{
18+
var nonDurableSession = session.NonDurablePersistenceSession();
19+
20+
var sagaData = nonDurableSession.GetSagaData<OrderSagaData>(
21+
context,
22+
data => data.OrderId == message.OrderId,
23+
cancellationToken);
24+
25+
return Task.FromResult(sagaData);
26+
}
27+
}
28+
29+
class OrderSagaFinderWithState :
30+
ISagaFinder<OrderSagaData, CompleteOrder>
31+
{
32+
public Task<OrderSagaData> FindBy(CompleteOrder message, ISynchronizedStorageSession session, IReadOnlyContextBag context, CancellationToken cancellationToken = default)
33+
{
34+
var nonDurableSession = session.NonDurablePersistenceSession();
35+
36+
var sagaData = nonDurableSession.GetSagaData<OrderSagaData, string>(
37+
context,
38+
message.OrderId,
39+
(data, orderId) => data.OrderId == orderId,
40+
cancellationToken);
41+
42+
return Task.FromResult(sagaData);
43+
}
44+
}
45+
46+
class CompleteOrder : IMessage
47+
{
48+
public string OrderId { get; set; } = string.Empty;
49+
}
50+
51+
class OrderSagaData : ContainSagaData
52+
{
53+
public string OrderId { get; set; } = string.Empty;
54+
}
55+
56+
#endregion
57+
}

‎Snippets/NonDurablePersistence/NonDurablePersistence_3/Usage.cs‎

Lines changed: 83 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
1-
namespace NonDurablePersistence_3
1+
namespace NonDurablePersistence_3
22
{
3+
using System.Text.Json;
4+
using System.Text.Json.Serialization;
35
using NServiceBus;
6+
using NServiceBus.Persistence.NonDurable;
47

58
class Usage
69
{
@@ -14,5 +17,84 @@ public Usage(EndpointConfiguration endpointConfiguration)
1417

1518
#endregion
1619
}
20+
21+
void ConfigureWithShortcut(EndpointConfiguration endpointConfiguration)
22+
{
23+
#region ConfiguringNonDurableShortcut
24+
25+
endpointConfiguration.UseNonDurablePersistence();
26+
27+
#endregion
28+
}
29+
30+
void ConfigureSharedStorage(EndpointConfiguration endpointConfiguration)
31+
{
32+
#region ConfiguringNonDurableSharedStorage
33+
34+
var sharedStorage = new NonDurableStorage();
35+
36+
var options = new NonDurablePersistenceOptions
37+
{
38+
Storage = sharedStorage
39+
};
40+
41+
endpointConfiguration.UseNonDurablePersistence(options);
42+
43+
#endregion
44+
}
45+
46+
void ConfigureWithOptions(EndpointConfiguration endpointConfiguration)
47+
{
48+
#region ConfiguringNonDurableOptions
49+
50+
var sharedStorage = new NonDurableStorage();
51+
52+
var options = new NonDurablePersistenceOptions
53+
{
54+
Storage = sharedStorage,
55+
TimeProvider = System.TimeProvider.System,
56+
Saga = new NonDurableSagaOptions
57+
{
58+
JsonSerializerOptions = new JsonSerializerOptions
59+
{
60+
TypeInfoResolverChain = { new SagaJsonContext() }
61+
}
62+
}
63+
};
64+
65+
endpointConfiguration.UseNonDurablePersistence(options);
66+
67+
#endregion
68+
}
69+
70+
void ConfigureSagaSerialization(EndpointConfiguration endpointConfiguration)
71+
{
72+
#region ConfiguringNonDurableSagaSerialization
73+
74+
var options = new NonDurablePersistenceOptions
75+
{
76+
Saga = new NonDurableSagaOptions
77+
{
78+
JsonSerializerOptions = new JsonSerializerOptions
79+
{
80+
TypeInfoResolverChain = { new SagaJsonContext() }
81+
}
82+
}
83+
};
84+
85+
endpointConfiguration.UseNonDurablePersistence(options);
86+
87+
#endregion
88+
}
89+
}
90+
91+
[JsonSerializable(typeof(MySagaData))]
92+
partial class SagaJsonContext : JsonSerializerContext
93+
{
94+
}
95+
96+
class MySagaData : ContainSagaData
97+
{
98+
public string OrderId { get; set; } = string.Empty;
1799
}
18100
}

‎persistence/non-durable/index.md‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
title: Non-durable persistence
33
summary: Non-durable persistence (previously known as In-Memory persistence) stores data in a non-durable manner
44
component: NonDurablePersistence
5-
reviewed: 2024-12-26
5+
reviewed: 2026-07-01
66
redirects:
77
- nservicebus/persistence/in-memory
88
---
@@ -25,15 +25,21 @@ For a description of each feature, see the [persistence at a glance legend](/per
2525

2626
## Configuration
2727

28+
Configure the endpoint to use non-durable persistence:
29+
2830
snippet: ConfiguringNonDurable
2931

32+
partial: configuration-shortcut
33+
3034
> [!CAUTION]
3135
> All information stored is discarded when the process ends.
3236
3337
partial: timeoutmanager
3438

3539
partial: gatewaydedupe
3640

41+
partial: extended
42+
3743
## Saga concurrency
3844

3945
When simultaneously handling messages, conflicts may occur. See below for examples of the exceptions which are thrown. _[Saga concurrency](/nservicebus/sagas/concurrency.md)_ explains how these conflicts are handled, and contains guidance for high-load scenarios.
@@ -51,3 +57,7 @@ System.InvalidOperationException: The saga with the correlation id 'Name: OrderI
5157
Non-durable persistence uses [optimistic concurrency control](https://en.wikipedia.org/wiki/Optimistic_concurrency_control) when updating or deleting saga data.
5258

5359
Example exception:
60+
61+
```
62+
System.InvalidOperationException: Saga with Id '7ac53d15-4742-4e38-9e2f-6d75c25b6621' can't be updated because it was updated by another process.
63+
```

‎persistence/non-durable/index_configuration-shortcut_nondurablepersistence_[3,).partial.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
Alternatively, use the non-durable-specific shorthand:
2+
3+
snippet: ConfiguringNonDurableShortcut
4+
5+
> [!NOTE]
6+
> The `UseNonDurablePersistence()` shorthand is unique to this persister. Other persistences use only the standard `UsePersistence<T>()` pattern.

‎persistence/non-durable/index_extended_nondurablepersistence_[3,).partial.md‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
## Advanced configuration
2+
3+
For scenarios that require additional control, use `UseNonDurablePersistence(NonDurablePersistenceOptions)`:
4+
5+
snippet: ConfiguringNonDurableOptions
6+
7+
### Shared storage
8+
9+
When multiple endpoints need to share the same in-memory state, provide a `NonDurableStorage` instance through the options:
10+
11+
snippet: ConfiguringNonDurableSharedStorage
12+
13+
Storage resolution follows this precedence:
14+
15+
1. A `NonDurableStorage` resolved from dependency injection
16+
2. The `Storage` property set on `NonDurablePersistenceOptions`
17+
3. A default shared storage instance
18+
19+
When using the [generic host or multi-endpoint hosting](/nservicebus/hosting/), register `NonDurableStorage` in the service collection. The persister automatically resolves it from dependency injection before falling back to the options or default storage:
20+
21+
snippet: NonDurableMultiEndpointStorage
22+
23+
### Time provider
24+
25+
Supply a custom `System.TimeProvider` to control how timestamps and outbox entry expiry are calculated. This is useful for testing scenarios that need deterministic time behavior.
26+
27+
### Saga serialization
28+
29+
Saga data is the only persistence state that is JSON-serialized. By default, `System.Text.Json` is used with reflection. For AOT-compatible deployments or trimmed applications, provide a source-generated serializer context:
30+
31+
snippet: ConfiguringNonDurableSagaSerialization
32+
33+
## Custom saga finders
34+
35+
Non-durable persistence supports [custom saga finders](/nservicebus/sagas/saga-finding.md) via `ISagaFinder<TSagaData, TMessage>`.
36+
37+
## Saga data projection
38+
39+
The non-durable persistence exposes `INonDurableStorageSession` to query saga data directly from within a custom saga finder. This is useful when correlation logic is too complex to express through the standard saga mapping API.
40+
41+
snippet: NonDurableSagaProjection
42+
43+
The query is evaluated against a moment-in-time snapshot of the underlying storage. Entries added or removed concurrently may or may not be included. The returned saga data is a copy of the stored entry, and optimistic concurrency checks still apply if the saga is later updated or completed.
44+
45+
For unit testing, use `TestableNonDurableSynchronizedStorageSession` to create a fake session backed by an in-memory store.
46+
47+
### Custom index with ISagaPersister
48+
49+
When maintaining a custom lookup index outside of the persister, resolve the saga ID from the index and delegate to `ISagaPersister.Get` to load the saga data. This still captures the saga entry for optimistic concurrency checks:
50+
51+
snippet: NonDurableSagaFinderWithPersister
52+
53+
## OpenTelemetry instrumentation
54+
55+
Non-durable persistence emits spans via the `NServiceBus.Persistence.NonDurable` activity source when an OpenTelemetry listener is configured.
56+
57+
### Saga spans
58+
59+
| Span name | Description |
60+
|:---|:---|
61+
| `NServiceBus.NonDurable.Persistence.Saga.GetById` | Loading a saga by its identifier |
62+
| `NServiceBus.NonDurable.Persistence.Saga.GetByProperty` | Loading a saga by a correlated property |
63+
| `NServiceBus.NonDurable.Persistence.Saga.Save` | Saving a new saga instance |
64+
| `NServiceBus.NonDurable.Persistence.Saga.Update` | Updating an existing saga instance |
65+
| `NServiceBus.NonDurable.Persistence.Saga.Complete` | Completing a saga instance |
66+
67+
### Outbox spans
68+
69+
| Span name | Description |
70+
|:---|:---|
71+
| `NServiceBus.NonDurable.Persistence.Outbox.BeginTransaction` | Beginning an outbox transaction |
72+
| `NServiceBus.NonDurable.Persistence.Outbox.Get` | Retrieving an outbox record |
73+
| `NServiceBus.NonDurable.Persistence.Outbox.Store` | Storing transport operations in the outbox |
74+
| `NServiceBus.NonDurable.Persistence.Outbox.SetAsDispatched` | Marking an outbox record as dispatched |
75+
76+
### Subscription spans
77+
78+
| Span name | Description |
79+
|:---|:---|
80+
| `NServiceBus.NonDurable.Persistence.Subscription.Subscribe` | Subscribing to a message type |
81+
| `NServiceBus.NonDurable.Persistence.Subscription.Unsubscribe` | Unsubscribing from a message type |
82+
| `NServiceBus.NonDurable.Persistence.Subscription.GetSubscribers` | Resolving subscribers for a message type |
83+
84+
See the [OpenTelemetry documentation](/nservicebus/operations/opentelemetry.md) for instructions on how to enable tracing in an endpoint.

0 commit comments

Comments
 (0)