Skip to content

fix: await MQTT unsubscribe tokens during shutdown - #673

Merged
xe-nvdk merged 3 commits into
Basekick-Labs:mainfrom
mah1104ahm:codex/fix-mqtt-unsubscribe
Sep 3, 2026
Merged

xe-nvdk merged 3 commits into
Basekick-Labs:mainfrom
mah1104ahm:codex/fix-mqtt-unsubscribe

Conversation

@mah1104ahm

Copy link
Copy Markdown
Contributor

Summary

  • await each MQTT unsubscribe token during subscriber shutdown
  • bound each wait with a one-second timeout and preserve the first failure
  • always disconnect after unsubscribe attempts
  • cover success, broker-error, and timeout paths

Verification

  • gofmt on changed Go files
  • go test ./internal/mqtt
  • git diff --check

Closes #303

@xe-nvdk xe-nvdk left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the fix, this closes the gap #303 describes and the shape is right: the token wait is bounded, the first failure is preserved while the remaining topics still get their unsubscribe attempt, and the disconnect still always runs. I checked the mock against paho's Token interface (all four methods covered) and ran go test ./internal/mqtt on the branch, which passes.

Two asks before merging:

  1. Please add a ## Fixed: entry to RELEASE_NOTES_2026.09.2.md for this, matching the entries the sibling PRs in this series carry (see #663 through #667 and #670).
  2. One behavioral follow-through from Stop() now returning an error, in the inline comment.

s.logger.Info().Msg("Disconnected from MQTT broker")
s.updateStatus(StatusStopped, "")

return unsubscribeErr

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Returning an error here changes behavior at two call sites in manager.go. StopSubscription and PauseSubscription return early when Stop() errors, which skips m.repo.UpdateStatus. At that point the subscriber is already deleted from m.subscribers and the client has disconnected, so the subscription is genuinely stopped while the persisted status keeps saying running, and retrying the stop returns ErrSubscriptionNotRunning, so the status stays wrong until the next start. (ListAutoStart selects on auto_start only, so nothing gets resurrected on restart; this is an API status consistency issue, not a data one.)

An unsubscribe timeout is most likely exactly when the broker is unreachable, so this path will happen in practice. Suggest having those two manager paths persist the status update even when Stop() reports an unsubscribe error, then return the error afterwards.

@mah1104ahm

mah1104ahm commented Sep 3, 2026 via email

Copy link
Copy Markdown
Contributor Author

@xe-nvdk xe-nvdk left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Re-reviewed a1ca9d3. Both asks are addressed: StopSubscription and PauseSubscription now persist the status update regardless of the unsubscribe outcome and return the joined error, and the release notes entry is in place with the credit line. Verified locally on the branch: go test ./internal/mqtt passes, and the new manager regression passes under -race for all four stop/pause x broker-error/timeout cases. Thanks for the thorough test coverage.

@xe-nvdk
xe-nvdk merged commit 3d96194 into Basekick-Labs:main Sep 3, 2026
1 check passed
xe-nvdk added a commit that referenced this pull request Sep 3, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

high(mqtt): Unsubscribe tokens not awaited in Stop()

2 participants