From 8a6147d51287654df6c8a8cf75a0ef3f6c17ea99 Mon Sep 17 00:00:00 2001
From: Cory McWilliams <cory@unprompted.com>
Date: Sun, 6 Oct 2024 11:14:37 -0400
Subject: [PATCH] ssb: Beginnings of a "sync now" mode for mobile.

---
 apps/ssb.json                  |  2 +-
 apps/ssb/app.js                |  3 ++
 apps/ssb/tf-tab-connections.js | 10 +++++
 src/ssb.c                      | 55 ++++++++++++++++++++---
 src/ssb.connections.c          | 82 +++++++++++++++++++++++++++++++++-
 src/ssb.connections.h          |  6 +++
 src/ssb.h                      | 36 +++++++++++++--
 src/ssb.js.c                   | 38 +++++-----------
 src/ssb.rpc.c                  |  2 +-
 src/ssb.tests.c                | 26 +++++------
 10 files changed, 207 insertions(+), 53 deletions(-)

diff --git a/apps/ssb.json b/apps/ssb.json
index cc9d61e4..efd3e8a9 100644
--- a/apps/ssb.json
+++ b/apps/ssb.json
@@ -1,5 +1,5 @@
 {
 	"type": "tildefriends-app",
 	"emoji": "🐌",
-	"previous": "&PK+UixgYQEYFKKPJ3BVacSKaDRRjiNO6M/fmgzDJaRM=.sha256"
+	"previous": "&O7Rf3apWJhDC4Zfjo1a5ZRk6AMNeCmFuEIiczXwmSYE=.sha256"
 }
diff --git a/apps/ssb/app.js b/apps/ssb/app.js
index 3868b534..afa58858 100644
--- a/apps/ssb/app.js
+++ b/apps/ssb/app.js
@@ -103,6 +103,9 @@ tfrpc.register(async function encrypt(id, recipients, content) {
 tfrpc.register(async function getActiveIdentity() {
 	return await ssb.getActiveIdentity();
 });
+tfrpc.register(async function sync() {
+	return await ssb.sync();
+});
 core.register('onBroadcastsChanged', async function () {
 	await tfrpc.rpc.set('broadcasts', await ssb.getBroadcasts());
 });
diff --git a/apps/ssb/tf-tab-connections.js b/apps/ssb/tf-tab-connections.js
index da424002..a28d0ba7 100644
--- a/apps/ssb/tf-tab-connections.js
+++ b/apps/ssb/tf-tab-connections.js
@@ -158,10 +158,20 @@ class TfTabConnectionsElement extends LitElement {
 		`;
 	}
 
+	refresh() {
+		tfrpc.rpc.sync();
+	}
+
 	render() {
 		let self = this;
 		return html`
 			<div class="w3-container" style="box-sizing: border-box">
+				<button
+					class="w3-button w3-theme-l3 w3-circle w3-ripple w3-large"
+					@click=${this.refresh}
+				>
+					🔃
+				</button>
 				<h2>New Connection</h2>
 				<textarea class="w3-input w3-theme-d1" id="code"></textarea>
 				<button
diff --git a/src/ssb.c b/src/ssb.c
index 508e87d6..ed760abc 100644
--- a/src/ssb.c
+++ b/src/ssb.c
@@ -365,6 +365,8 @@ typedef struct _tf_ssb_connection_t
 	int active_write_count;
 
 	uint64_t last_notified_active;
+
+	int flags;
 } tf_ssb_connection_t;
 
 static JSClassID _connection_class_id;
@@ -1148,11 +1150,11 @@ bool tf_ssb_verify_and_strip_signature(JSContext* context, JSValue val, char* ou
 	return false;
 }
 
-void tf_ssb_close_all(tf_ssb_t* ssb)
+void tf_ssb_close_all(tf_ssb_t* ssb, const char* reason)
 {
 	for (tf_ssb_connection_t* connection = ssb->connections; connection; connection = connection->next)
 	{
-		_tf_ssb_connection_close(connection, "tf_ssb_close_all");
+		_tf_ssb_connection_close(connection, reason);
 	}
 }
 
@@ -2755,7 +2757,7 @@ static void _tf_ssb_connection_tunnel_callback(
 	}
 }
 
-tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id)
+tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id, int connect_flags)
 {
 	tf_ssb_connection_t* connection = tf_ssb_connection_get(ssb, portal_id);
 
@@ -2765,6 +2767,7 @@ tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char*
 	memset(tunnel, 0, sizeof(*tunnel));
 	snprintf(tunnel->name, sizeof(tunnel->name), "tun%d", s_tunnel_index++);
 	tunnel->ssb = ssb;
+	tunnel->flags = connect_flags;
 	tunnel->tunnel_connection = connection;
 	tunnel->tunnel_request_number = -request_number;
 	tunnel->send_request_number = 1;
@@ -2807,6 +2810,7 @@ typedef struct _connect_t
 	uv_getaddrinfo_t req;
 	char host[256];
 	int port;
+	int flags;
 	uint8_t key[k_id_bin_len];
 } connect_t;
 
@@ -2819,7 +2823,11 @@ static void _tf_on_connect_getaddrinfo(uv_getaddrinfo_t* addrinfo, int result, s
 		{
 			struct sockaddr_in addr = *(struct sockaddr_in*)info->ai_addr;
 			addr.sin_port = htons(connect->port);
-			tf_ssb_connection_create(connect->ssb, connect->host, &addr, connect->key);
+			tf_ssb_connection_t* connection = tf_ssb_connection_create(connect->ssb, connect->host, &addr, connect->key);
+			if (connection)
+			{
+				connection->flags = connect->flags;
+			}
 		}
 		else
 		{
@@ -2831,7 +2839,7 @@ static void _tf_on_connect_getaddrinfo(uv_getaddrinfo_t* addrinfo, int result, s
 	tf_free(connect);
 }
 
-void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key)
+void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key, int connect_flags)
 {
 	if (ssb->shutting_down)
 	{
@@ -2841,6 +2849,7 @@ void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* ke
 	*connect = (connect_t) {
 		.ssb = ssb,
 		.port = port,
+		.flags = connect_flags,
 		.req.data = connect,
 	};
 	char id[k_id_base64_len] = { 0 };
@@ -3131,12 +3140,12 @@ static bool _tf_ssb_parse_broadcast(const char* in_broadcast, tf_ssb_broadcast_t
 	return false;
 }
 
-void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address)
+void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address, int connect_flags)
 {
 	tf_ssb_broadcast_t broadcast = { 0 };
 	if (_tf_ssb_parse_broadcast(address, &broadcast))
 	{
-		tf_ssb_connect(ssb, broadcast.host, ntohs(broadcast.addr.sin_port), broadcast.pub);
+		tf_ssb_connect(ssb, broadcast.host, ntohs(broadcast.addr.sin_port), broadcast.pub, connect_flags);
 	}
 	else
 	{
@@ -4282,3 +4291,35 @@ void tf_ssb_connection_adjust_write_count(tf_ssb_connection_t* connection, int d
 	connection->active_write_count += delta;
 	_tf_ssb_connection_dispatch_scheduled(connection);
 }
+
+void tf_ssb_sync_start(tf_ssb_t* ssb)
+{
+	tf_ssb_connections_sync_start(ssb->connections_tracker);
+}
+
+bool tf_ssb_tunnel_create(tf_ssb_t* ssb, const char* portal_id, const char* target_id, int connect_flags)
+{
+	tf_ssb_connection_t* connection = tf_ssb_connection_get(ssb, portal_id);
+	if (connection)
+	{
+		JSContext* context = ssb->context;
+		int32_t request_number = tf_ssb_connection_next_request_number(connection);
+		JSValue message = JS_NewObject(context);
+		JSValue name = JS_NewArray(context);
+		JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel"));
+		JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "connect"));
+		JS_SetPropertyStr(context, message, "name", name);
+		JSValue arg = JS_NewObject(context);
+		JS_SetPropertyStr(context, arg, "portal", JS_NewString(context, portal_id));
+		JS_SetPropertyStr(context, arg, "target", JS_NewString(context, target_id));
+		JSValue args = JS_NewArray(context);
+		JS_SetPropertyUint32(context, args, 0, arg);
+		JS_SetPropertyStr(context, message, "args", args);
+		JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex"));
+		tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream | k_ssb_rpc_flag_new_request, request_number, "tunnel.connect", message, NULL, NULL, NULL);
+		JS_FreeValue(context, message);
+
+		tf_ssb_connection_tunnel_create(ssb, portal_id, request_number, target_id, connect_flags);
+	}
+	return connection != NULL;
+}
diff --git a/src/ssb.connections.c b/src/ssb.connections.c
index 547ee27a..ff7c154a 100644
--- a/src/ssb.connections.c
+++ b/src/ssb.connections.c
@@ -106,7 +106,7 @@ static void _tf_ssb_connections_get_next_after_work(tf_ssb_t* ssb, int status, v
 		uint8_t key_bin[k_id_bin_len];
 		if (tf_ssb_id_str_to_bin(key_bin, next->key))
 		{
-			tf_ssb_connect(ssb, next->host, next->port, key_bin);
+			tf_ssb_connect(ssb, next->host, next->port, key_bin, 0);
 		}
 	}
 	tf_free(next);
@@ -267,3 +267,83 @@ void tf_ssb_connections_set_succeeded(tf_ssb_connections_t* connections, const c
 	snprintf(update->key, sizeof(update->key), "%s", key);
 	_tf_ssb_connections_queue_update(connections, update);
 }
+
+static void _tf_ssb_connections_sync_broadcast_visit(
+	const char* host, const struct sockaddr_in* addr, tf_ssb_broadcast_origin_t origin, tf_ssb_connection_t* tunnel, const uint8_t* pub, void* user_data)
+{
+	tf_ssb_t* ssb = user_data;
+	if (tunnel)
+	{
+		char target_id[k_id_base64_len] = { 0 };
+		if (tf_ssb_id_bin_to_str(target_id, sizeof(target_id), pub))
+		{
+			char portal_id[k_id_base64_len] = { 0 };
+			if (tf_ssb_connection_get_id(tunnel, portal_id, sizeof(portal_id)))
+			{
+				tf_ssb_tunnel_create(ssb, portal_id, target_id, k_tf_ssb_connect_flag_one_shot);
+			}
+		}
+	}
+	else
+	{
+		tf_ssb_connect(ssb, host, ntohs(addr->sin_port), pub, k_tf_ssb_connect_flag_one_shot);
+	}
+}
+
+typedef struct _tf_ssb_connections_get_all_work_t
+{
+	char** connections;
+	int connections_count;
+} tf_ssb_connections_get_all_work_t;
+
+static void _tf_ssb_connections_get_all_work(tf_ssb_t* ssb, void* user_data)
+{
+	tf_ssb_connections_get_all_work_t* work = user_data;
+	sqlite3_stmt* statement;
+	sqlite3* db = tf_ssb_acquire_db_reader(ssb);
+	if (sqlite3_prepare(db, "SELECT host, port, key FROM connections ORDER BY last_attempt", -1, &statement, NULL) == SQLITE_OK)
+	{
+		while (sqlite3_step(statement) == SQLITE_ROW)
+		{
+			const char* host = (const char*)sqlite3_column_text(statement, 0);
+			int port = sqlite3_column_int(statement, 1);
+			const char* key = (const char*)sqlite3_column_text(statement, 2);
+			char connection[1024] = { 0 };
+			snprintf(connection, sizeof(connection), "net:%s:%d~shs:%s", host, port, key);
+			char* dot = strrchr(connection, '.');
+			if (dot && strcmp(dot, ".ed25519") == 0)
+			{
+				*dot = '\0';
+			}
+			work->connections = tf_resize_vec(work->connections, sizeof(char*) * (work->connections_count + 1));
+			work->connections[work->connections_count++] = tf_strdup(connection);
+		}
+		sqlite3_finalize(statement);
+	}
+	else
+	{
+		tf_printf("prepare: %s\n", sqlite3_errmsg(db));
+	}
+	tf_ssb_release_db_reader(ssb, db);
+}
+
+static void _tf_ssb_connections_get_all_after_work(tf_ssb_t* ssb, int status, void* user_data)
+{
+	tf_ssb_connections_get_all_work_t* work = user_data;
+	for (int i = 0; i < work->connections_count; i++)
+	{
+		tf_printf("connections[%d] = %s\n", i, work->connections[i]);
+		tf_ssb_connect_str(ssb, work->connections[i], k_tf_ssb_connect_flag_one_shot);
+		tf_free(work->connections[i]);
+	}
+	tf_free(work->connections);
+	tf_free(work);
+}
+
+void tf_ssb_connections_sync_start(tf_ssb_connections_t* connections)
+{
+	tf_ssb_connections_get_all_work_t* work = tf_malloc(sizeof(tf_ssb_connections_get_all_work_t));
+	*work = (tf_ssb_connections_get_all_work_t) { 0 };
+	tf_ssb_run_work(connections->ssb, _tf_ssb_connections_get_all_work, _tf_ssb_connections_get_all_after_work, work);
+	tf_ssb_visit_broadcasts(connections->ssb, _tf_ssb_connections_sync_broadcast_visit, connections->ssb);
+}
diff --git a/src/ssb.connections.h b/src/ssb.connections.h
index 6be839bd..6789cee9 100644
--- a/src/ssb.connections.h
+++ b/src/ssb.connections.h
@@ -53,4 +53,10 @@ void tf_ssb_connections_set_attempted(tf_ssb_connections_t* connections, const c
 */
 void tf_ssb_connections_set_succeeded(tf_ssb_connections_t* connections, const char* host, int port, const char* key);
 
+/**
+** Initiate an immediate sync.
+** @param connections The connections tracker.
+*/
+void tf_ssb_connections_sync_start(tf_ssb_connections_t* connections);
+
 /** @} */
diff --git a/src/ssb.h b/src/ssb.h
index a5682f2b..a5e518fb 100644
--- a/src/ssb.h
+++ b/src/ssb.h
@@ -65,6 +65,14 @@ typedef enum _tf_ssb_message_flags_t
 	k_tf_ssb_message_flag_sequence_before_author = 1,
 } tf_ssb_message_flags_t;
 
+/**
+** Flags affecting an SSB connection.
+*/
+typedef enum _tf_ssb_connect_flags_t
+{
+	k_tf_ssb_connect_flag_one_shot = 0x1,
+} tf_ssb_connect_flags_t;
+
 /** An SSB instance. */
 typedef struct _tf_ssb_t tf_ssb_t;
 /** An SSB connection. */
@@ -346,15 +354,17 @@ int tf_ssb_get_connections(tf_ssb_t* ssb, tf_ssb_connection_t** out_connections,
 ** @param host The host name or address.
 ** @param port The host's SHS port.
 ** @param key The host's SSB identity.
+** @param connect_flags Flags affecting the connection.
 */
-void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key);
+void tf_ssb_connect(tf_ssb_t* ssb, const char* host, int port, const uint8_t* key, int connect_flags);
 
 /**
 ** Establish an SHS connection with a host by string address.
 ** @param ssb The SSB instance.
 ** @param address The address.
+** @param connect_flags Flags affecting the connection.
 */
-void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address);
+void tf_ssb_connect_str(tf_ssb_t* ssb, const char* address, int connect_flags);
 
 /**
 ** Begin listening for SHS connections on the given port.
@@ -380,8 +390,9 @@ void tf_ssb_server_close(tf_ssb_t* ssb);
 /**
 ** Close all active SHS connections.
 ** @param ssb The SSB instance.
+** @param reason Reason for the close.
 */
-void tf_ssb_close_all(tf_ssb_t* ssb);
+void tf_ssb_close_all(tf_ssb_t* ssb, const char* reason);
 
 /**
 ** Send a graceful close message to all active SHS connections.
@@ -867,9 +878,10 @@ void tf_ssb_connection_remove_room_attendant(tf_ssb_connection_t* connection, co
 ** @param portal_id The identity of the tunnel intermediary.
 ** @param request_number The tunnel request.
 ** @param target_id The identity being tunneled to.
+** @param connect_flags Flags affecting the connection.
 ** @return The new tunnel connection.
 */
-tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id);
+tf_ssb_connection_t* tf_ssb_connection_tunnel_create(tf_ssb_t* ssb, const char* portal_id, int32_t request_number, const char* target_id, int connect_flags);
 
 /**
 ** Get the request number on which to send EBT responses.
@@ -1073,4 +1085,20 @@ void tf_ssb_connection_adjust_read_backpressure(tf_ssb_connection_t* connection,
 */
 void tf_ssb_connection_adjust_write_count(tf_ssb_connection_t* connection, int delta);
 
+/**
+** Initiate a tunnel connection.
+** @param ssb The SSB instance.
+** @param portal_id The public key of the instance through which to tunnel.
+** @param target_id The public key of the instance with which to establish a connection.
+** @param connect_flags Flags affecting the connection.
+** @return true if the tunnel instance was found.
+*/
+bool tf_ssb_tunnel_create(tf_ssb_t* ssb, const char* portal_id, const char* target_id, int connect_flags);
+
+/**
+** Initiate a one time sync operation.
+** @param ssb The SSB instance.
+*/
+void tf_ssb_sync_start(tf_ssb_t* ssb);
+
 /** @} */
diff --git a/src/ssb.js.c b/src/ssb.js.c
index 8129b7f6..3d187962 100644
--- a/src/ssb.js.c
+++ b/src/ssb.js.c
@@ -1536,7 +1536,7 @@ static JSValue _tf_ssb_connect(JSContext* context, JSValueConst this_val, int ar
 		{
 			const char* address_str = JS_ToCString(context, args);
 			tf_printf("Connecting to %s\n", address_str);
-			tf_ssb_connect_str(ssb, address_str);
+			tf_ssb_connect_str(ssb, address_str, 0);
 			JS_FreeCString(context, address_str);
 		}
 		else
@@ -1553,7 +1553,7 @@ static JSValue _tf_ssb_connect(JSContext* context, JSValueConst this_val, int ar
 				tf_printf("Connecting to %s:%d\n", address_str, port_int);
 				uint8_t pubkey_bin[k_id_bin_len];
 				tf_ssb_id_str_to_bin(pubkey_bin, pubkey_str);
-				tf_ssb_connect(ssb, address_str, port_int, pubkey_bin);
+				tf_ssb_connect(ssb, address_str, port_int, pubkey_bin, 0);
 			}
 			else
 			{
@@ -1826,37 +1826,15 @@ static JSValue _tf_ssb_remove_event_listener(JSContext* context, JSValueConst th
 
 static JSValue _tf_ssb_createTunnel(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv)
 {
-	JSValue result = JS_UNDEFINED;
 	tf_ssb_t* ssb = JS_GetOpaque(this_val, _tf_ssb_classId);
 	const char* portal_id = JS_ToCString(context, argv[0]);
 	const char* target_id = JS_ToCString(context, argv[1]);
 
-	tf_ssb_connection_t* connection = tf_ssb_connection_get(ssb, portal_id);
-	if (connection)
-	{
-		int32_t request_number = tf_ssb_connection_next_request_number(connection);
-		JSValue message = JS_NewObject(context);
-		JSValue name = JS_NewArray(context);
-		JS_SetPropertyUint32(context, name, 0, JS_NewString(context, "tunnel"));
-		JS_SetPropertyUint32(context, name, 1, JS_NewString(context, "connect"));
-		JS_SetPropertyStr(context, message, "name", name);
-		JSValue arg = JS_NewObject(context);
-		JS_SetPropertyStr(context, arg, "portal", JS_NewString(context, portal_id));
-		JS_SetPropertyStr(context, arg, "target", JS_NewString(context, target_id));
-		JSValue args = JS_NewArray(context);
-		JS_SetPropertyUint32(context, args, 0, arg);
-		JS_SetPropertyStr(context, message, "args", args);
-		JS_SetPropertyStr(context, message, "type", JS_NewString(context, "duplex"));
-		tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream | k_ssb_rpc_flag_new_request, request_number, "tunnel.connect", message, NULL, NULL, NULL);
-		JS_FreeValue(context, message);
-
-		tf_ssb_connection_tunnel_create(ssb, portal_id, request_number, target_id);
-		result = JS_TRUE;
-	}
+	bool result = tf_ssb_tunnel_create(ssb, portal_id, target_id, 0);
 
 	JS_FreeCString(context, target_id);
 	JS_FreeCString(context, portal_id);
-	return result;
+	return result ? JS_TRUE : JS_FALSE;
 }
 
 enum
@@ -2306,6 +2284,13 @@ static JSValue _tf_ssb_following(JSContext* context, JSValueConst this_val, int
 	return result;
 }
 
+static JSValue _tf_ssb_sync(JSContext* context, JSValueConst this_val, int argc, JSValueConst* argv)
+{
+	tf_ssb_t* ssb = JS_GetOpaque(this_val, _tf_ssb_classId);
+	tf_ssb_sync_start(ssb);
+	return JS_UNDEFINED;
+}
+
 void tf_ssb_register(JSContext* context, tf_ssb_t* ssb)
 {
 	JS_NewClassID(&_tf_ssb_classId);
@@ -2350,6 +2335,7 @@ void tf_ssb_register(JSContext* context, tf_ssb_t* ssb)
 	JS_SetPropertyStr(context, object, "connect", JS_NewCFunction(context, _tf_ssb_connect, "connect", 1));
 	JS_SetPropertyStr(context, object, "createTunnel", JS_NewCFunction(context, _tf_ssb_createTunnel, "createTunnel", 3));
 	JS_SetPropertyStr(context, object, "following", JS_NewCFunction(context, _tf_ssb_following, "following", 2));
+	JS_SetPropertyStr(context, object, "sync", JS_NewCFunction(context, _tf_ssb_sync, "sync", 0));
 	/* Write. */
 	JS_SetPropertyStr(context, object, "storeMessage", JS_NewCFunction(context, _tf_ssb_storeMessage, "storeMessage", 1));
 	JS_SetPropertyStr(context, object, "blobStore", JS_NewCFunction(context, _tf_ssb_blobStore, "blobStore", 1));
diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c
index 3afcfb6e..c6e7d373 100644
--- a/src/ssb.rpc.c
+++ b/src/ssb.rpc.c
@@ -395,7 +395,7 @@ static void _tf_ssb_rpc_tunnel_connect(tf_ssb_connection_t* connection, uint8_t
 		const char* origin_str = JS_ToCString(context, origin);
 		const char* portal_str = JS_ToCString(context, portal);
 		const char* target_str = JS_ToCString(context, target);
-		tf_ssb_connection_tunnel_create(ssb, portal_str, -request_number, origin_str);
+		tf_ssb_connection_tunnel_create(ssb, portal_str, -request_number, origin_str, 0);
 		JS_FreeCString(context, origin_str);
 		JS_FreeCString(context, portal_str);
 		JS_FreeCString(context, target_str);
diff --git a/src/ssb.tests.c b/src/ssb.tests.c
index 4dc801e5..2ac2877d 100644
--- a/src/ssb.tests.c
+++ b/src/ssb.tests.c
@@ -245,7 +245,7 @@ void tf_ssb_test_ssb(const tf_test_options_t* options)
 
 	uint8_t id0bin[k_id_bin_len];
 	tf_ssb_id_str_to_bin(id0bin, id0);
-	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin);
+	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin, 0);
 
 	tf_printf("Waiting for connection.\n");
 	while (test.connection_count0 != 1 || test.connection_count1 != 1)
@@ -457,8 +457,8 @@ void tf_ssb_test_rooms(const tf_test_options_t* options)
 
 	uint8_t id0bin[k_id_bin_len];
 	tf_ssb_id_str_to_bin(id0bin, id0);
-	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin);
-	tf_ssb_connect(ssb2, "127.0.0.1", 12347, id0bin);
+	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin, 0);
+	tf_ssb_connect(ssb2, "127.0.0.1", 12347, id0bin, 0);
 
 	tf_printf("Waiting for connection.\n");
 	while (test.connection_count0 != 2 || test.connection_count1 != 1 || test.connection_count2 != 1)
@@ -497,7 +497,7 @@ void tf_ssb_test_rooms(const tf_test_options_t* options)
 	tf_ssb_connection_rpc_send_json(connections[0], k_ssb_rpc_flag_stream | k_ssb_rpc_flag_new_request, tunnel_request_number, "tunnel.connect", message, NULL, NULL, NULL);
 	JS_FreeValue(context, message);
 
-	tf_ssb_connection_t* tun0 = tf_ssb_connection_tunnel_create(ssb1, id0, tunnel_request_number, id2);
+	tf_ssb_connection_t* tun0 = tf_ssb_connection_tunnel_create(ssb1, id0, tunnel_request_number, id2, 0);
 	tf_printf("tun0 = %p\n", tun0);
 
 	tf_printf("Done.\n");
@@ -523,9 +523,9 @@ void tf_ssb_test_rooms(const tf_test_options_t* options)
 	tf_ssb_send_close(ssb1);
 	tf_ssb_send_close(ssb2);
 
-	tf_ssb_close_all(ssb0);
-	tf_ssb_close_all(ssb1);
-	tf_ssb_close_all(ssb2);
+	tf_ssb_close_all(ssb0, "end of test");
+	tf_ssb_close_all(ssb1, "end of test");
+	tf_ssb_close_all(ssb2, "end of test");
 
 	uv_run(&loop, UV_RUN_DEFAULT);
 
@@ -689,7 +689,7 @@ void tf_ssb_test_bench(const tf_test_options_t* options)
 	tf_ssb_register(tf_ssb_get_context(ssb1), ssb1);
 
 	tf_ssb_server_open(ssb0, 12347);
-	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin);
+	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin, 0);
 
 	tf_printf("Waiting for messages.\n");
 	clock_gettime(CLOCK_REALTIME, &start_time);
@@ -824,7 +824,7 @@ static void _ssb_test_room_broadcasts_visit(
 		JS_FreeValue(context, message);
 
 		tf_printf("tunnel create ssb=%p portal=%s rn=%d target=%s\n", ssb, portal, (int)tunnel_request_number, target);
-		tf_ssb_connection_tunnel_create(ssb, portal, tunnel_request_number, target);
+		tf_ssb_connection_tunnel_create(ssb, portal, tunnel_request_number, target, 0);
 		_break_in_a_bit(ssb, tunnel, target, tunnel_request_number);
 	}
 }
@@ -858,8 +858,8 @@ void tf_ssb_test_go_ssb_room(const tf_test_options_t* options)
 
 	tf_ssb_add_broadcasts_changed_callback(ssb0, _ssb_test_room_broadcasts_changed, NULL, NULL);
 
-	tf_ssb_connect_str(ssb0, "net:linode.unprompted.com:8008~shs:Q0pc/7kXQJGIlqJxuwayL2huayzddgkVDoGkYVWQS1Y=:SSB+Room+PSK3TLYC2T86EHQCUHBUHASCASE18JBV24=");
-	tf_ssb_connect_str(ssb1, "net:linode.unprompted.com:8008~shs:Q0pc/7kXQJGIlqJxuwayL2huayzddgkVDoGkYVWQS1Y=:SSB+Room+PSK3TLYC2T86EHQCUHBUHASCASE18JBV24=");
+	tf_ssb_connect_str(ssb0, "net:linode.unprompted.com:8008~shs:Q0pc/7kXQJGIlqJxuwayL2huayzddgkVDoGkYVWQS1Y=:SSB+Room+PSK3TLYC2T86EHQCUHBUHASCASE18JBV24=", 0);
+	tf_ssb_connect_str(ssb1, "net:linode.unprompted.com:8008~shs:Q0pc/7kXQJGIlqJxuwayL2huayzddgkVDoGkYVWQS1Y=:SSB+Room+PSK3TLYC2T86EHQCUHBUHASCASE18JBV24=", 0);
 
 	uv_run(&loop, UV_RUN_DEFAULT);
 
@@ -959,8 +959,8 @@ void tf_ssb_test_peer_exchange(const tf_test_options_t* options)
 	tf_ssb_whoami(ssb0, id0, sizeof(id0));
 	uint8_t id0bin[k_id_bin_len];
 	tf_ssb_id_str_to_bin(id0bin, id0);
-	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin);
-	tf_ssb_connect(ssb2, "127.0.0.1", 12347, id0bin);
+	tf_ssb_connect(ssb1, "127.0.0.1", 12347, id0bin, 0);
+	tf_ssb_connect(ssb2, "127.0.0.1", 12347, id0bin, 0);
 
 	while (_count_broadcasts(ssb0) != 2 || _count_broadcasts(ssb1) != 1 || _count_broadcasts(ssb2) != 1)
 	{