perf: increase job concurrency, add channel routing and identity keys

spp_programs/README.rst

@@ -254,6 +254,18 @@ Dependencies
 Changelog
 =========
 
+19.0.2.0.10
+~~~~~~~~~~~
+
+- Increase parallel-safe channel limits (cycle, eligibility_manager,
+  program_manager) from 1 to 4
+- Add serial ``entitlement_approval`` channel (limit=1) for fund balance
+  safety
+- Add serial ``statistics_refresh`` channel (limit=1) to prevent
+  concurrent refresh storms
+- Add ``identity_key`` to async job dispatchers to prevent duplicate
+  submission on double-click
+
 19.0.2.0.9
 ~~~~~~~~~~
 

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.9",
+    "version": "19.0.2.0.10",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_programs/data/queue_data.xml

@@ -1,16 +1,26 @@
 <odoo>
     <record id="limit_cycle" model="queue.limit">
         <field name="name">cycle</field>
-        <field name="limit">1</field>
+        <field name="limit">4</field>
         <field name="rate_limit">0</field>
     </record>
     <record id="limit_eligibility_manager" model="queue.limit">
         <field name="name">eligibility_manager</field>
-        <field name="limit">1</field>
+        <field name="limit">4</field>
         <field name="rate_limit">0</field>
     </record>
     <record id="limit_program_manager" model="queue.limit">
         <field name="name">program_manager</field>
+        <field name="limit">4</field>
+        <field name="rate_limit">0</field>
+    </record>
+    <record id="limit_entitlement_approval" model="queue.limit">
+        <field name="name">entitlement_approval</field>
+        <field name="limit">1</field>
+        <field name="rate_limit">0</field>
+    </record>
+    <record id="limit_statistics_refresh" model="queue.limit">
+        <field name="name">statistics_refresh</field>
         <field name="limit">1</field>
         <field name="rate_limit">0</field>
     </record>

spp_programs/models/managers/cycle_manager_base.py

@@ -527,9 +527,14 @@ def _check_eligibility_async(self, cycle, beneficiaries_count):
 
         jobs = []
         for min_id, max_id in id_ranges:
-            jobs.append(self.delayable(channel="cycle")._check_eligibility(cycle, min_id=min_id, max_id=max_id))
+            jobs.append(
+                self.delayable(
+                    channel="cycle",
+                    identity_key=f"check_elig_{cycle.id}_{min_id}",
+                )._check_eligibility(cycle, min_id=min_id, max_id=max_id)
+            )
         main_job = group(*jobs)
-        main_job.on_done(self.delayable(channel="cycle").mark_check_eligibility_as_done(cycle))
+        main_job.on_done(self.delayable(channel="statistics_refresh").mark_check_eligibility_as_done(cycle))
         main_job.delay()
 
     def _check_eligibility(
@@ -607,10 +612,17 @@ def _prepare_entitlements_async(self, cycle, beneficiaries_count):
 
         jobs = []
         for min_id, max_id in id_ranges:
-            jobs.append(self.delayable(channel="cycle")._prepare_entitlements(cycle, min_id=min_id, max_id=max_id))
+            jobs.append(
+                self.delayable(
+                    channel="cycle",
+                    identity_key=f"prepare_ent_{cycle.id}_{min_id}",
+                )._prepare_entitlements(cycle, min_id=min_id, max_id=max_id)
+            )
         main_job = group(*jobs)
         main_job.on_done(
-            self.delayable(channel="cycle").mark_prepare_entitlement_as_done(cycle, _("Entitlement Ready."))
+            self.delayable(channel="statistics_refresh").mark_prepare_entitlement_as_done(
+                cycle, _("Entitlement Ready.")
+            )
         )
         main_job.delay()
 
@@ -844,15 +856,20 @@ def _add_beneficiaries_async(self, cycle, beneficiaries, state):
         jobs = []
         for i in range(0, beneficiaries_count, self.MAX_ROW_JOB_QUEUE):
             jobs.append(
-                self.delayable(channel="cycle")._add_beneficiaries(
+                self.delayable(
+                    channel="cycle",
+                    identity_key=f"add_benef_{cycle.id}_{i}",
+                )._add_beneficiaries(
                     cycle,
                     beneficiaries[i : i + self.MAX_ROW_JOB_QUEUE],
                     state,
                 )
             )
 
         main_job = group(*jobs)
-        main_job.on_done(self.delayable(channel="cycle").mark_import_as_done(cycle, _("Beneficiary import finished.")))
+        main_job.on_done(
+            self.delayable(channel="statistics_refresh").mark_import_as_done(cycle, _("Beneficiary import finished."))
+        )
         main_job.delay()
 
     def _add_beneficiaries(self, cycle, beneficiaries, state="draft", do_count=False):

spp_programs/models/managers/eligibility_manager.py

@@ -155,12 +155,13 @@ def _import_registrants_async(self, new_beneficiaries, state="draft"):
         jobs = []
         for i in range(0, len(new_beneficiaries), 10000):
             jobs.append(
-                self.delayable(channel="eligibility_manager")._import_registrants(
-                    new_beneficiaries[i : i + 10000], state
-                )
+                self.delayable(
+                    channel="eligibility_manager",
+                    identity_key=f"import_reg_{program.id}_{i}",
+                )._import_registrants(new_beneficiaries[i : i + 10000], state)
             )
         main_job = group(*jobs)
-        main_job.on_done(self.delayable(channel="eligibility_manager").mark_import_as_done())
+        main_job.on_done(self.delayable(channel="statistics_refresh").mark_import_as_done())
         main_job.delay()
 
     def mark_import_as_done(self):

spp_programs/models/managers/entitlement_manager_base.py

@@ -89,7 +89,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements):
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
             jobs.append(
-                self.delayable()._set_pending_validation_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE])
+                self.delayable(channel="entitlement_approval")._set_pending_validation_entitlements(
+                    entitlements[i : i + self.MAX_ROW_JOB_QUEUE]
+                )
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation.")))
@@ -137,7 +139,11 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count):
 
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
-            jobs.append(self.delayable()._validate_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE]))
+            jobs.append(
+                self.delayable(channel="entitlement_approval")._validate_entitlements(
+                    entitlements[i : i + self.MAX_ROW_JOB_QUEUE]
+                )
+            )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
         main_job.delay()
@@ -197,7 +203,11 @@ def _cancel_entitlements_async(self, cycle, entitlements, entitlements_count):
 
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
-            jobs.append(self.delayable()._cancel_entitlements(entitlements[i : i + self.MAX_ROW_JOB_QUEUE]))
+            jobs.append(
+                self.delayable(channel="entitlement_approval")._cancel_entitlements(
+                    entitlements[i : i + self.MAX_ROW_JOB_QUEUE]
+                )
+            )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Cancelled.")))
         main_job.delay()

spp_programs/models/managers/entitlement_manager_cash.py

@@ -319,7 +319,11 @@ def _validate_entitlements_async(self, cycle, entitlements, entitlements_count):
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
             # Needs to override
-            jobs.append(self.delayable()._validate_entitlements(cycle, entitlements[i : i + self.MAX_ROW_JOB_QUEUE]))
+            jobs.append(
+                self.delayable(channel="entitlement_approval")._validate_entitlements(
+                    cycle, entitlements[i : i + self.MAX_ROW_JOB_QUEUE]
+                )
+            )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
         main_job.delay()

spp_programs/models/managers/entitlement_manager_inkind.py

@@ -216,7 +216,9 @@ def _set_pending_validation_entitlements_async(self, cycle, entitlements_count):
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
             jobs.append(
-                self.delayable()._set_pending_validation_entitlements(cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE)
+                self.delayable(channel="entitlement_approval")._set_pending_validation_entitlements(
+                    cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE
+                )
             )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Set to Pending Validation.")))
@@ -315,7 +317,11 @@ def _validate_entitlements_async(self, cycle, entitlements_count):
 
         jobs = []
         for i in range(0, entitlements_count, self.MAX_ROW_JOB_QUEUE):
-            jobs.append(self.delayable()._validate_entitlements(cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE))
+            jobs.append(
+                self.delayable(channel="entitlement_approval")._validate_entitlements(
+                    cycle, offset=i, limit=self.MAX_ROW_JOB_QUEUE
+                )
+            )
         main_job = group(*jobs)
         main_job.on_done(self.delayable().mark_job_as_done(cycle, _("Entitlements Validated and Approved.")))
         main_job.delay()

spp_programs/models/managers/program_manager.py

@@ -199,12 +199,13 @@ def _enroll_eligible_registrants_async(self, states, members_count):
         jobs = []
         for min_id, max_id in id_ranges:
             jobs.append(
-                self.delayable(channel="program_manager")._enroll_eligible_registrants(
-                    states, min_id=min_id, max_id=max_id
-                )
+                self.delayable(
+                    channel="program_manager",
+                    identity_key=f"enroll_eligible_{program.id}_{min_id}",
+                )._enroll_eligible_registrants(states, min_id=min_id, max_id=max_id)
             )
         main_job = group(*jobs)
-        main_job.on_done(self.delayable(channel="program_manager").mark_enroll_eligible_as_done())
+        main_job.on_done(self.delayable(channel="statistics_refresh").mark_enroll_eligible_as_done())
         main_job.delay()
 
     def _enroll_eligible_registrants(self, states, offset=0, limit=None, min_id=None, max_id=None, do_count=False):

spp_programs/readme/HISTORY.md

@@ -1,3 +1,10 @@
+### 19.0.2.0.10
+
+- Increase parallel-safe channel limits (cycle, eligibility_manager, program_manager) from 1 to 4
+- Add serial `entitlement_approval` channel (limit=1) for fund balance safety
+- Add serial `statistics_refresh` channel (limit=1) to prevent concurrent refresh storms
+- Add `identity_key` to async job dispatchers to prevent duplicate submission on double-click
+
 ### 19.0.2.0.9
 
 - Add context flags (`skip_registrant_statistics`, `skip_program_statistics`) to suppress expensive computed field recomputation during bulk operations

spp_programs/static/description/index.html

@@ -658,6 +658,19 @@ <h2><a class="toc-backref" href="#toc-entry-1">Changelog</a></h2>
 </div>
 </div>
 <div class="section" id="section-1">
+<h1>19.0.2.0.10</h1>
+<ul class="simple">
+<li>Increase parallel-safe channel limits (cycle, eligibility_manager,
+program_manager) from 1 to 4</li>
+<li>Add serial <tt class="docutils literal">entitlement_approval</tt> channel (limit=1) for fund balance
+safety</li>
+<li>Add serial <tt class="docutils literal">statistics_refresh</tt> channel (limit=1) to prevent
+concurrent refresh storms</li>
+<li>Add <tt class="docutils literal">identity_key</tt> to async job dispatchers to prevent duplicate
+submission on double-click</li>
+</ul>
+</div>
+<div class="section" id="section-2">
 <h1>19.0.2.0.9</h1>
 <ul class="simple">
 <li>Add context flags (<tt class="docutils literal">skip_registrant_statistics</tt>,
@@ -670,7 +683,7 @@ <h1>19.0.2.0.9</h1>
 <tt class="docutils literal">_compute_has_members</tt></li>
 </ul>
 </div>
-<div class="section" id="section-2">
+<div class="section" id="section-3">
 <h1>19.0.2.0.8</h1>
 <ul class="simple">
 <li>Replace OFFSET pagination with NTILE-based ID-range batching in all
@@ -681,7 +694,7 @@ <h1>19.0.2.0.8</h1>
 program and cycle</li>
 </ul>
 </div>
-<div class="section" id="section-3">
+<div class="section" id="section-4">
 <h1>19.0.2.0.7</h1>
 <ul class="simple">
 <li>Bulk membership creation using raw SQL INSERT ON CONFLICT DO NOTHING
@@ -690,7 +703,7 @@ <h1>19.0.2.0.7</h1>
 <tt class="docutils literal">_add_beneficiaries</tt> with bulk SQL path</li>
 </ul>
 </div>
-<div class="section" id="section-4">
+<div class="section" id="section-5">
 <h1>19.0.2.0.6</h1>
 <ul class="simple">
 <li>Remove unused entitlement_base_model.py (dead code, never imported)</li>
@@ -699,34 +712,34 @@ <h1>19.0.2.0.6</h1>
 payment, and fund tests (172 → 492 tests)</li>
 </ul>
 </div>
-<div class="section" id="section-5">
+<div class="section" id="section-6">
 <h1>19.0.2.0.5</h1>
 <ul class="simple">
 <li>Batch create entitlements and payments instead of one-by-one ORM
 creates</li>
 </ul>
 </div>
-<div class="section" id="section-6">
+<div class="section" id="section-7">
 <h1>19.0.2.0.4</h1>
 <ul class="simple">
 <li>Fetch fund balance once per approval batch instead of per entitlement</li>
 </ul>
 </div>
-<div class="section" id="section-7">
+<div class="section" id="section-8">
 <h1>19.0.2.0.3</h1>
 <ul class="simple">
 <li>Replace cycle computed fields (total_amount, entitlements_count,
 approval flags) with SQL aggregation queries</li>
 </ul>
 </div>
-<div class="section" id="section-8">
+<div class="section" id="section-9">
 <h1>19.0.2.0.2</h1>
 <ul class="simple">
 <li>Add composite indexes for frequent query patterns on entitlements and
 program memberships</li>
 </ul>
 </div>
-<div class="section" id="section-9">
+<div class="section" id="section-10">
 <h1>19.0.2.0.1</h1>
 <ul class="simple">
 <li>Replace Python-level uniqueness checks with SQL UNIQUE constraints for
@@ -735,7 +748,7 @@ <h1>19.0.2.0.1</h1>
 constraint creation</li>
 </ul>
 </div>
-<div class="section" id="section-10">
+<div class="section" id="section-11">
 <h1>19.0.2.0.0</h1>
 <ul class="simple">
 <li>Initial migration to OpenSPP2</li>

spp_programs/tests/__init__.py

@@ -35,3 +35,4 @@
 from . import test_bulk_membership
 from . import test_keyset_pagination
 from . import test_canary_patterns
+from . import test_concurrency