diff --git a/invoice_az/background_tasks.py b/invoice_az/background_tasks.py index bcaaa4d..6b5e830 100644 --- a/invoice_az/background_tasks.py +++ b/invoice_az/background_tasks.py @@ -5,18 +5,21 @@ # every long-running worker writes a progress snapshot here, the widget reads it # back and can request cancellation. # +# Storage: a per-user Redis HASH, one field per job_id. Each field is written/ +# deleted independently (HSET/HDEL are atomic per field), so MULTIPLE concurrent +# jobs for the same user never clobber each other. (An earlier version stored the +# whole registry as one pickled dict and did read-modify-write, which raced under +# concurrent loads — phantom/flickering rows. The hash fixes that.) +# # Design notes: # * Keyed by USER. `get_active_background_tasks()` / `cancel_background_task()` # derive the user from `frappe.session.user` — the client never passes a user, # so one user can neither see nor cancel another user's jobs. -# * The registry is the source of truth (survives page reloads). Realtime pings -# are only a "something changed, re-fetch" hint — anything published before the -# client subscribes is lost, so the widget ALSO polls this on a timer. +# * `hgetall` reads straight from Redis (no frappe.local cache), so the widget +# always sees a fresh, consistent snapshot. The registry is the source of +# truth (survives page reloads); realtime pings are just "re-fetch" hints. # * Cancellation is best-effort: the worker checks `is_cancel_requested()` # between steps and bails out after finishing the current step. -# * Generalising this to the other bulk imports (sales/purchase/vat/amas) is -# ~10 lines each: new_job_id() -> register_task -> update_task -> -# is_cancel_requested -> finish_task. import uuid @@ -24,7 +27,7 @@ import frappe REALTIME_EVENT = "background_tasks_changed" -_REGISTRY_TTL = 7200 # the per-user registry key self-expires after 2h +_REGISTRY_TTL = 7200 # the per-user hash self-expires after 2h (crash safety net) _CANCEL_TTL = 3600 @@ -38,12 +41,32 @@ def _cancel_key(job_id): return f"invoice_az:bg_cancel:{job_id}" -def _read(user): - return frappe.cache().get_value(_registry_key(user)) or {} +def _touch_ttl(user): + """Keep the per-user hash key alive while jobs are running.""" + try: + cache = frappe.cache() + cache.expire(cache.make_key(_registry_key(user)), _REGISTRY_TTL) + except Exception: + pass # TTL is only a crash safety net; finish_task removes tasks normally -def _write(user, tasks): - frappe.cache().set_value(_registry_key(user), tasks, expires_in_sec=_REGISTRY_TTL) +def _all_tasks(user): + """Fresh-from-Redis list of the user's task snapshots (hgetall bypasses + frappe.local cache).""" + return list((frappe.cache().hgetall(_registry_key(user)) or {}).values()) + + +def _get_task(user, job_id): + return frappe.cache().hget(_registry_key(user), job_id) + + +def _put_task(user, job_id, task): + frappe.cache().hset(_registry_key(user), job_id, task) + _touch_ttl(user) + + +def _del_task(user, job_id): + frappe.cache().hdel(_registry_key(user), job_id) def _ping(user, payload=None): @@ -62,8 +85,7 @@ def register_task(job_id, user, feature, label, total=0, cancellable=True): """Add a task to the user's registry in the `running` state.""" if not job_id: return - tasks = _read(user) - tasks[job_id] = { + _put_task(user, job_id, { "job_id": job_id, "feature": feature, "label": label, @@ -73,8 +95,7 @@ def register_task(job_id, user, feature, label, total=0, cancellable=True): "status": "running", # running | cancelling | done | error | cancelled "cancellable": bool(cancellable), "message": "", - } - _write(user, tasks) + }) _ping(user) @@ -83,8 +104,7 @@ def update_task(job_id, user, current=None, total=None, percent=None, message=No current/total unless an explicit `percent` is given.""" if not job_id: return - tasks = _read(user) - task = tasks.get(job_id) + task = _get_task(user, job_id) if not task: return if total is not None: @@ -97,8 +117,7 @@ def update_task(job_id, user, current=None, total=None, percent=None, message=No task["percent"] = max(0, min(100, int(percent))) elif task.get("total"): task["percent"] = max(0, min(100, int(round(task["current"] / task["total"] * 100)))) - tasks[job_id] = task - _write(user, tasks) + _put_task(user, job_id, task) _ping(user) @@ -106,20 +125,20 @@ def finish_task(job_id, user, status="done", message=None): """Remove the task from the registry and flash its final state to the widget. Removing it immediately keeps the poll-based fallback consistent: a task that - is gone from the registry is simply gone from the widget's list, so a lost - realtime ping can't leave a stale "running" row behind.""" + is gone from the registry is simply gone from the widget's list.""" if not job_id: return - tasks = _read(user) - task = tasks.pop(job_id, None) - _write(user, tasks) + task = _get_task(user, job_id) + _del_task(user, job_id) frappe.cache().delete_value(_cancel_key(job_id)) if task: task["status"] = status - if message is not None: - task["message"] = message if status == "done": task["percent"] = 100 + task["current"] = task.get("total") or task.get("current") or 0 + task["message"] = message if message is not None else "Done" + elif message is not None: + task["message"] = message _ping(user, {"finished": task}) else: _ping(user) @@ -128,15 +147,13 @@ def finish_task(job_id, user, status="done", message=None): def request_cancel(job_id, user): """Set the cancel flag for a task owned by `user`. Returns False if the task is not in this user's registry (so cross-user cancellation is impossible).""" - tasks = _read(user) - task = tasks.get(job_id) + task = _get_task(user, job_id) if not task: return False frappe.cache().set_value(_cancel_key(job_id), 1, expires_in_sec=_CANCEL_TTL) task["status"] = "cancelling" task["message"] = "Cancellation requested" - tasks[job_id] = task - _write(user, tasks) + _put_task(user, job_id, task) _ping(user) return True @@ -158,7 +175,7 @@ def is_cancel_requested(job_id): @frappe.whitelist() def get_active_background_tasks(): """Return the current user's in-flight tasks. Source of truth for the widget.""" - return list(_read(frappe.session.user).values()) + return _all_tasks(frappe.session.user) @frappe.whitelist()