fix(broker): never close a typed-nil logical connection
Opening a logical stream on the physical connection returned the logical connector's concrete *LogicalConnection directly, so an open failure handed back an interface that compares unequal to nil while holding no receiver. The pool saw a non-nil connection, closed it, and dereferenced the nil pointer: a panic in the goroutine serving a client stream, reachable from any failing open. - The physical connection returns an untyped nil on failure and treats an unexpected nil connection as an invalid admission, so the interface is nil whenever the open failed. - The pool stops calling Close on a connection that is nil behind the interface, using the package's existing typed-nil-aware check and the same check for the unexpected nil after a successful open, so any future adapter returning a concrete nil cannot panic the serving goroutine. - (*LogicalConnection).Close tolerates a nil receiver, matching the physical connection's existing guard. Verified by mutation: restoring the direct return fails the adapter test with "an open failure must not return a typed-nil connection", and restoring the nil-comparison fails the pool test with the nil pointer dereference itself.
- commit
- 976dd322f99898c9adcb1eaf649bf285ceae031b
- author
- brice · · 2026-09-19 12:31 UTC
- parent
- 0ede6cb
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +76 -3
M internal/adapters/daemonmux/connector.go +12 -1
| @@ -287,7 +287,18 @@ func (c *PhysicalConnection) OpenStream(ctx context.Context, request ports.Broke | |||
| 287 | 287 | if !request.Policy.Compatible(c.policy) { | |
| 288 | 288 | return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy} | |
| 289 | 289 | } | |
| 290 | - | return c.logical.Open(ctx, request) | |
| 290 | + | // Never hand back a typed nil: Open returns a concrete *LogicalConnection, so | |
| 291 | + | // returning it directly on the error path would wrap a nil pointer in a | |
| 292 | + | // non-nil interface, and a caller that checks the interface against nil would | |
| 293 | + | // then call a method on it. | |
| 294 | + | connection, err := c.logical.Open(ctx, request) | |
| 295 | + | if err != nil { | |
| 296 | + | return nil, err | |
| 297 | + | } | |
| 298 | + | if connection == nil { | |
| 299 | + | return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid} | |
| 300 | + | } | |
| 301 | + | return connection, nil | |
| 291 | 302 | } | |
| 292 | 303 | ||
| 293 | 304 | // Close tears the physical connection down exactly once: it closes the pump, | |
| @@ -287,7 +287,18 @@ func (c *PhysicalConnection) OpenStream(ctx context.Context, request ports.Broke | |||||
| 287 | if !request.Policy.Compatible(c.policy) { | 287 | if !request.Policy.Compatible(c.policy) { | ||
| 288 | return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy} | 288 | return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy} | ||
| 289 | } | 289 | } | ||
| 290 | - | return c.logical.Open(ctx, request) | 290 | + | // Never hand back a typed nil: Open returns a concrete *LogicalConnection, so |
| 291 | + | // returning it directly on the error path would wrap a nil pointer in a | |||
| 292 | + | // non-nil interface, and a caller that checks the interface against nil would | |||
| 293 | + | // then call a method on it. | |||
| 294 | + | connection, err := c.logical.Open(ctx, request) | |||
| 295 | + | if err != nil { | |||
| 296 | + | return nil, err | |||
| 297 | + | } | |||
| 298 | + | if connection == nil { | |||
| 299 | + | return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid} | |||
| 300 | + | } | |||
| 301 | + | return connection, nil | |||
| 291 | } | 302 | } | ||
| 292 | 303 | ||||
| 293 | // Close tears the physical connection down exactly once: it closes the pump, | 304 | // Close tears the physical connection down exactly once: it closes the pump, | ||
M internal/adapters/daemonmux/logical_conn.go +3 -0
| @@ -374,6 +374,9 @@ func (c *LogicalConnection) ReceiveServer() (protocol.ServerMessage, error) { | |||
| 374 | 374 | // stream. Close is idempotent and concurrent-safe; every caller observes the | |
| 375 | 375 | // same stored transport Close error. | |
| 376 | 376 | func (c *LogicalConnection) Close() error { | |
| 377 | + | if c == nil { | |
| 378 | + | return nil | |
| 379 | + | } | |
| 377 | 380 | c.closeOnce.Do(func() { | |
| 378 | 381 | // Discard what the engine already accepted before closing locally: this | |
| 379 | 382 | // side stops reading as soon as it closes, so draining first is what lets | |
| @@ -374,6 +374,9 @@ func (c *LogicalConnection) ReceiveServer() (protocol.ServerMessage, error) { | |||||
| 374 | // stream. Close is idempotent and concurrent-safe; every caller observes the | 374 | // stream. Close is idempotent and concurrent-safe; every caller observes the | ||
| 375 | // same stored transport Close error. | 375 | // same stored transport Close error. | ||
| 376 | func (c *LogicalConnection) Close() error { | 376 | func (c *LogicalConnection) Close() error { | ||
| 377 | + | if c == nil { | |||
| 378 | + | return nil | |||
| 379 | + | } | |||
| 377 | c.closeOnce.Do(func() { | 380 | c.closeOnce.Do(func() { | ||
| 378 | // Discard what the engine already accepted before closing locally: this | 381 | // Discard what the engine already accepted before closing locally: this | ||
| 379 | // side stops reading as soon as it closes, so draining first is what lets | 382 | // side stops reading as soon as it closes, so draining first is what lets | ||
A internal/adapters/daemonmux/typed_nil_test.go +31 -0
| @@ -0,0 +1,31 @@ | |||
| 1 | + | package daemonmux | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "testing" | |
| 6 | + | ||
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // TestOpenFailureReturnsNilInterface proves a failed open hands back an | |
| 11 | + | // interface that compares equal to nil. The logical connector returns a concrete | |
| 12 | + | // *LogicalConnection, so returning that value directly on the error path wraps a | |
| 13 | + | // nil pointer in a non-nil interface: a caller that checks the interface against | |
| 14 | + | // nil then calls a method on a nil receiver instead of reporting the failure. | |
| 15 | + | func TestOpenFailureReturnsNilInterface(t *testing.T) { | |
| 16 | + | // A connector without a pump fails every open with ErrLogicalConfig, which | |
| 17 | + | // is the failure shape that used to return a typed nil. | |
| 18 | + | physical := &PhysicalConnection{logical: &LogicalConnector{}, policy: logicalTestPolicy()} | |
| 19 | + | ||
| 20 | + | connection, err := physical.OpenStream(context.Background(), controlRequest(1)) | |
| 21 | + | require.Error(t, err) | |
| 22 | + | require.True(t, connection == nil, "an open failure must not return a typed-nil connection") | |
| 23 | + | } | |
| 24 | + | ||
| 25 | + | // TestNilLogicalConnectionCloseIsSafe proves the receiver guard: a caller that | |
| 26 | + | // received a typed-nil connection from anywhere can still release it, exactly as | |
| 27 | + | // a physical connection can. | |
| 28 | + | func TestNilLogicalConnectionCloseIsSafe(t *testing.T) { | |
| 29 | + | var connection *LogicalConnection | |
| 30 | + | require.NoError(t, connection.Close()) | |
| 31 | + | } | |
| @@ -0,0 +1,31 @@ | |||||
| 1 | + | package daemonmux | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "testing" | |||
| 6 | + | ||||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // TestOpenFailureReturnsNilInterface proves a failed open hands back an | |||
| 11 | + | // interface that compares equal to nil. The logical connector returns a concrete | |||
| 12 | + | // *LogicalConnection, so returning that value directly on the error path wraps a | |||
| 13 | + | // nil pointer in a non-nil interface: a caller that checks the interface against | |||
| 14 | + | // nil then calls a method on a nil receiver instead of reporting the failure. | |||
| 15 | + | func TestOpenFailureReturnsNilInterface(t *testing.T) { | |||
| 16 | + | // A connector without a pump fails every open with ErrLogicalConfig, which | |||
| 17 | + | // is the failure shape that used to return a typed nil. | |||
| 18 | + | physical := &PhysicalConnection{logical: &LogicalConnector{}, policy: logicalTestPolicy()} | |||
| 19 | + | ||||
| 20 | + | connection, err := physical.OpenStream(context.Background(), controlRequest(1)) | |||
| 21 | + | require.Error(t, err) | |||
| 22 | + | require.True(t, connection == nil, "an open failure must not return a typed-nil connection") | |||
| 23 | + | } | |||
| 24 | + | ||||
| 25 | + | // TestNilLogicalConnectionCloseIsSafe proves the receiver guard: a caller that | |||
| 26 | + | // received a typed-nil connection from anywhere can still release it, exactly as | |||
| 27 | + | // a physical connection can. | |||
| 28 | + | func TestNilLogicalConnectionCloseIsSafe(t *testing.T) { | |||
| 29 | + | var connection *LogicalConnection | |||
| 30 | + | require.NoError(t, connection.Close()) | |||
| 31 | + | } | |||
M internal/usecase/broker/pool.go +5 -2
| @@ -244,12 +244,15 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest | |||
| 244 | 244 | stopEntry() | |
| 245 | 245 | openCancel() | |
| 246 | 246 | if err != nil { | |
| 247 | - | if raw != nil { | |
| 247 | + | // A failed open may still carry a typed-nil connection, which compares | |
| 248 | + | // unequal to nil while holding no receiver: closing it would dereference | |
| 249 | + | // a nil pointer. | |
| 250 | + | if !nilDependency(raw) { | |
| 248 | 251 | _ = raw.Close() | |
| 249 | 252 | } | |
| 250 | 253 | return nil, p.streamTerminal(ctx, entry, req, normalizePoolError(err)) | |
| 251 | 254 | } | |
| 252 | - | if raw == nil { | |
| 255 | + | if nilDependency(raw) { | |
| 253 | 256 | return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid} | |
| 254 | 257 | } | |
| 255 | 258 | if terminal := p.streamTerminal(ctx, entry, req, nil); terminal != nil { | |
| @@ -244,12 +244,15 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest | |||||
| 244 | stopEntry() | 244 | stopEntry() | ||
| 245 | openCancel() | 245 | openCancel() | ||
| 246 | if err != nil { | 246 | if err != nil { | ||
| 247 | - | if raw != nil { | 247 | + | // A failed open may still carry a typed-nil connection, which compares |
| 248 | + | // unequal to nil while holding no receiver: closing it would dereference | |||
| 249 | + | // a nil pointer. | |||
| 250 | + | if !nilDependency(raw) { | |||
| 248 | _ = raw.Close() | 251 | _ = raw.Close() | ||
| 249 | } | 252 | } | ||
| 250 | return nil, p.streamTerminal(ctx, entry, req, normalizePoolError(err)) | 253 | return nil, p.streamTerminal(ctx, entry, req, normalizePoolError(err)) | ||
| 251 | } | 254 | } | ||
| 252 | - | if raw == nil { | 255 | + | if nilDependency(raw) { |
| 253 | return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid} | 256 | return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid} | ||
| 254 | } | 257 | } | ||
| 255 | if terminal := p.streamTerminal(ctx, entry, req, nil); terminal != nil { | 258 | if terminal := p.streamTerminal(ctx, entry, req, nil); terminal != nil { | ||
M internal/usecase/broker/pool_test.go +25 -0
| @@ -114,6 +114,31 @@ func TestPoolRefusedResolutionNeverDials(t *testing.T) { | |||
| 114 | 114 | require.Zero(t, dials.Load(), "a refused request must never dial") | |
| 115 | 115 | } | |
| 116 | 116 | ||
| 117 | + | // TestPoolOpenFailureWithTypedNilConnection proves a failed open that still | |
| 118 | + | // carries a typed-nil logical connection is not closed, and never panics the | |
| 119 | + | // serving goroutine. A concrete-typed adapter returning a nil pointer directly | |
| 120 | + | // produces an interface that compares unequal to nil while holding no receiver. | |
| 121 | + | func TestPoolOpenFailureWithTypedNilConnection(t *testing.T) { | |
| 122 | + | p, _ := setupPool(t, func(_ context.Context, e ports.BrokerResolvedEndpoint) (ports.BrokerPhysicalConnection, error) { | |
| 123 | + | return &fakePhysical{ | |
| 124 | + | endpoint: e, | |
| 125 | + | done: make(chan struct{}), | |
| 126 | + | open: func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 127 | + | var absent *fakeLogical | |
| 128 | + | return absent, errors.New("open failed") | |
| 129 | + | }, | |
| 130 | + | }, nil | |
| 131 | + | }) | |
| 132 | + | id, err := p.RegisterClient() | |
| 133 | + | require.NoError(t, err) | |
| 134 | + | ||
| 135 | + | stream, err := p.OpenStream(context.Background(), poolRequest(id, 1)) | |
| 136 | + | require.Error(t, err) | |
| 137 | + | require.True(t, stream == nil, "a failed open must not yield a stream") | |
| 138 | + | var typed ports.BrokerError | |
| 139 | + | require.ErrorAs(t, err, &typed) | |
| 140 | + | } | |
| 141 | + | ||
| 117 | 142 | func TestPoolHundredStreamsLossAndIsolation(t *testing.T) { | |
| 118 | 143 | var calls atomic.Int32 | |
| 119 | 144 | var physical *fakePhysical | |
| @@ -114,6 +114,31 @@ func TestPoolRefusedResolutionNeverDials(t *testing.T) { | |||||
| 114 | require.Zero(t, dials.Load(), "a refused request must never dial") | 114 | require.Zero(t, dials.Load(), "a refused request must never dial") | ||
| 115 | } | 115 | } | ||
| 116 | 116 | ||||
| 117 | + | // TestPoolOpenFailureWithTypedNilConnection proves a failed open that still | |||
| 118 | + | // carries a typed-nil logical connection is not closed, and never panics the | |||
| 119 | + | // serving goroutine. A concrete-typed adapter returning a nil pointer directly | |||
| 120 | + | // produces an interface that compares unequal to nil while holding no receiver. | |||
| 121 | + | func TestPoolOpenFailureWithTypedNilConnection(t *testing.T) { | |||
| 122 | + | p, _ := setupPool(t, func(_ context.Context, e ports.BrokerResolvedEndpoint) (ports.BrokerPhysicalConnection, error) { | |||
| 123 | + | return &fakePhysical{ | |||
| 124 | + | endpoint: e, | |||
| 125 | + | done: make(chan struct{}), | |||
| 126 | + | open: func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 127 | + | var absent *fakeLogical | |||
| 128 | + | return absent, errors.New("open failed") | |||
| 129 | + | }, | |||
| 130 | + | }, nil | |||
| 131 | + | }) | |||
| 132 | + | id, err := p.RegisterClient() | |||
| 133 | + | require.NoError(t, err) | |||
| 134 | + | ||||
| 135 | + | stream, err := p.OpenStream(context.Background(), poolRequest(id, 1)) | |||
| 136 | + | require.Error(t, err) | |||
| 137 | + | require.True(t, stream == nil, "a failed open must not yield a stream") | |||
| 138 | + | var typed ports.BrokerError | |||
| 139 | + | require.ErrorAs(t, err, &typed) | |||
| 140 | + | } | |||
| 141 | + | ||||
| 117 | func TestPoolHundredStreamsLossAndIsolation(t *testing.T) { | 142 | func TestPoolHundredStreamsLossAndIsolation(t *testing.T) { | ||
| 118 | var calls atomic.Int32 | 143 | var calls atomic.Int32 | ||
| 119 | var physical *fakePhysical | 144 | var physical *fakePhysical | ||