diff --git a/pysus/management/sync.py b/pysus/management/sync.py index fa84db25..eb0f3bd6 100644 --- a/pysus/management/sync.py +++ b/pysus/management/sync.py @@ -1250,11 +1250,15 @@ async def catalog_writer() -> None: else: for _ in range(workers): await raw_queue.put(None) - if not ftp_tasks and not gov_workers_tasks: - for _ in range(workers): - await write_queue.put(None) - writers_total = max(1, len(gov_workers_tasks)) + # The writer is the single consumer of ``write_queue`` and the only + # one that shuts itself down: it exits once it has seen + # ``writers_total`` sentinels. Nothing may queue a sentinel before + # every producer is done — the VACINACAO serial phase and the retry + # pass below both push onto this queue after the concurrent drain. + # The single sentinel is therefore queued at the very end, right + # before awaiting the writer. + writers_total = 1 writer_task = asyncio.create_task(catalog_writer()) await asyncio.gather(*ftp_tasks, *gov_workers_tasks) for _ in processor_tasks: diff --git a/pysus/tests/management/test_sync_more.py b/pysus/tests/management/test_sync_more.py index 70df5ad9..6a483dc5 100644 --- a/pysus/tests/management/test_sync_more.py +++ b/pysus/tests/management/test_sync_more.py @@ -1126,6 +1126,77 @@ async def _convert(file, raw, callback=None): assert outcomes assert journal.exists() + @pytest.mark.asyncio + async def test_run_vacinacao_only_syncs_every_file(self, engine, tmp_path): + """A VACINACAO-only run must still reach the catalog writer. + + VACINACAO items are deferred to a trailing serial phase that + pushes onto ``write_queue``. When the run has no FTP and no + gov items, the "no producers" branch queued the writer's shutdown + sentinel *before* that phase ran, so the writer exited early and + the deferred entries had no consumer. + """ + vacs = [ + _record( + "dadosgov", + f"VAC{i}BR25.csv.zip", + dataset="VACINACAO", + size=0, + file=_remote_file(basename=f"VAC{i}BR25.csv.zip"), + ) + for i in range(6) + ] + records = { + "ducklake": [], + "ftp": [], + "dadosgov": vacs, + "saude": [], + } + engine.dadosgov_token = "tok" + engine._ducklake = self._ducklake_run() + + raw = tmp_path / "raw.bin" + raw.write_bytes(b"x") + + async def _download(file, ftp_client=None): + return raw + + async def _convert(file, raw, callback=None): + return self._payload() + + engine._download_raw_with_retry = AsyncMock(side_effect=_download) + engine._convert_and_upload = AsyncMock(side_effect=_convert) + + writer = self._writer_run() + outcomes = [] + + with patch.object(engine, "_require_pysus", return_value=MagicMock()): + with patch( + "pysus.management.sync.Inventory", + return_value=self._inventory(records), + ): + with patch.object(engine, "_checkpoint", new=AsyncMock()): + with patch.object( + SyncEngine, + "writer", + new_callable=PropertyMock, + ) as mock_writer_prop: + mock_writer_prop.return_value = writer + report = await asyncio.wait_for( + engine.run( + datasets=["VACINACAO"], + on_outcome=outcomes.append, + ), + timeout=30, + ) + + assert report.summary()["uploaded"] == len(vacs) + assert report.summary()["failed"] == 0 + uploaded = [o for o in outcomes if o.status == "uploaded"] + assert len(uploaded) == len(vacs) + # The catalog row is only written when the writer is alive. + assert writer.upsert_file.call_count == len(vacs) + @pytest.mark.asyncio async def test_run_retries_failed_ftp_and_catalog_errors( self, engine, tmp_path