Refuse processing thread pushes after stop() and after a task threw

backend/include/backend/sftp/operation.hpp

@@ -22,6 +22,18 @@
 #include <system_error>
 #include <vector>
 
+template <>
+struct SecureShell::UnfinishedTaskResult<SharedData::OperationError>
+{
+    SharedData::OperationError operator()() const
+    {
+        return SharedData::OperationError{
+            .type = SharedData::OperationErrorType::ChannelClosed,
+            .extraInfo = "The task did not finish, the SFTP session is closed or its thread stopped",
+        };
+    }
+};
+
 class Operation
 {
   public:

backend/source/backend/sftp/operation_queue.cpp

@@ -1049,14 +1049,18 @@ bool OperationQueue::workQueue(std::deque<std::pair<Ids::OperationId, std::uniqu
     // every eligible op's workInStrand() call. Ops that do not use the strand run on the
     // caller afterwards: their work() blocks on that very thread and would deadlock inside it.
     SecureShell::ProcessingStrand* strand = nullptr;
+    std::size_t strandIndex = 0;
     for (const auto& item : eligible)
     {
         auto* operation = queue[item.index].second.get();
         operation->setParallelBudget(item.slots);
         if (!operation->hasWake())
             operation->setWake(wake_);
         if (operation->usesStrand() && !strand)
+        {
             strand = operation->strand();
+            strandIndex = item.index;
+        }
     }
 
     struct ItemResult
@@ -1103,17 +1107,42 @@ bool OperationQueue::workQueue(std::deque<std::pair<Ids::OperationId, std::uniqu
     std::vector<ItemResult> results;
     if (strand)
     {
-        auto fut = strand->pushPromiseTask(
-            [&runBatch]()
+        using Batch = std::optional<std::vector<ItemResult>>;
+        auto batch = strand
+                         ->pushPromiseTask(
+                             [&runBatch]() -> Batch
+                             {
+                                 return runBatch(true);
+                             },
+                             []() -> Batch
+                             {
+                                 return std::nullopt;
+                             }
+                         )
+                         .get();
+        // The processing thread no longer runs tasks, the operations that need it fail one per cycle.
+        if (!batch)
+        {
+            results.push_back(
+                {strandIndex,
+                    std::unexpected(
+                        Operation::Error{
+                            .type = Operation::ErrorType::ChannelClosed,
+                            .extraInfo = "The SFTP processing thread no longer runs tasks",
+                        }
+                    )}
+            );
+        }
+        else
+        {
+            results = std::move(batch).value();
+            if (std::ranges::none_of(results, isTerminal))
             {
-                return runBatch(true);
+                auto local = runBatch(false);
+                results.insert(
+                    results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end())
+                );
             }
-        );
-        results = fut.get();
-        if (std::ranges::none_of(results, isTerminal))
-        {
-            auto local = runBatch(false);
-            results.insert(results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end()));
         }
     }
     else

ssh/include/ssh/async/processing_strand.hpp

@@ -109,35 +109,35 @@ namespace SecureShell
         }
 
         /**
-         * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks.
+         * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks. The task
+         *        runs even when the thread no longer runs tasks, see ProcessingThread::pushCleanupTask.
          *
          * @param task The task to push.
          */
         void pushFinalTask(std::function<void()> task)
         {
             auto* processingThread = finalize();
             if (processingThread != nullptr)
-                processingThread->pushTask(std::move(task));
+                processingThread->pushCleanupTask(std::move(task));
         }
 
         /**
          * @brief Works like pushFinalTask but returns a future for the return value of the task.
          *
-         * @param task The task to run.
+         * @param func The task to run.
+         * @param unfinished Makes the future's value when the task does not finish, also when the strand was finalized
+         *        before. See UnfinishedTaskResult.
          */
-        template <typename FunctionT>
-        auto pushFinalPromiseTask(FunctionT&& func) -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
+        template <
+            typename FunctionT,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<FunctionT>>>>
+        auto pushFinalPromiseTask(FunctionT&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
         {
             auto* processingThread = finalize();
             if (processingThread == nullptr)
-            {
-                std::promise<std::invoke_result_t<std::decay_t<FunctionT>>> promise{};
-                promise.set_exception(
-                    std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand."))
-                );
-                return promise.get_future();
-            }
-            return processingThread->pushPromiseTask(std::forward<FunctionT>(func));
+                return Detail::unfinishedFuture<std::invoke_result_t<std::decay_t<FunctionT>>>(std::move(unfinished));
+            return processingThread->pushCleanupPromiseTask(std::forward<FunctionT>(func), std::move(unfinished));
         }
 
         /**
@@ -164,21 +164,20 @@ namespace SecureShell
          *
          * @tparam Func The type of the task.
          * @param func The task to push.
+         * @param unfinished Makes the future's value when the task does not finish, also when the strand was
+         *        finalized. See UnfinishedTaskResult.
          * @return std::future<std::invoke_result_t<std::decay_t<Func>>> The future of the return value of the task.
          */
-        template <typename Func>
-        auto pushPromiseTask(Func&& func) -> std::future<std::invoke_result_t<std::decay_t<Func>>>
+        template <
+            typename Func,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<Func>>>>
+        auto pushPromiseTask(Func&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<Func>>>
         {
             std::scoped_lock lock(mutex_);
             if (finalized_)
-            {
-                std::promise<std::invoke_result_t<std::decay_t<Func>>> promise{};
-                promise.set_exception(
-                    std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand."))
-                );
-                return promise.get_future();
-            }
-            return processingThread_->pushPromiseTask(std::forward<Func>(func));
+                return Detail::unfinishedFuture<std::invoke_result_t<std::decay_t<Func>>>(std::move(unfinished));
+            return processingThread_->pushPromiseTask(std::forward<Func>(func), std::move(unfinished));
         }
 
         /**