Read complete bounded Memcached responses

docs/wasm-runtime-architecture.md

@@ -214,6 +214,14 @@ hostcall loop therefore cannot renew an invocation indefinitely. Blocking host
 helpers retain operation-specific limits and cap them to the remaining
 invocation budget where the underlying operation is cancellable. Forked task
 callbacks receive a fresh invocation deadline capped by the task lifetime.
+The Memcached ASCII hostcall applies one deadline to the complete command write
+and protocol-framed response read. Retrieval payload lengths are parsed, so a
+payload containing `END` cannot terminate the response; multi-value and `stats`
+responses may span socket chunks up to the 8 MiB response bound. Timeout,
+malformed framing, or overflow shuts down the retained descriptor before
+returning an empty result, preventing unread response bytes from poisoning the
+next command on that connection. The native fallback uses the same parser and
+deadline contract.
 Synchronous compiler locks, transitive `#load` compilation, and compiler child
 processes consume that same deadline. Compiler children run in a dedicated
 process group; timeout kills the group, retains the previous generation, and

scripts/test_wasm_compile_timeout.sh

@@ -45,7 +45,7 @@ HTTP_PORT=
 HTTP_DOCUMENT_ROOT=$site
 SESSION_PATH=$root/session
 TMP_UPLOAD_PATH=$root/upload
-WASM_CORE_PATH=$(pwd)/bin/wasm/core.wasm
+WASM_CORE_PATH=$(pwd -P)/bin/wasm/core.wasm
 WASM_COMPILE_SCRIPT=$root/compile
 WASM_INVOCATION_TIMEOUT_MS=2000
 WASM_EPOCH_PERIOD_MS=20

site/doc/pages/memcache_command.txt

@@ -13,6 +13,7 @@ return value : string containing the Memcache server's response
 Executes a raw command on an open Memcache connection and returns the server response as a string.
 
 This is the low-level escape hatch for Memcache operations that are not covered by the dedicated helpers.
+UCE reads the complete framed ASCII response, including multi-value retrievals and `stats`, rather than returning a socket chunk. The complete write and read share a one-second deadline and responses are limited to 8 MiB. A timeout, malformed frame, or oversized response returns an empty string and makes that connection unusable so unread bytes cannot affect a later command. Commands ending in `noreply` and `quit` return immediately with an empty string.
 
 :example
 u64 conn = memcache_connect();

site/tests/services.uce

@@ -90,10 +90,25 @@ RENDER(Request& context)
 		StringMap mem_multi = memcache_get_multiple(memfd, {"site-tests-key", "site-tests-key2"});
 		memcache_delete(memfd, "site-tests-key");
 		memcache_delete(memfd, "site-tests-key2");
+		String large_mem_value(200 * 1024, 'x');
+		large_mem_value.replace(1024, 7, "\r\nEND\r\n");
+		bool large_mem_stored = memcache_set(memfd, "site-tests-large", large_mem_value);
+		String large_mem_loaded = memcache_get(memfd, "site-tests-large");
+		String second_large_mem_value(160 * 1024, 'y');
+		bool second_large_mem_stored = memcache_set(memfd, "site-tests-large-2", second_large_mem_value);
+		StringMap large_mem_multi = memcache_get_multiple(memfd, {"site-tests-large", "site-tests-large-2"});
+		String no_reply = memcache_command(memfd, "set site-tests-noreply 0 60 5 noreply\r\nvalue");
+		String no_reply_value = memcache_get(memfd, "site-tests-noreply");
+		String stats_mem = memcache_command(memfd, "stats");
+		String stats_reset_mem = memcache_command(memfd, "stats reset");
+		String after_stats_reset = memcache_get(memfd, "site-tests-noreply");
+		bool large_mem_deleted = memcache_delete(memfd, "site-tests-large");
+		bool second_large_mem_deleted = memcache_delete(memfd, "site-tests-large-2");
+		bool no_reply_deleted = memcache_delete(memfd, "site-tests-noreply");
 		mark(
 			"memcache_connect() / memcache_command() / memcache_get_multiple()",
-			mem_value == "value-1" && contains(raw_mem, "VALUE site-tests-key") && mem_multi["site-tests-key"] == "value-1" && mem_multi["site-tests-key2"] == "value-2" ? "pass" : "fail",
-			"memcache value=" + mem_value + " raw=" + raw_mem + " multi=" + var_dump(mem_multi)
+			mem_value == "value-1" && contains(raw_mem, "VALUE site-tests-key") && mem_multi["site-tests-key"] == "value-1" && mem_multi["site-tests-key2"] == "value-2" && large_mem_stored && large_mem_loaded == large_mem_value && second_large_mem_stored && large_mem_multi["site-tests-large"] == large_mem_value && large_mem_multi["site-tests-large-2"] == second_large_mem_value && no_reply == "" && no_reply_value == "value" && contains(stats_mem, "STAT ") && trim(stats_reset_mem) == "RESET" && after_stats_reset == "value" && large_mem_deleted && second_large_mem_deleted && no_reply_deleted ? "pass" : "fail",
+			"memcache value=" + mem_value + " raw=" + raw_mem + " multi=" + var_dump(mem_multi) + " large=" + std::to_string(large_mem_loaded.size()) + " multi-large=" + std::to_string(large_mem_multi["site-tests-large"].size() + large_mem_multi["site-tests-large-2"].size()) + " noreply=" + no_reply_value + " stats=" + std::to_string(stats_mem.size()) + "/" + trim(stats_reset_mem) + " after-reset=" + after_stats_reset + " deleted=" + (large_mem_deleted && second_large_mem_deleted && no_reply_deleted ? "true" : "false")
 		);
 	}
 

src/lib/sys.cpp

@@ -737,6 +737,7 @@ StringMap default_config()
 #include <fcntl.h>
 #include <limits.h>
 #include <sys/file.h>
+#include <poll.h>
 #include <errno.h>
 #include "sys.h"
 #include "hash.h"
@@ -1291,6 +1292,185 @@ String socket_read(u64 sockfd, u32 max_length, u32 timeout)
 	return("");
 }
 
+static bool memcache_decimal(String value, u64& result)
+{
+	if(value == "")
+		return(false);
+	result = 0;
+	for(char c : value)
+	{
+		if(c < '0' || c > '9' || result > (UINT64_MAX - (u64)(c - '0')) / 10)
+			return(false);
+		result = result * 10 + (u64)(c - '0');
+	}
+	return(true);
+}
+
+static String memcache_command_line(const String& command)
+{
+	size_t end = command.find("\r\n");
+	return(trim(command.substr(0, end)));
+}
+
+bool uce_memcache_command_has_no_reply(const String& command)
+{
+	StringList words = split(memcache_command_line(command), " ");
+	if(words.size() == 0)
+		return(false);
+	String verb = to_lower(words[0]);
+	return(verb == "quit" || to_lower(words.back()) == "noreply");
+}
+
+MemcacheResponseState uce_memcache_response_advance(const String& command, const String& response, MemcacheResponseParser& parser)
+{
+	while(true)
+	{
+		if(parser.mode == MemcacheResponseParser::Mode::FirstLine)
+		{
+			size_t line_end = response.find("\r\n");
+			if(line_end == String::npos)
+				return(MemcacheResponseState::Incomplete);
+			String line = response.substr(0, line_end);
+			if(line == "ERROR" || line.rfind("CLIENT_ERROR", 0) == 0 || line.rfind("SERVER_ERROR", 0) == 0 || line == "RESET" || line == "OK")
+				return(MemcacheResponseState::Complete);
+			if(line.rfind("VALUE ", 0) == 0)
+				parser.mode = MemcacheResponseParser::Mode::RetrievalHeader;
+			else if(line.rfind("VA ", 0) == 0)
+			{
+				StringList fields = split(line, " ");
+				u64 length = 0;
+				if(fields.size() < 2 || !memcache_decimal(fields[1], length) || length > SIZE_MAX - line_end - 4)
+					return(MemcacheResponseState::Malformed);
+				parser.data_end = line_end + 2 + (size_t)length;
+				parser.mode = MemcacheResponseParser::Mode::MetaData;
+			}
+			else
+			{
+				StringList request_words = split(memcache_command_line(command), " ");
+				String request_verb = request_words.size() ? to_lower(request_words[0]) : String("");
+				bool end_terminated = line.rfind("STAT ", 0) == 0 || line.rfind("ITEM ", 0) == 0 ||
+					request_verb == "stats" || (request_verb == "lru_crawler" && request_words.size() > 1 && to_lower(request_words[1]) == "metadump");
+				if(!end_terminated)
+					return(MemcacheResponseState::Complete);
+				parser.mode = MemcacheResponseParser::Mode::Lines;
+			}
+		}
+
+		if(parser.mode == MemcacheResponseParser::Mode::RetrievalHeader)
+		{
+			size_t line_end = response.find("\r\n", parser.offset);
+			if(line_end == String::npos)
+				return(MemcacheResponseState::Incomplete);
+			String line = response.substr(parser.offset, line_end - parser.offset);
+			if(line == "END")
+				return(MemcacheResponseState::Complete);
+			StringList fields = split(line, " ");
+			u64 length = 0;
+			if(fields.size() < 4 || fields[0] != "VALUE" || !memcache_decimal(fields[3], length) || length > SIZE_MAX - line_end - 4)
+				return(MemcacheResponseState::Malformed);
+			parser.data_end = line_end + 2 + (size_t)length;
+			parser.mode = MemcacheResponseParser::Mode::RetrievalData;
+		}
+
+		if(parser.mode == MemcacheResponseParser::Mode::RetrievalData)
+		{
+			if(response.size() < parser.data_end + 2)
+				return(MemcacheResponseState::Incomplete);
+			if(response.compare(parser.data_end, 2, "\r\n") != 0)
+				return(MemcacheResponseState::Malformed);
+			parser.offset = parser.data_end + 2;
+			parser.mode = MemcacheResponseParser::Mode::RetrievalHeader;
+		}
+
+		if(parser.mode == MemcacheResponseParser::Mode::MetaData)
+		{
+			if(response.size() < parser.data_end + 2)
+				return(MemcacheResponseState::Incomplete);
+			return(response.compare(parser.data_end, 2, "\r\n") == 0 ? MemcacheResponseState::Complete : MemcacheResponseState::Malformed);
+		}
+
+		if(parser.mode == MemcacheResponseParser::Mode::Lines)
+		{
+			size_t line_end = response.find("\r\n", parser.offset);
+			if(line_end == String::npos)
+				return(MemcacheResponseState::Incomplete);
+			if(response.compare(parser.offset, line_end - parser.offset, "END") == 0)
+				return(MemcacheResponseState::Complete);
+			parser.offset = line_end + 2;
+		}
+	}
+}
+
+static bool native_socket_wait(int fd, short events, u64 deadline)
+{
+	while(true)
+	{
+		u64 now = monotonic_ms();
+		if(now >= deadline)
+			return(false);
+		pollfd item = {fd, events, 0};
+		int timeout = (int)std::min<u64>(deadline - now, INT_MAX);
+		int ready = poll(&item, 1, timeout);
+		if(ready > 0)
+			return((item.revents & events) != 0);
+		if(ready == 0 || errno != EINTR)
+			return(false);
+	}
+}
+
+static String native_memcache_exchange(u64 connection, String command)
+{
+	static constexpr size_t max_response = 8 * 1024 * 1024;
+	int fd = (int)connection;
+	u64 deadline = monotonic_ms() + 1000;
+	String request = command + "\r\n";
+	size_t sent = 0;
+	while(sent < request.size())
+	{
+		if(!native_socket_wait(fd, POLLOUT, deadline))
+			break;
+		ssize_t count = send(fd, request.data() + sent, request.size() - sent, MSG_DONTWAIT | MSG_NOSIGNAL);
+		if(count > 0)
+			sent += (size_t)count;
+		else if(count < 0 && (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK))
+			continue;
+		else
+			break;
+	}
+	if(sent != request.size())
+	{
+		shutdown(fd, SHUT_RDWR);
+		return("");
+	}
+	if(uce_memcache_command_has_no_reply(command))
+		return("");
+
+	String response;
+	response.reserve(16 * 1024);
+	MemcacheResponseParser parser;
+	while(response.size() < max_response)
+	{
+		MemcacheResponseState state = uce_memcache_response_advance(command, response, parser);
+		if(state == MemcacheResponseState::Complete)
+			return(response);
+		if(state == MemcacheResponseState::Malformed || !native_socket_wait(fd, POLLIN, deadline))
+			break;
+		char buffer[64 * 1024];
+		size_t remaining = max_response - response.size();
+		ssize_t count = recv(fd, buffer, std::min(remaining, sizeof(buffer)), MSG_DONTWAIT);
+		if(count > 0)
+			response.append(buffer, (size_t)count);
+		else if(count < 0 && (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK))
+			continue;
+		else
+			break;
+	}
+	if(uce_memcache_response_advance(command, response, parser) == MemcacheResponseState::Complete)
+		return(response);
+	shutdown(fd, SHUT_RDWR);
+	return("");
+}
+
 String memcache_escape_key(String key)
 {
 	String result;
@@ -1320,26 +1500,18 @@ u64 memcache_connect(String host, u16 port)
 
 String memcache_command(u64 connection, String command)
 {
-	socket_write(connection, command+"\r\n");
-	return(socket_read(connection)); // FIXME: do multi-chunk until END line is received!
+	return(native_memcache_exchange(connection, command));
 }
 
 bool memcache_set(u64 connection, String key, String value, u64 expires_in)
 {
-	socket_write(connection,
-		// set KEY META_DATA EXPIRY_TIME LENGTH_IN_BYTES
-		String("set ") + memcache_escape_key(key) + " 0 " + std::to_string(expires_in) + " " + std::to_string(value.length()) + "\r\n" +
-		value + "\r\n");
-	return("STORED" == trim(socket_read(connection)));
+	return("STORED" == trim(memcache_command(connection,
+		String("set ") + memcache_escape_key(key) + " 0 " + std::to_string(expires_in) + " " + std::to_string(value.length()) + "\r\n" + value)));
 }
 
 bool memcache_delete(u64 connection, String key)
 {
-	socket_write(connection,
-		// set KEY META_DATA EXPIRY_TIME LENGTH_IN_BYTES
-		String("delete ") + memcache_escape_key(key) + "\r\n"
-		);
-	return("DELETED" == trim(socket_read(connection)));
+	return("DELETED" == trim(memcache_command(connection, String("delete ") + memcache_escape_key(key))));
 }
 
 String memcache_get(u64 connection, String key, String default_value)

src/lib/sys.h

@@ -128,6 +128,15 @@ String signal_name(s32 sig);
 String memcache_escape_key(String key);
 StringList memcache_escape_keys(StringList keys);
 u64 memcache_connect(String host = "127.0.0.1", u16 port = 11211);
+enum class MemcacheResponseState { Incomplete, Complete, Malformed };
+struct MemcacheResponseParser
+{
+	enum class Mode { FirstLine, RetrievalHeader, RetrievalData, Lines, MetaData } mode = Mode::FirstLine;
+	size_t offset = 0;
+	size_t data_end = 0;
+};
+__attribute__((visibility("hidden"))) bool uce_memcache_command_has_no_reply(const String& command);
+__attribute__((visibility("hidden"))) MemcacheResponseState uce_memcache_response_advance(const String& command, const String& response, MemcacheResponseParser& parser);
 String memcache_command(u64 connection, String command);
 bool memcache_set(u64 connection, String key, String value, u64 expires_in = 60*60);
 bool memcache_delete(u64 connection, String key);

src/wasm/worker.cpp

@@ -395,10 +395,9 @@ static u64 wasm_socket_connect_bounded(const String& host, u16 port, u64 timeout
 	return((u64)fd);
 }
 
-static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 timeout_ms)
+static bool wasm_socket_write_until(u64 socket_fd, const String& data, u64 deadline)
 {
 	int fd = (int)socket_fd;
-	u64 deadline = wasm_deadline_after_ms(timeout_ms);
 	size_t offset = 0;
 	while(offset < data.size())
 	{
@@ -413,6 +412,11 @@ static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 tim
 	return(true);
 }
 
+static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 timeout_ms)
+{
+	return(wasm_socket_write_until(socket_fd, data, wasm_deadline_after_ms(timeout_ms)));
+}
+
 static String wasm_socket_read_bounded(u64 socket_fd, u32 max_length, u64 timeout_ms)
 {
 	if(max_length == 0 || !wasm_socket_wait((int)socket_fd, POLLIN, wasm_deadline_after_ms(timeout_ms)))
@@ -422,6 +426,42 @@ static String wasm_socket_read_bounded(u64 socket_fd, u32 max_length, u64 timeou
 	return(count > 0 ? String(buffer.data(), (size_t)count) : String(""));
 }
 
+static String wasm_memcache_exchange(u64 socket_fd, const String& command, u64 timeout_ms)
+{
+	static constexpr size_t max_response = 8 * 1024 * 1024;
+	int fd = (int)socket_fd;
+	u64 deadline = wasm_deadline_after_ms(timeout_ms);
+	if(!wasm_socket_write_until(socket_fd, command + "\r\n", deadline))
+	{
+		shutdown(fd, SHUT_RDWR);
+		return("");
+	}
+	if(uce_memcache_command_has_no_reply(command))
+		return("");
+	String response;
+	response.reserve(16 * 1024);
+	MemcacheResponseParser parser;
+	while(response.size() < max_response)
+	{
+		MemcacheResponseState state = uce_memcache_response_advance(command, response, parser);
+		if(state == MemcacheResponseState::Complete)
+			return(response);
+		if(state == MemcacheResponseState::Malformed || !wasm_socket_wait(fd, POLLIN, deadline))
+			break;
+		char buffer[64 * 1024];
+		size_t remaining = max_response - response.size();
+		ssize_t count = recv(fd, buffer, std::min(remaining, sizeof(buffer)), MSG_DONTWAIT);
+		if(count > 0)
+			response.append(buffer, (size_t)count);
+		else if(count == 0 || (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK))
+			break;
+	}
+	if(uce_memcache_response_advance(command, response, parser) == MemcacheResponseState::Complete)
+		return(response);
+	shutdown(fd, SHUT_RDWR);
+	return("");
+}
+
 static f64 wasm_thread_cpu_time()
 {
 	struct timespec ts;
@@ -4254,8 +4294,7 @@ private:
 				else
 				{
 					u64 socket_fd = (u64)args[0].i64();
-					wasm_socket_write_bounded(socket_fd, command + "\r\n", self->bounded_hostcall_timeout_ms(1000));
-					out = wasm_socket_read_bounded(socket_fd, 1024 * 128, self->bounded_hostcall_timeout_ms(1000));
+					out = wasm_memcache_exchange(socket_fd, command, self->bounded_hostcall_timeout_ms(1000));
 					if(buf == 0)
 					{
 						self->staged_memcache_key = key;