Merge branch 'main' into sb/no-networksb/no-network

3 files changed, 103 insertions, 81 deletions

diff --git a/src/zenhttp/servers/httpasio.cpp b/src/zenhttp/servers/httpasio.cpp
index 643f33618..9f4875eaf 100644
--- a/src/zenhttp/servers/httpasio.cpp
+++ b/src/zenhttp/servers/httpasio.cpp
@@ -601,6 +601,7 @@ public:
 	bool						  m_IsLocalMachineRequest;
 	bool						  m_AllowZeroCopyFileSend = true;
 	std::string					  m_RemoteAddress;
+	std::string					  m_DecodedUri;	 // Percent-decoded URI; m_Uri/m_UriWithExtension point into this
 	std::unique_ptr<HttpResponse> m_Response;
 };
 
@@ -623,6 +624,7 @@ public:
 	~HttpResponse() = default;
 
 	void SetAllowZeroCopyFileSend(bool Allow) { m_AllowZeroCopyFileSend = Allow; }
+	void SetKeepAlive(bool KeepAlive) { m_IsKeepAlive = KeepAlive; }
 
 	/**
 	 * Initialize the response for sending a payload made up of multiple blobs
@@ -780,8 +782,8 @@ public:
 		return m_Headers;
 	}
 
-	template<typename SocketType>
-	void SendResponse(SocketType& Socket, std::function<void(const asio::error_code& Ec, std::size_t ByteCount)>&& Token)
+	template<typename SocketType, typename Executor>
+	void SendResponse(SocketType& Socket, Executor& Strand, std::function<void(const asio::error_code& Ec, std::size_t ByteCount)>&& Token)
 	{
 		ZEN_ASSERT(m_State == State::kInitialized);
 
@@ -791,11 +793,11 @@ public:
 		m_SendCb = std::move(Token);
 		m_State	 = State::kSending;
 
-		SendNextChunk(Socket);
+		SendNextChunk(Socket, Strand);
 	}
 
-	template<typename SocketType>
-	void SendNextChunk(SocketType& Socket)
+	template<typename SocketType, typename Executor>
+	void SendNextChunk(SocketType& Socket, Executor& Strand)
 	{
 		ZEN_ASSERT(m_State == State::kSending);
 
@@ -812,12 +814,12 @@ public:
 
 			auto CompletionToken = [Self = this, Token = std::move(m_SendCb), TotalBytes = m_TotalBytesSent] { Token({}, TotalBytes); };
 
-			asio::defer(Socket.get_executor(), std::move(CompletionToken));
+			asio::defer(Strand, std::move(CompletionToken));
 
 			return;
 		}
 
-		auto OnCompletion = [this, &Socket](const asio::error_code& Ec, std::size_t ByteCount) {
+		auto OnCompletion = asio::bind_executor(Strand, [this, &Socket, &Strand](const asio::error_code& Ec, std::size_t ByteCount) {
 			ZEN_ASSERT(m_State == State::kSending);
 
 			m_TotalBytesSent += ByteCount;
@@ -828,9 +830,9 @@ public:
 			}
 			else
 			{
-				SendNextChunk(Socket);
+				SendNextChunk(Socket, Strand);
 			}
-		};
+		});
 
 		const IoVec& Io = m_IoVecs[m_IoVecCursor++];
 
@@ -982,16 +984,14 @@ private:
 	void CloseConnection();
 	void SendInlineResponse(uint32_t RequestNumber, std::string_view StatusLine, std::string_view Headers = {}, std::string_view Body = {});
 
-	HttpAsioServerImpl&		 m_Server;
-	asio::streambuf			 m_RequestBuffer;
-	std::atomic<uint32_t>	 m_RequestCounter{0};
-	uint32_t				 m_ConnectionId = 0;
-	Ref<IHttpPackageHandler> m_PackageHandler;
-
-	RwLock									  m_ActiveResponsesLock;
+	HttpAsioServerImpl&						  m_Server;
+	std::unique_ptr<SocketType>				  m_Socket;
+	asio::strand<asio::any_io_executor>		  m_Strand;
+	asio::streambuf							  m_RequestBuffer;
+	uint32_t								  m_RequestCounter = 0;
+	uint32_t								  m_ConnectionId   = 0;
+	Ref<IHttpPackageHandler>				  m_PackageHandler;
 	std::deque<std::unique_ptr<HttpResponse>> m_ActiveResponses;
-
-	std::unique_ptr<SocketType> m_Socket;
 };
 
 std::atomic<uint32_t> g_ConnectionIdCounter{0};
@@ -999,8 +999,9 @@ std::atomic<uint32_t> g_ConnectionIdCounter{0};
 template<typename SocketType>
 HttpServerConnectionT<SocketType>::HttpServerConnectionT(HttpAsioServerImpl& Server, std::unique_ptr<SocketType>&& Socket)
 : m_Server(Server)
-, m_ConnectionId(g_ConnectionIdCounter.fetch_add(1))
 , m_Socket(std::move(Socket))
+, m_Strand(asio::make_strand(m_Socket->get_executor()))
+, m_ConnectionId(g_ConnectionIdCounter.fetch_add(1))
 {
 	ZEN_TRACE_VERBOSE("new connection #{}", m_ConnectionId);
 }
@@ -1008,8 +1009,6 @@ HttpServerConnectionT<SocketType>::HttpServerConnectionT(HttpAsioServerImpl& Ser
 template<typename SocketType>
 HttpServerConnectionT<SocketType>::~HttpServerConnectionT()
 {
-	RwLock::ExclusiveLockScope _(m_ActiveResponsesLock);
-
 	ZEN_TRACE_VERBOSE("destroying connection #{}", m_ConnectionId);
 }
 
@@ -1017,7 +1016,7 @@ template<typename SocketType>
 void
 HttpServerConnectionT<SocketType>::HandleNewRequest()
 {
-	EnqueueRead();
+	asio::dispatch(m_Strand, [Conn = AsSharedPtr()] { Conn->EnqueueRead(); });
 }
 
 template<typename SocketType>
@@ -1058,7 +1057,9 @@ HttpServerConnectionT<SocketType>::EnqueueRead()
 	asio::async_read(*m_Socket.get(),
 					 m_RequestBuffer,
 					 asio::transfer_at_least(1),
-					 [Conn = AsSharedPtr()](const asio::error_code& Ec, std::size_t ByteCount) { Conn->OnDataReceived(Ec, ByteCount); });
+					 asio::bind_executor(m_Strand, [Conn = AsSharedPtr()](const asio::error_code& Ec, std::size_t ByteCount) {
+						 Conn->OnDataReceived(Ec, ByteCount);
+					 }));
 }
 
 template<typename SocketType>
@@ -1091,7 +1092,7 @@ HttpServerConnectionT<SocketType>::OnDataReceived(const asio::error_code& Ec, [[
 
 	ZEN_TRACE_VERBOSE("on data received, connection: {}, request: {}, thread: {}, bytes: {}",
 					  m_ConnectionId,
-					  m_RequestCounter.load(std::memory_order_relaxed),
+					  m_RequestCounter,
 					  zen::GetCurrentThreadId(),
 					  NiceBytes(ByteCount));
 
@@ -1153,25 +1154,23 @@ HttpServerConnectionT<SocketType>::OnResponseDataSent(const asio::error_code&
 
 	if (ResponseToPop)
 	{
-		m_ActiveResponsesLock.WithExclusiveLock([&] {
-			// Once a response is sent we can release any referenced resources
-			//
-			// completion callbacks may be issued out-of-order so we need to
-			// remove the relevant entry from our active response list, it may
-			// not be the first
-
-			if (auto It = find_if(begin(m_ActiveResponses),
-								  end(m_ActiveResponses),
-								  [ResponseToPop](const auto& Item) { return Item.get() == ResponseToPop; });
-				It != end(m_ActiveResponses))
-			{
-				m_ActiveResponses.erase(It);
-			}
-			else
-			{
-				ZEN_WARN("response not found");
-			}
-		});
+		// Once a response is sent we can release any referenced resources
+		//
+		// completion callbacks may be issued out-of-order so we need to
+		// remove the relevant entry from our active response list, it may
+		// not be the first
+
+		if (auto It = find_if(begin(m_ActiveResponses),
+							  end(m_ActiveResponses),
+							  [ResponseToPop](const auto& Item) { return Item.get() == ResponseToPop; });
+			It != end(m_ActiveResponses))
+		{
+			m_ActiveResponses.erase(It);
+		}
+		else
+		{
+			ZEN_WARN("response not found");
+		}
 	}
 
 	if (!m_RequestData.IsKeepAlive())
@@ -1234,9 +1233,11 @@ HttpServerConnectionT<SocketType>::SendInlineResponse(uint32_t		   RequestNumber
 	asio::async_write(
 		*m_Socket,
 		Buffer,
-		[Conn = AsSharedPtr(), RequestNumber, Response = std::move(ResponseData)](const asio::error_code& Ec, std::size_t ByteCount) {
-			Conn->OnResponseDataSent(Ec, ByteCount, RequestNumber, /* ResponseToPop */ nullptr);
-		});
+		asio::bind_executor(
+			m_Strand,
+			[Conn = AsSharedPtr(), RequestNumber, Response = std::move(ResponseData)](const asio::error_code& Ec, std::size_t ByteCount) {
+				Conn->OnResponseDataSent(Ec, ByteCount, RequestNumber, /* ResponseToPop */ nullptr);
+			}));
 }
 
 template<typename SocketType>
@@ -1272,21 +1273,23 @@ HttpServerConnectionT<SocketType>::HandleRequest()
 				asio::async_write(
 					*m_Socket,
 					asio::buffer(ResponseStr->data(), ResponseStr->size()),
-					[Conn = AsSharedPtr(), WsHandler, OwnedResponse = ResponseStr](const asio::error_code& Ec, std::size_t) {
-						if (Ec)
-						{
-							ZEN_WARN("WebSocket 101 send failed: {}", Ec.message());
-							return;
-						}
-
-						Conn->m_Server.m_HttpServer->OnWebSocketConnectionOpened();
-						using WsConnType = WsAsioConnectionT<SocketType>;
-						Ref<WsConnType>			 WsConn(new WsConnType(std::move(Conn->m_Socket), *WsHandler, Conn->m_Server.m_HttpServer));
-						Ref<WebSocketConnection> WsConnRef(WsConn.Get());
-
-						WsHandler->OnWebSocketOpen(std::move(WsConnRef));
-						WsConn->Start();
-					});
+					asio::bind_executor(
+						m_Strand,
+						[Conn = AsSharedPtr(), WsHandler, OwnedResponse = ResponseStr](const asio::error_code& Ec, std::size_t) {
+							if (Ec)
+							{
+								ZEN_WARN("WebSocket 101 send failed: {}", Ec.message());
+								return;
+							}
+
+							Conn->m_Server.m_HttpServer->OnWebSocketConnectionOpened();
+							using WsConnType = WsAsioConnectionT<SocketType>;
+							Ref<WsConnType> WsConn(new WsConnType(std::move(Conn->m_Socket), *WsHandler, Conn->m_Server.m_HttpServer));
+							Ref<WebSocketConnection> WsConnRef(WsConn.Get());
+
+							WsHandler->OnWebSocketOpen(std::move(WsConnRef));
+							WsConn->Start();
+						}));
 
 				m_RequestState = RequestState::kDone;
 				return;
@@ -1312,7 +1315,7 @@ HttpServerConnectionT<SocketType>::HandleRequest()
 		m_RequestState = RequestState::kWriting;
 	}
 
-	const uint32_t RequestNumber = m_RequestCounter.fetch_add(1);
+	const uint32_t RequestNumber = m_RequestCounter++;
 
 	if (HttpService* Service = m_Server.RouteRequest(m_RequestData.Url()))
 	{
@@ -1444,31 +1447,34 @@ HttpServerConnectionT<SocketType>::HandleRequest()
 			{
 				ZEN_TRACE_CPU("asio::async_write");
 
-				std::string_view Headers = Response->GetHeaders();
+				HttpResponse* ResponseRaw = Response.get();
+				m_ActiveResponses.push_back(std::move(Response));
+
+				std::string_view Headers = ResponseRaw->GetHeaders();
 
 				std::vector<asio::const_buffer> AsioBuffers;
 				AsioBuffers.push_back(asio::const_buffer(Headers.data(), Headers.size()));
 
-				asio::async_write(*m_Socket.get(),
-								  AsioBuffers,
-								  asio::transfer_all(),
-								  [Conn = AsSharedPtr(), RequestNumber](const asio::error_code& Ec, std::size_t ByteCount) {
-									  Conn->OnResponseDataSent(Ec, ByteCount, RequestNumber, /* ResponseToPop */ nullptr);
-								  });
+				asio::async_write(
+					*m_Socket.get(),
+					AsioBuffers,
+					asio::transfer_all(),
+					asio::bind_executor(
+						m_Strand,
+						[Conn = AsSharedPtr(), ResponseRaw, RequestNumber](const asio::error_code& Ec, std::size_t ByteCount) {
+							Conn->OnResponseDataSent(Ec, ByteCount, RequestNumber, /* ResponseToPop */ ResponseRaw);
+						}));
 			}
 			else
 			{
 				ZEN_TRACE_CPU("asio::async_write");
 
 				HttpResponse* ResponseRaw = Response.get();
-
-				m_ActiveResponsesLock.WithExclusiveLock([&] {
-					// Keep referenced resources alive
-					m_ActiveResponses.push_back(std::move(Response));
-				});
+				m_ActiveResponses.push_back(std::move(Response));
 
 				ResponseRaw->SendResponse(
 					*m_Socket,
+					m_Strand,
 					[Conn = AsSharedPtr(), ResponseRaw, RequestNumber](const asio::error_code& Ec, std::size_t ByteCount) {
 						Conn->OnResponseDataSent(Ec, ByteCount, RequestNumber, /* ResponseToPop */ ResponseRaw);
 					});
@@ -1982,11 +1988,24 @@ HttpAsioServerRequest::HttpAsioServerRequest(HttpRequestParser& Request,
 {
 	const int PrefixLength = Service.UriPrefixLength();
 
-	std::string_view Uri = Request.Url();
-	Uri.remove_prefix(std::min(PrefixLength, static_cast<int>(Uri.size())));
-	m_Uri			   = Uri;
-	m_UriWithExtension = Uri;
-	m_QueryString	   = Request.QueryString();
+	std::string_view RawUri = Request.Url();
+	RawUri.remove_prefix(std::min(PrefixLength, static_cast<int>(RawUri.size())));
+
+	// Percent-decode the URI path so handlers see the same decoded paths regardless
+	// of whether the ASIO or http.sys backend is used (http.sys pre-decodes via CookedUrl).
+	// Skip the allocation when there is nothing to decode (common case).
+	if (RawUri.find('%') != std::string_view::npos)
+	{
+		m_DecodedUri	   = Decode(RawUri);
+		m_Uri			   = m_DecodedUri;
+		m_UriWithExtension = m_DecodedUri;
+	}
+	else
+	{
+		m_Uri			   = RawUri;
+		m_UriWithExtension = RawUri;
+	}
+	m_QueryString = Request.QueryString();
 
 	m_Verb			= Request.RequestVerb();
 	m_ContentLength = Request.Body().Size();
@@ -2083,6 +2102,7 @@ HttpAsioServerRequest::WriteResponse(HttpResponseCode ResponseCode)
 
 	m_Response.reset(new HttpResponse(HttpContentType::kBinary, m_RequestNumber));
 	m_Response->SetAllowZeroCopyFileSend(m_AllowZeroCopyFileSend);
+	m_Response->SetKeepAlive(m_Request.IsKeepAlive());
 	std::array<IoBuffer, 0> Empty;
 
 	m_Response->InitializeForPayload((uint16_t)ResponseCode, Empty);
@@ -2097,6 +2117,7 @@ HttpAsioServerRequest::WriteResponse(HttpResponseCode ResponseCode, HttpContentT
 
 	m_Response.reset(new HttpResponse(ContentType, m_RequestNumber));
 	m_Response->SetAllowZeroCopyFileSend(m_AllowZeroCopyFileSend);
+	m_Response->SetKeepAlive(m_Request.IsKeepAlive());
 	m_Response->InitializeForPayload((uint16_t)ResponseCode, Blobs);
 }
 
@@ -2108,6 +2129,7 @@ HttpAsioServerRequest::WriteResponse(HttpResponseCode ResponseCode, HttpContentT
 	ZEN_ASSERT(!m_Response);
 	m_Response.reset(new HttpResponse(ContentType, m_RequestNumber));
 	m_Response->SetAllowZeroCopyFileSend(m_AllowZeroCopyFileSend);
+	m_Response->SetKeepAlive(m_Request.IsKeepAlive());
 
 	IoBuffer				MessageBuffer(IoBuffer::Wrap, ResponseString.data(), ResponseString.size());
 	std::array<IoBuffer, 1> SingleBufferList({MessageBuffer});
diff --git a/src/zenhttp/servers/httpplugin.cpp b/src/zenhttp/servers/httpplugin.cpp
index 4bf8c61bb..31b0315d4 100644
--- a/src/zenhttp/servers/httpplugin.cpp
+++ b/src/zenhttp/servers/httpplugin.cpp
@@ -147,7 +147,7 @@ public:
 	HttpPluginServerRequest& operator=(const HttpPluginServerRequest&) = delete;
 
 	// As this is plugin transport connection used for specialized connections we assume it is not a machine local connection
-	virtual bool			 IsLocalMachineRequest() const /* override*/ { return false; }
+	virtual bool			 IsLocalMachineRequest() const override { return false; }
 	virtual std::string_view GetAuthorizationHeader() const override;
 	virtual Oid				 ParseSessionId() const override;
 	virtual uint32_t		 ParseRequestId() const override;
diff --git a/src/zenhttp/servers/httpsys.cpp b/src/zenhttp/servers/httpsys.cpp
index 4d6a53696..f8fb1c9be 100644
--- a/src/zenhttp/servers/httpsys.cpp
+++ b/src/zenhttp/servers/httpsys.cpp
@@ -1173,7 +1173,7 @@ HttpSysServer::RegisterHttpUrls(int BasePort)
 	{
 		Result = HttpAddUrlToUrlGroup(m_HttpUrlGroupId, WildcardUrlPath.c_str(), HTTP_URL_CONTEXT(0), 0);
 
-		if ((Result == ERROR_SHARING_VIOLATION))
+		if (Result == ERROR_SHARING_VIOLATION)
 		{
 			ZEN_INFO("Desired port {} is in use (HttpAddUrlToUrlGroup returned: {}), retrying",
 					 EffectivePort,