libutil/async-io: augment read type safety w.r.t to EOFs

Usually, EOFs are represented by returning 0 in the `read` APIs, at
least, this is what read(2) dictate.

As clever creature, we may sum zeroes sometimes (advanced form:
`buf->added(got)`) and forego handling the EOF condition.

To avoid the bug that lurked in remote-store.cc and caused busy looping
if the remote end disconnects suddenly, we return
`Result<Option<size_t>>` forcing the caller to perform a specific
processing for the EOF situation.

The conversion did not raise any other offending code path.

Change-Id: I185fdcb77aa82d87ab0802d66ac37c1363657a73
Signed-off-by: Raito Bezarius <raito@lix.systems>

lix/libstore/binary-cache-store.cc

@@ -399,12 +399,12 @@ try {
         {
         }
 
-        kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+        kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
         {
             return decompressed->read(buffer, size).then([&](auto r) {
                 if (r.has_value()) {
-                    if (r.value() > 0) {
-                        total += r.value();
+                    if (r.value()) {
+                        total += *r.value();
                     } else {
                         stats.narRead++;
                         // stats.narReadCompressedBytes += nar->size(); // FIXME

lix/libstore/build/derivation-goal.cc

@@ -1229,7 +1229,11 @@ try {
     while (true) {
         std::string_view data;
         try {
-            data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))};
+            if (const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size()))) {
+                data = {buf.begin(), *got};
+            } else {
+                co_return std::nullopt;
+            }
         } catch (SysError & e) {
             // the builder output stream may be a pty fd, and closing one pty
             // endpoint sends EIO to the other endpoint. this is a good exit.
@@ -1272,13 +1276,14 @@ DerivationGoal::handleHookOutput(AsyncInputStream & in) noexcept
 try {
     auto buf = kj::heapArray<char>(4096);
     while (true) {
-        std::string_view data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))};
-        lastChildActivity = AIO().provider.getTimer().now();
-
-        if (data.empty()) {
+        const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size()));
+        if (!got) {
             co_return std::nullopt;
         }
 
+        std::string_view data = {buf.begin(), *got};
+        lastChildActivity = AIO().provider.getTimer().now();
+
         for (auto c : data)
             if (c == '\n') {
                 auto json = parseJSONMessage(currentHookLine, "the derivation builder");

lix/libstore/filetransfer.cc

@@ -802,12 +802,17 @@ struct curlFileTransfer : public FileTransfer
                 {
                     AutoCloseFD fd;
                     OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {}
-                    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+                    kj::Promise<Result<std::optional<size_t>>>
+                    read(void * buffer, size_t size) override
                     {
                         // NOTE the synchronous implementation used to have a buffer for
                         // file data, but we cannot be bothered to treat this edge case.
                         if (const auto got = ::read(fd.get(), buffer, size); got >= 0) {
-                            return {result::success(got)};
+                            if (got == 0) {
+                                return {result::success(std::nullopt)};
+                            } else {
+                                return {result::success(got)};
+                            }
                         } else {
                             return {result::failure(std::make_exception_ptr(SysError("reading file")
                             ))};
@@ -1071,7 +1076,7 @@ struct curlFileTransfer : public FileTransfer
             co_return result::current_exception();
         }
 
-        kj::Promise<Result<size_t>> read(void * buffer, size_t len) override
+        kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t len) override
         try {
             TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len));
 
@@ -1084,7 +1089,11 @@ struct curlFileTransfer : public FileTransfer
                 total += available;
             }
 
-            co_return total;
+            if (total == 0) {
+                co_return std::nullopt;
+            } else {
+                co_return total;
+            }
         } catch (...) {
             co_return result::current_exception();
         }

lix/libstore/legacy-ssh-store.cc

@@ -356,7 +356,7 @@ struct LegacySSHStore final : public Store
 
             NarStream(Pool<Connection>::Handle conn) : conn(std::move(conn)) {}
 
-            kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+            kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
             {
                 return copier->read(buffer, size);
             }

lix/libstore/local-store.cc

@@ -1211,12 +1211,10 @@ try {
             free(toRealloc);
             throw std::bad_alloc();
         }
-        auto got = 0;
-        Finally cleanup([&]() {
-            dump = {dumpBuffer.get(), dump.size() + got};
-        });
+        std::optional<size_t> got = std::nullopt;
+        Finally cleanup([&]() { dump = {dumpBuffer.get(), dump.size() + got.value_or(0)}; });
         got = TRY_AWAIT(source.read(dumpBuffer.get() + oldSize, want));
-        if (got == 0) {
+        if (!got) {
             inMemory = true;
             break;
         }
@@ -1234,7 +1232,7 @@ try {
             bool useSecond = false;
             ChainSource(AsyncInputStream & s1, AsyncInputStream & s2) : source1(s1), source2(s2) {}
 
-            kj::Promise<Result<size_t>> read(void * data, size_t len) override
+            kj::Promise<Result<std::optional<size_t>>> read(void * data, size_t len) override
             try {
                 if (useSecond) {
                     co_return TRY_AWAIT(source2.read(data, len));

lix/libstore/remote-store.cc

@@ -737,7 +737,7 @@ try {
         {
         }
 
-        kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+        kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
         {
             return narCopier->read(buffer, size);
         }
@@ -788,7 +788,12 @@ try {
             KJ_DEFER(resetBlockingState(getFD(), oldState));
             while (fromBuf->used() < sizeof(uint64_t)) {
                 const auto available = fromBuf->getWriteBuffer();
-                fromBuf->added(TRY_AWAIT(stream.read(available.data(), available.size())));
+                const auto got = TRY_AWAIT(stream.read(available.data(), available.size()));
+                if (got) {
+                    fromBuf->added(*got);
+                } else {
+                    throw Error("Nix daemon disconnected while waiting for a response");
+                }
             }
         }
 

lix/libstore/store-api.cc

@@ -1020,11 +1020,13 @@ struct CopyPathStream : AsyncInputStream
     {
     }
 
-    kj::Promise<Result<size_t>> read(void * data, size_t len) override
+    kj::Promise<Result<std::optional<size_t>>> read(void * data, size_t len) override
     try {
         auto result = TRY_AWAIT(inner->read(data, len));
-        copied += result;
-        act.progress(copied, expected);
+        if (result) {
+            copied += *result;
+            act.progress(copied, expected);
+        }
         co_return result;
     } catch (...) {
         co_return result::current_exception();

lix/libutil/archive.cc

@@ -596,31 +596,32 @@ struct AsyncCopier : AsyncInputStream
 
     explicit AsyncCopier(AsyncInputStream & source) : source(source) {}
 
-    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+    kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
     try {
         while (current.pending == 0) {
             if (auto want = stream.next()) {
                 current = *want;
             } else {
-                co_return 0;
+                co_return std::nullopt;
             }
         }
 
         size = std::min<uint64_t>(current.pending, size);
         if (size == 0) {
-            co_return 0;
+            co_return std::nullopt;
         }
 
         auto got = TRY_AWAIT(source.read(buffer, size));
-        current.pending -= got;
-        if (got == 0) {
+        if (!got) {
             throw badArchive("truncated NAR encountered");
-        } else if (!current.pendingFileContents) {
-            auto end = this->buffer.size();
-            this->buffer.resize(end + got);
-            memcpy(this->buffer.data() + end, buffer, got);
         }
-        co_return got;
+        current.pending -= *got;
+        if (!current.pendingFileContents) {
+            auto end = this->buffer.size();
+            this->buffer.resize(end + *got);
+            memcpy(this->buffer.data() + end, buffer, *got);
+        }
+        co_return *got;
     } catch (...) {
         co_return result::current_exception();
     }
@@ -673,11 +674,12 @@ struct AsyncParser
     try {
         while (n > 0) {
             auto got = TRY_AWAIT(source.read(buffer, n));
-            if (got == 0) {
+            if (!got) {
                 throw badArchive("unexpected end of nar encountered");
             }
-            buffer += got;
-            n -= got;
+
+            buffer += *got;
+            n -= *got;
         }
         co_return result::success();
     } catch (...) {
@@ -814,10 +816,10 @@ try {
 
         NarSource(AsyncInputStream & source) : source(source) {}
 
-        kj::Promise<Result<size_t>> read(void * data, size_t len) override
+        kj::Promise<Result<std::optional<size_t>>> read(void * data, size_t len) override
         try {
             auto n = TRY_AWAIT(source.read(data, len));
-            pos += n;
+            pos += n.value_or(0);
             co_return n;
         } catch (...) {
             co_return result::current_exception();

lix/libutil/async-io.cc

@@ -13,7 +13,7 @@ try {
     constexpr size_t BUF_SIZE = 65536;
     auto buf = std::make_unique<char[]>(BUF_SIZE);
     while (auto r = TRY_AWAIT(read(buf.get(), BUF_SIZE))) {
-        sink(std::string_view(buf.get(), r));
+        sink(std::string_view(buf.get(), *r));
     }
     co_return result::success();
 } catch (...) {
@@ -25,7 +25,7 @@ try {
     constexpr size_t BUF_SIZE = 65536;
     auto buf = std::make_unique<char[]>(BUF_SIZE);
     while (auto r = TRY_AWAIT(read(buf.get(), BUF_SIZE))) {
-        TRY_AWAIT(stream.writeFull(buf.get(), r));
+        TRY_AWAIT(stream.writeFull(buf.get(), *r));
     }
     co_return result::success();
 } catch (...) {
@@ -41,7 +41,7 @@ try {
     co_return result::current_exception();
 }
 
-kj::Promise<Result<size_t>> AsyncSourceInputStream::read(void * buffer, size_t size)
+kj::Promise<Result<std::optional<size_t>>> AsyncSourceInputStream::read(void * buffer, size_t size)
 try {
     while (true) {
         if (auto got = inner.read(static_cast<char *>(buffer), size); got > 0) {
@@ -49,37 +49,45 @@ try {
         }
     }
 } catch (EndOfFile &) {
-    return {result::success(0)};
+    return {result::success(std::nullopt)};
 } catch (...) {
     return {result::current_exception()};
 }
 
-kj::Promise<Result<size_t>> AsyncStringInputStream::read(void * buffer, size_t size)
+kj::Promise<Result<std::optional<size_t>>> AsyncStringInputStream::read(void * buffer, size_t size)
 {
     size = std::min(size, s.size());
-    if (size > 0) {
-        memcpy(buffer, s.data(), size);
-        s.remove_prefix(size);
+    if (size == 0) {
+        return {result::success(std::nullopt)};
     }
+
+    memcpy(buffer, s.data(), size);
+    s.remove_prefix(size);
     return {result::success(size)};
 }
 
-kj::Promise<Result<size_t>> AsyncTeeInputStream::read(void * buffer, size_t size)
+kj::Promise<Result<std::optional<size_t>>> AsyncTeeInputStream::read(void * buffer, size_t size)
 try {
     auto got = TRY_AWAIT(inner.read(buffer, size));
-    sink({static_cast<char *>(buffer), got});
+    if (got) {
+        sink({static_cast<char *>(buffer), *got});
+    }
     co_return got;
 } catch (...) {
     co_return result::current_exception();
 }
 
-kj::Promise<Result<size_t>> AsyncGeneratorInputStream::read(void * data, size_t len)
+kj::Promise<Result<std::optional<size_t>>> AsyncGeneratorInputStream::read(void * data, size_t len)
 try {
+    if (len == 0) {
+        return {result::success(std::nullopt)};
+    }
+
     while (!buf.size()) {
         if (auto next = g.next()) {
             buf = *next;
         } else {
-            return {result::success(0)};
+            return {result::success(std::nullopt)};
         }
     }
 
@@ -91,15 +99,15 @@ try {
     return {result::current_exception()};
 }
 
-kj::Promise<Result<size_t>> AsyncBufferedInputStream::read(void * data, size_t size)
+kj::Promise<Result<std::optional<size_t>>> AsyncBufferedInputStream::read(void * data, size_t size)
 try {
     while (buffer->used() == 0) {
         const auto space = buffer->getWriteBuffer();
         const auto got = TRY_AWAIT(inner.read(space.data(), space.size()));
-        if (got == 0) {
-            co_return 0;
+        if (!got) {
+            co_return std::nullopt;
         }
-        buffer->added(got);
+        buffer->added(*got);
     }
 
     const auto available = buffer->getReadBuffer();
@@ -162,11 +170,13 @@ AsyncFdIoStream::~AsyncFdIoStream() noexcept(false)
     }
 }
 
-kj::Promise<Result<size_t>> AsyncFdIoStream::read(void * tgt, size_t size)
+kj::Promise<Result<std::optional<size_t>>> AsyncFdIoStream::read(void * tgt, size_t size)
 {
     auto got = ::read(fd, tgt, size);
-    if (got >= 0) {
+    if (got > 0) {
         return {result::success(got)};
+    } else if (got == 0) {
+        return {result::success(std::nullopt)};
     } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
         return observer.whenBecomesReadable().then([=, this] { return read(tgt, size); });
     } else {

lix/libutil/async-io.hh

@@ -27,8 +27,8 @@ class AsyncInputStream : private kj::AsyncObject
 public:
     virtual ~AsyncInputStream() noexcept(false) {}
 
-    // expected to return 0 only on EOF or when `size = 0` was explicitly set.
-    virtual kj::Promise<Result<size_t>> read(void * buffer, size_t size) = 0;
+    // expected to return none only on EOF or when `size = 0` was explicitly set.
+    virtual kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) = 0;
 
     kj::Promise<Result<void>> drainInto(Sink & sink);
     kj::Promise<Result<void>> drainInto(AsyncOutputStream & stream);
@@ -48,7 +48,7 @@ public:
     AsyncSourceInputStream(Source & inner) : inner(inner) {}
     AsyncSourceInputStream(box_ptr<Source> inner) : inner(*inner), owned(std::move(inner).take()) {}
 
-    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override;
 };
 
 class AsyncStringInputStream : public AsyncInputStream
@@ -58,7 +58,7 @@ class AsyncStringInputStream : public AsyncInputStream
 public:
     explicit AsyncStringInputStream(std::string_view s) : s(s) {}
 
-    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override;
 };
 
 // this writes to sources instead of async streams because none of the sinks
@@ -71,7 +71,7 @@ class AsyncTeeInputStream : public AsyncInputStream
 public:
     AsyncTeeInputStream(AsyncInputStream & inner, Sink & sink) : inner(inner), sink(sink) {}
 
-    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override;
 };
 
 class AsyncGeneratorInputStream : public AsyncInputStream
@@ -83,7 +83,7 @@ private:
 public:
     AsyncGeneratorInputStream(Generator<Bytes> && g) : g(std::move(g)) {}
 
-    kj::Promise<Result<size_t>> read(void * data, size_t len) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * data, size_t len) override;
 };
 
 class AsyncBufferedInputStream : public AsyncInputStream
@@ -105,7 +105,7 @@ public:
 
     KJ_DISALLOW_COPY_AND_MOVE(AsyncBufferedInputStream);
 
-    kj::Promise<Result<size_t>> read(void * data, size_t size) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * data, size_t size) override;
 };
 
 class AsyncOutputStream : private kj::AsyncObject
@@ -180,7 +180,7 @@ public:
         return fd;
     }
 
-    kj::Promise<Result<size_t>> read(void * tgt, size_t size) override;
+    kj::Promise<Result<std::optional<size_t>>> read(void * tgt, size_t size) override;
     kj::Promise<Result<size_t>> write(const void * src, size_t size) override;
 };
 

lix/libutil/compression.cc

@@ -345,8 +345,8 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream
             if (buf.used() == 0) {
                 const auto space = buf.getWriteBuffer();
                 const auto got = TRY_AWAIT(inner->read(space.data(), space.size()));
-                if (got > 0) {
-                    buf.added(got);
+                if (got) {
+                    buf.added(*got);
                 } else {
                     co_return;
                 }
@@ -371,7 +371,7 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream
         feedExc = std::current_exception();
     }
 
-    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+    kj::Promise<Result<std::optional<size_t>>> read(void * buffer, size_t size) override
     try {
         while (true) {
             if (const auto got = ::read(uncompressed.readSide.get(), buffer, size); got > 0) {
@@ -379,7 +379,7 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream
             } else if (got == 0) {
                 // decompresser must have finished, poll for any errors and return EOF
                 thread.get();
-                co_return 0;
+                co_return std::nullopt;
             } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
                 co_await readObserver->whenBecomesReadable();
             } else {

lix/libutil/file-system.cc

@@ -450,7 +450,7 @@ try {
     try {
         while (true) {
             if (auto n = TRY_AWAIT(source.read(buf.data(), buf.size()))) {
-                writeFull(fd.get(), {buf.data(), n});
+                writeFull(fd.get(), {buf.data(), *n});
             } else {
                 break;
             }

tests/unit/libstore/filetransfer.cc

@@ -243,7 +243,7 @@ TEST(FileTransfer, exceptionAbortsRead)
         aio.blockOn(
             aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10)
         ),
-        0
+        std::nullopt
     );
 }
 
@@ -329,10 +329,11 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers)
         size_t dropped = 0;
         while (size > 0) {
             auto round = std::min(size, sizeof(buf));
-            round = aio.blockOn(source.read(buf, round));
-            if (round == 0) {
+            auto got = aio.blockOn(source.read(buf, round));
+            if (!got) {
                 break;
             }
+            round = *got;
             size -= round;
             dropped += round;
         }

tests/unit/libutil/archive.cc

@@ -421,7 +421,7 @@ TEST_P(NarTest, parseCopiedAsync)
     }
 
     char buf;
-    ASSERT_EQ(input.read(&buf, 1).wait(ws).value(), 0);
+    ASSERT_EQ(input.read(&buf, 1).wait(ws).value(), std::nullopt);
 }
 
 TEST_P(NarTest, index)