fix: converge changing topic shards

Author: aiwarelife

scripts/sync_topic_plugins.py

@@ -272,23 +272,26 @@ def search_page(self, query_text: str, page: int) -> dict[str, Any]:
         return payload
 
     def collect_query(self, query_text: str, first: dict[str, Any]) -> list[dict[str, Any]]:
-        total = int(first.get("total_count", 0))
-        if total > 1000:
+        initial_total = int(first.get("total_count", 0))
+        if initial_total > 1000:
             raise SyncError(f"search shard still exceeds 1,000 repositories: {query_text}")
-        repositories = list(first.get("items", []))
-        page = 2
-        while len(repositories) < total:
-            payload = self.search_page(query_text, page)
-            items = payload.get("items", [])
-            if not items:
-                break
-            repositories.extend(items)
-            page += 1
-        if len(repositories) != total:
-            raise SyncError(
-                f"search shard reported {total} repositories but returned {len(repositories)}"
-            )
-        return repositories
+        seen: dict[int, dict[str, Any]] = {}
+        next_first = first
+        for _ in range(3):
+            latest_total = int(next_first.get("total_count", 0))
+            page = 1
+            while page <= max(1, (latest_total + 99) // 100):
+                payload = next_first if page == 1 else self.search_page(query_text, page)
+                latest_total = int(payload.get("total_count", latest_total))
+                for item in payload.get("items", []):
+                    seen[int(item["id"])] = item
+                page += 1
+            if len(seen) >= latest_total:
+                return list(seen.values())
+            next_first = self.search_page(query_text, 1)
+        raise SyncError(
+            f"search shard did not converge: latest={latest_total}, unique={len(seen)}"
+        )
 
     def collect_date_range(
         self,

tests/test_topic_sync.py

@@ -141,6 +141,26 @@ def test_deferred_candidate_is_not_publishable(self):
         self.assertFalse(candidate["enriched"])
         self.assertEqual(candidate["reasons"], ["enrichment_deferred"])
 
+    def test_search_shard_converges_when_topic_grows_during_pagination(self):
+        class GrowingClient(topic_sync.GitHubClient):
+            def __init__(self):
+                pass
+
+            def search_page(self, query_text, page):
+                if page == 2:
+                    return {
+                        "total_count": 104,
+                        "items": [{"id": value} for value in range(101, 105)],
+                    }
+                raise AssertionError("unexpected page")
+
+        first = {
+            "total_count": 102,
+            "items": [{"id": value} for value in range(1, 101)],
+        }
+        items = GrowingClient().collect_query("topic:dsh-plugin", first)
+        self.assertEqual(len(items), 104)
+
     def test_changelog_and_report_are_idempotent(self):
         promoted = [
             {