From 2e1e4f90e7068299f2dcb09a9cfbc79c82d7400a Mon Sep 17 00:00:00 2001
From: Cory McWilliams <cory@unprompted.com>
Date: Tue, 28 Dec 2021 15:25:04 +0000
Subject: [PATCH] Working toward less aggressive blob and feed fetching.

git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@3710 ed5197a5-7fde-0310-b194-c3ffbd925b24
---
 core/ssb.js | 61 ++++++++++++++++++++++++++++++++++++++++-------------
 1 file changed, 46 insertions(+), 15 deletions(-)

diff --git a/core/ssb.js b/core/ssb.js
index 516f4488..1d0fd1be 100644
--- a/core/ssb.js
+++ b/core/ssb.js
@@ -1,6 +1,8 @@
+"use strict";
 var g_wants_requests = {};
 var g_database = new Database('core');
 const k_use_create_history_stream = false;
+const k_blobs_concurrent_target = 8;
 
 function following(db, id) {
 	var o = db.get(id + ":following");
@@ -90,9 +92,12 @@ ssb.addEventListener('connections', function(change, connection) {
 				connection.send_json({"name": ["ebt", "replicate"], "args": [{"version": 3, "format": "classic"}], "type": "duplex"}, ebtReplicateClient);
 			}
 		}
+
+		connection.active_blob_wants = {};
 		connection.send_json({'name': ['blobs', 'createWants'], 'type': 'source', 'args': []}, function(message) {
 			Object.keys(message.message).forEach(function(id) {
 				if (message.message[id] < 0) {
+					delete connection.active_blob_wants[id];
 					var blob = ssb.blobGet(id);
 					if (blob) {
 						var out_message = {};
@@ -104,12 +109,19 @@ ssb.addEventListener('connections', function(change, connection) {
 					var expected_bytes = message.message[id];
 					var buffer = new Uint8Array(expected_bytes);
 					connection.send_json({'name': ['blobs', 'get'], 'type': 'source', 'args': [id]}, function(message) {
-						buffer.set(new Uint8Array(message.message, 0, message.message.byteLength), received_bytes);
-						received_bytes += message.message.byteLength;
-						if (received_bytes == expected_bytes) {
-							ssb.blobStore(buffer);
+						if (message.flags & 0x4 /* end */) {
+							delete connection.active_blob_wants[id];
+						} else {
+							buffer.set(new Uint8Array(message.message, 0, message.message.byteLength), received_bytes);
+							received_bytes += message.message.byteLength;
+							if (received_bytes == expected_bytes) {
+								ssb.blobStore(buffer);
+							}
 						}
 					});
+					if (Object.keys(connection.active_blob_wants).length < k_blobs_concurrent_target) {
+						requestMoreBlobs(g_wants_requests[connection.id]);
+					}
 				}
 			});
 		});
@@ -121,18 +133,27 @@ ssb.addEventListener('connections', function(change, connection) {
 	}
 });
 
+function blob_want_discovered(request, id) {
+	if (Object.keys(request.connection.active_blob_wants).length > k_blobs_concurrent_target) {
+		return;
+	}
+	var message = {};
+	message[id] = -1;
+	request.send_json(message);
+	request.connection.active_blob_wants[id] = true;
+}
+
+function requestMoreBlobs(request) {
+	ssb.sqlStream(
+		'SELECT id FROM blob_wants LIMIT ' + k_blobs_concurrent_target,
+		[],
+		row => blob_want_discovered(request, row.id));
+}
+
 ssb.addRpc(['blobs', 'createWants'], function(request) {
 	g_wants_requests[request.connection.id] = request;
-	function blob_want_discovered(id) {
-		var message = {};
-		message[id] = -1;
-		request.send_json(message);
-	}
-	ssb.addEventListener('blob_want_added', blob_want_discovered);
-	ssb.sqlStream(
-		'SELECT id FROM blob_wants',
-		[],
-		row => blob_want_discovered(row.id));
+	ssb.addEventListener('blob_want_added', id => blob_want_discovered(request, id));
+	requestMoreBlobs(request);
 });
 
 ssb.addRpc(['blobs', 'has'], function(request) {
@@ -166,10 +187,16 @@ ssb.addRpc(['tunnel', 'isRoom'], function(request) {
 function ebtReplicateSendClock(request, have) {
 	var me = ssb.whoami();
 	var message = {};
-	var ids = followingDeep(g_database, [me], 2).concat([request.connection.id]).concat(Object.keys(have));
+	var ids = followingDeep(g_database, [me], 2).concat([request.connection.id]);
 	for (let id of ids) {
 		message[id] = get_latest_sequence_for_author(id);
 	}
+	for (let id of Object.keys(have)) {
+		if (message[id] === undefined) {
+			var sequence = get_latest_sequence_for_author(id);
+			message[id] = sequence ? sequence : -1;
+		}
+	}
 
 	var last_sent = request.connection.sent_clock || {};
 	var to_send = {}
@@ -177,6 +204,10 @@ function ebtReplicateSendClock(request, have) {
 		if (last_sent[id] === undefined || message[id] > last_sent[id]) {
 			last_sent[id] = to_send[id] = message[id] === -1 ? -1 : message[id] << 1;
 		}
+		if (Object.keys(to_send).length >= 32) {
+			request.send_json(to_send);
+			to_send = {};
+		}
 	}
 	request.connection.sent_clock = last_sent;