fix(pipeline): recheck bucket quorum after promotion

Hold a community-only row again when bucketed reports appear below the three-bucket floor. Cover separate drains and a mixed same-drain queue.

Author: SysAdminDoc

CHANGELOG.md

@@ -6,7 +6,8 @@ All notable changes to CallShield will be documented in this file.
 
 - New community-only numbers now wait for reports on two UTC days before they
   enter the shipped database. Reports with reporter buckets also need three
-  distinct buckets. Pending reports expire after 30 days. The first pass held
+  distinct buckets, even when bucketed reports arrive after an earlier
+  bucketless promotion. Pending reports expire after 30 days. The first pass held
   449 older, uncorroborated rows, leaving 51,357 numbers in feed v47.
 - Setup now finishes once the required phone, message and call screening access
   is ready. You can skip notifications, caller ID overlay and notification

scripts/merge_community_reports.py

@@ -402,11 +402,12 @@ def main(argv: list[str] | None = None):
             pending.pop(number, None)
             continue
         state = pending.setdefault(number, legacy_community_state(entry, today))
-        if state["published"]:
+        if state["published"] and not any(event["bucket"] for event in state["events"]):
             continue
         if community_has_quorum(state):
             state["published"] = True
         else:
+            state["published"] = False
             del existing[number]
             demoted += 1
 
@@ -507,6 +508,10 @@ def main(argv: list[str] | None = None):
                         if state:
                             state["events"].append({"key": key, "day": reported_at, "bucket": bucket, "count": 1})
                             state["entry"] = entry
+                            if bucket and state["published"] and not community_has_quorum(state):
+                                state["published"] = False
+                                del existing[number]
+                                demoted += 1
                         updated += 1
                     else:
                         if state is None:

scripts/test_report_pipeline.py

@@ -228,6 +228,7 @@ def assert_community_promotion(data_dir: Path) -> None:
     legacy_number = "+12122340671"
     bucket_number = "+12122340672"
     old_number = "+12122340673"
+    mixed_batch_number = "+12122340674"
 
     write_report(data_dir, "first.json", legacy_number, None, TIMES[0])
     run_drain(data_dir)
@@ -243,6 +244,22 @@ def assert_community_promotion(data_dir: Path) -> None:
     run_drain(data_dir)
     assert reports_for(data_dir, legacy_number) == 3, "two UTC days did not promote bucketless reports"
 
+    write_report(data_dir, "legacy_bucket_first.json", legacy_number, BUCKETS[0], f"{TODAY}T00:01:00+00:00")
+    run_drain(data_dir)
+    assert reports_for(data_dir, legacy_number) == 0, "one bucket left a bucketless promotion shipped"
+    write_report(data_dir, "legacy_bucket_second.json", legacy_number, BUCKETS[1], f"{TODAY}T00:02:00+00:00")
+    run_drain(data_dir)
+    assert reports_for(data_dir, legacy_number) == 0, "two buckets restored a demoted row"
+    write_report(data_dir, "legacy_bucket_third.json", legacy_number, BUCKETS[2], f"{TODAY}T00:03:00+00:00")
+    run_drain(data_dir)
+    assert reports_for(data_dir, legacy_number) == 6, "three buckets did not restore a demoted row"
+
+    write_report(data_dir, "mixed_batch_first.json", mixed_batch_number, None, TIMES[0])
+    write_report(data_dir, "mixed_batch_second.json", mixed_batch_number, None, f"{TODAY}T00:00:00+00:00")
+    write_report(data_dir, "mixed_batch_bucket.json", mixed_batch_number, BUCKETS[0], f"{TODAY}T00:02:00+00:00")
+    run_drain(data_dir)
+    assert reports_for(data_dir, mixed_batch_number) == 0, "a mixed same-drain batch shipped with one bucket"
+
     write_report(data_dir, "bucket_first.json", bucket_number, BUCKETS[0], TIMES[0])
     run_drain(data_dir)
     write_report(data_dir, "bucket_repeat.json", bucket_number, BUCKETS[0], TIMES[2])
@@ -264,7 +281,7 @@ def assert_community_promotion(data_dir: Path) -> None:
     write_json(data_dir / "spam_numbers.json", database)
     run_script("merge_community_reports.py", data_dir)
     assert reports_for(data_dir, old_number) == 0, "a legacy single-report row stayed in the shipped database"
-    assert reports_for(data_dir, legacy_number) == 3, "a promoted row was demoted on the next merge"
+    assert reports_for(data_dir, legacy_number) == 6, "a promoted row was demoted on the next merge"
 
     pending_path = data_dir / "community_pending.json"
     pending = json.loads(pending_path.read_text(encoding="utf-8"))