feat: support delete-on-empty config for streams

src/streamstore/_lib/s2/v1alpha/s2_pb2.pyi

@@ -637,7 +637,7 @@ class ReadSessionResponse(_message.Message):
     ) -> None: ...
 
 class StreamConfig(_message.Message):
-    __slots__ = ("storage_class", "age", "timestamping")
+    __slots__ = ("storage_class", "age", "timestamping", "delete_on_empty")
     class Timestamping(_message.Message):
         __slots__ = ("mode", "uncapped")
         MODE_FIELD_NUMBER: _ClassVar[int]
@@ -650,17 +650,26 @@ class StreamConfig(_message.Message):
             uncapped: bool = ...,
         ) -> None: ...
 
+    class DeleteOnEmpty(_message.Message):
+        __slots__ = ("min_age_secs",)
+        MIN_AGE_SECS_FIELD_NUMBER: _ClassVar[int]
+        min_age_secs: int
+        def __init__(self, min_age_secs: _Optional[int] = ...) -> None: ...
+
     STORAGE_CLASS_FIELD_NUMBER: _ClassVar[int]
     AGE_FIELD_NUMBER: _ClassVar[int]
     TIMESTAMPING_FIELD_NUMBER: _ClassVar[int]
+    DELETE_ON_EMPTY_FIELD_NUMBER: _ClassVar[int]
     storage_class: StorageClass
     age: int
     timestamping: StreamConfig.Timestamping
+    delete_on_empty: StreamConfig.DeleteOnEmpty
     def __init__(
         self,
         storage_class: _Optional[_Union[StorageClass, str]] = ...,
         age: _Optional[int] = ...,
         timestamping: _Optional[_Union[StreamConfig.Timestamping, _Mapping]] = ...,
+        delete_on_empty: _Optional[_Union[StreamConfig.DeleteOnEmpty, _Mapping]] = ...,
     ) -> None: ...
 
 class BasinConfig(_message.Message):

src/streamstore/_mappers.py

@@ -27,6 +27,8 @@
     StreamInfo,
     TailOffset,
     Timestamp,
+    Timestamping,
+    TimestampingMode,
 )
 
 _ReadStart = SeqNum | Timestamp | TailOffset
@@ -125,6 +127,7 @@ def stream_config_message(
         storage_class = config.storage_class
         retention_age = config.retention_age
         timestamping = config.timestamping
+        delete_on_empty_min_age = config.delete_on_empty_min_age
         if storage_class is not None:
             paths.append(f"{mask_path_prefix}storage_class")
             stream_config.storage_class = msgs.StorageClass(storage_class.value)
@@ -142,6 +145,9 @@ def stream_config_message(
             if timestamping.uncapped is not None:
                 paths.append(f"{mask_path_prefix}timestamping.uncapped")
                 stream_config.timestamping.uncapped = timestamping.uncapped
+        if delete_on_empty_min_age is not None:
+            paths.append(f"{mask_path_prefix}delete_on_empty.min_age_secs")
+            stream_config.delete_on_empty.min_age_secs = delete_on_empty_min_age
     if return_mask_paths:
         return (stream_config, paths)
     return stream_config
@@ -181,6 +187,11 @@ def stream_config_schema(config: msgs.StreamConfig) -> StreamConfig:
     return StreamConfig(
         StorageClass(config.storage_class),
         config.age,
+        Timestamping(
+            mode=TimestampingMode(config.timestamping.mode),
+            uncapped=config.timestamping.uncapped,
+        ),
+        config.delete_on_empty.min_age_secs,
     )
 
 

src/streamstore/schemas.py

@@ -302,6 +302,10 @@ class StreamConfig:
     retention_age: int | None = None
     #: Timestamping behavior for appends to this stream, which influences how timestamps are handled.
     timestamping: Timestamping | None = None
+    #: Minimum age in seconds before this stream can be automatically deleted if empty.
+    #:
+    #: If not specified or set to ``0``, this stream will not be automatically deleted.
+    delete_on_empty_min_age: int | None = None
 
 
 @dataclass(slots=True)