Cancel pending TLS handshake task when streams closed in both directions (#1497)

Author: jfallows

runtime/binding-tls/src/main/java/io/aklivity/zilla/runtime/binding/tls/internal/stream/TlsClientFactory.java

@@ -1529,6 +1529,8 @@ private void onNetEnd(
                         doAppReset(traceId);
                         doEncodeCloseOutbound(traceId, budgetId);
                         doNetEnd(traceId);
+
+                        cancelHandshakeTask();
                     }
 
                     decoder = decodeIgnoreAll;
@@ -1565,6 +1567,8 @@ private void onNetAbort(
                 doAppReset(traceId);
 
                 doNetAbort(traceId);
+
+                cancelHandshakeTask();
             }
 
             private void onNetReset(
@@ -1589,6 +1593,8 @@ private void onNetReset(
                 doAppAbort(traceId);
 
                 doNetReset(traceId);
+
+                cancelHandshakeTask();
             }
 
             private void onNetWindow(
@@ -2194,6 +2200,17 @@ private void cleanupEncodeSlot()
                 }
             }
 
+            private void cancelHandshakeTask()
+            {
+                assert TlsState.closed(state);
+
+                if (handshakeTaskFutureId != NO_CANCEL_ID)
+                {
+                    signaler.cancel(handshakeTaskFutureId);
+                    handshakeTaskFutureId = NO_CANCEL_ID;
+                }
+            }
+
             private void cancelHandshakeTimeout()
             {
                 if (handshakeTimeoutFutureId != NO_CANCEL_ID)

runtime/binding-tls/src/main/java/io/aklivity/zilla/runtime/binding/tls/internal/stream/TlsServerFactory.java

@@ -1233,6 +1233,8 @@ private void onNetEnd(
                 {
                     doEncodeCloseOutbound(traceId, budgetId);
                     doNetEnd(traceId);
+
+                    cancelHandshakeTask();
                 }
 
                 decoder = decodeIgnoreAll;
@@ -1268,6 +1270,8 @@ private void onNetAbort(
             stream.ifPresent(s -> s.doAppReset(traceId));
 
             doNetAbort(traceId);
+
+            cancelHandshakeTask();
         }
 
         private void onNetReset(
@@ -1292,6 +1296,8 @@ private void onNetReset(
             stream.ifPresent(s -> s.doAppAbort(traceId));
 
             doNetReset(traceId);
+
+            cancelHandshakeTask();
         }
 
         private void onNetWindow(
@@ -1841,6 +1847,17 @@ private void cleanupEncodeSlot()
             }
         }
 
+        private void cancelHandshakeTask()
+        {
+            assert TlsState.closed(state);
+
+            if (handshakeTaskFutureId != NO_CANCEL_ID)
+            {
+                signaler.cancel(handshakeTaskFutureId);
+                handshakeTaskFutureId = NO_CANCEL_ID;
+            }
+        }
+
         private void cancelHandshakeTimeout()
         {
             if (handshakeTimeoutFutureId != NO_CANCEL_ID)