Skip to content

fix(sync): keep the catalog writer alive for the VACINACAO phase - #351

Closed
devgtv wants to merge 1 commit into
AlertaDengue:mainfrom
devgtv:fix/sync-writer-queue-vacinacao-deadlock
Closed

devgtv wants to merge 1 commit into
AlertaDengue:mainfrom
devgtv:fix/sync-writer-queue-vacinacao-deadlock

Conversation

@devgtv

@devgtv devgtv commented Sep 29, 2026

Copy link
Copy Markdown
Contributor

Summary

SyncEngine.run() has a single consumer of write_queue — the serial catalog_writer coroutine — which shuts itself down after seeing writers_total sentinels.

The "no producers" branch queued the writer's shutdown sentinel too early:

else:
    for _ in range(workers):
        await raw_queue.put(None)      # stops the processors
if not ftp_tasks and not gov_workers_tasks:
    for _ in range(workers):
        await write_queue.put(None)    # stops the writer

writers_total = max(1, len(gov_workers_tasks))   # == 1 here

writers_total is 1 when there are no gov items, so the writer consumed one sentinel and exited. The second block is a copy-paste of the first, which is correct for raw_queue.

Impact

VACINACAO items are deliberately deferred to a serial phase that runs after the concurrent drain and pushes onto write_queue (as does the retry pass). With the writer already gone, those entries had no reader:

  1. The parquet was uploaded to S3, but no catalog row was written and no SyncOutcome was emitted — the artifact stayed invisible to PySUS and the run reported nothing happening.
  2. Once the bounded queue (maxsize=workers*2) filled, put blocked forever. The run hung with no timeout.

This affects any run whose pending items are all VACINACAO — e.g. sync_clients.py --saude-only, or run(datasets=["VACINACAO"]). The existing tests always pair VACINACAO with a dadosgov item, which keeps gov_workers_tasks non-empty and hides the bug.

Fix

The writer's single shutdown sentinel is now queued at the end of run(), immediately before awaiting the writer, and writers_total is set to 1 to match the one consumer that exists.

Test

test_run_vacinacao_only_syncs_every_file runs a VACINACAO-only sync of 6 items and asserts all 6 are reported uploaded and all 6 reach the catalog. It fails with assert 0 == 6 against the previous code.

  • 1734 passed, 6 skipped
  • black and isort clean

run() terminated the single catalog writer by queueing `workers`
sentinels on write_queue in the "no producers" branch, then computed
writers_total = max(1, len(gov_workers_tasks)) -- which is 1 when there
are no gov items. The writer breaks on the first sentinel it sees, so it
exited before the trailing phases ran.

VACINACAO items are deliberately deferred to a serial phase that pushes
onto write_queue after the concurrent drain, and the retry pass does the
same. With neither of those consumers alive, their entries had no reader:
the parquet was uploaded to S3 but no catalog row was written and no
SyncOutcome was emitted, so the artifact stayed invisible to PySUS. Once
the bounded queue (maxsize=workers*2) filled, the put blocked forever and
the run hung with no timeout.

The branch was also a copy-paste of the one above it, which correctly
sends sentinels to raw_queue to stop the processors.

Sent the writer's single shutdown sentinel at the end of run(), right
before awaiting the writer, and set writers_total to 1 to match the one
consumer that actually exists.

Test added for a VACINACAO-only run: it asserts every deferred item is
reported as uploaded and reaches the catalog. It fails with 0 uploaded
against the previous code.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants