Make the relay acknowledge all received commands

To that effect, bump liberty and the xC relay protocol version.
Relay events have been reordered to improve forward compatibility.

Also prevent use-after-free when serialization fails.

xP now slightly throttles activity notifications,
and indicates when there are unacknowledged commands.

1 files changed, 49 insertions, 26 deletions

diff --git a/xC.c b/xC.c
index aef693e..fb3cf99 100644
--- a/xC.c
+++ b/xC.c
@@ -1818,6 +1818,7 @@ struct client
 
 	uint32_t event_seq;                 ///< Outgoing message counter
 	bool initialized;                   ///< Initial sync took place
+	bool closing;                       ///< We're closing the connection
 
 	struct poller_fd socket_event;      ///< The socket can be read/written to
 };
@@ -1875,7 +1876,7 @@ enum server_state
 	IRC_CONNECTED,                      ///< Trying to register
 	IRC_REGISTERED,                     ///< We can chat now
 	IRC_CLOSING,                        ///< Flushing output before shutdown
-	IRC_HALF_CLOSED                     ///< Connection shutdown from our side
+	IRC_HALF_CLOSED                     ///< Connection shut down from our side
 };
 
 /// Convert an IRC identifier character to lower-case
@@ -2263,14 +2264,6 @@ struct app_context
 
 	struct str_map servers;             ///< Our servers
 
-	// Relay:
-
-	int relay_fd;                       ///< Listening socket FD
-	struct client *clients;             ///< Our relay clients
-
-	/// A single message buffer to prepare all outcoming messages within
-	struct relay_event_message relay_message;
-
 	// Events:
 
 	struct poller_fd tty_event;         ///< Terminal input event
@@ -2322,6 +2315,14 @@ struct app_context
 	char *editor_filename;              ///< The file being edited by user
 	int terminal_suspended;             ///< Terminal suspension level
 
+	// Relay:
+
+	int relay_fd;                       ///< Listening socket FD
+	struct client *clients;             ///< Our relay clients
+
+	/// A single message buffer to prepare all outcoming messages within
+	struct relay_event_message relay_message;
+
 	// Plugins:
 
 	struct plugin *plugins;             ///< Loaded plugins
@@ -2392,8 +2393,6 @@ app_context_init (struct app_context *self)
 	self->config = config_make ();
 	poller_init (&self->poller);
 
-	self->relay_fd = -1;
-
 	self->servers = str_map_make ((str_map_free_fn) server_unref);
 	self->servers.key_xfrm = tolower_ascii_strxfrm;
 
@@ -2417,6 +2416,8 @@ app_context_init (struct app_context *self)
 
 	self->nick_palette =
 		filter_color_cube_for_acceptable_nick_colors (&self->nick_palette_len);
+
+	self->relay_fd = -1;
 }
 
 static void
@@ -4152,8 +4153,11 @@ client_kill (struct client *c)
 static void
 client_update_poller (struct client *c, const struct pollfd *pfd)
 {
+	// In case of closing without any data in the write buffer,
+	// we don't actually need to be able to write to the socket,
+	// but the condition should be quick to satisfy.
 	int new_events = POLLIN;
-	if (c->write_buffer.len)
+	if (c->write_buffer.len || c->closing)
 		new_events |= POLLOUT;
 
 	hard_assert (new_events != 0);
@@ -4168,9 +4172,7 @@ relay_send (struct client *c)
 {
 	struct relay_event_message *m = &c->ctx->relay_message;
 	m->event_seq = c->event_seq++;
-
-	// TODO: Also don't try sending anything if half-closed.
-	if (!c->initialized || c->socket_fd == -1)
+	if (!c->initialized || c->closing || c->socket_fd == -1)
 		return;
 
 	// liberty has msg_{reader,writer} already, but they use 8-byte lengths.
@@ -4180,12 +4182,18 @@ relay_send (struct client *c)
 	 || (frame_len = c->write_buffer.len - frame_len_pos - 4) > UINT32_MAX)
 	{
 		print_error ("serialization failed, killing client");
-		client_kill (c);
-		return;
+
+		// We can't kill the client immediately,
+		// because more relay_send() calls may follow.
+		c->write_buffer.len = frame_len_pos;
+		c->closing = true;
+	}
+	else
+	{
+		uint32_t len = htonl (frame_len);
+		memcpy (c->write_buffer.str + frame_len_pos, &len, sizeof len);
 	}
 
-	uint32_t len = htonl (frame_len);
-	memcpy (c->write_buffer.str + frame_len_pos, &len, sizeof len);
 	client_update_poller (c, NULL);
 }
 
@@ -15604,28 +15612,31 @@ client_process_message (struct client *c,
 		return true;
 	}
 
+	bool acknowledge = true;
 	switch (m->data.command)
 	{
 	case RELAY_COMMAND_HELLO:
+		c->initialized = true;
 		if (m->data.hello.version != RELAY_VERSION)
 		{
-			// TODO: This should send back an error message and shut down.
 			log_global_error (c->ctx,
 				"Protocol version mismatch, killing client");
-			return false;
+			relay_prepare_error (c->ctx,
+				m->command_seq, "Protocol version mismatch");
+			relay_send (c);
+
+			c->closing = true;
+			return true;
 		}
-		c->initialized = true;
 		client_resync (c);
 		break;
 	case RELAY_COMMAND_PING:
-		relay_prepare_response (c->ctx, m->command_seq)
-			->data.command = RELAY_COMMAND_PING;
-		relay_send (c);
 		break;
 	case RELAY_COMMAND_ACTIVE:
 		reset_autoaway (c->ctx);
 		break;
 	case RELAY_COMMAND_BUFFER_COMPLETE:
+		acknowledge = false;
 		client_process_buffer_complete (c, m->command_seq, buffer,
 			&m->data.buffer_complete);
 		break;
@@ -15639,13 +15650,21 @@ client_process_message (struct client *c,
 		buffer_toggle_unimportant (c->ctx, buffer);
 		break;
 	case RELAY_COMMAND_BUFFER_LOG:
+		acknowledge = false;
 		client_process_buffer_log (c, m->command_seq, buffer);
 		break;
 	default:
+		acknowledge = false;
 		log_global_debug (c->ctx, "Unhandled client command");
 		relay_prepare_error (c->ctx, m->command_seq, "Unknown command");
 		relay_send (c);
 	}
+	if (acknowledge)
+	{
+		relay_prepare_response (c->ctx, m->command_seq)
+			->data.command = m->data.command;
+		relay_send (c);
+	}
 	return true;
 }
 
@@ -15667,7 +15686,7 @@ client_process_buffer (struct client *c)
 			break;
 
 		struct relay_command_message m = {};
-		bool ok = client_process_message (c, &r, &m);
+		bool ok = c->closing || client_process_message (c, &r, &m);
 		relay_command_message_free (&m);
 		if (!ok)
 			return false;
@@ -15739,7 +15758,11 @@ on_client_ready (const struct pollfd *pfd, void *user_data)
 {
 	struct client *c = user_data;
 	if (client_try_read (c) && client_try_write (c))
+	{
 		client_update_poller (c, pfd);
+		if (c->closing && !c->write_buffer.len)
+			client_kill (c);
+	}
 }
 
 // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -