Stop rebilling completed Instagram recheck batches

pipeline/README.md

@@ -413,3 +413,8 @@ over half of a batch reports deleted is treated as an outage and changes nothing
 Withdrawn sources carry `assessment.deleted`, which publication skips so the
 archived copy is not republished. Progress is kept in `data/post_rechecks.json`.
 A post is checked once, so a deletion after the check is not seen.
+Each batch saves its check results before marking the paid run consumed.
+An interrupted batch reuses its original run and post list; completed batches
+use the next due list. Replaying a run cannot count as a second deletion
+confirmation. Empty or unrelated output stops further batches and retains the
+paid run for investigation rather than silently buying the same lookup again.

pipeline/apify_posts.py

@@ -154,7 +154,10 @@ def details(self, codes: list[str], *, max_charge: float, timeout: int) -> tuple
         # This intent stays unconsumed even for FAILED runs: replacing it within
         # a batch could spend its reserved detail ceiling twice. A fresh plan's
         # cycle-specific filename permits a new attempt for incomplete profiles.
-        codes = saved.get("usernames", sorted(codes))
+        if saved and not saved.get("consumed"):
+            codes = saved.get("usernames", sorted(codes))
+        else:
+            codes = sorted(codes)
         client = ApifyClient(self.session.headers["Authorization"].removeprefix("Bearer "), path)
         run = client._run(hpix_input(posts=[f"https://www.instagram.com/p/{code}/" for code in codes]),
                           {"usernames": codes, "actor_id": HPIX_ACTOR_ID,

pipeline/post_recheck.py

@@ -120,7 +120,12 @@ def main() -> None:
     registry = assessed_events.load_registry()
     state = read_json(STATE_FILE) if STATE_FILE.exists() else {}
     due = due_sources(events, registry, state, now)
-    if not due:
+    details_file = RUN_FILE.with_name(RUN_FILE.stem + "-details.json")
+    saved = read_json(details_file) if details_file.exists() else {}
+    # Finish already-paid work before buying a new batch, even if its events
+    # have expired or its check state was saved just before an interruption.
+    recovery = saved.get("usernames", []) if saved and not saved.get("consumed") else []
+    if not due and not recovery:
         log.info("post recheck: nothing due")
         return
     media_ids = [key.rsplit(":", 1)[-1] for key in due]
@@ -131,46 +136,63 @@ def main() -> None:
     token = os.environ.get("APIFY_TOKEN", "")
     api = ApifyClient(token, RUN_FILE)
     results: dict[str, str] = {}
-    pending = sorted(by_code)
-    for i in range(0, len(pending), BATCH_SIZE):
-        batch = pending[i:i + BATCH_SIZE]
+    withdrawn = 0
+    pending = sorted(set(by_code) - set(recovery))
+    batches = ([recovery] if recovery else []) + [
+        pending[i:i + BATCH_SIZE] for i in range(0, len(pending), BATCH_SIZE)]
+    for batch in batches:
         run, rows = api.details(batch, max_charge=MAX_CHARGE_USD, timeout=600)
-        details_file = RUN_FILE.with_name(RUN_FILE.stem + "-details.json")
-        write_json(details_file, {**read_json(details_file), "consumed": True})
         # Count against the lookup's own ceiling; a charge-limited run may
         # have stopped before reaching every post, so nothing can be concluded.
         if run.get("status") != "SUCCEEDED":
             raise RuntimeError(f"Post recheck run {run.get('id')} ended {run.get('status')}")
         reason = charge_limit_reason(run, MAX_CHARGE_USD, len(rows), details=True)
         if reason:
             raise CollectionDeferred(f"Post recheck: {reason}")
-        batch_result = classify(rows, set(pending))
+        if not rows:
+            raise RuntimeError(f"Post recheck run {run['id']}: empty output; paid batch retained")
+        for row in rows:
+            node = row.get("data") if isinstance(row, dict) else None
+            code = (node.get("code") or node.get("shortcode")) if isinstance(node, dict) else None
+            match = POST_URL.search(str(row.get("input") or "")) if isinstance(row, dict) else None
+            identities = {value for value in (code, match.group(1) if match else None) if value}
+            if not identities or not identities.issubset(batch):
+                raise RuntimeError(f"Post recheck run {run['id']}: unrelated or unidentifiable output; paid batch retained")
+        batch_result = classify(rows, set(batch))
         gone = sum(1 for v in batch_result.values() if v == "missing")
         if len(batch) >= 4 and gone * 2 > len(batch):
             # Posts this old almost never vanish together; the lookup is broken.
             raise RuntimeError(f"Post recheck: {gone} of {len(batch)} posts reported deleted; "
                                "treating the response as an outage and changing nothing")
         results.update(batch_result)
-
-    updates = []
-    for code, outcome in results.items():
-        media_id = by_code[code]
-        key = f"instagram:post:{media_id}"
-        entry = state.get(media_id) or {}
-        if outcome == "present":
-            state[media_id] = {"status": "present", "checked_at": iso(now)}
-            continue
-        misses = entry.get("misses", 0) + 1
-        state[media_id] = {"status": "missing", "misses": misses, "checked_at": iso(now)}
-        log.info("post %s (%s): reported deleted (%d/%d)", media_id, code, misses, CONFIRMATIONS)
-        if misses >= CONFIRMATIONS:
-            updates.append(withdrawal(key, registry[key], now))
-            state[media_id]["status"] = "withdrawn"
-    write_json(STATE_FILE, state)
-    if updates:
-        log.info("post recheck: withdrawing %d deleted post(s): %s", len(updates),
-                 ", ".join(u["source_key"] for u in updates))
-        log.info("Assessed publication: %s", assessed_events.publish(updates))
+        updates = []
+        for code, outcome in batch_result.items():
+            # Recovery may contain sources that are no longer due. Its paid
+            # dataset is still consumed, but expired listings are left alone.
+            if code not in by_code:
+                continue
+            media_id = by_code[code]
+            key = f"instagram:post:{media_id}"
+            entry = state.get(media_id) or {}
+            if entry.get("run_id") == run["id"]:
+                continue
+            state[media_id] = {"status": outcome, "checked_at": iso(now), "run_id": run["id"]}
+            if outcome == "missing":
+                misses = entry.get("misses", 0) + 1
+                state[media_id]["misses"] = misses
+                log.info("post %s (%s): reported deleted (%d/%d)", media_id, code, misses, CONFIRMATIONS)
+                if misses >= CONFIRMATIONS:
+                    updates.append(withdrawal(key, registry[key], now))
+                    state[media_id]["status"] = "withdrawn"
+        if updates:
+            log.info("post recheck: withdrawing %d deleted post(s): %s", len(updates),
+                     ", ".join(u["source_key"] for u in updates))
+            log.info("Assessed publication: %s", assessed_events.publish(updates))
+            withdrawn += len(updates)
+        # Persist each batch before marking its paid run consumed. If either
+        # write fails, replay retains the run ID and cannot double-count misses.
+        write_json(STATE_FILE, state)
+        write_json(details_file, {**read_json(details_file), "consumed": True})
     log.info("post recheck: %d looked up, %d present, %d reported deleted, %d withdrawn",
              len(results), sum(v == "present" for v in results.values()),
-             sum(v == "missing" for v in results.values()), len(updates))
+             sum(v == "missing" for v in results.values()), withdrawn)

pipeline/tests/test_post_recheck.py

@@ -2,15 +2,18 @@
 from __future__ import annotations
 
 import sys
+import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from unittest.mock import MagicMock, patch
 
 PIPELINE_ROOT = Path(__file__).resolve().parents[1]
 if str(PIPELINE_ROOT) not in sys.path:
     sys.path.insert(0, str(PIPELINE_ROOT))
 
 import post_recheck as pr  # noqa: E402
+import apify_posts as apify  # noqa: E402
 
 NOW = datetime(2026, 10, 9, 12, tzinfo=timezone.utc)
 
@@ -62,5 +65,219 @@ def test_publishes_nothing_and_marks_the_source_deleted(self):
         self.assertEqual("complete", update["assessment"]["status"])
 
 
+class RecheckIntegration(unittest.TestCase):
+    """Exercise real batching, paid-run recovery and durable check state."""
+
+    def setUp(self):
+        directory = self.enterContext(tempfile.TemporaryDirectory())
+        self.state_file = Path(directory) / "state.json"
+        self.run_file = Path(directory) / "recheck.json"
+        self.details_file = Path(directory) / "recheck-details.json"
+        self.enterContext(patch.object(pr, "STATE_FILE", self.state_file))
+        self.enterContext(patch.object(pr, "RUN_FILE", self.run_file))
+        self.enterContext(patch.dict(pr.os.environ, {"APIFY_TOKEN": "test"}))
+        self.clock = self.enterContext(patch.object(pr, "datetime", wraps=datetime))
+        self.clock.now.return_value = NOW
+        self.events = []
+        self.registry = {}
+        self.codes = {}
+        self.database = MagicMock()
+        query = self.database.table.return_value.select.return_value.eq.return_value
+        query.gte.return_value.lte.return_value.execute.return_value.data = self.events
+        self.enterContext(patch("db.client", return_value=self.database))
+        self.enterContext(patch("assessed_events.load_registry", return_value=self.registry))
+        self.publish = self.enterContext(patch("assessed_events.publish"))
+        self.enterContext(patch.object(pr, "shortcodes", side_effect=lambda ids: {
+            key: self.codes[key] for key in ids if key in self.codes}))
+        self.purchases = []
+        self.runs = {}
+        self.datasets = {}
+        self.requests = []
+        self.response_rows = None
+        self.fail_purchase = None
+        self.fail_dataset = None
+        self.run_updates = {}
+        self.enterContext(patch.object(apify.ApifyClient, "request", new=self.request))
+
+    def add_posts(self, count, offset=0):
+        for i in range(offset, offset + count):
+            key = str(i)
+            self.codes[key] = f"POST_{i:03d}"
+            self.events.append(event(key, 2))
+            self.registry[f"instagram:post:{key}"] = source(key)
+
+    def request(self, method, path, **kwargs):
+        self.requests.append((method, path))
+        if method == "POST":
+            if len(self.purchases) == self.fail_purchase:
+                raise TimeoutError("ambiguous paid start")
+            codes = [url.rsplit("/", 2)[1] for url in kwargs["json"]["posts"]]
+            run_id = str(len(self.purchases))
+            self.purchases.append(codes)
+            self.datasets[run_id] = (self.response_rows if self.response_rows is not None else
+                                     [{"kind": "post", "data": {"code": code}} for code in codes])
+            run = {"id": run_id, "actId": apify.HPIX_ACTOR_ID, "status": "SUCCEEDED",
+                   "startedAt": NOW.isoformat(), "defaultDatasetId": run_id,
+                   "options": {"maxTotalChargeUsd": kwargs["params"]["maxTotalChargeUsd"]},
+                   "chargedEventCounts": {"individual_post_scraped": len(codes), "apify-actor-start": 1},
+                   "pricingInfo": {"pricingPerEvent": {"actorChargeEvents": {
+                       "individual_post_scraped": {"eventPriceUsd": .00149},
+                       "apify-actor-start": {"eventPriceUsd": .00005}}}}}
+            run.update(self.run_updates)
+            self.runs[run_id] = run
+            return {"data": run}
+        if path.startswith("actor-runs/"):
+            return {"data": self.runs[path.split("/")[1]]}
+        if path.startswith("datasets/"):
+            if path.split("/")[1] == self.fail_dataset:
+                raise RuntimeError("dataset temporarily unavailable")
+            return self.datasets[path.split("/")[1]]
+        raise AssertionError((method, path))
+
+    def test_distinct_batches_and_later_cycle_do_not_recheck_present_posts(self):
+        self.add_posts(60)
+        pr.main()
+        self.assertEqual([50, 10], [len(batch) for batch in self.purchases])
+        self.assertEqual(60, len(set(code for batch in self.purchases for code in batch)))
+        self.assertEqual(60, len(pr.read_json(self.state_file)))
+        pr.main()
+        self.assertEqual(2, len(self.purchases))
+        self.add_posts(1, offset=60)
+        pr.main()
+        self.assertEqual(["POST_060"], self.purchases[-1])
+
+    def test_completed_legacy_list_does_not_replace_new_work(self):
+        pr.write_json(self.details_file, {"consumed": True, "usernames": ["EXPIRED"]})
+        self.add_posts(1)
+        pr.main()
+        self.assertEqual([["POST_000"]], self.purchases)
+
+    def test_state_write_failure_replays_paid_run_without_another_purchase(self):
+        self.add_posts(1)
+        write = pr.write_json
+        def fail_state(path, value):
+            if path == self.state_file:
+                raise OSError("disk unavailable")
+            write(path, value)
+        with patch.object(pr, "write_json", side_effect=fail_state):
+            with self.assertRaises(OSError):
+                pr.main()
+        self.assertFalse(pr.read_json(self.details_file)["consumed"])
+        pr.main()
+        self.assertEqual(1, len(self.purchases))
+        self.assertEqual("present", pr.read_json(self.state_file)["0"]["status"])
+
+    def test_later_batch_failure_keeps_completed_checks(self):
+        self.add_posts(60)
+        self.fail_purchase = 1
+        with self.assertRaises(TimeoutError):
+            pr.main()
+        self.assertEqual(50, len(pr.read_json(self.state_file)))
+        with self.assertRaisesRegex(RuntimeError, "outcome unknown"):
+            pr.main()
+        self.assertEqual(1, len(self.purchases))
+
+    def test_pending_paid_run_is_consumed_even_when_its_events_expired(self):
+        self.add_posts(1)
+        with patch.object(pr, "write_json", side_effect=OSError("disk unavailable")):
+            with self.assertRaises(OSError):
+                pr.main()
+        self.events.clear()
+        pr.main()
+        self.assertEqual(1, len(self.purchases))
+        self.assertTrue(pr.read_json(self.details_file)["consumed"])
+
+    def test_second_batch_resumes_original_list_after_due_work_changes(self):
+        self.add_posts(60)
+        self.fail_dataset = "1"
+        with self.assertRaisesRegex(RuntimeError, "dataset temporarily unavailable"):
+            pr.main()
+        self.assertEqual(50, len(pr.read_json(self.state_file)))
+        self.add_posts(1, offset=60)
+        self.fail_dataset = None
+        pr.main()
+        self.assertEqual([50, 10, 1], [len(batch) for batch in self.purchases])
+        self.assertEqual(["POST_060"], self.purchases[-1])
+        self.assertEqual(61, len(pr.read_json(self.state_file)))
+
+    def test_failed_or_charge_limited_run_is_not_consumed_or_rebought(self):
+        self.add_posts(60)
+        for changes in ({"status": "FAILED"}, {"options": {"maxTotalChargeUsd": .05}}):
+            with self.subTest(changes=changes):
+                self.run_updates = changes
+                if self.details_file.exists():
+                    self.details_file.unlink()
+                before = len(self.purchases)
+                with self.assertRaises(RuntimeError):
+                    pr.main()
+                self.assertFalse(pr.read_json(self.details_file)["consumed"])
+                with self.assertRaises(RuntimeError):
+                    pr.main()
+                self.assertEqual(before + 1, len(self.purchases))
+                self.assertFalse(self.state_file.exists())
+
+    def test_withdrawal_failure_replays_paid_results_before_marking_withdrawn(self):
+        self.add_posts(1)
+        pr.write_json(self.state_file, {"0": {"status": "missing", "misses": 1,
+                      "checked_at": (NOW - timedelta(hours=8)).isoformat(), "run_id": "old"}})
+        self.response_rows = [{"kind": "post", "input": "https://www.instagram.com/p/POST_000/",
+                               "error": "Post deleted"}]
+        self.publish.side_effect = RuntimeError("database unavailable")
+        with self.assertRaisesRegex(RuntimeError, "database unavailable"):
+            pr.main()
+        self.assertEqual("missing", pr.read_json(self.state_file)["0"]["status"])
+        self.assertFalse(pr.read_json(self.details_file)["consumed"])
+        self.publish.side_effect = None
+        pr.main()
+        self.assertEqual(1, len(self.purchases))
+        self.assertEqual("withdrawn", pr.read_json(self.state_file)["0"]["status"])
+
+    def test_matching_rate_limit_stays_unknown_without_withdrawing(self):
+        self.add_posts(1)
+        self.response_rows = [{"input": "https://www.instagram.com/p/POST_000/", "error": "Rate limited"}]
+        pr.main()
+        self.assertEqual({}, pr.read_json(self.state_file))
+        self.assertTrue(pr.read_json(self.details_file)["consumed"])
+        self.publish.assert_not_called()
+
+    def test_unrelated_or_empty_output_stops_without_buying_more_batches(self):
+        self.add_posts(60)
+        for rows in ([{"kind": "post", "data": {"code": "UNRELATED"}}], []):
+            with self.subTest(rows=rows):
+                self.response_rows = rows
+                if self.details_file.exists():
+                    self.details_file.unlink()
+                before = len(self.purchases)
+                with self.assertRaisesRegex(RuntimeError, "output"):
+                    pr.main()
+                self.assertEqual(before + 1, len(self.purchases))
+                with self.assertRaisesRegex(RuntimeError, "output"):
+                    pr.main()
+                self.assertEqual(before + 1, len(self.purchases))
+                self.assertFalse(self.state_file.exists())
+
+    def test_missing_confirmation_is_not_counted_twice_after_consumption_write_failure(self):
+        self.add_posts(1)
+        self.response_rows = [{"kind": "post", "input": "https://www.instagram.com/p/POST_000/",
+                               "error": "Post deleted"}]
+        write = pr.write_json
+        def fail_consumption(path, value):
+            if path == self.details_file and value.get("consumed"):
+                raise OSError("disk unavailable")
+            write(path, value)
+        with patch.object(pr, "write_json", side_effect=fail_consumption):
+            with self.assertRaises(OSError):
+                pr.main()
+        self.clock.now.return_value = NOW + timedelta(hours=8)
+        pr.main()
+        self.assertEqual(1, len(self.purchases))
+        self.assertEqual(1, pr.read_json(self.state_file)["0"]["misses"])
+        self.publish.assert_not_called()
+        pr.main()
+        self.assertEqual(2, len(self.purchases))
+        self.assertEqual("withdrawn", pr.read_json(self.state_file)["0"]["status"])
+        self.publish.assert_called_once()
+
+
 if __name__ == "__main__":
     unittest.main()