Streaming deserialization part 3: sink

[bjester]

Summary

Completes the streaming deserialization migration. It adds the sink, switches all callers to the new pipeline, and removes the legacy _deserialize_from_store path. Deserialization now runs as a stream, like serialization:

StoreModelSource → AppModelDeserialize → AppModelValidate → Buffer(500, by model) → DeserializeSink

What's new (morango/sync/stream/deserialize.py)

• DeserializeSink applies each task's resolved outcome per chunk:
  • SAVE: per-record save in its own savepoint, with pre_save/post_save muted.
  • DELETE: one Django Collector delete per chunk.
  • PROPAGATE_DELETE/PROPAGATE_HARD_DELETE: batched bulk_create(ignore_conflicts=True) into the tracking tables.
  • ERROR: an error write on the Store record.
• Store clean writes are one QuerySet.update per chunk. Error writes are per record, because bulk_update measured slower than per-record saves.
• deserialize_from_store(profile, sync_filter=None, skip_errored=False) is the entry point, shaped like serialize_into_store. It logs a per-outcome summary.
• Write helpers delete_app_models and record_propagated_deletions are shared by the sink and Store._deserialize_store_model.

Why the sink checks saves against two sets. The buffer means a record is validated up to one chunk before it is written. FK constraints are DEFERRABLE INITIALLY DEFERRED, so a save that breaks one succeeds inside its savepoint and fails the whole transaction at commit. Two sink writes can remove a target that a later save was validated against:

1. A parent's save fails (for example, a unique violation) after its children passed validation.
2. A DELETE cascades to app rows that a dirty record references.

The sink tracks failed IDs and cascaded row keys, and gives an error write to any SAVE that references them instead of saving it. Each problem record gets a per-record error, and the run always commits.

Callers and removal

• ReceiverDeserializeOperation and MorangoProfileController.deserialize_from_store call the new entry point. The controller's public signature is unchanged.
• _deserialize_from_store, _save_deserialized_record, _validate_store_foreign_keys, _validate_missing_store_foreign_keys and _handle_deleted_store_foreign_keys are removed, along with the imports only they used.
• Store._deserialize_store_model keeps its signature and (app_model | None, dict) return so it can still be called by hand. It is rebuilt on the transforms and write helpers. defer_fks=True is ignored with a DeprecationWarning, and the second element is always {}.

Behavior differences from legacy

• A record that propagates the deletion of its FK target now ends with dirty_bit=False. Legacy left it dirty.
• A child of a dirty parent that is left out of the run (by skip_errored or the sync filter) is now saved, because it validates against the parent's older app row. Legacy never deserialized children of a dirty parent, and failed them with MorangoMissingParent. This difference comes from the Part 2 transforms (Streaming deserialization part 2: transformations #390).
• Models whose clean_fields loads a related object. A child in the same chunk as its dirty parent can't load the parent yet, so it fails with MorangoDirtyParent. A dirty chain of depth d in one chunk takes d runs. Kolibri models don't load related objects in clean_fields, so Kolibri is not affected. The test Facility.clean_dereferences_parent now defaults to False to match, and tests patch it to True to cover such models.
• A dirty grandchild of a cascade-deleted record always gets a "was deleted" error and stays dirty. Legacy's result depended on query order.
• A failed parent save with children in flight fails the children per record. Legacy's cross-model path saved them and failed at commit.
• Store._deserialize_store_model raises the error the transforms resolved, which can be MorangoDirtyParent, MorangoMissingParent or IntegrityError.
• The deserialization warning logger moves from morango.sync.operations to morango.sync.stream.deserialize.

Not in scope

• Recording cascaded deletions in the tracking tables is a separate issue. post_delete stays muted, as in legacy.
• deferred_clean_fields, ForeignKeyReference, TemporaryTable and the backend bulk helpers lose their only callers. They are kept for a later cleanup.

References

• Umbrella: Morango memory performance improvements #279. Related: Reduce memory overhead of _deserialize_from_store #210 (deserialization memory use).
• Part 1: Streaming deserialize: add source stream and task #318. Part 2: Streaming deserialization part 2: transformations #390.
• Stacked on Streaming deserialization part 2: transformations #390. The first 4 commits belong to Part 2. This PR's changes start at "Move deserialize FK helpers to module level for reuse by the sink".

Reviewer guidance

Verification performed

• Full suite on SQLite: 642 passed, 2 skipped. On PostgreSQL: 636 passed, 8 skipped.
• prek run --all-files is clean, and makemigrations --check reports no changes.
• With the sink guards turned off, all 4 DeserializeFromStoreTestCase tests that check the run commits fail at commit with FOREIGN KEY constraint failed. This shows they cover the deferred-constraint case.

Test changes

• Before switching callers, the Part 2 parity tests were frozen as literal values captured from legacy. They now run against the new pipeline. The only changed expectations are:
  • the 3 propagated-deletion rows, where dirty_bit becomes False;
  • test_dirty_parent_not_in_run, where the child is now saved and ends clean, instead of failing with MorangoMissingParent.
• DeserializationTestCases keeps the same expected results; only the call and its keyword arguments change.
• test_controller.py changes only the logger name in test_unique_violation_logs_warning.
• One existing test passes only because of the new Facility default: SelfReferentialFKDeserializationTestCase::test_models_created_successfully. It deserializes a whole dirty chain in one run, which is the multi-run difference described above.

Worth a close look

• DeserializeSink.consume order: errors, then deletes, then propagated deletions, then saves, then Store writes. Deletes run before saves so the save guard can see the rows that a delete in the same chunk cascades to.
• delete_app_models builds its Collector while post_delete is muted. This matters because Django checks for signal receivers when choosing to fast-delete. Fast-deleted querysets are read with one extra SELECT each, only in chunks that have deletes.
• The integration tests use shared on-disk databases, so don't run two test processes against them at once.

AI usage

I used Claude Code (Claude Opus 5.5) throughout this part, in a spec → plan → implementation workflow. The design spec and implementation plan were drafted with Claude, and I reviewed and approved both. An additional AI review of the spec found two problems that we fixed before implementation: a missing cascade case and an impossible query-count criterion. I directed key decisions, including the buffered sink, per-record saves over bulk_update, and keeping Store._deserialize_store_model for manual use. Claude then implemented the plan task by task, writing failing tests first. It also captured the legacy parity values and ran mutation checks to confirm the new tests catch the deferred-constraint failures.