libstore: remove ssh{,-ng}:// connection pooling

it's used very rarely, on ssh it's pretty much broken, and on ssh-ng
it's standing squarely in the way of the rpc transition. we have not
found significant use of this feature in public configs (with all of
*two* repositories using this store parameter), so why even keep it?

Change-Id: Ifcafdf794d815001a1a55a771e5823010e5b174b

doc/manual/rl-next/no-max-connections.md

@@ -0,0 +1,11 @@
+---
+synopsis: "Remove `max-connections` store parameters for `ssh://` and `ssh-ng://` stores"
+cls: []
+category: Miscellany
+credits: [horrors]
+---
+
+The `max-connections` parameter was undocumented, untested, and (in the case of `ssh`) even ignored
+entirely for remote builds. During a survey of public nixos configurations we have found *two* uses
+of `max-connections` for `ssh-ng`, and none at all for `ssh`. Since it is so rarely used but brings
+significant internal complexity that hinders improvements we have decided to remove these features.

lix/libstore/legacy-ssh-store.cc

@@ -1,6 +1,7 @@
 #include "lix/libstore/legacy-ssh-store.hh"
 #include "libutil/error.hh"
 #include "libutil/logging.hh"
+#include "libutil/sync.hh"
 #include "lix/libutil/archive.hh"
 #include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
@@ -86,9 +87,6 @@ struct LegacySSHStoreConfig : CommonSSHStoreConfig
     const Setting<Path> remoteProgram{this, "nix-store", "remote-program",
         "Path to the `nix-store` executable on the remote machine."};
 
-    const Setting<int> maxConnections{this, 1, "max-connections",
-        "Maximum number of concurrent SSH connections."};
-
     const std::string name() override { return "SSH Store"; }
 
     std::string doc() override
@@ -256,7 +254,7 @@ struct LegacySSHStore final : public Store
 
     std::string host;
 
-    ref<Pool<Connection>> connections;
+    Sync<Connection, AsyncMutex> connection;
 
     SSH ssh;
 
@@ -266,20 +264,18 @@ struct LegacySSHStore final : public Store
         : Store(config)
         , config_(std::move(config))
         , host(host)
-        , connections(
-              make_ref<Pool<Connection>>(
-                  std::max(1, (int) config_.maxConnections),
-                  [this]() { return openConnection(); },
-                  [](const ref<Connection> & r) { return r->good; }
-              )
-          )
         , ssh(host, config_.port, config_.sshKey, config_.sshPublicHostKey, config_.compress)
     {
     }
 
-    kj::Promise<Result<ref<Connection>>> openConnection()
+    kj::Promise<Result<decltype(connection)::Lock>> getConnection()
     try {
-        auto conn = make_ref<Connection>();
+        auto conn = co_await connection.lock();
+        if (conn->good && conn->sshConn) {
+            co_return conn;
+        }
+
+        *conn = {};
         conn->sshConn = ssh.startCommand(
             fmt("%s --serve --write", config_.remoteProgram)
             + (config_.remoteStore.get() == "" ? "" : " --store " + shellEscape(config_.remoteStore.get()))
@@ -314,14 +310,14 @@ struct LegacySSHStore final : public Store
 
             conn->act.emplace_back(logger->startActivity(lvlDebug, actUnknown, "remote store " + getUri()));
             conn->logHandlerPromise = conn->logHandler(host);
-            return {conn};
+            co_return conn;
         } catch (Error & e) {
             std::string msg = chomp(drainFD(conn->sshConn->stderrPipe.get(), false));
             throw Error("cannot connect to %s: %s (%s)", getUri(), e.msg(), msg);
         }
     } catch (...) {
-        return {result::current_exception()};
+        co_return result::current_exception();
-    };
+    }
 
     std::string getUri() override
     {
@@ -331,7 +327,7 @@ struct LegacySSHStore final : public Store
     kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
     queryPathInfoUncached(const StorePath & path, const Activity * context) override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         debug("querying remote host '%s' for info on '%s'", host, printStorePath(path));
 
@@ -359,7 +355,7 @@ struct LegacySSHStore final : public Store
     try {
         debug("adding path '%s' to remote host '%s'", printStorePath(info.path), host);
 
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
         unsigned result;
 
         if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 5) {
@@ -403,16 +399,16 @@ struct LegacySSHStore final : public Store
     kj::Promise<Result<box_ptr<AsyncInputStream>>>
     narFromPath(const StorePath & path, const Activity * context) override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         struct NarStream : AsyncInputStream
         {
-            Pool<Connection>::Handle conn;
+            Sync<Connection, AsyncMutex>::Lock conn;
             AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn->sshConn->socket.get()};
             AsyncBufferedInputStream buffered{stream, conn->fromBuf};
             box_ptr<AsyncInputStream> copier{copyNAR(buffered)};
 
-            NarStream(Pool<Connection>::Handle conn) : conn(std::move(conn)) {}
+            NarStream(Sync<Connection, AsyncMutex>::Lock conn) : conn(std::move(conn)) {}
 
             kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
             {
@@ -476,7 +472,7 @@ public:
         const StorePath & drvPath, const BasicDerivation & drv, BuildMode buildMode
     ) override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         // this is a duplicate of DerivationGoal::buildDescription because ugh.
         // getting that information into here where needed is nigh *impossible*
@@ -509,7 +505,7 @@ public:
         if (evalStore && evalStore.get() != this)
             throw Error("building on an SSH store is incompatible with '--eval-store'");
 
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         Strings ss;
         for (auto & p : drvPaths) {
@@ -568,7 +564,7 @@ public:
             co_return result::success();
         }
 
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         out.merge(TRY_AWAIT(conn->sendCommand<StorePathSet>(
             ServeProto::Command::QueryClosure, includeOutputs, ServeProto::write(*conn, paths)
@@ -582,7 +578,7 @@ public:
     kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
         SubstituteFlag maybeSubstitute = NoSubstitute) override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
 
         co_return TRY_AWAIT(conn->sendCommand<StorePathSet>(
             ServeProto::Command::QueryValidPaths,
@@ -596,7 +592,7 @@ public:
 
     kj::Promise<Result<void>> init() override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
         co_return result::success();
     } catch (...) {
         co_return result::current_exception();
@@ -604,7 +600,7 @@ public:
 
     kj::Promise<Result<unsigned int>> getProtocol() override
     try {
-        auto conn(TRY_AWAIT(connections->get()));
+        auto conn(TRY_AWAIT(getConnection()));
         co_return conn->remoteVersion;
     } catch (...) {
         co_return result::current_exception();

lix/libstore/machines.cc

@@ -37,10 +37,6 @@ bool Machine::mandatoryMet(const std::set<std::string> & features) const
 kj::Promise<Result<ref<Store>>> Machine::openStore() const
 try {
     StoreConfig::Params storeParams;
-    if (storeUri.starts_with("ssh://")) {
-        // Remote builds become flakey, when having more than one ssh connection
-        storeParams["max-connections"] = "1";
-    }
 
     if (storeUri.starts_with("ssh://") || storeUri.starts_with("ssh-ng://")) {
         if (sshKey != "")

lix/libstore/remote-store-connection.hh

@@ -121,12 +121,21 @@ struct RemoteStore::Connection
  */
 struct RemoteStore::ConnectionHandle
 {
-    Pool<RemoteStore::Connection>::Handle handle;
+    Sync<std::shared_ptr<Connection>, AsyncMutex>::Lock handle;
 
-    ConnectionHandle(Pool<RemoteStore::Connection>::Handle && handle) : handle(std::move(handle)) {}
+    ConnectionHandle(Sync<std::shared_ptr<Connection>, AsyncMutex>::Lock && handle)
+        : handle(std::move(handle))
+    {
+    }
 
-    RemoteStore::Connection & operator * () { return *handle; }
+    RemoteStore::Connection & operator*()
-    RemoteStore::Connection * operator -> () { return &*handle; }
+    {
+        return **handle;
+    }
+    RemoteStore::Connection * operator->()
+    {
+        return &**handle;
+    }
 
     kj::Promise<Result<void>> processStderr(AsyncFdIoStream & stream);
 
@@ -148,11 +157,11 @@ struct RemoteStore::ConnectionHandle
         // the stack, but that's sufficiently suspect to warrant being as careful.
         auto invalidateOnCancel = kj::defer([&] {
             if (std::uncaught_exceptions() == 0) {
-                handle.markBad();
+                *handle = nullptr;
             }
         });
 
-        AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()};
+        AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, (*handle)->getFD()};
 
         // if the last argument can be serialized normally we will serialize *all*
         // arguments at once and hand off to the remote. if the last argument does
@@ -165,7 +174,7 @@ struct RemoteStore::ConnectionHandle
                 ((msg << std::forward<Args>(args)), ...);
                 TRY_AWAIT(stream.writeFull(msg.s.data(), msg.s.size()));
             } catch (...) {
-                handle.markBad();
+                *handle = nullptr;
                 throw;
             }
             LIX_TRY_AWAIT(processStderr(stream));
@@ -183,7 +192,7 @@ struct RemoteStore::ConnectionHandle
                     }(ImmediateArgsIdxs{});
                     TRY_AWAIT(stream.writeFull(msg.s.data(), msg.s.size()));
                 } catch (...) {
-                    handle.markBad();
+                    *handle = nullptr;
                     throw;
                 }
 
@@ -195,14 +204,16 @@ struct RemoteStore::ConnectionHandle
             co_return result::success();
         } else {
             try {
-                AsyncBufferedInputStream from{stream, handle->fromBuf};
+                AsyncBufferedInputStream from{stream, (*handle)->fromBuf};
-                auto result = LIX_TRY_AWAIT(WorkerProto::readAsync(
+                auto result = LIX_TRY_AWAIT(
-                    from, *handle->store, handle->daemonVersion, WorkerProto::Serialise<R>::read
+                    WorkerProto::readAsync(
-                ));
+                        from, *(*handle)->store, (*handle)->daemonVersion, WorkerProto::Serialise<R>::read
+                    )
+                );
                 invalidateOnCancel.cancel();
                 co_return result;
             } catch (...) {
-                handle.markBad();
+                *handle = nullptr;
                 throw;
             }
         }

lix/libstore/remote-store.cc

@@ -1,4 +1,5 @@
 #include "libutil/fmt.hh"
+#include "libutil/ref.hh"
 #include "lix/libutil/async-collect.hh"
 #include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
@@ -39,32 +40,23 @@
 namespace nix {
 
 /* TODO: Separate these store impls into different files, give them better names */
-RemoteStore::RemoteStore(const RemoteStoreConfig & config)
+RemoteStore::RemoteStore(const RemoteStoreConfig & config) : Store(config) {}
-    : Store(config)
-    , connections(make_ref<Pool<Connection>>(
-          std::max(1, (int) config.maxConnections),
-          [this]() { return openAndInitConnection(); },
-          [this](const ref<Connection> & r) {
-              return std::chrono::duration_cast<std::chrono::seconds>(
-                         std::chrono::steady_clock::now() - r->startTime
-                     )
-                         .count()
-                  < this->config().maxConnectionAge;
-          }
-      ))
-{
-}
 
 kj::Promise<Result<ref<RemoteStore::Connection>>> RemoteStore::openConnectionForDaemonForwarding()
 {
     return openConnection();
 }
 
-kj::Promise<Result<ref<RemoteStore::Connection>>> RemoteStore::openAndInitConnection()
+kj::Promise<Result<RemoteStore::ConnectionHandle>> RemoteStore::getConnection()
 try {
-    auto conn = TRY_AWAIT(openConnection());
+    auto conn = co_await connection.lock();
-    TRY_AWAIT(initConnection(*conn));
+    if (*conn) {
-    co_return conn;
+        co_return {std::move(conn)};
+    }
+
+    *conn = TRY_AWAIT(openConnection());
+    TRY_AWAIT(initConnection(**conn));
+    co_return {std::move(conn)};
 } catch (...) {
     co_return result::current_exception();
 }
@@ -184,26 +176,19 @@ try {
 
 kj::Promise<Result<void>> RemoteStore::ConnectionHandle::processStderr(AsyncFdIoStream & stream)
 try {
-    auto ex = TRY_AWAIT(handle->processStderr(stream));
+    auto ex = TRY_AWAIT((*handle)->processStderr(stream));
     if (ex.e) {
         co_return result::failure(ex.e);
     }
     co_return result::success();
 } catch (...) {
-    handle.markBad();
+    *handle = nullptr;
-    co_return result::current_exception();
-}
-
-kj::Promise<Result<RemoteStore::ConnectionHandle>> RemoteStore::getConnection()
-try {
-    co_return ConnectionHandle(TRY_AWAIT(connections->get()));
-} catch (...) {
     co_return result::current_exception();
 }
 
 kj::Promise<Result<void>> RemoteStore::setOptions()
 try {
-    TRY_AWAIT(setOptions(*(TRY_AWAIT(getConnection()).handle)));
+    TRY_AWAIT(setOptions(*(TRY_AWAIT(getConnection()))));
     co_return result::success();
 } catch (...) {
     co_return result::current_exception();
@@ -373,10 +358,6 @@ kj::Promise<Result<ref<const ValidPathInfo>>> RemoteStore::addCAToStore(
 try {
     auto conn(TRY_AWAIT(getConnection()));
 
-    // The dump source may invoke the store, so we need to make some room.
-    connections->incCapacity();
-    Finally cleanup([&]() { connections->decCapacity(); });
-
     co_return make_ref<ValidPathInfo>(TRY_AWAIT(conn.sendCommand<ValidPathInfo>(
         WorkerProto::Op::AddToStore,
         name,
@@ -708,8 +689,7 @@ try {
 
 kj::Promise<Result<unsigned int>> RemoteStore::getProtocol()
 try {
-    auto conn(TRY_AWAIT(connections->get()));
+    co_return TRY_AWAIT(getConnection())->daemonVersion;
-    co_return conn->daemonVersion;
 } catch (...) {
     co_return result::current_exception();
 }
@@ -856,7 +836,7 @@ try {
             TRY_AWAIT(framed.finish());
             co_return result::success();
         } catch (...) {
-            handle.markBad();
+            *handle = nullptr;
             co_return result::current_exception();
         }
     };

lix/libstore/remote-store.hh

@@ -5,6 +5,7 @@
 #include <string>
 #include <utility>
 
+#include "lix/libutil/sync.hh"
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/gc-store.hh"
 #include "lix/libstore/log-store.hh"
@@ -24,14 +25,6 @@ template<typename T> class Pool;
 struct RemoteStoreConfig : virtual StoreConfig
 {
     using StoreConfig::StoreConfig;
-
-    const Setting<int> maxConnections{this, 1, "max-connections",
-        "Maximum number of concurrent connections to the Nix daemon."};
-
-    const Setting<unsigned int> maxConnectionAge{this,
-        std::numeric_limits<unsigned int>::max(),
-        "max-connection-age",
-        "Maximum age of a connection before it is closed."};
 };
 
 /**
@@ -184,11 +177,9 @@ protected:
 
     virtual kj::Promise<Result<ref<Connection>>> openConnection() = 0;
 
-    kj::Promise<Result<ref<Connection>>> openAndInitConnection();
-
     virtual kj::Promise<Result<void>> initConnection(Connection & conn);
 
-    ref<Pool<Connection>> connections;
+    Sync<std::shared_ptr<Connection>, AsyncMutex> connection;
 
     virtual kj::Promise<Result<void>> setOptions(Connection & conn);