fix: improve setup, logging, and task resiliency [no ci]

Author: PromiseFru

install.sh

@@ -95,7 +95,11 @@ clone_repository() {
 setup_virtualenv() {
   log "Setting up virtual environment"
   cd "$INSTALL_DIR"
-  [ -d "venv" ] || python3 -m venv venv || error "Failed to create venv"
+  if [ -d "venv" ]; then
+    log "Removing existing venv for a clean rebuild"
+    rm -rf venv
+  fi
+  python3 -m venv venv || error "Failed to create venv"
   venv/bin/pip install --quiet --upgrade pip || error "Failed to upgrade pip"
   venv/bin/pip install --quiet -r requirements.txt || error "Failed to install Python dependencies"
 }

logutils.py

@@ -13,10 +13,14 @@
 if not isinstance(numeric_level, int):
     raise ValueError(f"Invalid log level: {LOG_LEVEL}")
 
-logging.basicConfig(
-    level=numeric_level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+_LOG_FORMAT = (
+    "%(name)s - %(levelname)s - %(message)s"
+    if os.getenv("JOURNAL_STREAM")
+    else "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 
+logging.basicConfig(level=numeric_level, format=_LOG_FORMAT)
+
 
 def get_logger(name: str = None) -> logging.Logger:
     """

publications.py

@@ -308,18 +308,19 @@ def _store_segment_and_try_join(
 
         try:
             joined = rrs.V1Payloads.join(segment_data)
-        except Exception:
-            logger.exception("Failed to join segments for session %s.", sess_id)
+        except Exception as exc:
             logger.warning(
-                "Session %s incomplete: %d segments saved.", sess_id, len(segment_data)
+                "Session %s not ready yet (%d segment(s) stored so far): %s: %s",
+                sess_id,
+                len(segment_data),
+                type(exc).__name__,
+                exc or "-",
             )
             return None
 
         delete_session(payload_session, self.session)
         logger.info(
-            "Session %s assembled successfully from %d segments.",
-            sess_id,
-            len(segment_data),
+            "Session %s assembled from %d segments.", sess_id, len(segment_data)
         )
         return joined
 

tasks/celery_app.py

@@ -85,6 +85,7 @@ def make_celery() -> Celery:
         worker_enable_remote_control=False,
         worker_concurrency=concurrency,
         task_acks_late=True,
+        task_reject_on_worker_lost=True,
         worker_prefetch_multiplier=1,
         task_ignore_result=True,
         beat_schedule_filename=schedule_path,