fix(inmemory): stabilize task ordering and lifecycle

[GefMar]

Ensure post_send completes before local execution while preserving per-invocation ownership and await_inplace semantics.

Drain accepted tasks during shutdown, retain background failures, and clean up middleware, result backend, and executor resources reliably.

Add concurrency, cancellation, lifecycle, and compatibility coverage.

Closes: #586

Why This Approach Differs from #639

PR #639 correctly identifies the original problem and the required hook ordering. However, its implementation covers only the sequential happy path and does not establish reliable ownership of locally executed tasks.

Per-Invocation Ownership

PR #639 stores inline tasks in a broker-wide _inplace_tasks set shared by all concurrent .kiq() calls. One invocation can therefore collect and await another invocation’s task, while the second invocation observes an empty set and returns before its own task completes.

This breaks the per-invocation contract of await_inplace=True.

The current implementation assigns each invocation its own execution task and synchronization gate. Every caller waits only for the work it submitted.

Strict post_send Ordering

Creating the receiver with asyncio.create_task() before running post_send does not guarantee strict ordering. When an asynchronous post_send hook suspends, the event loop may start pre_execute and the task body concurrently.

The implementation in this PR keeps the receiver behind a per-invocation gate. Execution cannot begin until the complete post_send middleware chain for that invocation has finished.

Error and Cancellation Ownership

If post_send fails or is cancelled after the message has been accepted, the execution task still needs an explicit owner.

This implementation transfers such work to broker-managed background ownership. wait_all() and shutdown() can then drain it and propagate execution failures instead of leaving detached tasks behind.

It also preserves the existing cancellation behavior for normal inline execution.

Lifecycle Integration

The ownership model is integrated with the complete InMemoryBroker lifecycle:

• completed background failures remain observable until drained;
• cancelling wait_all() does not cancel broker-owned work;
• shutdown() drains accepted tasks before closing middleware, the result backend, and the executor;
• cleanup continues after failures while preserving the first raised exception;
• reentrant drain attempts fail explicitly instead of deadlocking.

Test Coverage

The behavior is protected by deterministic tests covering:

• suspending asynchronous post_send hooks;
• concurrent .kiq() calls;
• post_send failures and cancellation;
• delayed background execution failures;
• overlapping sends and shutdown;
• direct kick() compatibility;
• unchanged behavior for regular external brokers.

The additional implementation is therefore not ceremony around hook ordering. It defines the complete lifecycle of an accepted message: gated execution, per-invocation or broker ownership, failure propagation, draining, and cleanup.

[codecov]

Codecov Report

❌ Patch coverage is 97.07602% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.36%. Comparing base (ae2b788) to head (f1b2ab8).

Files with missing lines	Patch %	Lines
taskiq/brokers/inmemory_broker.py	95.93%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #643      +/-   ##
==========================================
+ Coverage   81.29%   82.36%   +1.07%     
==========================================
  Files          69       69              
  Lines        2577     2711     +134     
==========================================
+ Hits         2095     2233     +138     
+ Misses        482      478       -4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.