fix(db): concurrency-safe set_user_resource_selection + enforce OpenCRE (Part of #586)

application/database/db.py

@@ -1352,20 +1352,43 @@ def set_user_resource_selection(
         for name in standard_names:
             if name not in deduped:
                 deduped.append(name)
-        self.session.query(UserResourceSelection).filter(
-            UserResourceSelection.user_id == user_id
-        ).delete()
-        for name in deduped:
-            self.session.add(
-                UserResourceSelection(
-                    id=generate_uuid(),
-                    user_id=user_id,
-                    standard_name=name,
-                    created_at=now,
+
+        def _replace() -> List[str]:
+            # Serialize concurrent replacements for this user by locking the
+            # parent User row: a second PUT for the same user waits here until
+            # ours commits, then its DELETE sees our rows. Without this, two
+            # *disjoint* selections could both commit (no IntegrityError) and
+            # merge, breaking replacement semantics. FOR UPDATE is a no-op on
+            # SQLite (dev/tests).
+            self.session.query(User).filter(
+                User.id == user_id
+            ).with_for_update().first()
+            self.session.query(UserResourceSelection).filter(
+                UserResourceSelection.user_id == user_id
+            ).delete()
+            for name in deduped:
+                self.session.add(
+                    UserResourceSelection(
+                        id=generate_uuid(),
+                        user_id=user_id,
+                        standard_name=name,
+                        created_at=now,
+                    )
                 )
-            )
-        self.session.commit()
-        return self.get_user_resource_selection(user_id)
+            # Capture our selection while still holding the lock, so the return
+            # value can't reflect a later concurrent write.
+            stored = self.get_user_resource_selection(user_id)
+            self.session.commit()
+            return stored
+
+        try:
+            return _replace()
+        except IntegrityError:
+            # Same-key collision on uq_user_resource_selection; roll back and
+            # retry the replace exactly once (mirrors upsert_user). A second
+            # failure propagates — no retry loop.
+            self.session.rollback()
+            return _replace()
 
     def __get_external_links(self) -> List[Tuple[CRE, Node, str]]:
         external_links: List[Tuple[CRE, Node, str]] = []

application/tests/user_model_test.py

@@ -1,7 +1,10 @@
 """Tests for user persistence and per-user resource selection (issue #586, RFC #876 TODO 1/2)."""
 
 import os
+import threading
 import unittest
+from typing import Any, List
+from unittest.mock import patch
 
 from sqlalchemy.exc import IntegrityError
 
@@ -159,6 +162,107 @@ def test_deleting_user_cascades_to_selection(self) -> None:
         sqla.session.commit()
         self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 0)
 
+    def test_set_resource_selection_recovers_from_integrity_error(self) -> None:
+        # A concurrent PUT can make the first commit raise IntegrityError on
+        # uq_user_resource_selection. The method must roll back and retry once,
+        # then return the correct selection.
+        user = self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+        real_commit = self.collection.session.commit
+        calls = {"n": 0}
+
+        def flaky_commit(*args: Any, **kwargs: Any) -> None:
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise IntegrityError(
+                    "stmt", {}, Exception("uq_user_resource_selection")
+                )
+            real_commit()
+
+        with patch.object(self.collection.session, "commit", side_effect=flaky_commit):
+            result = self.collection.set_user_resource_selection(
+                user.id, ["ASVS", "CWE"]
+            )
+
+        self.assertEqual(sorted(result), ["ASVS", "CWE"])
+        self.assertEqual(calls["n"], 2)  # retried exactly once
+
+    def test_set_resource_selection_reraises_on_persistent_integrity_error(
+        self,
+    ) -> None:
+        # If the retry also fails, the error propagates — the recovery must not
+        # loop indefinitely (retry exactly once).
+        user = self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+
+        def always_raise(*args: Any, **kwargs: Any) -> None:
+            raise IntegrityError("stmt", {}, Exception("uq_user_resource_selection"))
+
+        with patch.object(self.collection.session, "commit", side_effect=always_raise):
+            with self.assertRaises(IntegrityError):
+                self.collection.set_user_resource_selection(user.id, ["ASVS"])
+        self.collection.session.rollback()
+
+    def test_set_resource_selection_blocks_on_locked_user_row(self) -> None:
+        # The replacement serializes per user by locking the User row: a second
+        # writer must WAIT for the first to commit (so its DELETE then sees the
+        # first's rows, instead of the two selections merging). Reproduced
+        # deterministically: hold the user-row lock here, then assert a worker's
+        # set_user_resource_selection blocks until we release it. Postgres-only —
+        # FOR UPDATE is a no-op on SQLite, so nothing would block there.
+        if "postgresql" not in str(sqla.engine.url):
+            self.skipTest("row-lock serialization requires Postgres (FOR UPDATE)")
+
+        user = self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+        user_id = user.id
+        self.collection.set_user_resource_selection(user_id, ["ASVS"])  # seed
+        self.collection.session.rollback()
+
+        # Acquire and HOLD the user-row lock on this session (open transaction).
+        self.collection.session.query(db.User).filter(
+            db.User.id == user_id
+        ).with_for_update().first()
+
+        done = threading.Event()
+        errors: List[Exception] = []
+
+        def worker() -> None:
+            with self.app.app_context():
+                try:
+                    # Empty replacement = DELETE only, no INSERT. So the only
+                    # thing that can make this touch (and block on) the User row
+                    # is the method's own FOR UPDATE — which isolates it from the
+                    # FK lock an INSERT would otherwise take.
+                    db.Node_collection().set_user_resource_selection(user_id, [])
+                except Exception as e:  # noqa: BLE001 - report worker failures to test
+                    errors.append(e)
+                finally:
+                    sqla.session.remove()
+                    done.set()
+
+        t = threading.Thread(target=worker)
+        t.start()
+
+        # While we hold the lock the worker must block — it cannot finish.
+        self.assertFalse(
+            done.wait(timeout=2), "worker did not block on the user-row lock"
+        )
+
+        # Release the lock; the worker now proceeds and completes.
+        self.collection.session.rollback()
+        self.assertTrue(
+            done.wait(timeout=15), "worker did not finish after lock release"
+        )
+        t.join(timeout=5)
+
+        self.assertEqual(errors, [])
+        self.collection.session.rollback()  # fresh snapshot
+        self.assertEqual(self.collection.get_user_resource_selection(user_id), [])
+
 
 if __name__ == "__main__":
     unittest.main()

application/tests/user_resources_api_test.py

@@ -12,6 +12,7 @@
 
 from application import create_app, sqla
 from application.database import db
+from application.utils.gap_analysis import OPENCRE_STANDARD_NAME
 
 
 class TestUserResourcesApi(unittest.TestCase):
@@ -145,11 +146,13 @@ def test_put_persists_and_returns_selection(self) -> None:
                 )
                 self.assertEqual(resp.status_code, 200)
                 self.assertEqual(
-                    sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"]
+                    sorted(json.loads(resp.data)["selected"]),
+                    sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]),
                 )
                 get = client.get("/rest/v1/user/resources")
                 self.assertEqual(
-                    sorted(json.loads(get.data)["selected"]), ["ASVS", "CWE"]
+                    sorted(json.loads(get.data)["selected"]),
+                    sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]),
                 )
 
     def test_put_replaces_previous_selection(self) -> None:
@@ -169,7 +172,10 @@ def test_put_replaces_previous_selection(self) -> None:
                 self._login(client, "sub-1", "U")
                 client.put("/rest/v1/user/resources", json={"selected": ["SAMM"]})
                 get = client.get("/rest/v1/user/resources")
-                self.assertEqual(json.loads(get.data)["selected"], ["SAMM"])
+                self.assertEqual(
+                    sorted(json.loads(get.data)["selected"]),
+                    sorted(["SAMM", OPENCRE_STANDARD_NAME]),
+                )
 
     def test_put_dedupes_input(self) -> None:
         self.collection.upsert_user(
@@ -190,7 +196,8 @@ def test_put_dedupes_input(self) -> None:
                     json={"selected": ["ASVS", "ASVS", "CWE"]},
                 )
                 self.assertEqual(
-                    sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"]
+                    sorted(json.loads(resp.data)["selected"]),
+                    sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]),
                 )
 
     def test_put_trims_and_dedupes_whitespace_variants(self) -> None:
@@ -215,9 +222,73 @@ def test_put_trims_and_dedupes_whitespace_variants(self) -> None:
                 )
                 self.assertEqual(resp.status_code, 200)
                 self.assertEqual(
-                    sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"]
+                    sorted(json.loads(resp.data)["selected"]),
+                    sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]),
+                )
+        # ASVS, CWE, and the always-injected OpenCRE.
+        self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 3)
+
+    def test_put_always_persists_opencre(self) -> None:
+        # A non-empty selection without OpenCRE gets OpenCRE injected at write.
+        self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "CRE_ENABLE_LOGIN": "1",
+                "CRE_ENABLE_MYOPENCRE": "1",
+                "INSECURE_REQUESTS": "1",
+            },
+        ):
+            with self.app.test_client() as client:
+                self._login(client, "sub-1", "U")
+                resp = client.put(
+                    "/rest/v1/user/resources", json={"selected": ["ASVS"]}
+                )
+                self.assertEqual(resp.status_code, 200)
+                self.assertIn(OPENCRE_STANDARD_NAME, json.loads(resp.data)["selected"])
+
+    def test_put_opencre_not_duplicated(self) -> None:
+        self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "CRE_ENABLE_LOGIN": "1",
+                "CRE_ENABLE_MYOPENCRE": "1",
+                "INSECURE_REQUESTS": "1",
+            },
+        ):
+            with self.app.test_client() as client:
+                self._login(client, "sub-1", "U")
+                resp = client.put(
+                    "/rest/v1/user/resources",
+                    json={"selected": ["ASVS", OPENCRE_STANDARD_NAME]},
                 )
-        self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 2)
+                selected = json.loads(resp.data)["selected"]
+                self.assertEqual(selected.count(OPENCRE_STANDARD_NAME), 1)
+
+    def test_put_empty_selection_stays_empty(self) -> None:
+        # Empty PUT must remain [] (PR3 treats [] as "show all") — OpenCRE must
+        # NOT be injected, or that would become "show only OpenCRE".
+        self.collection.upsert_user(
+            google_sub="sub-1", email="a@x.com", display_name="U"
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "CRE_ENABLE_LOGIN": "1",
+                "CRE_ENABLE_MYOPENCRE": "1",
+                "INSECURE_REQUESTS": "1",
+            },
+        ):
+            with self.app.test_client() as client:
+                self._login(client, "sub-1", "U")
+                resp = client.put("/rest/v1/user/resources", json={"selected": []})
+                self.assertEqual(resp.status_code, 200)
+                self.assertEqual(json.loads(resp.data)["selected"], [])
 
     def test_put_400_on_invalid_body(self) -> None:
         self.collection.upsert_user(

application/web/web_main.py

@@ -1382,6 +1382,11 @@ def put_user_resources() -> Any:
     # Normalize before storing: otherwise " ASVS " and "ASVS" both validate but
     # persist as distinct rows, defeating the dedupe.
     selected = [name.strip() for name in raw_selected]
+    # OpenCRE is always part of a non-empty selection (matches the read filter and
+    # the "OpenCRE is always included" UI copy). An empty selection stays empty —
+    # [] means "show everything", so injecting OpenCRE would wrongly narrow it.
+    if selected and OPENCRE_STANDARD_NAME not in selected:
+        selected.append(OPENCRE_STANDARD_NAME)
     database = db.Node_collection()
     user = _resolve_current_user(database)
     if user is None: