Use do_orm_execute for DML driven invalidation

[whabanks]

Summary | Résumé

Experimentally branched off of: #3033

Many dao methods for updating / deleting leverage legacy sqlalchemy DML statements Query.update() and Query.delete(). These DML queries bypass the typical session ORM events like after_commit and after_flush. This prevents us from relying solely on ORM events for cache invalidation. Modernized DML syntax (session.execute(<someQuery>)) can be intercepted via do_orm_execute allowing us to follow the same event driven invalidation patterns for both session ORM changes and changes resulting from DML statements

• Add do_orm_execute event listener
• Add cache_invalidating_dml wrapper to execute DML statements and attach cache metadata to the session.info
• Add _queue_cache_invalidation to store deduplicated cache invalidation actions to be executed post-commit
• Migrate / modernize DML statements from legacy Query.update()/delete() to session.execute(<query))

How does this work?

Bulk DML queries in SQLAlchemy bypass session.dirty and after_flush instance detection. Statements like the one below are both legacy syntax from v1.4 and do not trigger typical session ORM events. (after_flush, after_commit, etc.)

# Legacy syntax & no event triggers here
db.session.query(User).filter_by(id=user.id).update(updates)

The helper method cache_invalidating_dml() adds cache metadata to the SQL statement that can be later extracted. Note the updated query syntax, this facilitates listening to do_orm_execute events and is central to enabling event driven invalidation for DML queries.

statement = cache_invalidating_dml(
    delete(Permission).where(Permission.user_id == user.id),
    user_id=user.id,
)

db.session.execute(statement)

Behind the scenes cache metadata is included:

{
    "cache_invalidation_entity_ids": {
        "user_id": user.id,
    }
}

_intercept_bulk_operations() is attached to the do_orm_execute event. When db.session.execute(statement) is run, it executes and:

1. Verifies the statement is an update or delete
2. Determines the mapped model is of type Permission
3. calls _queue_model_invalidations()
   • Note: We queue the invalidation(s) in the transaction so that we can wait for either the after_commit or after_rollback events to fire, so that we only invalidate after successful mutation, or drop the invalidations when rolling back
4. Looks up the Permission entity in the CACHE_INVALIDATION_REGISTRY
5. Validates that the required id's are present and builds the invalidation actions
6. after_commit event fires, signalling successful data changes and _invalidate_cache_after_commit is called to invalidate the key

At a high level

Benefits of this approach

1. Developers need only to do two things in any given situation to leverage dogpile while everything else is generically taken care of in the background.

ORM session based actions

1. Annotate the dao method, specifying the group it belongs to. User for example:

@cache_on_arguments(namespace="user", group_by="user_id")
def dao_get_user_by_id(user_id) -> dict:
    . . .

2. Check that the CACHE_INVALIDATION_REGISTRY entries for the associated entity exist and clear the keys that it should.

    User: [
        CacheInvalidationRule(namespace="user", entity_id_attribute="id"),
    ],

DML based statements

1. Wrap any bulk update or delete statements that affect cached values with the helper method

statement = update(User).where(User.id == usr.id).values(**updates)
db.session.execute(cache_invalidating_dml(statement, id=usr.id))

2. Check that the CACHE_INVALIDATION_REGISTRY entries for the associated entity exist and clear the keys that it should.

    User: [
        CacheInvalidationRule(namespace="user", entity_id_attribute="id"),
    ],

2. The current approach is compatible with threaded workers

• SQLAlchemy sessions are scoped per request context. Concurrent requests should have distinct sessions
  • Therefore the events we are listening to are also run against distinct, scoped sessions
• While the CACHE_INVALIDATION_REGISTRY is technically global, it's read-only at runtime and will never be mutated.
• Redis clients (used for scan/delete invalidations) use thread-safe connection pools. Each invalidate_group_keys call maintains it's own local scan cursor, therefore parallel scans should not corrupt one another.

Drawbacks

1. Cache invalidation is synchronous with the SQLAlchemy commit path, and we add a small amount of overhead to DB write operations.

database COMMIT
    -> SQLAlchemy after_commit
    -> Redis SCAN
    -> Redis DELETE
    -> commit() returns to caller

This would add some latency to API responses when executing an update or delete. The silver lining here is that all subsequent gets on that entity would hit cache cache instead of the DB and thus should offset added latency. You trade one slightly longer API call for many subsequent faster API calls.

Test instructions | Instructions pour tester la modification

TODO: Fill in test instructions for the reviewer.

Release Instructions | Instructions pour le déploiement

None.

Reviewer checklist | Liste de vérification du réviseur

• This PR does not break existing functionality.
• This PR does not violate GCNotify's privacy policies.
• This PR does not raise new security concerns. Refer to our GC Notify Risk Register document on our Google drive.
• This PR does not significantly alter performance.
• Additional required documentation resulting of these changes is covered (such as the README, setup instructions, a related ADR or the technical documentation).

⚠ If boxes cannot be checked off before merging the PR, they should be moved to the "Release Instructions" section with appropriate steps required to verify before release. For example, changes to celery code may require tests on staging to verify that performance has not been affected.