Merge pull request #2029 from atsign-foundation/perf-tweaking

Author: gkc

packages/dart/noports_core/lib/src/npt/npt.dart

@@ -192,6 +192,11 @@ class _NptImpl extends NptBase
     /// Start the sshnpd payload handler
     await sshnpdChannel.callInitialization();
 
+    if (sshnpdChannel.cachedPingResponse != null) {
+      _srvdChannel.cachedDaemonPublicSigningKeyUri =
+          sshnpdChannel.cachedPingResponse!['publicSigningKeyUri'];
+    }
+
     List<DaemonFeature> requiredFeatures = [
       DaemonFeature.srAuth,
       DaemonFeature.srE2ee,

packages/dart/noports_core/lib/src/srvd/isolates/port_pair_isolate.dart

@@ -104,7 +104,8 @@ class PortPairWorker extends RelayWorker {
     if (lookups.containsKey(atKey)) {
       return lookups[atKey];
     } else {
-      final resp = await rpcToMain(IIRequest.create('lookup', atKey));
+      final resp = await rpcToMain(
+          IIRequest.create('lookup', {'key': atKey, 'sessionId': sessionId}));
       lookups[atKey] = resp.payload;
       return resp.payload;
     }

packages/dart/noports_core/lib/src/srvd/isolates/relay_worker.dart

@@ -56,6 +56,8 @@ abstract class RelayWorker implements RelayAuthVerifyHelper {
           } else {
             rpcCompleters[msg.id]!.complete(msg);
           }
+        } else {
+          logger.shout('Got an unexpected IIResponse (${msg.toString()})');
         }
         return;
       } else {

packages/dart/noports_core/lib/src/srvd/isolates/shared_single_port_isolate.dart

@@ -71,7 +71,8 @@ class SinglePortWorker extends RelayWorker {
     if (si.lookups[atKey] != null) {
       return si.lookups[atKey]!;
     } else {
-      final resp = await rpcToMain(IIRequest.create('lookup', atKey));
+      final resp = await rpcToMain(
+          IIRequest.create('lookup', {'key': atKey, 'sessionId': sessionId}));
       si.lookups[atKey] = resp.payload;
       return resp.payload;
     }

packages/dart/noports_core/lib/src/srvd/srvd_impl.dart

@@ -18,7 +18,7 @@ import 'srvd_session_params.dart';
 @protected
 class SrvdImpl implements Srvd {
   @override
-  final AtSignLogger logger = AtSignLogger(' srvd ');
+  final AtSignLogger logger = AtSignLogger(' srvd main ');
   @override
   AtClient atClient;
   @override
@@ -90,6 +90,9 @@ class SrvdImpl implements Srvd {
       if (p.verbose) {
         AtSignLogger.root_level = 'INFO';
       }
+      if (p.debug) {
+        AtSignLogger.root_level = 'FINEST';
+      }
 
       if (atClient == null && atClientGenerator == null) {
         throw StateError('atClient and atClientGenerator are both null');
@@ -217,41 +220,49 @@ class SrvdImpl implements Srvd {
       return;
     }
 
-    var mutexKey = AtKey.fromString('${sessionParams.sessionId}'
-        '.session_mutexes.${Srvd.namespace}'
-        '${atClient.getCurrentAtSign()!}')
-      ..metadata = (Metadata()
-        ..immutable = true // only one srvd will succeed in doing this
-        ..ttl = 30000); // expire after 30 seconds to keep datastore clean
-    PutRequestOptions pro = PutRequestOptions()
-      ..shouldEncrypt = false
-      ..useRemoteAtServer = true;
-
-    try {
-      await atClient.put(
-        mutexKey,
-        'lock',
-        putRequestOptions: pro,
-      );
+    if (sessionParams.multipleAcksOk) {
+      // client can handle multiple acks, no need to lock a mutex
       logger.shout('😎 Will handle request from ${notification.from}'
-          '; acquired mutex $mutexKey');
-    } catch (err) {
-      if (err.toString().toLowerCase().contains('immutable')) {
-        logger.shout('🤷‍♂️ Will not handle request from ${notification.from}'
-            '; did not acquire mutex $mutexKey');
-        ppiSendToSpawned?.send(IIRequest.create('stop', null));
-      } else {
-        logger.shout('Will not handle; did not acquire mutex $mutexKey : $err');
+          ' which can handle multiple acks (no mutex required)');
+    } else {
+      // client cannot handle multiple acks, so we need to lock a mutex
+      var mutexKey = AtKey.fromString('${sessionParams.sessionId}'
+          '.session_mutexes.${Srvd.namespace}'
+          '${atClient.getCurrentAtSign()!}')
+        ..metadata = (Metadata()
+          ..immutable = true // only one srvd will succeed in doing this
+          ..ttl = 30000); // expire after 30 seconds to keep datastore clean
+      PutRequestOptions pro = PutRequestOptions()
+        ..shouldEncrypt = false
+        ..useRemoteAtServer = true;
+
+      try {
+        await atClient.put(
+          mutexKey,
+          'lock',
+          putRequestOptions: pro,
+        );
+        logger.shout('😎 Will handle request from ${notification.from}'
+            '; acquired mutex $mutexKey');
+      } catch (err) {
+        if (err.toString().toLowerCase().contains('immutable')) {
+          logger.shout('🤷‍♂️ Will not handle request from ${notification.from}'
+              '; did not acquire mutex $mutexKey');
+          ppiSendToSpawned?.send(IIRequest.create('stop', null));
+        } else {
+          logger
+              .shout('Will not handle; did not acquire mutex $mutexKey : $err');
+        }
+        return;
       }
-      return;
     }
 
     if (sessionParams.only443) {
       if (!bind443) {
         var message = 'Client requested port 443'
             ' but this relay is not bound to port 443';
         logger.shout(message);
-        if (sessionParams.sendNacks) {
+        if (sessionParams.multipleAcksOk) {
           try {
             await sendNack(
               sessionId: sessionParams.sessionId,
@@ -299,20 +310,47 @@ class SrvdImpl implements Srvd {
           waitForFinalDeliveryStatus: false,
           checkForFinalDeliveryStatus: false);
     } catch (e) {
-      stderr.writeln("Error writing session ${notification.value} atKey");
+      logger.shout("Error sending response to client");
     }
+
+    preFetched[sessionParams.sessionId] = {};
+    for (final s in sessionParams.preFetch) {
+      try {
+        final AtValue value = await _lookup(AtKey.fromString(s));
+        preFetched[sessionParams.sessionId]![s] = value.value;
+      } catch (e) {
+        logger.shout('$e while preFetching $s');
+      }
+    }
+    unawaited(Future.delayed(Duration(seconds: 30))
+        .whenComplete(() => preFetched.remove(sessionParams.sessionId)));
+  }
+
+  Map<String, Map<String, dynamic>> preFetched = {};
+
+  Future<AtValue> _lookup(AtKey atKey) async {
+    logger.info('Looking up $atKey on atServer');
+    return await atClient.get(
+      atKey,
+      getRequestOptions: GetRequestOptions()..useRemoteAtServer = true,
+    );
   }
 
   @override
   Future<void> lookup(IIRequest msg, SendPort toSpawned) async {
-    final AtValue value;
     try {
       logger.info('request: "lookup" : ${msg.payload}');
-      value = await atClient.get(
-        AtKey.fromString(msg.payload),
-        getRequestOptions: GetRequestOptions()..useRemoteAtServer = true,
-      );
-      logger.info('request: "lookup" : success ${value.value}');
+      String sessionId = msg.payload['sessionId'];
+      String key = msg.payload['key'];
+      AtValue value;
+      String fromPreFetch = '';
+      if (preFetched[sessionId]?[key] != null) {
+        value = AtValue()..value = preFetched[sessionId]?[key];
+        fromPreFetch = ' (pre-fetched)';
+      } else {
+        value = await _lookup(AtKey.fromString(key));
+      }
+      logger.info('request: "lookup" : success$fromPreFetch: ${value.value}');
       toSpawned.send(IIResponse(
         id: msg.id,
         isError: false,

packages/dart/noports_core/lib/src/srvd/srvd_params.dart

@@ -14,6 +14,7 @@ class SrvdParams {
   final bool logTraffic;
   final String rootDomain;
   final bool perSessionStorage;
+  final bool debug;
 
   /// Whether to start an isolate where all connections are to the same port
   final bool bind443;
@@ -37,6 +38,7 @@ class SrvdParams {
     required this.perSessionStorage,
     required this.bind443,
     required this.localBindPort443,
+    required this.debug,
   });
 
   static Future<SrvdParams> fromArgs(List<String> args) async {
@@ -65,6 +67,7 @@ class SrvdParams {
       bind443: r['443'],
       localBindPort443:
           r['443-bind-port'] == null ? 443 : int.parse(r['443-bind-port']),
+      debug: r['debug'],
     );
   }
 
@@ -105,7 +108,12 @@ class SrvdParams {
     parser.addFlag(
       'verbose',
       abbr: 'v',
-      help: 'More logging',
+      help: 'Show more logs (INFO and above)',
+    );
+    parser.addFlag(
+      'debug',
+      defaultsTo: false,
+      help: 'Show all logs (FINEST and above)',
     );
     if (BuildEnv.enableSnoop) {
       parser.addFlag(

packages/dart/noports_core/lib/src/srvd/srvd_session_params.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 
 import 'package:at_client/at_client.dart';
+import 'package:at_utils/at_logger.dart';
 import 'package:noports_core/sshnp_foundation.dart';
 import 'package:noports_core/srvd.dart';
 
@@ -18,7 +19,8 @@ class SrvdSessionParams {
   final RelayAuthMode relayAuthMode;
   final String? relayAuthAesKey;
   final bool only443;
-  final bool sendNacks;
+  final bool multipleAcksOk;
+  final List<String> preFetch;
 
   SrvdSessionParams({
     required this.sessionId,
@@ -33,7 +35,8 @@ class SrvdSessionParams {
     this.relayAuthMode = RelayAuthMode.payload,
     this.relayAuthAesKey,
     required this.only443,
-    required this.sendNacks,
+    required this.multipleAcksOk,
+    required this.preFetch,
   });
 
   @override
@@ -52,11 +55,13 @@ class SrvdSessionParams {
         'relayAuthMode': relayAuthMode.name,
         'relayAuthAesKey': relayAuthAesKey,
         'only443': only443,
-        'sendNacks': sendNacks,
+        'multipleAcksOk': multipleAcksOk,
+        'preFetch': preFetch,
       };
 }
 
 class SrvdUtil {
+  static AtSignLogger logger = AtSignLogger(' SrvdUtil ');
   final AtClient atClient;
 
   SrvdUtil(this.atClient);
@@ -79,7 +84,8 @@ class SrvdUtil {
       sessionId: notification.value!,
       atSignA: notification.from,
       only443: false,
-      sendNacks: false,
+      multipleAcksOk: false,
+      preFetch: [],
     );
   }
 
@@ -93,6 +99,7 @@ class SrvdUtil {
   Future<SrvdSessionParams> _sessionParamsForJsonRequest(
       AtNotification notification) async {
     dynamic json = jsonDecode(notification.value ?? '');
+    logger.info('Received session request JSON: $json');
 
     assertValidMapValue(json, 'sessionId', String);
     assertValidMapValue(json, 'atSignA', String);
@@ -135,7 +142,8 @@ class SrvdUtil {
       relayAuthMode: relayAuthMode,
       relayAuthAesKey: json['relayAuthAesKey'],
       only443: json['only443'] ?? false,
-      sendNacks: json['sendNacks'] ?? false,
+      multipleAcksOk: json['multipleAcksOk'] ?? false,
+      preFetch: List<String>.from(json['preFetch'] ?? []),
     );
   }
 

packages/dart/noports_core/lib/src/sshnp/sshnp_core.dart

@@ -161,6 +161,11 @@ abstract class SshnpCore
     }
     await sshnpdChannel.sharePublicKeyIfRequired(identityKeyPair);
 
+    if (sshnpdChannel.cachedPingResponse != null) {
+      srvdChannel.cachedDaemonPublicSigningKeyUri =
+          sshnpdChannel.cachedPingResponse!['publicSigningKeyUri'];
+    }
+
     /// Retrieve the srvd host and port pair
     sendProgress('Fetching host and port from srvd');
     await srvdChannel.callInitialization();

packages/dart/noports_core/lib/src/sshnp/util/srvd_channel/notification_request_message.dart

@@ -12,7 +12,8 @@ class SocketRendezvousRequestMessage {
   final RelayAuthMode relayAuthMode;
   final String? relayAuthAesKey;
   final bool only443;
-  final bool sendNacks;
+  final bool multipleAcksOk;
+  final List<String> preFetch;
 
   SocketRendezvousRequestMessage({
     required this.sessionId,
@@ -24,7 +25,8 @@ class SocketRendezvousRequestMessage {
     required this.relayAuthMode,
     required this.relayAuthAesKey,
     required this.only443,
-    required this.sendNacks,
+    required this.multipleAcksOk,
+    required this.preFetch,
   });
 
   @override
@@ -39,7 +41,8 @@ class SocketRendezvousRequestMessage {
     m['relayAuthMode'] = relayAuthMode.name;
     m['relayAuthAesKey'] = relayAuthAesKey;
     m['only443'] = only443;
-    m['sendNacks'] = sendNacks;
+    m['multipleAcksOk'] = multipleAcksOk;
+    m['preFetch'] = preFetch;
     return jsonEncode(m);
   }
 }

packages/dart/noports_core/lib/src/sshnp/util/srvd_channel/srvd_channel.dart

@@ -38,6 +38,8 @@ abstract class SrvdChannel<T>
   final String sessionId;
   final String clientNonce = DateTime.now().toIso8601String();
 
+  String? cachedDaemonPublicSigningKeyUri;
+
   Completer acked = Completer();
 
   bool fetched = false;
@@ -110,10 +112,12 @@ abstract class SrvdChannel<T>
 
   @override
   Future<void> initialize() async {
-    await publishPublicSigningKey();
+    Future publishPSKFuture = publishPublicSigningKey();
 
     await getHostAndPortFromSrvd();
 
+    await publishPSKFuture;
+
     completeInitialization();
   }
 
@@ -233,6 +237,16 @@ abstract class SrvdChannel<T>
           ..namespaceAware = false
           ..ttl = 10000);
 
+      List<String> preFetch = [];
+
+      // Currently prefetch is only needed if auth mode is ESCR
+      if (params.relayAuthMode == RelayAuthMode.escr) {
+        preFetch.add(publicSigningKeyUri);
+        if (cachedDaemonPublicSigningKeyUri != null) {
+          preFetch.add(cachedDaemonPublicSigningKeyUri!);
+        }
+      }
+
       var message = SocketRendezvousRequestMessage(
         sessionId: sessionId,
         atSignA: params.clientAtSign,
@@ -243,7 +257,8 @@ abstract class SrvdChannel<T>
         relayAuthMode: params.relayAuthMode,
         relayAuthAesKey: relayAuthAesKey,
         only443: params.only443,
-        sendNacks: true,
+        multipleAcksOk: true,
+        preFetch: preFetch,
       );
 
       rvdRequestValue = message.toString();