Repository: relayrides/pushy (java)

Issue:
InternalServerError from APNs
We are using Pushy in our production environment and it works great! Besides we have a staging environment where tests are performed prior to launching new versions. Staging environment is not always used and most of the time it is in idle state.
Today when we started to use staging, we got `InternalServerError` errors (stack trace given below). I guess it is related to long idle state of pushy, not sure though. We didn't have any choice but to restart the app.

In which cases APNs may return such an error? Should pushy be closing and reconnecting in such cases? Or is it a duty of our side to reconnect upon such an error?

Using pushy v0.9, java 1.8.0_102 (Oracle)

    2017-01-06 05:23:27,715  WARN [nioEventLoopGroup-2-4] c.r.p.a.ApnsClient APNs server reported an internal error when sending ... .
    2017-01-06 05:23:27,715 ERROR [nioEventLoopGroup-2-4] o.g.g.a.ApnsResponseListener Failed to send notification id=117_963041067
    java.util.concurrent.ExecutionException: com.relayrides.pushy.apns.ApnsServerException: {"reason":"InternalServerError"}
	    at io.netty.util.concurrent.AbstractFuture.get(AbstractFuture.java:54)
	    at org.grouvi.gpns.apns.ApnsResponseListener.operationComplete(ApnsResponseListener.java:59)
	    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:514)
	    at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:507)
	    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:486)
	    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:427)
	    at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:129)
	    at com.relayrides.pushy.apns.ApnsClient.handleServerError(ApnsClient.java:965)
	    at com.relayrides.pushy.apns.ApnsClientHandler$ApnsClientHandlerFrameAdapter.onDataRead(ApnsClientHandler.java:173)
	    at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder$FrameReadListener.onDataRead(DefaultHttp2ConnectionDecoder.java:229)
	    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readDataFrame(DefaultHttp2FrameReader.java:411)
	    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.processPayloadState(DefaultHttp2FrameReader.java:245)
	    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:155)
	    at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder.decodeFrame(DefaultHttp2ConnectionDecoder.java:118)
	    at io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:335)
	    at io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:395)
	    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
	    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:373)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:351)
	    at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:266)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:373)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:351)
	    at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1069)
	    at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:902)
	    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
	    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:373)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:351)
	    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:373)
	    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
	    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:129)
	    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:651)
	    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:574)
	    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:488)
	    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:450)
	    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:873)
	    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
	    at java.lang.Thread.run(Thread.java:745)
    Caused by: com.relayrides.pushy.apns.ApnsServerException: {"reason":"InternalServerError"}
	    ... 37 common frames omitted
    


Pull request description:
Close connections after receiving an `InternalServerError`
For a long time, we've been working around #408, an upstream issue where connections could go into a permanently-unhealthy state by responding to all notifications with an `InternalServerError` after replacing an expired authentication token. We hesitated to build a fix into Pushy itself for several reasons:

1. We had reason to believe that the issue would receive a timely upstream fix, and so it was best not to commit too hard to a client-side fix.
2. It wasn't obvious that the same solution would work for everybody.
3. Users had a viable workaround in that they could close and re-open connections if things got stuck in an unhealthy state.

Now that we've moved to a connection-pooling model, users no longer have direct control over connections. We've also gone a looooooong time without an upstream fix, and the new connection model allows for easy behind-the-scenes connection replacement. This means that we have a more acute need for a client-side fix and fewer reasons to hold off.

Long story short: this pull request introduces behavior where clients will quietly replace unhealthy connections behind the scenes, which fixes #408.

Patch:
diff --git a/pushy/src/main/java/com/turo/pushy/apns/ApnsClientHandler.java b/pushy/src/main/java/com/turo/pushy/apns/ApnsClientHandler.java
index 5ab60249..003b4813 100644
--- a/pushy/src/main/java/com/turo/pushy/apns/ApnsClientHandler.java
+++ b/pushy/src/main/java/com/turo/pushy/apns/ApnsClientHandler.java
@@ -75,6 +75,8 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
     private static final IOException STREAM_CLOSED_BEFORE_REPLY_EXCEPTION =
             new IOException("Stream closed before a reply was received");
 
+    private static final ApnsServerException APNS_SERVER_EXCEPTION = new ApnsServerException();
+
     private static final Gson GSON = new GsonBuilder()
             .registerTypeAdapter(Date.class, new DateAsTimeSinceEpochTypeAdapter(TimeUnit.MILLISECONDS))
             .create();
@@ -287,13 +289,7 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
 
         if (endOfStream) {
             final Http2Stream stream = this.connection().stream(streamId);
-
-            final Http2Headers headers = stream.getProperty(this.responseHeadersPropertyKey);
-            final ApnsPushNotification pushNotification = stream.getProperty(this.pushNotificationPropertyKey);
-
-            final ErrorResponse errorResponse = GSON.fromJson(data.toString(StandardCharsets.UTF_8), ErrorResponse.class);
-
-            this.handleErrorResponse(context, streamId, headers, pushNotification, errorResponse);
+            this.handleEndOfStream(context, this.connection().stream(streamId), (Http2Headers) stream.getProperty(this.responseHeadersPropertyKey), data);
         } else {
             log.error("Gateway sent a DATA frame that was not the end of a stream.");
         }
@@ -301,21 +297,6 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
         return bytesProcessed;
     }
 
-    protected void handleErrorResponse(final ChannelHandlerContext context, final int streamId, final Http2Headers headers, final ApnsPushNotification pushNotification, final ErrorResponse errorResponse) {
-        final Promise<PushNotificationResponse<ApnsPushNotification>> responsePromise =
-                this.connection().stream(streamId).getProperty(this.responsePromisePropertyKey);
-
-        final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
-
-        if (HttpResponseStatus.INTERNAL_SERVER_ERROR.equals(status)) {
-            log.warn("APNs server reported an internal error when sending {}.", pushNotification);
-            responsePromise.tryFailure(new ApnsServerException(GSON.toJson(errorResponse)));
-        } else {
-            responsePromise.trySuccess(new SimplePushNotificationResponse<>(pushNotification,
-                    HttpResponseStatus.OK.equals(status), errorResponse.getReason(), errorResponse.getTimestamp()));
-        }
-    }
-
     @Override
     public void onHeadersRead(final ChannelHandlerContext context, final int streamId, final Http2Headers headers, final int streamDependency, final short weight, final boolean exclusive, final int padding, final boolean endOfStream) throws Http2Exception {
         this.onHeadersRead(context, streamId, headers, padding, endOfStream);
@@ -327,25 +308,50 @@ class ApnsClientHandler extends Http2ConnectionHandler implements Http2FrameList
         final Http2Stream stream = this.connection().stream(streamId);
 
         if (endOfStream) {
-            final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
-            final boolean success = HttpResponseStatus.OK.equals(status);
+            this.handleEndOfStream(context, stream, headers, null);
+        } else {
+            stream.setProperty(this.responseHeadersPropertyKey, headers);
+        }
+    }
 
-            if (!success) {
-                log.warn("Gateway sent an end-of-stream HEADERS frame for an unsuccessful notification.");
-            }
+    private void handleEndOfStream(final ChannelHandlerContext context, final Http2Stream stream, final Http2Headers headers, final ByteBuf data) {
+
+        final ApnsPushNotification pushNotification = stream.getProperty(this.pushNotificationPropertyKey);
+        final Promise<PushNotificationResponse<ApnsPushNotification>> responsePromise = stream.getProperty(this.responsePromisePropertyKey);
 
-            final ApnsPushNotification pushNotification = stream.getProperty(this.pushNotificationPropertyKey);
-            final Promise<PushNotificationResponse<ApnsPushNotification>> responsePromise = stream.getProperty(this.responsePromisePropertyKey);
+        final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
 
+        if (HttpResponseStatus.OK.equals(status)) {
+            responsePromise.trySuccess(
+                    new SimplePushNotificationResponse<>(pushNotification, true, null, null));
+        } else {
             if (HttpResponseStatus.INTERNAL_SERVER_ERROR.equals(status)) {
                 log.warn("APNs server reported an internal error when sending {}.", pushNotification);
-                responsePromise.tryFailure(new ApnsServerException());
+                responsePromise.tryFailure(APNS_SERVER_EXCEPTION);
+                context.channel().close();
             } else {
-                responsePromise.trySuccess(
-                        new SimplePushNotificationResponse<>(pushNotification, success, null, null));
+                if (data != null) {
+                    final ErrorResponse errorResponse = GSON.fromJson(data.toString(StandardCharsets.UTF_8), ErrorResponse.class);
+                    this.handleErrorResponse(context, stream.id(), headers, pushNotification, errorResponse);
+                } else {
+                    log.warn("Gateway sent an end-of-stream HEADERS frame for an unsuccessful notification.");
+                }
             }
+        }
+    }
+
+    protected void handleErrorResponse(final ChannelHandlerContext context, final int streamId, final Http2Headers headers, final ApnsPushNotification pushNotification, final ErrorResponse errorResponse) {
+        final Promise<PushNotificationResponse<ApnsPushNotification>> responsePromise =
+                this.connection().stream(streamId).getProperty(this.responsePromisePropertyKey);
+
+        final HttpResponseStatus status = HttpResponseStatus.parseLine(headers.status());
+
+        if (HttpResponseStatus.INTERNAL_SERVER_ERROR.equals(status)) {
+            log.warn("APNs server reported an internal error when sending {}.", pushNotification);
+            responsePromise.tryFailure(new ApnsServerException(GSON.toJson(errorResponse)));
         } else {
-            stream.setProperty(this.responseHeadersPropertyKey, headers);
+            responsePromise.trySuccess(new SimplePushNotificationResponse<>(pushNotification,
+                    HttpResponseStatus.OK.equals(status), errorResponse.getReason(), errorResponse.getTimestamp()));
         }
     }
 


Test patch:
diff --git a/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java b/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
index 3605ca01..7ed241dc 100644
--- a/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
+++ b/pushy/src/test/java/com/turo/pushy/apns/ApnsClientTest.java
@@ -511,11 +511,14 @@ public class ApnsClientTest {
         try {
             terribleTerribleServer.registerVerificationKey(this.verificationKey, DEFAULT_TOPIC);
 
+            final TestMetricsListener metricsListener = new TestMetricsListener();
+
             final ApnsClient unfortunateClient = new ApnsClientBuilder()
                     .setApnsServer(HOST, PORT)
                     .setSigningKey(this.signingKey)
                     .setTrustedServerCertificateChain(CA_CERTIFICATE)
                     .setEventLoopGroup(EVENT_LOOP_GROUP)
+                    .setMetricsListener(metricsListener)
                     .build();
 
             try {
@@ -530,6 +533,11 @@ public class ApnsClientTest {
                 assertTrue(future.isDone());
                 assertFalse(future.isSuccess());
                 assertTrue(future.cause() instanceof ApnsServerException);
+
+                unfortunateClient.sendNotification(pushNotification).await();
+
+                assertTrue("Connections should be replaced after receiving an InternalServerError.",
+                        metricsListener.getConnectionsRemoved().get() == 1);
             } finally {
                 unfortunateClient.close().await();
                 Thread.sleep(10);