Mqtt: wait for retained message after subscribe before first read

[andig]

Mqtt getters without timeout returned outdated on first read when a retained message had not yet arrived — retained delivery is asynchronous, queued by the broker right after suback.

Since Listen already waits for the suback, the receiver now simply waits up to 100ms for the first value before returning when no timeout is configured. A retained message resolves the wait immediately; topics without retained value fall through after the grace period with unchanged behavior.

Refs #32001

🤖 Generated with Claude Code

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• In Barrier, entries in syncWait are only removed on successful echo; if the context or timeout fires first, the map will leak entries over time, so consider cleaning up the nonce entry on non-success paths.
• The use of sync.Once with Listen means that if the initial Listen on syncTopic fails, subsequent Barrier calls will never retry and will keep returning the original error; you may want to avoid marking sync.Once as done on failure or otherwise support retrying the subscription.
• Barrier currently races between ctx.Done and a separate time.After(request.Timeout); if both represent timeouts, consider unifying them (e.g. by deriving a context with the request timeout) to simplify behavior and avoid ambiguous timeout sources.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In Barrier, entries in syncWait are only removed on successful echo; if the context or timeout fires first, the map will leak entries over time, so consider cleaning up the nonce entry on non-success paths.
- The use of sync.Once with Listen means that if the initial Listen on syncTopic fails, subsequent Barrier calls will never retry and will keep returning the original error; you may want to avoid marking sync.Once as done on failure or otherwise support retrying the subscription.
- Barrier currently races between ctx.Done and a separate time.After(request.Timeout); if both represent timeouts, consider unifying them (e.g. by deriving a context with the request timeout) to simplify behavior and avoid ambiguous timeout sources.

## Individual Comments

### Comment 1
<location path="plugin/mqtt/client.go" line_range="274-283" />
<code_context>
+// its echo, relying on the broker delivering a connection's messages in order.
+func (m *Client) Barrier(ctx context.Context) error {
+	var err error
+	m.syncOnce.Do(func() {
+		err = m.Listen(m.syncTopic, func(payload string) {
+			m.syncMu.Lock()
+			defer m.syncMu.Unlock()
+
+			if ch, ok := m.syncWait[payload]; ok {
+				close(ch)
+				delete(m.syncWait, payload)
+			}
+		})
+	})
+	if err != nil {
+		return err
</code_context>
<issue_to_address>
**issue (bug_risk):** Consider allowing retry if the sync subscription setup fails the first time.

Wrapping `Listen` in `sync.Once` means a transient failure (e.g., connection issue) on the first `Barrier` call will set `syncOnce` as done, and all future `Barrier` calls will skip subscription, breaking sync for the client lifetime. Consider only marking `syncOnce` after a successful subscription, or tracking subscription state separately so failed setups can be retried.
</issue_to_address>

### Comment 2
<location path="plugin/mqtt/client.go" line_range="293-294" />
<code_context>
+	ch := make(chan struct{})
+
+	m.syncMu.Lock()
+	m.syncWait[nonce] = ch
+	m.syncMu.Unlock()
+
+	m.client.Publish(m.syncTopic, m.Qos, false, nonce)
</code_context>
<issue_to_address>
**suggestion:** Clean up syncWait entries when the wait ends without a response to avoid potential leaks.

When `Barrier` exits via `ctx.Done()` or timeout and the echo never arrives, the `nonce` entry stays in `syncWait` indefinitely, and repeated failures can accumulate stale entries. Consider cleaning up the map entry when the wait ends (e.g., with a `defer` that locks `syncMu` and deletes `syncWait[nonce]`), while having the listener check the map before closing the channel to safely handle late messages.

Suggested implementation:

```golang
	nonce := fmt.Sprintf("%d", rand.Int64())
	ch := make(chan struct{})

	m.syncMu.Lock()
	m.syncWait[nonce] = ch
	m.syncMu.Unlock()

	// Ensure we don't leak syncWait entries when the barrier exits
	defer func() {
		m.syncMu.Lock()
		delete(m.syncWait, nonce)
		m.syncMu.Unlock()
	}()

	m.client.Publish(m.syncTopic, m.Qos, false, nonce)

	select {
	case <-ch:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(request.Timeout):
		return fmt.Errorf("sync %s: %w", m.syncTopic, api.ErrTimeout)
	}
}

```

To fully implement the suggestion, the MQTT listener code that handles sync echo messages should:
1. Lock `syncMu`, look up `syncWait[nonce]`, and only proceed if the entry exists.
2. If present, read the channel, unlock, and close the channel; if absent (barrier already cleaned up), simply unlock and ignore the late message.

This ensures the deferred deletion here is safe and that late messages don't attempt to close channels that are no longer tracked.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}