RCORE-2060 Enabling 'cancel_waits_on_nonfatal_error' does not cancel waits during location update while offline

CHANGELOG.md

@@ -5,7 +5,7 @@
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
-* None.
+* Enabling 'cancel_waits_on_nonfatal_error' does not cancel waits during location update while offline ([#7527](https://github.com/realm/realm-core/issues/7527), since v13.26.0)
 
 ### Breaking changes
 * None.

src/realm/exceptions.cpp

@@ -64,9 +64,12 @@ Exception::Exception(Status status)
 {
 }
 
-Status exception_to_status() noexcept
+Status exception_to_status(std::exception_ptr exc_ptr) noexcept
 {
     try {
+        if (exc_ptr) {
+            std::rethrow_exception(exc_ptr);
+        }
         throw;
     }
     catch (const Exception& e) {

src/realm/exceptions.hpp

@@ -21,6 +21,7 @@
 
 #include <realm/status.hpp>
 
+#include <exception>
 #include <stdexcept>
 #include <system_error>
 
@@ -49,7 +50,7 @@ class Exception : public std::exception {
  *
  * Currently this works for exceptions that derive from std::exception or Exception only.
  */
-Status exception_to_status() noexcept;
+Status exception_to_status(std::exception_ptr exc_ptr = nullptr) noexcept;
 
 
 /// The UnsupportedFileFormatVersion exception is thrown by DB::open()

src/realm/object-store/sync/app.cpp

@@ -823,7 +823,7 @@ void App::log_out(const std::shared_ptr<SyncUser>& user, UniqueFunction<void(Opt
         return completion(util::none);
     }
 
-    log_debug("App: log_out(%1)", user->user_profile().name());
+    log_debug("App: log_out(%1)", user->identity());
     auto refresh_token = user->refresh_token();
     user->log_out();
 
@@ -1255,15 +1255,18 @@ void App::refresh_access_token(const std::shared_ptr<SyncUser>& sync_user, bool
         return;
     }
 
-    log_debug("App: refresh_access_token: email: %1 %2", sync_user->user_profile().email(),
+    log_debug("App: refresh_access_token: email: %1 %2", sync_user->identity(),
               update_location ? "(updating location)" : "");
 
     // If update_location is set, force the location info to be updated before sending the request
     do_request(
         {HttpMethod::post, url_for_path("/auth/session"), m_request_timeout_ms,
          get_request_headers(sync_user, RequestTokenType::RefreshToken)},
-        [completion = std::move(completion), sync_user](const Response& response) {
+        [self = shared_from_this(), completion = std::move(completion), sync_user](const Response& response) {
             if (auto error = AppUtils::check_for_errors(response)) {
+                self->log_error("App: refresh_access_token: %1 -> %2 ERROR: %3", sync_user->identity(),
+                                response.http_status_code, error->what());
+
                 return completion(std::move(error));
             }
 

src/realm/object-store/sync/sync_session.cpp

@@ -343,9 +343,15 @@ SyncSession::handle_refresh(const std::shared_ptr<SyncSession>& session, bool re
                 // internal backoff timer which will happen automatically so nothing needs to
                 // happen here.
                 util::CheckedUniqueLock lock(session->m_state_mutex);
+                // If updating access token while opening realm, just become active at this point
+                // and try to use the current access token.
                 if (session->m_state == State::WaitingForAccessToken) {
                     session->become_active();
                 }
+                // If `cancel_waits_on_nonfatal_error` is true, then cancel the waiters and pass along the error
+                else if (session->config(&SyncConfig::cancel_waits_on_nonfatal_error)) {
+                    session->cancel_pending_waits(std::move(lock), error->to_status()); // unlocks the mutex
+                }
             }
         }
         else {

test/object-store/realm.cpp

@@ -45,6 +45,7 @@
 
 #if REALM_ENABLE_SYNC
 #include <util/sync/flx_sync_harness.hpp>
+#include <util/sync/sync_test_utils.hpp>
 
 #include <realm/object-store/sync/async_open_task.hpp>
 #include <realm/object-store/sync/impl/sync_metadata.hpp>
@@ -1206,6 +1207,117 @@ TEST_CASE("Get Realm using Async Open", "[sync][pbs][async open]") {
         REQUIRE(got_error);
     }
 
+    SECTION("waiters are cancelled if cancel_waits_on_nonfatal_error") {
+        auto logger = util::Logger::get_default_logger();
+        auto transport = std::make_shared<HookedUnitTestTransport>();
+        auto socket_provider = std::make_shared<HookedSocketProvider>(logger, "some user agent");
+        enum TestMode { location_fails, token_fails, token_not_authorized };
+        auto txt_test_mode = [](TestMode mode) {
+            switch (mode) {
+                case TestMode::location_fails:
+                    return "location_fails";
+                case TestMode::token_fails:
+                    return "token_fails";
+                case TestMode::token_not_authorized:
+                    return "token_not_authorized";
+                default:
+                    return "Unknown TestMode";
+            }
+        };
+
+        OfflineAppSession::Config oas_config;
+        oas_config.transport = transport;
+        oas_config.socket_provider = socket_provider;
+        OfflineAppSession oas(oas_config);
+
+        SyncTestFile config(oas, "realm");
+        auto user = config.sync_config->user;
+        config.sync_config->cancel_waits_on_nonfatal_error = true;
+        config.sync_config->error_handler = [&logger](std::shared_ptr<SyncSession> session, SyncError error) {
+            logger->debug("The sync error handler caught an error: '%1' for '%2'", error.status, session->path());
+            // Ignore connection failed non-fatal errors and check for access token refresh unauthorized fatal errors
+            if (error.status.code() == ErrorCodes::SyncConnectFailed) {
+                REQUIRE_FALSE(error.is_fatal);
+                return;
+            }
+            if (error.status.code() == ErrorCodes::AuthError) {
+                REQUIRE(error.is_fatal);
+            }
+        };
+
+        // User should be logged in at this point
+        REQUIRE(user->is_logged_in());
+        bool not_authorized = false;
+        bool token_refresh_called = false;
+        bool location_refresh_called = false;
+
+        TestMode mode = GENERATE(location_fails, token_fails, token_not_authorized);
+
+        SECTION(util::format("access token expired when realm is opened - mode: %1", txt_test_mode(mode))) {
+            logger->trace(">>> access token expired when realm is opened - mode: %1", txt_test_mode(mode));
+            // invalidate the user's cached access token
+            user->update_access_token(std::move(expired_token));
+        }
+        SECTION(util::format("access token expired by websocket - mode: %1", txt_test_mode(mode))) {
+            logger->trace(">>> access token expired by websocket - mode: %1", txt_test_mode(mode));
+            // tell websocket to return not authorized to refresh access token
+            not_authorized = true;
+        }
+        SECTION(util::format("websocket returns connection failed - mode: %1", txt_test_mode(mode))) {
+            logger->trace(">>> websocket returns connection failed - mode: %1", txt_test_mode(mode));
+            // default case
+        }
+
+        transport->request_hook = [&](const app::Request& req) -> std::optional<app::Response> {
+            static constexpr int CURLE_OPERATION_TIMEDOUT = 28;
+            std::lock_guard<std::mutex> lock(mutex);
+            if (req.url.find("/auth/session") != std::string::npos) {
+                token_refresh_called = true;
+                if (mode == token_not_authorized) {
+                    return app::Response{403, 0, {}, "403 not authorized"};
+                }
+                if (mode == token_fails) {
+                    return app::Response{0, CURLE_OPERATION_TIMEDOUT, {}, "Operation timed out"};
+                }
+            }
+            else if (req.url.find("/location") != std::string::npos) {
+                location_refresh_called = true;
+                if (mode == location_fails) {
+                    // Fake "offline/request timed out" custom error response
+                    return app::Response{0, CURLE_OPERATION_TIMEDOUT, {}, "Operation timed out"};
+                }
+            }
+            return std::nullopt;
+        };
+
+        socket_provider->websocket_connect_func = [&]() -> std::optional<SocketProviderError> {
+            if (not_authorized) {
+                not_authorized = false; // one shot
+                return SocketProviderError(sync::websocket::WebSocketError::websocket_unauthorized,
+                                           "403 not authorized");
+            }
+            return SocketProviderError(sync::websocket::WebSocketError::websocket_connection_failed,
+                                       "Operation timed out");
+        };
+
+        auto task = Realm::get_synchronized_realm(config);
+        auto pf = util::make_promise_future<std::exception_ptr>();
+        task->start([&pf](auto ref, auto error) mutable {
+            REQUIRE(!ref);
+            REQUIRE(error);
+            pf.promise.emplace_value(error);
+        });
+
+        auto result = pf.future.get_no_throw();
+        REQUIRE(result.is_ok());
+        REQUIRE(result.get_value());
+        std::lock_guard<std::mutex> lock(mutex);
+        REQUIRE(location_refresh_called);
+        if (mode != TestMode::location_fails) {
+            REQUIRE(token_refresh_called);
+        }
+    }
+
     SECTION("read-only mode sets the schema version") {
         {
             SharedRealm realm = Realm::get_shared_realm(config);

test/object-store/sync/app.cpp

@@ -2685,7 +2685,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
         }
     }
     {
-        auto redir_transport = std::make_shared<HookedTransport>();
+        auto redir_transport = std::make_shared<HookedSynchronousTransport>();
         AutoVerifiedEmailCredentials creds;
 
         auto app_config = get_config(redir_transport, session.app_session());
@@ -2852,7 +2852,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
         }
     }
     SECTION("Test app redirect with no metadata") {
-        auto redir_transport = std::make_shared<HookedTransport>();
+        auto redir_transport = std::make_shared<HookedSynchronousTransport>();
         AutoVerifiedEmailCredentials creds, creds2;
 
         auto app_config = get_config(redir_transport, session.app_session());
@@ -2943,7 +2943,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
         const std::string redirect_host = "fakerealm.example.com:9090";
         const std::string redirect_url = "http://fakerealm.example.com:9090";
 
-        auto redir_transport = std::make_shared<HookedTransport>();
+        auto redir_transport = std::make_shared<HookedSynchronousTransport>();
         auto redir_provider = std::make_shared<HookedSocketProvider>(logger, "");
         redir_provider->websocket_endpoint_resolver = [&](sync::WebSocketEndpoint&& ep) {
             ep.address = original_address;
@@ -3164,7 +3164,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
             original_address = original_host.substr(0, port_pos);
         }
 
-        auto redir_transport = std::make_shared<HookedTransport>();
+        auto redir_transport = std::make_shared<HookedSynchronousTransport>();
         auto redir_provider = std::make_shared<HookedSocketProvider>(logger, "");
         redir_provider->websocket_endpoint_resolver = [&](sync::WebSocketEndpoint&& ep) {
             ep.address = original_address;
@@ -3262,7 +3262,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
             REQUIRE(get_dogs(r).size() == 1);
         }
 
-        auto transport = std::make_shared<HookedTransport>();
+        auto transport = std::make_shared<HookedSynchronousTransport>();
         TestAppSession hooked_session(session.app_session(), transport, DeleteApp{false});
         auto app = hooked_session.app();
         std::shared_ptr<SyncUser> user = app->current_user();
@@ -3321,7 +3321,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
             REQUIRE(token.expired(now));
         }
 
-        auto transport = std::make_shared<HookedTransport>();
+        auto transport = std::make_shared<HookedSynchronousTransport>();
         TestAppSession hooked_session(session.app_session(), transport, DeleteApp{false});
         auto app = hooked_session.app();
         std::shared_ptr<SyncUser> user = app->current_user();

test/object-store/sync/flx_sync.cpp

@@ -2678,7 +2678,7 @@ TEST_CASE("flx: connect to PBS as FLX returns an error", "[sync][flx][protocol][
 }
 
 TEST_CASE("flx: commit subscription while refreshing the access token", "[sync][flx][token][baas]") {
-    auto transport = std::make_shared<HookedTransport>();
+    auto transport = std::make_shared<HookedSynchronousTransport>();
     FLXSyncTestHarness harness("flx_wait_access_token2", FLXSyncTestHarness::default_server_schema(), transport);
     auto app = harness.app();
     std::shared_ptr<SyncUser> user = app->current_user();

test/object-store/util/sync/baas_admin_api.cpp

@@ -346,8 +346,11 @@ app::Response do_http_request(const app::Request& request)
 
     auto logger = util::Logger::get_default_logger();
     if (response_code != CURLE_OK) {
+        std::string message = curl_easy_strerror(response_code);
         logger->error("curl_easy_perform() failed when sending request to '%1' with body '%2': %3", request.url,
-                      request.body, curl_easy_strerror(response_code));
+                      request.body, message);
+        // Return a failing response with the CURL error as the custom code
+        return {0, response_code, {}, message};
     }
     if (logger->would_log(util::Logger::Level::trace)) {
         std::string coid = [&] {

test/object-store/util/sync/sync_test_utils.hpp

@@ -194,8 +194,14 @@ class SynchronousTestTransport : public app::GenericNetworkTransport {
 };
 
 
-class HookedTransport : public SynchronousTestTransport {
+// Converted to a templated class to allow creating against the UnitTestTransport or
+// SynchronousTestTransport (or other custom) GenericNetworkTransport base class.
+template <typename Parent>
+class HookedTransport : public Parent {
 public:
+    static_assert(std::is_base_of<app::GenericNetworkTransport, Parent>::value,
+                  "HookedTransport must be derived from a class whose parent is app::GenericNetworkTransport");
+
     void send_request_to_server(const app::Request& request,
                                 util::UniqueFunction<void(const app::Response&)>&& completion) override
     {
@@ -204,7 +210,7 @@ class HookedTransport : public SynchronousTestTransport {
                 return completion(*simulated_response);
             }
         }
-        SynchronousTestTransport::send_request_to_server(request, [&](const app::Response& response) mutable {
+        Parent::send_request_to_server(request, [&](const app::Response& response) mutable {
             if (response_hook) {
                 response_hook(request, response);
             }
@@ -218,6 +224,9 @@ class HookedTransport : public SynchronousTestTransport {
     std::function<std::optional<app::Response>(const app::Request&)> request_hook;
 };
 
+using HookedSynchronousTransport = HookedTransport<SynchronousTestTransport>;
+using HookedUnitTestTransport = HookedTransport<UnitTestTransport>;
+
 
 struct SocketProviderError {
     SocketProviderError(sync::HTTPStatus code, std::string message = "")