append: fix possible deadlock

This construct enters into a deadlock if
1) the writer is waiting because the queue is full
2) the reader is waiting on the writer to quit

The reason is that this status is not communicated through the queue
but with a separate variable.

This commit makes it so the status is communicated only through the
queue itself.  Special note has to be taken in the case of a timeout
in the reader (res == fd->Timeout) since normally this branch does not
look at the queue at all.

Because of this we needed to add a function the the channel that
actively checks whether the other side is still alive.

core/src/lib/channel.h

@@ -1,7 +1,7 @@
 /*
    BAREOS® - Backup Archiving REcovery Open Sourced
 
-   Copyright (C) 2023-2023 Bareos GmbH & Co. KG
+   Copyright (C) 2023-2024 Bareos GmbH & Co. KG
 
    This program is Free Software; you can redistribute it and/or
    modify it under the terms of version three of the GNU Affero General Public
@@ -211,6 +211,18 @@ template <typename T> class input {
     return false;
   }
 
+  void try_update_status()
+  {
+    if (did_close) { return; }
+
+    bool should_close = false;
+    if (auto result = shared->try_input_lock()) {
+      if (std::get_if<channel_closed>(&result.value())) { should_close = true; }
+    }
+
+    if (should_close) { close(); }
+  }
+
   void close()
   {
     if (!did_close) {

core/src/stored/append.cc

@@ -2,7 +2,7 @@
    BAREOS® - Backup Archiving REcovery Open Sourced
 
    Copyright (C) 2000-2012 Free Software Foundation Europe e.V.
-   Copyright (C) 2016-2023 Bareos GmbH & Co. KG
+   Copyright (C) 2016-2024 Bareos GmbH & Co. KG
 
    This program is Free Software; you can redistribute it and/or
    modify it under the terms of version three of the GNU Affero General Public
@@ -158,7 +158,7 @@ class MessageHandler {
 
   BareosSocket* close_and_get_sock()
   {
-    end.store(true);
+    output.close();
     receive_thread.join();
     return fd;
   }
@@ -174,9 +174,6 @@ class MessageHandler {
   {
   }
 
-  bool error_while_reading{false};
-  std::atomic<bool> end{false};
-
   BareosSocket* fd;
   channel::input<result_type> input;
   channel::output<result_type> output;
@@ -216,17 +213,22 @@ class MessageHandler {
         fd->msg = nullptr;
 
         if (!input.emplace(std::move(result))) {
-          Dmsg1(20,
-                "Tried to put message into queue; but it did not succeed.\n");
+          if (input.closed()) {
+            Dmsg1(20, "Tried to put message into closed queue.\n");
+          } else {
+            Dmsg1(20,
+                  "Tried to put message into queue; but it did not succeed.\n");
+          }
           cont = false;
         }
       } else if (res == fd->Error) {
         cont = false;
-        error_while_reading = true;
       } else {
         ASSERT(res == fd->Timeout);
+        input.try_update_status();
       }
-      if (end.load()) { cont = false; }
+
+      if (input.closed()) { cont = false; }
     }
 
     input.close();