http: state machine fix

When client sends second request without waiting for response of the
first request http_ts_rx_callback should drop request (pipelining is
not supported) instead of invoking return to state machine which can
lead to erroneous state, e.g. reading random data from server app
fifo.

Added simple http static server url handler for testing to simulate
long running request processing, for now hardcoded delay 5 seconds.

Type: fix

Change-Id: Ied9f7e2e4ee64c982f045c0f7f99a2dc5d7a2108
Signed-off-by: Matus Fabian <matfabia@cisco.com>

5 files changed, 287 insertions, 4 deletions

diff --git a/extras/hs-test/http_test.go b/extras/hs-test/http_test.go
index e20efd6f35c..29bd272fe99 100644
--- a/extras/hs-test/http_test.go
+++ b/extras/hs-test/http_test.go
@@ -5,7 +5,10 @@ import (
 	"fmt"
 	"github.com/onsi/gomega/gmeasure"
 	"io"
+	"net"
 	"net/http"
+	"net/http/httptrace"
+	"os"
 	"time"
 
 	. "fd.io/hs-test/infra"
@@ -13,7 +16,7 @@ import (
 
 func init() {
 	RegisterVethTests(HttpCliTest, HttpCliConnectErrorTest)
-	RegisterNoTopoTests(HeaderServerTest,
+	RegisterNoTopoTests(HeaderServerTest, HttpPersistentConnectionTest, HttpPipeliningTest,
 		HttpStaticMovedTest, HttpStaticNotFoundTest, HttpCliMethodNotAllowedTest,
 		HttpCliBadRequestTest, HttpStaticBuildInUrlGetIfStatsTest, HttpStaticBuildInUrlPostIfStatsTest,
 		HttpInvalidRequestLineTest, HttpMethodNotImplementedTest, HttpInvalidHeadersTest,
@@ -56,6 +59,98 @@ func HttpTpsTest(s *NoTopoSuite) {
 	s.RunBenchmark("HTTP tps 10M", 10, 0, httpDownloadBenchmark, url)
 }
 
+func HttpPersistentConnectionTest(s *NoTopoSuite) {
+	// testing url handler app do not support multi-thread
+	s.SkipIfMultiWorker()
+	vpp := s.GetContainerByName("vpp").VppInstance
+	serverAddress := s.GetInterfaceByName(TapInterfaceName).Peer.Ip4AddressString()
+	s.Log(vpp.Vppctl("http static server uri tcp://" + serverAddress + "/80 url-handlers"))
+	s.Log(vpp.Vppctl("test-url-handler enable"))
+
+	transport := http.DefaultTransport
+	transport.(*http.Transport).Proxy = nil
+	transport.(*http.Transport).DisableKeepAlives = false
+	client := &http.Client{
+		Transport: transport,
+		Timeout:   time.Second * 30,
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			return http.ErrUseLastResponse
+		}}
+
+	req, err := http.NewRequest("GET", "http://"+serverAddress+":80/test1", nil)
+	s.AssertNil(err, fmt.Sprint(err))
+	resp, err := client.Do(req)
+	s.AssertNil(err, fmt.Sprint(err))
+	defer resp.Body.Close()
+	s.Log(DumpHttpResp(resp, true))
+	s.AssertEqual(200, resp.StatusCode)
+	s.AssertEqual(false, resp.Close)
+	body, err := io.ReadAll(resp.Body)
+	s.AssertNil(err, fmt.Sprint(err))
+	s.AssertEqual(string(body), "hello")
+	o1 := vpp.Vppctl("show session verbose proto http state ready")
+	s.Log(o1)
+	s.AssertContains(o1, "ESTABLISHED")
+
+	req, err = http.NewRequest("GET", "http://"+serverAddress+":80/test2", nil)
+	s.AssertNil(err, fmt.Sprint(err))
+	clientTrace := &httptrace.ClientTrace{
+		GotConn: func(info httptrace.GotConnInfo) {
+			s.AssertEqual(true, info.Reused, "connection not reused")
+		},
+	}
+	req = req.WithContext(httptrace.WithClientTrace(req.Context(), clientTrace))
+	resp, err = client.Do(req)
+	s.AssertNil(err, fmt.Sprint(err))
+	defer resp.Body.Close()
+	s.Log(DumpHttpResp(resp, true))
+	s.AssertEqual(200, resp.StatusCode)
+	s.AssertEqual(false, resp.Close)
+	body, err = io.ReadAll(resp.Body)
+	s.AssertNil(err, fmt.Sprint(err))
+	s.AssertEqual(string(body), "some data")
+	s.AssertNil(err, fmt.Sprint(err))
+	o2 := vpp.Vppctl("show session verbose proto http state ready")
+	s.Log(o2)
+	s.AssertContains(o2, "ESTABLISHED")
+	s.AssertEqual(o1, o2)
+}
+
+func HttpPipeliningTest(s *NoTopoSuite) {
+	// testing url handler app do not support multi-thread
+	s.SkipIfMultiWorker()
+	vpp := s.GetContainerByName("vpp").VppInstance
+	serverAddress := s.GetInterfaceByName(TapInterfaceName).Peer.Ip4AddressString()
+	s.Log(vpp.Vppctl("http static server uri tcp://" + serverAddress + "/80 url-handlers debug"))
+	s.Log(vpp.Vppctl("test-url-handler enable"))
+
+	req1 := "GET /test_delayed HTTP/1.1\r\nHost:" + serverAddress + ":80\r\nUser-Agent:test\r\n\r\n"
+	req2 := "GET /test1 HTTP/1.1\r\nHost:" + serverAddress + ":80\r\nUser-Agent:test\r\n\r\n"
+
+	conn, err := net.DialTimeout("tcp", serverAddress+":80", time.Second*30)
+	s.AssertNil(err, fmt.Sprint(err))
+	defer conn.Close()
+	err = conn.SetDeadline(time.Now().Add(time.Second * 15))
+	s.AssertNil(err, fmt.Sprint(err))
+	n, err := conn.Write([]byte(req1))
+	s.AssertNil(err, fmt.Sprint(err))
+	s.AssertEqual(n, len([]rune(req1)))
+	// send second request a bit later so first is already in progress
+	time.Sleep(500 * time.Millisecond)
+	n, err = conn.Write([]byte(req2))
+	s.AssertNil(err, fmt.Sprint(err))
+	s.AssertEqual(n, len([]rune(req2)))
+	reply := make([]byte, 1024)
+	n, err = conn.Read(reply)
+	s.AssertNil(err, fmt.Sprint(err))
+	s.Log(string(reply))
+	s.AssertContains(string(reply), "delayed data", "first request response not received")
+	s.AssertNotContains(string(reply), "hello", "second request response received")
+	// make sure response for second request is not received later
+	_, err = conn.Read(reply)
+	s.AssertMatchError(err, os.ErrDeadlineExceeded, "second request response received")
+}
+
 func HttpCliTest(s *VethsSuite) {
 	serverContainer := s.GetContainerByName("server-vpp")
 	clientContainer := s.GetContainerByName("client-vpp")
diff --git a/extras/hs-test/infra/hst_suite.go b/extras/hs-test/infra/hst_suite.go
index 2cf241afa64..1f1d54b1b94 100644
--- a/extras/hs-test/infra/hst_suite.go
+++ b/extras/hs-test/infra/hst_suite.go
@@ -224,6 +224,10 @@ func (s *HstSuite) AssertNotEmpty(object interface{}, msgAndArgs ...interface{})
 	Expect(object).ToNot(BeEmpty(), msgAndArgs...)
 }
 
+func (s *HstSuite) AssertMatchError(actual, expected error, msgAndArgs ...interface{}) {
+	Expect(actual).To(MatchError(expected))
+}
+
 func (s *HstSuite) CreateLogger() {
 	suiteName := s.GetCurrentSuiteName()
 	var err error
diff --git a/src/plugins/hs_apps/CMakeLists.txt b/src/plugins/hs_apps/CMakeLists.txt
index 179c9c7a4c4..3553a25837e 100644
--- a/src/plugins/hs_apps/CMakeLists.txt
+++ b/src/plugins/hs_apps/CMakeLists.txt
@@ -23,6 +23,7 @@ add_vpp_plugin(hs_apps
   http_client_cli.c
   http_tps.c
   proxy.c
+  test_builtins.c
 )
 
 ##############################################################################
diff --git a/src/plugins/hs_apps/test_builtins.c b/src/plugins/hs_apps/test_builtins.c
new file mode 100644
index 00000000000..cd6b00d0251
--- /dev/null
+++ b/src/plugins/hs_apps/test_builtins.c
@@ -0,0 +1,171 @@
+/* SPDX-License-Identifier: Apache-2.0
+ * Copyright(c) 2024 Cisco Systems, Inc.
+ */
+
+#include <http_static/http_static.h>
+#include <vppinfra/tw_timer_2t_1w_2048sl.h>
+
+typedef struct
+{
+  u32 stop_timer_handle;
+  hss_session_handle_t sh;
+} tw_timer_elt_t;
+
+typedef struct tb_main_
+{
+  tw_timer_elt_t *delayed_resps;
+  tw_timer_wheel_2t_1w_2048sl_t tw;
+  hss_session_send_fn send_data;
+} tb_main_t;
+
+static tb_main_t tb_main;
+
+static uword
+test_builtins_timer_process (vlib_main_t *vm, vlib_node_runtime_t *rt,
+			     vlib_frame_t *f)
+{
+  tb_main_t *tbm = &tb_main;
+  f64 now, timeout = 1.0;
+  uword *event_data = 0;
+  uword __clib_unused event_type;
+
+  while (1)
+    {
+      vlib_process_wait_for_event_or_clock (vm, timeout);
+      now = vlib_time_now (vm);
+      event_type = vlib_process_get_events (vm, (uword **) &event_data);
+
+      /* expire timers */
+      tw_timer_expire_timers_2t_1w_2048sl (&tbm->tw, now);
+
+      vec_reset_length (event_data);
+    }
+  return 0;
+}
+
+VLIB_REGISTER_NODE (test_builtins_timer_process_node) = {
+  .function = test_builtins_timer_process,
+  .type = VLIB_NODE_TYPE_PROCESS,
+  .name = "test-builtins-timer-process",
+  .state = VLIB_NODE_STATE_DISABLED,
+};
+
+static void
+send_data_to_hss (hss_session_handle_t sh, u8 *data)
+{
+  tb_main_t *tbm = &tb_main;
+  hss_url_handler_args_t args = {};
+
+  args.sh = sh;
+  args.data = data;
+  args.data_len = vec_len (data);
+  args.ct = HTTP_CONTENT_TEXT_PLAIN;
+  args.sc = HTTP_STATUS_OK;
+  args.free_vec_data = 1;
+
+  tbm->send_data (&args);
+}
+
+static hss_url_handler_rc_t
+handle_get_test1 (hss_url_handler_args_t *args)
+{
+  u8 *data;
+
+  clib_warning ("get request on test1");
+  data = format (0, "hello");
+  send_data_to_hss (args->sh, data);
+
+  return HSS_URL_HANDLER_ASYNC;
+}
+
+static hss_url_handler_rc_t
+handle_get_test2 (hss_url_handler_args_t *args)
+{
+  u8 *data;
+
+  clib_warning ("get request on test2");
+  data = format (0, "some data");
+  send_data_to_hss (args->sh, data);
+
+  return HSS_URL_HANDLER_ASYNC;
+}
+
+static void
+delayed_resp_cb (u32 *expired_timers)
+{
+  tb_main_t *tbm = &tb_main;
+  int i;
+  u32 pool_index;
+  tw_timer_elt_t *e;
+  u8 *data;
+
+  for (i = 0; i < vec_len (expired_timers); i++)
+    {
+      pool_index = expired_timers[i] & 0x7FFFFFFF;
+      e = pool_elt_at_index (tbm->delayed_resps, pool_index);
+      clib_warning ("sending delayed data");
+      data = format (0, "delayed data");
+      send_data_to_hss (e->sh, data);
+      pool_put (tbm->delayed_resps, e);
+    }
+}
+
+static hss_url_handler_rc_t
+handle_get_test_delayed (hss_url_handler_args_t *args)
+{
+  tb_main_t *tbm = &tb_main;
+  tw_timer_elt_t *e;
+
+  clib_warning ("get request on test_delayed");
+  pool_get (tbm->delayed_resps, e);
+  e->sh = args->sh;
+  e->stop_timer_handle =
+    tw_timer_start_2t_1w_2048sl (&tbm->tw, e - tbm->delayed_resps, 0, 5);
+
+  return HSS_URL_HANDLER_ASYNC;
+}
+
+static void
+test_builtins_init (vlib_main_t *vm)
+{
+  tb_main_t *tbm = &tb_main;
+  hss_register_url_fn fp;
+  vlib_node_t *n;
+
+  fp = vlib_get_plugin_symbol ("http_static_plugin.so",
+			       "hss_register_url_handler");
+
+  if (fp == 0)
+    {
+      clib_warning ("http_static_plugin.so not loaded...");
+      return;
+    }
+
+  (*fp) (handle_get_test1, "test1", HTTP_REQ_GET);
+  (*fp) (handle_get_test2, "test2", HTTP_REQ_GET);
+  (*fp) (handle_get_test_delayed, "test_delayed", HTTP_REQ_GET);
+
+  tbm->send_data =
+    vlib_get_plugin_symbol ("http_static_plugin.so", "hss_session_send_data");
+
+  tw_timer_wheel_init_2t_1w_2048sl (&tbm->tw, delayed_resp_cb, 1.0, ~0);
+
+  vlib_node_set_state (vm, test_builtins_timer_process_node.index,
+		       VLIB_NODE_STATE_POLLING);
+  n = vlib_get_node (vm, test_builtins_timer_process_node.index);
+  vlib_start_process (vm, n->runtime_index);
+}
+
+static clib_error_t *
+test_builtins_enable_command_fn (vlib_main_t *vm, unformat_input_t *input,
+				 vlib_cli_command_t *cmd)
+{
+  test_builtins_init (vm);
+  return 0;
+}
+
+VLIB_CLI_COMMAND (test_builtins_enable_command, static) = {
+  .path = "test-url-handler enable",
+  .short_help = "test-url-handler enable",
+  .function = test_builtins_enable_command_fn,
+};
diff --git a/src/plugins/http/http.c b/src/plugins/http/http.c
index db101539ac4..416ba8fe817 100644
--- a/src/plugins/http/http.c
+++ b/src/plugins/http/http.c
@@ -82,11 +82,19 @@ http_state_is_tx_valid (http_conn_t *hc)
 {
   http_state_t state = hc->http_state;
   return (state == HTTP_STATE_APP_IO_MORE_DATA ||
-	  state == HTTP_STATE_CLIENT_IO_MORE_DATA ||
 	  state == HTTP_STATE_WAIT_APP_REPLY ||
 	  state == HTTP_STATE_WAIT_APP_METHOD);
 }
 
+static inline int
+http_state_is_rx_valid (http_conn_t *hc)
+{
+  http_state_t state = hc->http_state;
+  return (state == HTTP_STATE_WAIT_SERVER_REPLY ||
+	  state == HTTP_STATE_CLIENT_IO_MORE_DATA ||
+	  state == HTTP_STATE_WAIT_CLIENT_METHOD);
+}
+
 static inline http_worker_t *
 http_worker_get (u32 thread_index)
 {
@@ -1060,7 +1068,6 @@ http_state_wait_app_reply (http_conn_t *hc, transport_send_params_t *sp)
   return HTTP_SM_CONTINUE;
 
 error:
-  clib_warning ("unexpected msg type from app %u", msg.type);
   http_send_error (hc, sc);
   http_state_change (hc, HTTP_STATE_WAIT_CLIENT_METHOD);
   session_transport_closing_notify (&hc->connection);
@@ -1294,12 +1301,16 @@ http_ts_rx_callback (session_t *ts)
       return -1;
     }
 
-  if (hc->state == HTTP_CONN_STATE_CLOSED)
+  if (!http_state_is_rx_valid (hc))
     {
+      if (hc->state != HTTP_CONN_STATE_CLOSED)
+	clib_warning ("app data req state '%U' session state %u",
+		      format_http_state, hc->http_state, hc->state);
       svm_fifo_dequeue_drop_all (ts->tx_fifo);
       return 0;
     }
 
+  HTTP_DBG (1, "run state machine");
   http_req_run_state_machine (hc, 0);
 
   if (hc->state == HTTP_CONN_STATE_TRANSPORT_CLOSED)
@@ -1605,6 +1616,7 @@ http_app_tx_callback (void *session, transport_send_params_t *sp)
   max_burst_sz = sp->max_burst_size * TRANSPORT_PACER_MIN_MSS;
   sp->max_burst_size = max_burst_sz;
 
+  HTTP_DBG (1, "run state machine");
   http_req_run_state_machine (hc, sp);
 
   if (hc->state == HTTP_CONN_STATE_APP_CLOSED)