xC/xP: implement session resumption

Mobile clients pretty much need this, because they keep interrupting
their connections to the relay.

NEWS

@@ -6,8 +6,17 @@ Unreleased
 
  * xC: added support for Lua 5.5
 
+ * xC: relay sessions can now be resumed after a dropped connection (adjustable
+   with general.relay_resume_timeout and general.relay_resume_log_limit);
+   dead links are detected with periodic pings
+
+ * xP: added a button to reconnect and resume the session if possible;
+   page reloads start a new session as before
+
  * prime.lua: prevented a DoS opportunity on excessively large numbers
 
+ * Bumped relay protocol version
+
 
 2.2.0 (2026-01-10) "Just Doing the Needful"
 

common.c

@@ -110,6 +110,11 @@ tokenize_host_port (char *address, const char **port)
 
 // --- To be moved to liberty --------------------------------------------------
 
+#define LIST_FOR_EACH_REVERSED(type, iter, tail)                               \
+	for (type *iter = (tail), *prev;                                           \
+		(iter && (prev = iter->prev)) || iter;                                 \
+		iter = prev)
+
 // FIXME: in xssl_get_error() we rely on error reasons never being NULL (i.e.,
 //   all loaded), which isn't very robust.
 // TODO: check all places where this is used and see if we couldn't gain better

xC.c

@@ -1572,32 +1572,56 @@ REF_COUNTABLE_METHODS (buffer)
 
 // ~~~ Relay ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
+#define RELAY_RESUME_MAX_SESSIONS 8
+#define RELAY_PING_INTERVAL_MS (60 * 1000)
+#define RELAY_PING_TIMEOUT_MS (59 * 1000)
+
 struct client
 {
 	LIST_HEADER (struct client)
 	struct app_context *ctx;            ///< Application context
 
-	// TODO: Convert this all to TLS, and only TLS, with required client cert.
+	// TODO: Convert this all to TLS, and only TLS, with a required client cert.
 	//   That means replacing plumbing functions with the /other/ set from xD.
 
-	int socket_fd;                      ///< The TCP socket
+	// |   R e t a i n e d   | Buffered |
+	// +----------+----------+----------+   The write buffer serves a second
+	// | Event #3 | Event #4 | Event #5 |   purpose as a session replay log.
+	// +--------^-+----------+----------+
+	//          '- start_seq ^- sent_offset
+
+	int socket_fd;                      ///< The TCP socket, or -1 if detached
 	struct str read_buffer;             ///< Unprocessed input
-	struct str write_buffer;            ///< Output yet to be sent out
+	struct str write_log;               ///< Retained outgoing event log
+	size_t sent_offset;                 ///< How far into the log we've written
+	uint32_t write_log_start_seq;       ///< The first retained event
 
 	uint32_t event_seq;                 ///< Outgoing message counter
+	uint8_t session_id[16];             ///< Session identity
 	bool initialized;                   ///< Initial sync took place
-	bool closing;                       ///< We're closing the connection
+	bool done;                          ///< We're closing the session
 
 	struct poller_fd socket_event;      ///< The socket can be read/written to
+	struct poller_timer ping_tmr;       ///< We should send a ping
+	struct poller_timer timeout_tmr;    ///< Ping response is overdue
+	struct poller_timer expire_tmr;     ///< Detached session should be dropped
 };
 
+static void client_init_timers (struct client *self);
+static void client_kill (struct client *c);
+
 static struct client *
-client_new (void)
+client_new (struct app_context *ctx)
 {
 	struct client *self = xcalloc (1, sizeof *self);
+	self->ctx = ctx;
 	self->socket_fd = -1;
 	self->read_buffer = str_make ();
-	self->write_buffer = str_make ();
+	self->write_log = str_make ();
+	hard_assert (random_bytes (self->session_id, sizeof self->session_id,
+		NULL));
+
+	client_init_timers (self);
 	return self;
 }
 
@@ -1607,13 +1631,15 @@ client_destroy (struct client *self)
 	if (!soft_assert (self->socket_fd == -1))
 		xclose (self->socket_fd);
 
+	poller_timer_reset (&self->ping_tmr);
+	poller_timer_reset (&self->timeout_tmr);
+	poller_timer_reset (&self->expire_tmr);
+
 	str_free (&self->read_buffer);
-	str_free (&self->write_buffer);
+	str_free (&self->write_log);
 	free (self);
 }
 
-static void client_kill (struct client *c);
-
 // ~~~ Server ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 // The only real purpose of this is to abstract away TLS
@@ -2085,6 +2111,9 @@ struct app_context
 
 	int relay_fd;                       ///< Listening socket FD
 	struct client *clients;             ///< Our relay clients
+	struct client *clients_tail;        ///< End of the clients list
+	unsigned relay_resume_timeout;      ///< Detached session TTL in seconds
+	size_t relay_resume_log_limit;      ///< Max retained event log size
 
 	/// A single message buffer to prepare all outcoming messages within
 	struct relay_event_message relay_message;
@@ -2259,6 +2288,8 @@ on_config_show_all_prefixes_change (struct config_item *item)
 }
 
 static void on_config_relay_bind_change (struct config_item *item);
+static void on_config_relay_resume_timeout_change (struct config_item *item);
+static void on_config_relay_resume_log_limit_change (struct config_item *item);
 static void on_config_backlog_limit_change (struct config_item *item);
 static void on_config_theme_change (struct config_item *item);
 static void on_config_logging_change (struct config_item *item);
@@ -2454,11 +2485,25 @@ static const struct config_schema g_config_general[] =
 	  .comment   = "Plugins to automatically load on start",
 	  .type      = CONFIG_ITEM_STRING_ARRAY,
 	  .validate  = config_validate_nonjunk_string },
+
+	// Relay:
 	{ .name      = "relay_bind",
 	  .comment   = "Address to bind to for a user interface relay point",
 	  .type      = CONFIG_ITEM_STRING,
 	  .validate  = config_validate_nonjunk_string,
 	  .on_change = on_config_relay_bind_change },
+	{ .name      = "relay_resume_log_limit",
+	  .comment   = "Maximum retained relay event log size per session in bytes",
+	  .type      = CONFIG_ITEM_INTEGER,
+	  .validate  = config_validate_nonnegative,
+	  .default_  = "1048576",
+	  .on_change = on_config_relay_resume_log_limit_change },
+	{ .name      = "relay_resume_timeout",
+	  .comment   = "Seconds to retain relay sessions for connection resumption",
+	  .type      = CONFIG_ITEM_INTEGER,
+	  .validate  = config_validate_nonnegative,
+	  .default_  = "86400",
+	  .on_change = on_config_relay_resume_timeout_change },
 
 	// Buffer history:
 	{ .name      = "backlog_limit",
@@ -3910,26 +3955,64 @@ formatter_flush (struct formatter *self, FILE *stream, int flush_opts)
 
 // --- Relay output ------------------------------------------------------------
 
+#define CLIENT_DETACHED(c) ((c)->socket_fd == -1)
+
+static void
+client_close_socket (struct client *c)
+{
+	if (CLIENT_DETACHED (c))
+		return;
+
+	poller_fd_reset (&c->socket_event);
+	xclose (c->socket_fd);
+	c->socket_fd = -1;
+	poller_timer_reset (&c->ping_tmr);
+	poller_timer_reset (&c->timeout_tmr);
+}
+
 static void
 client_kill (struct client *c)
 {
 	struct app_context *ctx = c->ctx;
-	poller_fd_reset (&c->socket_event);
-	xclose (c->socket_fd);
-	c->socket_fd = -1;
-
-	LIST_UNLINK (ctx->clients, c);
+	client_close_socket (c);
+	LIST_UNLINK_WITH_TAIL (ctx->clients, ctx->clients_tail, c);
 	client_destroy (c);
 }
 
+static void
+client_detach (struct client *c)
+{
+	hard_assert (!CLIENT_DETACHED (c));
+
+	struct app_context *ctx = c->ctx;
+	if (!c->initialized || c->done || !ctx->relay_resume_timeout)
+	{
+		client_kill (c);
+		return;
+	}
+
+	client_close_socket (c);
+	poller_timer_set (&c->expire_tmr, ctx->relay_resume_timeout * 1000);
+
+	// Registration is append-ordered, so this drops the oldest sessions
+	// beyond the limit.  Those may include this client.
+	unsigned detached = 0;
+	LIST_FOR_EACH_REVERSED (struct client, iter, ctx->clients_tail)
+		if (CLIENT_DETACHED (iter) && ++detached > RELAY_RESUME_MAX_SESSIONS)
+			client_kill (iter);
+}
+
 static void
 client_update_poller (struct client *c, const struct pollfd *pfd)
 {
-	// In case of closing without any data in the write buffer,
+	if (CLIENT_DETACHED (c))
+		return;
+
+	// In case of closing without any data left to write,
 	// 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 || c->closing)
+	if (c->sent_offset < c->write_log.len || c->done)
 		new_events |= POLLOUT;
 
 	hard_assert (new_events != 0);
@@ -3939,34 +4022,112 @@ client_update_poller (struct client *c, const struct pollfd *pfd)
 
 // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 
+// XXX: This write log reprocessing a bit annoying, but the alternative seems
+//   to be keeping the same data in an extra data structure.
+
+static bool
+client_write_log_frame_at (const struct client *c,
+	size_t offset, uint32_t *frame_len)
+{
+	if (offset + 4 > c->write_log.len)
+		return false;
+
+	uint32_t len;
+	memcpy (&len, c->write_log.str + offset, sizeof len);
+	*frame_len = 4 + ntohl (len);
+	return offset + *frame_len <= c->write_log.len;
+}
+
+static bool
+client_write_log_offset_for_seq
+	(const struct client *c, uint32_t seq, size_t *offset_out)
+{
+	uint32_t retained = c->event_seq - c->write_log_start_seq;
+	uint32_t skip = seq - c->write_log_start_seq;
+	if (skip > retained)
+		return false;
+
+	size_t offset = 0;
+	while (skip--)
+	{
+		uint32_t frame_len = 0;
+		if (!client_write_log_frame_at (c, offset, &frame_len))
+			return false;
+		offset += frame_len;
+	}
+	*offset_out = offset;
+	return true;
+}
+
+static bool
+client_trim_write_log (struct client *c)
+{
+	size_t limit = c->ctx->relay_resume_log_limit;
+	if (!limit || c->write_log.len <= limit)
+		return true;
+
+	size_t offset = 0;
+	uint32_t dropped = 0;
+	while (c->write_log.len - offset > limit)
+	{
+		uint32_t frame_len = 0;
+		if (!client_write_log_frame_at (c, offset, &frame_len))
+			break;
+
+		size_t next = offset + frame_len;
+		if (next > c->sent_offset)
+			break;
+
+		offset = next;
+		dropped++;
+	}
+	if (offset)
+	{
+		str_remove_slice (&c->write_log, 0, offset);
+		c->sent_offset -= offset;
+		c->write_log_start_seq += dropped;
+	}
+	return c->write_log.len <= limit;
+}
+
 static void
 relay_send (struct client *c)
 {
 	struct relay_event_message *m = &c->ctx->relay_message;
-	m->event_seq = c->event_seq++;
-	if (!c->initialized || c->closing || c->socket_fd == -1)
+	if (!c->initialized || c->done)
 		return;
+	m->event_seq = c->event_seq++;
 
 	// liberty has msg_{reader,writer} already, but they use 8-byte lengths.
-	size_t frame_len_pos = c->write_buffer.len, frame_len = 0;
-	str_pack_u32 (&c->write_buffer, 0);
-	if (!relay_event_message_serialize (m, &c->write_buffer)
-	 || (frame_len = c->write_buffer.len - frame_len_pos - 4) > UINT32_MAX)
+	size_t frame_len_pos = c->write_log.len, frame_len = 0;
+	str_pack_u32 (&c->write_log, 0);
+	if (!relay_event_message_serialize (m, &c->write_log)
+	 || (frame_len = c->write_log.len - frame_len_pos - 4) > UINT32_MAX)
 	{
 		print_error ("serialization failed, killing client");
 
 		// We can't kill the client immediately,
 		// because more relay_send() calls may follow.
-		c->write_buffer.len = frame_len_pos;
-		c->closing = true;
+		c->write_log.len = frame_len_pos;
+		c->event_seq--;
+		c->done = true;
 	}
 	else
 	{
 		uint32_t len = htonl (frame_len);
-		memcpy (c->write_buffer.str + frame_len_pos, &len, sizeof len);
+		memcpy (c->write_log.str + frame_len_pos, &len, sizeof len);
+
+		// We may eventually trim too much for the client to resume.
+		//
+		// We might instead trim only within the PING_RESPONSE handler,
+		// but that might not be soon enough to obey the log limit.
+		if ((c->done = !client_trim_write_log (c) && CLIENT_DETACHED (c)))
+			print_debug ("write log overflow, killing detached client");
 	}
 
 	client_update_poller (c, NULL);
+	if (c->done && CLIENT_DETACHED (c))
+		poller_timer_set (&c->expire_tmr, 0);
 }
 
 static void
@@ -14995,6 +15156,52 @@ init_poller_events (struct app_context *ctx)
 
 // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
 
+static void
+on_client_ping (void *user_data)
+{
+	struct client *c = user_data;
+	if (c->done)
+		return;
+
+	relay_prepare_ping (c->ctx);
+	relay_send (c);
+
+	poller_timer_set (&c->ping_tmr, RELAY_PING_INTERVAL_MS);
+	poller_timer_set (&c->timeout_tmr, RELAY_PING_TIMEOUT_MS);
+}
+
+static void
+on_client_timeout (void *user_data)
+{
+	struct client *c = user_data;
+	log_global_debug (c->ctx, "Relay client ping timeout");
+	client_detach (c);
+}
+
+static void
+on_client_expire (void *user_data)
+{
+	client_kill (user_data);
+}
+
+static void
+client_init_timers (struct client *self)
+{
+	self->ping_tmr = poller_timer_make (&self->ctx->poller);
+	self->ping_tmr.dispatcher = on_client_ping;
+	self->ping_tmr.user_data = self;
+
+	self->timeout_tmr = poller_timer_make (&self->ctx->poller);
+	self->timeout_tmr.dispatcher = on_client_timeout;
+	self->timeout_tmr.user_data = self;
+
+	self->expire_tmr = poller_timer_make (&self->ctx->poller);
+	self->expire_tmr.dispatcher = on_client_expire;
+	self->expire_tmr.user_data = self;
+}
+
+// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
+
 static void
 client_resync_buffer_input (struct client *c, struct buffer *buffer)
 {
@@ -15038,6 +15245,20 @@ client_resync (struct client *c)
 	relay_send (c);
 }
 
+static struct client *
+client_find_by_session_id (struct app_context *ctx, const uint8_t *token,
+	uint32_t token_len, struct client *except)
+{
+	if (token_len != sizeof except->session_id)
+		return NULL;
+
+	LIST_FOR_EACH (struct client, c, ctx->clients)
+		if (c != except && c->initialized && !c->done
+		 && !memcmp (c->session_id, token, token_len))
+			return c;
+	return NULL;
+}
+
 static const char *
 client_message_buffer_name (const struct relay_command_message *m)
 {
@@ -15058,6 +15279,71 @@ client_message_buffer_name (const struct relay_command_message *m)
 	}
 }
 
+static void
+client_process_hello (struct client *c, uint32_t seq,
+	struct relay_command_data_hello *req)
+{
+	if (req->version != RELAY_VERSION)
+	{
+		c->initialized = true;
+
+		log_global_error (c->ctx,
+			"Protocol version mismatch, killing client");
+		relay_prepare_error (c->ctx, seq, "Protocol version mismatch");
+		relay_send (c);
+
+		c->done = true;
+		return;
+	}
+	if (c->initialized)
+	{
+		relay_prepare_error (c->ctx, seq, "You've already introduced yourself");
+		relay_send (c);
+		return;
+	}
+
+	bool resumed = false;
+	struct client *old = client_find_by_session_id (c->ctx,
+		req->session_id, req->session_id_len, c);
+	if (old)
+	{
+		size_t offset;
+		if (client_write_log_offset_for_seq (old, req->event_seq, &offset))
+		{
+			struct str tmp = c->write_log;
+			c->write_log = old->write_log;
+			old->write_log = tmp;
+
+			c->sent_offset = offset;
+			c->write_log_start_seq = old->write_log_start_seq;
+			c->event_seq = old->event_seq;
+			memcpy (c->session_id, old->session_id, sizeof c->session_id);
+
+			resumed = true;
+		}
+		client_kill (old);
+	}
+
+	c->initialized = true;
+
+	struct relay_event_data_response *e = relay_prepare_response (c->ctx, seq);
+	e->data.command = RELAY_COMMAND_HELLO;
+	e->data.hello.session_id_len = sizeof c->session_id;
+	e->data.hello.session_id = xmalloc (sizeof c->session_id);
+	memcpy (e->data.hello.session_id, c->session_id, sizeof c->session_id);
+	relay_send (c);
+
+	// Otherwise the write log gets replayed as appropriate.
+	if (!resumed)
+		client_resync (c);
+
+	if (!c->done && !CLIENT_DETACHED (c))
+	{
+		poller_timer_set (&c->ping_tmr, RELAY_PING_INTERVAL_MS);
+		poller_timer_reset (&c->timeout_tmr);
+	}
+}
+
 static void
 client_process_buffer_complete (struct client *c, uint32_t seq,
 	struct buffer *buffer, struct relay_command_data_buffer_complete *req)
@@ -15164,22 +15450,15 @@ client_process_message (struct client *c,
 	switch (m->data.command)
 	{
 	case RELAY_COMMAND_HELLO:
-		c->initialized = true;
-		if (m->data.hello.version != RELAY_VERSION)
-		{
-			log_global_error (c->ctx,
-				"Protocol version mismatch, killing client");
-			relay_prepare_error (c->ctx,
-				m->command_seq, "Protocol version mismatch");
-			relay_send (c);
-
-			c->closing = true;
-			return true;
-		}
-		client_resync (c);
+		acknowledge = false;
+		client_process_hello (c, m->command_seq, &m->data.hello);
 		break;
 	case RELAY_COMMAND_PING:
 		break;
+	case RELAY_COMMAND_PING_RESPONSE:
+		// TODO(p): We should check m->data.ping_response.event_seq
+		poller_timer_reset (&c->timeout_tmr);
+		break;
 	case RELAY_COMMAND_ACTIVE:
 		reset_autoaway (c->ctx);
 		break;
@@ -15234,7 +15513,7 @@ client_process_buffer (struct client *c)
 			break;
 
 		struct relay_command_message m = {};
-		bool ok = c->closing || client_process_message (c, &r, &m);
+		bool ok = c->done || client_process_message (c, &r, &m);
 		relay_command_message_free (&m);
 		if (!ok)
 			return false;
@@ -15272,22 +15551,23 @@ client_try_read (struct client *c)
 			"#s: #s: #l", __func__, "read", strerror (errno));
 	}
 
-	client_kill (c);
+	client_detach (c);
 	return false;
 }
 
 static bool
 client_try_write (struct client *c)
 {
-	struct str *buf = &c->write_buffer;
+	struct str *buf = &c->write_log;
 	ssize_t n_written;
 
-	while (buf->len)
+	while (c->sent_offset < buf->len)
 	{
-		n_written = write (c->socket_fd, buf->str, buf->len);
+		n_written = write (c->socket_fd, buf->str + c->sent_offset,
+			buf->len - c->sent_offset);
 		if (n_written >= 0)
 		{
-			str_remove_slice (buf, 0, n_written);
+			c->sent_offset += n_written;
 			continue;
 		}
 		if (errno == EAGAIN || errno == EINTR)
@@ -15295,7 +15575,7 @@ client_try_write (struct client *c)
 
 		log_global_debug (c->ctx,
 			"#s: #s: #l", __func__, "write", strerror (errno));
-		client_kill (c);
+		client_detach (c);
 		return false;
 	}
 	return true;
@@ -15308,7 +15588,7 @@ on_client_ready (const struct pollfd *pfd, void *user_data)
 	if (client_try_read (c) && client_try_write (c))
 	{
 		client_update_poller (c, pfd);
-		if (c->closing && !c->write_buffer.len)
+		if (c->done && c->sent_offset >= c->write_log.len)
 			client_kill (c);
 	}
 }
@@ -15350,10 +15630,12 @@ relay_try_fetch_client (struct app_context *ctx, int listen_fd)
 	soft_assert (setsockopt (fd, IPPROTO_TCP, TCP_NODELAY,
 		&yes, sizeof yes) != -1);
 
-	struct client *c = client_new ();
-	c->ctx = ctx;
+	struct client *c = client_new (ctx);
 	c->socket_fd = fd;
-	LIST_PREPEND (ctx->clients, c);
+	LIST_APPEND_WITH_TAIL (ctx->clients, ctx->clients_tail, c);
+
+	// We treat the HELLO command like PING_RESPONSE.
+	poller_timer_set (&c->timeout_tmr, RELAY_PING_TIMEOUT_MS);
 
 	c->socket_event = poller_fd_make (&c->ctx->poller, c->socket_fd);
 	c->socket_event.dispatcher = (poller_fd_fn) on_client_ready;
@@ -15478,6 +15760,27 @@ on_config_relay_bind_change (struct config_item *item)
 	free (address);
 }
 
+static void
+on_config_relay_resume_timeout_change (struct config_item *item)
+{
+	struct app_context *ctx = item->user_data;
+	ctx->relay_resume_timeout = MIN (item->value.integer, INT_MAX / 1000);
+
+	if (!ctx->relay_resume_timeout)
+	{
+		LIST_FOR_EACH (struct client, c, ctx->clients)
+			if (CLIENT_DETACHED (c))
+				client_kill (c);
+	}
+}
+
+static void
+on_config_relay_resume_log_limit_change (struct config_item *item)
+{
+	struct app_context *ctx = item->user_data;
+	ctx->relay_resume_log_limit = (size_t) item->value.integer;
+}
+
 // --- Tests -------------------------------------------------------------------
 
 // The application is quite monolithic and can only be partially unit-tested.

xC.lxdr

@@ -1,5 +1,5 @@
 // Backwards-compatible protocol version.
-const VERSION = 2;
+const VERSION = 3;
 
 // From the frontend to the relay.
 // All commands receive either an Event.RESPONSE, or an Event.ERROR.
@@ -13,16 +13,20 @@ struct CommandMessage {
 		BUFFER_ACTIVATE,
 		BUFFER_INPUT,
 		BUFFER_TOGGLE_UNIMPORTANT,
-		PING_RESPONSE,
 		PING,
+		PING_RESPONSE,
 		BUFFER_COMPLETE,
 		BUFFER_LOG,
 	} command) {
-	// If the version check succeeds, the client will receive
-	// an initial stream of SERVER_UPDATE, BUFFER_UPDATE, BUFFER_STATS,
-	// BUFFER_LINE, and finally a BUFFER_ACTIVATE message.
+	// When a session is resumed, the RESPONSE is preceded by an event log
+	// replay starting at event_seq.
 	case HELLO:
 		u32 version;
+		// Empty to start a new session, otherwise the session to resume.
+		u8 session_id<>;
+		// The event_seq to continue from;
+		// a new session is started when that point is no longer available.
+		u32 event_seq;
 	case ACTIVE:
 		void;
 	case BUFFER_ACTIVATE:
@@ -74,6 +78,12 @@ struct EventMessage {
 	case RESPONSE:
 		u32 command_seq;
 		union ResponseData switch (Command command) {
+		case HELLO:
+			// The session to present on the next connection.  If it differs
+			// from the command, the session is new, and a full resync follows:
+			// an initial stream of SERVER_UPDATE, BUFFER_UPDATE, BUFFER_STATS,
+			// BUFFER_LINE, and finally a BUFFER_ACTIVATE message.
+			u8 session_id<>;
 		case BUFFER_COMPLETE:
 			u32 start;
 			string completions<>;

xM/main.swift

@@ -697,7 +697,8 @@ func refreshBuffer(b: Buffer) {
 // --- Event processing --------------------------------------------------------
 
 relayRPC.onConnected = {
-	let hello = RelayCommandDataHello(version: UInt32(relayVersion))
+	let hello = RelayCommandDataHello(
+		version: UInt32(relayVersion), sessionId: [], eventSeq: 0)
 	relayRPC.send(data: hello)
 }
 

xP/public/xP.css

@@ -70,7 +70,7 @@ body {
 	visibility: hidden;
 }
 
-.toolbar {
+.toolbar, .title div {
 	display: flex;
 	align-items: baseline;
 	margin-right: -.3em;

xP/public/xP.js

@@ -9,14 +9,40 @@ class RelayRPC extends EventTarget {
 		super()
 		this.url = url
 		this.commandSeq = 0
+		this.promised = {}
+
+		// Session to resume after a lost connection, and our progress in it.
+		this.sessionId = ''
+		this.lastEventSeq = 0
+		this.connecting = false
 	}
 
+	// FIXME: The m.redraws are contamination for properties
+	//   that should have an event: this.connecting, this.busy
+
+	// When we can't resume our session, a 'reset' event will ask for the model
+	// to be reinitialized.
 	connect() {
-		// We can't close the connection immediately, as that queues a task.
+		if (this.connecting)
+			return
+		this.connecting = true
+		m.redraw()
+
+		// We can't reopen the socket immediately, as closing queues a task.
+		if (this.ws === undefined)
+			this._connect()
+		else {
+			this.ws.addEventListener('close',
+				() => this._connect(), {once: true})
+			this.ws.close()
+		}
+	}
+
+	_connect() {
 		if (this.ws !== undefined)
 			throw "Already connecting or connected"
 
-		return new Promise((resolve, reject) => {
+		new Promise((resolve, reject) => {
 			let ws = this.ws = new WebSocket(this.url)
 			ws.onopen = event => {
 				this._initialize()
@@ -27,6 +53,20 @@ class RelayRPC extends EventTarget {
 				this.ws = undefined
 				reject()
 			}
+		}).then(() => {
+			return this.send({
+				command: 'Hello',
+				version: Relay.version,
+				sessionId: this.sessionId,
+				// Continue right after the last event we have processed.
+				eventSeq: this.sessionId ? (this.lastEventSeq + 1) >>> 0 : 0,
+			})
+		}).catch(error => {
+			// Failures manifest as a closed connection.
+			console.error(error)
+		}).finally(() => {
+			this.connecting = false
+			m.redraw()
 		})
 	}
 
@@ -42,8 +82,6 @@ class RelayRPC extends EventTarget {
 		this.ws.onclose = event => {
 			let message = "Connection closed: " +
 				event.reason + " (" + event.code + ")"
-			for (const seq in this.promised)
-				this.promised[seq].reject(message)
 
 			this.ws = undefined
 			this.dispatchEvent(new CustomEvent('close', {
@@ -52,8 +90,6 @@ class RelayRPC extends EventTarget {
 
 			// Now connect() can be called again.
 		}
-
-		this.promised = {}
 	}
 
 	_process(data) {
@@ -66,6 +102,8 @@ class RelayRPC extends EventTarget {
 	}
 
 	_processOne(message) {
+		this.lastEventSeq = message.eventSeq
+
 		let e = message.data
 		let p
 		switch (e.event) {
@@ -76,6 +114,9 @@ class RelayRPC extends EventTarget {
 				p.reject(e.error)
 			break
 		case Relay.Event.Response:
+			if (e.data.command === Relay.Command.Hello)
+				this._adoptSession(base64Encode(e.data.sessionId), e.commandSeq)
+
 			if ((p = this.promised[e.commandSeq]) === undefined)
 				console.error("Unawaited response")
 			else if (p !== true)
@@ -90,15 +131,33 @@ class RelayRPC extends EventTarget {
 		delete this.promised[e.commandSeq]
 		for (const seq in this.promised) {
 			// We don't particularly care about wraparound issues.
-			if (seq >= e.commandSeq)
-				continue
-
-			this.promised[seq].reject("No response")
-			delete this.promised[seq]
+			if (seq < e.commandSeq)
+				this._abandon(seq, "No response")
 		}
 		m.redraw()
 	}
 
+	_adoptSession(sessionId, seq) {
+		if (sessionId === this.sessionId)
+			return
+
+		// Nothing else can be answered from within the new session.
+		for (const pending in this.promised)
+			if (pending != seq)
+				this._abandon(pending, "Session lost")
+
+		this.sessionId = sessionId
+		this.dispatchEvent(new CustomEvent('reset'))
+	}
+
+	// Stop waiting for a reply that is never going to arrive.
+	_abandon(seq, message) {
+		const p = this.promised[seq]
+		if (p !== true)
+			p.reject(message)
+		delete this.promised[seq]
+	}
+
 	get busy() {
 		for (const seq in this.promised)
 			return true
@@ -138,6 +197,8 @@ class RelayRPC extends EventTarget {
 function utf8Encode(s) { return new TextEncoder().encode(s) }
 function utf8Decode(s) { return new TextDecoder().decode(s) }
 
+function base64Encode(bytes) { return btoa(String.fromCharCode(...bytes)) }
+
 function hasShortcutModifiers(event) {
 	return (event.altKey || event.escapePrefix) &&
 		!event.metaKey && !event.ctrlKey
@@ -202,6 +263,16 @@ let bufferAutoscroll = true
 
 let servers = new Map()
 
+rpc.addEventListener('reset', event => {
+	buffers.clear()
+	bufferLast = undefined
+	bufferCurrent = undefined
+	bufferLog = undefined
+	bufferAutoscroll = true
+
+	servers.clear()
+})
+
 let lastActive = undefined
 
 function notifyActive() {
@@ -259,23 +330,7 @@ function bufferToggleLog() {
 	})
 }
 
-let connecting = true
-rpc.connect().then(result => {
-	buffers.clear()
-	bufferLast = undefined
-	bufferCurrent = undefined
-	bufferLog = undefined
-	bufferAutoscroll = true
-
-	servers.clear()
-
-	rpc.send({command: 'Hello', version: Relay.version})
-	connecting = false
-	m.redraw()
-}).catch(error => {
-	connecting = false
-	m.redraw()
-})
+rpc.connect()
 
 rpc.addEventListener('close', event => {
 	m.redraw()
@@ -1115,7 +1170,7 @@ let Input = {
 let Main = {
 	view: vnode => {
 		let overlay = undefined
-		if (connecting)
+		if (rpc.connecting)
 			overlay = m('.overlay', {}, "Connecting...")
 		else if (rpc.ws === undefined)
 			overlay = m('.overlay', {}, [
@@ -1126,11 +1181,12 @@ let Main = {
 		return m('.xP', {}, [
 			overlay,
 			m('.title', {}, [
-				m('span', [
+				m('div', [
 					// Midline Horizontal Ellipsis, No-Break Space
 					m('span', {class: rpc.busy ? undefined : 'invisible'},
 						`\u22EF\u00A0`),
 					m('b', {}, `xP`),
+					m('button', {onclick: () => rpc.connect()}, '⟳'),
 				]),
 				m(Topic),
 			]),