KV per-key TTL (parity with nats.go #1864)

src/kv.c

@@ -267,6 +267,15 @@ js_CreateKeyValue(kvStore **new_kv, jsCtx *js, kvConfig *cfg)
         sc.AllowDirect = true;
         sc.RePublish = cfg->RePublish;
 
+        // LimitMarkerTTL enables per-message TTL on the bucket (so
+        // kvStore_CreateWithTTL works) and auto-expires the
+        // delete/purge tombstone markers after the given duration (nanoseconds).
+        if (cfg->LimitMarkerTTL > 0)
+        {
+            sc.AllowMsgTTL = true;
+            sc.SubjectDeleteMarkerTTL = cfg->LimitMarkerTTL;
+        }
+
         if (cfg->Mirror != NULL)
         {
             jsStreamSource *m = cfg->Mirror;
@@ -684,8 +693,33 @@ kvStore_PutString(uint64_t *rev, kvStore *kv, const char *key, const char *data)
     return NATS_UPDATE_ERR_STACK(s);
 }
 
+// CAS update carrying an optional per-key TTL (milliseconds). ttl <= 0 means
+// "no TTL" (identical to the pre-existing update). The TTL rides
+// jsPubOptions.MsgTTL; combining it with the CAS predicate in the same publish
+// is what callers would otherwise have to hand-roll.
 natsStatus
-kvStore_Create(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len)
+kvStore_UpdateWithTTL(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len,
+                      uint64_t last, int64_t ttl)
+{
+    natsStatus      s;
+    jsPubOptions    po;
+
+    jsPubOptions_Init(&po);
+    if (last == 0)
+        po.ExpectNoMessage = true;
+    else
+        po.ExpectLastSubjectSeq = last;
+    if (ttl > 0)
+        po.MsgTTL = ttl;
+    s = _putEntry(rev, kv, &po, key, data, len);
+    return NATS_UPDATE_ERR_STACK(s);
+}
+
+// create-if-absent carrying an optional per-key TTL. Reuses the marker-aware
+// retry so a re-create over a DEL/PURGE/MaxAge tombstone succeeds AND keeps the
+// TTL on both the initial attempt and the re-create-over-marker attempt.
+natsStatus
+kvStore_CreateWithTTL(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len, int64_t ttl)
 {
     natsStatus s;
     natsStatus ls;
@@ -695,7 +729,7 @@ kvStore_Create(uint64_t *rev, kvStore *kv, const char *key, const void *data, in
     if (kv == NULL)
         return nats_setDefaultError(NATS_INVALID_ARG);
 
-    s = kvStore_Update(rev, kv, key, data, len, 0);
+    s = kvStore_UpdateWithTTL(rev, kv, key, data, len, 0, ttl);
     if (s == NATS_OK)
         return s;
 
@@ -705,13 +739,20 @@ kvStore_Create(uint64_t *rev, kvStore *kv, const char *key, const void *data, in
     if (ls == NATS_OK)
     {
         if (deleted)
-            s = kvStore_Update(rev, kv, key, data, len, kvEntry_Revision(e));
+            s = kvStore_UpdateWithTTL(rev, kv, key, data, len, kvEntry_Revision(e), ttl);
 
         kvEntry_Destroy(e);
     }
     return NATS_UPDATE_ERR_STACK(s);
 }
 
+natsStatus
+kvStore_Create(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len)
+{
+    natsStatus s = kvStore_CreateWithTTL(rev, kv, key, data, len, 0);
+    return NATS_UPDATE_ERR_STACK(s);
+}
+
 natsStatus
 kvStore_CreateString(uint64_t *rev, kvStore *kv, const char *key, const char *data)
 {
@@ -720,17 +761,16 @@ kvStore_CreateString(uint64_t *rev, kvStore *kv, const char *key, const char *da
 }
 
 natsStatus
-kvStore_Update(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len, uint64_t last)
+kvStore_CreateStringWithTTL(uint64_t *rev, kvStore *kv, const char *key, const char *data, int64_t ttl)
 {
-    natsStatus      s;
-    jsPubOptions    po;
+    natsStatus s = kvStore_CreateWithTTL(rev, kv, key, (const void*) data, (int) strlen(data), ttl);
+    return NATS_UPDATE_ERR_STACK(s);
+}
 
-    jsPubOptions_Init(&po);
-    if (last == 0)
-        po.ExpectNoMessage = true;
-    else
-        po.ExpectLastSubjectSeq = last;
-    s = _putEntry(rev, kv, &po, key, data, len);
+natsStatus
+kvStore_Update(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len, uint64_t last)
+{
+    natsStatus s = kvStore_UpdateWithTTL(rev, kv, key, data, len, last, 0);
     return NATS_UPDATE_ERR_STACK(s);
 }
 
@@ -741,6 +781,13 @@ kvStore_UpdateString(uint64_t *rev, kvStore *kv, const char *key, const char *da
     return NATS_UPDATE_ERR_STACK(s);
 }
 
+natsStatus
+kvStore_UpdateStringWithTTL(uint64_t *rev, kvStore *kv, const char *key, const char *data, uint64_t last, int64_t ttl)
+{
+    natsStatus s = kvStore_UpdateWithTTL(rev, kv, key, (const void*) data, (int) strlen(data), last, ttl);
+    return NATS_UPDATE_ERR_STACK(s);
+}
+
 static natsStatus
 _delete(kvStore *kv, const char *key, bool purge, const kvPurgeOptions *opts)
 {
@@ -770,10 +817,13 @@ _delete(kvStore *kv, const char *key, bool purge, const kvPurgeOptions *opts)
             s = natsMsgHeader_Set(msg, kvOpHeader, kvOpDeleteStr);
         }
     }
-    if (purge && (opts != NULL) && (opts->Timeout > 0))
+    if (purge && (opts != NULL) && ((opts->Timeout > 0) || (opts->TTL > 0)))
     {
         jsPubOptions_Init(&o);
-        o.MaxWait = opts->Timeout;
+        if (opts->Timeout > 0)
+            o.MaxWait = opts->Timeout;
+        if (opts->TTL > 0)              // PurgeTTL: the purge marker auto-expires
+            o.MsgTTL = opts->TTL;
         po = &o;
     }
     IFOK(s, js_PublishMsg(NULL, kv->js, msg, po, NULL));

src/nats.h

@@ -1601,6 +1601,12 @@ typedef struct kvConfig
         jsStreamSource  **Sources;
         int             SourcesLen;
 
+        // When > 0 (nanoseconds), enables per-message TTL on the bucket's
+        // backing stream and auto-expires the
+        // delete/purge tombstone markers after this duration. Required for
+        // kvStore_CreateWithTTL() to take effect. Requires nats-server v2.11+.
+        int64_t         LimitMarkerTTL;
+
 } kvConfig;
 
 /**
@@ -1689,6 +1695,14 @@ typedef struct kvPurgeOptions
         // The value is expressed as a time in nanoseconds.
         int64_t         DeleteMarkersOlderThan;
 
+        // When > 0, the purge marker
+        // written by kvStore_Purge() is published with this per-message TTL (in
+        // milliseconds), so the server auto-expires the purge tombstone after the
+        // duration. Requires the bucket created with kvConfig.LimitMarkerTTL > 0 and
+        // nats-server v2.11+. Ignored by kvStore_Delete() (delete markers use the
+        // bucket-level LimitMarkerTTL).
+        int64_t         TTL;
+
 } kvPurgeOptions;
 
 /** \brief A list of KeyValue store entries.
@@ -8489,6 +8503,33 @@ kvStore_Create(uint64_t *rev, kvStore *kv, const char *key, const void *data, in
 NATS_EXTERN natsStatus
 kvStore_CreateString(uint64_t *rev, kvStore *kv, const char *key, const char *data);
 
+/** \brief Create-if-absent with a per-key TTL.
+ *
+ * Like #kvStore_Create, but the new value is published with a per-message TTL so the
+ * server expires this key after `ttl` milliseconds. Re-creating over an existing
+ * delete/purge/MaxAge marker succeeds and keeps the TTL (the marker-aware create path
+ * is reused). Requires the bucket to have been created with `kvConfig.LimitMarkerTTL > 0`
+ * and nats-server v2.11+; otherwise the server rejects the TTL (JSMessageTTLDisabledErr).
+ * The server requires the TTL to be at least 1 second (1000 ms); a smaller value is
+ * rejected with JSMessageTTLInvalidErr.
+ *
+ * @param rev the location where to store the revision of this value, or `NULL`.
+ * @param kv the pointer to the #kvStore object.
+ * @param key the name of the key.
+ * @param data the pointer to the data in memory.
+ * @param len the number of bytes to copy from the data's memory location.
+ * @param ttl per-key time-to-live in milliseconds (`<= 0` means no TTL — same as #kvStore_Create).
+ */
+NATS_EXTERN natsStatus
+kvStore_CreateWithTTL(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len, int64_t ttl);
+
+/** \brief String variant of #kvStore_CreateWithTTL.
+ *
+ * \note Equivalent of calling #kvStore_CreateWithTTL with `(int) strlen(data)`.
+ */
+NATS_EXTERN natsStatus
+kvStore_CreateStringWithTTL(uint64_t *rev, kvStore *kv, const char *key, const char *data, int64_t ttl);
+
 /** \brief Updates the value for the key into the store if and only if the latest revision matches.
  *
  * Updates the value for the key into the store if and only if the latest revision matches.
@@ -8520,6 +8561,33 @@ kvStore_Update(uint64_t *rev, kvStore *kv, const char *key, const void *data, in
 NATS_EXTERN natsStatus
 kvStore_UpdateString(uint64_t *rev, kvStore *kv, const char *key, const char *data, uint64_t last);
 
+/** \brief Updates a key with a per-key TTL if and only if the latest revision matches.
+ *
+ * Like #kvStore_Update, but the new value is published with a per-message TTL so the
+ * server expires this key after `ttl` milliseconds. The TTL is set on the same CAS
+ * publish, so a plain update can adopt a TTL atomically. Requires the bucket to have
+ * been created with `kvConfig.LimitMarkerTTL > 0` and nats-server v2.11+; otherwise the
+ * server rejects the TTL (JSMessageTTLDisabledErr). The server requires the TTL to be at
+ * least 1 second (1000 ms); a smaller value is rejected with JSMessageTTLInvalidErr.
+ *
+ * @param rev the location where to store the revision of this value, or `NULL`.
+ * @param kv the pointer to the #kvStore object.
+ * @param key the name of the key.
+ * @param data the pointer to the data in memory.
+ * @param len the number of bytes to copy from the data's memory location.
+ * @param last the expected latest revision prior to the update.
+ * @param ttl per-key time-to-live in milliseconds (`<= 0` means no TTL — same as #kvStore_Update).
+ */
+NATS_EXTERN natsStatus
+kvStore_UpdateWithTTL(uint64_t *rev, kvStore *kv, const char *key, const void *data, int len, uint64_t last, int64_t ttl);
+
+/** \brief String variant of #kvStore_UpdateWithTTL.
+ *
+ * \note Equivalent of calling #kvStore_UpdateWithTTL with `(int) strlen(data)`.
+ */
+NATS_EXTERN natsStatus
+kvStore_UpdateStringWithTTL(uint64_t *rev, kvStore *kv, const char *key, const char *data, uint64_t last, int64_t ttl);
+
 /** \brief Deletes a key by placing a delete marker and leaving all revisions.
  *
  * Deletes a key by placing a delete marker and leaving all revisions.

test/list_test.txt

@@ -138,21 +138,22 @@ _test(JetStreamUnmarshalStreamConfig)
 _test(JetStreamUnmarshalStreamInfo)
 _test(JetStreamUnmarshalStreamState)
 _test(KeyValueBasics)
+_test(KeyValueCreateWithTTL)
 _test(KeyValueCrossAccount)
 _test(KeyValueDeleteTombstones)
 _test(KeyValueDeleteVsPurge)
 _test(KeyValueDiscardOldToNew)
 _test(KeyValueHistory)
 _test(KeyValueKeys)
 _test(KeyValueKeysWithFilters)
+_test(KeyValueLimitMarkerTTL)
 _test(KeyValueManager)
 _test(KeyValueMirrorCrossDomains)
 _test(KeyValueMirrorDirectGet)
 _test(KeyValuePurgeDeletesMarkerThreshold)
 _test(KeyValueRePublish)
 _test(KeyValueWatch)
 _test(KeyValueWatchMulti)
-_test(KeyValueLimitMarkerTTL)
 _test(LameDuckMode)
 _test(MessageBufferPadding)
 _test(MicroAddService)