PR #339 review: automatic connection and channel recovery

pr339_review.md

Code Review: PR #339 - Automatic Connection and Channel Recovery

PR: rabbitmq/amqp091-go#339 Author: suchitd (Suchit Dhakate) Reviewed: 2026-06-04 Reviewer: AI (Claude) with @lukebakken guidance Size: +1824 / -63 lines across 11 files

This review was produced by an AI assistant (Claude Code, Anthropic) under the direction of @lukebakken. Findings were verified against the actual source code and each includes a confidence estimate. Human judgement should be applied before acting on any recommendation.

Critical Bugs

1. Channel ID allocator not synchronized with recovered channels

Confidence: 99%

Location: connection.go:1178 / connection.go:1440-1453 / connection.go:986-992

During Reconnect(), calling c.open() -> openTune() creates a brand-new allocator (line 1178). The existing channels in c.channels retain their old IDs but the new allocator has no knowledge of them being in use. When conn.Channel() is subsequently called after a successful recovery, allocateChannel() -> c.allocator.next() can return an ID already occupied by a recovered channel. At line 992, c.channels[uint16(id)] = ch silently overwrites the recovered channel in the map.

Impact: The recovered channel becomes unreachable (lost from the map), and the new channel conflicts with the recovered channel's ID on the wire, causing protocol-level corruption or silent message loss.

Fix: After c.open() succeeds in Reconnect(), iterate c.channels and call c.allocator.reserve(int(ch.id)) for each existing channel ID. Alternatively, preserve the old allocator's state across reconnections instead of creating a new one.

---

2. SetRecoverableErrorCodes / AddRecoverableErrorCodes have data races

Confidence: 99%

Location: connection.go:1541-1561 / connection.go:1516

SetRecoverableErrorCodes (line 1548) and AddRecoverableErrorCodes (line 1560) write to RecoverableErrorCodes without any lock. Meanwhile, isRecoverable() (line 1516) reads the same slice without a lock, from the watchConnection/watchChannel goroutines running concurrently.

AddRecoverableErrorCodes using append is particularly dangerous: if the underlying array has spare capacity, append mutates the existing backing array in-place while isRecoverable might be iterating over the same memory.

Impact: Data race detectable by -race. Could result in corrupted reads, panics, or incorrect recovery decisions.

Fix: Protect RecoverableErrorCodes access with a mutex (either c.m or a dedicated lock). Alternatively, make these methods copy-on-write: build a new slice and assign it atomically.

---

3. DefaultRecoverableErrorCodes shared by reference

Confidence: 95%

Location: recovery.go:20-21 / connection.go:321

When the user provides a ReconnectionConfig with nil RecoverableErrorCodes, line 321 assigns DefaultRecoverableErrorCodes directly without copying:

config.Recovery.ReconnectionConfig.RecoverableErrorCodes = DefaultRecoverableErrorCodes

The connection's slice now shares its backing array with the exported package-level variable. A subsequent AddRecoverableErrorCodes call would append to this shared slice. Whether it corrupts the global depends on whether the backing array has spare capacity (for a literal []int{320, 541}, the capacity equals the length in current Go implementations, so append would allocate a new array). However, the DefaultRecoverableErrorCodes variable is exported and mutable, so any external code doing amqp091.DefaultRecoverableErrorCodes = append(amqp091.DefaultRecoverableErrorCodes, 999) or amqp091.DefaultRecoverableErrorCodes[0] = 0 would affect connections that share the reference.

The DefaultReconnectionConfig.Clone() path (line 319) correctly deep-copies the slice, but the line 321 path does not.

Impact: Mutations to the global variable affect existing connections, or vice versa. Unlikely in typical usage but violates the principle of isolation.

Fix: Copy the slice at line 321:

codes := make([]int, len(DefaultRecoverableErrorCodes))
copy(codes, DefaultRecoverableErrorCodes)
config.Recovery.ReconnectionConfig.RecoverableErrorCodes = codes

---

Significant Issues

4. Fixed retry interval - consider exponential backoff with shorter initial delay

Confidence: 100%

Location: connection.go:1379-1380

for i := 0; i < c.MaxRetryCount(); i++ {
    ...
    jitter := time.Duration(rand.Intn(500)) * time.Millisecond
    time.Sleep(c.RetryInterval() + jitter)

The first connection recovery attempt waits the full 5-second RetryInterval before trying. Sleeping before the first attempt is reasonable - when CONNECTION_FORCED fires the broker may be overloaded or shutting down, and immediate reconnection from many clients creates a thundering herd. However, 5 seconds is arguably too long for the initial attempt, especially in cases where the broker is already back (e.g., a brief network blip).

A better strategy would be a shorter initial sleep (e.g., 1-2 seconds) with larger jitter on the first attempt (to spread out the herd), then exponential backoff to a ceiling of 10-30 seconds for subsequent retries. The current fixed interval means attempt 1 and attempt 5 wait the same duration despite very different likelihood of success.

The channel Reconnect() (line 1978) skips the sleep entirely on i==0, which is appropriate since channel recovery operates on an already-established connection and does not risk flooding the broker's TCP accept queue.

Impact: First recovery attempt is slower than necessary. Under sustained outages, the fixed interval provides no backpressure - all retries arrive at the same cadence regardless of how many have already failed.

Fix: Use exponential backoff: start with a shorter initial interval (e.g., 1 second + larger jitter for thundering-herd spread), then double on each retry up to a maximum (e.g., 30 seconds). Consider making the backoff strategy configurable via ReconnectionConfig.

---

5. LifeCycle.deliverLoop can block indefinitely (goroutine leak)

Confidence: 95%

Location: lifecycle.go:152

deliverLoop() sends to the user's channel without any timeout or drop policy:

ch <- sc  // blocks if channel is full and consumer is not reading

If the user registers a NotifyStateChange listener but stops reading from it (or uses an insufficiently buffered channel), the goroutine blocks forever. Subsequent SetState() calls append to l.queue indefinitely (since l.sending remains true and no new goroutine is started), causing unbounded memory growth.

Impact: Goroutine leak and unbounded memory growth if the user neglects the state change channel. Additionally, if deliverLoop is blocked, SetState() still updates l.state (line 121) but the notification is deferred - so the state and the notifications can diverge significantly.

Fix: Document the contract (user MUST consume), or add a non-blocking send with a drop policy, or add a timeout.

---

6. NotifyStateChange silently replaces previous listener

Confidence: 100%

Location: lifecycle.go:156-159

func (l *LifeCycle) notifyStateChange(channel chan *StateChanged) {
    l.mutex.Lock()
    defer l.mutex.Unlock()
    l.chStatusChanged = channel
}

Calling NotifyStateChange replaces any previous channel. This differs from NotifyClose, NotifyBlocked, NotifyReturn, and NotifyCancel, which all append to a slice and support multiple listeners. If two parts of an application both register, the first one silently stops receiving.

Additionally, if deliverLoop is in-flight when the channel is replaced, it reads l.chStatusChanged on each loop iteration (line 149). Queued state transitions partially delivered to the old channel will switch mid-stream to the new channel, splitting the transition history non-deterministically.

Impact: Surprising API inconsistency. Silent data loss for replaced listeners.

Fix: Either support multiple listeners (append to a slice, matching other Notify* methods) or document clearly that only one listener is supported and additional calls replace the previous one. Return the old channel so callers can detect replacement.

---

7. Dead code: state deduplication check uses pointer equality

Confidence: 100%

Location: lifecycle.go:116-118

if l.state == value {
    return
}

Every call site creates a fresh pointer allocation: &StateOpen{}, &StateReconnecting{}, etc. In Go, interface equality checks both the dynamic type and the dynamic value. For pointer types, the value is the memory address. Two separate &StateOpen{} allocations will never be ==. This check can never be true as the code is currently written - it is dead code.

Impact: No functional impact, but misleads readers into thinking duplicate state transitions are suppressed.

Fix: If deduplication is intended, compare l.state.getState() == value.getState(). If not intended, remove the check.

---

8. No cancellation mechanism for in-progress recovery

Confidence: 100%

Location: connection.go:1380 / connection.go:485-516

Reconnect() uses time.Sleep() which cannot be interrupted. There is no context parameter, no select on a done channel, and no mechanism for Close() to signal the retry loop to abort promptly.

If Close() is called while Reconnect() is sleeping:

• If c.closed is still true (before resetState() on the first attempt), Close() returns ErrClosed immediately - the user cannot stop recovery at all.
• If c.closed is false (after resetState()), Close() sets closeInit = true and proceeds to send a connection.close frame. The loop will exit on the next iteration's condition check (c.MaxRetryCount() returns 0 because IsRecoveryEnabled() sees closeInit == true), but the current iteration runs to completion, potentially racing with Close() on the same socket.

Impact: Users cannot cleanly cancel recovery. Shutdown can be delayed up to RetryInterval + 500ms per remaining retry. Concurrent Close() and Reconnect() can race on socket operations.

Fix: Accept a context.Context parameter (or use an internal done channel). Replace time.Sleep() with a select on a timer and the cancellation signal. Check for cancellation after each major step in the retry loop.

---

9. Latent race: reader goroutine's defer close(c.rpc) vs. resetState()

Confidence: 85% (theoretically valid, not practically exploitable with default config)

Location: connection.go:891 / connection.go:1488

The reader goroutine has defer close(c.rpc) which references the struct field, not a captured channel value. After a failed retry attempt, the reader calls c.shutdown() (which holds destructorM), then shutdown() returns (releasing destructorM), then the reader function returns, and the defer fires.

If the Go scheduler gives the Reconnect() goroutine a turn between shutdown() returning and the defer executing, Reconnect() could acquire destructorM, call resetState() (replacing c.rpc), and then the old reader's defer closes the new c.rpc.

In practice, this is not exploitable with the current default 5-second retry interval because the next iteration sleeps before calling resetState(), and the defer fires within microseconds of shutdown() returning. The race window is nanoseconds vs. a 5-second sleep.

Impact: Under default configuration, none. With very short retry intervals (or under extreme scheduling pressure), the new connection's RPC channel could be closed, breaking the AMQP handshake.

Fix: Capture c.rpc in a local variable at reader start time and close that instead, or move the close(c.rpc) into shutdown() itself (which already holds the appropriate locks).

---

Minor Issues

10. ILifeCycleState / IConnectionRecovery naming not idiomatic Go

Confidence: 100%

Location: lifecycle.go:17 / recovery.go:63

The I prefix for interfaces is a Java/C# convention. Go convention uses just the name (e.g., LifeCycleState, ConnectionRecovery). This is a public API surface that will be difficult to change after release.

Additionally, ILifeCycleState.getState() is unexported, which prevents external packages from implementing the interface. If this is intentional (sealed interface pattern), it should be documented. If not, getState() should be exported.

---

11. Credentials logged in plain text

Confidence: 100%

Location: recovery.go:82,91,95,98

The full AMQP URL (including username and password) is logged in four places via conn.url:

Logger.Printf("Connection %s recovery is not enabled, skipping reconnect. ", conn.url)
Logger.Printf("Connection %s closed with non-recoverable error code %d, skipping reconnect.", conn.url, code)
Logger.Printf("Initiating connection recovery for %s.", conn.url)
Logger.Printf("Connection %s recovery failed: %v.", conn.url, err)

A URL like amqp://admin:s3cret@prod-rabbit:5672/ would appear in logs.

Fix: Log only the host:port portion, or redact credentials from the URL before logging.

---

12. Zero-value MaxRetryCount silently disables recovery

Confidence: 100%

Location: connection.go:1504

If the user passes a custom ReconnectionConfig without explicitly setting MaxRetryCount (leaving it at the zero value), IsRecoveryEnabled() returns false because of: c.Config.Recovery.ReconnectionConfig.MaxRetryCount > 0. Recovery appears configured but silently does nothing.

This only affects the case where the user provides their own ReconnectionConfig. When ReconnectionConfig is nil, DefaultReconnectionConfig.Clone() provides a MaxRetryCount of 5.

Fix: Either validate at DialConfig time and return an error, or treat 0 as "use default" (similar to how ChannelMax == 0 means "use default" elsewhere in the codebase).

---

13. DefaultConnectionRecovery.config field is unused

Confidence: 100%

Location: recovery.go:76 / connection.go:325

type DefaultConnectionRecovery struct {
    config *ReconnectionConfig
}

The config field is set at creation (connection.go:325: config: config.Recovery.ReconnectionConfig) but never accessed in OnConnectionClose or OnChannelClose. Both methods access configuration through the Connection or Channel parameter. This is dead code.

Fix: Remove the field, or use it instead of reaching through conn.IsRecoveryEnabled() / conn.isRecoverable().

---

14. Typos in example

Confidence: 100%

Location: _examples/recovery/recovery.go:105,143,189

"continuosly" appears three times. Should be "continuously".

---

15. Test utilities hardcode host, port, and credentials

Confidence: 100%

Location: test/utils/http.go:26 / recovery_test.go:318

Connections() hardcodes http://localhost:15672 and guest:guest. There is no way to override these via environment variables. DropConnection accepts a port parameter but Connections() does not, creating an inconsistency.

This is adequate for CI (where the workflow maps port 15672) but breaks for anyone running RabbitMQ on non-default configuration or in containers with different port mappings.

Fix: Read host/port/credentials from environment variables (e.g., RABBITMQ_MANAGEMENT_URL) with localhost:15672 as fallback, consistent with how amqpURL is handled elsewhere in the test suite.

---

Design Observations

---

17. Redundant recovery triggers during retry loop

Confidence: 90%

When a failed retry attempt's reader calls shutdown(), it sends an error to c.closes listeners. The watchConnection goroutine receives this and calls Reconnect() again, which blocks on c.reconnecting.Lock(). After the original Reconnect() returns, the redundant call proceeds but exits immediately via if !c.IsClosed() { return nil } (if recovery succeeded) or retries (if it failed).

This is functionally safe but generates unnecessary log noise ("Connection closed unexpectedly" messages for internal retry failures) and creates goroutines that block on the reconnecting mutex for the duration of the recovery.

Fix: Consider not sending errors from retry-internal readers to c.closes, or have watchConnection check whether recovery is already in progress before calling OnConnectionClose.