Revert "libutil: drop Fs{Source,Sink}::good"

This reverts commit 1340807e30dba4b3972c31f02861bbaeaeb60e61.

Change-Id: I34d2a80eb3c3e9d79cb02b92cd1189da32d18cb6

src/libstore/remote-store.cc

@@ -39,7 +39,9 @@ RemoteStore::RemoteStore(const Params & params)
             },
             [this](const ref<Connection> & r) {
                 return
-                    std::chrono::duration_cast<std::chrono::seconds>(
+                    r->to.good()
+                    && r->from.good()
+                    && std::chrono::duration_cast<std::chrono::seconds>(
                         std::chrono::steady_clock::now() - r->startTime).count() < maxConnectionAge;
             }
             ))
@@ -178,10 +180,6 @@ void RemoteStore::ConnectionHandle::processStderr(Sink * sink, Source * source,
                     m.find("Derive([") != std::string::npos)
                     throw Error("%s, this might be because the daemon is too old to understand dependencies on dynamic derivations. Check to see if the raw derivation is in the form '%s'", std::move(m), "DrvWithVersion(..)");
             }
-            // the daemon can still handle more requests, so the connection itself
-            // is still valid. the current *handle* however should be considered a
-            // lost cause and abandoned entirely.
-            handle.release();
             throw;
         }
     }

src/libutil/serialise.cc

@@ -52,7 +52,18 @@ FdSink::~FdSink()
 void FdSink::writeUnbuffered(std::string_view data)
 {
     written += data.size();
+    try {
         writeFull(fd, data);
+    } catch (SysError & e) {
+        _good = false;
+        throw;
+    }
+}
+
+
+bool FdSink::good()
+{
+    return _good;
 }
 
 
@@ -117,13 +128,19 @@ size_t FdSource::readUnbuffered(char * data, size_t len)
         checkInterrupt();
         n = ::read(fd, data, len);
     } while (n == -1 && errno == EINTR);
-    if (n == -1) { throw SysError("reading from file"); }
-    if (n == 0) { throw EndOfFile(std::string(*endOfFileError)); }
+    if (n == -1) { _good = false; throw SysError("reading from file"); }
+    if (n == 0) { _good = false; throw EndOfFile(std::string(*endOfFileError)); }
     read += n;
     return n;
 }
 
 
+bool FdSource::good()
+{
+    return _good;
+}
+
+
 size_t StringSource::read(char * data, size_t len)
 {
     if (pos == s.size()) throw EndOfFile("end of string reached");

src/libutil/serialise.hh

@@ -18,6 +18,7 @@ struct Sink
 {
     virtual ~Sink() { }
     virtual void operator () (std::string_view data) = 0;
+    virtual bool good() { return true; }
 };
 
 /**
@@ -79,6 +80,8 @@ struct Source
      */
     virtual size_t read(char * data, size_t len) = 0;
 
+    virtual bool good() { return true; }
+
     void drainInto(Sink & sink);
 
     std::string drain();
@@ -133,6 +136,11 @@ struct FdSink : BufferedSink
     ~FdSink();
 
     void writeUnbuffered(std::string_view data) override;
+
+    bool good() override;
+
+private:
+    bool _good = true;
 };
 
 
@@ -157,8 +165,11 @@ struct FdSource : BufferedSource
         return *this;
     }
 
+    bool good() override;
 protected:
     size_t readUnbuffered(char * data, size_t len) override;
+private:
+    bool _good = true;
 };