Make 'nix copy --to daemon' run in constant memory (daemon side)

Continuation of 97002b6. This makes
the daemon use constant memory. For example, it reduces the daemon's
maximum RSS on

  $ nix copy --from ~/my-nix --to daemon /nix/store/1n7x0yv8vq6zi90hfmian84vdhd04bgp-blender-2.79a

from 264 MiB to 7 MiB.

We now use a TunnelSource to prevent the connection from ending up in
an undefined state if an exception is thrown while the NAR is being
sent.

Issue #1681.

Author: edolstra

src/libstore/remote-store.cc

@@ -411,8 +411,9 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
                  << info.references << info.registrationTime << info.narSize
                  << info.ultimate << info.sigs << info.ca
                  << repair << !checkSigs;
-        copyNAR(source, conn->to);
-        conn->processStderr();
+        bool tunnel = GET_PROTOCOL_MINOR(conn->daemonVersion) >= 21;
+        if (!tunnel) copyNAR(source, conn->to);
+        conn->processStderr(0, tunnel ? &source : nullptr);
     }
 }
 

src/libstore/worker-protocol.hh

@@ -6,7 +6,7 @@ namespace nix {
 #define WORKER_MAGIC_1 0x6e697863
 #define WORKER_MAGIC_2 0x6478696f
 
-#define PROTOCOL_VERSION 0x114
+#define PROTOCOL_VERSION 0x115
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)
 

src/nix-daemon/nix-daemon.cc

@@ -690,12 +690,22 @@ static void performOp(TunnelLogger * logger, ref<LocalStore> store,
         if (!trusted)
             info.ultimate = false;
 
-        TeeSink tee(from);
-        parseDump(tee, tee.source);
+        std::string saved;
+        std::unique_ptr<Source> source;
+        if (GET_PROTOCOL_MINOR(clientVersion) >= 21)
+            source = std::make_unique<TunnelSource>(from);
+        else {
+            TeeSink tee(from);
+            parseDump(tee, tee.source);
+            saved = std::move(*tee.source.data);
+            source = std::make_unique<StringSource>(saved);
+        }
 
         logger->startWork();
-        store.cast<Store>()->addToStore(info, tee.source.data, (RepairFlag) repair,
+
+        store.cast<Store>()->addToStore(info, *source, (RepairFlag) repair,
             dontCheckSigs ? NoCheckSigs : CheckSigs, nullptr);
+
         logger->stopWork();
         break;
     }