Remove lock from data stream

firehose/data_stream.py

@@ -156,8 +156,7 @@ def on_message_handler(message: firehose_models.MessageFrame) -> None:
             # Update the client's parameters with the new cursor
             client.update_params(models.ComAtprotoSyncSubscribeRepos.Params(cursor=commit.seq))
             # Persist the new cursor in the database
-            with db.atomic():
-                SubscriptionState.update(cursor=commit.seq).where(SubscriptionState.service == name).execute()
+            SubscriptionState.update(cursor=commit.seq).where(SubscriptionState.service == name).execute()
 
         if not commit.blocks:
             # Skip if there are no blocks to process

start_stream.py

@@ -18,12 +18,12 @@ def handle_termination(signum, frame):
     for sig in [signal.SIGINT, signal.SIGTERM, signal.SIGHUP]:
         signal.signal(sig, handle_termination)
 
-    try:
-        data_stream.run(config.SERVICE_DID, operations_callback, stream_stop_event)
-    except Exception as e:
-        logger.error(f"An exception occurred in the firehose: {e}")
-        # try to restart the stream
-        main()
+    while not stream_stop_event.is_set():
+        try:
+            data_stream.run(config.SERVICE_DID, operations_callback, stream_stop_event)
+        except Exception as e:
+            logger.error(f"An exception occurred in the firehose: {e}")
+            logger.info("Restarting the firehose stream...")
 
 if __name__ == '__main__':
     main()