diff --git a/README.md b/README.md index a5ab86d..e13b7ed 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,23 @@ submission. Option removal and option changes invalidate only answers bound to that option, and repeated removal/revocation is an idempotent replay even after the Poll has moved out of an editable lifecycle state. +## Identified response invariant + +Poll stores at most one active response for each identified respondent in a +Poll. PostgreSQL and SQLite enforce this with the partial unique index +`uq_poll_responses_active_respondent`; anonymous and tombstoned responses do +not participate in that invariant. If two submissions race, the losing insert +is rolled back to a savepoint and follows the ordinary update policy against +the winning row. Unrelated integrity errors are not converted into response +updates. + +The migration deterministically retains the latest active row by +`submitted_at DESC, id DESC` and tombstones older duplicates. Apply it while +Poll response writes are quiesced or while the platform maintenance lock is +held. The released migration-head baseline must only be advanced as part of +the reviewed release that includes this migration; it is not a development +head ledger. + ## Scheduling As A Poll-Backed Workflow Scheduling should use Poll as the reusable response collection primitive, not @@ -172,3 +189,11 @@ Focused manifest verification: cd /mnt/DATA/git/govoplan-poll PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan-core/.venv/bin/python -m unittest discover -s tests ``` + +Run the optional two-session PostgreSQL race check against a disposable or +development database account that may create schemas: + +```bash +GOVOPLAN_POLL_TEST_POSTGRES_URL=postgresql+psycopg://user@localhost/database \ + /mnt/DATA/git/govoplan/.venv/bin/python -m pytest -q tests/test_response_uniqueness.py +``` diff --git a/src/govoplan_poll/backend/participation_service.py b/src/govoplan_poll/backend/participation_service.py index c95e680..6fe5a48 100644 --- a/src/govoplan_poll/backend/participation_service.py +++ b/src/govoplan_poll/backend/participation_service.py @@ -32,8 +32,10 @@ from govoplan_poll.backend.service import ( _assert_governed_invitation_gateway_allowed, _assert_poll_accepts_responses, _existing_response, + _insert_or_reconcile_identified_response, _lock_poll_for_response, _now, + _update_existing_response, assert_no_sensitive_participation_metadata, get_poll_invitation_by_token, normalize_response_answers, @@ -632,25 +634,32 @@ def _submit_locked_governed_response( trusted_metadata["comment"] = comment now = _now() if existing is not None: - if not poll.allow_response_update: - raise PollError("Response updates are not allowed for this poll") - existing.answers = normalized_answers - existing.respondent_label = payload.respondent_label - existing.submitted_at = now - existing.metadata_ = trusted_metadata - response = existing + response = _update_existing_response( + session, + poll=poll, + response=existing, + answers=normalized_answers, + respondent_label=payload.respondent_label, + submitted_at=now, + metadata=trusted_metadata, + ) else: - response = PollResponse( - tenant_id=poll.tenant_id, - poll_id=poll.id, + response, _reconciled = _insert_or_reconcile_identified_response( + session, + poll=poll, respondent_id=respondent_id, respondent_label=payload.respondent_label, answers=normalized_answers, submitted_at=now, - metadata_=trusted_metadata, + metadata=trusted_metadata, + conflict_validator=lambda winner: _enforce_capacity( + session, + poll=poll, + existing=winner, + normalized_answers=normalized_answers, + limit=policy.max_participants_per_option, + ), ) - session.add(response) - session.flush() if idempotency_key is not None: session.add( PollParticipationSubmission( diff --git a/src/govoplan_poll/backend/service.py b/src/govoplan_poll/backend/service.py index 5702e7d..8358fd3 100644 --- a/src/govoplan_poll/backend/service.py +++ b/src/govoplan_poll/backend/service.py @@ -5,8 +5,9 @@ import re import secrets from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any +from typing import Any, Callable +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from govoplan_core.db.base import utcnow @@ -50,6 +51,7 @@ POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ran POLL_INITIAL_STATUSES = {"draft", "open"} CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"} AVAILABILITY_VALUES = {"available", "maybe", "unavailable"} +ACTIVE_RESPONSE_UNIQUE_INDEX = "uq_poll_responses_active_respondent" YES_NO_OPTIONS = ( PollOptionInput(key="yes", label="Yes"), PollOptionInput(key="no", label="No"), @@ -1081,13 +1083,37 @@ def _existing_response(session: Session, *, poll: Poll, respondent_id: str | Non return ( session.query(PollResponse) .filter(PollResponse.poll_id == poll.id, PollResponse.respondent_id == respondent_id, PollResponse.deleted_at.is_(None)) - .order_by(PollResponse.submitted_at.desc()) + .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc()) .first() ) +def _share_lock_poll_for_response( + session: Session, + *, + tenant_id: str, + poll_id: str, +) -> Poll: + """Coordinate response writes with lifecycle changes without serializing peers.""" + + poll = ( + session.query(Poll) + .filter( + Poll.tenant_id == tenant_id, + Poll.id == poll_id, + Poll.deleted_at.is_(None), + ) + .populate_existing() + .with_for_update(read=True) + .first() + ) + if poll is None: + raise PollError("Poll not found") + return poll + + def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -> Poll: - """Serialize the read-or-create path for identified respondents.""" + """Serialize policy-sensitive response and option mutations.""" poll = ( session.query(Poll) @@ -1105,6 +1131,106 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) - return poll +def _is_active_response_uniqueness_conflict( + session: Session, + error: IntegrityError, +) -> bool: + """Recognize only the active identified-response invariant violation.""" + + original = error.orig + constraint_name = getattr( + getattr(original, "diag", None), + "constraint_name", + None, + ) + if constraint_name is not None: + return constraint_name == ACTIVE_RESPONSE_UNIQUE_INDEX + + if session.get_bind().dialect.name != "sqlite": + return False + message = " ".join(str(original).casefold().split()) + return ( + "unique constraint failed:" in message + and "poll_responses.poll_id" in message + and "poll_responses.respondent_id" in message + ) + + +def _update_existing_response( + session: Session, + *, + poll: Poll, + response: PollResponse, + answers: list[dict[str, Any]], + respondent_label: str | None, + submitted_at: datetime, + metadata: dict[str, Any], +) -> PollResponse: + if not poll.allow_response_update: + raise PollError("Response updates are not allowed for this poll") + response.answers = answers + response.respondent_label = respondent_label + response.submitted_at = submitted_at + response.metadata_ = metadata + session.flush() + return response + + +def _insert_or_reconcile_identified_response( + session: Session, + *, + poll: Poll, + respondent_id: str, + respondent_label: str | None, + answers: list[dict[str, Any]], + submitted_at: datetime, + metadata: dict[str, Any], + conflict_validator: Callable[[PollResponse], None] | None = None, +) -> tuple[PollResponse, bool]: + """Insert under a savepoint, or update the row that won the same race.""" + + try: + with session.begin_nested(): + response = PollResponse( + tenant_id=poll.tenant_id, + poll_id=poll.id, + respondent_id=respondent_id, + respondent_label=respondent_label, + answers=answers, + submitted_at=submitted_at, + metadata_=metadata, + ) + session.add(response) + session.flush() + except IntegrityError as error: + if not _is_active_response_uniqueness_conflict(session, error): + raise + winner = _existing_response( + session, + poll=poll, + respondent_id=respondent_id, + ) + if winner is None: + # A correctly identified conflict always has a visible winner after + # the savepoint rollback. Preserve the database failure otherwise. + raise + if conflict_validator is not None: + conflict_validator(winner) + return ( + _update_existing_response( + session, + poll=poll, + response=winner, + answers=answers, + respondent_label=respondent_label, + submitted_at=submitted_at, + metadata=metadata, + ), + True, + ) + return response, False + + def poll_requires_governed_participation(*, poll: Poll) -> bool: """Whether responses must pass through the owning module's gateway.""" @@ -1411,7 +1537,7 @@ def _synchronize_mutable_choice_bounds( def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse: assert_no_sensitive_participation_metadata(payload.metadata) - poll = _lock_poll_for_response( + poll = _share_lock_poll_for_response( session, tenant_id=tenant_id, poll_id=poll_id, @@ -1422,26 +1548,39 @@ def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payl raise PollError("Anonymous responses are not allowed for this poll") answers = normalize_response_answers(poll, payload) existing = _existing_response(session, poll=poll, respondent_id=payload.respondent_id) + submitted_at = _now() if existing is not None: - if not poll.allow_response_update: - raise PollError("Response updates are not allowed for this poll") - existing.answers = answers - existing.respondent_label = payload.respondent_label - existing.submitted_at = _now() - existing.metadata_ = payload.metadata + return _update_existing_response( + session, + poll=poll, + response=existing, + answers=answers, + respondent_label=payload.respondent_label, + submitted_at=submitted_at, + metadata=payload.metadata, + ) + if payload.respondent_id is None: + response = PollResponse( + tenant_id=tenant_id, + poll_id=poll.id, + respondent_id=None, + respondent_label=payload.respondent_label, + answers=answers, + submitted_at=submitted_at, + metadata_=payload.metadata, + ) + session.add(response) session.flush() - return existing - response = PollResponse( - tenant_id=tenant_id, - poll_id=poll.id, + return response + response, _reconciled = _insert_or_reconcile_identified_response( + session, + poll=poll, respondent_id=payload.respondent_id, respondent_label=payload.respondent_label, answers=answers, - submitted_at=_now(), - metadata_=payload.metadata, + submitted_at=submitted_at, + metadata=payload.metadata, ) - session.add(response) - session.flush() return response diff --git a/tests/test_response_uniqueness.py b/tests/test_response_uniqueness.py new file mode 100644 index 0000000..90df174 --- /dev/null +++ b/tests/test_response_uniqueness.py @@ -0,0 +1,353 @@ +from __future__ import annotations + +import os +import sqlite3 +import threading +import unittest +import uuid +from concurrent.futures import ThreadPoolExecutor +from unittest.mock import MagicMock, patch + +from sqlalchemy import create_engine, text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.db.base import Base, utcnow +from govoplan_poll.backend import service as poll_service +from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse +from govoplan_poll.backend.schemas import ( + PollAnswerInput, + PollCreateRequest, + PollOptionInput, + PollSubmitResponseRequest, +) +from govoplan_poll.backend.service import ( + PollError, + _share_lock_poll_for_response, + create_poll, + submit_poll_response, +) + + +class PollResponseUniquenessTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__] + Base.metadata.create_all(self.engine, tables=self.tables) + self.Session = sessionmaker(bind=self.engine) + self.session: Session = self.Session() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables))) + self.engine.dispose() + + def _poll( + self, + *, + allow_anonymous: bool = False, + allow_response_update: bool = True, + ) -> Poll: + return create_poll( + self.session, + tenant_id="tenant-1", + user_id="owner-1", + payload=PollCreateRequest( + title="One response each", + kind="single_choice", + status="open", + allow_anonymous=allow_anonymous, + allow_response_update=allow_response_update, + options=[ + PollOptionInput(key="yes", label="Yes"), + PollOptionInput(key="no", label="No"), + ], + ), + ) + + @staticmethod + def _payload( + respondent_id: str | None, + *, + option_key: str = "yes", + ) -> PollSubmitResponseRequest: + return PollSubmitResponseRequest( + respondent_id=respondent_id, + respondent_label=respondent_id, + answers=[PollAnswerInput(option_key=option_key)], + ) + + def _hide_first_lookup(self): + original = poll_service._existing_response + calls = 0 + + def hide_once(session, *, poll, respondent_id): + nonlocal calls + calls += 1 + if calls == 1: + return None + return original( + session, + poll=poll, + respondent_id=respondent_id, + ) + + return patch.object(poll_service, "_existing_response", side_effect=hide_once) + + def test_sqlite_invariant_reconciles_a_racing_insert_to_normal_update(self) -> None: + poll = self._poll() + winner = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1"), + ) + + with self._hide_first_lookup(): + reconciled = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1", option_key="no"), + ) + + self.assertEqual(reconciled.id, winner.id) + self.assertEqual(reconciled.answers[0]["option_key"], "no") + self.assertEqual( + self.session.query(PollResponse) + .filter(PollResponse.deleted_at.is_(None)) + .count(), + 1, + ) + + def test_racing_insert_obeys_update_disabled_policy(self) -> None: + poll = self._poll() + winner = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1"), + ) + poll.allow_response_update = False + self.session.flush() + + with self._hide_first_lookup(): + with self.assertRaisesRegex( + PollError, + "Response updates are not allowed", + ): + submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1", option_key="no"), + ) + + self.assertEqual(winner.answers[0]["option_key"], "yes") + self.assertEqual(self.session.query(PollResponse).count(), 1) + + def test_anonymous_and_tombstoned_responses_do_not_conflict(self) -> None: + poll = self._poll(allow_anonymous=True) + anonymous_one = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload(None), + ) + anonymous_two = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload(None, option_key="no"), + ) + identified_one = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1"), + ) + identified_one.deleted_at = utcnow() + self.session.flush() + identified_two = submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("person-1", option_key="no"), + ) + + self.assertNotEqual(anonymous_one.id, anonymous_two.id) + self.assertNotEqual(identified_one.id, identified_two.id) + self.assertEqual(self.session.query(PollResponse).count(), 4) + + def test_unrelated_integrity_error_is_not_reconciled(self) -> None: + poll = self._poll() + self.session.execute( + text( + """ + CREATE TRIGGER reject_blocked_poll_response + BEFORE INSERT ON poll_responses + WHEN NEW.respondent_id = 'blocked' + BEGIN + SELECT RAISE(ABORT, 'blocked by unrelated invariant'); + END + """ + ) + ) + + with self.assertRaises(IntegrityError) as caught: + submit_poll_response( + self.session, + tenant_id=poll.tenant_id, + poll_id=poll.id, + payload=self._payload("blocked"), + ) + + self.assertIsInstance(caught.exception.orig, sqlite3.IntegrityError) + self.assertIn("blocked by unrelated invariant", str(caught.exception.orig)) + + def test_response_read_lock_is_shared(self) -> None: + session = MagicMock() + query = MagicMock() + poll = MagicMock() + session.query.return_value = query + query.filter.return_value = query + query.populate_existing.return_value = query + query.with_for_update.return_value = query + query.first.return_value = poll + + result = _share_lock_poll_for_response( + session, + tenant_id="tenant-1", + poll_id="poll-1", + ) + + self.assertIs(result, poll) + query.with_for_update.assert_called_once_with(read=True) + + def test_orm_mirrors_both_partial_index_predicates(self) -> None: + index = next( + index + for index in PollResponse.__table__.indexes + if index.name == "uq_poll_responses_active_respondent" + ) + + self.assertTrue(index.unique) + self.assertEqual( + str(index.dialect_options["sqlite"]["where"]), + "deleted_at IS NULL AND respondent_id IS NOT NULL", + ) + self.assertEqual( + str(index.dialect_options["postgresql"]["where"]), + "deleted_at IS NULL AND respondent_id IS NOT NULL", + ) + + +@unittest.skipUnless( + os.environ.get("GOVOPLAN_POLL_TEST_POSTGRES_URL"), + "set GOVOPLAN_POLL_TEST_POSTGRES_URL for the two-session PostgreSQL check", +) +class PollResponsePostgresConcurrencyTests(unittest.TestCase): + def setUp(self) -> None: + database_url = os.environ["GOVOPLAN_POLL_TEST_POSTGRES_URL"] + self.schema = f"poll_response_race_{uuid.uuid4().hex}" + self.admin_engine = create_engine(database_url) + with self.admin_engine.begin() as connection: + connection.execute(text(f'CREATE SCHEMA "{self.schema}"')) + self.engine = create_engine( + database_url, + connect_args={"options": f"-c search_path={self.schema}"}, + ) + self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__] + Base.metadata.create_all(self.engine, tables=self.tables) + + def tearDown(self) -> None: + try: + Base.metadata.drop_all( + self.engine, + tables=list(reversed(self.tables)), + ) + finally: + self.engine.dispose() + with self.admin_engine.begin() as connection: + connection.execute(text(f'DROP SCHEMA "{self.schema}"')) + self.admin_engine.dispose() + + def test_two_sessions_converge_on_one_active_response(self) -> None: + with Session(self.engine) as session: + poll = create_poll( + session, + tenant_id="tenant-1", + user_id="owner-1", + payload=PollCreateRequest( + title="Concurrent response", + kind="single_choice", + status="open", + options=[ + PollOptionInput(key="yes", label="Yes"), + PollOptionInput(key="no", label="No"), + ], + ), + ) + poll_id = poll.id + session.commit() + + original = poll_service._existing_response + barrier = threading.Barrier(2) + thread_state = threading.local() + + def synchronize_first_lookup(session, *, poll, respondent_id): + response = original( + session, + poll=poll, + respondent_id=respondent_id, + ) + lookup_count = getattr(thread_state, "lookup_count", 0) + 1 + thread_state.lookup_count = lookup_count + if lookup_count == 1: + self.assertIsNone(response) + barrier.wait(timeout=10) + return response + + def submit(option_key: str) -> str: + with Session(self.engine) as session: + response = submit_poll_response( + session, + tenant_id="tenant-1", + poll_id=poll_id, + payload=PollSubmitResponseRequest( + respondent_id="person-1", + respondent_label="Person One", + answers=[PollAnswerInput(option_key=option_key)], + ), + ) + response_id = response.id + session.commit() + return response_id + + with patch.object( + poll_service, + "_existing_response", + side_effect=synchronize_first_lookup, + ): + with ThreadPoolExecutor(max_workers=2) as executor: + response_ids = tuple( + executor.map(submit, ("yes", "no")) + ) + + self.assertEqual(len(set(response_ids)), 1) + with Session(self.engine) as session: + active = ( + session.query(PollResponse) + .filter( + PollResponse.poll_id == poll_id, + PollResponse.respondent_id == "person-1", + PollResponse.deleted_at.is_(None), + ) + .all() + ) + self.assertEqual(len(active), 1) + self.assertIn(active[0].answers[0]["option_key"], {"yes", "no"}) + + +if __name__ == "__main__": + unittest.main()