fix(mqtt/greengrass): release pooled subscriptions via linger eviction - #132
Merged
Merged
Conversation
Pooled IPC subscription streams were never terminated, so every distinct topic ever subscribed stayed subscribed at the cloud broker for the life of the process. Each is a cloud-side MQTT subscription counting toward AWS IoT's per-connection limit; past 50 the nucleus opens a `<thing>#2` connection that a thing-policy-variable IoT policy rejects, leaving the device online-but-unmanageable (factbird-edge-applications#191, section C). Give each pooled slot a lifecycle instead of immortality: - On the last subscriber's unsubscribe/drop, the slot goes idle and arms a `linger` timer rather than terminating immediately. Reuse within the window bumps a generation counter so the pending timer no-ops and the live stream is reused — no cloud churn, so the resubscribe race never arises for the common request/response pattern. - A slot idle for the full `linger` is evicted: removed from the pool and its forwarder aborted, dropping the StreamOperation which sends TERMINATE_STREAM. - Because terminate is un-acked, a `settle` guard delays the next subscribe to a just-terminated topic so the stale teardown is processed by the core first. Defaults: linger 5s, settle 500ms, both overridable via with_linger/with_settle. Adds pooled_subscription_count()/pooled_topics() for observability. The pool->slot cycle is broken with a Weak handle; eviction and claim serialize on the pool write lock and eviction is ptr_eq-guarded against replacement slots.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
The Greengrass IPC pool never terminated subscription streams (
src/mqtt/greengrass.rs), so every distinct topic ever subscribed stayed subscribed at the cloud broker for the life of the process. Each one is a cloud-side MQTT subscription counting toward AWS IoT's per-connection limit. Past 50 the nucleus opens a second connection under client ID<thing>#2, which a thing-policy-variable IoT policy rejects (NOT_AUTHORIZED), leaving the device online-but-unmanageable. This is the rustot-side hygiene work called out in factbird-edge-applications#191, section C.This makes
unsubscribe/Dropactually release subscriptions — bounded, and without reintroducing the resubscribe race the pool was originally built to avoid.Lifecycle: linger eviction
Each pooled slot now has a lifecycle instead of being immortal:
lingertimer instead of terminating immediately.generationcounter (bumped on every claim, under the pool write lock) makes a resubscribe-within-lingera pure reuse: the pending timer sees the new generation and no-ops, so the live stream is reused with no cloud churn — the terminate is never sent, so the race can't occur for the common request/response cycle.lingeris evicted: removed from the pool and its forwarder aborted, which drops theStreamOperationand sendsTERMINATE_STREAM, releasing the cloud subscription.Settle guard
Because
TERMINATE_STREAMis un-acked, a fresh subscribe to a just-terminated topic could overlap the in-flight teardown. The next subscribe to a topic terminated less thansettleago waits out the remainder, so the stale terminate is processed by the core first.Knobs & observability
lingerdefault 5s,settledefault 500ms — overridable viawith_linger/with_settle.pooled_subscription_count()/pooled_topics()for monitoring the live subscription footprint.Correctness notes
Weakhandle in the slot, so slots don't leak.Arc::ptr_eq-guarded so it can't remove a replacement slot.subscribe_to_iot_coreawaits theSubscribeToTopicResponsebefore returning, so a fresh resubscribe is live before any publish — no solicited response can be missed.Scope
This is section C only — it bounds and reclaims subscriptions to buy headroom. A busy device can still legitimately exceed 50, so it does not replace the IoT policy fix (options A/B in the issue). The multi-shadow manager still subscribes per-shadow; moving those request/response paths onto the existing wildcard mechanism (making the count constant rather than merely bounded) is a possible follow-up.
One item to verify against a real core before fleet rollout: whether the GG core refcounts cloud subscriptions per-topic across streams. The settle guard keeps us safe under the pessimistic reading either way.