From bcfbd400035c8a7bb31eafc8146cddbaa645d82b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C5=99emysl=20Eric=20Janouch?= Date: Sun, 26 Jul 2026 16:03:49 +0200 Subject: [PATCH] xC/xP: implement session resumption Mobile clients pretty much need this, because they keep interrupting their connections to the relay. --- NEWS | 9 ++ common.c | 5 + xC.c | 401 +++++++++++++++++++++++++++++++++++++++++------ xC.lxdr | 20 ++- xM/main.swift | 3 +- xP/public/xP.css | 2 +- xP/public/xP.js | 116 ++++++++++---- 7 files changed, 470 insertions(+), 86 deletions(-) diff --git a/NEWS b/NEWS index 828b742..4ff4a9b 100644 --- a/NEWS +++ b/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" diff --git a/common.c b/common.c index 2362c8b..c6b5f02 100644 --- a/common.c +++ b/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 diff --git a/xC.c b/xC.c index f1c5f51..2f4fbb7 100644 --- a/xC.c +++ b/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. diff --git a/xC.lxdr b/xC.lxdr index eba914f..d9dc018 100644 --- a/xC.lxdr +++ b/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<>; diff --git a/xM/main.swift b/xM/main.swift index 39b66c5..be8ca0e 100644 --- a/xM/main.swift +++ b/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) } diff --git a/xP/public/xP.css b/xP/public/xP.css index bf5a66e..f6c6734 100644 --- a/xP/public/xP.css +++ b/xP/public/xP.css @@ -70,7 +70,7 @@ body { visibility: hidden; } -.toolbar { +.toolbar, .title div { display: flex; align-items: baseline; margin-right: -.3em; diff --git a/xP/public/xP.js b/xP/public/xP.js index 6806081..7199648 100644 --- a/xP/public/xP.js +++ b/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), ]),