diff --git a/apps/api/app/main.py b/apps/api/app/main.py index e5844c7..798afbc 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -1649,7 +1649,13 @@ def _run_source_discovery(db, job, handler): selected=[linked["kind"]] if linked else [] placeholders=','.join('?'*len(selected)) or "NULL" sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND circuit_open=0 AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall() if selected else [] - total=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100))) + if selected and not sources: + blocked=db.execute("SELECT source_code,kind,circuit_open,enabled FROM sources WHERE organization_id=? AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall() + code="SOURCE_CIRCUIT_OPEN" if any(row["circuit_open"] for row in blocked) else "SOURCE_DISABLED" + handler.add_job_event(db,job["id"],org,"source.blocked","No selected source is eligible to run",10,code) + if run: db.execute("UPDATE discovery_runs SET lifecycle='failed',updated_at=CURRENT_TIMESTAMP WHERE id=?",(run["id"],)) + raise RuntimeError(code) + total=0; blocked_count=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100))) for source in sources: handler.add_job_event(db,job["id"],org,"source.started",f"Starting {source['display_name'] or source['kind']}",5) try: @@ -1664,9 +1670,10 @@ def _run_source_discovery(db, job, handler): handler.add_job_event(db,job["id"],org,"source.quota_exceeded",f"Quota reached for {source['kind']}",10,"SOURCE_QUOTA_EXCEEDED"); continue page=adapter_for(source["source_code"] or source["kind"]).discover(source_config_with_credentials(db, source, config)) except Exception as exc: - failures=int(source["consecutive_failures"])+1 + blocked_count+=1; failures=int(source["consecutive_failures"])+1 db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=?,circuit_open=?,last_failure_at=CURRENT_TIMESTAMP,last_error=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(failures,int(circuit_is_open(failures)),str(exc)[:300],source["id"])) - handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",10,"SOURCE_NOT_CONFIGURED"); continue + detail=str(exc).lower(); code="SOURCE_NETWORK_ERROR" if any(token in detail for token in ("urlopen", "gaierror", "timed out", "temporary failure", "network is unreachable")) else "SOURCE_EXECUTION_FAILED" + handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['display_name'] or source['kind']} could not be reached",10,code); continue limit=min(max_records-total, per_run_limit, max(0,daily_limit-used_today)) for record in page.records[:limit]: raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); normalized=normalize_business(normalize_record(record)); norm=json.dumps(normalized,sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest() @@ -1697,7 +1704,10 @@ def _run_source_discovery(db, job, handler): if not db.execute("SELECT 1 FROM contacts WHERE organization_id=? AND business_id=? AND email=? AND phone=?",(org,bid,normalized["email"],normalized["phone"])).fetchone(): db.execute("INSERT INTO contacts(business_id,organization_id,email,phone) VALUES(?,?,?,?)",(bid,org,normalized["email"],normalized["phone"])) if record_id: db.execute("UPDATE source_records SET processing_status='processed',normalized_json=?,normalized_key=? WHERE id=? AND organization_id=?",(norm,nkey,record_id,org)); db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id,status) VALUES(?,?,?)",(org,record_id,"completed")); db.execute("UPDATE enrichment_queue SET status='completed',updated_at=CURRENT_TIMESTAMP WHERE source_record_id=? AND organization_id=?",(record_id,org)) handler.add_job_event(db,job["id"],org,"enrichment.queued",f"Enrichment complete for business {bid}",75); handler.add_job_event(db,job["id"],org,"review.queued",f"Business {bid} queued for review",90) - if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle='succeeded',updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,run["id"])) + if blocked_count and total==0: + if run: db.execute("UPDATE discovery_runs SET lifecycle='failed',updated_at=CURRENT_TIMESTAMP WHERE id=?",(run["id"],)) + raise RuntimeError("SOURCE_NETWORK_ERROR") + if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,'partial' if blocked_count else 'succeeded',run["id"])) handler.add_job_event(db,job["id"],org,"discovery.completed",f"Persisted {total} source records",100) diff --git a/apps/api/tests/test_sources_phase5.py b/apps/api/tests/test_sources_phase5.py index 93553c3..b349662 100644 --- a/apps/api/tests/test_sources_phase5.py +++ b/apps/api/tests/test_sources_phase5.py @@ -113,6 +113,16 @@ class SourceApiTests(unittest.TestCase): self.assertEqual(status, 200) self.assertFalse(health['configured']) + def test_disabled_selected_source_fails_instead_of_succeeding_with_zero_records(self): + status, source=self.req('POST','/api/v1/sources',{'name':'Disabled manual source','kind':'manual','config':{'rows':[]}}); self.assertEqual(status,201) + status, query=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'disabled-source-run','query':{},'selected_adapters':['manual']}); self.assertEqual(status,201) + status, job=self.req('POST',f"/api/v1/discovery-queries/{query['id']}/run",{}); self.assertEqual(status,202) + for _ in range(100): + _, current=self.req('GET',f"/api/v1/jobs/{job['id']}") + if current['status'] in ('succeeded','failed'): break + threading.Event().wait(.01) + self.assertEqual(current['status'],'failed'); self.assertEqual(current['error_code'],'SOURCE_DISABLED') + def test_source_discovery_persists_pipeline_and_is_idempotent(self): status, source = self.req('POST', '/api/v1/sources', { 'name': 'Manual leads', 'kind': 'manual', 'enabled': True, diff --git a/docker-compose.yml b/docker-compose.yml index 0b8c096..6d062b0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -56,6 +56,7 @@ services: restart: unless-stopped networks: - prospect_internal + - source_egress web: build: @@ -115,5 +116,6 @@ volumes: networks: prospect_internal: internal: true + source_egress: searxng_egress: web_ingress: