realm / realm-core / jonathan.reams_2947

#ifndef REALM_NOINST_CLIENT_IMPL_BASE_HPP

#define REALM_NOINST_CLIENT_IMPL_BASE_HPP

#include <realm/sync/client_base.hpp>

#include <realm/binary_data.hpp>

#include <realm/sync/history.hpp>

#include <realm/sync/network/default_socket.hpp>

#include <realm/sync/network/network_ssl.hpp>

#include <realm/sync/noinst/client_history_impl.hpp>

#include <realm/sync/noinst/migration_store.hpp>

#include <realm/sync/noinst/migration_store.hpp>

#include <realm/sync/noinst/protocol_codec.hpp>

#include <realm/sync/protocol.hpp>

#include <realm/sync/subscriptions.hpp>

#include <realm/sync/trigger.hpp>

#include <realm/util/buffer_stream.hpp>

#include <realm/util/logger.hpp>

#include <realm/util/optional.hpp>

#include <realm/util/span.hpp>

#include <cstdint>

#include <deque>

#include <functional>

#include <list>

#include <map>

#include <random>

#include <string>

#include <utility>

namespace realm::sync {

// (protocol, address, port, session_multiplex_ident)

//

// `protocol` is included for convenience, even though it is not strictly part

// of an endpoint.

struct ServerEndpoint {

    ProtocolEnvelope envelope;

    std::string address;

    network::Endpoint::port_type port;

    std::string user_id;

    SyncServerMode server_mode = SyncServerMode::PBS;

private:

    auto to_tuple() const

public:

    friend inline bool operator==(const ServerEndpoint& lhs, const ServerEndpoint& rhs)

    friend inline bool operator<(const ServerEndpoint& lhs, const ServerEndpoint& rhs)

};

class SessionWrapper;

class SessionWrapperStack {

public:

    bool empty() const noexcept;

    void push(util::bind_ptr<SessionWrapper>) noexcept;

    util::bind_ptr<SessionWrapper> pop() noexcept;

    void clear() noexcept;

    SessionWrapperStack(SessionWrapperStack&&) noexcept;

    ~SessionWrapperStack();

    friend void swap(SessionWrapperStack& q_1, SessionWrapperStack& q_2) noexcept

private:

    SessionWrapper* m_back = nullptr;

};

template <typename ErrorType, typename RandomEngine>

struct ErrorBackoffState {

    ErrorBackoffState() = default;

    explicit ErrorBackoffState(ResumptionDelayInfo default_delay_info, RandomEngine& random_engine)

        : default_delay_info(std::move(default_delay_info))

        , m_random_engine(random_engine)

    void update(ErrorType new_error, std::optional<ResumptionDelayInfo> new_delay_info)

    void reset()

    std::chrono::milliseconds delay_interval()

    ResumptionDelayInfo default_delay_info;

    ResumptionDelayInfo delay_info;

    util::Optional<std::chrono::milliseconds> cur_delay_interval;

    util::Optional<ErrorType> triggering_error;

private:

    std::chrono::milliseconds jitter_value(std::chrono::milliseconds value)

    std::reference_wrapper<RandomEngine> m_random_engine;

};

class ClientImpl {

public:

    enum class ConnectionTerminationReason;

    class Connection;

    class Session;

    using port_type = network::Endpoint::port_type;

    using OutputBuffer = util::ResettableExpandableBufferOutputStream;

    using ClientProtocol = _impl::ClientProtocol;

    using RandomEngine = std::mt19937_64;

    /// Per-server endpoint information used to determine reconnect delays.

    class ReconnectInfo {

    public:

        explicit ReconnectInfo(ReconnectMode mode, ResumptionDelayInfo default_delay_info, RandomEngine& random)

            : m_reconnect_mode(mode)

            , m_backoff_state(default_delay_info, random)

        void reset() noexcept;

        void update(ConnectionTerminationReason reason, std::optional<ResumptionDelayInfo> new_delay_info);

        std::chrono::milliseconds delay_interval();

        const std::optional<ConnectionTerminationReason> reason() const noexcept

        // Set this flag to true to schedule a postponed invocation of reset(). See

        // Connection::cancel_reconnect_delay() for details and rationale.

        //

        // Will be set back to false when a PONG message arrives, and the

        // corresponding PING message was sent while `m_scheduled_reset` was

        // true. See receive_pong().

        bool scheduled_reset = false;

    private:

        ReconnectMode m_reconnect_mode;

        ErrorBackoffState<ConnectionTerminationReason, RandomEngine> m_backoff_state;

    };

    static constexpr milliseconds_type default_connect_timeout = 120000;        // 2 minutes

    static constexpr milliseconds_type default_connection_linger_time = 30000;  // 30 seconds

    static constexpr milliseconds_type default_ping_keepalive_period = 60000;   // 1 minute

    static constexpr milliseconds_type default_pong_keepalive_timeout = 120000; // 2 minutes

    static constexpr milliseconds_type default_fast_reconnect_limit = 60000;    // 1 minute

    const std::shared_ptr<util::Logger> logger_ptr;

    util::Logger& logger;

    ClientImpl(ClientConfig);

    ~ClientImpl();

    static constexpr int get_oldest_supported_protocol_version() noexcept;

    void shutdown() noexcept;

    void shutdown_and_wait();

    const std::string& get_user_agent_string() const noexcept;

    ReconnectMode get_reconnect_mode() const noexcept;

    bool is_dry_run() const noexcept;

    // Functions to post onto the event loop and create an event loop timer using the

    // SyncSocketProvider

    void post(SyncSocketProvider::FunctionHandler&& handler);

    SyncSocketProvider::SyncTimer create_timer(std::chrono::milliseconds delay,

                                               SyncSocketProvider::FunctionHandler&& handler);

    using SyncTrigger = std::unique_ptr<Trigger<ClientImpl>>;

    SyncTrigger create_trigger(SyncSocketProvider::FunctionHandler&& handler);

    RandomEngine& get_random() noexcept;

    /// Returns false if the specified URL is invalid.

    bool decompose_server_url(const std::string& url, ProtocolEnvelope& protocol, std::string& address,

                              port_type& port, std::string& path) const;

    void cancel_reconnect_delay();

    bool wait_for_session_terminations_or_client_stopped();

    void voluntary_disconnect_all_connections();

private:

    using connection_ident_type = std::int_fast64_t;

    const ReconnectMode m_reconnect_mode; // For testing purposes only

    const milliseconds_type m_connect_timeout;

    const milliseconds_type m_connection_linger_time;

    const milliseconds_type m_ping_keepalive_period;

    const milliseconds_type m_pong_keepalive_timeout;

    const milliseconds_type m_fast_reconnect_limit;

    const ResumptionDelayInfo m_reconnect_backoff_info;

    const bool m_disable_upload_activation_delay;

    const bool m_dry_run; // For testing purposes only

    const bool m_enable_default_port_hack;

    const bool m_disable_upload_compaction;

    const bool m_fix_up_object_ids;

    const std::function<RoundtripTimeHandler> m_roundtrip_time_handler;

    const std::string m_user_agent_string;

    std::shared_ptr<SyncSocketProvider> m_socket_provider;

    ClientProtocol m_client_protocol;

    session_ident_type m_prev_session_ident = 0;

    const bool m_one_connection_per_session;

    RandomEngine m_random;

    SyncTrigger m_actualize_and_finalize;

    // Note: There is one server slot per server endpoint (hostname, port,

    // session_multiplex_ident), and it survives from one connection object to

    // the next, which is important because it carries information about a

    // possible reconnect delay applying to the new connection object (server

    // hammering protection).

    //

    // Note: Due to a particular load balancing scheme that is currently in use,

    // every session is forced to open a seperate connection (via abuse of

    // `m_one_connection_per_session`, which is only intended for testing

    // purposes). This disables part of the hammering protection scheme built in

    // to the client.

    struct ServerSlot {

        explicit ServerSlot(ReconnectInfo reconnect_info)

            : reconnect_info(std::move(reconnect_info))

        ReconnectInfo reconnect_info; // Applies exclusively to `connection`.

        std::unique_ptr<ClientImpl::Connection> connection;

        // Used instead of `connection` when `m_one_connection_per_session` is

        // true.

        std::map<connection_ident_type, std::unique_ptr<ClientImpl::Connection>> alt_connections;

    };

    // Must be accessed only by event loop thread

    std::map<ServerEndpoint, ServerSlot> m_server_slots;

    // Must be accessed only by event loop thread

    connection_ident_type m_prev_connection_ident = 0;

    std::mutex m_drain_mutex;

    std::condition_variable m_drain_cv;

    bool m_drained = false;

    uint64_t m_outstanding_posts = 0;

    uint64_t m_num_connections = 0;

    std::mutex m_mutex;

    bool m_stopped = false;                       // Protected by `m_mutex`

    bool m_sessions_terminated = false;           // Protected by `m_mutex`

    bool m_actualize_and_finalize_needed = false; // Protected by `m_mutex`

    // The set of session wrappers that are not yet wrapping a session object,

    // and are not yet abandoned (still referenced by the application).

    //

    // Protected by `m_mutex`.

    std::map<SessionWrapper*, ServerEndpoint> m_unactualized_session_wrappers;

    // The set of session wrappers that were successfully actualized, but are

    // now abandoned (no longer referenced by the application), and have not yet

    // been finalized. Order in queue is immaterial.

    //

    // Protected by `m_mutex`.

    SessionWrapperStack m_abandoned_session_wrappers;

    // Protected by `m_mutex`.

    std::condition_variable m_wait_or_client_stopped_cond;

    void register_unactualized_session_wrapper(SessionWrapper*, ServerEndpoint);

    void register_abandoned_session_wrapper(util::bind_ptr<SessionWrapper>) noexcept;

    void actualize_and_finalize_session_wrappers();

    // Get or create a connection. If a connection exists for the specified

    // endpoint, it will be returned, otherwise a new connection will be

    // created. If `m_one_connection_per_session` is true (testing only), a new

    // connection will be created every time.

    //

    // Must only be accessed from event loop thread.

    //

    // FIXME: Passing these SSL parameters here is confusing at best, since they

    // are ignored if a connection is already available for the specified

    // endpoint. Also, there is no way to check that all the specified SSL

    // parameters are in agreement with a preexisting connection. A better

    // approach would be to allow for per-endpoint SSL parameters to be

    // specifiable through public member functions of ClientImpl from where they

    // could then be picked up as new connections are created on demand.

    //

    // FIXME: `session_multiplex_ident` should be eliminated from ServerEndpoint

    // as it effectively disables part of the hammering protection scheme if it

    // is used to ensure that each session gets a separate connection. With the

    // alternative approach outlined in the previous FIXME (specify per endpoint

    // SSL parameters at the client object level), there seems to be no more use

    // for `session_multiplex_ident`.

    ClientImpl::Connection& get_connection(ServerEndpoint, const std::string& authorization_header_name,

                                           const std::map<std::string, std::string>& custom_http_headers,

                                           bool verify_servers_ssl_certificate,

                                           util::Optional<std::string> ssl_trust_certificate_path,

                                           std::function<SyncConfig::SSLVerifyCallback>,

                                           util::Optional<SyncConfig::ProxyConfig>, bool& was_created);

    // Destroys the specified connection.

    void remove_connection(ClientImpl::Connection&) noexcept;

    void drain_connections();

    void drain_connections_on_loop();

    session_ident_type get_next_session_ident() noexcept;

    friend class ClientImpl::Connection;

    friend class SessionWrapper;

};

constexpr int ClientImpl::get_oldest_supported_protocol_version() noexcept

static_assert(ClientImpl::get_oldest_supported_protocol_version() >= 1, "");

static_assert(ClientImpl::get_oldest_supported_protocol_version() <= get_current_protocol_version(), "");

/// Information about why a connection (or connection initiation attempt) was

/// terminated. This is used to determinte the delay until the next connection

/// initiation attempt.

enum class ClientImpl::ConnectionTerminationReason {

    connect_operation_failed,          ///< Failure during connect operation

    closed_voluntarily,                ///< Voluntarily closed or connection operation canceled

    read_or_write_error,               ///< Read/write error after successful TCP connect operation

    ssl_certificate_rejected,          ///< Client rejected the SSL certificate of the server

    ssl_protocol_violation,            ///< A violation of the SSL protocol

    websocket_protocol_violation,      ///< A violation of the WebSocket protocol

    http_response_says_fatal_error,    ///< Status code in HTTP response says "fatal error"

    http_response_says_nonfatal_error, ///< Status code in HTTP response says "nonfatal error"

    bad_headers_in_http_response,      ///< Missing or bad headers in HTTP response

    sync_protocol_violation,           ///< Client received a bad message from the server

    sync_connect_timeout,              ///< Sync connection was not fully established in time

    server_said_try_again_later,       ///< Client received ERROR message with try_again=yes

    server_said_do_not_reconnect,      ///< Client received ERROR message with try_again=no

    pong_timeout,                      ///< Client did not receive PONG after PING

    /// The application requested a feature that is unavailable in the

    /// negotiated protocol version.

    missing_protocol_feature,

};

/// All use of connection objects, including construction and destruction, must

/// occur on behalf of the event loop thread of the associated client object.

// TODO: The parent will be updated to WebSocketObserver once the WebSocket integration is complete

class ClientImpl::Connection {

public:

    using connection_ident_type = std::int_fast64_t;

    using SSLVerifyCallback = SyncConfig::SSLVerifyCallback;

    using ProxyConfig = SyncConfig::ProxyConfig;

    using ReconnectInfo = ClientImpl::ReconnectInfo;

    std::shared_ptr<util::Logger> logger_ptr;

    util::Logger& logger;

    ClientImpl& get_client() noexcept;

    ReconnectInfo get_reconnect_info() const noexcept;

    ClientProtocol& get_client_protocol() noexcept;

    /// Activate this connection object. No attempt is made to establish a

    /// connection before the connection object is activated.

    void activate();

    /// Activate the specified session.

    ///

    /// Prior to being activated, no messages will be sent or received on behalf

    /// of this session, and the associated Realm file will not be accessed,

    /// i.e., `Session::get_db()` will not be called.

    ///

    /// If activation is successful, the connection keeps the session alive

    /// until the application calls initiated_session_deactivation() or until

    /// the application destroys the connection object, whichever comes first.

    void activate_session(std::unique_ptr<Session>);

    /// Initiate the deactivation process which eventually (or immediately)

    /// leads to destruction of this session object.

    ///

    /// IMPORTANT: The session object may get destroyed before this function

    /// returns.

    ///

    /// The deactivation process must be considered initiated even if this

    /// function throws.

    ///

    /// The deactivation process is guaranteed to not be initiated until the

    /// application calls this function. So from the point of view of the

    /// application, after successful activation, a pointer to a session object

    /// remains valid until the application calls

    /// initiate_session_deactivation().

    ///

    /// After the initiation of the deactivation process, the associated Realm

    /// file will no longer be accessed, i.e., `get_db()` will not be called

    /// again, and a previously returned reference will also not be accessed

    /// again.

    ///

    /// The initiation of the deactivation process must be preceded by a

    /// successful invocation of activate_session(). It is an error to call

    /// initiate_session_deactivation() twice.

    void initiate_session_deactivation(Session*);

    /// Cancel the reconnect delay for this connection, if one is currently in

    /// effect. If a reconnect delay is not currently in effect, ensure that the

    /// delay before the next reconnection attempt will be canceled. This is

    /// necessary as an apparently established connection, or ongoing connection

    /// attempt can be about to fail for a reason that precedes the invocation

    /// of this function.

    ///

    /// It is an error to call this function before the connection has been

    /// activated.

    void cancel_reconnect_delay();

    void force_close();

    /// Returns zero until the HTTP response is received. After that point in

    /// time, it returns the negotiated protocol version, which is based on the

    /// contents of the `Sec-WebSocket-Protocol` header in the HTTP

    /// response. The negotiated protocol version is guaranteed to be greater

    /// than or equal to get_oldest_supported_protocol_version(), and be less

    /// than or equal to get_current_protocol_version().

    int get_negotiated_protocol_version() noexcept;

    // Methods from WebSocketObserver interface for websockets from the Socket Provider

    void websocket_connected_handler(const std::string& protocol);

    bool websocket_binary_message_received(util::Span<const char> data);

    void websocket_error_handler();

    bool websocket_closed_handler(bool, websocket::WebSocketError, std::string_view msg);

    connection_ident_type get_ident() const noexcept;

    const ServerEndpoint& get_server_endpoint() const noexcept;

    ConnectionState get_state() const noexcept;

    SyncServerMode get_sync_server_mode() const noexcept;

    bool is_flx_sync_connection() const noexcept;

    void update_connect_info(const std::string& http_request_path_prefix, const std::string& signed_access_token);

    void resume_active_sessions();

    void voluntary_disconnect();

    std::string get_active_appservices_connection_id();

    Connection(ClientImpl&, connection_ident_type, ServerEndpoint, const std::string& authorization_header_name,

               const std::map<std::string, std::string>& custom_http_headers, bool verify_servers_ssl_certificate,

               util::Optional<std::string> ssl_trust_certificate_path, std::function<SSLVerifyCallback>,

               util::Optional<ProxyConfig>, ReconnectInfo);

    ~Connection();

private:

    struct LifecycleSentinel : public util::AtomicRefCountBase {

        bool destroyed = false;

    };

    struct WebSocketObserverShim;

    using ReceivedChangesets = ClientProtocol::ReceivedChangesets;

    template <class H>

    void for_each_active_session(H handler);

    /// \brief Called when the connection becomes idle.

    ///

    /// The connection is considered idle when all of the following conditions

    /// are true:

    ///

    /// - The connection is activated.

    ///

    /// - The connection has no sessions in the Active state.

    ///

    /// - The connection is closed (in the disconnected state).

    ///

    /// From the point of view of this class, an overriding function is allowed

    /// to commit suicide (`delete this`).

    ///

    /// The default implementation of this function does nothing.

    ///

    /// This function is always called by the event loop thread of the

    /// associated client object.

    void on_idle();

    std::string get_http_request_path() const;

    void initiate_reconnect_wait();

    void handle_reconnect_wait(Status status);

    void initiate_reconnect();

    void initiate_connect_wait();

    void handle_connect_wait(Status status);

    void handle_connection_established();

    void schedule_urgent_ping();

    void initiate_ping_delay(milliseconds_type now);

    void handle_ping_delay();

    void initiate_pong_timeout();

    void handle_pong_timeout();

    void initiate_write_message(const OutputBuffer&, Session*);

    void handle_write_message();

    void send_next_message();

    void send_ping();

    void initiate_write_ping(const OutputBuffer&);

    void handle_write_ping();

    void handle_message_received(util::Span<const char> data);

    void initiate_disconnect_wait();

    void handle_disconnect_wait(Status status);

    void close_due_to_protocol_error(Status status);

    void close_due_to_client_side_error(Status, IsFatal is_fatal, ProtocolErrorInfo::Action error_action,

                                        ConnectionTerminationReason reason);

    void close_due_to_transient_error(Status status, ConnectionTerminationReason reason);

    void close_due_to_server_side_error(ProtocolError, const ProtocolErrorInfo& info);

    void involuntary_disconnect(const SessionErrorInfo& info, ConnectionTerminationReason reason);

    void disconnect(const SessionErrorInfo& info);

    void change_state_to_disconnected() noexcept;

    // These are only called from ClientProtocol class.

    void receive_pong(milliseconds_type timestamp);

    void receive_error_message(const ProtocolErrorInfo& info, session_ident_type);

    void receive_query_error_message(int error_code, std::string_view message, int64_t query_version,

                                     session_ident_type);

    void receive_ident_message(session_ident_type, SaltedFileIdent);

    void receive_download_message(session_ident_type, const SyncProgress&, std::uint_fast64_t downloadable_bytes,

                                  int64_t query_version, DownloadBatchState batch_state, const ReceivedChangesets&);

    void receive_mark_message(session_ident_type, request_ident_type);

    void receive_unbound_message(session_ident_type);

    void receive_test_command_response(session_ident_type, request_ident_type, std::string_view body);

    void receive_server_log_message(session_ident_type, util::Logger::Level, std::string_view body);

    void receive_appservices_request_id(std::string_view coid);

    void handle_protocol_error(Status status);

    // These are only called from Session class.

    void enlist_to_send(Session*);

    void one_more_active_unsuspended_session();

    void one_less_active_unsuspended_session();

    void finish_session_deactivation(Session* sess);

    OutputBuffer& get_output_buffer() noexcept;

    Session* get_session(session_ident_type) const noexcept;

    Session* find_and_validate_session(session_ident_type session_ident, std::string_view message) noexcept;

    static bool was_voluntary(ConnectionTerminationReason) noexcept;

    static std::string make_logger_prefix(connection_ident_type);

    void report_connection_state_change(ConnectionState, util::Optional<SessionErrorInfo> error_info = util::none);

    friend ClientProtocol;

    friend class Session;

    ClientImpl& m_client;

    util::bind_ptr<LifecycleSentinel> m_websocket_sentinel;

    std::unique_ptr<WebSocketInterface> m_websocket;

    /// DEPRECATED - These will be removed in a future release

    const bool m_verify_servers_ssl_certificate;

    const util::Optional<std::string> m_ssl_trust_certificate_path;

    const std::function<SSLVerifyCallback> m_ssl_verify_callback;

    const util::Optional<ProxyConfig> m_proxy_config;

    ReconnectInfo m_reconnect_info;

    int m_negotiated_protocol_version = 0;

    ConnectionState m_state = ConnectionState::disconnected;

    std::size_t m_num_active_unsuspended_sessions = 0;

    std::size_t m_num_active_sessions = 0;

    ClientImpl::SyncTrigger m_on_idle;

    // activate() has been called

    bool m_activated = false;

    // A reconnect delay is in progress

    bool m_reconnect_delay_in_progress = false;

    // Has no meaning when m_reconnect_delay_in_progress is false.

    bool m_nonzero_reconnect_delay = false;

    // A disconnect (linger) delay is in progress. This is for keeping the

    // connection open for a while after there are no more active unsuspended

    // sessions.

    bool m_disconnect_delay_in_progress = false;

    bool m_disconnect_has_occurred = false;

    // A message is currently being sent, i.e., the sending of a message has

    // been initiated, but not yet completed.

    bool m_sending = false;

    bool m_ping_delay_in_progress = false;

    bool m_waiting_for_pong = false;

    bool m_send_ping = false;

    bool m_minimize_next_ping_delay = false;

    bool m_ping_after_scheduled_reset_of_reconnect_info = false;

    // At least one PING message was sent since connection was established

    bool m_ping_sent = false;

    bool m_websocket_error_received = false;

    bool m_force_closed = false;

    // The timer will be constructed on demand, and will only be destroyed when

    // canceling a reconnect or disconnect delay.

    //

    // It is necessary to destroy and recreate the timer when canceling a wait

    // operation, because the next wait operation might need to be initiated

    // before the completion handler of the previous canceled wait operation

    // starts executing. Such an overlap is not allowed for wait operations on

    // the same timer instance.

    SyncSocketProvider::SyncTimer m_reconnect_disconnect_timer;

    // Timer for connect operation watchdog. For why this timer is optional, see

    // `m_reconnect_disconnect_timer`.

    SyncSocketProvider::SyncTimer m_connect_timer;

    // This timer is used to schedule the sending of PING messages, and as a

    // watchdog for timely reception of PONG messages. For why this timer is

    // optional, see `m_reconnect_disconnect_timer`.

    SyncSocketProvider::SyncTimer m_heartbeat_timer;

    milliseconds_type m_pong_wait_started_at = 0;

    milliseconds_type m_last_ping_sent_at = 0;

    // Round-trip time, in milliseconds, for last PING message for which a PONG

    // message has been received, or zero if no PONG message has been received.

    milliseconds_type m_previous_ping_rtt = 0;

    // Only valid when `m_disconnect_has_occurred` is true.

    milliseconds_type m_disconnect_time = 0;

    // The set of sessions associated with this connection. A session becomes

    // associated with a connection when it is activated.

    std::map<session_ident_type, std::unique_ptr<Session>> m_sessions;

    // Keep track of previously used sessions idents to see if a stale message was

    // received for a closed session

    std::unordered_set<session_ident_type> m_session_history;

    // A queue of sessions that have enlisted for an opportunity to send a

    // message to the server. Sessions will be served in the order that they

    // enlist. A session is only allowed to occur once in this queue. If the

    // connection is open, and the queue is not empty, and no message is

    // currently being written, the first session is taken out of the queue, and

    // then granted an opportunity to send a message.

    std::deque<Session*> m_sessions_enlisted_to_send;

    Session* m_sending_session = nullptr;

    std::unique_ptr<char[]> m_input_body_buffer;

    OutputBuffer m_output_buffer;

    const connection_ident_type m_ident;

    const ServerEndpoint m_server_endpoint;

    std::string m_appservices_coid;

    /// DEPRECATED - These will be removed in a future release

    const std::string m_authorization_header_name;

    const std::map<std::string, std::string> m_custom_http_headers;

    std::string m_http_request_path_prefix;

    std::string m_signed_access_token;

};

/// A synchronization session between a local and a remote Realm file.

///

/// All use of session objects, including construction and destruction, must

/// occur on the event loop thread of the associated client object.

class ClientImpl::Session {

public:

    using ReceivedChangesets = ClientProtocol::ReceivedChangesets;

    std::shared_ptr<util::Logger> logger_ptr;

    util::Logger& logger;

    ClientImpl& get_client() noexcept;

    Connection& get_connection() noexcept;

    session_ident_type get_ident() const noexcept;

    /// Inform this client about new changesets in the history.

    ///

    /// The type of the version specified here is the one that identifies an

    /// entry in the sync history. Whether this is the same as the snapshot

    /// version of the Realm depends on the history implementation.

    ///

    /// The application is supposed to call this function to inform the client

    /// about a new version produced by a transaction that was not performed on

    /// behalf of this client. If the application does not call this function,

    /// the client will not discover and upload new changesets in a timely

    /// manner.

    ///

    /// It is an error to call this function before activation of the session,

    /// or after initiation of deactivation.

    void recognize_sync_version(version_type);

    /// \brief Request notification when all changesets in the local history

    /// have been uploaded to the server.

    ///

    /// When uploading completes, on_upload_completion() will be called by the

    /// thread that processes the event loop (as long as such a thread exists).

    ///

    /// IMPORTANT: on_upload_completion() may get called before

    /// request_upload_completion_notification() returns (reentrant callback).

    ///

    /// If request_upload_completion_notification() is called while a previously

    /// requested completion notification has not yet occurred, the previous

    /// request is canceled and the corresponding notification will never

    /// occur. This ensure that there is no ambiguity about the meaning of each

    /// completion notification.

    ///

    /// The application must be prepared for "spurious" invocations of

    /// on_upload_completion() before the client's first invocation of

    /// request_upload_completion_notification(), or after a previous invocation

    /// of on_upload_completion(), as long as it is before the subsequent

    /// invocation by the client of

    /// request_upload_completion_notification(). This is possible because the

    /// client reserves the right to request upload completion notifications

    /// internally.

    ///

    /// Upload is considered complete when all changesets in the history, that

    /// are supposed to be uploaded, and that precede `current_client_version`,

    /// have been uploaded and acknowledged by the

    /// server. `current_client_version` is generally the version that refers to

    /// the last changeset in the history, but more precisely, it may be any

    /// version between the last version reported by the application through

    /// recognize_sync_version() and the version referring to the last history

    /// entry (both ends inclusive).

    ///

    /// If new changesets are added to the history while a previously requested

    /// completion notification has not yet occurred, it is unspecified whether

    /// the addition of those changesets will cause `current_client_version` to

    /// be bumped or stay fixed, regardless of whether they are advertised via

    /// recognize_sync_version().

    ///

    /// It is an error to call this function before activation of the session,

    /// or after initiation of deactivation.

    void request_upload_completion_notification();

    /// \brief Request notification when all changesets currently avaialble on

    /// the server have been downloaded.

    ///

    /// When downloading completes, on_download_completion() will be called by

    /// the thread that processes the event loop (as long as such a thread

    /// exists).

    ///

    /// If request_download_completion_notification() is called while a

    /// previously requested completion notification has not yet occurred, the

    /// previous request is canceled and the corresponding notification will

    /// never occur. This ensure that there is no ambiguity about the meaning of

    /// each completion notification.

    ///

    /// The application must be prepared for "spurious" invocations of

    /// on_download_completion() before the client's first invocation of

    /// request_download_completion_notification(), or after a previous

    /// invocation of on_download_completion(), as long as it is before the

    /// subsequent invocation by the client of

    /// request_download_completion_notification(). This is possible because the

    /// client reserves the right to request download completion notifications

    /// internally.

    ///

    /// Download is considered complete when all changesets in the server-side

    /// history, that are supposed to be downloaded, and that precede

    /// `current_server_version`, have been downloaded and integrated into the

    /// local history. `current_server_version` is the version that refers to

    /// the last changeset in the server-side history at the time the server

    /// receives the first MARK message that is sent by the client after the

    /// invocation of request_download_completion_notification().

    ///

    /// Every invocation of request_download_completion_notification() will

    /// cause a new MARK message to be sent to the server, to redetermine

    /// `current_server_version`.

    ///

    /// It is an error to call this function before activation of the session,

    /// or after initiation of deactivation.

    void request_download_completion_notification();

    /// \brief Gets the subscription store associated with this Session.

    SubscriptionStore* get_flx_subscription_store();

    /// \brief Gets the migration store associated with this Session.

    MigrationStore* get_migration_store();

    /// Update internal client state when a flx subscription becomes complete outside

    /// of the normal sync process. This can happen during client reset.

    void on_flx_sync_version_complete(int64_t version);

    /// If this session is currently suspended, resume it immediately.

    ///

    /// It is an error to call this function before activation of the session,

    /// or after initiation of deactivation.

    void cancel_resumption_delay();

    /// To be used in connection with implementations of

    /// initiate_integrate_changesets().

    void integrate_changesets(const SyncProgress&, std::uint_fast64_t downloadable_bytes, const ReceivedChangesets&,

                              VersionInfo&, DownloadBatchState last_in_batch);

    /// To be used in connection with implementations of

    /// initiate_integrate_changesets().

    ///

    /// If \a success is true, the value of \a error does not matter. If \a

    /// success is false, the values of \a client_version and \a

    /// download_progress do not matter.

    ///

    /// It is an error to call this function before activation of the session

    /// (Connection::activate_session()), or after initiation of deactivation

    /// (Connection::initiate_session_deactivation()).

    void on_changesets_integrated(version_type client_version, const SyncProgress& progress);

    void on_integration_failure(const IntegrationException& e);

    void on_connection_state_changed(ConnectionState, const util::Optional<SessionErrorInfo>&);

    /// The application must ensure that the new session object is either

    /// activated (Connection::activate_session()) or destroyed before the

    /// specified connection object is destroyed.

    ///

    /// The specified transaction reporter (via the config object) is guaranteed

    /// to not be called before activation, and also not after initiation of

    /// deactivation.

    Session(SessionWrapper&, ClientImpl::Connection&);

    ~Session();

    void force_close();

    util::Future<std::string> send_test_command(std::string body);

private:

    struct PendingTestCommand {

        request_ident_type id;

        std::string body;

        util::Promise<std::string> promise;

        bool pending = true;

    };

    /// Fetch a reference to the remote virtual path of the Realm associated

    /// with this session.

    ///

    /// This function is always called by the event loop thread of the

    /// associated client object.

    ///

    /// This function is guaranteed to not be called before activation, and also

    /// not after initiation of deactivation.

    const std::string& get_virt_path() const noexcept;

    const std::string& get_realm_path() const noexcept;

    // Can only be called if the session is active or being activated

    DBRef get_db() const noexcept;

    ClientReplication& get_repl() const noexcept;

    ClientHistory& get_history() const noexcept;

    // client_reset_config() returns the config for client

    // reset. If it returns none, ordinary sync is used. If it returns a

    // Config::ClientReset, the session will be initiated with a state Realm

    // transfer from the server.

    util::Optional<ClientReset>& get_client_reset_config() noexcept;

    // Get the reason a synchronization session is used for (regular sync or client reset)

    // - Client reset state means the session is going to be used to download a fresh realm.

    SessionReason get_session_reason() noexcept;

    /// \brief Initiate the integration of downloaded changesets.

    ///

    /// This function must provide for the passed changesets (if any) to

    /// eventually be integrated, and without unnecessary delay. If no

    /// changesets are passed, the purpose of this function reduces to causing

    /// the current synchronization progress (SyncProgress) to be persisted.

    ///

    /// When all changesets have been integrated, and the synchronization

    /// progress has been persisted, this function must provide for

    /// on_changesets_integrated() to be called without unnecessary delay,

    /// although never after initiation of session deactivation.

    ///

    /// The implementation is allowed, but not obliged to aggregate changesets

    /// from multiple invocations of initiate_integrate_changesets() and pass

    /// them to ClientReplication::integrate_server_changesets() at once.

    ///

    /// The synchronization progress passed to

    /// ClientReplication::integrate_server_changesets() must be obtained

    /// by calling get_status(), and that call must occur after the last

    /// invocation of initiate_integrate_changesets() whose changesets are

    /// included in what is passed to

    /// ClientReplication::integrate_server_changesets().

    ///

    /// The download cursor passed to on_changesets_integrated() must be

    /// SyncProgress::download of the synchronization progress passed to the

    /// last invocation of

    /// ClientReplication::integrate_server_changesets().

    ///

    /// The default implementation integrates the specified changesets and calls

    /// on_changesets_integrated() immediately (i.e., from the event loop thread

    /// of the associated client object, and before

    /// initiate_integrate_changesets() returns), and via the history accessor

    /// made available by access_realm().

    ///

    /// This function is always called by the event loop thread of the

    /// associated client object, and on_changesets_integrated() must always be

    /// called by that thread too.

    ///

    /// This function is guaranteed to not be called before activation, and also

    /// not after initiation of deactivation.

    void initiate_integrate_changesets(std::uint_fast64_t downloadable_bytes, DownloadBatchState batch_state,

                                       const SyncProgress& progress, const ReceivedChangesets&);

    /// See request_upload_completion_notification().

    ///

    /// The default implementation does nothing.

    void on_upload_completion();

    /// See request_download_completion_notification().

    ///

    /// The default implementation does nothing.

    void on_download_completion();

    //@{

    /// These are called as the state of the session changes between

    /// "suspended" and "resumed". The initial state is

    /// always "resumed".

    ///

    /// A switch to the suspended state only happens when an error occurs,

    /// and information about that error is passed to on_suspended().

    ///

    /// The default implementations of these functions do nothing.

    ///

    /// These functions are always called by the event loop thread of the

    /// associated client object.

    ///

    /// These functions are guaranteed to not be called before activation, and also

    /// not after initiation of deactivation.

    void on_suspended(const SessionErrorInfo& error_info);

    void on_resumed();

    //@}

    void on_flx_sync_error(int64_t version, std::string_view err_msg);

    void on_flx_sync_progress(int64_t version, DownloadBatchState batch_state);

    // Processes an FLX download message, if it's a bootstrap message. If it's not a bootstrap

    // message then this is a noop and will return false. Otherwise this will return true

    // and no further processing of the download message should take place.

    bool process_flx_bootstrap_message(const SyncProgress& progress, DownloadBatchState batch_state,