Test new atproto with firehose restarting

docker-compose.yml

@@ -22,10 +22,10 @@ services:
     restart: unless-stopped
     healthcheck:
       test: ["CMD", "python", "health_check.py"] # Command to check the container's health
-      interval: 60s # Time between each health check
-      timeout: 10s # Time allowed for the health check to complete
+      interval: 15m # Time between each health check
+      timeout: 30s # Time allowed for the health check to complete
       retries: 2 # Number of failed health checks before marking as unhealthy
-      start_period: 60s # Grace period before starting health checks after container starts
+      start_period: 60s # Time to wait before starting the health checks
     depends_on:
       - postgres
     labels:

firehose/data_stream.py

@@ -130,7 +130,7 @@ def _run(name, operations_callback, stream_stop_event=None):
         # If no state exists, create an initial subscription state with cursor 0
         SubscriptionState.create(service=name, cursor=0)
 
-    # Initialize the firehose client w/o a cursor for now
+    # Initialize the firehose client
     client = FirehoseSubscribeReposClient(params)
 
     def on_message_handler(message: firehose_models.MessageFrame) -> None:
@@ -163,7 +163,7 @@ def on_message_handler(message: firehose_models.MessageFrame) -> None:
             logger.info(f'Cursor -> {commit.seq}')
             # 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
+            # Persist the new cursor in the database with the last indexed timestamp
             with db.atomic():
                 SubscriptionState.update(
                     cursor=commit.seq,

firehose/health_check.py

@@ -11,7 +11,10 @@ def is_healthy():
 
     if not state or not state.last_indexed_at:
         return False
-    return (datetime.now(timezone.utc) - state.last_indexed_at) < timedelta(minutes=5)
+
+    # Check if the firehose last log is within the last 15 minutes
+    # If it is, the health check is considered healthy
+    return (datetime.now(timezone.utc) - state.last_indexed_at) < timedelta(minutes=15)
 
 if __name__ == "__main__":
     if is_healthy():

firehose/requirements.txt

@@ -1,4 +1,4 @@
 psycopg2-binary
-atproto==0.0.56
+atproto
 peewee
 python-dotenv

requirements.txt

@@ -1,4 +1,4 @@
-atproto==0.0.56
+atproto
 peewee
 Flask
 python-dotenv

scheduler/requirements.txt

@@ -1,4 +1,4 @@
-atproto==0.0.56
+atproto
 apscheduler
 peewee
 psycopg2-binary