From cc560704deb5077923b7cf9694148ef027927009 Mon Sep 17 00:00:00 2001
From: eldritch horrors <pennae@lix.systems>
Date: Wed, 11 Jun 2025 17:22:33 +0200
Subject: [PATCH] libstore: have SSH use a socketpair, not two pipes

remote store async io will need to set O_NONBLOCK on the connection fds,
and right now the number of fds can vary between connection types: local
connections have one one fd for the sink/source pair since they use unix
sockets, but ssh connections have two because ssh uses pipes. this makes
it rather hard to manage flags correctly, and even harder to wait for io
readiness on both directions using kj. using sockets for ssh fixes this.

Change-Id: I0f563ece7627cd3fbd0f5ce21c25140469729e5a
---
 lix/libstore/legacy-ssh-store.cc |  4 ++--
 lix/libstore/ssh-store.cc        |  4 ++--
 lix/libstore/ssh.cc              | 39 ++++++++++++++++++++++----------
 lix/libstore/ssh.hh              |  2 +-
 4 files changed, 32 insertions(+), 17 deletions(-)

diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc
index 1b9ab541e..47be65317 100644
--- a/lix/libstore/legacy-ssh-store.cc
+++ b/lix/libstore/legacy-ssh-store.cc
@@ -135,8 +135,8 @@ struct LegacySSHStore final : public Store
                    ? ""
                    : " --store " + shellEscape(config_.remoteStore.get()))
         );
-        conn->to = FdSink(conn->sshConn->in.get());
-        conn->from = FdSource(conn->sshConn->out.get());
+        conn->to = FdSink(conn->sshConn->socket.get());
+        conn->from = FdSource(conn->sshConn->socket.get());
         conn->store = this;
 
         try {
diff --git a/lix/libstore/ssh-store.cc b/lix/libstore/ssh-store.cc
index a8a09ef46..7d3810e12 100644
--- a/lix/libstore/ssh-store.cc
+++ b/lix/libstore/ssh-store.cc
@@ -97,8 +97,8 @@ ref<RemoteStore::Connection> SSHStore::openConnection()
         command += " --store " + shellEscape(config_.remoteStore.get());
 
     conn->sshConn = ssh.startCommand(command);
-    conn->to = FdSink(conn->sshConn->in.get());
-    conn->from = FdSource(conn->sshConn->out.get());
+    conn->to = FdSink(conn->sshConn->socket.get());
+    conn->from = FdSource(conn->sshConn->socket.get());
     return conn;
 }
 
diff --git a/lix/libstore/ssh.cc b/lix/libstore/ssh.cc
index 06ad66bc3..b43cc50a9 100644
--- a/lix/libstore/ssh.cc
+++ b/lix/libstore/ssh.cc
@@ -1,10 +1,13 @@
 #include "lix/libutil/current-process.hh"
 #include "lix/libutil/environment-variables.hh"
 #include "lix/libstore/ssh.hh"
+#include "lix/libutil/error.hh"
+#include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/finally.hh"
 #include "lix/libutil/logging.hh"
 #include "lix/libutil/strings.hh"
 #include "lix/libstore/temporary-dir.hh"
+#include <sys/socket.h>
 
 namespace nix {
 
@@ -47,9 +50,23 @@ void SSH::addCommonSSHOpts(Strings & args)
 
 std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
 {
-    Pipe in, out;
-    in.create();
-    out.create();
+    int sp[2];
+    // only linux and bsd support SOCK_CLOEXEC in socketpair type.
+#if __linux__ || __FreeBSD__
+    constexpr int sock_type = SOCK_STREAM | SOCK_CLOEXEC;
+#else
+    constexpr int sock_type = SOCK_STREAM;
+#endif
+    if (socketpair(AF_UNIX, sock_type, 0, sp) < 0) {
+        throw SysError("socketpair() for ssh");
+    }
+
+    AutoCloseFD parent(sp[0]), child(sp[1]);
+#if !(__linux__ || __FreeBSD__)
+    if (fcntl(parent.get(), F_SETFD, O_CLOEXEC) < 0 || fcntl(child.get(), F_SETFD, O_CLOEXEC) < 0) {
+        throw SysError("making socketpair O_CLOEXEC");
+    }
+#endif
 
     auto conn = std::make_unique<Connection>();
     ProcessOptions options;
@@ -64,13 +81,14 @@ std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
     conn->sshPid = startProcess([&]() {
         restoreProcessContext();
 
-        close(in.writeSide.get());
-        close(out.readSide.get());
+        parent.close();
 
-        if (dup2(in.readSide.get(), STDIN_FILENO) == -1)
+        if (dup2(child.get(), STDIN_FILENO) == -1) {
             throw SysError("duping over stdin");
-        if (dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
+        }
+        if (dup2(child.get(), STDOUT_FILENO) == -1) {
             throw SysError("duping over stdout");
+        }
         if (logFD != -1 && dup2(logFD, STDERR_FILENO) == -1)
             throw SysError("duping over stderr");
 
@@ -93,12 +111,9 @@ std::unique_ptr<SSH::Connection> SSH::startCommand(const std::string & command)
         throw SysError("unable to execute '%s'", args.front());
     }, options);
 
+    child.close();
 
-    in.readSide.reset();
-    out.writeSide.reset();
-
-    conn->out = std::move(out.readSide);
-    conn->in = std::move(in.writeSide);
+    conn->socket = std::move(parent);
 
     return conn;
 }
diff --git a/lix/libstore/ssh.hh b/lix/libstore/ssh.hh
index d8a42dd57..a3bf8358b 100644
--- a/lix/libstore/ssh.hh
+++ b/lix/libstore/ssh.hh
@@ -36,7 +36,7 @@ public:
     struct Connection
     {
         Pid sshPid;
-        AutoCloseFD out, in;
+        AutoCloseFD socket;
     };
 
     std::unique_ptr<Connection> startCommand(const std::string & command);