WIP: Make the relay acknowledge received commands

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

Relay events have been reordered to improve forward compatibility.

As can be observed, even the smallest protocol change has
great consequences.

WIP:
 - xC: fix connection killing

1 files changed, 29 insertions, 10 deletions

diff --git a/xC.c b/xC.c
index 73ddd12..60da463 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 shutdown;                      ///< Shutting down
 
 	struct poller_fd socket_event;      ///< The socket can be read/written to
 };
@@ -4168,9 +4169,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->shutdown || c->socket_fd == -1)
 		return;
 
 	// liberty has msg_{reader,writer} already, but they use 8-byte lengths.
@@ -4180,7 +4179,10 @@ 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);
+		// FIXME: This must not be done immediately!
+		//client_kill (c);
+		// TODO: Perhaps set an idle task that collects shutdown clients.
+		c->shutdown = true;
 		return;
 	}
 
@@ -15716,28 +15718,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->shutdown = 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;
@@ -15751,13 +15756,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;
 }
 
@@ -15851,7 +15864,13 @@ 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);
+
+		// There must be something in the write buffer if you request shutdown.
+		if (c->shutdown && !c->write_buffer.len)
+			client_kill (c);
+	}
 }
 
 // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -