Merge remote-tracking branch 'origin/topic/awelzel/4440-websocket-pong-timeout'

* origin/topic/awelzel/4440-websocket-pong-timeout:
  btest/cluster/websocket: Update tests for new event signature
  cluster/websocket: Propagate code and reason to websocket_client_lost()
  cluster/websocket: Support configurable ping interval
  IXWebSocket: Bump to improve O^2 rxbuf erase()

(cherry picked from commit 8f60d37ad0)

testing/btest/Baseline.zam/cluster.websocket.listen-idempotent/.stderr

@@ -1,5 +1,5 @@
 ### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
-error in <...>/main.zeek, line 666: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_x))
-error in <...>/main.zeek, line 666: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_wss_port))
-error in <...>/main.zeek, line 666: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_qs))
+error in <...>/main.zeek, line 673: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_x))
+error in <...>/main.zeek, line 673: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_wss_port))
+error in <...>/main.zeek, line 673: Already listening on 127.0.0.1:<port> (Cluster::__listen_websocket(ws_opts_qs))
 received termination signal

testing/btest/Baseline.zam/cluster.websocket.tls-usage-error/.stderr

@@ -1,3 +1,3 @@
 ### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
-error in <...>/main.zeek, line 666: Invalid tls_options: No key_file field (Cluster::__listen_websocket(Cluster::options.0))
-error in <...>/main.zeek, line 666: Invalid tls_options: No cert_file field (Cluster::__listen_websocket(Cluster::options.3))
+error in <...>/main.zeek, line 673: Invalid tls_options: No key_file field (Cluster::__listen_websocket(Cluster::options.0))
+error in <...>/main.zeek, line 673: Invalid tls_options: No cert_file field (Cluster::__listen_websocket(Cluster::options.3))

testing/btest/Baseline/cluster.websocket.cluster-log/..manager.cluster.log.cannonified

@@ -2,6 +2,6 @@
 manager	WebSocket client <nodeid> (127.0.0.1:<port>) subscribed to [/topic/ws/1, /topic/ws/all]
 manager	WebSocket client <nodeid> (127.0.0.1:<port>) subscribed to [/topic/ws/2, /topic/ws/all]
 manager	WebSocket client <nodeid> (127.0.0.1:<port>) subscribed to [/topic/ws/3, /topic/ws/all]
-manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone
-manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone
-manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone
+manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone with code 1000
+manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone with code 1000
+manager	WebSocket client <nodeid> (127.0.0.1:<port>) gone with code 1000

testing/btest/Baseline/cluster.websocket.ping-timeout/..client..stderr

@@ -0,0 +1 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.

testing/btest/Baseline/cluster.websocket.ping-timeout/..client.out

@@ -0,0 +1,3 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+received 1011 (internal error) Ping timeout; then sent 1011 (internal error) Ping timeout
+received 1011 (internal error) Ping timeout; then sent 1011 (internal error) Ping timeout

testing/btest/Baseline/cluster.websocket.ping-timeout/..manager..stderr

@@ -0,0 +1,2 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+received termination signal

testing/btest/Baseline/cluster.websocket.ping-timeout/..manager.cluster.log.cannonified

@@ -0,0 +1,5 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+WebSocket client <nodeid> (127.0.0.1:<port>) subscribed to []
+WebSocket client <nodeid> (127.0.0.1:<port>) subscribed to []
+WebSocket client <nodeid> (127.0.0.1:<port>) gone with code 1011 and reason 'Ping timeout'
+WebSocket client <nodeid> (127.0.0.1:<port>) gone with code 1011 and reason 'Ping timeout'

testing/btest/Baseline/cluster.websocket.ping-timeout/..manager.out

@@ -0,0 +1,5 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+Cluster::websocket_client_added, []
+Cluster::websocket_client_added, []
+Cluster::websocket_client_lost, 1011, Ping timeout
+Cluster::websocket_client_lost, 1011, Ping timeout

testing/btest/cluster/websocket/bad-event-args.zeek

@@ -50,7 +50,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "Cluster::websocket_client_lost";
 	terminate();

testing/btest/cluster/websocket/bad-subscriptions.zeek

@@ -47,7 +47,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", added, subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print "Cluster::websocket_client_lost", lost;

testing/btest/cluster/websocket/bad-url.zeek

@@ -53,7 +53,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "Cluster::websocket_client_lost";
 	terminate();

testing/btest/cluster/websocket/broker/manager-recvs.zeek

@@ -63,7 +63,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print fmt("%s: Cluster::websocket_client_added %s %s", current_time(), added, subscriptions);
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print fmt("%s: Cluster::websocket_client_lost %s", current_time(), lost);

testing/btest/cluster/websocket/broker/manager-sends.zeek

@@ -66,7 +66,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 		event send_ping();
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print fmt("%s: Cluster::websocket_client_lost %s", current_time(), lost);

testing/btest/cluster/websocket/broker/manager-worker-ping-pong.zeek

@@ -94,7 +94,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print fmt("%s: Cluster::websocket_client_added %s %s", current_time(), added, subscriptions);
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print fmt("%s: Cluster::websocket_client_lost %s", current_time(), lost);

testing/btest/cluster/websocket/cluster-log.zeek

@@ -63,7 +63,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", added, subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print "Cluster::websocket_client_lost", lost;

testing/btest/cluster/websocket/one-pipelining.zeek

@@ -53,7 +53,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "Cluster::websocket_client_lost";
 	terminate();

testing/btest/cluster/websocket/one.zeek

@@ -53,7 +53,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "Cluster::websocket_client_lost";
 	terminate();

testing/btest/cluster/websocket/ping-timeout.zeek

@@ -0,0 +1,100 @@
+# @TEST-DOC: Ensure the websocket_client_lost() event contains code and reason. This starts two WebSocket client that aren't replying to PING frames.
+#
+# @TEST-REQUIRES: python3 -c 'import websockets.sync'
+#
+# @TEST-PORT: WEBSOCKET_PORT
+#
+# @TEST-EXEC: cp $FILES/ws/wstest.py .
+# @TEST-EXEC: zeek -b --parse-only manager.zeek
+# @TEST-EXEC: python3 -m py_compile client.py
+#
+# @TEST-EXEC: btest-bg-run manager "ZEEKPATH=$ZEEKPATH:.. && zeek -b ../manager.zeek >out"
+# @TEST-EXEC: btest-bg-run client "python3 ../client.py >out"
+#
+# @TEST-EXEC: btest-bg-wait 30
+# @TEST-EXEC: btest-diff ./manager/out
+# @TEST-EXEC: btest-diff ./manager/.stderr
+# @TEST-EXEC: zeek-cut message <  ./manager/cluster.log | sed -r "s/client '.+' /client <nodeid> /g" | sed -r "s/:[0-9]+/:<port>/g" > ./manager/cluster.log.cannonified
+# @TEST-EXEC: btest-diff ./manager/cluster.log.cannonified
+# @TEST-EXEC: btest-diff ./client/out
+# @TEST-EXEC: btest-diff ./client/.stderr
+
+# @TEST-START-FILE manager.zeek
+redef exit_only_after_terminate = T;
+
+global lost = 0;
+
+event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions: string_vec)
+	{
+	print "Cluster::websocket_client_added", subscriptions;
+	}
+
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
+	{
+	++lost;
+	print "Cluster::websocket_client_lost", code, reason;
+	if ( lost == 2 )
+		terminate();
+	}
+
+event zeek_init()
+	{
+	Cluster::listen_websocket([$listen_host="127.0.0.1", $listen_port=to_port(getenv("WEBSOCKET_PORT")), $ping_interval=1sec]);
+	Cluster::subscribe("/test/pings/");
+	}
+# @TEST-END-FILE
+
+
+# @TEST-START-FILE client.py
+import json
+import functools
+
+import wstest
+
+from websockets.sync.client import connect
+from websockets.sync.client import ClientConnection
+from websockets.frames import OP_PONG
+
+class MyClientConnection(ClientConnection):
+    """
+    Custom Client class patching the protocol.send_frame() function
+    to discard any PONG frames. The websocket library responds
+    automatically to these in a thread and can't easily turn this off,
+    but we want to test Zeek behavior when a client fails to respond
+    with PONG frames quickly enough.
+    """
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.__orig_send_frame = self.protocol.send_frame
+        def __my_send_frame(_self, frame):
+            if frame.opcode != OP_PONG:
+                self.__orig_send_frame(frame)
+
+        self.protocol.send_frame = functools.partial(__my_send_frame, self.protocol)
+
+def run(ws_url):
+    with (
+        connect(ws_url, create_connection=MyClientConnection) as c1,
+        connect(ws_url, create_connection=MyClientConnection) as c2,
+    ):
+        c1.send(json.dumps([]))
+        ack1 = json.loads(c1.recv())
+        assert ack1["type"] == "ack", repr(ack1)
+
+        c2.send(json.dumps([]))
+        ack2 = json.loads(c2.recv())
+        assert ack2["type"] == "ack", repr(ack2)
+
+        try:
+            c1.recv()
+        except Exception as e:
+            print(e)
+        try:
+            c2.recv()
+        except Exception as e:
+            print(e)
+
+
+if __name__ == "__main__":
+    wstest.main(run, wstest.WS4_URL_V1)
+# @TEST-END-FILE

testing/btest/cluster/websocket/terminate-while-queuing.zeek

@@ -46,7 +46,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "B Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "E Cluster::websocket_client_lost";
 	}

testing/btest/cluster/websocket/three.zeek

@@ -52,7 +52,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "B Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print "D Cluster::websocket_client_lost", lost;

testing/btest/cluster/websocket/tls.zeek

@@ -67,7 +67,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print "Cluster::websocket_client_added", subscriptions;
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	print "Cluster::websocket_client_lost";
 	terminate();

testing/btest/cluster/websocket/two-pipelining.zeek

@@ -84,7 +84,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	drain_if_ready();
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print "C Cluster::websocket_client_lost";

testing/btest/cluster/websocket/zeromq/manager-worker-ping-pong.zeek

@@ -93,7 +93,7 @@ event Cluster::websocket_client_added(info: Cluster::EndpointInfo, subscriptions
 	print fmt("%s: Cluster::websocket_client_added %s %s", current_time(), added, subscriptions);
 	}
 
-event Cluster::websocket_client_lost(info: Cluster::EndpointInfo)
+event Cluster::websocket_client_lost(info: Cluster::EndpointInfo, code: count, reason: string)
 	{
 	++lost;
 	print fmt("%s: Cluster::websocket_client_lost %s", current_time(), lost);