Add persist_mode to StreamConfig (#773)

Signed-off-by: Casper Beyer <[email protected]>

Author: caspervonb

nats/src/nats/js/api.py

@@ -237,6 +237,28 @@ class StoreCompression(str, Enum):
     S2 = "s2"
 
 
+class PersistMode(str, Enum):
+    """
+    PersistMode defines the consistency and durability guarantees for stream persistence.
+
+    See ADR-56 for details: https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-56.md
+
+    Currently only applicable to R1 (single replica) streams.
+    Introduced in nats-server 2.12.0.
+    """
+
+    # DEFAULT represents the strongest consistency guarantee.
+    # Uses synchronous writes with fsync for maximum durability.
+    # Server does not store this value - it's the implied default when unset.
+    DEFAULT = "default"
+
+    # ASYNC enables asynchronous flushing of data to disk.
+    # Returns PubAck before disk persistence occurs, batching writes in memory.
+    # Provides significantly improved performance at the cost of potential data loss
+    # during infrastructure failures. Incompatible with batch publishing.
+    ASYNC = "async"
+
+
 @dataclass
 class RePublish(Base):
     """
@@ -317,6 +339,10 @@ class StreamConfig(Base):
     # Allow batched publishing. Introduced in nats-server 2.12.0.
     allow_batched: Optional[bool] = None
 
+    # Persistence mode for stream. Only applicable to R1 streams.
+    # Introduced in nats-server 2.12.0.
+    persist_mode: Optional[PersistMode] = None
+
     # Metadata are user defined string key/value pairs.
     metadata: Optional[Dict[str, str]] = None
 

nats/tests/test_js.py

@@ -4743,6 +4743,48 @@ async def test_stream_allow_atomic(self):
 
         await nc.close()
 
+    @async_test
+    async def test_stream_persist_mode(self):
+        nc = await nats.connect()
+
+        server_version = nc.connected_server_version
+        if server_version.major == 2 and server_version.minor < 12:
+            pytest.skip("persist_mode requires nats-server v2.12.0 or later")
+
+        js = nc.jetstream()
+
+        # Test setting async consistency model on R1 stream
+        await js.add_stream(
+            name="ASYNC",
+            subjects=["test"],
+            num_replicas=1,
+            persist_mode=nats.js.api.PersistMode.ASYNC,
+        )
+        sinfo = await js.stream_info("ASYNC")
+        assert sinfo.config.persist_mode == nats.js.api.PersistMode.ASYNC
+
+        # Test that default consistency model works
+        await js.add_stream(
+            name="DEFAULT_CONSISTENCY",
+            subjects=["foo"],
+            num_replicas=1,
+            persist_mode=nats.js.api.PersistMode.DEFAULT,
+        )
+        sinfo = await js.stream_info("DEFAULT_CONSISTENCY")
+        # Server doesn't store default value, so it may be None
+        assert sinfo.config.persist_mode in [None, nats.js.api.PersistMode.DEFAULT]
+
+        # Test that it defaults to None when not set
+        await js.add_stream(
+            name="UNSET_CONSISTENCY",
+            subjects=["bar"],
+            num_replicas=1,
+        )
+        sinfo = await js.stream_info("UNSET_CONSISTENCY")
+        assert sinfo.config.persist_mode in [None, nats.js.api.PersistMode.DEFAULT]
+
+        await nc.close()
+
     @async_test
     async def test_fetch_pull_subscribe_bind(self):
         nc = NATS()