ssb: Attempt to show EBT replication progress in the connections tab, mainly to help track down bugs, which there are.

apps/ssb.json

@@ -1,5 +1,5 @@
 {
 	"type": "tildefriends-app",
 	"emoji": "🦀",
-	"previous": "&Qae0CxJGEH7OspuapuXX/GurmV+VBtQiMhHRmCBKCwU=.sha256"
+	"previous": "&wOd/+1l5wpywBlfxC1Lm0i+HhYidrgSfrn9LRX7qy2w=.sha256"
 }

apps/ssb/tf-tab-connections.js

@@ -103,6 +103,21 @@ class TfTabConnectionsElement extends LitElement {
 		</div>`;
 	}
 
+	render_progress(name, value, max) {
+		if (max && value != max) {
+			return html`
+				<div class="w3-theme-d1 w3-small">
+					<div
+						class="w3-container w3-theme-l1"
+						style="width: ${Math.floor((100.0 * value) / max)}%; text-wrap: nowrap"
+					>
+						${name} ${value} / ${max} (${Math.round((100.0 * value) / max)}%)
+					</div>
+				</div>
+			`;
+		}
+	}
+
 	render_broadcast(connection) {
 		let self = this;
 		return html`
@@ -154,6 +169,16 @@ class TfTabConnectionsElement extends LitElement {
 				: undefined}
 			${connection.flags.one_shot ? '🔃' : undefined}
 			<tf-user id=${connection.id} .users=${this.users}></tf-user>
+			${this.render_progress(
+				'recv',
+				connection.progress.in.total - connection.progress.in.current,
+				connection.progress.in.total
+			)}
+			${this.render_progress(
+				'send',
+				connection.progress.out.total - connection.progress.out.current,
+				connection.progress.out.total
+			)}
 			${connection.tunnel !== undefined
 				? '🚇'
 				: html`(${connection.host}:${connection.port})`}

src/ssb.ebt.c

@@ -30,6 +30,9 @@ typedef struct _tf_ssb_ebt_t
 	int entries_count;
 
 	int send_clock_pending;
+
+	int max_in;
+	int max_out;
 } tf_ssb_ebt_t;
 
 tf_ssb_ebt_t* tf_ssb_ebt_create(tf_ssb_connection_t* connection)
@@ -56,6 +59,25 @@ static int _ebt_entry_compare(const void* a, const void* b)
 	return strcmp(id, entry->id);
 }
 
+static void _ebt_count_messages(tf_ssb_ebt_t* ebt, int* in, int* out)
+{
+	for (int i = 0; i < ebt->entries_count; i++)
+	{
+		ebt_entry_t* entry = &ebt->entries[i];
+		if (entry->in >= 0 && entry->out >= 0)
+		{
+			if (entry->in > entry->out)
+			{
+				*in += entry->in - entry->out;
+			}
+			else if (entry->out > entry->in)
+			{
+				*out += entry->out - entry->in;
+			}
+		}
+	}
+}
+
 static ebt_entry_t* _ebt_get_entry(tf_ssb_ebt_t* ebt, const char* id)
 {
 	uint8_t bin[k_id_bin_len];
@@ -135,6 +157,13 @@ void tf_ssb_ebt_receive_clock(tf_ssb_ebt_t* ebt, JSContext* context, JSValue clo
 			}
 			JS_FreeValue(context, in_clock);
 		}
+
+		int in = 0;
+		int out = 0;
+		_ebt_count_messages(ebt, &in, &out);
+		ebt->max_in = tf_max(in, ebt->max_in);
+		ebt->max_out = tf_max(out, ebt->max_out);
+
 		uv_mutex_unlock(&ebt->mutex);
 		for (uint32_t i = 0; i < plen; ++i)
 		{
@@ -164,13 +193,15 @@ static void _ebt_add_to_clock(ebt_get_clock_t* work, const char* id, int64_t val
 {
 	int count = work->clock ? work->clock->count : 0;
 	ebt_entry_t* entry = _ebt_get_entry(work->ebt, id);
-	if (entry &&
+	if (entry)
-		((replicate && !entry->out_replicate) || (receive && !entry->out_receive) || ((replicate || receive || entry->out_replicate || entry->out_receive) && entry->out != value)))
 	{
 		entry->out = value;
 		entry->out_replicate = entry->out_replicate || replicate;
 		entry->out_receive = entry->out_receive || receive;
 
+		if ((replicate && !entry->out_replicate) || (receive && !entry->out_receive) ||
+			((replicate || receive || entry->out_replicate || entry->out_receive) && entry->out != value))
+		{
 			int index = tf_util_insert_index(id, count ? work->clock->entries : NULL, count, sizeof(tf_ssb_ebt_clock_entry_t), _ebt_compare_entry);
 			int64_t out_value = entry->out_replicate ? ((value << 1) | (entry->out_receive ? 0 : 1)) : -1;
 			if (index < count && strcmp(id, work->clock->entries[index].id) == 0)
@@ -190,6 +221,7 @@ static void _ebt_add_to_clock(ebt_get_clock_t* work, const char* id, int64_t val
 			}
 		}
 	}
+}
 
 static void _tf_ssb_ebt_get_send_clock_work(tf_ssb_connection_t* connection, void* user_data)
 {
@@ -268,6 +300,14 @@ static void _tf_ssb_ebt_get_send_clock_work(tf_ssb_connection_t* connection, voi
 		uv_mutex_unlock(&work->ebt->mutex);
 		tf_free(requested);
 	}
+
+	uv_mutex_lock(&work->ebt->mutex);
+	int in = 0;
+	int out = 0;
+	_ebt_count_messages(work->ebt, &in, &out);
+	work->ebt->max_in = tf_max(in, work->ebt->max_in);
+	work->ebt->max_out = tf_max(out, work->ebt->max_out);
+	uv_mutex_unlock(&work->ebt->mutex);
 }
 
 static void _tf_ssb_ebt_get_send_clock_after_work(tf_ssb_connection_t* connection, int status, void* user_data)
@@ -356,3 +396,12 @@ void tf_ssb_ebt_debug_clock(tf_ssb_ebt_t* ebt, JSContext* context, JSValue debug
 	}
 	uv_mutex_unlock(&ebt->mutex);
 }
+
+void tf_ssb_ebt_get_progress(tf_ssb_ebt_t* ebt, int* in_pending, int* in_total, int* out_pending, int* out_total)
+{
+	uv_mutex_lock(&ebt->mutex);
+	_ebt_count_messages(ebt, in_pending, out_pending);
+	*in_total = ebt->max_in;
+	*out_total = ebt->max_out;
+	uv_mutex_unlock(&ebt->mutex);
+}

src/ssb.ebt.h

@@ -106,3 +106,13 @@ void tf_ssb_ebt_set_send_clock_pending(tf_ssb_ebt_t* ebt, int pending);
 ** @param debug A JS object populated with the information.
 */
 void tf_ssb_ebt_debug_clock(tf_ssb_ebt_t* ebt, JSContext* context, JSValue debug);
+
+/**
+** Get a representation of sync progress.
+** @param ebt The EBT instance.
+** @param in_pending Populated with the number of messages remaining to be received.
+** @param in_total Populated with the total number of messages to receive this session.
+** @param out_pending Populated with the number of messages remaining to send.
+** @param out_total Populated with the total number of messages to send this session.
+*/
+void tf_ssb_ebt_get_progress(tf_ssb_ebt_t* ebt, int* in_pending, int* in_total, int* out_pending, int* out_total);

src/ssb.js.c

@@ -3,6 +3,7 @@
 #include "log.h"
 #include "mem.h"
 #include "ssb.db.h"
+#include "ssb.ebt.h"
 #include "ssb.h"
 #include "util.js.h"
 
@@ -942,6 +943,21 @@ static JSValue _tf_ssb_connections(JSContext* context, JSValueConst this_val, in
 			{
 				JS_SetPropertyStr(context, object, "destroy_reason", JS_NewString(context, destroy_reason));
 			}
+			int in = 0;
+			int out = 0;
+			int max_in = 0;
+			int max_out = 0;
+			tf_ssb_ebt_get_progress(tf_ssb_connection_get_ebt(connection), &in, &max_in, &out, &max_out);
+			JSValue progress = JS_NewObject(context);
+			JSValue in_progress = JS_NewObject(context);
+			JS_SetPropertyStr(context, in_progress, "current", JS_NewInt32(context, in));
+			JS_SetPropertyStr(context, in_progress, "total", JS_NewInt32(context, max_in));
+			JS_SetPropertyStr(context, progress, "in", in_progress);
+			JSValue out_progress = JS_NewObject(context);
+			JS_SetPropertyStr(context, out_progress, "current", JS_NewInt32(context, out));
+			JS_SetPropertyStr(context, out_progress, "total", JS_NewInt32(context, max_out));
+			JS_SetPropertyStr(context, progress, "out", out_progress);
+			JS_SetPropertyStr(context, object, "progress", progress);
 			JS_SetPropertyUint32(context, result, i, object);
 		}
 	}