diff --git a/src/kv.c b/src/kv.c index 27146d704..d8fec330d 100644 --- a/src/kv.c +++ b/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)); diff --git a/src/nats.h b/src/nats.h index c3d8f4264..d23ff8df7 100644 --- a/src/nats.h +++ b/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. diff --git a/test/list_test.txt b/test/list_test.txt index 05d6f1abb..4804a8f04 100644 --- a/test/list_test.txt +++ b/test/list_test.txt @@ -138,6 +138,7 @@ _test(JetStreamUnmarshalStreamConfig) _test(JetStreamUnmarshalStreamInfo) _test(JetStreamUnmarshalStreamState) _test(KeyValueBasics) +_test(KeyValueCreateWithTTL) _test(KeyValueCrossAccount) _test(KeyValueDeleteTombstones) _test(KeyValueDeleteVsPurge) @@ -145,6 +146,7 @@ _test(KeyValueDiscardOldToNew) _test(KeyValueHistory) _test(KeyValueKeys) _test(KeyValueKeysWithFilters) +_test(KeyValueLimitMarkerTTL) _test(KeyValueManager) _test(KeyValueMirrorCrossDomains) _test(KeyValueMirrorDirectGet) @@ -152,7 +154,6 @@ _test(KeyValuePurgeDeletesMarkerThreshold) _test(KeyValueRePublish) _test(KeyValueWatch) _test(KeyValueWatchMulti) -_test(KeyValueLimitMarkerTTL) _test(LameDuckMode) _test(MessageBufferPadding) _test(MicroAddService) diff --git a/test/test.c b/test/test.c index 8e5b4b170..eb7302b10 100644 --- a/test/test.c +++ b/test/test.c @@ -37400,6 +37400,158 @@ void test_KeyValueLimitMarkerTTL(void) JS_TEARDOWN; } +// Exercises the per-key TTL write helpers (kvStore_CreateWithTTL / +// kvStore_CreateStringWithTTL / kvStore_UpdateWithTTL / kvStore_UpdateStringWithTTL), +// the kvConfig.LimitMarkerTTL bucket option, and kvPurgeOptions.TTL. Requires +// nats-server v2.11+ (per-message TTL); JS_SETUP(2,11,0) skips it on older servers. +void test_KeyValueCreateWithTTL(void) +{ + natsStatus s; + kvStore *kv = NULL; + kvEntry *e = NULL; + kvConfig kvc; + kvPurgeOptions po; + uint64_t rev = 0; + + JS_SETUP(2, 11, 0); + + test("Create KV with LimitMarkerTTL: "); + kvConfig_Init(&kvc); + kvc.Bucket = "KVCTTL"; + kvc.LimitMarkerTTL = (int64_t)1000000000; // 1s (ns): enables per-message TTL + marker auto-expire + s = js_CreateKeyValue(&kv, js, &kvc); + testCond(s == NATS_OK); + + test("CreateWithTTL rejects NULL kv: "); + s = kvStore_CreateWithTTL(NULL, NULL, "k", (const void*)"v", 1, 1000); + testCond(s == NATS_INVALID_ARG); + nats_clearLastError(); + + test("UpdateWithTTL rejects NULL kv: "); + s = kvStore_UpdateWithTTL(NULL, NULL, "k", (const void*)"v", 1, 0, 1000); + testCond(s == NATS_INVALID_ARG); + nats_clearLastError(); + + test("CreateStringWithTTL k=v ttl=1s: "); + s = kvStore_CreateStringWithTTL(&rev, kv, "k", "v", 1000); + testCond((s == NATS_OK) && (rev == 1)); + + test("Get before expiry returns the value: "); + s = kvStore_Get(&e, kv, "k"); + testCond((s == NATS_OK) && (e != NULL) + && (strcmp(kvEntry_ValueString(e), "v") == 0) + && (kvEntry_Operation(e) == kvOp_Put)); + kvEntry_Destroy(e); + e = NULL; + + test("CreateWithTTL over a live key fails (create-if-absent preserved): "); + s = kvStore_CreateStringWithTTL(&rev, kv, "k", "v-dup", 1000); + testCond(s != NATS_OK); + nats_clearLastError(); + + test("Key expires server-side after its per-key TTL: "); + nats_Sleep(2500); + s = kvStore_Get(&e, kv, "k"); + testCond((s == NATS_NOT_FOUND) && (e == NULL)); + nats_clearLastError(); + + // Re-create over the (expired) tombstone must succeed on the first attempt — the + // marker-aware create retry the helper inherits from kvStore_Create — and keep the TTL. + test("Re-create over the expired marker (first attempt): "); + s = kvStore_CreateStringWithTTL(&rev, kv, "k", "v2", 1000); + testCond(s == NATS_OK); + + test("Get the re-created value: "); + s = kvStore_Get(&e, kv, "k"); + testCond((s == NATS_OK) && (e != NULL) + && (strcmp(kvEntry_ValueString(e), "v2") == 0)); + kvEntry_Destroy(e); + e = NULL; + + // kvStore_UpdateWithTTL: set a per-key TTL on an existing (no-TTL) key via a CAS + // update — the update-side counterpart of kvStore_CreateWithTTL (a plain + // kvStore_Update cannot add a TTL). + test("Seed a no-TTL key for UpdateWithTTL: "); + s = kvStore_CreateString(&rev, kv, "u", "u0"); + testCond((s == NATS_OK) && (rev >= 1)); + + test("UpdateWithTTL sets a per-key TTL under a CAS predicate: "); + s = kvStore_UpdateWithTTL(&rev, kv, "u", (const void*)"u1", 2, rev, 1000); + testCond(s == NATS_OK); + + test("Updated value is present before expiry: "); + s = kvStore_Get(&e, kv, "u"); + testCond((s == NATS_OK) && (e != NULL) + && (strcmp(kvEntry_ValueString(e), "u1") == 0)); + kvEntry_Destroy(e); + e = NULL; + + test("UpdateStringWithTTL writes the value under a CAS predicate: "); + s = kvStore_UpdateStringWithTTL(&rev, kv, "u", "u2", rev, 1000); + testCond(s == NATS_OK); + + test("String-updated value is present before expiry: "); + s = kvStore_Get(&e, kv, "u"); + testCond((s == NATS_OK) && (e != NULL) + && (strcmp(kvEntry_ValueString(e), "u2") == 0)); + kvEntry_Destroy(e); + e = NULL; + + test("Key expires server-side after the update's per-key TTL: "); + nats_Sleep(2500); + s = kvStore_Get(&e, kv, "u"); + testCond((s == NATS_NOT_FOUND) && (e == NULL)); + nats_clearLastError(); + + // E: a stale CAS predicate (wrong `last`) must fail rather than overwrite. + test("Seed a key for the stale-CAS check: "); + s = kvStore_CreateString(&rev, kv, "c", "c0"); + testCond(s == NATS_OK); + + test("UpdateWithTTL fails on a stale CAS revision: "); + s = kvStore_UpdateWithTTL(&rev, kv, "c", (const void*)"c1", 2, rev + 100, 1000); + testCond(s != NATS_OK); + nats_clearLastError(); + + // D: the server requires a per-key TTL of at least 1 second; a sub-second ttl is + // rejected. (The constraint is a 1s minimum, not whole-seconds — see the 1500ms case + // below.) The marker-aware create path runs a _getEntry on retry which clears the + // server's error text, so assert the server message on the single-publish update path. + test("Sub-second per-key TTL is rejected on create: "); + s = kvStore_CreateStringWithTTL(&rev, kv, "sub", "x", 500); + testCond(s != NATS_OK); + nats_clearLastError(); + + test("Sub-second per-key TTL on update surfaces 'invalid per-message TTL': "); + s = kvStore_CreateString(&rev, kv, "sub2", "s0"); + if (s == NATS_OK) + s = kvStore_UpdateWithTTL(&rev, kv, "sub2", (const void*)"s1", 2, rev, 500); + testCond((s != NATS_OK) + && (strstr(nats_GetLastError(NULL), "invalid per-message TTL") != NULL)); + nats_clearLastError(); + + // A non-whole-second ttl >= 1s is accepted (confirms the 1s-minimum rule). + test("Non-whole-second TTL (1500ms) is accepted: "); + s = kvStore_CreateStringWithTTL(&rev, kv, "ws", "x", 1500); + testCond(s == NATS_OK); + nats_clearLastError(); + + test("Purge with PurgeTTL: "); + kvPurgeOptions_Init(&po); + po.TTL = 1000; // ms: purge marker auto-expires + s = kvStore_Purge(kv, "k", &po); + testCond(s == NATS_OK); + + test("Get after purge returns NOT_FOUND: "); + s = kvStore_Get(&e, kv, "k"); + testCond((s == NATS_NOT_FOUND) && (e == NULL)); + nats_clearLastError(); + + kvStore_Destroy(kv); + + JS_TEARDOWN; +} + void test_natsHeader(void) { natsStatus s = NATS_OK;