Tolerate idling from the active state

Motivation:

In certain conditions it's possible for a connection to idle when in the
'active' state resulting in a precondition failure.

The active state is not user facing and represents the state where we
have an active connection to a remote peer but have not yet seen the
initial settings frame. It's not usually possible for users to get a
connection in this state since we normally only vend connections in the
'ready' state (where we have seen the initial settings frame). However,
in the 'fastFailure' mode this constraint is lifted.

If keepalive is configured, the connection is in the active state, and
the client starts an RPC then it is possible for the keepalive timeout
to fire, idling the connection before it reaches the ready state,
resulting in a precondition failure.

Modifications:

- Allow the connection manager to idle from the active state

Result:

- Idling from the active state is tolerated
- Resolves grpc#949

Sources/GRPC/ConnectionManager.swift

@@ -514,10 +514,16 @@ internal class ConnectionManager {
     ])
 
     switch self.state {
+    case let .active(state):
+      // This state is reachable if the keepalive timer fires before we reach the ready state.
+      self.state = .idle(IdleState(configuration: state.configuration))
+      state.readyChannelPromise
+        .fail(GRPCStatus(code: .unavailable, message: "Idled before reaching ready state"))
+
     case let .ready(state):
       self.state = .idle(IdleState(configuration: state.configuration))
 
-    case .idle, .connecting, .transientFailure, .active, .shutdown:
+    case .idle, .connecting, .transientFailure, .shutdown:
       self.invalidState()
     }
   }

Tests/GRPCTests/GRPCKeepaliveTests.swift

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2020, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import EchoImplementation
+import EchoModel
+@testable import GRPC
+import NIO
+import XCTest
+
+class GRPCClientKeepaliveTests: GRPCTestCase {
+  func testKeepaliveTimeoutFiresBeforeConnectionIsReady() throws {
+    // This test relates to https://github.com/grpc/grpc-swift/issues/949
+    //
+    // When a stream is created, a ping may be sent on the connection. If a ping is sent we then
+    // schedule a task for some time in the future to close the connection (if we don't receive the
+    // ping ack in the meantime).
+    //
+    // The task to close actually fires an event which is picked up by the idle handler; this will
+    // tell the connection manager to idle the connection. However, the connection manager only
+    // tolerates being idled from the ready state. Since we protect from idling multiple times in
+    // the handler we must be in a state where we have connection but are not yet ready (i.e.
+    // channel active has fired but we have not seen the initial settings frame). To be in this
+    // state the user must be using the 'fastFailure' call start behaviour (if this is not the case
+    // then no channel will be vended until we reach the ready state, so it would not be possible
+    // to create the stream).
+    let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    defer {
+      XCTAssertNoThrow(try group.syncShutdownGracefully())
+    }
+
+    // Setup a server.
+    let server = try Server.insecure(group: group)
+      .withServiceProviders([EchoProvider()])
+      .withLogger(self.serverLogger)
+      .bind(host: "localhost", port: 0)
+      .wait()
+    defer {
+      XCTAssertNoThrow(try server.close().wait())
+    }
+
+    // Setup a connection. We'll add a handler to drop all reads, this is somewhat equivalent to
+    // simulating bad network conditions and allows us to setup a connection and have our keepalive
+    // timeout expire.
+    let connection = ClientConnection.insecure(group: group)
+      .withBackgroundActivityLogger(self.clientLogger)
+      // See above comments for why we need this.
+      .withCallStartBehavior(.fastFailure)
+      .withKeepalive(.init(interval: .seconds(1), timeout: .milliseconds(100)))
+      .withDebugChannelInitializer { channel in
+        channel.pipeline.addHandler(ReadDroppingHandler(), position: .first)
+      }
+      .connect(host: "localhost", port: server.channel.localAddress!.port!)
+    defer {
+      XCTAssertNoThrow(try connection.close().wait())
+    }
+
+    let client = Echo_EchoClient(channel: connection)
+    let get = client.get(.with { $0.text = "Hello" })
+    XCTAssertThrowsError(try get.response.wait())
+    XCTAssertEqual(try get.status.map { $0.code }.wait(), .unavailable)
+  }
+
+  class ReadDroppingHandler: ChannelDuplexHandler {
+    typealias InboundIn = Any
+    typealias OutboundIn = Any
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {}
+  }
+}

Tests/GRPCTests/XCTestManifests.swift

@@ -382,6 +382,15 @@ extension FunctionalTestsMutualAuthenticationNIOTS {
     ]
 }
 
+extension GRPCClientKeepaliveTests {
+    // DO NOT MODIFY: This is autogenerated, use:
+    //   `swift test --generate-linuxmain`
+    // to regenerate.
+    static let __allTests__GRPCClientKeepaliveTests = [
+        ("testKeepaliveTimeoutFiresBeforeConnectionIsReady", testKeepaliveTimeoutFiresBeforeConnectionIsReady),
+    ]
+}
+
 extension GRPCClientStateMachineTests {
     // DO NOT MODIFY: This is autogenerated, use:
     //   `swift test --generate-linuxmain`
@@ -962,6 +971,7 @@ public func __allTests() -> [XCTestCaseEntry] {
         testCase(FunctionalTestsInsecureTransportNIOTS.__allTests__FunctionalTestsInsecureTransportNIOTS),
         testCase(FunctionalTestsMutualAuthentication.__allTests__FunctionalTestsMutualAuthentication),
         testCase(FunctionalTestsMutualAuthenticationNIOTS.__allTests__FunctionalTestsMutualAuthenticationNIOTS),
+        testCase(GRPCClientKeepaliveTests.__allTests__GRPCClientKeepaliveTests),
         testCase(GRPCClientStateMachineTests.__allTests__GRPCClientStateMachineTests),
         testCase(GRPCCustomPayloadTests.__allTests__GRPCCustomPayloadTests),
         testCase(GRPCIdleTests.__allTests__GRPCIdleTests),