.

wcfs/client/wcfs_watchlink.cpp

@@ -86,6 +86,7 @@ error _WatchLink::close() {
     //
     // TODO -> better pthread_kill(SIGINT) instead of relying on wcfs proper behaviour?
     error err2 = wlink._serveWG->wait();
+    // XXX getting ErrLinkDown from wait - ok?
     if (errors::Is(err2, context::canceled)) // canceled is expected and ok
         err2 = nil;
 
@@ -130,7 +131,7 @@ error _WatchLink::_serveRX(context::Context ctx) {
     defer([&]() {
         wlink._acceptq.close();
         wlink._rxmu.lock();
-        wlink._rxdown = true;   // don't allow new rxtab registers
+        wlink._rxdown = true;   // don't allow new rxtab registers, mark the link as down
         wlink._rxmu.unlock();
         for (auto _ : wlink._rxtab) {   // FIXME iterates without lock
             auto rxq = _.second;
@@ -247,9 +248,12 @@ error _WatchLink::replyReq(context::Context ctx, const PinReq *req, const string
 
     wlink._rxmu.lock();
     bool ok   = wlink._accepted.has(req->stream);
+    bool down = wlink._rxdown;
     wlink._rxmu.unlock();
     if (!ok)
         panic("reply to not accepted stream");
+    if (down)
+        return E(ErrLinkDown);
 
     error err = wlink._send(req->stream, answer);
 
@@ -290,8 +294,13 @@ pair</*reply*/string, error> _WatchLink::sendReq(context::Context ctx, const str
     if (_ == 0)
         return make_pair("", E(ctx->err()));
 
-    if (!ok)
-        return make_pair("", E(io::ErrUnexpectedEOF));
+    if (!ok) {
+        wlink._rxmu.lock();
+        bool down = wlink._rxdown;
+        wlink._rxmu.unlock();
+
+        return make_pair("", E(down ? ErrLinkDown : io::ErrUnexpectedEOF));
+    }
     string reply = rx.to_string();
     return make_pair(reply, nil);
 }