fix(pesacheck_meedan_bridge): Stop retrying fact-checks Check already has

.github/workflows/pesacheck-meedan-bridge-ci.yaml

@@ -0,0 +1,60 @@
+# yamllint disable rule:line-length
+name: PesaCheck Meedan Bridge | Continuous Integration
+"on":
+  pull_request:
+    types: [opened, synchronize, reopened]
+    paths:
+      - "pesacheck_meedan_bridge/**"
+      - ".github/workflows/pesacheck-meedan-bridge-ci.yaml"
+# The job only checks out the repo.
+permissions:
+  contents: read
+
+# This allows a subsequently queued workflow run to interrupt previous runs
+concurrency:
+  # yamllint disable-line rule:line-length
+  group: "${{ github.workflow }} @ ${{ github.event.pull_request.head.label || github.head_ref || github.ref }}"
+  cancel-in-progress: true
+
+jobs:
+  build:
+    name: Perform CI Checks
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Setup Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Initialize Pants
+        uses: pantsbuild/actions/init-pants@main
+        with:
+          # cache0 makes it easy to bust the cache if needed
+          gha-cache-key: cache0-py${{ matrix.python_version }}
+          named-caches-hash: ${{ hashFiles('lockfiles/*.json', '**/something-else.lock') }}
+
+      - name: Bootstrap Pants
+        run: pants --version
+
+      - name: Check BUILD files
+        run: pants tailor --check update-build-files --check
+
+      - name: Lint
+        run: |
+          pants lint --lint-skip-formatters pesacheck_meedan_bridge/::
+
+      - name: Test
+        run: |
+          pants test pesacheck_meedan_bridge/::
+
+      - name: Build
+        run: |
+          pants package pesacheck_meedan_bridge/py:pesacheck-deps pesacheck_meedan_bridge/py:pesacheck-srcs

pesacheck_meedan_bridge/README.md

@@ -55,6 +55,7 @@ Articles are tracked by `status`:
 | `Pending` | Stored, not yet accepted by Check. Retried on every run. |
 | `Posting` | Sent to Check, but the outcome could not be recorded (e.g. a timeout). **Not** retried automatically: check the article in Check, then set the row to `Completed` (with the Check ids) or back to `Pending`. Each run reports these to Sentry. |
 | `Completed` | Accepted by Check, with the Check ids stored. |
+| `Duplicate` | Check already has this fact-check (it rejects a repeat of the same content). Terminal: never retried. The Check ids are not recorded, so find the article in Check by title if you need them. |
 
 ### Known limitation: backdated articles
 

pesacheck_meedan_bridge/py/BUILD

@@ -8,6 +8,11 @@ python_sources(
     ],
 )
 
+python_tests(
+    name="tests",
+    dependencies=[":lib"],
+)
+
 pex_binary(
     name="pesacheck-deps",
     environment=parametrize("__local__", "linux"),

pesacheck_meedan_bridge/py/VERSION

@@ -1 +1 @@
-0.1.21
+0.1.22

pesacheck_meedan_bridge/py/check_api.py

@@ -3,6 +3,35 @@
 import requests
 import settings
 
+# Check refuses a fact-check it already has, via a unique index on the
+# fact-check signature. The article is in Check, so there is nothing to retry.
+# This is Check's own Postgres index name, surfaced in the GraphQL error text:
+# it is not part of any API contract, so a rename upstream would stop it
+# matching and duplicates would go back to being retried every run.
+DUPLICATE_CONSTRAINT = "index_fact_checks_on_signature"
+
+
+def error_message(error):
+    """The text of one GraphQL error, whatever shape the server sent."""
+    if isinstance(error, dict):
+        return str(error.get("message") or "")
+    return str(error or "")
+
+
+def is_duplicate(errors):
+    """True when the ONLY thing wrong is that Check already has this one.
+
+    A response can carry several errors; treating it as a duplicate because one
+    of them matches would mark the row terminally and silently drop the rest.
+    """
+    if not errors:
+        return False
+    return all(DUPLICATE_CONSTRAINT in error_message(error) for error in errors)
+
+
+class DuplicateFactCheckError(Exception):
+    pass
+
 
 def create_mutation_query(
     media_type="Blank",
@@ -70,7 +99,10 @@ def post_to_check(data):
     url = settings.PESACHECK_CHECK_URL
     response = requests.post(url, headers=headers, json=body, timeout=60)
     res = response.json()
-    if response.status_code != 200 or res.get("errors"):
+    errors = res.get("errors") or []
+    if is_duplicate(errors):
+        raise DuplicateFactCheckError(response.text)
+    if response.status_code != 200 or errors:
         raise Exception(response.text)
     project_media = ((res.get("data") or {}).get("createProjectMedia") or {}).get(
         "project_media"

pesacheck_meedan_bridge/py/database.py

@@ -81,13 +81,17 @@ def insert_pesacheck_feed(self, feed):
         finally:
             conn.close()
 
-    def update_pesacheck_feed(self, guid, new_feed):
+    def update_pesacheck_feed(self, guid, new_feed, expected_status=None):
         conn = self.create_connection()
         sql = """UPDATE pesacheck_feeds
                 SET title = ?, pubDate = ?, author = ?, link = ?, thumbnail = ?,
                 description = ?, status = ?, categories = ?,
                 check_project_media_id = ?, check_full_url = ?,
                 claim_description_id = ? WHERE guid = ?"""
+        params_tail = []
+        if expected_status is not None:
+            sql += " AND status = ?"
+            params_tail.append(expected_status)
         try:
             cur = conn.cursor()
             cur.execute(
@@ -105,24 +109,59 @@ def update_pesacheck_feed(self, guid, new_feed):
                     new_feed.check_full_url,
                     new_feed.claim_description_id,
                     guid,
+                    *params_tail,
                 ),
             )
             conn.commit()
             if cur.rowcount != 1:
-                raise Error(f"No pesacheck_feeds row with guid {guid}")
+                raise Error(
+                    f"No pesacheck_feeds row with guid {guid}"
+                    + (f" in status {expected_status}" if expected_status else "")
+                )
         finally:
             conn.close()
 
-    def update_pesacheck_feed_status(self, guid, status):
+    def claim_pending_feed(self, guid):
+        """Move a row from Pending to Posting, returning whether we won it.
+
+        Two overlapping runs can both read the same Pending row. SQLite
+        serializes the conditional update, so only one of them sees rowcount 1
+        and calls Check; the loser leaves the row alone.
+        """
         conn = self.create_connection()
         try:
             cur = conn.cursor()
             cur.execute(
-                "UPDATE pesacheck_feeds SET status = ? WHERE guid = ?", (status, guid)
+                "UPDATE pesacheck_feeds SET status = 'Posting' "
+                "WHERE guid = ? AND status = 'Pending'",
+                (guid,),
             )
             conn.commit()
+            return cur.rowcount == 1
+        finally:
+            conn.close()
+
+    def update_pesacheck_feed_status(self, guid, status, expected_status=None):
+        """Set a row's status, optionally only from an expected one.
+
+        Terminal transitions pass expected_status="Posting" so a row another
+        run has already finished can't be overwritten.
+        """
+        conn = self.create_connection()
+        sql = "UPDATE pesacheck_feeds SET status = ? WHERE guid = ?"
+        params = [status, guid]
+        if expected_status is not None:
+            sql += " AND status = ?"
+            params.append(expected_status)
+        try:
+            cur = conn.cursor()
+            cur.execute(sql, params)
+            conn.commit()
             if cur.rowcount != 1:
-                raise Error(f"No pesacheck_feeds row with guid {guid}")
+                raise Error(
+                    f"No pesacheck_feeds row with guid {guid}"
+                    + (f" in status {expected_status}" if expected_status else "")
+                )
         finally:
             conn.close()
 

pesacheck_meedan_bridge/py/main.py

@@ -6,7 +6,7 @@
 import requests
 import sentry_sdk
 import settings
-from check_api import post_to_check
+from check_api import DuplicateFactCheckError, post_to_check
 from database import PesacheckDatabase, PesacheckFeed
 
 
@@ -142,22 +142,42 @@ def build_check_input(feed):
     }
 
 
+def mark_terminal(feed, db, status):
+    """Record a terminal status without losing the outcome it describes.
+
+    The caller is about to raise the exception this status explains; a failure
+    here must not replace it, or a known duplicate would look like a row whose
+    outcome could not be recorded.
+    """
+    feed.status = status
+    try:
+        db.update_pesacheck_feed_status(feed.guid, status, expected_status="Posting")
+    except Exception as exception:
+        sentry_sdk.capture_exception(exception)
+
+
 def post_to_check_and_update(feed, db):
     # Build the request first: a failure here means nothing was sent, so the row
     # must stay "Pending" rather than be marked as maybe-posted.
     input_data = build_check_input(feed)
-    # Mark the row before posting so that, if the result cannot be recorded, the
-    # row is not re-posted on the next run and is flagged for reconciliation.
-    db.update_pesacheck_feed_status(feed.guid, "Posting")
+    # Claim the row before posting: it marks the row as maybe-posted so a failed
+    # run doesn't re-post it, and it stops an overlapping run posting it twice.
+    if not db.claim_pending_feed(feed.guid):
+        return None
     try:
         res = post_to_check(input_data)
+    except DuplicateFactCheckError:
+        # Check already has this fact-check, so retrying it every run would
+        # fail forever. Mark it terminally instead.
+        mark_terminal(feed, db, "Duplicate")
+        raise
     except requests.RequestException:
         # e.g. a timeout: Check may still have created the item, so leave the
         # row as "Posting" rather than risk a duplicate.
         raise
     except Exception:
         # Check rejected the mutation, so nothing was created.
-        db.update_pesacheck_feed_status(feed.guid, "Pending")
+        mark_terminal(feed, db, "Pending")
         raise
     # post_to_check() has validated the response, so from here on the item
     # exists in Check: a failure to record it leaves the row as "Posting".
@@ -166,25 +186,41 @@ def post_to_check_and_update(feed, db):
     feed.check_full_url = project_media.get("full_url")
     feed.claim_description_id = project_media["claim_description"]["fact_check"]["id"]
     feed.status = "Completed"
-    db.update_pesacheck_feed(feed.guid, feed)
+    db.update_pesacheck_feed(feed.guid, feed, expected_status="Posting")
     return feed
 
 
+def count_phrase(count):
+    """Shared wording so the run's messages can't drift apart."""
+    return f"{count} PesaCheck article(s)"
+
+
+def post_and_record(feed, db, success_posts, duplicates):
+    """Post one stored article, recording the outcome for the run summary."""
+    try:
+        posted = post_to_check_and_update(feed, db=db)
+        if posted is not None:
+            success_posts.append(posted.link)
+    except DuplicateFactCheckError:
+        duplicates.append(feed.link)
+    except Exception as exception:
+        sentry_sdk.capture_exception(exception)
+
+
 def main(db):
     success_posts = []
+    duplicates = []
     try:
         unreconciled = db.get_pesacheck_feeds_by_status("Posting")
         if unreconciled:
             sentry_sdk.capture_message(
-                f"{len(unreconciled)} PesaCheck article(s) may have been posted to "
-                f"Check without being recorded: {[feed.link for feed in unreconciled]}",
+                f"{count_phrase(len(unreconciled))} may have been posted to "
+                "Check without being recorded: "
+                f"{[feed.link for feed in unreconciled]}",
                 level="warning",
             )
         for pending in db.get_pesacheck_feeds_by_status("Pending"):
-            try:
-                success_posts.append(post_to_check_and_update(pending, db=db))
-            except Exception as exception:
-                sentry_sdk.capture_exception(exception)
+            post_and_record(pending, db, success_posts, duplicates)
         from_pesacheck = fetch_from_pesacheck(since=get_checkpoint(db))
         # Oldest first, so the checkpoint never moves past an unstored article.
         for post in reversed(from_pesacheck):
@@ -219,21 +255,21 @@ def main(db):
                 # next run starts again from the current checkpoint.
                 sentry_sdk.capture_exception(exception)
                 break
-            try:
-                # The article is stored, so a failure here is retried next run.
-                success_posts.append(post_to_check_and_update(feed, db=db))
-            except Exception as exception:
-                sentry_sdk.capture_exception(exception)
+            # The article is stored, so a failure here is retried next run.
+            post_and_record(feed, db, success_posts, duplicates)
     except Exception as e:
         # Re-raised so that the process exits non-zero and cron/monitoring can
         # see that the whole run failed.
         sentry_sdk.capture_exception(e)
         raise
     finally:
-        sentry_sdk.capture_message(
-            f"Posted {len(success_posts)} PesaCheck article(s) to Check: "
-            f"{[post.link for post in success_posts]}"
-        )
+        message = f"Posted {count_phrase(len(success_posts))} to Check: {success_posts}"
+        if duplicates:
+            message += (
+                f". Skipped {count_phrase(len(duplicates))} "
+                f"Check already has: {duplicates}"
+            )
+        sentry_sdk.capture_message(message)
 
 
 if __name__ == "__main__":