Improve Active Workers table and recover stale tasks

Backend:
- Add _recover_stale_tasks() to reset tasks stuck in 'in_progress'
  from previous crashes (tasks >5 min old get reset to pending)
- Called automatically on startup

Frontend:
- Fix dark mode colors using CSS variables instead of hardcoded values
- Add elapsed time column showing how long task has been running
- Add spinning indicator next to package name
- Add status badge (Running/Stale?)
- Highlight stale tasks (>5 min) in amber
- Auto-updates every 5 seconds with existing refresh
This commit is contained in:
Mondo Diaz
2026-02-02 14:29:17 -06:00
parent a485852a6f
commit 1f98caa73c
3 changed files with 143 additions and 23 deletions

View File

@@ -33,6 +33,45 @@ _cache_worker_running: bool = False
_dispatcher_thread: Optional[threading.Thread] = None
def _recover_stale_tasks():
"""
Recover tasks stuck in 'in_progress' state from a previous crash.
Called on startup to reset tasks that were being processed when
the server crashed. Resets them to 'pending' so they can be retried.
"""
db = SessionLocal()
try:
# Find tasks that have been in_progress for more than 5 minutes
# These are likely from a crashed worker
stale_threshold = datetime.utcnow() - timedelta(minutes=5)
stale_count = (
db.query(PyPICacheTask)
.filter(
PyPICacheTask.status == "in_progress",
or_(
PyPICacheTask.started_at == None,
PyPICacheTask.started_at < stale_threshold,
),
)
.update(
{
"status": "pending",
"started_at": None,
}
)
)
db.commit()
if stale_count > 0:
logger.warning(f"Recovered {stale_count} stale in_progress tasks from previous crash")
except Exception as e:
logger.error(f"Error recovering stale tasks: {e}")
finally:
db.close()
def init_cache_worker_pool(max_workers: Optional[int] = None):
"""
Initialize the cache worker pool. Called on app startup.
@@ -47,6 +86,9 @@ def init_cache_worker_pool(max_workers: Optional[int] = None):
logger.warning("Cache worker pool already initialized")
return
# Recover any stale tasks from previous crash before starting workers
_recover_stale_tasks()
workers = max_workers or settings.PYPI_CACHE_WORKERS
_cache_worker_pool = ThreadPoolExecutor(
max_workers=workers,