Skip to content

Update KV per key TTL #831

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mtmk merged 2 commits into from
Apr 24, 2025
Merged

Update KV per key TTL #831

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7087d75
Update KV per key TTL
mtmk Apr 16, 2025
4626989
Fix format
mtmk Apr 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions src/NATS.Client.JetStream/Models/ApiStats.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,14 @@ namespace NATS.Client.JetStream.Models;

public record ApiStats
{
/// <summary>
/// Supported API level.
/// </summary>
[System.Text.Json.Serialization.JsonPropertyName("level")]
[System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.Never)]
[System.ComponentModel.DataAnnotations.Range(0, int.MaxValue)]
public int Level { get; set; }

/// <summary>
/// Total number of API requests received for this account
/// </summary>
Expand Down
82 changes: 24 additions & 58 deletions src/NATS.Client.KeyValueStore/INatsKVStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,18 +26,6 @@ public interface INatsKVStore
/// <returns>Revision number</returns>
ValueTask<ulong> PutAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Put a value into the bucket using the key
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
/// <returns>Revision number</returns>
ValueTask<ulong> PutAsync<T>(string key, T value, TimeSpan ttl, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

Comment on lines -29 to -40

@stebet stebet Apr 23, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since Put is no longer supported, does that mean to update a TTLd key one should delete and recreate it.

@mtmk mtmk Apr 23, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think so though not sure how that goes with delete markers etc. let me ask around.

@mtmk mtmk Apr 24, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok the proper solution is in the pipeline but for a workaround we purge with a TTL (to keep bucket size maintained) then recreate with new TTL. also for caching we probably want a 1-history buket.

@stebet stebet Apr 24, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll take a look at the IDistributedCache implementation I have to take that into account :)

@mtmk mtmk Apr 24, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

released in 2.6.0-preview.5

/// <summary>
/// Tries to put a value into the bucket using the key
/// </summary>
Expand All @@ -52,21 +40,6 @@ public interface INatsKVStore
/// </remarks>
ValueTask<NatsResult<ulong>> TryPutAsync<T>(string key, T value, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Tries to put a value into the bucket using the key
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
/// <returns>Revision number</returns>
/// <remarks>
/// Use this method to avoid exceptions
/// </remarks>
ValueTask<NatsResult<ulong>> TryPutAsync<T>(string key, T value, TimeSpan ttl, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Create a new entry in the bucket only if it doesn't exist
/// </summary>
Expand All @@ -83,7 +56,7 @@ public interface INatsKVStore
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.LimitMarkerTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
Expand All @@ -109,7 +82,7 @@ public interface INatsKVStore
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.LimitMarkerTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
Expand All @@ -131,19 +104,6 @@ public interface INatsKVStore
/// <returns>The revision number of the updated entry</returns>
ValueTask<ulong> UpdateAsync<T>(string key, T value, ulong revision, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Update an entry in the bucket only if last update revision matches
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="revision">Last revision number to match</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
/// <returns>The revision number of the updated entry</returns>
ValueTask<ulong> UpdateAsync<T>(string key, T value, ulong revision, TimeSpan ttl, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Tries to update an entry in the bucket only if last update revision matches
/// </summary>
Expand All @@ -159,22 +119,6 @@ public interface INatsKVStore
/// </remarks>
ValueTask<NatsResult<ulong>> TryUpdateAsync<T>(string key, T value, ulong revision, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Tries to update an entry in the bucket only if last update revision matches
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="value">Value of the entry</param>
/// <param name="revision">Last revision number to match</param>
/// <param name="ttl">Time to live for the entry (requires the <see cref="NatsKVConfig.AllowMsgTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="serializer">Serializer to use for the message type.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <typeparam name="T">Serialized value type</typeparam>
/// <returns>A NatsResult object representing the revision number of the updated entry or an error.</returns>
/// <remarks>
/// Use this method to avoid exceptions
/// </remarks>
ValueTask<NatsResult<ulong>> TryUpdateAsync<T>(string key, T value, ulong revision, TimeSpan ttl, INatsSerialize<T>? serializer = default, CancellationToken cancellationToken = default);

/// <summary>
/// Delete an entry from the bucket
/// </summary>
Expand Down Expand Up @@ -203,6 +147,15 @@ public interface INatsKVStore
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
ValueTask PurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);

/// <summary>
/// Purge an entry from the bucket
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="ttl">Time to live for the purge marker (requires the <see cref="NatsKVConfig.LimitMarkerTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="opts">Delete options</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
ValueTask PurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);

/// <summary>
/// Tries to purge an entry from the bucket
/// </summary>
Expand All @@ -215,6 +168,19 @@ public interface INatsKVStore
/// </remarks>
ValueTask<NatsResult> TryPurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);

/// <summary>
/// Tries to purge an entry from the bucket
/// </summary>
/// <param name="key">Key of the entry</param>
/// <param name="ttl">Time to live for the purge marker (requires the <see cref="NatsKVConfig.LimitMarkerTTL"/> to be set to true). For a key that should never expire, use the <see cref="TimeSpan.MaxValue"/> constant. This feature is only available on NATS server v2.11 and later.</param>
/// <param name="opts">Delete options</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
/// <returns>A NatsResult object representing success or an error.</returns>
/// <remarks>
/// Use this method to avoid exceptions
/// </remarks>
ValueTask<NatsResult> TryPurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default);

/// <summary>
/// Get an entry from the bucket using the key
/// </summary>
Expand Down
10 changes: 2 additions & 8 deletions src/NATS.Client.KeyValueStore/NatsKVConfig.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,16 +84,10 @@ public record NatsKVConfig
public IDictionary<string, string>? Metadata { get; set; }

/// <summary>
/// If true, the bucket will allow TTL on individual keys.
/// How long the bucket keeps markers when keys are removed by the TTL setting, 0 meaning markers are not supported.
/// </summary>
/// <remarks>This feature is only available on NATS server v2.11 and later.</remarks>
public bool AllowMsgTTL { get; set; }

/// <summary>
/// Enables and sets a duration for adding server markers for delete, purge and max age limits.
/// </summary>
/// <remarks>This feature is only available on NATS server v2.11 and later.</remarks>
public TimeSpan SubjectDeleteMarkerTTL { get; set; }
public TimeSpan LimitMarkerTTL { get; init; }
}

/// <summary>
Expand Down
21 changes: 18 additions & 3 deletions src/NATS.Client.KeyValueStore/NatsKVContext.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,20 @@ public async ValueTask<INatsKVStore> CreateStoreAsync(NatsKVConfig config, Cance
{
ValidateBucketName(config.Bucket);

if (config.LimitMarkerTTL < TimeSpan.Zero || (config.LimitMarkerTTL > TimeSpan.Zero && config.LimitMarkerTTL < TimeSpan.FromSeconds(1)))
{
throw new NatsKVException("Invalid LimitMarkerTTL");
}

if (config.LimitMarkerTTL > TimeSpan.Zero)
{
var info = await JetStreamContext.JSRequestResponseAsync<object, AccountInfoResponse>("$JS.API.INFO", null, cancellationToken);
if (info.Api.Level < 1)
{
throw new NatsKVException("API doesn't support LimitMarkerTTL");
}
}

var streamConfig = NatsKVContext.CreateStreamConfig(config);

var stream = await JetStreamContext.CreateStreamAsync(streamConfig, cancellationToken);
Expand Down Expand Up @@ -133,7 +147,8 @@ public async IAsyncEnumerable<NatsKVStatus> GetStatusesAsync([EnumeratorCancella
{
var stream = await JetStreamContext.GetStreamAsync(name, cancellationToken: cancellationToken);
var isCompressed = stream.Info.Config.Compression != StreamConfigCompression.None;
yield return new NatsKVStatus(name, isCompressed, stream.Info);

yield return new NatsKVStatus(name, isCompressed, NatsKVStore.GetLimitMarkerTTL(stream.Info.Config), stream.Info);
}
}

Expand Down Expand Up @@ -264,8 +279,8 @@ private static StreamConfig CreateStreamConfig(NatsKVConfig config)
Sources = sources,
Retention = StreamConfigRetention.Limits, // from ADR-8
Metadata = config.Metadata,
AllowMsgTTL = config.AllowMsgTTL,
SubjectDeleteMarkerTTL = config.SubjectDeleteMarkerTTL,
AllowMsgTTL = config.LimitMarkerTTL > TimeSpan.Zero,
SubjectDeleteMarkerTTL = config.LimitMarkerTTL,
};

return streamConfig;
Expand Down
Loading