diff --git a/src/NATS.Client.JetStream/Models/ApiStats.cs b/src/NATS.Client.JetStream/Models/ApiStats.cs index 261272e91..e555e67b9 100644 --- a/src/NATS.Client.JetStream/Models/ApiStats.cs +++ b/src/NATS.Client.JetStream/Models/ApiStats.cs @@ -2,6 +2,14 @@ namespace NATS.Client.JetStream.Models; public record ApiStats { + /// + /// Supported API level. + /// + [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; } + /// /// Total number of API requests received for this account /// diff --git a/src/NATS.Client.KeyValueStore/INatsKVStore.cs b/src/NATS.Client.KeyValueStore/INatsKVStore.cs index f98d3ec9d..566406c6c 100644 --- a/src/NATS.Client.KeyValueStore/INatsKVStore.cs +++ b/src/NATS.Client.KeyValueStore/INatsKVStore.cs @@ -26,18 +26,6 @@ public interface INatsKVStore /// Revision number ValueTask PutAsync(string key, T value, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// - /// Put a value into the bucket using the key - /// - /// Key of the entry - /// Value of the entry - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. - /// Serializer to use for the message type. - /// A used to cancel the API call. - /// Serialized value type - /// Revision number - ValueTask PutAsync(string key, T value, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// /// Tries to put a value into the bucket using the key /// @@ -52,21 +40,6 @@ public interface INatsKVStore /// ValueTask> TryPutAsync(string key, T value, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// - /// Tries to put a value into the bucket using the key - /// - /// Key of the entry - /// Value of the entry - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. - /// Serializer to use for the message type. - /// A used to cancel the API call. - /// Serialized value type - /// Revision number - /// - /// Use this method to avoid exceptions - /// - ValueTask> TryPutAsync(string key, T value, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// /// Create a new entry in the bucket only if it doesn't exist /// @@ -83,7 +56,7 @@ public interface INatsKVStore /// /// Key of the entry /// Value of the entry - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. + /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. /// Serializer to use for the message type. /// A used to cancel the API call. /// Serialized value type @@ -109,7 +82,7 @@ public interface INatsKVStore /// /// Key of the entry /// Value of the entry - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. + /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. /// Serializer to use for the message type. /// A used to cancel the API call. /// Serialized value type @@ -131,19 +104,6 @@ public interface INatsKVStore /// The revision number of the updated entry ValueTask UpdateAsync(string key, T value, ulong revision, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// - /// Update an entry in the bucket only if last update revision matches - /// - /// Key of the entry - /// Value of the entry - /// Last revision number to match - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. - /// Serializer to use for the message type. - /// A used to cancel the API call. - /// Serialized value type - /// The revision number of the updated entry - ValueTask UpdateAsync(string key, T value, ulong revision, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// /// Tries to update an entry in the bucket only if last update revision matches /// @@ -159,22 +119,6 @@ public interface INatsKVStore /// ValueTask> TryUpdateAsync(string key, T value, ulong revision, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// - /// Tries to update an entry in the bucket only if last update revision matches - /// - /// Key of the entry - /// Value of the entry - /// Last revision number to match - /// Time to live for the entry (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. - /// Serializer to use for the message type. - /// A used to cancel the API call. - /// Serialized value type - /// A NatsResult object representing the revision number of the updated entry or an error. - /// - /// Use this method to avoid exceptions - /// - ValueTask> TryUpdateAsync(string key, T value, ulong revision, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default); - /// /// Delete an entry from the bucket /// @@ -203,6 +147,15 @@ public interface INatsKVStore /// A used to cancel the API call. ValueTask PurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default); + /// + /// Purge an entry from the bucket + /// + /// Key of the entry + /// Time to live for the purge marker (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. + /// Delete options + /// A used to cancel the API call. + ValueTask PurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default); + /// /// Tries to purge an entry from the bucket /// @@ -215,6 +168,19 @@ public interface INatsKVStore /// ValueTask TryPurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default); + /// + /// Tries to purge an entry from the bucket + /// + /// Key of the entry + /// Time to live for the purge marker (requires the to be set to true). For a key that should never expire, use the constant. This feature is only available on NATS server v2.11 and later. + /// Delete options + /// A used to cancel the API call. + /// A NatsResult object representing success or an error. + /// + /// Use this method to avoid exceptions + /// + ValueTask TryPurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default); + /// /// Get an entry from the bucket using the key /// diff --git a/src/NATS.Client.KeyValueStore/NatsKVConfig.cs b/src/NATS.Client.KeyValueStore/NatsKVConfig.cs index 9cbb47508..61aac595a 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVConfig.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVConfig.cs @@ -84,16 +84,10 @@ public record NatsKVConfig public IDictionary? Metadata { get; set; } /// - /// 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. /// /// This feature is only available on NATS server v2.11 and later. - public bool AllowMsgTTL { get; set; } - - /// - /// Enables and sets a duration for adding server markers for delete, purge and max age limits. - /// - /// This feature is only available on NATS server v2.11 and later. - public TimeSpan SubjectDeleteMarkerTTL { get; set; } + public TimeSpan LimitMarkerTTL { get; init; } } /// diff --git a/src/NATS.Client.KeyValueStore/NatsKVContext.cs b/src/NATS.Client.KeyValueStore/NatsKVContext.cs index dd31b76b0..83712287d 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVContext.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVContext.cs @@ -58,6 +58,20 @@ public async ValueTask 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("$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); @@ -133,7 +147,8 @@ public async IAsyncEnumerable 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); } } @@ -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; diff --git a/src/NATS.Client.KeyValueStore/NatsKVStore.cs b/src/NATS.Client.KeyValueStore/NatsKVStore.cs index cf1cd0618..ae9b5bd33 100644 --- a/src/NATS.Client.KeyValueStore/NatsKVStore.cs +++ b/src/NATS.Client.KeyValueStore/NatsKVStore.cs @@ -44,7 +44,7 @@ public class NatsKVStore : INatsKVStore private const string NatsSubject = "Nats-Subject"; private const string NatsSequence = "Nats-Sequence"; private const string NatsTimeStamp = "Nats-Time-Stamp"; - private const string NatsTtl = "Nats-TTL"; + private const string NatsTTL = "Nats-TTL"; private static readonly Regex ValidKeyRegex = new(pattern: @"\A[-/_=\.a-zA-Z0-9]+\z", RegexOptions.Compiled); private static readonly NatsKVException MissingSequenceHeaderException = new("Missing sequence header"); private static readonly NatsKVException MissingTimestampHeaderException = new("Missing timestamp header"); @@ -57,10 +57,12 @@ public class NatsKVStore : INatsKVStore private static readonly NatsKVException KeyCannotBeEmptyException = new("Key cannot be empty"); private static readonly NatsKVException KeyCannotStartOrEndWithPeriodException = new("Key cannot start or end with a period"); private static readonly NatsKVException KeyContainsInvalidCharactersException = new("Key contains invalid characters"); + private static readonly NatsKVException ThisStoreDoesNotSupportTTLException = new("This store does not support TTL"); private readonly INatsJSStream _stream; private readonly NatsKVOpts _opts; private readonly string _kvBucket; private readonly string _streamName; + private bool _supportsTTL; internal NatsKVStore(string bucket, INatsJSContext context, INatsJSStream stream, NatsKVOpts opts) { @@ -70,6 +72,7 @@ internal NatsKVStore(string bucket, INatsJSContext context, INatsJSStream stream _opts = opts; _kvBucket = $"$KV.{Bucket}."; _streamName = NatsKVContext.KvStreamNamePrefix + Bucket; + _supportsTTL = GetLimitMarkerTTL(stream.Info.Config) > TimeSpan.Zero; } /// @@ -124,7 +127,7 @@ public async ValueTask> TryPutAsync(string key, T value, Ti { headers = new NatsHeaders { - { NatsTtl, ttl == TimeSpan.MaxValue ? "never" : $"{(int)ttl.TotalSeconds:D}" }, + { NatsTTL, ToTTLString(ttl) }, }; } @@ -165,11 +168,17 @@ public async ValueTask CreateAsync(string key, T value, TimeSpan ttl = } /// - public ValueTask> TryCreateAsync(string key, T value, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) => TryCreateAsync(key, value, default, serializer, cancellationToken); + public ValueTask> TryCreateAsync(string key, T value, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) + => TryCreateAsync(key, value, TimeSpan.Zero, serializer, cancellationToken); /// - public async ValueTask> TryCreateAsync(string key, T value, TimeSpan ttl = default, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) + public async ValueTask> TryCreateAsync(string key, T value, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) { + if (ttl > TimeSpan.Zero && !_supportsTTL) + { + return ThisStoreDoesNotSupportTTLException; + } + var keyValidResult = TryValidateKey(key); if (!keyValidResult.Success) { @@ -177,7 +186,7 @@ public async ValueTask> TryCreateAsync(string key, T value, } // First try to create a new entry - var resultUpdate = await TryUpdateAsync(key, value, revision: 0, ttl, serializer, cancellationToken); + var resultUpdate = await TryUpdateInternalAsync(key, value, revision: 0, ttl, serializer, cancellationToken); if (resultUpdate.Success) { return resultUpdate; @@ -194,7 +203,7 @@ public async ValueTask> TryCreateAsync(string key, T value, else if (resultReadExisting.Error is NatsKVKeyDeletedException deletedException) { // If our previous call errored because the last entry is deleted, then that's ok, we update with the deleted revision - return await TryUpdateAsync(key, value, deletedException.Revision, ttl, serializer, cancellationToken); + return await TryUpdateInternalAsync(key, value, deletedException.Revision, ttl, serializer, cancellationToken); } else { @@ -208,7 +217,7 @@ public async ValueTask> TryCreateAsync(string key, T value, /// public async ValueTask UpdateAsync(string key, T value, ulong revision, TimeSpan ttl = default, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) { - var result = await TryUpdateAsync(key, value, revision, ttl, serializer, cancellationToken); + var result = await TryUpdateInternalAsync(key, value, revision, ttl, serializer, cancellationToken); if (!result.Success) { ThrowException(result.Error); @@ -218,47 +227,8 @@ public async ValueTask UpdateAsync(string key, T value, ulong revision } /// - public ValueTask> TryUpdateAsync(string key, T value, ulong revision, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) => TryUpdateAsync(key, value, revision, default, serializer, cancellationToken); - - /// - public async ValueTask> TryUpdateAsync(string key, T value, ulong revision, TimeSpan ttl = default, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) - { - var keyValidResult = TryValidateKey(key); - if (!keyValidResult.Success) - { - return keyValidResult.Error; - } - - var headers = new NatsHeaders { { NatsExpectedLastSubjectSequence, revision.ToString() } }; - if (ttl != default) - { - headers.Add(NatsTtl, ttl == TimeSpan.MaxValue ? "never" : $"{(int)ttl.TotalSeconds:D}"); - } - - var publishResult = await JetStreamContext.TryPublishAsync(_kvBucket + key, value, headers: headers, serializer: serializer, cancellationToken: cancellationToken); - if (publishResult.Success) - { - var ack = publishResult.Value; - if (ack.Error is { ErrCode: 10071, Code: 400, Description: not null } && ack.Error.Description.StartsWith("wrong last sequence", StringComparison.OrdinalIgnoreCase)) - { - return new NatsKVWrongLastRevisionException(); - } - else if (ack.Error != null) - { - return new NatsJSApiException(ack.Error); - } - else if (ack.Duplicate) - { - return new NatsJSDuplicateMessageException(ack.Seq); - } - - return ack.Seq; - } - else - { - return publishResult.Error; - } - } + public ValueTask> TryUpdateAsync(string key, T value, ulong revision, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) + => TryUpdateInternalAsync(key, value, revision, TimeSpan.Zero, serializer, cancellationToken); /// public async ValueTask DeleteAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) @@ -271,66 +241,31 @@ public async ValueTask DeleteAsync(string key, NatsKVDeleteOpts? opts = default, } /// - public async ValueTask TryDeleteAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) - { - var keyValidResult = TryValidateKey(key); - if (!keyValidResult.Success) - { - return keyValidResult.Error; - } - - opts ??= new NatsKVDeleteOpts(); - - var headers = new NatsHeaders(); + public ValueTask TryDeleteAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) + => TryDeleteInternalAsync(key, TimeSpan.Zero, opts, cancellationToken); - if (opts.Purge) - { - headers.Add(KVOperation, OperationPurge); - headers.Add(NatsRollup, RollupSub); - } - else - { - headers.Add(KVOperation, OperationDel); - } - - if (opts.Revision != default) - { - headers.Add(NatsExpectedLastSubjectSequence, opts.Revision.ToString()); - } - - var publishResult = await JetStreamContext.TryPublishAsync(_kvBucket + key, null, headers: headers, cancellationToken: cancellationToken); - if (publishResult.Success) - { - var ack = publishResult.Value; - if (ack.Error is { ErrCode: 10071, Code: 400, Description: not null } && ack.Error.Description.StartsWith("wrong last sequence", StringComparison.OrdinalIgnoreCase)) - { - return new NatsKVWrongLastRevisionException(); - } - else if (ack.Error != null) - { - return new NatsJSApiException(ack.Error); - } - else if (ack.Duplicate) - { - return new NatsJSDuplicateMessageException(ack.Seq); - } + /// + public ValueTask PurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) => + DeleteAsync(key, (opts ?? new NatsKVDeleteOpts()) with { Purge = true }, cancellationToken); - return NatsResult.Default; - } - else + /// + public async ValueTask PurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) + { + var result = await TryDeleteInternalAsync(key, ttl, (opts ?? new NatsKVDeleteOpts()) with { Purge = true }, cancellationToken); + if (!result.Success) { - return publishResult.Error; + ThrowException(result.Error); } } - /// - public ValueTask PurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) => - DeleteAsync(key, (opts ?? new NatsKVDeleteOpts()) with { Purge = true }, cancellationToken); - /// public ValueTask TryPurgeAsync(string key, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) => TryDeleteAsync(key, (opts ?? new NatsKVDeleteOpts()) with { Purge = true }, cancellationToken); + /// + public ValueTask TryPurgeAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) => + TryDeleteInternalAsync(key, ttl, (opts ?? new NatsKVDeleteOpts()) with { Purge = true }, cancellationToken); + /// public async ValueTask> GetEntryAsync(string key, ulong revision = default, INatsDeserialize? serializer = default, CancellationToken cancellationToken = default) { @@ -562,7 +497,9 @@ public async ValueTask GetStatusAsync(CancellationToken cancellati { await _stream.RefreshAsync(cancellationToken); var isCompressed = _stream.Info.Config.Compression != StreamConfigCompression.None; - return new NatsKVStatus(Bucket, isCompressed, _stream.Info); + var limitMarkerTTL = GetLimitMarkerTTL(_stream.Info.Config); + _supportsTTL = limitMarkerTTL > TimeSpan.Zero; + return new NatsKVStatus(Bucket, isCompressed, limitMarkerTTL, _stream.Info); } /// @@ -640,6 +577,16 @@ public async IAsyncEnumerable GetKeysAsync(IEnumerable filters, } } + internal static TimeSpan GetLimitMarkerTTL(StreamConfig config) + { + if (config.AllowMsgTTL) + { + return config.SubjectDeleteMarkerTTL; + } + + return TimeSpan.Zero; + } + internal async ValueTask> WatchInternalAsync(IEnumerable keys, INatsDeserialize? serializer = default, NatsKVWatchOpts? opts = default, CancellationToken cancellationToken = default) { opts ??= NatsKVWatchOpts.Default; @@ -661,6 +608,14 @@ internal async ValueTask> WatchInternalAsync(IEnumerable + /// For the TTL header, we need to convert the TimeSpan to a Go time.ParseDuration string. + /// + /// TTL + /// String representing the number of seconds Go time.ParseDuration() can understand. + private static string ToTTLString(TimeSpan ttl) + => ttl == TimeSpan.MaxValue ? "never" : $"{(int)ttl.TotalSeconds:D}s"; + /// /// Valid keys are \A[-/_=\.a-zA-Z0-9]+\z, additionally they may not start or end in . /// @@ -686,6 +641,106 @@ private static NatsResult TryValidateKey(string key) [MethodImpl(MethodImplOptions.NoInlining)] private static void ThrowException(Exception exception) => throw exception; + + private async ValueTask> TryUpdateInternalAsync(string key, T value, ulong revision, TimeSpan ttl, INatsSerialize? serializer = default, CancellationToken cancellationToken = default) + { + var keyValidResult = TryValidateKey(key); + if (!keyValidResult.Success) + { + return keyValidResult.Error; + } + + var headers = new NatsHeaders { { NatsExpectedLastSubjectSequence, revision.ToString() } }; + if (ttl > TimeSpan.Zero) + { + headers.Add(NatsTTL, ToTTLString(ttl)); + } + + var publishResult = await JetStreamContext.TryPublishAsync(_kvBucket + key, value, headers: headers, serializer: serializer, cancellationToken: cancellationToken); + if (publishResult.Success) + { + var ack = publishResult.Value; + if (ack.Error is { ErrCode: 10071, Code: 400, Description: not null } && ack.Error.Description.StartsWith("wrong last sequence", StringComparison.OrdinalIgnoreCase)) + { + return new NatsKVWrongLastRevisionException(); + } + else if (ack.Error != null) + { + return new NatsJSApiException(ack.Error); + } + else if (ack.Duplicate) + { + return new NatsJSDuplicateMessageException(ack.Seq); + } + + return ack.Seq; + } + else + { + return publishResult.Error; + } + } + + private async ValueTask TryDeleteInternalAsync(string key, TimeSpan ttl, NatsKVDeleteOpts? opts = default, CancellationToken cancellationToken = default) + { + var keyValidResult = TryValidateKey(key); + if (!keyValidResult.Success) + { + return keyValidResult.Error; + } + + opts ??= new NatsKVDeleteOpts(); + + var headers = new NatsHeaders(); + + if (opts.Purge) + { + headers.Add(KVOperation, OperationPurge); + headers.Add(NatsRollup, RollupSub); + if (ttl > TimeSpan.Zero) + { + if (!_supportsTTL) + { + return ThisStoreDoesNotSupportTTLException; + } + + headers.Add(NatsTTL, ToTTLString(ttl)); + } + } + else + { + headers.Add(KVOperation, OperationDel); + } + + if (opts.Revision != default) + { + headers.Add(NatsExpectedLastSubjectSequence, opts.Revision.ToString()); + } + + var publishResult = await JetStreamContext.TryPublishAsync(_kvBucket + key, null, headers: headers, cancellationToken: cancellationToken); + if (publishResult.Success) + { + var ack = publishResult.Value; + if (ack.Error is { ErrCode: 10071, Code: 400, Description: not null } && ack.Error.Description.StartsWith("wrong last sequence", StringComparison.OrdinalIgnoreCase)) + { + return new NatsKVWrongLastRevisionException(); + } + else if (ack.Error != null) + { + return new NatsJSApiException(ack.Error); + } + else if (ack.Duplicate) + { + return new NatsJSDuplicateMessageException(ack.Seq); + } + + return NatsResult.Default; + } + else + { + return publishResult.Error; + } + } } -public record NatsKVStatus(string Bucket, bool IsCompressed, StreamInfo Info); +public record NatsKVStatus(string Bucket, bool IsCompressed, TimeSpan LimitMarkerTTL, StreamInfo Info); diff --git a/tests/NATS.Client.KeyValueStore.Tests/KeyValueStoreTest.cs b/tests/NATS.Client.KeyValueStore.Tests/KeyValueStoreTest.cs index a6620de43..c78e71305 100644 --- a/tests/NATS.Client.KeyValueStore.Tests/KeyValueStoreTest.cs +++ b/tests/NATS.Client.KeyValueStore.Tests/KeyValueStoreTest.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using NATS.Client.Core.Tests; using NATS.Client.JetStream.Models; using NATS.Client.Platform.Windows.Tests; @@ -461,6 +462,23 @@ await Assert.ThrowsAsync(async () => } } + [SkipIfNatsServer(versionLaterThan: "2.11")] + public async Task TestMessageTTLApiNotSupportedupport() + { + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + await using var server = await NatsServerProcess.StartAsync(); + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + + var js = new NatsJSContext(nats); + var kv = new NatsKVContext(js); + + // Config validation + var exception = await Assert.ThrowsAsync(() => kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(10) }, cancellationToken: cancellationToken).AsTask()); + _output.WriteLine(exception.Message); + } + [SkipIfNatsServer(versionEarlierThan: "2.11")] public async Task TestMessageTTL() { @@ -473,24 +491,52 @@ public async Task TestMessageTTL() var js = new NatsJSContext(nats); var kv = new NatsKVContext(js); - var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { AllowMsgTTL = true }, cancellationToken: cancellationToken); + // Check TTL support + var exception = await Assert.ThrowsAsync(async () => + { + var store1 = await kv.CreateStoreAsync(new NatsKVConfig("kv0") { LimitMarkerTTL = TimeSpan.Zero }, cancellationToken: cancellationToken); + await store1.CreateAsync("k1", "v1", ttl: TimeSpan.FromSeconds(60), cancellationToken: cancellationToken); + }); + Assert.Equal("This store does not support TTL", exception.Message); + + // Check API version + var info = await js.JSRequestResponseAsync("$JS.API.INFO", null, cancellationToken); + Assert.True(info.Api.Level >= 1); + + // Config validation + await Assert.ThrowsAsync(() => kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(-1) }, cancellationToken: cancellationToken).AsTask()); + await Assert.ThrowsAsync(() => kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(.99) }, cancellationToken: cancellationToken).AsTask()); + + var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(2) }, cancellationToken: cancellationToken); for (var i = 0; i < 10; i++) { - await store.PutAsync($"k{i}", $"v{i}", TimeSpan.FromSeconds(1), cancellationToken: cancellationToken); + await store.CreateAsync($"k{i}", $"v{i}", TimeSpan.FromSeconds(2), cancellationToken: cancellationToken); } - var state = await store.GetStatusAsync(); + var state = await store.GetStatusAsync(cancellationToken); Assert.Equal(10, state.Info.State.Messages); Assert.Equal(1ul, state.Info.State.FirstSeq); Assert.Equal(10ul, state.Info.State.LastSeq); - // Sleep for two seconds, now all the messages should be gone - await Task.Delay(2000); - state = await store.GetStatusAsync(); + await Retry.Until( + reason: "messages are deleted", + condition: async () => + { + var state1 = await store.GetStatusAsync(cancellationToken); + _output.WriteLine($"Messages: {state1.Info.State.Messages}"); + _output.WriteLine($"FirstSeq: {state1.Info.State.FirstSeq}"); + _output.WriteLine($"LastSeq: {state1.Info.State.LastSeq}"); + + return state1.Info.State is { Messages: 0, FirstSeq: 21, LastSeq: 20 }; + }, + retryDelay: TimeSpan.FromSeconds(2), + timeout: TimeSpan.FromSeconds(30)); + + state = await store.GetStatusAsync(cancellationToken); Assert.Equal(0, state.Info.State.Messages); - Assert.Equal(11ul, state.Info.State.FirstSeq); - Assert.Equal(10ul, state.Info.State.LastSeq); + Assert.Equal(21ul, state.Info.State.FirstSeq); + Assert.Equal(20ul, state.Info.State.LastSeq); } [SkipIfNatsServer(versionEarlierThan: "2.11")] @@ -505,9 +551,9 @@ public async Task TestTTLMessageWhenTTLDisabledOnStream() var js = new NatsJSContext(nats); var kv = new NatsKVContext(js); - var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { AllowMsgTTL = false }, cancellationToken: cancellationToken); - var exception = await Assert.ThrowsAsync(async () => await store.PutAsync($"somekey", $"somevalue", TimeSpan.FromSeconds(1), cancellationToken: cancellationToken)); - Assert.Equal("per-message TTL is disabled", exception.Message); + var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.Zero }, cancellationToken: cancellationToken); + var exception = await Assert.ThrowsAsync(async () => await store.CreateAsync($"somekey", $"somevalue", TimeSpan.FromSeconds(1), cancellationToken: cancellationToken)); + Assert.Equal("This store does not support TTL", exception.Message); } [SkipIfNatsServer(versionEarlierThan: "2.11")] @@ -522,7 +568,7 @@ public async Task SetsSubjectDeleteMarkerTTL() var js = new NatsJSContext(nats); var kv = new NatsKVContext(js); - var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { AllowMsgTTL = true, SubjectDeleteMarkerTTL = TimeSpan.FromSeconds(2) }, cancellationToken: cancellationToken); + var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(2) }, cancellationToken: cancellationToken); var info = await js.GetStreamAsync("KV_kv1"); Assert.Equal(TimeSpan.FromSeconds(2), info.Info.Config.SubjectDeleteMarkerTTL); } @@ -542,8 +588,7 @@ public async Task SubjectDeleteMarkerTTL_enabled_removals_should_be_interpreted_ var store = await kv.CreateStoreAsync( new NatsKVConfig("kv1") { - AllowMsgTTL = true, - SubjectDeleteMarkerTTL = TimeSpan.FromHours(1), + LimitMarkerTTL = TimeSpan.FromHours(1), MaxAge = TimeSpan.FromSeconds(4), }, cancellationToken: cancellationToken); @@ -589,27 +634,37 @@ public async Task TestMessageNeverExpire() var js = new NatsJSContext(nats); var kv = new NatsKVContext(js); - var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { AllowMsgTTL = true, MaxAge = TimeSpan.FromSeconds(1) }, cancellationToken: cancellationToken); + var store = await kv.CreateStoreAsync(new NatsKVConfig("kv1") { LimitMarkerTTL = TimeSpan.FromSeconds(2) }, cancellationToken: cancellationToken); // The first message we publish is set to "never expire", therefore it won't age out with the MaxAge policy. - await store.PutAsync($"k0", $"v0", TimeSpan.MaxValue, cancellationToken: cancellationToken); + await store.CreateAsync($"k0", $"v0", TimeSpan.MaxValue, cancellationToken: cancellationToken); + + await Task.Delay(1000); for (var i = 1; i < 11; i++) { - await store.PutAsync($"k{i}", $"v{i}", TimeSpan.FromSeconds(1), cancellationToken: cancellationToken); + await store.CreateAsync($"k{i}", $"v{i}", TimeSpan.FromSeconds(2), cancellationToken: cancellationToken); } - var state = await store.GetStatusAsync(); + var state = await store.GetStatusAsync(cancellationToken); Assert.Equal(11, state.Info.State.Messages); Assert.Equal(1ul, state.Info.State.FirstSeq); Assert.Equal(11ul, state.Info.State.LastSeq); - // Sleep for two seconds, only the first message should be there - await Task.Delay(2000); - state = await store.GetStatusAsync(); + await Retry.Until( + reason: "messages are deleted", + condition: async () => + { + var state1 = await store.GetStatusAsync(cancellationToken); + return state1.Info.State is { Messages: 1, FirstSeq: 1, LastSeq: 21 }; + }, + retryDelay: TimeSpan.FromSeconds(2), + timeout: TimeSpan.FromSeconds(30)); + + state = await store.GetStatusAsync(cancellationToken); Assert.Equal(1, state.Info.State.Messages); Assert.Equal(1ul, state.Info.State.FirstSeq); - Assert.Equal(11ul, state.Info.State.LastSeq); + Assert.Equal(21ul, state.Info.State.LastSeq); } [Fact]