fix: do not return stopped service instances from the instance cache - #22995
fix: do not return stopped service instances from the instance cache#22995devin-ai-integration[bot] wants to merge 3 commits into
Conversation
A queue service that stops during startup removes itself from the instance cache before the caller of instance() stores it there, so the stopped instance stays cached and all later callers get a service that cannot accept items. Co-Authored-By: bot_apk <apk@cognition.ai>
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
devin-ai-integration
Bot
added
the
development
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| if instance._stopped: | ||
| cls._instances.pop(key, None) |
There was a problem hiding this comment.
🟡 Stopped replacement still reaches callers
If _new_instance stops before returning, instance uncaches it but still returns it. The first submission still fails and can lose an event or log.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This is intentional and out of scope here. If a service cannot start, the caller must see that failure; a retry loop inside instance() would hide a persistent startup problem. The bug this PR fixes is that the failure was permanent for every later caller, because the stopped service stayed in the cache. After this change the first caller can still fail, but the next instance() call gets a working service.
…lacement Co-Authored-By: bot_apk <apk@cognition.ai>
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…instance Co-Authored-By: bot_apk <apk@cognition.ai>
| # A stopped service never accepts items again, so a cached instance that | ||
| # is stopped must be replaced instead of returned | ||
| if instance is None or instance._stopped: | ||
| instance = cls._new_instance(*args) | ||
| cls._instances[key] = instance | ||
|
|
||
| # A service that stops while it is created removes itself from | ||
| # `_instances` before it is stored there, so the caller must remove it | ||
| if instance._stopped: | ||
| cls._instances.pop(key, None) |
There was a problem hiding this comment.
🔍 Bug fix lacks tracked issue
The contribution guide requires an issue before bug fixes. This PR explicitly states that no issue exists.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
This fix comes from a flaky test on main (workflow run 33405376931), not from a user report, so there is no issue to link. The failing run is the tracking record and is described in the PR body.
| @classmethod | ||
| def _new_instance(cls, *args): | ||
| instance = super()._new_instance(*args) | ||
| if cls.fail: | ||
| # Wait for the failed service to finish so that it removes itself from | ||
| # the instance cache before `instance()` stores it there | ||
| instance.drain() | ||
| return instance | ||
|
|
||
| try: | ||
| failed_instance = FailingStartupService.instance() | ||
| assert failed_instance._stopped | ||
| assert FailingStartupService._instances.get(failed_instance._key) is None |
There was a problem hiding this comment.
🔍 Regression tests mirror private internals
The tests override _new_instance and mutate _instances directly. The testing standard prefers public behavior over private implementation details.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
QueueService has no public surface for its lifecycle: instance, _instances, _new_instance, _stop and _stopped are all private, and the existing tests in this file use them the same way (for example test_instance_returns_new_instance_after_stopping and test_lifespan). The new test follows that convention, and the behavior it checks — instance() gives a working service, and the cached service is the one that works — is the behavior callers depend on.
Merging this PR will not alter performance
Comparing 
Footnotes
|
tests/test_assets.py::test_nested_materializationfailed onmain(run, group 10/18, python 3.12 / postgres 14) with:Root cause
QueueService.instance()can keep a stopped service in the class-level_instancescache:instance()calls_new_instance(), which starts the service on the global loop thread._lifespanfails during startup),_runcalls_remove_instance(), which removes the key from_instances.instance()stores the new instance in_instances, so the caller writes the stopped instance into the cache.The stopped instance stays in the cache. Each later
instance()call gets that dead service,send()raisesRuntimeError, and no items are processed. For theEventsWorker, the client also never enters its async context, soAssertingEventsClient.eventsdoes not exist and the test fails with theAttributeError.Fix
instance()now replaces a cached instance that is stopped, and does not cache a new instance that is already stopped. A stopped service can never accept items again, so the cache must not keep it.Reproduction before the fix (a delay between the start of the service and the cache write makes the race deterministic):
After the fix,
instance()gives a new, running service.Verification
tests/_internal/concurrency/test_services.py: 39 passedtests/test_assets.pyandtests/events/client/test_events_worker.py: 78 passedtests/test_assets.py::test_nested_materialization: 20 consecutive passestests/_internal/concurrency,tests/events/client,tests/test_logging.py: onlytests/events/client/test_automations_server_compatibility.py::test_trigger_round_trippingfails, and it fails the same way onmainwithout this changeChecklist
<link to issue>"main.mint.json.Link to Devin session: https://app.devin.ai/sessions/fa2f90b080764e1ba44805f21217bda3
Open in Devin Desktop: https://app.devin.ai/desktop/session/fa2f90b080764e1ba44805f21217bda3?variant=devin