Unix: synchronize encryption fragment exceptions

Serialize updates to the shared exception pointer and retain the first
reported error. Test concurrent failures through all three handlers,
exception cleanup and subsequent successful work.

Tests/encryption_thread_pool_exception_test.cpp

@@ -0,0 +1,127 @@
+/* Copyright (c) 2026 AM Crypto. Licensed under the Apache License 2.0. */
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdlib>
+#include <cstring>
+#include <iostream>
+#include <mutex>
+#include <stdexcept>
+#include <vector>
+#include "Volume/EncryptionThreadPool.h"
+
+using namespace VeraCrypt;
+
+static std::atomic<int> Clones (0), LiveClones (0);
+
+static void Check (bool condition, const char *message)
+{
+    if (!condition)
+    {
+        std::cerr << "FAIL: " << message << std::endl;
+        std::exit (1);
+    }
+}
+
+struct TrackedException : Exception
+{
+    explicit TrackedException (bool stored = false) : Stored (stored)
+    { if (Stored) ++LiveClones; }
+    TrackedException (const TrackedException &other) : Exception (other), Stored (false) { }
+    ~TrackedException () throw () { if (Stored) --LiveClones; }
+    Exception *CloneNew () override { ++Clones; return new TrackedException (true); }
+    void Throw () const override { throw *this; }
+    bool Stored;
+};
+
+class PoolView : public EncryptionThreadPool
+{
+public:
+    static size_t Workers () { return ThreadCount; }
+};
+
+class TestMode : public EncryptionMode
+{
+public:
+    enum Failure { None, Native, Standard, Unknown };
+    TestMode (size_t fragments, Failure failure)
+        : Expected (fragments), Arrived (0), Kind (failure) { }
+
+    void Encrypt (uint8 *, uint64) const override { throw NotApplicable (); }
+    void Decrypt (uint8 *, uint64) const override { throw NotApplicable (); }
+    void EncryptSectorsCurrentThread (uint8 *data, uint64, uint64 count, size_t sectorSize) const override
+    { Process (data, count, sectorSize); }
+    void DecryptSectorsCurrentThread (uint8 *data, uint64, uint64 count, size_t sectorSize) const override
+    { Process (data, count, sectorSize); }
+    size_t GetKeySize () const override { return 0; }
+    std::wstring GetName () const override { return L"exception-test"; }
+    shared_ptr<EncryptionMode> GetNew () const override { return shared_ptr<EncryptionMode> (); }
+    void SetKey (const ConstBufferPtr &) override { }
+
+private:
+    void Process (uint8 *data, uint64 count, size_t sectorSize) const
+    {
+        {
+            std::unique_lock<std::mutex> lock (Gate);
+            ++Arrived;
+            Ready.notify_all ();
+            Check (Ready.wait_for (lock, std::chrono::seconds (10), [this] { return Arrived == Expected; }),
+                "fragment barrier timed out");
+        }
+        switch (Kind)
+        {
+        case Native: throw TrackedException ();
+        case Standard: throw std::runtime_error ("fragment failure");
+        case Unknown: throw 7;
+        case None: std::memset (data, 0xa5, count * sectorSize); break;
+        }
+    }
+
+    size_t Expected;
+    mutable size_t Arrived;
+    Failure Kind;
+    mutable std::mutex Gate;
+    mutable std::condition_variable Ready;
+};
+
+int main ()
+{
+    EncryptionThreadPool::Start ();
+    if (!EncryptionThreadPool::IsRunning ())
+    {
+        std::cout << "SKIP: at least two logical CPUs required" << std::endl;
+        return 77;
+    }
+    size_t widths[] = { 2, std::min<size_t> (8, PoolView::Workers ()) };
+    int checks = 0;
+    for (size_t width : widths)
+    {
+        std::vector<uint8> data (width * ENCRYPTION_DATA_UNIT_SIZE);
+        for (int direction = 0; direction < 2; ++direction)
+        {
+            auto type = direction ? EncryptionThreadPool::WorkType::DecryptDataUnits : EncryptionThreadPool::WorkType::EncryptDataUnits;
+            for (int kind = TestMode::Native; kind <= TestMode::Unknown; ++kind)
+            {
+                TestMode mode (width, static_cast<TestMode::Failure> (kind));
+                int observed = TestMode::None;
+                Clones = 0;
+                try { EncryptionThreadPool::DoWork (type, &mode, data.data (), 0, width, ENCRYPTION_DATA_UNIT_SIZE); }
+                catch (const TrackedException &) { observed = TestMode::Native; }
+                catch (const ExternalException &) { observed = TestMode::Standard; }
+                catch (const UnknownException &) { observed = TestMode::Unknown; }
+                Check (observed == kind, "requester received the wrong exception");
+                Check (LiveClones == 0, "stored exception leaked");
+                if (kind == TestMode::Native) Check (Clones == 1, "more than one fragment exception retained");
+                ++checks;
+            }
+            std::fill (data.begin (), data.end (), 0);
+            TestMode success (width, TestMode::None);
+            EncryptionThreadPool::DoWork (type, &success, data.data (), 0, width, ENCRYPTION_DATA_UNIT_SIZE);
+            for (uint8 byte : data) Check (byte == 0xa5, "pool failed after exception propagation");
+            ++checks;
+        }
+    }
+    EncryptionThreadPool::Stop ();
+    std::cout << "PASS: " << checks << " fragment exception and recovery checks" << std::endl;
+}

Tests/test_encryption_thread_pool_exceptions.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Check concurrent fragment exceptions in the production Unix encryption pool.
+
+Uses real pool, encryption-mode dispatch and platform synchronization code with
+an event-controlled test cipher. Requires Linux, Python 3 and a C++11 compiler.
+Run: python3 Tests/test_encryption_thread_pool_exceptions.py [--sanitize]
+"""
+import argparse
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+import tempfile
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--build-dir", type=Path)
+    parser.add_argument("--sanitize", action="store_true", help="enable AddressSanitizer")
+    args = parser.parse_args()
+    if not sys.platform.startswith("linux"):
+        parser.error("This build runner requires Linux")
+    build = (args.build_dir or Path(tempfile.mkdtemp(prefix="vc-pool-exceptions-"))).resolve()
+    build.mkdir(parents=True, exist_ok=True)
+    sources = ["Volume/EncryptionThreadPool.cpp", "Volume/EncryptionMode.cpp"]
+    sources += ["Platform/" + name + ".cpp" for name in (
+        "Buffer", "Exception", "Memory", "Serializable", "Serializer", "SerializerFactory",
+        "StringConverter", "Unix/Mutex", "Unix/SyncEvent", "Unix/SystemException",
+        "Unix/SystemLog", "Unix/Thread")]
+    exe = build / "encryption_thread_pool_exception_test"
+    command = shlex.split(os.environ.get("CXX", "c++")) + [
+        "-std=c++11", "-O2", "-g", "-pthread", "-ffunction-sections", "-fdata-sections",
+        "-Wl,--gc-sections", "-DTC_UNIX", "-DTC_LINUX", "-DARGON2_NO_THREADS",
+        "-D_FILE_OFFSET_BITS=64", "-I" + str(ROOT / "src"),
+        "-I" + str(ROOT / "src/Crypto"), "-I" + str(ROOT / "src/Crypto/Argon2/include")]
+    if args.sanitize:
+        command += ["-fsanitize=address", "-fno-omit-frame-pointer"]
+    command += [str(ROOT / "Tests/encryption_thread_pool_exception_test.cpp")]
+    command += [str(ROOT / "src" / source) for source in sources]
+    command += ["-o", str(exe)]
+    result = subprocess.run(command, capture_output=True, text=True, timeout=180)
+    (build / "build.log").write_text(result.stdout + result.stderr)
+    if result.returncode:
+        raise RuntimeError(result.stdout + result.stderr)
+    result = subprocess.run([str(exe)], timeout=60)
+    if result.returncode not in (0, 77):
+        raise SystemExit(result.returncode)
+
+
+if __name__ == "__main__":
+    main()

src/Volume/EncryptionThreadPool.cpp

@@ -344,21 +344,33 @@ namespace VeraCrypt
 					if (workItem->Type == WorkType::DeriveKey)
 						workItem->KeyDerivation.WorkItem->ItemException.reset (e.CloneNew());
 					else
-						workItem->FirstFragment->ItemException.reset (e.CloneNew());
+					{
+						ScopeLock exceptionLock (FragmentExceptionMutex);
+						if (!workItem->FirstFragment->ItemException.get())
+							workItem->FirstFragment->ItemException.reset (e.CloneNew());
+					}
 				}
 				catch (exception &e)
 				{
 					if (workItem->Type == WorkType::DeriveKey)
 						workItem->KeyDerivation.WorkItem->ItemException.reset (new ExternalException (SRC_POS, StringConverter::ToExceptionString (e)));
 					else
-						workItem->FirstFragment->ItemException.reset (new ExternalException (SRC_POS, StringConverter::ToExceptionString (e)));
+					{
+						ScopeLock exceptionLock (FragmentExceptionMutex);
+						if (!workItem->FirstFragment->ItemException.get())
+							workItem->FirstFragment->ItemException.reset (new ExternalException (SRC_POS, StringConverter::ToExceptionString (e)));
+					}
 				}
 				catch (...)
 				{
 					if (workItem->Type == WorkType::DeriveKey)
 						workItem->KeyDerivation.WorkItem->ItemException.reset (new UnknownException (SRC_POS));
 					else
-						workItem->FirstFragment->ItemException.reset (new UnknownException (SRC_POS));
+					{
+						ScopeLock exceptionLock (FragmentExceptionMutex);
+						if (!workItem->FirstFragment->ItemException.get())
+							workItem->FirstFragment->ItemException.reset (new UnknownException (SRC_POS));
+					}
 				}
 
 				if (workItem->Type == WorkType::DeriveKey)
@@ -409,6 +421,7 @@ namespace VeraCrypt
 
 	Mutex EncryptionThreadPool::EnqueueMutex;
 	Mutex EncryptionThreadPool::DequeueMutex;
+	Mutex EncryptionThreadPool::FragmentExceptionMutex;
 	Mutex EncryptionThreadPool::KeyDerivationCompletionMutex;
 
 	SyncEvent EncryptionThreadPool::WorkItemReadyEvent;

src/Volume/EncryptionThreadPool.h

@@ -111,6 +111,8 @@ namespace VeraCrypt
 		static volatile size_t DequeuePosition;
 		static volatile size_t EnqueuePosition;
 		static Mutex EnqueueMutex;
+		// Fragments of one request share the first fragment's exception pointer.
+		static Mutex FragmentExceptionMutex;
 		// Orders KDF outstanding-count transitions against no-outstanding event updates.
 		static Mutex KeyDerivationCompletionMutex;
 		static list < shared_ptr <Thread> > RunningThreads;