Expire sheets and objref stores

Author: ajhai

llmstack/assets/stream.py

@@ -35,6 +35,9 @@ def append_chunk(self, chunk: bytes):
         )
         self._id += 1
 
+        # Expire the stream after 20 minutes
+        objref_stream_client.expire(self.objref, 1200)
+
     def finalize(self):
         # Read the stream and finalize the asset
         file_bytes = b""
@@ -56,6 +59,9 @@ def finalize(self):
             {"chunk": b"", "id": -1},
         )
 
+        # Expire the stream after 20 minutes
+        objref_stream_client.expire(self.objref, 1200)
+
         return self._asset.finalize_streaming_asset(file_bytes)
 
     def read(self, start_index=0, timeout=1000):

llmstack/sheets/consumers.py

@@ -78,6 +78,9 @@ async def sheet_error(self, event):
     async def _append_event_to_redis(self, event):
         try:
             sheet_run_data_store.rpush(self.run_id, json.dumps(event))
+
+            # Expire the stream after 1 day
+            sheet_run_data_store.expire(self.run_id, 86400)
         except Exception as e:
             logger.error(f"Error appending event to Redis: {str(e)}")