querygraph.hpp

Go to the documentation of this file.

// Copyright (c) 2025 Ryan P. Nicholl <rnicholl@protonmail.com>
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#ifndef NODEGRAPH2_NODEGRAPH_HPP
#define NODEGRAPH2_NODEGRAPH_HPP

 
#include "rpnx/querygraph/byte_format.hpp"
#include "rpnx/result.hpp"
#include "rpnx/typelist.hpp"
 
#include <algorithm>
#include <any>
#include <array>
#include <atomic>
#include <cassert>
#include <chrono>
#include <condition_variable>
#include <concepts>
#include <coroutine>
#include <cstddef>
#include <deque>
#include <exception>
#include <filesystem>
#include <format>
#include <fstream>
#include <functional>
#include <iostream>
#include <iterator>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <print>
#include <set>
#include <shared_mutex>
#include <source_location>
#include <stdexcept>
#include <string>
#include <thread>
#include <tuple>
#include <type_traits>
#include <typeindex>
#include <variant>
#include <vector>
 
#include "rpnx/querygraph/graphdata.hpp"
#include "rpnx/querygraph/hasher.hpp"
#include "rpnx/serialization4.hpp"
 
#ifdef RPNX_QUERYGRAPH_USE_CONC_UNORDERED_MAP
#include "rpnx/sharded_unordered_map.hpp"
#endif
 
#ifdef NDEBUG
#define RPNX_DEBUGONLY(x)
#define RPNX_RELEASEONLY(x) x
#else
#define RPNX_DEBUGONLY(x) x
#define RPNX_RELEASEONLY(x)
#endif
 
 
// TODO: Attach estimated memory class to node spec, to allow for scheduling nodes.
RPNX_ENUM(rpnx::querygraph, memory_class, std::uint8_t, unknown, tiny, small, medium, large, huge);

 
 
namespace rpnx::querygraph
{
    using shared_mutex = std::shared_mutex;
 
    // Win32 std::mutex is rather slow, std::shared_mutex was added later
    // and is generally faster. However, the improvements couldn't be backported to std::mutex
    // because it would break ABI compatibility.
    // On most other platforms, std::mutex is generally faster than std::shared_mutex for exclusive locking.
#ifdef _WIN32
    using fast_mutex = std::shared_mutex;
    using fast_cv = std::condition_variable_any;
#else
    using fast_mutex = std::mutex;
    using fast_cv = std::condition_variable;
#endif

    template < typename T >
    concept query_spec_c = requires {
        typename T::input_type;
        typename T::output_type;
        { T::query_id } -> std::convertible_to< std::string >;
    };

    template < typename T >
    concept subquery_spec_c = requires {
        typename T::parent_query;
        typename T::input_type;
        typename T::output_type;
        requires query_spec_c< typename T::parent_query >;
        { T::subquery_id } -> std::convertible_to< std::string >;
    };
 
    namespace detail
    {
        template < query_spec_c QuerySpec, rpnx::typelist_c Deps, rpnx::typelist_c ProducedSubqueries = rpnx::typelist<> >
        struct query_handler_spec
        {
            using query = QuerySpec;
            using dependencies = Deps;
            using produced_subqueries = ProducedSubqueries;
        };
    } // namespace detail

    template < typename T >
    concept query_handler_spec_struct_c = requires {
        typename T::query;
        typename T::dependencies;
    } && query_spec_c< typename T::query > && rpnx::typelist_c< typename T::dependencies >;

    template < typename T >
    concept query_handler_produced_subqueries_struct_c = requires {
        typename T::produced_subqueries;
    } && rpnx::typelist_c< typename T::produced_subqueries >;

    template < typename T >
    concept query_handler_spec_c = query_handler_spec_struct_c< T >;

    template < typename T >
    struct query_handler_traits;

    template < typename T >
    struct query_handler_produced_subquery_traits
    {
        using produced_subqueries = rpnx::typelist<>;
    };

    template < query_handler_produced_subqueries_struct_c T >
    struct query_handler_produced_subquery_traits< T >
    {
        using produced_subqueries = typename T::produced_subqueries;
    };

    template < query_handler_spec_struct_c T >
    struct query_handler_traits< T >
    {
        using query = typename T::query;
        using dependencies = typename T::dependencies;
        using produced_subqueries = typename query_handler_produced_subquery_traits< T >::produced_subqueries;
    };

    template < query_handler_spec_c HandlerSpec >
    using query_handler_query_t = typename query_handler_traits< HandlerSpec >::query;

    template < query_handler_spec_c HandlerSpec >
    using query_handler_dependencies_t = typename query_handler_traits< HandlerSpec >::dependencies;

    template < query_handler_spec_c HandlerSpec >
    using query_handler_produced_subqueries_t = typename query_handler_traits< HandlerSpec >::produced_subqueries;

    template < query_handler_spec_c HandlerSpec >
    using query_handler_input_t = typename query_handler_query_t< HandlerSpec >::input_type;

    template < query_handler_spec_c HandlerSpec >
    using query_handler_output_t = typename query_handler_query_t< HandlerSpec >::output_type;

    template < typename T >
    struct is_query_handler_spec : std::bool_constant< query_handler_spec_c< T > >
    {
    };

    template < typename T >
    struct binary_traits
    {
        static std::vector< std::byte > serialize_to_binary(T const& value)
        {
            std::vector< std::byte > buffer;
            rpnx::serial4::serialize_iter(value, std::back_inserter(buffer));
            return buffer;
        }
    };
 
    template <>
    struct binary_traits< std::monostate >
    {
        static std::vector< std::byte > serialize_to_binary(std::monostate const&)
        {
            return {};
        }
    };

    template < typename T >
    concept type_with_to_debug_string = requires(T const& t) {
        { t.to_debug_string() } -> std::convertible_to< std::string >;
    };

    template < typename T >
    struct debug_traits
    {
        static std::string to_debug_string(T const& value)
        {
            if constexpr (type_with_to_debug_string< T >)
            {
                return value.to_debug_string();
            }
            else if constexpr (std::formattable< T, char >)
            {
                return std::format("{}", value);
            }
            else
            {
                return "<no debug string available>";
            }
        }
    };
 
    template <>
    struct debug_traits< std::string >
    {
        static std::string to_debug_string(std::string const& value)
        {
            return std::format("{:?}", value);
        }
    };
 
    template <>
    struct debug_traits< std::monostate >
    {
        static std::string to_debug_string(std::monostate const&)
        {
            return "std::monostate{}";
        }
    };

    struct debug_message
    {
        std::string message;
        std::source_location location;
 
      public:
        template < typename... Ts >
        debug_message(std::format_string< Ts... > format_string, Ts&&... args) : message(std::format(format_string, std::forward< Ts >(args)...))
        {
        }

        debug_message(debug_message const& other) = default;
    };

    std::string format_debug_message_with_source(debug_message const& msg);

    struct canonical_error_storage
    {
        tsig m_error_typesig{};
        std::any m_error_value;
        std::function< void(std::any const&) > m_throw_copy;
        std::function< std::vector< std::byte >(std::any const&) > m_serialize;
        std::function< std::string(std::any const&) > m_message;

        void throw_copy() const;

        std::exception_ptr make_exception_ptr() const;

        std::string message() const;

        canonical_error_result_data dump() const;
    };

    template < typename Error >
    canonical_error_storage make_canonical_error_storage(Error const& error)
    {
        static_assert(std::derived_from< Error, std::exception >, "Canonical errors must derive from std::exception");
        static_assert(std::copy_constructible< Error >, "Canonical errors must be copy constructible");
 
        canonical_error_storage storage;
        storage.m_error_typesig = rpnx::serial4::get_type_typesig_hash16< Error >();
        storage.m_error_value.emplace< Error >(error);
        storage.m_throw_copy = [](std::any const& value)
        {
            throw std::any_cast< Error const& >(value);
        };
        storage.m_serialize = [](std::any const& value)
        {
            return binary_traits< Error >::serialize_to_binary(std::any_cast< Error const& >(value));
        };
        storage.m_message = [](std::any const& value)
        {
            return std::string(std::any_cast< Error const& >(value).what());
        };
        return storage;
    }

    class error_result
    {
        std::optional< canonical_error_storage > m_canonical_error;
        std::exception_ptr m_unexpected_exception;
 
      public:
        error_result() = default;

        static error_result canonical(canonical_error_storage error);

        static error_result unexpected(std::exception_ptr exception);

        bool has_error() const;

        bool has_canonical_error() const;

        bool has_unexpected_exception() const;

        status result_status() const;

        std::exception_ptr unexpected_exception() const;

        std::exception_ptr make_exception_ptr() const;

        void throw_error() const;

        std::string message() const;

        std::optional< canonical_error_result_data > canonical_dump() const;
    };

    template < typename T >
    class result
    {
        std::optional< T > m_value;
        error_result m_error;
 
      public:
        result() = default;

        result(T value) : m_value(std::move(value))
        {
        }

        result(std::exception_ptr exception) : m_error(error_result::unexpected(exception))
        {
        }

        void test() const
        {
            if (m_error.has_error())
            {
                m_error.throw_error();
            }
        }

        T& get() &
        {
            test();
            assert(m_value.has_value());
            return m_value.value();
        }

        T const& get() const&
        {
            test();
            assert(m_value.has_value());
            return m_value.value();
        }

        T&& get() &&
        {
            test();
            assert(m_value.has_value());
            return std::move(m_value.value());
        }

        T const&& get() const&&
        {
            test();
            assert(m_value.has_value());
            return std::move(m_value.value());
        }

        T& value() &
        {
            return get();
        }

        T const& value() const&
        {
            return get();
        }

        T&& value() &&
        {
            return std::move(*this).get();
        }

        T const&& value() const&&
        {
            return std::move(*this).get();
        }

        void set_value(T value)
        {
            m_error = error_result();
            m_value = std::move(value);
        }

        void set_error(error_result error)
        {
            m_value.reset();
            m_error = std::move(error);
        }

        void set_error(std::exception_ptr exception)
        {
            set_error(error_result::unexpected(exception));
        }

        void set_exception(std::exception_ptr exception)
        {
            set_error(exception);
        }

        bool has_result() const
        {
            return m_value.has_value() || m_error.has_error();
        }

        bool has_value() const
        {
            return m_value.has_value();
        }

        bool has_error() const
        {
            return m_error.has_error();
        }

        bool has_exception() const
        {
            return has_error();
        }

        bool has_canonical_error() const
        {
            return m_error.has_canonical_error();
        }

        bool has_unexpected_exception() const
        {
            return m_error.has_unexpected_exception();
        }

        error_result const& get_error_result() const
        {
            return m_error;
        }

        std::exception_ptr get_error() const
        {
            return m_error.make_exception_ptr();
        }

        status result_status() const
        {
            if (m_value.has_value())
            {
                return status::completed;
            }
            return m_error.result_status();
        }

        explicit operator bool() const
        {
            return has_result();
        }
    };

    template <>
    class result< void >
    {
        bool m_has_value = false;
        error_result m_error;
 
      public:
        result() = default;

        result(std::exception_ptr exception) : m_error(error_result::unexpected(exception))
        {
        }

        void test() const
        {
            if (m_error.has_error())
            {
                m_error.throw_error();
            }
        }

        void get() const
        {
            test();
            if (!m_has_value)
            {
                throw std::logic_error("No value");
            }
        }

        void value() const
        {
            get();
        }

        void set_value()
        {
            m_error = error_result();
            m_has_value = true;
        }

        void set_error(error_result error)
        {
            m_has_value = false;
            m_error = std::move(error);
        }

        void set_error(std::exception_ptr exception)
        {
            set_error(error_result::unexpected(exception));
        }

        void set_exception(std::exception_ptr exception)
        {
            set_error(exception);
        }

        bool has_result() const
        {
            return m_has_value || m_error.has_error();
        }

        bool has_value() const
        {
            return m_has_value;
        }

        bool has_error() const
        {
            return m_error.has_error();
        }

        bool has_exception() const
        {
            return has_error();
        }

        bool has_canonical_error() const
        {
            return m_error.has_canonical_error();
        }

        bool has_unexpected_exception() const
        {
            return m_error.has_unexpected_exception();
        }

        error_result const& get_error_result() const
        {
            return m_error;
        }

        std::exception_ptr get_error() const
        {
            return m_error.make_exception_ptr();
        }

        status result_status() const
        {
            if (m_has_value)
            {
                return status::completed;
            }
            return m_error.result_status();
        }

        explicit operator bool() const
        {
            return has_result();
        }
    };
 
    inline namespace RPNX_DEBUGONLY(debug_abi) RPNX_RELEASEONLY(release_abi)
    {
        class graph;

        template < query_spec_c QuerySpec >
        struct request
        {
            typename QuerySpec::input_type m_input;
 
          public:
            request(typename QuerySpec::input_type input) : m_input(std::move(input))
            {
            }
        };

        template < query_spec_c QuerySpec >
        struct dependency
        {
            request< QuerySpec > m_request;
 
          public:
            dependency(request< QuerySpec > request) : m_request(std::move(request))
            {
            }
        };

        template < subquery_spec_c SubquerySpec >
        struct subquery_request
        {
            typename SubquerySpec::parent_query::input_type m_parent_input;
            typename SubquerySpec::input_type m_input;
 
          public:
            subquery_request(typename SubquerySpec::parent_query::input_type parent_input, typename SubquerySpec::input_type input) : m_parent_input(std::move(parent_input)), m_input(std::move(input))
            {
            }
        };

        template < subquery_spec_c SubquerySpec >
        struct subquery_result
        {
            typename SubquerySpec::input_type m_input;
            typename SubquerySpec::output_type m_output;
 
          public:
            subquery_result(typename SubquerySpec::input_type input, typename SubquerySpec::output_type output) : m_input(std::move(input)), m_output(std::move(output))
            {
            }
        };

        class subquery_does_not_exist : public std::logic_error
        {
            std::string m_subquery_id;
            std::string m_message;
 
          public:
            RPNX_MEMBER_METADATA(subquery_does_not_exist, m_subquery_id);

            subquery_does_not_exist() : std::logic_error("")
            {
                post_serialize();
            }

            explicit subquery_does_not_exist(std::string subquery_id) : std::logic_error(""), m_subquery_id(std::move(subquery_id))
            {
                post_serialize();
            }

            std::string const& subquery_id() const noexcept
            {
                return m_subquery_id;
            }

            void post_serialize()
            {
                m_message = std::format("Subquery does not exist: {}", m_subquery_id);
            }

            char const* what() const noexcept override
            {
                return m_message.c_str();
            }
        };

        class subquery_parent_failed : public std::runtime_error
        {
            std::string m_subquery_id;
 
          public:
            RPNX_MEMBER_METADATA(subquery_parent_failed, m_subquery_id);

            explicit subquery_parent_failed(std::string subquery_id) : std::runtime_error(std::format("Parent query failed before producing subquery: {}", subquery_id)), m_subquery_id(std::move(subquery_id))
            {
            }

            std::string const& subquery_id() const noexcept
            {
                return m_subquery_id;
            }
        };

        class recursive_dependency_error : public std::logic_error
        {
          public:
            recursive_dependency_error() : std::logic_error("recursive dependency")
            {
            }
        };

        class bad_continuation : public std::logic_error
        {
          public:
            bad_continuation() : std::logic_error("no continuation")
            {
            }
        };
 
        template < typename Result >
        struct query_result;
 
        template < subquery_spec_c SubquerySpec >
        struct subquery_query_result;
 
        template < query_handler_spec_c HandlerSpec >
        struct coroutine;

        class graph
        {
 
            bool m_enable_node_debug_io_capture = false;
            bool m_enable_node_serialization = true;
            std::chrono::nanoseconds m_long_running_node_threshold;
            std::mutex m_diagnostic_mutex;
 
            static constexpr std::size_t long_running_input_max_length = 200;
            static constexpr std::chrono::seconds long_running_progress_minimum = std::chrono::seconds(1);
 
          public:
            explicit graph(std::chrono::nanoseconds long_running_node_threshold = std::chrono::milliseconds(100));

            struct promise_base;
            struct query_node;
 
            struct resumption
            {
                promise_base* m_promise = nullptr;
                std::coroutine_handle< void > handle;
            };
 
            struct runner;
 
            struct alignas(std::hardware_destructive_interference_size) executor
            {
                runner* m_runner = nullptr;
                RPNX_DEBUGONLY(std::thread::id m_id);
                fast_mutex m_mutex;
                std::deque< resumption > m_runnables;
                RPNX_DEBUGONLY(bool m_idle = false;)
 
                static constexpr std::size_t workgroup_threshold = 128;
 
                executor(runner* runner);
 
                executor(executor const& other) = delete;
 
                void check_legal_access();
 
                void run();
 
                void declare_runnable(std::coroutine_handle< void > h, promise_base* ptr, bool immediate);
 
                void declare_runnable(resumption r, bool immediate);
 
                bool try_exec_one();
 
                std::vector< resumption > try_steal(std::size_t max_items);
            };
 
            struct runner
            {
                static std::size_t default_executor_count();

                std::size_t global_workitems_to_publish(std::size_t available_items) const;

                std::size_t global_workitems_to_take(std::size_t queued_work) const;
 
                graph* m_graph = nullptr;
                std::size_t m_total_executors = 0;
                std::atomic< std::size_t > m_idle_executors = 0;
                std::atomic< std::size_t > m_global_workitems = 0;
                std::size_t m_global_workitem_threshold = 0;
                fast_mutex m_mutex;
                std::deque< executor > m_executors;
                std::vector< std::thread > m_executor_threads;
                std::deque< resumption > m_runnables;
                fast_cv m_runner_cv;
 
                static constexpr std::size_t taskgroup_max_size = 1024;
 
                runner(graph* g, std::size_t total_executors = default_executor_count(), std::size_t global_workitem_threshold = 0);
 
                executor* initial_executor();
 
                void run();
 
                void push_group(std::vector< resumption > vals);
 
                void try_steal_work();
 
                std::vector< resumption > get_work(executor* ex);
            };

            template < query_spec_c QuerySpec >
            auto make_request(typename QuerySpec::input_type input) -> typename QuerySpec::output_type
            {
                runner runner(this);
                query_descriptor< QuerySpec >& desc = get_query_descriptor< QuerySpec >();
                query_result< typename QuerySpec::output_type > query_result = desc.m_get(runner.initial_executor(), input, false);
                if (query_result.started_by_current_request())
                {
                    runner.run();
                    if (!query_result.available())
                    {
                        abort_query_node_with_unexpected_exception< typename QuerySpec::output_type >(query_result.m_node, runner.initial_executor(), std::make_exception_ptr(recursive_dependency_error()));
                    }
                }
                query_result.wait();
                return query_result.await_resume();
            }

            template < subquery_spec_c SubquerySpec >
            auto make_subquery_request(typename SubquerySpec::parent_query::input_type parent_input, typename SubquerySpec::input_type input) -> typename SubquerySpec::output_type
            {
                register_subquery_descriptors< SubquerySpec >();
                runner runner(this);
                query_descriptor< typename SubquerySpec::parent_query >& desc = get_query_descriptor< typename SubquerySpec::parent_query >();
                query_result< typename SubquerySpec::parent_query::output_type > parent_result = desc.m_get(runner.initial_executor(), parent_input, false);
                subquery_query_result< SubquerySpec > query_result = get_subquery_result< SubquerySpec >(runner.initial_executor(), parent_result.m_node, input);
                if (parent_result.started_by_current_request())
                {
                    runner.run();
                    if (!query_result.available())
                    {
                        abort_query_node_with_unexpected_exception< typename SubquerySpec::parent_query::output_type >(parent_result.m_node, runner.initial_executor(), std::make_exception_ptr(recursive_dependency_error()));
                    }
                }
                query_result.wait();
                return query_result.await_resume();
            }

            template < query_spec_c QuerySpec >
            auto dump_query_to_file(std::filesystem::path output_path, typename QuerySpec::input_type input) -> std::filesystem::path
            {
                runner runner(this);
                query_descriptor< QuerySpec >& desc = get_query_descriptor< QuerySpec >();
                query_result< typename QuerySpec::output_type > query_result = desc.m_get(runner.initial_executor(), input, false);
                if (query_result.started_by_current_request())
                {
                    runner.run();
                    if (!query_result.available())
                    {
                        abort_query_node_with_unexpected_exception< typename QuerySpec::output_type >(query_result.m_node, runner.initial_executor(), std::make_exception_ptr(recursive_dependency_error()));
                    }
                }
                query_result.wait();
 
                std::vector< std::byte > const marshaled_dump = marshall(query_result.m_node);
                std::ofstream out(output_path, std::ios::binary | std::ios::trunc);
                if (!out)
                {
                    throw std::runtime_error(std::format("Failed to open dump file for writing: {}", output_path.string()));
                }
 
                if (!marshaled_dump.empty())
                {
                    out.write(reinterpret_cast< char const* >(marshaled_dump.data()), static_cast< std::streamsize >(marshaled_dump.size()));
                }
                if (!out)
                {
                    throw std::runtime_error(std::format("Failed to write dump file: {}", output_path.string()));
                }
 
                return output_path;
            }
 
            // struct handler_spec_typeinfo
            // {
            //     std::type_index question_type;
            //     std::vector< std::type_index > dependencies;
            // };
 
            // bindable_type_descriptor is a runtime type descriptor
            // for types that can be used as inputs or outputs
            // to queries.
            // struct bindable_type_descriptor
            // {
            //     std::function< std::vector< std::byte >(std::any const&) > serialize_binary;
            //     std::function< void(std::any&, std::vector< std::byte > const&) > deserialize_binary;
            //     std::function< std::string(std::any const&) > to_string;
            // };
 
            // struct handler_entry_any
            // {
            //     std::type_index type;
            //     std::any handler;
            // };

            template < subquery_spec_c SubquerySpec >
            struct subquery_slot
            {
                std::any m_input;
                std::any m_result;
                std::string m_input_string;
                std::string m_output_string;
                std::vector< resumption > waiters;
            };
 
            template < subquery_spec_c SubquerySpec >
            struct subquery_state
            {
                std::map< typename SubquerySpec::input_type, subquery_slot< SubquerySpec > > m_instances;
            };
 
            struct subquery_state_any
            {
                std::any m_state;
                std::type_index m_input_type = std::type_index(typeid(void));
                std::type_index m_output_type = std::type_index(typeid(void));
                std::string m_subquery_id;
                std::function< std::vector< resumption >(std::any&) > m_finalize_missing;
                std::function< std::vector< resumption >(std::any&, error_result const&) > m_finalize_parent_failed;
                std::function< void(std::any&, std::function< void(std::any&, std::any&) >) > m_iterate_all;
            };
 
            struct query_node
            {
                std::shared_mutex m_mutex;
                std::condition_variable_any m_external_cv;
 
                std::type_index m_query_type = typeid(void);
                std::string m_input_string;
                std::string m_output_string;
 
                // Nodes store a pointer to their own input so the input can be reached from dependency pointers without doing a lookup.
                std::any m_input;
                // m_result stores the output of the node as a result<T> object where T is the output type.
                std::any m_result;
                // m_cort is used for lifetime management of the coroutine object.
                std::any m_cort;
                std::vector< resumption > waiters;
                std::vector< debug_message > m_messages;
                std::set< query_node* > dependencies;
                std::set< query_node* > dependents;
                std::map< std::type_index, subquery_state_any > m_subquery_states;
                std::atomic< std::chrono::nanoseconds::rep > m_self_time = 0;
                std::atomic< bool > m_long_running_threshold_reported = false;
                std::atomic< bool > m_long_running_completion_reported = false;
                std::atomic< bool > m_finished = false;
 
                void unblock_waiters(executor* executor);
 
                void unblock_waiters(executor* executor, std::unique_lock< std::shared_mutex >& lock);
 
                void abort_waiters(executor* executor, std::unique_lock< std::shared_mutex >& lock, std::exception_ptr exception);
            };
 
            static void declare_dependency(query_node* dependent, query_node* dependency);
 
            template < query_spec_c HandlerSpec >
            struct query_descriptor;
 
            struct promise_base
            {
                executor* m_exec = nullptr;
                query_node* m_node = nullptr;
                std::mutex m_run_mutex;
                virtual std::string query_name() const = 0;
                virtual std::string input_text() const = 0;
                virtual void abort_with_unexpected_exception(executor* executor, std::exception_ptr exception) = 0;
                virtual ~promise_base() = default;
            };
 
            template < query_spec_c QuerySpec >
            struct query_descriptor
            {
                std::function< query_result< typename QuerySpec::output_type >(executor*, typename QuerySpec::input_type const&, bool) > m_get;
                std::any m_query_handler_impl;
            };
 
            struct query_descriptor_any
            {
                std::function< void(std::function< void(query_node& node) >) > m_iterate_all;
                std::type_index m_input_type = std::type_index(typeid(void));
                std::type_index m_output_type = std::type_index(typeid(void));
                std::string m_query_id;
            };
 
            template < typename T >
            struct dependency_descriptor_type;
 
            template < query_spec_c QuerySpec >
            struct dependency_descriptor_type< QuerySpec >
            {
                using type = query_descriptor< QuerySpec >*;
            };
 
            template < subquery_spec_c SubquerySpec >
            struct dependency_descriptor_type< SubquerySpec >
            {
                using type = query_descriptor< typename SubquerySpec::parent_query >*;
            };
 
            template < typename T >
            struct dependency_descriptor_table;
 
            template < typename... Ts >
            struct dependency_descriptor_table< rpnx::typelist< Ts... > >
            {
                using type = std::tuple< typename dependency_descriptor_type< Ts >::type... >;
            };
 
            template < rpnx::typelist_c Typelist >
            using dependency_descriptor_table_from_typelist = typename dependency_descriptor_table< Typelist >::type;
 
            template < typename T >
            std::string datavalue_to_string(T const& input)
            {
                auto desc = this->m_text_descriptors.get_or_create(std::type_index(typeid(T)),
                                                                   []()
                                                                   {
                                                                       return text_descriptor{};
                                                                   });
 
                if (desc.m_to_string == nullptr)
                {
                    return "<no to_string available>";
                }
 
                return desc.m_to_string(std::any(input), std::any());
            }
 
            template < query_handler_spec_c HandlerSpec >
            struct query_handler_descriptor_memoized_impl
            {
#ifdef RPNX_QUERYGRAPH_USE_CONC_UNORDERED_MAP
                rpnx::conc_sharded_unordered_map< query_handler_input_t< HandlerSpec >, query_node, hasher< query_handler_input_t< HandlerSpec > > > m_mt_cache;
#else
                std::shared_mutex mutex;
                std::map< query_handler_input_t< HandlerSpec >, query_node > cache;
#endif
 
                query_descriptor< query_handler_query_t< HandlerSpec > >* m_handler_entry = nullptr;
 
                dependency_descriptor_table_from_typelist< query_handler_dependencies_t< HandlerSpec > > dependency_descriptors = {};
 
                std::function< coroutine< HandlerSpec >(query_handler_input_t< HandlerSpec >) > m_startup;
 
#ifdef RPNX_QUERYGRAPH_USE_CONC_UNORDERED_MAP
                query_result< query_handler_output_t< HandlerSpec > > get(executor* exec, query_handler_input_t< HandlerSpec > const& input_ask, bool immediate)
                {
                    std::coroutine_handle< void > cr;
                    bool started = false;
                    promise_base* pbase = nullptr;
                    auto& node = m_mt_cache.get_or_init_iter(input_ask,
                                                             [&](query_handler_input_t< HandlerSpec > const& key, query_node& node)
                                                             {
                                                                 // Identify the query type for the node so UI shows a proper name.
                                                                 node.m_query_type = std::type_index(typeid(query_handler_query_t< HandlerSpec >));
 
                                                                 node.m_input.emplace< query_handler_input_t< HandlerSpec > const* >(&key);
 
                                                                 if (exec->m_runner->m_graph->m_enable_node_debug_io_capture)
                                                                 {
                                                                     node.m_input_string = exec->m_runner->m_graph->datavalue_to_string(input_ask);
                                                                 }
                                                                 coroutine< HandlerSpec > co = m_startup(input_ask);
                                                                 node.m_result = result< query_handler_output_t< HandlerSpec > >();
 
                                                                 co.m_promise->m_node = &node;
                                                                 co.m_promise->m_handler_entry = m_handler_entry;
                                                                 cr = std::coroutine_handle< typename coroutine< HandlerSpec >::promise_type >::from_promise(*co.m_promise);
                                                                 pbase = co.m_promise;
                                                                 node.m_cort = std::make_shared< coroutine< HandlerSpec > >(std::move(co));
 
                                                                 started = true;
                                                             });
                    if (started)
                    {
                        exec->declare_runnable(cr, pbase, immediate);
                    }
 
                    return query_result< query_handler_output_t< HandlerSpec > >(&node, started);
                }
 
                void iterate_all(std::function< void(query_node& node) > func)
                {
                    auto iter = m_mt_cache.range_exclusive();
                    for (auto it = iter.begin(); it != iter.end(); ++it)
                    {
                        func(it->second);
                    }
                }
#else
 
                query_result< query_handler_output_t< HandlerSpec > > get(executor* exec, query_handler_input_t< HandlerSpec > const& input_ask)
                {
                    std::shared_lock lock(mutex);
                    typename std::map< query_handler_input_t< HandlerSpec >, query_node >::iterator it = cache.find(input_ask);
                    if (it != cache.end())
                    {
                        return query_result< query_handler_output_t< HandlerSpec > >(&it->second, false);
                    }
                    lock.unlock();
                    std::unique_lock ulock(mutex);
                    // Double-check pattern
                    it = cache.find(input_ask);
                    if (it != cache.end())
                    {
                        return query_result< query_handler_output_t< HandlerSpec > >(&it->second, false);
                    }
                    auto [node_it, inserted] = cache.try_emplace(input_ask);
                    assert(inserted);
                    coroutine< HandlerSpec > co = m_startup(input_ask);
                    query_node& node = node_it->second;
                    // Identify the query type and record input string (for debug UI) similarly to the concurrent path.
                    node.m_query_type = std::type_index(typeid(query_handler_query_t< HandlerSpec >));
                    node.m_input.emplace< query_handler_input_t< HandlerSpec > const* >(&node_it->first);
                    if (exec->m_runner->m_graph->m_enable_node_debug_io_capture)
                    {
                        node.m_input_string = exec->m_runner->m_graph->datavalue_to_string(input_ask);
                    }
                    node.m_result = result< query_handler_output_t< HandlerSpec > >();
                    co.m_promise->m_node = &node;
                    co.m_promise->m_handler_entry = m_handler_entry;
                    std::coroutine_handle< void > cr = std::coroutine_handle< typename coroutine< HandlerSpec >::promise_type >::from_promise(*co.m_promise);
                    promise_base* pbase = co.m_promise;
                    node.m_cort = std::make_shared< coroutine< HandlerSpec > >(std::move(co));
                    // ulock.unlock();
                    exec->declare_runnable(cr, pbase);
                    return query_result< query_handler_output_t< HandlerSpec > >(&node, true);
                }
#endif
                template < query_spec_c QuerySpec >
                auto get_dependency_descriptor() -> query_descriptor< QuerySpec >&
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Failed to resolve index of dependency descriptor: QuerySpec is not listed in the HandlerSpec dependencies, no entry in dependency_descriptors exists");
 
                    constexpr std::size_t index = *query_handler_dependencies_t< HandlerSpec >::template index_of< QuerySpec >();
 
                    // Descriptor dependencies should be set during the binding stage, which should occur before
                    // any queries are executed
                    return *std::get< index >(dependency_descriptors);
                }
 
                template < subquery_spec_c SubquerySpec >
                auto get_subquery_parent_descriptor() -> query_descriptor< typename SubquerySpec::parent_query >&
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< SubquerySpec >(), "Failed to resolve index of subquery descriptor: SubquerySpec is not listed in HandlerSpec dependencies");
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< typename SubquerySpec::parent_query >(), "Making a subquery request requires declaring the subquery's parent query in HandlerSpec dependencies");
 
                    constexpr std::size_t index = *query_handler_dependencies_t< HandlerSpec >::template index_of< SubquerySpec >();
 
                    return *std::get< index >(dependency_descriptors);
                }
 
                void bind(graph* graph)
                {
                    bind_spec< 0 >(graph);
                }
 
                template < size_t N >
                void bind_spec(graph* graph)
                {
                    if constexpr (N < query_handler_dependencies_t< HandlerSpec >::size)
                    {
                        using dep = query_handler_dependencies_t< HandlerSpec >::template type_at< N >;
                        if constexpr (query_spec_c< dep >)
                        {
                            query_descriptor< dep >& dep_descriptor = graph->get_query_descriptor< dep >();
                            std::get< N >(dependency_descriptors) = &dep_descriptor;
                        }
                        else if constexpr (subquery_spec_c< dep >)
                        {
                            static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< typename dep::parent_query >(), "A handler that depends on a subquery must also list the subquery's parent query in HandlerSpec dependencies");
                            query_descriptor< typename dep::parent_query >& dep_descriptor = graph->get_query_descriptor< typename dep::parent_query >();
                            std::get< N >(dependency_descriptors) = &dep_descriptor;
                        }
                        bind_spec< N + 1 >(graph);
                    }
                }
            };
 
            struct text_descriptor
            {
                std::function< std::string(std::any, std::any) > m_to_string;
            };
 
            struct binary_descriptor
            {
                struct result_binary_data
                {
                    std::vector< std::byte > m_output_data;
                    status m_status = status::pending;
                    std::optional< canonical_error_result_data > m_error_result;
                };
 
                std::function< std::vector< std::byte >(std::any&) > m_serialize_from_cptr;
                std::function< result_binary_data(std::any&) > m_serialize_from_result;
 
                std::function< std::any(std::vector< std::byte > const&) > m_deserialize;
 
                std::array< std::byte, 16 > m_typesig;
            };
 
            struct canonical_error_descriptor
            {
                std::function< error_result(std::exception const&) > m_capture;
            };
 
            std::map< std::type_index, std::any > query_handler_registry;
            std::map< std::type_index, query_descriptor_any > query_descriptor_any_registry;
            std::vector< std::function< void(graph*) > > m_pending_binds;
            rpnx::conc_sharded_unordered_map< std::type_index, text_descriptor, hasher< std::type_index > > m_text_descriptors;
            std::map< std::type_index, binary_descriptor > m_binary_descriptors;
            std::map< std::type_index, canonical_error_descriptor > m_canonical_error_descriptors;
            std::map< std::type_index, std::string > m_query_names;

            auto long_running_node_description(promise_base* promise) -> std::string;

            void print_long_running_node_progress_diagnostic(promise_base* promise);

            void print_long_running_node_completion_diagnostic(promise_base* promise, std::chrono::nanoseconds self_time);

            void print_noncanonical_exception_diagnostic(promise_base* promise, error_result const& error);
 
            static void declare_resumptions(executor* executor, std::vector< resumption > waiters);
 
            static void abort_resumptions(executor* executor, std::vector< resumption > waiters, std::exception_ptr exception);
 
            template < subquery_spec_c SubquerySpec >
            static std::vector< resumption > take_subquery_waiters(subquery_slot< SubquerySpec >& slot)
            {
                std::vector< resumption > waiters;
                waiters.assign(std::make_move_iterator(slot.waiters.begin()), std::make_move_iterator(slot.waiters.end()));
                slot.waiters.clear();
                return waiters;
            }
 
            static canonical_error_storage make_subquery_parent_failed_storage(std::string subquery_id, error_result parent_error);
 
            static error_result make_subquery_does_not_exist_error(std::string subquery_id);
 
            static error_result make_subquery_parent_failed_error(std::string subquery_id, error_result const& parent_error);
 
            template < subquery_spec_c SubquerySpec >
            static auto make_subquery_state_any() -> subquery_state_any
            {
                subquery_state_any state_any;
                state_any.m_state.emplace< subquery_state< SubquerySpec > >();
                state_any.m_input_type = std::type_index(typeid(typename SubquerySpec::input_type));
                state_any.m_output_type = std::type_index(typeid(typename SubquerySpec::output_type));
                state_any.m_subquery_id = std::string(SubquerySpec::subquery_id);
                state_any.m_finalize_missing = [](std::any& state_value) -> std::vector< resumption >
                {
                    std::vector< resumption > waiters;
                    subquery_state< SubquerySpec >& state = std::any_cast< subquery_state< SubquerySpec >& >(state_value);
                    for (auto& [input, slot] : state.m_instances)
                    {
                        result< typename SubquerySpec::output_type >& res = std::any_cast< result< typename SubquerySpec::output_type >& >(slot.m_result);
                        if (!res.has_result())
                        {
                            res.set_error(make_subquery_does_not_exist_error(std::string(SubquerySpec::subquery_id)));
                            std::vector< resumption > slot_waiters = take_subquery_waiters< SubquerySpec >(slot);
                            waiters.insert(waiters.end(), std::make_move_iterator(slot_waiters.begin()), std::make_move_iterator(slot_waiters.end()));
                        }
                    }
                    return waiters;
                };
                state_any.m_finalize_parent_failed = [](std::any& state_value, error_result const& parent_error) -> std::vector< resumption >
                {
                    std::vector< resumption > waiters;
                    subquery_state< SubquerySpec >& state = std::any_cast< subquery_state< SubquerySpec >& >(state_value);
                    for (auto& [input, slot] : state.m_instances)
                    {
                        result< typename SubquerySpec::output_type >& res = std::any_cast< result< typename SubquerySpec::output_type >& >(slot.m_result);
                        if (!res.has_result())
                        {
                            res.set_error(make_subquery_parent_failed_error(std::string(SubquerySpec::subquery_id), parent_error));
                            std::vector< resumption > slot_waiters = take_subquery_waiters< SubquerySpec >(slot);
                            waiters.insert(waiters.end(), std::make_move_iterator(slot_waiters.begin()), std::make_move_iterator(slot_waiters.end()));
                        }
                    }
                    return waiters;
                };
                state_any.m_iterate_all = [](std::any& state_value, std::function< void(std::any&, std::any&) > callback)
                {
                    subquery_state< SubquerySpec >& state = std::any_cast< subquery_state< SubquerySpec >& >(state_value);
                    for (auto& [input, slot] : state.m_instances)
                    {
                        callback(slot.m_input, slot.m_result);
                    }
                };
                return state_any;
            }
 
            template < subquery_spec_c SubquerySpec >
            auto get_or_create_subquery_slot_locked(query_node* parent_node, typename SubquerySpec::input_type const& input) -> subquery_slot< SubquerySpec >&
            {
                std::type_index const state_key = std::type_index(typeid(SubquerySpec));
                auto state_it = parent_node->m_subquery_states.find(state_key);
                if (state_it == parent_node->m_subquery_states.end())
                {
                    auto [inserted_it, inserted] = parent_node->m_subquery_states.emplace(state_key, make_subquery_state_any< SubquerySpec >());
                    assert(inserted);
                    state_it = inserted_it;
                }
 
                subquery_state< SubquerySpec >& state = std::any_cast< subquery_state< SubquerySpec >& >(state_it->second.m_state);
                auto [slot_it, inserted] = state.m_instances.try_emplace(input);
                subquery_slot< SubquerySpec >& slot = slot_it->second;
                if (inserted)
                {
                    slot.m_input.template emplace< typename SubquerySpec::input_type const* >(&slot_it->first);
                    slot.m_result.template emplace< result< typename SubquerySpec::output_type > >();
                    if (node_debug_io_capture_enabled())
                    {
                        slot.m_input_string = datavalue_to_string(input);
                    }
                }
                return slot;
            }
 
            template < subquery_spec_c SubquerySpec >
            std::vector< resumption > complete_subquery_from_parent_result_locked(query_node* parent_node, subquery_slot< SubquerySpec >& slot)
            {
                std::vector< resumption > waiters;
                result< typename SubquerySpec::output_type >& subquery_res = std::any_cast< result< typename SubquerySpec::output_type >& >(slot.m_result);
                if (subquery_res.has_result())
                {
                    return waiters;
                }
 
                result< typename SubquerySpec::parent_query::output_type > const& parent_res = std::any_cast< result< typename SubquerySpec::parent_query::output_type > const& >(parent_node->m_result);
                if (parent_res.has_value())
                {
                    subquery_res.set_error(make_subquery_does_not_exist_error(std::string(SubquerySpec::subquery_id)));
                    waiters = take_subquery_waiters< SubquerySpec >(slot);
                }
                else if (parent_res.has_exception())
                {
                    subquery_res.set_error(make_subquery_parent_failed_error(std::string(SubquerySpec::subquery_id), parent_res.get_error_result()));
                    waiters = take_subquery_waiters< SubquerySpec >(slot);
                }
                return waiters;
            }
 
            template < subquery_spec_c SubquerySpec >
            auto get_subquery_result(executor* executor, query_node* parent_node, typename SubquerySpec::input_type const& input) -> subquery_query_result< SubquerySpec >
            {
                std::unique_lock lock(parent_node->m_mutex);
                subquery_slot< SubquerySpec >& slot = get_or_create_subquery_slot_locked< SubquerySpec >(parent_node, input);
                std::vector< resumption > waiters = complete_subquery_from_parent_result_locked< SubquerySpec >(parent_node, slot);
                bool const abort_waiters = std::any_cast< result< typename SubquerySpec::output_type > const& >(slot.m_result).has_unexpected_exception();
                std::exception_ptr const unexpected_exception = abort_waiters ? std::any_cast< result< typename SubquerySpec::output_type > const& >(slot.m_result).get_error_result().unexpected_exception() : nullptr;
                lock.unlock();
                if (abort_waiters)
                {
                    abort_resumptions(executor, std::move(waiters), unexpected_exception);
                }
                else
                {
                    declare_resumptions(executor, std::move(waiters));
                }
                return subquery_query_result< SubquerySpec >(parent_node, &slot);
            }
 
            template < subquery_spec_c SubquerySpec >
            void set_subquery_result(executor* executor, query_node* parent_node, typename SubquerySpec::input_type input, typename SubquerySpec::output_type output)
            {
                std::unique_lock lock(parent_node->m_mutex);
                subquery_slot< SubquerySpec >& slot = get_or_create_subquery_slot_locked< SubquerySpec >(parent_node, input);
                result< typename SubquerySpec::output_type >& subquery_res = std::any_cast< result< typename SubquerySpec::output_type >& >(slot.m_result);
                if (subquery_res.has_result())
                {
                    throw std::logic_error(std::format("Subquery result already produced: {}", std::string(SubquerySpec::subquery_id)));
                }
                if (node_debug_io_capture_enabled())
                {
                    slot.m_output_string = datavalue_to_string(output);
                }
                subquery_res.set_value(std::move(output));
                std::vector< resumption > waiters = take_subquery_waiters< SubquerySpec >(slot);
                lock.unlock();
                parent_node->m_external_cv.notify_all();
                declare_resumptions(executor, std::move(waiters));
            }
 
            std::vector< resumption > finalize_missing_subqueries_locked(query_node* node);
 
            std::vector< resumption > finalize_parent_failed_subqueries_locked(query_node* node, error_result const& parent_error);
 
            template < subquery_spec_c SubquerySpec >
            void register_subquery_descriptors()
            {
                register_inputoutput_text_descriptor< typename SubquerySpec::input_type >();
                register_inputoutput_text_descriptor< typename SubquerySpec::output_type >();
                register_binary_descriptors< typename SubquerySpec::input_type >();
                register_binary_descriptors< typename SubquerySpec::output_type >();
            }
 
            template < rpnx::typelist_c Typelist, std::size_t N = 0 >
            void register_subquery_descriptors_from_typelist()
            {
                if constexpr (N < Typelist::size)
                {
                    using dep = typename Typelist::template type_at< N >;
                    if constexpr (subquery_spec_c< dep >)
                    {
                        register_subquery_descriptors< dep >();
                    }
                    register_subquery_descriptors_from_typelist< Typelist, N + 1 >();
                }
            }
 
            std::optional< error_result > capture_subquery_parent_failed(subquery_parent_failed const& exception) const;
 
            error_result capture_exception(std::exception_ptr exception) const;
 
            error_result capture_current_exception() const;
 
            template < typename Output >
            void abort_query_node_with_unexpected_exception(query_node* node, executor* executor, std::exception_ptr exception)
            {
                std::vector< resumption > subquery_waiters;
                {
                    std::unique_lock lock(node->m_mutex);
                    result< Output >& res = std::any_cast< result< Output >& >(node->m_result);
                    if (res.has_result())
                    {
                        return;
                    }
                    res.set_error(error_result::unexpected(exception));
                    subquery_waiters = finalize_parent_failed_subqueries_locked(node, res.get_error_result());
                    node->abort_waiters(executor, lock, exception);
                }
                abort_resumptions(executor, std::move(subquery_waiters), exception);
            }
 
          public:
            template < typename Error >
            void register_canonical_error()
            {
                static_assert(std::derived_from< Error, std::exception >, "Canonical errors must derive from std::exception");
                static_assert(std::copy_constructible< Error >, "Canonical errors must be copy constructible");
 
                canonical_error_descriptor descriptor;
                descriptor.m_capture = [](std::exception const& exception)
                {
                    return error_result::canonical(make_canonical_error_storage(static_cast< Error const& >(exception)));
                };
                m_canonical_error_descriptors[std::type_index(typeid(Error))] = std::move(descriptor);
            }

            template < typename T >
            void register_inputoutput_text_descriptor()
            {
                m_text_descriptors.get_or_init(std::type_index(typeid(T)),
                                               [](text_descriptor& desc)
                                               {
                                                   desc.m_to_string = [](std::any val, std::any codec) -> std::string
                                                   {
                                                       return debug_traits< T >::to_debug_string(std::any_cast< T >(val));
                                                   };
                                               });
            }

            template < typename T >
            void register_binary_descriptors()
            {
                binary_descriptor desc;
                desc.m_serialize_from_cptr = [](std::any& val) -> std::vector< std::byte >
                {
                    T const* value = std::any_cast< T const* >(val);
                    return binary_traits< T >::serialize_to_binary(*value);
                };
                desc.m_serialize_from_result = [](std::any& val) -> binary_descriptor::result_binary_data
                {
                    result< T >& res = std::any_cast< result< T >& >(val);
                    if (res.has_value())
                    {
                        std::vector< std::byte > data = binary_traits< T >::serialize_to_binary(res.value());
                        return binary_descriptor::result_binary_data{
                            .m_output_data = std::move(data),
                            .m_status = status::completed,
                            .m_error_result = std::nullopt,
                        };
                    }
                    if (res.has_exception())
                    {
                        std::vector< std::byte > data;
                        std::string str = "Exception: ";
                        str += res.get_error_result().message();
                        for (auto c : str)
                        {
                            data.push_back(std::byte(c));
                        }
                        return binary_descriptor::result_binary_data{
                            .m_output_data = std::move(data),
                            .m_status = res.result_status(),
                            .m_error_result = res.get_error_result().canonical_dump(),
                        };
                    }
                    return binary_descriptor::result_binary_data{
                        .m_output_data = {},
                        .m_status = status::pending,
                        .m_error_result = std::nullopt,
                    };
                };
                desc.m_deserialize = [](std::vector< std::byte > const& data) -> std::any
                {
                    T value{};
                    rpnx::serial4::deserialize_iter(value, data.begin(), data.end());
                    return std::any(std::move(value));
                };
                desc.m_typesig = rpnx::serial4::get_type_typesig_hash16< T >();
 
                m_binary_descriptors[std::type_index(typeid(T))] = desc;
            }

            template < query_handler_spec_c QuerySpec >
            void register_handler_map(std::map< query_handler_input_t< QuerySpec >, query_handler_output_t< QuerySpec > > handler_map)
            {
                // TODO: Validate that this works.
                register_handler_function< QuerySpec >(
                    [handler_map = std::make_shared< std::map< query_handler_input_t< QuerySpec >, query_handler_output_t< QuerySpec > > >(std::move(handler_map))](query_handler_input_t< QuerySpec > const& input) -> coroutine< QuerySpec >
                    {
                        auto it = handler_map->find(input);
                        if (it == handler_map->end())
                        {
                            throw std::logic_error("No handler found for the given input in handler map");
                        }
                        co_return it->second;
                    });
            }

            template < query_handler_spec_c HandlerSpec >
                requires std::same_as< query_handler_input_t< HandlerSpec >, std::monostate >
            void register_handler_singleton(query_handler_output_t< HandlerSpec > value)
            {
                using output_type = query_handler_output_t< HandlerSpec >;
 
                register_handler_function< HandlerSpec >(
                    [value = std::make_shared< output_type >(std::move(value))](std::monostate const&) -> coroutine< HandlerSpec >
                    {
                        co_return *value;
                    });
            }

            template < query_spec_c QuerySpec >
                requires std::same_as< typename QuerySpec::input_type, std::monostate >
            void register_handler_singleton(typename QuerySpec::output_type value)
            {
                register_handler_singleton< detail::query_handler_spec< QuerySpec, rpnx::typelist<> > >(std::move(value));
            }

            template < query_handler_spec_c HandlerSpec, typename Handler >
            void register_handler_function(Handler h)
            {
                auto query_index = std::type_index(typeid(query_handler_query_t< HandlerSpec >));
                if (query_handler_registry.find(query_index) != query_handler_registry.end())
                {
                    throw std::logic_error("Handler already registered");
                }
                query_handler_registry[query_index].emplace< query_descriptor< query_handler_query_t< HandlerSpec > > >();
                query_descriptor_any& qd_any = query_descriptor_any_registry[query_index];
 
                query_descriptor< query_handler_query_t< HandlerSpec > >& descriptor = std::any_cast< query_descriptor< query_handler_query_t< HandlerSpec > >& >(query_handler_registry[query_index]);
                descriptor.m_query_handler_impl = std::make_shared< query_handler_descriptor_memoized_impl< HandlerSpec > >();
                query_handler_descriptor_memoized_impl< HandlerSpec >& memoized_impl = *std::any_cast< std::shared_ptr< query_handler_descriptor_memoized_impl< HandlerSpec > >& >(descriptor.m_query_handler_impl);
                memoized_impl.m_startup = std::move(h);
                memoized_impl.m_handler_entry = &descriptor;
 
                // This is safe because the memoized_impl lifetime is tied to the query_handler_descriptor lifetime
                descriptor.m_get = [&memoized_impl](executor* exec, query_handler_input_t< HandlerSpec > const& input_ask, bool immediate)
                {
                    return memoized_impl.get(exec, input_ask, immediate);
                };
 
                qd_any.m_iterate_all = [&memoized_impl](std::function< void(query_node&) > callback)
                {
                    memoized_impl.iterate_all(std::move(callback));
                };
 
                qd_any.m_input_type = std::type_index(typeid(query_handler_input_t< HandlerSpec >));
                qd_any.m_output_type = std::type_index(typeid(query_handler_output_t< HandlerSpec >));
                qd_any.m_query_id = query_handler_query_t< HandlerSpec >::query_id;
 
                register_inputoutput_text_descriptor< query_handler_input_t< HandlerSpec > >();
                register_inputoutput_text_descriptor< query_handler_output_t< HandlerSpec > >();
                register_binary_descriptors< query_handler_input_t< HandlerSpec > >();
                register_binary_descriptors< query_handler_output_t< HandlerSpec > >();
                register_subquery_descriptors_from_typelist< query_handler_dependencies_t< HandlerSpec > >();
                register_subquery_descriptors_from_typelist< query_handler_produced_subqueries_t< HandlerSpec > >();
 
                m_query_names[std::type_index(typeid(query_handler_query_t< HandlerSpec >))] = query_handler_query_t< HandlerSpec >::query_id;
 
                m_pending_binds.push_back(
                    [&memoized_impl](graph* exec)
                    {
                        memoized_impl.bind(exec);
                    });
            }

            template < query_spec_c QuerySpec >
            auto get_query_descriptor() -> query_descriptor< QuerySpec >&
            {
                auto it = query_handler_registry.find(std::type_index(typeid(QuerySpec)));
                if (it == query_handler_registry.end())
                {
                    throw std::logic_error("No handler registered for the requested query spec");
                }
                return std::any_cast< query_descriptor< QuerySpec >& >(it->second);
            }

            void bind_handlers();

            template < typename Visitor >
            void for_each_marshaled_type_descriptor(Visitor visitor)
            {
                for (auto const& [type_index, binary_desc] : m_binary_descriptors)
                {
                    auto text_desc = m_text_descriptors.get_or_create(type_index,
                                                                      []()
                                                                      {
                                                                          return text_descriptor{};
                                                                      });
 
                    visitor(binary_desc.m_typesig, binary_desc.m_deserialize, text_desc.m_to_string);
                }
            }

            graph_data dump();

            graph_data dump(query_node* start_node);

            std::vector< std::byte > marshall();

            std::vector< std::byte > marshall(query_node* start_node);

            bool node_debug_io_capture_enabled() const noexcept;
        };
 
        template < typename Result >
        struct query_result
        {
            graph::query_node* m_node;
            bool m_started_by_current_request = false;
 
          public:
            explicit query_result(graph::query_node* node_ptr, bool started_by_current_request = false) : m_node(node_ptr), m_started_by_current_request(started_by_current_request)
            {
            }
 
            bool started_by_current_request() const noexcept
            {
                return m_started_by_current_request;
            }
 
            bool available() const
            {
                std::shared_lock lock(m_node->m_mutex);
                result< Result > const& res = std::any_cast< result< Result > const& >(m_node->m_result);
                return res.has_result();
            }
 
            void wait()
            {
                std::unique_lock lock(m_node->m_mutex);
                m_node->m_external_cv.wait(lock,
                                           [this]()
                                           {
                                               result< Result > const& res = std::any_cast< result< Result > const& >(m_node->m_result);
                                               return res.has_result();
                                           });
            }
 
            bool await_ready()
            {
                std::shared_lock lock(m_node->m_mutex);
                result< Result > const& res = std::any_cast< result< Result > const& >(m_node->m_result);
                return res.has_result() && !res.has_unexpected_exception();
            }
 
            bool await_ready([[maybe_unused]] std::unique_lock< std::shared_mutex >& lock)
            {
                return std::any_cast< result< Result >& >(m_node->m_result).has_result();
            }
 
            bool await_ready([[maybe_unused]] std::shared_lock< std::shared_mutex >& lock)
            {
                return std::any_cast< result< Result > const& >(m_node->m_result).has_result();
            }
 
            template < typename U >
            void await_suspend(std::coroutine_handle< U > h)
            {
                await_suspend_helper< typename U::coroutine_type >(h.promise(), h);
            }
 
            template < typename C >
            void await_suspend_helper(typename C::promise_type& p, std::coroutine_handle< typename C::promise_type > h)
            {
                await_suspend_helper2< C, typename C::spec_type >(p, h);
            }
 
            template < typename C, typename HandlerSpec >
            void await_suspend_helper2(typename C::promise_type& p, std::coroutine_handle< typename C::promise_type > h)
            {
                std::unique_lock lock(m_node->m_mutex);
 
                graph::executor* current_executor = p.m_exec;
                assert(current_executor != nullptr);
                current_executor->check_legal_access();
 
                if (await_ready(lock))
                {
                    result< Result > const& res = std::any_cast< result< Result > const& >(m_node->m_result);
                    if (res.has_unexpected_exception())
                    {
                        std::exception_ptr unexpected_exception = res.get_error_result().unexpected_exception();
                        lock.unlock();
                        p.abort_with_unexpected_exception(current_executor, unexpected_exception);
                    }
                    else
                    {
                        current_executor->declare_runnable(graph::resumption{.m_promise = &p, .handle = h}, false);
                    }
                }
                else
                {
                    m_node->waiters.push_back(graph::resumption{.m_promise = &p, .handle = h});
                }
            }
 
            Result const& await_resume()
            {
                std::shared_lock lock(m_node->m_mutex);
                auto& res = std::any_cast< result< Result > const& >(m_node->m_result);
                return res.value();
            }
        };
 
        template < subquery_spec_c SubquerySpec >
        struct subquery_query_result
        {
            graph::query_node* m_parent_node;
            graph::subquery_slot< SubquerySpec >* m_slot;
 
          public:
            subquery_query_result(graph::query_node* parent_node, graph::subquery_slot< SubquerySpec >* slot) : m_parent_node(parent_node), m_slot(slot)
            {
            }
 
            bool available() const
            {
                std::shared_lock lock(m_parent_node->m_mutex);
                result< typename SubquerySpec::output_type > const& res = std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result);
                return res.has_result();
            }
 
            void wait()
            {
                std::unique_lock lock(m_parent_node->m_mutex);
                m_parent_node->m_external_cv.wait(lock,
                                                  [this]()
                                                  {
                                                      result< typename SubquerySpec::output_type > const& res = std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result);
                                                      return res.has_result();
                                                  });
            }
 
            bool await_ready()
            {
                std::shared_lock lock(m_parent_node->m_mutex);
                result< typename SubquerySpec::output_type > const& res = std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result);
                return res.has_result() && !res.has_unexpected_exception();
            }
 
            bool await_ready([[maybe_unused]] std::unique_lock< std::shared_mutex >& lock)
            {
                return std::any_cast< result< typename SubquerySpec::output_type >& >(m_slot->m_result).has_result();
            }
 
            bool await_ready([[maybe_unused]] std::shared_lock< std::shared_mutex >& lock)
            {
                return std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result).has_result();
            }
 
            template < typename U >
            void await_suspend(std::coroutine_handle< U > h)
            {
                await_suspend_helper< typename U::coroutine_type >(h.promise(), h);
            }
 
            template < typename C >
            void await_suspend_helper(typename C::promise_type& p, std::coroutine_handle< typename C::promise_type > h)
            {
                std::unique_lock lock(m_parent_node->m_mutex);
 
                graph::executor* current_executor = p.m_exec;
                assert(current_executor != nullptr);
                current_executor->check_legal_access();
 
                if (await_ready(lock))
                {
                    result< typename SubquerySpec::output_type > const& res = std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result);
                    if (res.has_unexpected_exception())
                    {
                        std::exception_ptr unexpected_exception = res.get_error_result().unexpected_exception();
                        lock.unlock();
                        p.abort_with_unexpected_exception(current_executor, unexpected_exception);
                    }
                    else
                    {
                        current_executor->declare_runnable(graph::resumption{.m_promise = &p, .handle = h}, false);
                    }
                }
                else
                {
                    m_slot->waiters.push_back(graph::resumption{.m_promise = &p, .handle = h});
                }
            }
 
            typename SubquerySpec::output_type const& await_resume()
            {
                std::shared_lock lock(m_parent_node->m_mutex);
                result< typename SubquerySpec::output_type > const& res = std::any_cast< result< typename SubquerySpec::output_type > const& >(m_slot->m_result);
                return res.value();
            }
        };

        template < query_handler_spec_c HandlerSpec, typename Ret >
        class cosubroutine_impl;

        template < query_handler_spec_c HandlerSpec >
        struct coroutine
        {
            using spec_type = HandlerSpec;

            template < typename Ret >
            using cosubroutine = cosubroutine_impl< HandlerSpec, Ret >;

            struct promise_type : public graph::promise_base
            {
                using root_coroutine_type = coroutine< HandlerSpec >;
                using coroutine_type = coroutine< HandlerSpec >;
 
                typename graph::template query_descriptor< query_handler_query_t< HandlerSpec > >* m_handler_entry = nullptr;
                coroutine< HandlerSpec >* m_coroutine = nullptr;
 
                std::string query_name() const override
                {
                    return std::string(query_handler_query_t< HandlerSpec >::query_id);
                }
 
                std::string input_text() const override
                {
                    using input_type = query_handler_input_t< HandlerSpec >;
                    input_type const* input = std::any_cast< input_type const* >(m_node->m_input);
                    return m_exec->m_runner->m_graph->datavalue_to_string(*input);
                }
 
                template < query_spec_c QuerySpec >
                auto await_transform(request< QuerySpec > a) -> query_result< typename QuerySpec::output_type >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
 
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, a.m_input, false);
 
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return await_obj;
                }
 
                template < subquery_spec_c SubquerySpec >
                auto await_transform(subquery_request< SubquerySpec > a) -> subquery_query_result< SubquerySpec >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< SubquerySpec >(), "Making a subquery request requires listing the subquery in query_handler_dependencies_t< HandlerSpec >");
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< typename SubquerySpec::parent_query >(), "Making a subquery request requires listing the subquery parent query in query_handler_dependencies_t< HandlerSpec >");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< typename SubquerySpec::parent_query >& parent_query_descriptor = handler_impl.template get_subquery_parent_descriptor< SubquerySpec >();
 
                    query_result< typename SubquerySpec::parent_query::output_type > parent_await_obj = parent_query_descriptor.m_get(m_exec, a.m_parent_input, false);
 
                    graph::query_node* parent_node = parent_await_obj.m_node;
                    graph::declare_dependency(m_node, parent_node);
                    return m_exec->m_runner->m_graph->template get_subquery_result< SubquerySpec >(m_exec, parent_node, a.m_input);
                }
 
                template < query_spec_c QuerySpec >
                auto yield_value(dependency< QuerySpec > dep)
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, dep.m_request.m_input, true);
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return std::suspend_never{};
                }
 
                template < subquery_spec_c SubquerySpec >
                auto yield_value(subquery_result< SubquerySpec > res)
                {
                    static_assert(query_handler_produced_subqueries_t< HandlerSpec >::template contains< SubquerySpec >(), "Yielding a subquery result requires listing the subquery in query_handler_produced_subqueries_t< HandlerSpec >");
                    static_assert(std::same_as< typename SubquerySpec::parent_query, query_handler_query_t< HandlerSpec > >, "A handler can only yield subqueries whose parent_query is the handler query");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    m_exec->m_runner->m_graph->template set_subquery_result< SubquerySpec >(m_exec, m_node, std::move(res.m_input), std::move(res.m_output));
                    return std::suspend_never{};
                }
 
                template < typename Ret >
                auto await_transform(cosubroutine< Ret >&& sub)
                {
                    assert(m_node != nullptr);
                    auto* sub_promise = sub.m_promise;
                    sub_promise->m_node = m_node;
                    sub_promise->m_handler_entry = m_handler_entry;
                    m_exec->declare_runnable(std::coroutine_handle< typename cosubroutine< Ret >::promise_type >::from_promise(*sub_promise), sub_promise, false);
                    return typename cosubroutine< Ret >::awaitable(sub_promise);
                }
 
                void return_value(query_handler_output_t< HandlerSpec > value)
                {
                    assert(m_node != nullptr);
                    std::unique_lock lock(m_node->m_mutex);
                    result< query_handler_output_t< HandlerSpec > >& res = std::any_cast< result< query_handler_output_t< HandlerSpec > >& >(m_node->m_result);
                    if (m_exec->m_runner->m_graph->node_debug_io_capture_enabled())
                    {
                        m_node->m_output_string = m_exec->m_runner->m_graph->datavalue_to_string(value);
                    }
                    res.set_value(std::move(value));
                    std::vector< graph::resumption > subquery_waiters = m_exec->m_runner->m_graph->finalize_missing_subqueries_locked(m_node);
                    m_node->unblock_waiters(m_exec, lock);
                    graph::declare_resumptions(m_exec, std::move(subquery_waiters));
                }
 
                auto initial_suspend()
                {
                    return std::suspend_always{};
                }
 
                auto final_suspend() noexcept
                {
                    std::unique_lock lock(m_node->m_mutex);
                    result< query_handler_output_t< HandlerSpec > >& res = std::any_cast< result< query_handler_output_t< HandlerSpec > >& >(m_node->m_result);
                    if (!res.has_result() && m_node->dependencies.empty())
                    {
                        std::exception_ptr exception = std::make_exception_ptr(bad_continuation());
                        res.set_error(error_result::unexpected(exception));
                        std::vector< graph::resumption > subquery_waiters = m_exec->m_runner->m_graph->finalize_parent_failed_subqueries_locked(m_node, res.get_error_result());
                        m_node->abort_waiters(m_exec, lock, exception);
                        graph::abort_resumptions(m_exec, std::move(subquery_waiters), exception);
                    }
                    return std::suspend_always{};
                }
 
                auto get_return_object()
                {
                    coroutine< HandlerSpec > co(this);
                    assert(m_coroutine == &co);
                    assert(co.m_promise == this);
                    return co;
                }
 
                auto yield_value(debug_message msg, std::source_location loc = std::source_location::current())
                {
                    assert(m_node != nullptr);
                    std::unique_lock lock(m_node->m_mutex);
                    msg.location = loc;
                    m_node->m_messages.push_back(std::move(msg));
                    return std::suspend_never{};
                }
 
                auto unhandled_exception()
                {
                    assert(m_node != nullptr);
                    assert(m_exec != nullptr);
                    std::unique_lock lock(m_node->m_mutex);
                    result< query_handler_output_t< HandlerSpec > >& res = std::any_cast< result< query_handler_output_t< HandlerSpec > >& >(m_node->m_result);
                    error_result captured_error = m_exec->m_runner->m_graph->capture_current_exception();
                    std::exception_ptr unexpected_exception = captured_error.unexpected_exception();
                    if (unexpected_exception != nullptr)
                    {
                        m_exec->m_runner->m_graph->print_noncanonical_exception_diagnostic(this, captured_error);
                    }
                    res.set_error(std::move(captured_error));
                    std::vector< graph::resumption > subquery_waiters = m_exec->m_runner->m_graph->finalize_parent_failed_subqueries_locked(m_node, res.get_error_result());
                    if (unexpected_exception != nullptr)
                    {
                        m_node->abort_waiters(m_exec, lock, unexpected_exception);
                        graph::abort_resumptions(m_exec, std::move(subquery_waiters), unexpected_exception);
                    }
                    else
                    {
                        m_node->unblock_waiters(m_exec, lock);
                        graph::declare_resumptions(m_exec, std::move(subquery_waiters));
                    }
                }
 
                void abort_with_unexpected_exception(graph::executor* executor, std::exception_ptr exception) override
                {
                    this->m_exec = executor;
                    assert(m_node != nullptr);
                    m_exec->m_runner->m_graph->template abort_query_node_with_unexpected_exception< query_handler_output_t< HandlerSpec > >(m_node, executor, exception);
                }
 
                template < typename Res >
                void await_suspend_on(query_result< Res > q)
                {
                    std::unique_lock lock(q.m_node->m_mutex);
 
                    graph::executor* current_executor = m_exec;
                    assert(current_executor != nullptr);
                    current_executor->check_legal_access();
                    auto h = std::coroutine_handle< promise_type >::from_promise(*this);
 
                    if (q.await_ready(lock))
                    {
                        result< Res > const& res = std::any_cast< result< Res > const& >(q.m_node->m_result);
                        if (res.has_unexpected_exception())
                        {
                            std::exception_ptr unexpected_exception = res.get_error_result().unexpected_exception();
                            lock.unlock();
                            abort_with_unexpected_exception(current_executor, unexpected_exception);
                        }
                        else
                        {
                            lock.unlock();
                            current_executor->declare_runnable(graph::resumption{.m_promise = this, .handle = h}, false);
                        }
                    }
                    else
                    {
                        q.m_node->waiters.push_back(graph::resumption{.m_promise = this, .handle = h});
                    }
                }
            };

            promise_type* m_promise = nullptr;
 
            coroutine() noexcept : m_promise(nullptr)
            {
            }
 
            coroutine(coroutine&) = delete;
 
            coroutine(coroutine&& other) noexcept : m_promise(other.m_promise)
            {
                m_promise = other.m_promise;
                other.m_promise = nullptr;
                if (m_promise)
                {
                    m_promise->m_coroutine = this;
                }
            }
 
            ~coroutine()
            {
                if (m_promise)
                {
                    m_promise->m_coroutine = nullptr;
                    std::coroutine_handle< promise_type >::from_promise(*m_promise).destroy();
                }
            }
 
            coroutine(promise_type* p) : m_promise(p)
            {
                assert(p->m_coroutine == nullptr);
                p->m_coroutine = this;
            }
        };

        template < query_handler_spec_c HandlerSpec, typename Ret >
        class cosubroutine_impl
        {
            friend struct promise_type;
            friend struct coroutine< HandlerSpec >;
            template < query_handler_spec_c OtherHandlerSpec, typename OtherRet >
            friend class cosubroutine_impl;
 
          public:
            using spec_type = HandlerSpec;
            struct promise_type : public graph::promise_base
            {
                using root_coroutine_type = coroutine< HandlerSpec >;
                using coroutine_type = cosubroutine_impl< HandlerSpec, Ret >;
 
                std::mutex m_mutex;
 
                typename graph::template query_descriptor< query_handler_query_t< HandlerSpec > >* m_handler_entry = nullptr;
                cosubroutine_impl* m_srt = nullptr;
                result< Ret > m_result;
                std::optional< graph::resumption > m_waiter;
 
                std::string query_name() const override
                {
                    return std::string(query_handler_query_t< HandlerSpec >::query_id);
                }
 
                std::string input_text() const override
                {
                    using input_type = query_handler_input_t< HandlerSpec >;
                    input_type const* input = std::any_cast< input_type const* >(m_node->m_input);
                    return m_exec->m_runner->m_graph->datavalue_to_string(*input);
                }
 
                auto get_return_object()
                {
                    return cosubroutine_impl(this);
                }
 
                auto initial_suspend()
                {
                    return std::suspend_always{};
                }
 
                void return_value(Ret value)
                {
                    assert(m_node != nullptr);
                    assert(this->m_exec != nullptr);
 
                    {
                        std::unique_lock lock(m_mutex);
                        m_result.set_value(std::move(value));
                        unblock_waiter(lock);
                    }
                }
 
                auto yield_value(debug_message msg, std::source_location loc = std::source_location::current())
                {
                    assert(m_node != nullptr);
                    std::unique_lock lock(m_node->m_mutex);
                    msg.location = loc;
                    m_node->m_messages.push_back(std::move(msg));
                    return std::suspend_never{};
                }

                template < query_spec_c QuerySpec >
                auto yield_value(dependency< QuerySpec > dep)
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, dep.m_request.m_input, true);
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return std::suspend_never{};
                }
 
                auto final_suspend() noexcept
                {
                    std::unique_lock lock(m_mutex);
                    assert(m_result.has_result());
                    return std::suspend_always{};
                }
 
                auto unhandled_exception()
                {
                    std::unique_lock lock(m_mutex);
                    assert(m_node != nullptr);
                    assert(this->m_exec != nullptr);
                    m_result.set_error(m_exec->m_runner->m_graph->capture_current_exception());
                    unblock_waiter(lock);
                }
 
                void abort_with_unexpected_exception(graph::executor* executor, std::exception_ptr exception) override
                {
                    this->m_exec = executor;
                    {
                        std::unique_lock lock(m_mutex);
                        if (!m_result.has_result())
                        {
                            m_result.set_error(error_result::unexpected(exception));
                        }
                    }
                    assert(m_node != nullptr);
                    m_exec->m_runner->m_graph->template abort_query_node_with_unexpected_exception< query_handler_output_t< HandlerSpec > >(m_node, executor, exception);
                }
 
                auto unblock_waiter(std::unique_lock< std::mutex >& lock)
                {
                    if (m_waiter.has_value())
                    {
                        auto resumption = m_waiter.value();
                        m_waiter.reset();
                        m_exec->check_legal_access();
                        m_exec->declare_runnable(resumption, false);
 
                        lock.unlock();
                    }
                }
 
                template < typename Ret2 >
                auto await_transform(typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >&& sub)
                {
                    assert(m_node != nullptr);
                    sub.m_promise->m_node = m_node;
                    sub.m_promise->m_handler_entry = m_handler_entry;
                    m_exec->declare_runnable(std::coroutine_handle< typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >::promise_type >::from_promise(*sub.m_promise), sub.m_promise, false);
                    return typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >::awaitable(sub.m_promise);
                }
 
                template < query_spec_c QuerySpec >
                auto await_transform(request< QuerySpec > a) -> query_result< typename QuerySpec::output_type >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
 
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, a.m_input, false);
 
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return await_obj;
                }
 
                template < subquery_spec_c SubquerySpec >
                auto await_transform(subquery_request< SubquerySpec > a) -> subquery_query_result< SubquerySpec >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< SubquerySpec >(), "Making a subquery request requires listing the subquery in query_handler_dependencies_t< HandlerSpec >");
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< typename SubquerySpec::parent_query >(), "Making a subquery request requires listing the subquery parent query in query_handler_dependencies_t< HandlerSpec >");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< typename SubquerySpec::parent_query >& parent_query_descriptor = handler_impl.template get_subquery_parent_descriptor< SubquerySpec >();
 
                    query_result< typename SubquerySpec::parent_query::output_type > parent_await_obj = parent_query_descriptor.m_get(m_exec, a.m_parent_input, false);
 
                    graph::query_node* parent_node = parent_await_obj.m_node;
                    graph::declare_dependency(m_node, parent_node);
                    return m_exec->m_runner->m_graph->template get_subquery_result< SubquerySpec >(m_exec, parent_node, a.m_input);
                }
 
                template < subquery_spec_c SubquerySpec >
                auto yield_value(subquery_result< SubquerySpec > res)
                {
                    static_assert(query_handler_produced_subqueries_t< HandlerSpec >::template contains< SubquerySpec >(), "Yielding a subquery result requires listing the subquery in query_handler_produced_subqueries_t< HandlerSpec >");
                    static_assert(std::same_as< typename SubquerySpec::parent_query, query_handler_query_t< HandlerSpec > >, "A handler can only yield subqueries whose parent_query is the handler query");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    m_exec->m_runner->m_graph->template set_subquery_result< SubquerySpec >(m_exec, m_node, std::move(res.m_input), std::move(res.m_output));
                    return std::suspend_never{};
                }
 
                ~promise_type()
                {
                    assert(m_srt == nullptr);
                }
            };
 
            struct awaitable
            {
                promise_type* m_promise;
 
              public:
                awaitable(promise_type* promise) : m_promise(promise)
                {
                }
 
                bool await_ready()
                {
                    std::unique_lock lck(m_promise->m_mutex);
                    return await_ready(lck);
                }
 
                bool await_ready([[maybe_unused]] std::unique_lock< std::mutex >& lock) noexcept
                {
                    assert(m_promise != nullptr);
                    return m_promise->m_result.has_result();
                }
 
                Ret await_resume()
                {
                    std::unique_lock lck(m_promise->m_mutex);
                    assert(m_promise != nullptr);
                    assert(m_promise->m_result.has_result());
                    return m_promise->m_result.value();
                }
 
                template < typename U >
                void await_suspend(std::coroutine_handle< U > h)
                {
                    await_suspend_helper< typename U::coroutine_type >(h.promise(), h);
                }
 
                template < typename C >
                void await_suspend_helper(typename C::promise_type& p, std::coroutine_handle< typename C::promise_type > h)
                {
                    assert(m_promise != nullptr);
 
                    auto current_executor = p.m_exec;
                    assert(current_executor != nullptr);
                    current_executor->check_legal_access();
 
                    graph::resumption r{.m_promise = &h.promise(), .handle = h};
                    std::unique_lock lck(m_promise->m_mutex);
 
                    if (await_ready(lck))
                    {
                        p.m_exec->declare_runnable(r, false);
                    }
                    else
                    {
                        assert(m_promise->m_waiter.has_value() == false);
                        m_promise->m_waiter = r;
                    }
                }
            };
 
          private:
            promise_type* m_promise;
 
          public:
            cosubroutine_impl(promise_type* p) : m_promise(p)
            {
                assert(p->m_srt == nullptr);
                p->m_srt = this;
            }
 
            cosubroutine_impl(const cosubroutine_impl&) = delete;
            cosubroutine_impl& operator=(const cosubroutine_impl&) = delete;
 
            ~cosubroutine_impl()
            {
                if (m_promise)
                {
                    std::unique_lock lock(m_promise->m_run_mutex);
                    lock.unlock();
 
                    m_promise->m_srt = nullptr;
                    std::coroutine_handle< promise_type >::from_promise(*m_promise).destroy();
                }
            }
        };
 
        template < query_handler_spec_c HandlerSpec >
        class cosubroutine_impl< HandlerSpec, void >
        {
            friend struct promise_type;
            friend struct coroutine< HandlerSpec >;
            template < query_handler_spec_c OtherHandlerSpec, typename OtherRet >
            friend class cosubroutine_impl;
 
          public:
            using spec_type = HandlerSpec;
            struct promise_type : public graph::promise_base
            {
                using root_coroutine_type = coroutine< HandlerSpec >;
                using coroutine_type = cosubroutine_impl< HandlerSpec, void >;
 
                std::mutex m_mutex;
 
                typename graph::template query_descriptor< query_handler_query_t< HandlerSpec > >* m_handler_entry = nullptr;
                cosubroutine_impl* m_srt = nullptr;
                result< void > m_result;
                std::optional< graph::resumption > m_waiter;
 
                std::string query_name() const override
                {
                    return std::string(query_handler_query_t< HandlerSpec >::query_id);
                }
 
                std::string input_text() const override
                {
                    using input_type = query_handler_input_t< HandlerSpec >;
                    input_type const* input = std::any_cast< input_type const* >(m_node->m_input);
                    return m_exec->m_runner->m_graph->datavalue_to_string(*input);
                }
 
                auto get_return_object()
                {
                    return cosubroutine_impl(this);
                }
 
                auto initial_suspend()
                {
                    return std::suspend_always{};
                }
 
                void return_void()
                {
                    assert(m_node != nullptr);
                    assert(this->m_exec != nullptr);
 
                    {
                        std::unique_lock lock(m_mutex);
                        m_result.set_value();
                        unblock_waiter(lock);
                    }
                }
 
                auto yield_value(debug_message msg, std::source_location loc = std::source_location::current())
                {
                    assert(m_node != nullptr);
                    std::unique_lock lock(m_node->m_mutex);
                    msg.location = loc;
                    m_node->m_messages.push_back(std::move(msg));
                    return std::suspend_never{};
                }

                template < query_spec_c QuerySpec >
                auto yield_value(dependency< QuerySpec > dep)
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, dep.m_request.m_input, true);
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return std::suspend_never{};
                }
 
                auto final_suspend() noexcept
                {
                    std::unique_lock lock(m_mutex);
                    assert(m_result.has_result());
                    return std::suspend_always{};
                }
 
                auto unhandled_exception()
                {
                    std::unique_lock lock(m_mutex);
                    assert(m_node != nullptr);
                    assert(this->m_exec != nullptr);
                    m_result.set_error(m_exec->m_runner->m_graph->capture_current_exception());
                    unblock_waiter(lock);
                }
 
                void abort_with_unexpected_exception(graph::executor* executor, std::exception_ptr exception) override
                {
                    this->m_exec = executor;
                    {
                        std::unique_lock lock(m_mutex);
                        if (!m_result.has_result())
                        {
                            m_result.set_error(error_result::unexpected(exception));
                        }
                    }
                    assert(m_node != nullptr);
                    m_exec->m_runner->m_graph->template abort_query_node_with_unexpected_exception< query_handler_output_t< HandlerSpec > >(m_node, executor, exception);
                }
 
                auto unblock_waiter(std::unique_lock< std::mutex >& lock)
                {
                    if (m_waiter.has_value())
                    {
                        auto resumption = m_waiter.value();
                        m_waiter.reset();
                        m_exec->check_legal_access();
                        m_exec->declare_runnable(resumption, false);
 
                        lock.unlock();
                    }
                }
 
                template < typename Ret2 >
                auto await_transform(typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >&& sub)
                {
                    assert(m_node != nullptr);
                    sub.m_promise->m_node = m_node;
                    sub.m_promise->m_handler_entry = m_handler_entry;
                    m_exec->declare_runnable(std::coroutine_handle< typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >::promise_type >::from_promise(*sub.m_promise), sub.m_promise, false);
                    return typename coroutine< HandlerSpec >::template cosubroutine< Ret2 >::awaitable(sub.m_promise);
                }
 
                template < query_spec_c QuerySpec >
                auto await_transform(request< QuerySpec > a) -> query_result< typename QuerySpec::output_type >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< QuerySpec >(), "Making a request to a query which is not listed in the HandlerSpec::dependencies");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< QuerySpec >& dep_query_descriptor = handler_impl.template get_dependency_descriptor< QuerySpec >();
 
                    query_result< typename QuerySpec::output_type > await_obj = dep_query_descriptor.m_get(m_exec, a.m_input, false);
 
                    graph::query_node* await_node = await_obj.m_node;
                    graph::declare_dependency(m_node, await_node);
                    return await_obj;
                }
 
                template < subquery_spec_c SubquerySpec >
                auto await_transform(subquery_request< SubquerySpec > a) -> subquery_query_result< SubquerySpec >
                {
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< SubquerySpec >(), "Making a subquery request requires listing the subquery in query_handler_dependencies_t< HandlerSpec >");
                    static_assert(query_handler_dependencies_t< HandlerSpec >::template contains< typename SubquerySpec::parent_query >(), "Making a subquery request requires listing the subquery parent query in query_handler_dependencies_t< HandlerSpec >");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec >& handler_impl = *std::any_cast< std::shared_ptr< typename graph::template query_handler_descriptor_memoized_impl< HandlerSpec > >& >(m_handler_entry->m_query_handler_impl);
 
                    typename graph::template query_descriptor< typename SubquerySpec::parent_query >& parent_query_descriptor = handler_impl.template get_subquery_parent_descriptor< SubquerySpec >();
 
                    query_result< typename SubquerySpec::parent_query::output_type > parent_await_obj = parent_query_descriptor.m_get(m_exec, a.m_parent_input, false);
 
                    graph::query_node* parent_node = parent_await_obj.m_node;
                    graph::declare_dependency(m_node, parent_node);
                    return m_exec->m_runner->m_graph->template get_subquery_result< SubquerySpec >(m_exec, parent_node, a.m_input);
                }
 
                template < subquery_spec_c SubquerySpec >
                auto yield_value(subquery_result< SubquerySpec > res)
                {
                    static_assert(query_handler_produced_subqueries_t< HandlerSpec >::template contains< SubquerySpec >(), "Yielding a subquery result requires listing the subquery in query_handler_produced_subqueries_t< HandlerSpec >");
                    static_assert(std::same_as< typename SubquerySpec::parent_query, query_handler_query_t< HandlerSpec > >, "A handler can only yield subqueries whose parent_query is the handler query");
                    assert(m_exec != nullptr);
                    assert(m_node != nullptr);
 
                    m_exec->m_runner->m_graph->template set_subquery_result< SubquerySpec >(m_exec, m_node, std::move(res.m_input), std::move(res.m_output));
                    return std::suspend_never{};
                }
 
                ~promise_type()
                {
                    assert(m_srt == nullptr);
                }
            };
 
            struct awaitable
            {
                promise_type* m_promise;
 
              public:
                awaitable(promise_type* promise) : m_promise(promise)
                {
                }
 
                bool await_ready()
                {
                    std::unique_lock lck(m_promise->m_mutex);
                    return await_ready(lck);
                }
 
                bool await_ready([[maybe_unused]] std::unique_lock< std::mutex >& lock) noexcept
                {
                    assert(m_promise != nullptr);
                    return m_promise->m_result.has_result();
                }
 
                void await_resume()
                {
                    std::unique_lock lck(m_promise->m_mutex);
                    assert(m_promise != nullptr);
                    assert(m_promise->m_result.has_result());
                    m_promise->m_result.test();
                }
 
                template < typename U >
                void await_suspend(std::coroutine_handle< U > h)
                {
                    await_suspend_helper< typename U::coroutine_type >(h.promise(), h);
                }
 
                template < typename C >
                void await_suspend_helper(typename C::promise_type& p, std::coroutine_handle< typename C::promise_type > h)
                {
                    assert(m_promise != nullptr);
 
                    auto current_executor = p.m_exec;
                    assert(current_executor != nullptr);
                    current_executor->check_legal_access();
 
                    graph::resumption r{.m_promise = &h.promise(), .handle = h};
                    std::unique_lock lck(m_promise->m_mutex);
 
                    if (await_ready(lck))
                    {
                        p.m_exec->declare_runnable(r, false);
                    }
                    else
                    {
                        assert(m_promise->m_waiter.has_value() == false);
                        m_promise->m_waiter = r;
                    }
                }
            };
 
          private:
            promise_type* m_promise;
 
          public:
            cosubroutine_impl(promise_type* p) : m_promise(p)
            {
                assert(p->m_srt == nullptr);
                p->m_srt = this;
            }
 
            cosubroutine_impl(const cosubroutine_impl&) = delete;
            cosubroutine_impl& operator=(const cosubroutine_impl&) = delete;
 
            ~cosubroutine_impl()
            {
                if (m_promise)
                {
                    std::unique_lock lock(m_promise->m_run_mutex);
                    lock.unlock();
 
                    m_promise->m_srt = nullptr;
                    std::coroutine_handle< promise_type >::from_promise(*m_promise).destroy();
                }
            }
        };
 
    } // namespace RPNX_DEBUGONLY(debug_abi)RPNX_RELEASEONLY(release_abi)
 
} // namespace rpnx::querygraph
 
#endif // NODEGRAPH2_NODEGRAPH_HPP