Deliver VCONN_CLOSE for parked TLS hooks; fix SNI queue accounting

plugins/experimental/rate_limit/limiter.h

@@ -327,6 +327,30 @@ template <class T> class RateLimiter
     return item;
   }
 
+  // Remove a still-queued element (e.g. a connection that closed before it was resumed).
+  // Returns true if it was found in the queue, so the caller can tell a queued element
+  // (which never reserved a slot) from one that was already resumed.
+  //
+  // Linear in the queue depth, and only reached when an element closes while queued, which
+  // requires the limiter to be at its limit. The depth is bounded by the configured queue size;
+  // note that a "queue" without a "size" leaves _max_queue at UINT32_MAX, in which case the only
+  // bound is proxy.config.net.connections_throttle.
+  bool
+  remove(const T &elem)
+  {
+    std::lock_guard<std::mutex> lock(_queue_lock);
+
+    for (auto it = _queue.begin(); it != _queue.end(); ++it) {
+      if (std::get<0>(*it) == elem) {
+        _queue.erase(it);
+        --_size;
+        return true;
+      }
+    }
+
+    return false;
+  }
+
   void
   incrementMetric(uint metric)
   {

plugins/experimental/rate_limit/sni_limiter.cc

@@ -171,7 +171,12 @@ sni_limit_cont(TSCont contp, TSEvent event, void *edata)
 
     if (limiter) {
       TSUserArgSet(vc, gVCIdx, nullptr);
-      limiter->free();
+      // A connection that is still queued never reserved a slot, so only release one if it
+      // is not in the queue (either it reserved at CLIENT_HELLO or the sweep resumed it into
+      // a reserved slot). Dropping it from the queue also avoids a stale entry.
+      if (!limiter->remove(vc)) {
+        limiter->free();
+      }
       limiter->selector()->release(); // Release the selector, such that it can be deleted later
     }
     TSVConnReenable(vc);

plugins/experimental/rate_limit/sni_selector.cc

@@ -219,9 +219,17 @@ sni_queue_cont(TSCont cont, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_
     QueueTime now     = std::chrono::system_clock::now(); // Only do this once per limiter
 
     if (owner) { // Don't operate on the aliases
-      // Try to enable some queued VCs (if any) if there are slots available
+      // Try to enable some queued VCs (if any) if there are slots available. Reserving before
+      // dequeuing means a resumed VC owns the slot it was granted, so its VCONN_CLOSE releases
+      // exactly that slot.
       while (limiter->size() > 0 && limiter->reserve() == ReserveStatus::RESERVED) {
-        auto [vc, contp, start_time]    = limiter->pop();
+        auto [vc, contp, start_time] = limiter->pop();
+
+        if (nullptr == vc) { // A concurrent close emptied the queue; give the slot back
+          limiter->free();
+          break;
+        }
+
         std::chrono::milliseconds delay = std::chrono::duration_cast<std::chrono::milliseconds>(now - start_time);
 
         (void)contp; // Ugly, but silences some compilers.
@@ -236,11 +244,18 @@ sni_queue_cont(TSCont cont, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_
 
         while (limiter->size() > 0 && limiter->hasOldEntity(now)) {
           // The oldest object on the queue is too old on the queue, so "kill" it.
-          auto [vc, contp, start_time]  = limiter->pop();
+          auto [vc, contp, start_time] = limiter->pop();
+
+          if (nullptr == vc) { // A concurrent close emptied the queue
+            break;
+          }
+
           std::chrono::milliseconds age = std::chrono::duration_cast<std::chrono::milliseconds>(now - start_time);
 
           (void)contp;
           Dbg(dbg_ctl, "Queued VC is too old (%ldms), erroring out", static_cast<long>(age.count()));
+          // This VC never reserved a slot; detach it (clear the arg and release the selector
+          // lease) so its VCONN_CLOSE does not release a slot it never held.
           TSUserArgSet(vc, gVCIdx, nullptr);
           limiter->selector()->release();
           TSVConnReenableEx(vc, TS_EVENT_ERROR);

src/iocore/net/TLSEventSupport.cc

@@ -153,10 +153,20 @@ TLSEventSupport::callHooks(TSEvent eventId)
   Dbg(dbg_ctl_ssl, "sslHandshakeHookState=%s eventID=%d", get_ssl_handshake_hook_state_name(this->sslHandshakeHookState), eventId);
 
   // Move state if it is appropriate
-  if (eventId == TS_EVENT_VCONN_CLOSE) {
+  if (eventId == TS_EVENT_VCONN_CLOSE || eventId == TS_EVENT_VCONN_OUTBOUND_CLOSE) {
     // Regardless of state, if the connection is closing, then transition to
     // the DONE state. This will trigger us to call the appropriate cleanup
     // routines.
+    //
+    // A connection can close while it is parked in a handshake hook, waiting for a plugin to
+    // reenable it. curHook then still points into that handshake hook's list. Each hook id owns
+    // a separate list, so advancing curHook below would walk the handshake list rather than the
+    // close list: the close event is dropped when the handshake list is exhausted, or delivered
+    // to the wrong plugin when it is not. Restart from the head of the close list unless we are
+    // already iterating it.
+    if (this->sslHandshakeHookState != SSLHandshakeHookState::HANDSHAKE_HOOKS_DONE) {
+      this->curHook = nullptr;
+    }
     this->sslHandshakeHookState = SSLHandshakeHookState::HANDSHAKE_HOOKS_DONE;
   } else {
     switch (this->sslHandshakeHookState) {

tests/gold_tests/pluginTest/rate_limit/rate_limit_sni_expiry.test.py

@@ -0,0 +1,87 @@
+'''
+Regression test for the max_age expiry branch of the rate_limit SNI queue accounting. A
+queued connection never reserves a slot, so when the sweep expires it the plugin must
+detach it rather than release a slot it never held; otherwise the expiry underflows the
+active-slot counter and the next reserve() trips a release assertion, aborting the server.
+ATS must survive the expiry.
+'''
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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 os
+
+Test.Summary = __doc__
+
+Test.SkipUnless(Condition.PluginExists('rate_limit.so'))
+
+
+class RateLimitSniExpiryTest:
+    """Age a queued connection out via max_age and assert the active-slot counter stays balanced."""
+
+    def __init__(self) -> None:
+        tr = Test.AddTestRun('rate_limit SNI queue max_age expiry')
+        self._configure_trafficserver()
+        self._configure_client(tr)
+
+    def _configure_trafficserver(self) -> None:
+        ts = Test.MakeATSProcess('ts', enable_tls=True, enable_cache=False)
+        self._ts = ts
+        ts.addDefaultSSLFiles()
+        for line in ['ssl_multicert:', '  - dest_ip: "*"', '    ssl_cert_name: server.pem', '    ssl_key_name: server.key']:
+            ts.Disk.ssl_multicert_yaml.AddLine(line)
+
+        # One concurrent handshake for this SNI, a one-deep queue, and a 1s max age so the
+        # sweep expires the queued connection. Named .config (not .yaml) so autest treats it
+        # as a plain config file; the plugin parses it as YAML regardless.
+        ts.Disk.MakeConfigFile('rate_limit.config').AddLines(
+            [
+                'selector:',
+                '  - sni: rate.limited.com',
+                '    limit: 1',
+                '    queue:',
+                '      size: 1',
+                '      max_age: 1',
+            ])
+        ts.Disk.plugin_config.AddLine(f'rate_limit.so {ts.Variables.CONFIGDIR}/rate_limit.config')
+
+        # Disable the freelist / ProxyAllocator so allocation behavior is not a confound; the
+        # abort under test is a release-assertion, and this keeps the run representative of CI.
+        ts.Command += ' -f -F'
+
+        ts.Disk.records_config.update(
+            {
+                'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir,
+                'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir,
+                'proxy.config.diags.debug.enabled': 1,
+                'proxy.config.diags.debug.tags': 'rate_limit',
+            })
+
+        # The expiry branch is actually reached...
+        ts.Disk.traffic_out.Content = Testers.ContainsExpression('too old', 'a queued connection was expired')
+        # ...and expiring it does not underflow the active-slot counter into the release assertion.
+        ts.Disk.traffic_out.Content += Testers.ExcludesExpression(
+            '_active <= _limit|received signal', 'expiring a queued connection must not underflow and abort ATS')
+
+    def _configure_client(self, tr: 'TestRun') -> None:
+        ts = self._ts
+        client = os.path.join(Test.TestDirectory, 'rate_limit_sni_expiry_client.sh')
+        tr.Processes.Default.Command = f'bash {client} 127.0.0.1 {ts.Variables.ssl_port} rate.limited.com'
+        tr.Processes.Default.ReturnCode = 0
+        tr.Processes.Default.StartBefore(ts)
+        tr.Processes.Default.Streams.stdout = Testers.ContainsExpression('rate_limit-expiry-done', 'the client ran to completion')
+
+
+RateLimitSniExpiryTest()

tests/gold_tests/pluginTest/rate_limit/rate_limit_sni_expiry_client.sh

@@ -0,0 +1,77 @@
+#!/usr/bin/env bash
+#
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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.
+#
+# Exercise the max_age EXPIRY branch of the rate_limit SNI queue accounting. A queued
+# connection never reserves a slot; when the sweep expires it (max_age), it must be
+# detached so its close does not release a slot it never held. Otherwise the expiry is an
+# unmatched decrement of the active-slot counter, and -- combined with the holder's own
+# close -- the counter wraps below zero and the limiter's release assertion aborts ATS.
+#
+#   1. holder completes its handshake and holds the single slot (counter = 1);
+#   2. one connection enqueues (slot full) and stays parked -- it is NOT disconnected, so
+#      only the sweep's max_age expiry removes it;
+#   3. after max_age the sweep errors it out -> (unfixed) unmatched decrement -> counter 1->0;
+#   4. the holder is closed; its matched decrement lands on the understated counter -> wrap;
+#   5. a probe connection's reserve() observes the wrapped counter and the assertion aborts.
+#
+# args: host port sni
+set -u
+host="$1"
+port="$2"
+sni="$3"
+
+OSSL="openssl s_client -connect ${host}:${port} -servername ${sni} -quiet -no_ign_eof"
+
+# Run a command in the background and terminate it after a deadline. coreutils "timeout" is not
+# available everywhere (notably macOS), so do it with sleep and kill.
+run_for() {
+  deadline="$1"
+  shift
+  "$@" &
+  target=$!
+  (
+    sleep "${deadline}"
+    kill -TERM "${target}" 2>/dev/null
+  ) &
+}
+
+# 1. Holder: hold the single slot. Its stdin is a FIFO on fd 3 so we end it in step 4.
+fifo_dir="$(mktemp -d "${TMPDIR:-/tmp}/rl_holder.XXXXXX")"
+fifo="${fifo_dir}/fifo"
+mkfifo "$fifo"
+${OSSL} <"$fifo" >/dev/null 2>&1 &
+exec 3<>"$fifo"
+rm -rf "$fifo_dir"
+sleep 3 # let the holder reserve the one slot
+
+# 2. One queued connection: enqueues and stays parked at the ClientHello hook (not killed),
+#    so the sweep's max_age expiry -- not a disconnect or a resume -- is what removes it.
+${OSSL} </dev/null >/dev/null 2>&1 &
+queued=$!
+sleep 3 # > max_age (1s) + sweeps: the expiry path errors the queued connection out
+
+# 4. End the holder: its matched decrement lands on the (unfixed) understated counter.
+exec 3>&-
+sleep 2
+
+# 5. Probe: its reserve() reads the counter; if it wrapped, the release assertion aborts.
+run_for 2 sh -c "${OSSL} </dev/null >/dev/null 2>&1"
+kill "${queued}" 2>/dev/null || true
+sleep 1
+
+echo "rate_limit-expiry-done"

tests/gold_tests/pluginTest/rate_limit/rate_limit_sni_queue.test.py

@@ -0,0 +1,89 @@
+'''
+Regression test for a queue-accounting balance bug in the rate_limit SNI limiter: a
+queued connection never reserves a slot, but its VCONN_CLOSE unconditionally releases
+one, so a queued connection that closes underflows the active-slot counter and the next
+reserve() trips a release assertion, aborting the server. ATS must survive the queue
+churn.
+'''
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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 os
+
+Test.Summary = __doc__
+
+Test.SkipUnless(Condition.PluginExists('rate_limit.so'))
+
+
+class RateLimitSniQueueTest:
+    """Churn the rate_limit SNI queue and assert the active-slot counter never underflows."""
+
+    def __init__(self) -> None:
+        tr = Test.AddTestRun('rate_limit SNI queue accounting')
+        self._configure_trafficserver()
+        self._configure_client(tr)
+
+    def _configure_trafficserver(self) -> None:
+        ts = Test.MakeATSProcess('ts', enable_tls=True, enable_cache=False)
+        self._ts = ts
+        ts.addDefaultSSLFiles()
+        for line in ['ssl_multicert:', '  - dest_ip: "*"', '    ssl_cert_name: server.pem', '    ssl_key_name: server.key']:
+            ts.Disk.ssl_multicert_yaml.AddLine(line)
+
+        # One concurrent handshake for this SNI and a queue that admits exactly one more.
+        # No rate and no max_age -- the sweep's resume path alone drives the scenario, with
+        # no rate-bucket or expiry timing to confound it. Named .config (not .yaml) so autest
+        # treats it as a plain config file; the plugin parses it as YAML regardless.
+        ts.Disk.MakeConfigFile('rate_limit.config').AddLines(
+            [
+                'selector:',
+                '  - sni: rate.limited.com',
+                '    limit: 1',
+                '    queue:',
+                '      size: 1',
+            ])
+        ts.Disk.plugin_config.AddLine(f'rate_limit.so {ts.Variables.CONFIGDIR}/rate_limit.config')
+
+        # Disable the freelist / ProxyAllocator so freed objects are really released rather
+        # than recycled, keeping allocation reuse from masking a stale access.
+        ts.Command += ' -f -F'
+
+        ts.Disk.records_config.update(
+            {
+                'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir,
+                'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir,
+                'proxy.config.diags.debug.enabled': 1,
+                'proxy.config.diags.debug.tags': 'rate_limit',
+            })
+
+        # The queue path is reached...
+        ts.Disk.traffic_out.Content = Testers.ContainsExpression('Queueing the VC', 'a connection was queued')
+        # ...and the active-slot counter never underflows into the release assertion. Match
+        # both the specific assertion (pins the failure to this bug) and the generic abort.
+        ts.Disk.traffic_out.Content += Testers.ExcludesExpression(
+            '_active <= _limit|received signal', 'the active-slot counter must not underflow and abort ATS')
+
+    def _configure_client(self, tr: 'TestRun') -> None:
+        ts = self._ts
+        client = os.path.join(Test.TestDirectory, 'rate_limit_sni_queue_client.sh')
+        tr.Processes.Default.Command = f'bash {client} 127.0.0.1 {ts.Variables.ssl_port} rate.limited.com'
+        tr.Processes.Default.ReturnCode = 0
+        tr.Processes.Default.StartBefore(ts)
+        tr.Processes.Default.Streams.stdout = Testers.ContainsExpression(
+            'rate_limit-queue-crash-done', 'the client ran to completion')
+
+
+RateLimitSniQueueTest()