[ADDED] Per-message TTL support (nats-server v2.11 feature)

src/js.c

@@ -1,4 +1,4 @@
-// Copyright 2021-2024 The NATS Authors
+// Copyright 2021-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -476,6 +476,13 @@ _setHeadersFromOptions(natsMsg *msg, jsPubOptions *opts)
             s = natsMsgHeader_Set(msg, jsExpectedLastSubjSeqHdr, temp);
         }
     }
+    if ((s == NATS_OK) && (opts->MsgTTL != 0))
+    {
+        if (opts->MsgTTL < 0)
+            s = nats_setError(NATS_INVALID_ARG, "option 'MsgTTL' (%" PRId64 ") cannot be negative", opts->MsgTTL);
+        else
+            s = natsMsgHeader_Set(msg, jsMsgTTLHdr, nats_formatDurationMilli(temp, opts->MsgTTL));
+    }
 
     return NATS_UPDATE_ERR_STACK(s);
 }

src/js.h

@@ -1,4 +1,4 @@
-// Copyright 2021-2024 The NATS Authors
+// Copyright 2021-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -34,6 +34,7 @@ extern const char*      jsDefaultAPIPrefix;
 extern const int64_t    jsDefaultRequestWait;
 
 #define jsMsgIdHdr                  "Nats-Msg-Id"
+#define jsMsgTTLHdr                 "Nats-TTL"
 #define jsExpectedStreamHdr         "Nats-Expected-Stream"
 #define jsExpectedLastSeqHdr        "Nats-Expected-Last-Sequence"
 #define jsExpectedLastSubjSeqHdr    "Nats-Expected-Last-Subject-Sequence"

src/jsm.c

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 The NATS Authors
+// Copyright 2021-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -763,6 +763,8 @@ js_unmarshalStreamConfig(nats_JSON *json, const char *fieldName, jsStreamConfig
     obj = NULL;
     IFOK(s, nats_JSONGetObject(jcfg, "consumer_limits", &obj));
     IFOK(s, _unmarshalStreamConsumerLimits(obj, &(cfg->ConsumerLimits)));
+    IFOK(s, nats_JSONGetLong(jcfg, "subject_delete_marker_ttl", &(cfg->SubjectDeleteMarkerTTL)));
+    IFOK(s, nats_JSONGetBool(jcfg, "allow_msg_ttl", &(cfg->AllowMsgTTL)));
 
     if (s == NATS_OK)
         *new_cfg = cfg;
@@ -891,6 +893,11 @@ js_marshalStreamConfig(natsBuffer **new_buf, jsStreamConfig *cfg)
     IFOK(s, _marshalSubjectTransformConfig(&cfg->SubjectTransform, buf));
     IFOK(s, _marshalStreamConsumerLimits(&cfg->ConsumerLimits, buf));
 
+    if ((s == NATS_OK) && cfg->AllowMsgTTL)
+        s = natsBuf_Append(buf, ",\"allow_msg_ttl\":true", -1);
+    if ((s == NATS_OK) && cfg->SubjectDeleteMarkerTTL > 0)
+        s = nats_marshalLong(buf, true, "subject_delete_marker_ttl", cfg->SubjectDeleteMarkerTTL);
+
     IFOK(s, natsBuf_AppendByte(buf, '}'));
 
     if (s == NATS_OK)

src/msg.c

@@ -1,4 +1,4 @@
-// Copyright 2015-2021 The NATS Authors
+// Copyright 2015-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -213,7 +213,9 @@ _processKeyValue(int line, natsMsg *msg, char *endPtr, char **pPtr, char **lastK
     start = ptr;
     if (*ptr == '\r')
     {
-        if ((++ptr == endPtr) || ((*ptr == '\n') && (++ptr == endPtr)))
+        if ((++ptr == endPtr) ||                     // case: "\r"
+            ((*ptr == '\n') && ((++ptr == endPtr) || // case: "\r\n"
+            ((*ptr == '\r') && (++ptr < endPtr) && (*ptr == '\n') && (++ptr == endPtr))))) // case: "\r\n\r\n"
         {
             *pPtr = ptr;
             return NATS_OK;

src/nats.h

@@ -1,4 +1,4 @@
-// Copyright 2015-2024 The NATS Authors
+// Copyright 2015-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -295,6 +295,7 @@ typedef struct jsPubOptions
         uint64_t        ExpectLastSeq;          ///< Expected last message sequence in the stream.
         uint64_t        ExpectLastSubjectSeq;   ///< Expected last message sequence for the subject in the stream.
         bool            ExpectNoMessage;        ///< Expected no message (that is, sequence == 0) for the subject in the stream.
+        int64_t         MsgTTL;                 ///< Message time to live (TTL) in milliseconds, used by the server to expire the message. Requires nats-server v2.11.0 or later.
 
 } jsPubOptions;
 
@@ -555,7 +556,7 @@ typedef struct jsStreamConfig {
         int64_t                 MaxConsumers;
         int64_t                 MaxMsgs;
         int64_t                 MaxBytes;
-        int64_t                 MaxAge;
+        int64_t                 MaxAge;         ///< Max age of messages in nanoseconds.
         int64_t                 MaxMsgsPerSubject;
         int32_t                 MaxMsgSize;
         jsDiscardPolicy         Discard;
@@ -611,6 +612,16 @@ typedef struct jsStreamConfig {
         /// @brief Sets the limits on certain options on all consumers of the
         /// stream.
         jsStreamConsumerLimits  ConsumerLimits;
+
+        /// @brief Allow the message to be sent with a time to live (TTL) value.
+        /// Requires nats-server v2.11.0 or later.
+        bool                    AllowMsgTTL;
+
+        /// @brief Enables and sets a duration for adding server markers for
+        /// delete, purge and max age limits. In nanoseconds. Requires
+        /// nats-server v2.11.0 or later.
+        int64_t                 SubjectDeleteMarkerTTL;
+
 } jsStreamConfig;
 
 /**

src/status.h

@@ -1,4 +1,4 @@
-// Copyright 2015-2022 The NATS Authors
+// Copyright 2015-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -268,6 +268,39 @@ typedef enum {
     JSConsumerCreateFilterSubjectMismatchErr = 10131,   ///< Consumer create request did not match filtered subject from create subject
     JSConsumerCreateDurableAndNameMismatchErr = 10132,  ///< Consumer Durable and Name have to be equal if both are provided
     JSReplicasCountCannotBeNegativeErr = 10133,         ///< Replicas count cannot be negative
+    JSConsumerReplicasShouldMatchStreamErr = 10134,     ///< Consumer config replicas must match interest retention stream's replicas
+    JSConsumerMetadataLengthErr = 10135,                ///< Consumer metadata exceeds maximum size 
+    JSConsumerDuplicateFilterSubjectsErr = 10136,       ///< Consumer cannot have both FilterSubject and FilterSubjects specified
+    JSConsumerMultipleFiltersNotAllowedErr = 10137,     ///< Consumer with multiple subject filters cannot use subject based API
+    JSConsumerOverlappingSubjectFiltersErr = 10138,     ///< Consumer subject filters cannot overlap
+    JSConsumerEmptyFilterErr = 10139,                   ///< Consumer filter in FilterSubjects cannot be empty
+    JSSourceDuplicateDetectedErr = 10140,               ///< Source stream, filter and transform (plus external if present) must form a unique combination
+    JSSourceInvalidStreamNameErr = 10141,               ///< Sourced stream name is invalid
+    JSMirrorInvalidStreamNameErr = 10142,               ///< Mirrored stream name is invalid
+    JSSourceMultipleFiltersNotAllowedErr = 10144,       ///< Source with multiple subject transforms cannot also have a single subject filter
+    JSSourceInvalidSubjectFilterErr = 10145,            ///< Source transform source error
+    JSSourceInvalidTransformDestinationErr = 10146,     ///< Source transform error
+    JSSourceOverlappingSubjectFiltersErr = 10147,       ///< Source subject filters cannot overlap
+    JSConsumerAlreadyExistsErr = 10148,                 ///< Action CREATE is used for a existing consumer with a different config
+    JSConsumerDoesNotExistErr = 10149,                  ///< Action UPDATE is used for a non-existing consumer
+    JSMirrorMultipleFiltersNotAllowedErr = 10150,       ///< Mirror with multiple subject transforms cannot also have a single subject filter
+    JSMirrorInvalidSubjectFilterErr = 10151,            ///< Mirror transform source error
+    JSMirrorOverlappingSubjectFiltersErr = 10152,       ///< Mirror subject filters cannot overlap
+    JSConsumerInactiveThresholdExcessErr = 10153,       ///< Consumer inactive threshold exceeds system limit
+    JSMirrorInvalidTransformDestinationErr = 10154,     ///< Mirror transform error
+    JSStreamTransformInvalidSourceErr = 10155,          ///< Stream transform source error
+    JSStreamTransformInvalidDestinationErr = 10156,     ///< Stream transform destination error
+    JSPedanticErr = 10157,                              ///< Pedantic mode error
+    JSStreamDuplicateMessageConflictErr = 10158,        ///< Duplicate message id is in process
+    JSConsumerPriorityPolicyWithoutGroupErr = 10159,    ///< Setting PriorityPolicy requires at least one PriorityGroup to be set
+    JSConsumerInvalidPriorityGroupErr = 10160,          ///< Provided priority group does not exist for this consumer"
+    JSConsumerEmptyGroupNameErr = 10161,                ///< Group name cannot be an empty string
+    JSConsumerInvalidGroupNameErr = 10162,              ///< Valid priority group name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters
+    JSStreamExpectedLastSeqPerSubjectNotReadyErr = 10163, ///< Expected last sequence per subject temporarily unavailable
+    JSStreamWrongLastSequenceConstantErr = 10164,       ///< Wrong last sequence
+    JSMessageTTLInvalidErr = 10165,                     ///< Invalid per-message TTL
+    JSMessageTTLDisabledErr = 10166,                    ///< Per-message TTL is disabled
+    JSStreamTooManyRequestsErr = 10167,                 ///< Too many requests
 
 } jsErrCode;
 

src/util.c

@@ -1,4 +1,4 @@
-// Copyright 2015-2023 The NATS Authors
+// Copyright 2015-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -2287,17 +2287,22 @@ fmt_int(char *buf, int w, uint64_t v)
     return w;
 }
 
-natsStatus
-nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d)
+// Converts a number of nanoseconds to a go duration string.
+// Note: buf must be at least 32 bytes long.
+// This is the maximum size of a duration string.
+// The format is: "2540400h10m10.000000000s".
+// Returns the pointer to the beginning of the string (in buf).
+char *
+nats_formatDuration(char *buf, int64_t d)
 {
-    // Largest time is 2540400h10m10.000000000s
-    char buf[32];
-    int w = 32;
     bool neg = d < 0;
     uint64_t u = (uint64_t) (neg ? -d : d);
     int prec;
-    natsStatus s = NATS_OK;
-    const char *start = (comma ? ",\"" : "\"");
+
+    // Largest time is 2540400h10m10.000000000s
+    int w = 32;
+    w--;
+    buf[w] = '\0';
 
     if (u < 1000000000)
     {
@@ -2307,12 +2312,7 @@ nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, in
         buf[w] = 's';
         w--;
         if (u == 0)
-        {
-            s = natsBuf_Append(out_buf, start, -1);
-            IFOK(s, natsBuf_Append(out_buf, field_name, -1));
-            IFOK(s, natsBuf_Append(out_buf, "\":\"0s\"", -1));
-            return NATS_UPDATE_ERR_STACK(s);
-        }
+            return buf+w;
         else if (u < 1000)
         {
             // print nanoseconds
@@ -2373,10 +2373,23 @@ nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, in
         buf[w] = '-';
     }
 
+    return buf+w;
+}
+
+natsStatus
+nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d)
+{
+    // Largest time is 2540400h10m10.000000000s
+    char buf[32];
+    natsStatus s = NATS_OK;
+    const char *start = (comma ? ",\"" : "\"");
+
+    char *dur = nats_formatDuration(buf, d);
+
     s = natsBuf_Append(out_buf, start, -1);
     IFOK(s, natsBuf_Append(out_buf, field_name, -1));
     IFOK(s, natsBuf_Append(out_buf, "\":\"", -1));
-    IFOK(s, natsBuf_Append(out_buf, buf + w, sizeof(buf) - w));
+    IFOK(s, natsBuf_Append(out_buf, dur, -1));
     IFOK(s, natsBuf_Append(out_buf, "\"", -1));
     return NATS_UPDATE_ERR_STACK(s);
 }

src/util.h

@@ -1,4 +1,4 @@
-// Copyright 2015-2023 The NATS Authors
+// Copyright 2015-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -240,6 +240,14 @@ nats_marshalLong(natsBuffer *buf, bool comma, const char *fieldName, int64_t lva
 natsStatus
 nats_marshalULong(natsBuffer *buf, bool comma, const char *fieldName, uint64_t uval);
 
+char *
+nats_formatDuration(char *buf, int64_t nano);
+
+static inline char *nats_formatDurationMilli(char *buf, int64_t milli)
+{
+    return nats_formatDuration(buf, milli * 1000 * 1000);
+}
+
 natsStatus
 nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d);
 

test/list_test.txt

@@ -90,6 +90,7 @@ _test(JetStreamInfoWithSubjects)
 _test(JetStreamMarshalStreamConfig)
 _test(JetStreamMgtConsumers)
 _test(JetStreamMgtStreams)
+_test(JetStreamMsgDeleteMarkerMaxAge)
 _test(JetStreamNakWithDelay)
 _test(JetStreamOrderedConsSrvRestart)
 _test(JetStreamOrderedConsumer)
@@ -98,6 +99,7 @@ _test(JetStreamOrderedConsumerWithErrors)
 _test(JetStreamPublish)
 _test(JetStreamPublishAckHandler)
 _test(JetStreamPublishAsync)
+_test(JetStreamPublishTTL)
 _test(JetStreamStreamsSealAndRollup)
 _test(JetStreamSubscribe)
 _test(JetStreamSubscribeConfigCheck)