Skip to content

Commit 2fe3f42

Browse files
fabianfettfabianfett
authored
Crash fix: HTTP2Connections emit events after the pool has closed them. (#481)
1 parent 1f3f141 commit 2fe3f42

7 files changed

+104
-19
lines changed

‎Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ extension HTTPConnectionPool {
108108
)
109109
}
110110

111-
// MARK: - Events
111+
// MARK: - Events -
112112

113113
mutating func executeRequest(_ request: Request) -> Action {
114114
switch self.state {
@@ -519,16 +519,20 @@ extension HTTPConnectionPool {
519519
// MARK: HTTP2
520520

521521
mutating func newHTTP2MaxConcurrentStreamsReceived(_ connectionID: Connection.ID, newMaxStreams: Int) -> Action {
522-
// It is save to bang the http2Connections here. If we get this callback but we don't have
523-
// http2 connections something has gone terribly wrong.
524-
_ = self.http2Connections!.newHTTP2MaxConcurrentStreamsReceived(connectionID, newMaxStreams: newMaxStreams)
522+
// The `http2Connections` are optional here:
523+
// Connections report events back to us, if they are in a shutdown that was
524+
// initiated by the state machine. For this reason this callback might be invoked
525+
// even though all references to HTTP2Connections have already been cleared.
526+
_ = self.http2Connections?.newHTTP2MaxConcurrentStreamsReceived(connectionID, newMaxStreams: newMaxStreams)
525527
return .none
526528
}
527529

528530
mutating func http2ConnectionGoAwayReceived(_ connectionID: Connection.ID) -> Action {
529-
// It is save to bang the http2Connections here. If we get this callback but we don't have
530-
// http2 connections something has gone terribly wrong.
531-
_ = self.http2Connections!.goAwayReceived(connectionID)
531+
// The `http2Connections` are optional here:
532+
// Connections report events back to us, if they are in a shutdown that was
533+
// initiated by the state machine. For this reason this callback might be invoked
534+
// even though all references to HTTP2Connections have already been cleared.
535+
_ = self.http2Connections?.goAwayReceived(connectionID)
532536
return .none
533537
}
534538

‎Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -523,9 +523,14 @@ extension HTTPConnectionPool {
523523
/// Sets the connection with the given `connectionId` to the draining state.
524524
/// - Returns: the `EventLoop` to create a new connection on if applicable
525525
/// - Precondition: connection with given `connectionId` must be either `.active` or already in the `.draining` state
526-
mutating func goAwayReceived(_ connectionID: Connection.ID) -> GoAwayContext {
526+
mutating func goAwayReceived(_ connectionID: Connection.ID) -> GoAwayContext? {
527527
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
528-
preconditionFailure("go away recieved for a connection that does not exists")
528+
// When a connection close is initiated by the connection pool (e.g. because the
529+
// connection was idle for too long), the connection will still report further
530+
// events to the state machine even though we don't care about its state anymore.
531+
//
532+
// This is because the HTTP2Connection has a strong let reference to its delegate.
533+
return nil
529534
}
530535
let eventLoop = self.connections[index].goAwayReceived()
531536
return GoAwayContext(eventLoop: eventLoop)
@@ -540,9 +545,13 @@ extension HTTPConnectionPool {
540545
mutating func newHTTP2MaxConcurrentStreamsReceived(
541546
_ connectionID: Connection.ID,
542547
newMaxStreams: Int
543-
) -> (Int, EstablishedConnectionContext) {
548+
) -> (Int, EstablishedConnectionContext)? {
544549
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
545-
preconditionFailure("We tried to update the maximum number of concurrent streams for a connection that does not exists")
550+
// When a connection close is initiated by the connection pool (e.g. because the
551+
// connection was idle for too long), the connection will still report its events to
552+
// the state machine and hence to this `HTTP2Connections` struct. In those cases we
553+
// must ignore the event.
554+
return nil
546555
}
547556
let availableStreams = self.connections[index].newMaxConcurrentStreams(newMaxStreams)
548557
let context = EstablishedConnectionContext(
@@ -661,8 +670,10 @@ extension HTTPConnectionPool {
661670

662671
mutating func failConnection(_ connectionID: Connection.ID) -> (Int, FailedConnectionContext)? {
663672
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
664-
/// When a connection close is initiated by the connection pool (e.g. because the connection was idle for too long), the connection will
665-
/// still report its close to the state machine and then to us. In those cases we must ignore the event.
673+
// When a connection close is initiated by the connection pool (e.g. because the
674+
// connection was idle for too long), the connection will still report its close to
675+
// the state machine and then to this `HTTP2Connections` struct. In those cases we
676+
// must ignore the event.
666677
return nil
667678
}
668679
self.connections[index].fail()

‎Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -293,12 +293,22 @@ extension HTTPConnectionPool {
293293
}
294294

295295
mutating func newHTTP2MaxConcurrentStreamsReceived(_ connectionID: Connection.ID, newMaxStreams: Int) -> Action {
296-
let (index, context) = self.connections.newHTTP2MaxConcurrentStreamsReceived(connectionID, newMaxStreams: newMaxStreams)
296+
guard let (index, context) = self.connections.newHTTP2MaxConcurrentStreamsReceived(connectionID, newMaxStreams: newMaxStreams) else {
297+
// When a connection close is initiated by the connection pool, the connection will
298+
// still report further events (like newMaxConcurrentStreamsReceived) to the state
299+
// machine. In those cases we must ignore the event.
300+
return .none
301+
}
297302
return .init(self.nextActionForAvailableConnection(at: index, context: context))
298303
}
299304

300305
mutating func http2ConnectionGoAwayReceived(_ connectionID: Connection.ID) -> Action {
301-
let context = self.connections.goAwayReceived(connectionID)
306+
guard let context = self.connections.goAwayReceived(connectionID) else {
307+
// When a connection close is initiated by the connection pool, the connection will
308+
// still report further events (like GOAWAY received) to the state machine. In those
309+
// cases we must ignore the event.
310+
return .none
311+
}
302312
return self.nextActionForClosingConnection(on: context.eventLoop)
303313
}
304314

‎Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest+XCTest.swift

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ extension HTTPConnectionPool_HTTP2ConnectionsTests {
3838
("testLeasingAllConnections", testLeasingAllConnections),
3939
("testGoAway", testGoAway),
4040
("testNewMaxConcurrentStreamsSetting", testNewMaxConcurrentStreamsSetting),
41+
("testEventsAfterConnectionIsClosed", testEventsAfterConnectionIsClosed),
4142
("testLeaseOnPreferredEventLoopWithoutAnyAvailable", testLeaseOnPreferredEventLoopWithoutAnyAvailable),
4243
("testMigrationFromHTTP1", testMigrationFromHTTP1),
4344
("testMigrationToHTTP1", testMigrationToHTTP1),

‎Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest.swift

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -371,7 +371,7 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase {
371371
XCTAssertEqual(leasedConn1, conn1)
372372
XCTAssertEqual(leasdConnContext1.wasIdle, true)
373373

374-
XCTAssertTrue(connections.goAwayReceived(conn1ID).eventLoop === el1)
374+
XCTAssertTrue(connections.goAwayReceived(conn1ID)?.eventLoop === el1)
375375

376376
XCTAssertEqual(
377377
connections.stats,
@@ -389,7 +389,7 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase {
389389
XCTAssertNil(connections.leaseStream(onRequired: el1), "we should not be able to lease a stream because the connection is draining")
390390

391391
// a server can potentially send more than one connection go away and we should not crash
392-
XCTAssertTrue(connections.goAwayReceived(conn1ID).eventLoop === el1)
392+
XCTAssertTrue(connections.goAwayReceived(conn1ID)?.eventLoop === el1)
393393
XCTAssertEqual(
394394
connections.stats,
395395
.init(
@@ -454,7 +454,9 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase {
454454

455455
XCTAssertNil(connections.leaseStream(onRequired: el1), "all streams are in use")
456456

457-
let (_, newSettingsContext1) = connections.newHTTP2MaxConcurrentStreamsReceived(conn1ID, newMaxStreams: 2)
457+
guard let (_, newSettingsContext1) = connections.newHTTP2MaxConcurrentStreamsReceived(conn1ID, newMaxStreams: 2) else {
458+
return XCTFail("Expected to get a new settings context")
459+
}
458460
XCTAssertEqual(newSettingsContext1.availableStreams, 1)
459461
XCTAssertTrue(newSettingsContext1.eventLoop === el1)
460462
XCTAssertFalse(newSettingsContext1.isIdle)
@@ -465,7 +467,9 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase {
465467
XCTAssertEqual(leasedConn2, conn1)
466468
XCTAssertEqual(leaseContext2.wasIdle, false)
467469

468-
let (_, newSettingsContext2) = connections.newHTTP2MaxConcurrentStreamsReceived(conn1ID, newMaxStreams: 1)
470+
guard let (_, newSettingsContext2) = connections.newHTTP2MaxConcurrentStreamsReceived(conn1ID, newMaxStreams: 1) else {
471+
return XCTFail("Expected to get a new settings context")
472+
}
469473
XCTAssertEqual(newSettingsContext2.availableStreams, 0)
470474
XCTAssertTrue(newSettingsContext2.eventLoop === el1)
471475
XCTAssertFalse(newSettingsContext2.isIdle)
@@ -489,6 +493,41 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase {
489493
XCTAssertEqual(leaseContext3.wasIdle, true)
490494
}
491495

496+
func testEventsAfterConnectionIsClosed() {
497+
let elg = EmbeddedEventLoopGroup(loops: 2)
498+
var connections = HTTPConnectionPool.HTTP2Connections(generator: .init())
499+
let el1 = elg.next()
500+
501+
let conn1ID = connections.createNewConnection(on: el1)
502+
let conn1: HTTPConnectionPool.Connection = .__testOnly_connection(id: conn1ID, eventLoop: el1)
503+
let (conn1Index, conn1CreatedContext) = connections.newHTTP2ConnectionEstablished(conn1, maxConcurrentStreams: 1)
504+
XCTAssertEqual(conn1CreatedContext.availableStreams, 1)
505+
506+
let (leasedConn1, leasdConnContext1) = connections.leaseStreams(at: conn1Index, count: 1)
507+
XCTAssertEqual(leasedConn1, conn1)
508+
XCTAssertEqual(leasdConnContext1.wasIdle, true)
509+
510+
XCTAssertNil(connections.leaseStream(onRequired: el1), "all streams are in use")
511+
512+
let (_, releaseContext) = connections.releaseStream(conn1ID)
513+
XCTAssertTrue(releaseContext.eventLoop === el1)
514+
XCTAssertEqual(releaseContext.availableStreams, 1)
515+
XCTAssertEqual(releaseContext.connectionID, conn1ID)
516+
XCTAssertEqual(releaseContext.isIdle, true)
517+
518+
// schedule timeout... this should remove the connection from http2Connections
519+
520+
XCTAssertEqual(connections.closeConnectionIfIdle(conn1ID), conn1)
521+
522+
// events race with the complete shutdown
523+
524+
XCTAssertNil(connections.newHTTP2MaxConcurrentStreamsReceived(conn1ID, newMaxStreams: 2))
525+
XCTAssertNil(connections.goAwayReceived(conn1ID))
526+
527+
// finally close event
528+
XCTAssertNil(connections.failConnection(conn1ID))
529+
}
530+
492531
func testLeaseOnPreferredEventLoopWithoutAnyAvailable() {
493532
let elg = EmbeddedEventLoopGroup(loops: 4)
494533
var connections = HTTPConnectionPool.HTTP2Connections(generator: .init())

‎Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests+XCTest.swift

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ extension HTTPConnectionPool_HTTP2StateMachineTests {
4141
("testHTTP2toHTTP1Migration", testHTTP2toHTTP1Migration),
4242
("testConnectionIsImmediatelyCreatedAfterBackoffTimerFires", testConnectionIsImmediatelyCreatedAfterBackoffTimerFires),
4343
("testMaxConcurrentStreamsIsRespected", testMaxConcurrentStreamsIsRespected),
44+
("testEventsAfterConnectionIsClosed", testEventsAfterConnectionIsClosed),
4445
]
4546
}
4647
}

‎Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1088,4 +1088,23 @@ class HTTPConnectionPool_HTTP2StateMachineTests: XCTestCase {
10881088
XCTAssertNotNil(connections.randomParkedConnection())
10891089
XCTAssertEqual(connections.count, 1)
10901090
}
1091+
1092+
func testEventsAfterConnectionIsClosed() {
1093+
let elg = EmbeddedEventLoopGroup(loops: 2)
1094+
guard var (connections, state) = try? MockConnectionPool.http2(elg: elg, maxConcurrentStreams: 100) else {
1095+
return XCTFail("Test setup failed")
1096+
}
1097+
1098+
let connection = connections.randomParkedConnection()!
1099+
XCTAssertNoThrow(try connections.closeConnection(connection))
1100+
1101+
let idleTimeoutAction = state.connectionIdleTimeout(connection.id)
1102+
XCTAssertEqual(idleTimeoutAction.connection, .closeConnection(connection, isShutdown: .no))
1103+
XCTAssertEqual(idleTimeoutAction.request, .none)
1104+
1105+
XCTAssertEqual(state.newHTTP2MaxConcurrentStreamsReceived(connection.id, newMaxStreams: 50), .none)
1106+
XCTAssertEqual(state.http2ConnectionGoAwayReceived(connection.id), .none)
1107+
1108+
XCTAssertEqual(state.http2ConnectionClosed(connection.id), .none)
1109+
}
10911110
}

0 commit comments

Comments
 (0)