Add subject_delete_marker_ttl and limit_marker_ttl support with review fixes

Implements PR nats-io#814 feature (closes nats-io#725) with improvements from code review:

- Add subject_delete_marker_ttl field to StreamConfig with nanosecond
  conversion in from_response/as_dict
- Add limit_marker_ttl field to KeyValueConfig, passed through to
  StreamConfig in create_key_value
- Add headers field to KeyValue.Entry for accessing message metadata
- Detect TTL-expired markers via Nats-Marker-Reason header in watcher

Review improvements applied:
- Extract "Nats-Marker-Reason" and "MaxAge" magic strings to module
  constants (NATS_MARKER_REASON, NATS_MARKER_MAX_AGE) in kv.py
- Add limit_marker_ttl nanosecond conversion in KeyValueConfig.as_dict()
  for consistency with ttl field
- Use Dict[str, str] type annotation for Entry.headers instead of bare dict
- Add server version comments on KeyValueConfig.limit_marker_ttl

https://claude.ai/code/session_01MteA8J2yH3ky5Y6coBpEAf

Author: claude

nats/src/nats/js/api.py

@@ -346,10 +346,15 @@ class StreamConfig(Base):
     # Metadata are user defined string key/value pairs.
     metadata: Optional[Dict[str, str]] = None
 
+    # TTL of subject delete markers.
+    # Introduced in nats-server 2.11.0.
+    subject_delete_marker_ttl: Optional[float] = None  # in seconds
+
     @classmethod
     def from_response(cls, resp: Dict[str, Any]):
         cls._convert_nanoseconds(resp, "max_age")
         cls._convert_nanoseconds(resp, "duplicate_window")
+        cls._convert_nanoseconds(resp, "subject_delete_marker_ttl")
         cls._convert(resp, "placement", Placement)
         cls._convert(resp, "mirror", StreamSource)
         cls._convert(resp, "sources", StreamSource)
@@ -361,6 +366,7 @@ def as_dict(self) -> Dict[str, object]:
         result = super().as_dict()
         result["duplicate_window"] = self._to_nanoseconds(self.duplicate_window)
         result["max_age"] = self._to_nanoseconds(self.max_age)
+        result["subject_delete_marker_ttl"] = self._to_nanoseconds(self.subject_delete_marker_ttl)
         if self.sources:
             result["sources"] = [src.as_dict() for src in self.sources]
         if self.compression and (self.compression != StoreCompression.NONE and self.compression != StoreCompression.S2):
@@ -748,9 +754,14 @@ class KeyValueConfig(Base):
     republish: Optional[RePublish] = None
     direct: Optional[bool] = None
 
+    # TTL of limit (delete) markers for KV.
+    # Introduced in nats-server 2.11.0.
+    limit_marker_ttl: Optional[float] = None  # in seconds
+
     def as_dict(self) -> Dict[str, object]:
         result = super().as_dict()
         result["ttl"] = self._to_nanoseconds(self.ttl)
+        result["limit_marker_ttl"] = self._to_nanoseconds(self.limit_marker_ttl)
         return result
 
 

nats/src/nats/js/client.py

@@ -1416,6 +1416,7 @@ async def create_key_value(
             num_replicas=config.replicas,
             storage=config.storage,
             republish=config.republish,
+            subject_delete_marker_ttl=config.limit_marker_ttl,
         )
         si = await self.add_stream(stream)
         assert stream.name is not None

nats/src/nats/js/kv.py

@@ -19,7 +19,7 @@
 import logging
 import re
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import nats.errors
 import nats.js.errors
@@ -32,6 +32,8 @@
 KV_DEL = "DEL"
 KV_PURGE = "PURGE"
 MSG_ROLLUP_SUBJECT = "sub"
+NATS_MARKER_REASON = "Nats-Marker-Reason"
+NATS_MARKER_MAX_AGE = "MaxAge"
 
 logger = logging.getLogger(__name__)
 
@@ -86,6 +88,7 @@ class Entry:
         delta: Optional[int]
         created: Optional[int]
         operation: Optional[str]
+        headers: Optional[Dict[str, str]] = None
 
     @dataclass(frozen=True)
     class BucketStatus:
@@ -472,8 +475,11 @@ async def watch_updates(msg):
 
             meta = msg.metadata
             op = None
-            if msg.header and KV_OP in msg.header:
-                op = msg.header.get(KV_OP)
+            if msg.header:
+                if KV_OP in msg.header:
+                    op = msg.header.get(KV_OP)
+                elif msg.header.get(NATS_MARKER_REASON) == NATS_MARKER_MAX_AGE:
+                    op = KV_DEL
 
                 # keys() uses this
                 if ignore_deletes:
@@ -491,6 +497,7 @@ async def watch_updates(msg):
                 delta=meta.num_pending,
                 created=meta.timestamp,
                 operation=op,
+                headers=msg.header,
             )
             await watcher._updates.put(entry)
 

nats/tests/test_js.py

@@ -5160,3 +5160,70 @@ async def test_add_stream_invalid_names(self):
                 ),
             ):
                 await js.add_stream(name=name)
+
+
+class V211FeaturesTest(SingleJetStreamServerTestCase):
+
+    @async_test
+    async def test_subject_delete_marker_ttl(self):
+        nc = NATS()
+        await nc.connect()
+
+        js = nc.jetstream()
+        stream = await js.add_stream(
+            nats.js.api.StreamConfig(subject_delete_marker_ttl=1),
+            name="test-subject-delete-marker-ttl",
+            subjects=["delete.marker"],
+        )
+        assert isinstance(stream, nats.js.api.StreamInfo)
+        assert isinstance(stream.config, nats.js.api.StreamConfig)
+        assert stream.config.subject_delete_marker_ttl == 1
+
+        messages: list[Msg] = []
+
+        async def save_msg(msg):
+            messages.append(msg)
+
+        sub = await js.subscribe("delete.marker", cb=save_msg)
+        await js.publish("delete.marker", b"test-message", msg_ttl=1)
+
+        await asyncio.sleep(1.1)
+        assert len(messages) == 2
+        assert messages[0].data == b"test-message"
+        assert messages[0].headers and messages[0].headers["Nats-TTL"] == "1"
+
+        assert messages[1].data == b""
+        m1_headers = messages[1].headers
+        assert m1_headers
+        assert m1_headers["Nats-Marker-Reason"] == "MaxAge"
+        assert m1_headers["Nats-TTL"] == "1s"
+
+    @async_test
+    async def test_create_key_value_ttl(self):
+        nc = NATS()
+        await nc.connect()
+
+        js = nc.jetstream()
+
+        kv = await js.create_key_value(
+            nats.js.api.KeyValueConfig("delete_marker_kv", ttl=1, limit_marker_ttl=1)
+        )
+
+        watcher = await kv.watchall()
+        await watcher.updates()
+        await kv.create("key", b"value", msg_ttl=1)
+
+        update = await watcher.updates()
+        assert update is not None
+        assert update.key == "key"
+        assert update.revision == 1
+        assert update.headers and update.headers["Nats-TTL"] == "1"
+
+        await asyncio.sleep(1.1)
+
+        update = await watcher.updates()
+        assert update is not None
+        assert update.key == "key"
+        assert not update.value
+        assert update.headers and update.headers["Nats-Marker-Reason"] == "MaxAge"
+        assert update.operation == "DEL"