Merge pull request #188 from OpenSPP/fix/cycle-stale-lock-recovery

fix(spp_programs): release stuck cycle/program lock when async pipeline fails

Author: gonzalesedwin1123

spp_programs/__manifest__.py

@@ -4,7 +4,7 @@
     "name": "OpenSPP Programs",
     "summary": "Manage programs, cycles, beneficiary enrollment, entitlements (cash and in-kind), payments, and fund tracking for social protection.",
     "category": "OpenSPP/Core",
-    "version": "19.0.2.0.11",
+    "version": "19.0.2.1.0",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_programs/models/cycle.py

@@ -1062,6 +1062,27 @@ def _get_related_job_domain(self):
         related_jobs = jobs.filtered(lambda r: self in r.args[0])
         return [("id", "in", related_jobs.ids)]
 
+    def action_force_unlock(self):
+        """Manager-only escape hatch: clear a stuck "Operation in progress" lock.
+
+        Use when an async pipeline (entitlement processing, payment prep, etc.)
+        died without firing its on_done/on_error callback — for example after
+        a hard server restart or before this fix was deployed. Posts an audit
+        line to chatter so admins can see who unstuck the cycle.
+        """
+        for rec in self:
+            if not rec.is_locked:
+                continue
+            previous_reason = rec.locked_reason
+            rec.write({"is_locked": False, "locked_reason": False})
+            rec.message_post(
+                body=_(
+                    "Lock manually cleared by %(user)s. Previous reason: %(reason)s",
+                    user=self.env.user.display_name,
+                    reason=previous_reason or _("(none)"),
+                )
+            )
+
     def unlink(self):
         # Admin also not able to delete the cycle bcz of beneficiaries mapped
         # So this function common for who are all having delete access.

spp_programs/models/managers/cycle_manager_base.py

@@ -322,13 +322,24 @@ def mark_import_as_done(self, cycle, msg):
         :return:
         """
         self.ensure_one()
-        cycle.is_locked = False
-        cycle.locked_reason = None
-        cycle.message_post(body=msg)
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post completion chatter on cycle %s", cycle.id)
 
         # Refresh statistics after bulk operations
         cycle.refresh_statistics()
 
+    def mark_import_as_failed(self, cycle, msg):
+        """Run via on_error() when async beneficiary import fails."""
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post failure chatter on cycle %s", cycle.id)
+
     def mark_prepare_entitlement_as_done(self, cycle, msg):
         """Complete the preparation of entitlements.
         Base :meth:`mark_prepare_entitlement_as_done`.
@@ -340,13 +351,24 @@ def mark_prepare_entitlement_as_done(self, cycle, msg):
         :return:
         """
         self.ensure_one()
-        cycle.is_locked = False
-        cycle.locked_reason = None
-        cycle.message_post(body=msg)
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post completion chatter on cycle %s", cycle.id)
 
         # Update Statistics
         cycle._compute_entitlements_count()
 
+    def mark_prepare_entitlement_as_failed(self, cycle, msg):
+        """Run via on_error() when async entitlement preparation fails."""
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post failure chatter on cycle %s", cycle.id)
+
     def mark_check_eligibility_as_done(self, cycle):
         """Complete the enrollment of eligible beneficiaries.
         Base :meth:`mark_check_eligibility_as_done`.
@@ -356,13 +378,25 @@ def mark_check_eligibility_as_done(self, cycle):
         :param cycle: A recordset of cycle
         :return:
         """
-        cycle.is_locked = False
-        cycle.locked_reason = None
-        cycle.message_post(body=_("Eligibility check finished."))
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=_("Eligibility check finished."))
+        except Exception:
+            _logger.exception("Failed to post completion chatter on cycle %s", cycle.id)
 
         # Compute Statistics
         cycle._compute_members_count()
 
+    def mark_check_eligibility_as_failed(self, cycle):
+        """Run via on_error() when async eligibility check fails."""
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=_("Eligibility check failed."))
+        except Exception:
+            _logger.exception("Failed to post failure chatter on cycle %s", cycle.id)
+
 
 class DefaultCycleManager(models.Model):
     _name = "spp.cycle.manager.default"
@@ -535,6 +569,7 @@ def _check_eligibility_async(self, cycle, beneficiaries_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable(channel="statistics_refresh").mark_check_eligibility_as_done(cycle))
+        main_job.on_error(self.delayable(channel="statistics_refresh").mark_check_eligibility_as_failed(cycle))
         main_job.delay()
 
     def _check_eligibility(
@@ -624,6 +659,11 @@ def _prepare_entitlements_async(self, cycle, beneficiaries_count):
                 cycle, _("Entitlement Ready.")
             )
         )
+        main_job.on_error(
+            self.delayable(channel="statistics_refresh").mark_prepare_entitlement_as_failed(
+                cycle, _("Entitlement preparation failed.")
+            )
+        )
         main_job.delay()
 
     def _prepare_entitlements(self, cycle, offset=0, limit=None, min_id=None, max_id=None, do_count=False):
@@ -870,6 +910,9 @@ def _add_beneficiaries_async(self, cycle, beneficiaries, state):
         main_job.on_done(
             self.delayable(channel="statistics_refresh").mark_import_as_done(cycle, _("Beneficiary import finished."))
         )
+        main_job.on_error(
+            self.delayable(channel="statistics_refresh").mark_import_as_failed(cycle, _("Beneficiary import failed."))
+        )
         main_job.delay()
 
     def _add_beneficiaries(self, cycle, beneficiaries, state="draft", do_count=False):

spp_programs/models/managers/entitlement_manager_base.py

@@ -95,6 +95,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation.")))
+        main_job.on_error(
+            self.delayable().mark_job_as_failed(cycle, _("Setting entitlements to pending validation failed."))
+        )
         main_job.delay()
 
     def _set_pending_validation_entitlements(self, entitlements):
@@ -146,6 +149,9 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
+        main_job.on_error(
+            self.delayable().mark_job_as_failed(cycle, _("Validation and approval of entitlements failed."))
+        )
         main_job.delay()
 
     def _validate_entitlements(self, entitlements):
@@ -210,6 +216,7 @@ def _cancel_entitlements_async(self, cycle, entitlements, entitlements_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Cancelled.")))
+        main_job.on_error(self.delayable().mark_job_as_failed(cycle, _("Cancelling entitlements failed.")))
         main_job.delay()
 
     def _cancel_entitlements(self, entitlements):
@@ -233,9 +240,30 @@ def mark_job_as_done(self, cycle, msg):
         :return:
         """
         self.ensure_one()
-        cycle.is_locked = False
-        cycle.locked_reason = None
-        cycle.message_post(body=msg)
+        # Clear the lock first so a chatter-side failure can't leave the
+        # cycle stuck with "Operation in progress".
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post completion chatter on cycle %s", cycle.id)
+
+    def mark_job_as_failed(self, cycle, msg):
+        """Run via on_error() when the async pipeline fails.
+
+        Clears the cycle lock and posts a failure note to chatter so the
+        user understands the operation finished without success — instead
+        of the lock remaining set indefinitely (the bug this fix targets).
+
+        :param cycle: A recordset of cycle
+        :param msg: A string to be posted in the chatter
+        """
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post failure chatter on cycle %s", cycle.id)
 
     def open_entitlements_form(self, cycle):
         """

spp_programs/models/managers/entitlement_manager_cash.py

@@ -326,6 +326,9 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
+        main_job.on_error(
+            self.delayable().mark_job_as_failed(cycle, _("Validation and approval of entitlements failed."))
+        )
         main_job.delay()
 
     def _validate_entitlements(self, cycle, entitlements):

spp_programs/models/managers/entitlement_manager_inkind.py

@@ -222,6 +222,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation.")))
+        main_job.on_error(
+            self.delayable().mark_job_as_failed(cycle, _("Setting entitlements to pending validation failed."))
+        )
         main_job.delay()
 
     def _set_pending_validation_entitlements(self, cycle, offset=0, limit=None):
@@ -324,6 +327,9 @@ def _validate_entitlements_async(self, cycle, entitlements_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
+        main_job.on_error(
+            self.delayable().mark_job_as_failed(cycle, _("Validation and approval of entitlements failed."))
+        )
         main_job.delay()
 
     def _validate_entitlements(self, cycle, offset=0, limit=None):

spp_programs/models/managers/payment_manager.py

@@ -71,9 +71,20 @@ def mark_job_as_done(self, cycle, msg):
         :return:
         """
         self.ensure_one()
-        cycle.is_locked = False
-        cycle.locked_reason = None
-        cycle.message_post(body=msg)
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post completion chatter on cycle %s", cycle.id)
+
+    def mark_job_as_failed(self, cycle, msg):
+        """Run via on_error() when the async payment pipeline fails."""
+        self.ensure_one()
+        cycle.write({"is_locked": False, "locked_reason": False})
+        try:
+            cycle.message_post(body=msg)
+        except Exception:
+            _logger.exception("Failed to post failure chatter on cycle %s", cycle.id)
 
 
 class DefaultFilePaymentManager(models.Model):
@@ -278,6 +289,7 @@ def _prepare_payments_async(self, cycle, entitlements, entitlements_count):
         ]
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Prepared payments.")))
+        main_job.on_error(self.delayable().mark_job_as_failed(cycle, _("Preparing payments failed.")))
         main_job.delay()
 
     def send_payments(self, batches):
@@ -392,6 +404,7 @@ def _send_payments_async(self, cycle, batches):
         ]
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Send payments completed.")))
+        main_job.on_error(self.delayable().mark_job_as_failed(cycle, _("Sending payments failed.")))
         main_job.delay()
 
     @api.model

spp_programs/models/managers/program_manager.py

@@ -73,13 +73,26 @@ def mark_enroll_eligible_as_done(self):
         :return:
         """
         self.ensure_one()
-        self.program_id.is_locked = False
-        self.program_id.locked_reason = None
-        self.program_id.message_post(body=_("Eligibility check finished."))
+        program = self.program_id
+        program.write({"is_locked": False, "locked_reason": False})
+        try:
+            program.message_post(body=_("Eligibility check finished."))
+        except Exception:
+            _logger.exception("Failed to post completion chatter on program %s", program.id)
 
         # Compute Statistics
-        self.program_id._compute_eligible_beneficiary_count()
-        self.program_id._compute_beneficiary_count()
+        program._compute_eligible_beneficiary_count()
+        program._compute_beneficiary_count()
+
+    def mark_enroll_eligible_as_failed(self):
+        """Run via on_error() when async eligibility enrollment fails."""
+        self.ensure_one()
+        program = self.program_id
+        program.write({"is_locked": False, "locked_reason": False})
+        try:
+            program.message_post(body=_("Eligibility check failed."))
+        except Exception:
+            _logger.exception("Failed to post failure chatter on program %s", program.id)
 
 
 class DefaultProgramManager(models.Model):
@@ -215,6 +228,7 @@ def _enroll_eligible_registrants_async(self, states, members_count):
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable(channel="statistics_refresh").mark_enroll_eligible_as_done())
+        main_job.on_error(self.delayable(channel="statistics_refresh").mark_enroll_eligible_as_failed())
         main_job.delay()
 
     def _enroll_eligible_registrants(self, states, offset=0, limit=None, min_id=None, max_id=None, do_count=False):

spp_programs/models/programs.py

@@ -725,6 +725,25 @@ def _get_related_job_domain(self):
         related_jobs = jobs.filtered(lambda r: self in r.records.program_id)
         return [("id", "in", related_jobs.ids)]
 
+    def action_force_unlock(self):
+        """Manager-only escape hatch: clear a stuck "Operation in progress" lock.
+
+        Use when an async pipeline died without firing its on_done/on_error
+        callback. Posts an audit line to chatter for traceability.
+        """
+        for rec in self:
+            if not rec.is_locked:
+                continue
+            previous_reason = rec.locked_reason
+            rec.write({"is_locked": False, "locked_reason": False})
+            rec.message_post(
+                body=_(
+                    "Lock manually cleared by %(user)s. Previous reason: %(reason)s",
+                    user=self.env.user.display_name,
+                    reason=previous_reason or _("(none)"),
+                )
+            )
+
     @api.constrains(
         "entitlement_manager_ids",
         "program_manager_ids",

spp_programs/tests/__init__.py

@@ -37,3 +37,4 @@
 from . import test_keyset_pagination
 from . import test_canary_patterns
 from . import test_concurrency
+from . import test_async_lock_recovery