realm/realm-core | Build 2214 | test/object-store/sync/flx_sync.cpp | Coveralls

14✔

    std::vector<ObjectId> ret;

14✔

116

    harness.load_initial_data([&](SharedRealm realm) {

14✔

117

        CppContext c(realm);

14✔

118

        for (int i = 0; i < 5; ++i) {

84✔

119

            auto id = ObjectId::gen();

70✔

120

            auto obj = Object::create(c, realm, "TopLevel",

70✔

121

                                      std::any(AnyDict{{"_id", id},

70✔

122

                                                       {"list_of_strings", AnyVector{}},

70✔

123

                                                       {"queryable_int_field", static_cast<int64_t>(i * 5)}}));

70✔

124

            List str_list(obj, realm->schema().find("TopLevel")->property_for_name("list_of_strings"));

70✔

125

            for (int j = 0; j < 1024; ++j) {

71,750✔

126

                str_list.add(c, std::any(std::string(1024, 'a' + (j % 26))));

71,680✔

127

71,680✔

128

35✔

129

            ret.push_back(id);

70✔

130

70✔

131

});

14✔

132

    return ret;

14✔

133

14✔

137

TEST_CASE("flx: connect to FLX-enabled app", "[sync][flx][baas]") {

2✔

138

    FLXSyncTestHarness harness("basic_flx_connect");

2✔

139

1✔

140

    auto foo_obj_id = ObjectId::gen();

2✔

141

    auto bar_obj_id = ObjectId::gen();

2✔

142

    harness.load_initial_data([&](SharedRealm realm) {

2✔

143

        CppContext c(realm);

2✔

144

        Object::create(c, realm, "TopLevel",

2✔

145

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

146

                                        {"queryable_str_field", "foo"s},

2✔

147

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

148

                                        {"non_queryable_field", "non queryable 1"s}}));

2✔

149

        Object::create(c, realm, "TopLevel",

2✔

150

                       std::any(AnyDict{{"_id", bar_obj_id},

2✔

151

                                        {"queryable_str_field", "bar"s},

2✔

152

                                        {"queryable_int_field", static_cast<int64_t>(10)},

2✔

153

                                        {"non_queryable_field", "non queryable 2"s}}));

2✔

154

});

2✔

155

1✔

156

1✔

157

    harness.do_with_new_realm([&](SharedRealm realm) {

2✔

158

2✔

159

            auto empty_subs = realm->get_latest_subscription_set();

2✔

162

            empty_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

163

2✔

164

1✔

165

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

166

        auto col_key = table->get_column_key("queryable_str_field");

2✔

167

        Query query_foo(table);

2✔

168

        query_foo.equal(col_key, "foo");

2✔

169

2✔

170

            auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

171

            new_subs.insert_or_assign(query_foo);

2✔

172

            auto subs = new_subs.commit();

2✔

173

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

174

2✔

175

1✔

176

2✔

177

            wait_for_advance(*realm);

2✔

178

            Results results(realm, table);

2✔

180

            auto obj = results.get<Obj>(0);

2✔

183

2✔

184

1✔

185

2✔

186

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

187

            Query new_query_bar(table);

2✔

188

            new_query_bar.equal(col_key, "bar");

2✔

189

            mut_subs.insert_or_assign(new_query_bar);

2✔

190

            auto subs = mut_subs.commit();

2✔

191

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

192

2✔

193

1✔

194

2✔

195

            wait_for_advance(*realm);

2✔

196

            Results results(realm, Query(table));

2✔

198

2✔

199

1✔

200

2✔

201

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

203

            Query new_query_bar(table);

2✔

204

            new_query_bar.equal(col_key, "bar");

2✔

205

            mut_subs.insert_or_assign(new_query_bar);

2✔

206

            auto subs = mut_subs.commit();

2✔

207

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

208

2✔

209

1✔

210

2✔

211

            wait_for_advance(*realm);

2✔

212

            Results results(realm, Query(table));

2✔

214

            auto obj = results.get<Obj>(0);

2✔

217

2✔

218

1✔

219

2✔

220

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

221

            mut_subs.clear();

2✔

222

            auto subs = mut_subs.commit();

2✔

223

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

224

2✔

225

1✔

226

2✔

227

            wait_for_advance(*realm);

2✔

228

            Results results(realm, table);

2✔

230

2✔

231

});

2✔

232

2✔

234

TEST_CASE("flx: test commands work", "[sync][flx][test command][baas]") {

2✔

235

    FLXSyncTestHarness harness("test_commands");

2✔

236

    harness.do_with_new_realm([&](const SharedRealm& realm) {

2✔

237

        wait_for_upload(*realm);

2✔

238

        nlohmann::json command_request = {

2✔

239

            {"command", "PAUSE_ROUTER_SESSION"},

2✔

240

};

2✔

241

        auto resp_body =

2✔

242

            SyncSession::OnlyForTesting::send_test_command(*realm->sync_session(), command_request.dump()).get();

2✔

244

1✔

245

        auto bad_status =

2✔

246

            SyncSession::OnlyForTesting::send_test_command(*realm->sync_session(), "foobar: }").get_no_throw();

2✔

248

        REQUIRE_THAT(bad_status.get_status().reason(),

2✔

249

                     Catch::Matchers::ContainsSubstring("Invalid json input to send_test_command"));

2✔

250

1✔

251

        bad_status =

2✔

252

            SyncSession::OnlyForTesting::send_test_command(*realm->sync_session(), "{\"cmd\": \"\"}").get_no_throw();

2✔

256

                "Must supply command name in \"command\" field of test command json object");

2✔

257

});

2✔

258

2✔

262

40✔

263

    auto [error_promise, error_future] = util::make_promise_future<SyncError>();

40✔

264

    auto shared_promise = std::make_shared<decltype(error_promise)>(std::move(error_promise));

40✔

265

    auto fn = [error_promise = std::move(shared_promise)](std::shared_ptr<SyncSession>, SyncError err) {

38✔

266

        error_promise->emplace_value(std::move(err));

36✔

267

};

36✔

268

    return std::make_pair(std::move(error_future), std::move(fn));

40✔

269

40✔

272

18✔

273

    auto [reset_promise, reset_future] = util::make_promise_future<ClientResyncMode>();

18✔

274

    auto shared_promise = std::make_shared<decltype(reset_promise)>(std::move(reset_promise));

18✔

275

    auto fn = [reset_promise = std::move(shared_promise)](SharedRealm, ThreadSafeReference, bool did_recover) {

18✔

276

        reset_promise->emplace_value(did_recover ? ClientResyncMode::Recover : ClientResyncMode::DiscardLocal);

14✔

277

};

18✔

278

    return std::make_pair(std::move(reset_future), std::move(fn));

18✔

279

18✔

282

TEST_CASE("app: error handling integration test", "[sync][flx][baas]") {

18✔

283

    static std::optional<FLXSyncTestHarness> harness{"error_handling"};

18✔

284

    create_user_and_log_in(harness->app());

18✔

285

    SyncTestFile config(harness->app()->current_user(), harness->schema(), SyncConfig::FLXSyncEnabled{});

18✔

286

    auto&& [error_future, error_handler] = make_error_handler();

18✔

287

    config.sync_config->error_handler = std::move(error_handler);

18✔

288

    config.sync_config->client_resync_mode = ClientResyncMode::Manual;

18✔

289

9✔

290

    SECTION("Resuming while waiting for session to auto-resume") {

18✔

291

        enum class TestState { InitialSuspend, InitialResume, SecondBind, SecondSuspend, SecondResume, Done };

2✔

292

        TestingStateMachine<TestState> state(TestState::InitialSuspend);

2✔

293

        config.sync_config->on_sync_client_event_hook = [&](std::weak_ptr<SyncSession>,

2✔

294

                                                            const SyncClientHookData& data) {

30✔

295

            std::optional<TestState> wait_for;

30✔

296

            auto event = data.event;

30✔

297

            state.transition_with([&](TestState state) -> std::optional<TestState> {

30✔

298

                if (state == TestState::InitialSuspend && event == SyncClientHookEvent::SessionSuspended) {

30✔

299

                    // If we're getting suspended for the first time, notify the test thread that we're

1✔

300

                    // ready to be resumed.

1✔

301

                    wait_for = TestState::SecondBind;

2✔

302

                    return TestState::InitialResume;

2✔

303

2✔

304

                else if (state == TestState::SecondBind && data.event == SyncClientHookEvent::BindMessageSent) {

28✔

305

                    return TestState::SecondSuspend;

2✔

306

2✔

307

                else if (state == TestState::SecondSuspend && event == SyncClientHookEvent::SessionSuspended) {

26✔

308

                    wait_for = TestState::Done;

2✔

309

                    return TestState::SecondResume;

2✔

310

2✔

311

                return std::nullopt;

24✔

312

});

24✔

313

            if (wait_for) {

30✔

314

                state.wait_for(*wait_for);

4✔

315

4✔

316

            return SyncClientHookAction::NoAction;

30✔

317

};

30✔

318

        auto r = Realm::get_shared_realm(config);

2✔

319

        wait_for_upload(*r);

2✔

320

        nlohmann::json error_body = {

2✔

321

            {"tryAgain", true},           {"message", "fake error"},

2✔

322

            {"shouldClientReset", false}, {"isRecoveryModeDisabled", false},

2✔

323

            {"action", "Transient"},      {"backoffIntervalSec", 900},

2✔

324

            {"backoffMaxDelaySec", 900},  {"backoffMultiplier", 1},

2✔

325

};

2✔

326

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

2✔

327

                                       {"args", nlohmann::json{{"errorCode", 229}, {"errorBody", error_body}}}};

2✔

328

1✔

329

        // First we trigger a retryable transient error that should cause the client to try to resume the

1✔

330

        // session in 5 minutes.

1✔

331

        auto test_cmd_res =

2✔

332

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

333

                .get();

2✔

335

1✔

336

        // Wait for the

1✔

337

        state.wait_for(TestState::InitialResume);

2✔

338

1✔

339

        // Once we're suspended, immediately tell the sync client to resume the session. This should cancel the

1✔

340

        // timer that would have auto-resumed the session.

1✔

341

        r->sync_session()->handle_reconnect();

2✔

342

        state.transition_with([&](TestState cur_state) {

2✔

344

            return TestState::SecondBind;

2✔

345

});

2✔

346

        state.wait_for(TestState::SecondSuspend);

2✔

347

1✔

348

        // Once we're connected again trigger another retryable transient error. Before RCORE-1770 the timer

1✔

349

        // to auto-resume the session would have still been active here and we would crash when trying to start

1✔

350

        // a second timer to auto-resume after this error.

1✔

351

        test_cmd_res =

2✔

352

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

353

                .get();

2✔

355

        state.wait_for(TestState::SecondResume);

2✔

356

1✔

357

        // Finally resume the session again which should cancel the second timer and the session should auto-resume

1✔

358

        // normally without crashing.

1✔

359

        r->sync_session()->handle_reconnect();

2✔

360

        state.transition_with([&](TestState cur_state) {

2✔

362

            return TestState::Done;

2✔

363

});

2✔

364

        wait_for_download(*r);

2✔

365

2✔

366

9✔

367

    SECTION("handles unknown errors gracefully") {

18✔

368

        auto r = Realm::get_shared_realm(config);

2✔

369

        wait_for_download(*r);

2✔

370

        nlohmann::json error_body = {

2✔

371

            {"tryAgain", false},         {"message", "fake error"},

2✔

372

            {"shouldClientReset", true}, {"isRecoveryModeDisabled", false},

2✔

373

            {"action", "ClientReset"},

2✔

374

};

2✔

375

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

2✔

376

                                       {"args", nlohmann::json{{"errorCode", 299}, {"errorBody", error_body}}}};

2✔

377

        auto test_cmd_res =

2✔

378

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

379

                .get();

2✔

381

        auto error = wait_for_future(std::move(error_future)).get();

2✔

385

        REQUIRE_THAT(error.status.reason(),

2✔

386

                     Catch::Matchers::ContainsSubstring("Unknown sync protocol error code 299"));

2✔

387

        REQUIRE_THAT(error.status.reason(), Catch::Matchers::ContainsSubstring("fake error"));

2✔

388

2✔

389

9✔

390

    SECTION("unknown errors without actions are application bugs") {

18✔

391

        auto r = Realm::get_shared_realm(config);

2✔

392

        wait_for_download(*r);

2✔

393

        nlohmann::json error_body = {

2✔

394

            {"tryAgain", false},

2✔

395

            {"message", "fake error"},

2✔

396

            {"shouldClientReset", false},

2✔

397

            {"isRecoveryModeDisabled", false},

2✔

398

};

2✔

399

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

2✔

400

                                       {"args", nlohmann::json{{"errorCode", 299}, {"errorBody", error_body}}}};

2✔

401

        auto test_cmd_res =

2✔

402

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

403

                .get();

2✔

405

        auto error = wait_for_future(std::move(error_future)).get();

2✔

409

        REQUIRE_THAT(error.status.reason(),

2✔

410

                     Catch::Matchers::ContainsSubstring("Unknown sync protocol error code 299"));

2✔

411

        REQUIRE_THAT(error.status.reason(), Catch::Matchers::ContainsSubstring("fake error"));

2✔

412

2✔

413

9✔

414

    SECTION("handles unknown actions gracefully") {

18✔

415

        auto r = Realm::get_shared_realm(config);

2✔

416

        wait_for_download(*r);

2✔

417

        nlohmann::json error_body = {

2✔

418

            {"tryAgain", false},

2✔

419

            {"message", "fake error"},

2✔

420

            {"shouldClientReset", true},

2✔

421

            {"isRecoveryModeDisabled", false},

2✔

422

            {"action", "FakeActionThatWillNeverExist"},

2✔

423

};

2✔

424

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

2✔

425

                                       {"args", nlohmann::json{{"errorCode", 201}, {"errorBody", error_body}}}};

2✔

426

        auto test_cmd_res =

2✔

427

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

428

                .get();

2✔

430

        auto error = wait_for_future(std::move(error_future)).get();

2✔

434

        REQUIRE_THAT(error.status.reason(), !Catch::Matchers::ContainsSubstring("Unknown sync protocol error code"));

2✔

435

        REQUIRE_THAT(error.status.reason(), Catch::Matchers::ContainsSubstring("fake error"));

2✔

436

2✔

437

9✔

438

9✔

439

    SECTION("unknown connection-level errors are still errors") {

18✔

440

        auto r = Realm::get_shared_realm(config);

2✔

441

        wait_for_download(*r);

2✔

442

        nlohmann::json error_body = {{"tryAgain", false},

2✔

443

                                     {"message", "fake error"},

2✔

444

                                     {"shouldClientReset", false},

2✔

445

                                     {"isRecoveryModeDisabled", false},

2✔

446

                                     {"action", "ApplicationBug"}};

2✔

447

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

2✔

448

                                       {"args", nlohmann::json{{"errorCode", 199}, {"errorBody", error_body}}}};

2✔

449

        auto test_cmd_res =

2✔

450

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

2✔

451

                .get();

2✔

453

        auto error = wait_for_future(std::move(error_future)).get();

2✔

457

2✔

458

9✔

459

    SECTION("client reset errors") {

18✔

460

        auto r = Realm::get_shared_realm(config);

6✔

461

        wait_for_download(*r);

6✔

462

        nlohmann::json error_body = {{"tryAgain", false},

6✔

463

                                     {"message", "fake error"},

6✔

464

                                     {"shouldClientReset", true},

6✔

465

                                     {"isRecoveryModeDisabled", false},

6✔

466

                                     {"action", "ClientReset"}};

6✔

467

        auto code = GENERATE(sync::ProtocolError::bad_client_file_ident, sync::ProtocolError::bad_server_version,

6✔

468

                             sync::ProtocolError::diverging_histories);

6✔

469

        nlohmann::json test_command = {{"command", "ECHO_ERROR"},

6✔

470

                                       {"args", nlohmann::json{{"errorCode", code}, {"errorBody", error_body}}}};

6✔

471

        auto test_cmd_res =

6✔

472

            wait_for_future(SyncSession::OnlyForTesting::send_test_command(*r->sync_session(), test_command.dump()))

6✔

473

                .get();

6✔

475

        auto error = wait_for_future(std::move(error_future)).get();

6✔

480

6✔

481

9✔

482

9✔

483

    SECTION("teardown") {

18✔

484

        harness.reset();

2✔

485

2✔

486

18✔

489

TEST_CASE("flx: client reset", "[sync][flx][client reset][baas]") {

44✔

490

    std::vector<ObjectSchema> schema{

44✔

491

        {"TopLevel",

44✔

492

44✔

493

             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

44✔

494

             {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

44✔

495

             {"queryable_int_field", PropertyType::Int | PropertyType::Nullable},

44✔

496

             {"non_queryable_field", PropertyType::String | PropertyType::Nullable},

44✔

497

             {"list_of_ints_field", PropertyType::Int | PropertyType::Array},

44✔

498

             {"sum_of_list_field", PropertyType::Int},

44✔

499

}},

44✔

500

        {"TopLevel2",

44✔

501

44✔

502

             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

44✔

503

             {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

44✔

504

}},

44✔

505

};

44✔

506

22✔

507

    // some of these tests make additive schema changes which is only allowed in dev mode

22✔

508

    constexpr bool dev_mode = true;

44✔

509

    FLXSyncTestHarness harness("flx_client_reset",

44✔

510

                               {schema, {"queryable_str_field", "queryable_int_field"}, {}, dev_mode});

44✔

511

22✔

512

    auto add_object = [](SharedRealm realm, std::string str_field, int64_t int_field,

44✔

513

                         ObjectId oid = ObjectId::gen()) {

122✔

514

        CppContext c(realm);

122✔

515

        realm->begin_transaction();

122✔

516

61✔

517

        int64_t r1 = random_int();

122✔

518

        int64_t r2 = random_int();

122✔

519

        int64_t r3 = random_int();

122✔

520

        int64_t sum = uint64_t(r1) + r2 + r3;

122✔

521

61✔

522

        Object::create(c, realm, "TopLevel",

122✔

523

                       std::any(AnyDict{{"_id", oid},

122✔

524

                                        {"queryable_str_field", str_field},

122✔

525

                                        {"queryable_int_field", int_field},

122✔

526

                                        {"non_queryable_field", "non queryable 1"s},

122✔

527

                                        {"list_of_ints_field", std::vector<std::any>{r1, r2, r3}},

122✔

528

                                        {"sum_of_list_field", sum}}));

122✔

529

        realm->commit_transaction();

122✔

530

};

122✔

531

22✔

532

    auto subscribe_to_and_add_objects = [&](SharedRealm realm, size_t num_objects) {

42✔

533

        auto table = realm->read_group().get_table("class_TopLevel");

40✔

534

        auto id_col = table->get_primary_key_column();

40✔

535

        auto sub_set = realm->get_latest_subscription_set();

40✔

536

        for (size_t i = 0; i < num_objects; ++i) {

130✔

537

            auto oid = ObjectId::gen();

90✔

538

            auto mut_sub = sub_set.make_mutable_copy();

90✔

539

            mut_sub.clear();

90✔

540

            mut_sub.insert_or_assign(Query(table).equal(id_col, oid));

90✔

541

            sub_set = mut_sub.commit();

90✔

542

            add_object(realm, util::format("added _id='%1'", oid), 0, oid);

90✔

543

90✔

544

};

40✔

545

22✔

546

    auto add_subscription_for_new_object = [&](SharedRealm realm, std::string str_field,

44✔

547

                                               int64_t int_field) -> sync::SubscriptionSet {

36✔

548

        auto table = realm->read_group().get_table("class_TopLevel");

28✔

549

        auto queryable_str_field = table->get_column_key("queryable_str_field");

28✔

550

        auto sub_set = realm->get_latest_subscription_set().make_mutable_copy();

28✔

551

        sub_set.insert_or_assign(Query(table).equal(queryable_str_field, StringData(str_field)));

28✔

552

        auto resulting_set = sub_set.commit();

28✔

553

        add_object(realm, str_field, int_field);

28✔

554

        return resulting_set;

28✔

555

};

28✔

556

22✔

557

    auto add_invalid_subscription = [](SharedRealm realm) -> sync::SubscriptionSet {

23✔

558

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

559

        auto queryable_str_field = table->get_column_key("non_queryable_field");

2✔

560

        auto sub_set = realm->get_latest_subscription_set().make_mutable_copy();

2✔

561

        sub_set.insert_or_assign(Query(table).equal(queryable_str_field, "foo"));

2✔

562

        auto resulting_set = sub_set.commit();

2✔

563

        return resulting_set;

2✔

564

};

2✔

565

22✔

566

    auto count_queries_with_str = [](sync::SubscriptionSet subs, std::string_view str) {

26✔

567

        size_t count = 0;

8✔

568

        for (auto sub : subs) {

14✔

569

            if (sub.query_string.find(str) != std::string::npos) {

14✔

570

                ++count;

6✔

571

6✔

572

14✔

573

        return count;

8✔

574

};

8✔

575

    create_user_and_log_in(harness.app());

44✔

576

    auto user1 = harness.app()->current_user();

44✔

577

    create_user_and_log_in(harness.app());

44✔

578

    auto user2 = harness.app()->current_user();

44✔

579

    SyncTestFile config_local(user1, harness.schema(), SyncConfig::FLXSyncEnabled{});

44✔

580

    config_local.path += ".local";

44✔

581

    SyncTestFile config_remote(user2, harness.schema(), SyncConfig::FLXSyncEnabled{});

44✔

582

    config_remote.path += ".remote";

44✔

583

    const std::string str_field_value = "foo";

44✔

584

    const int64_t local_added_int = 100;

44✔

585

    const int64_t local_added_int2 = 150;

44✔

586

    const int64_t remote_added_int = 200;

44✔

587

    size_t before_reset_count = 0;

44✔

588

    size_t after_reset_count = 0;

44✔

589

    config_local.sync_config->notify_before_client_reset = [&before_reset_count](SharedRealm) {

36✔

590

        ++before_reset_count;

28✔

591

};

28✔

592

    config_local.sync_config->notify_after_client_reset = [&after_reset_count](SharedRealm, ThreadSafeReference,

44✔

593

                                                                               bool) {

22✔

596

22✔

597

    SECTION("Recover: offline writes and subscription (single subscription)") {

44✔

598

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

599

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

600

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

601

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

602

        test_reset

2✔

603

            ->populate_initial_object([&](SharedRealm realm) {

2✔

604

                auto pk_of_added_object = ObjectId::gen();

2✔

605

                auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

606

                auto table = realm->read_group().get_table(ObjectStore::table_name_for_object_type("TopLevel"));

2✔

607

                REALM_ASSERT(table);

2✔

608

                mut_subs.insert_or_assign(Query(table));

2✔

609

                mut_subs.commit();

2✔

610

1✔

611

                realm->begin_transaction();

2✔

612

                CppContext c(realm);

2✔

613

                int64_t r1 = random_int();

2✔

614

                int64_t r2 = random_int();

2✔

615

                int64_t r3 = random_int();

2✔

616

                int64_t sum = uint64_t(r1) + r2 + r3;

2✔

617

1✔

618

                Object::create(c, realm, "TopLevel",

2✔

619

                               std::any(AnyDict{{"_id"s, pk_of_added_object},

2✔

620

                                                {"queryable_str_field"s, "initial value"s},

2✔

621

                                                {"list_of_ints_field", std::vector<std::any>{r1, r2, r3}},

2✔

622

                                                {"sum_of_list_field", sum}}));

2✔

623

1✔

624

                realm->commit_transaction();

2✔

625

                wait_for_upload(*realm);

2✔

626

                return pk_of_added_object;

2✔

627

})

2✔

628

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

629

                add_object(local_realm, str_field_value, local_added_int);

2✔

630

})

2✔

631

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

632

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

633

                sync::SubscriptionSet::State actual =

2✔

634

                    remote_realm->get_latest_subscription_set()

2✔

635

                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

636

                        .get();

2✔

638

})

2✔

639

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {

2✔

640

                wait_for_advance(*local_realm);

2✔

641

                ClientResyncMode mode = client_reset_future.get();

2✔

643

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

644

                auto str_col = table->get_column_key("queryable_str_field");

2✔

645

                auto int_col = table->get_column_key("queryable_int_field");

2✔

646

                auto tv = table->where().equal(str_col, StringData(str_field_value)).find_all();

2✔

647

                tv.sort(int_col);

2✔

648

                // the object we created while offline was recovered, and the remote object was downloaded

1✔

652

})

2✔

653

            ->run();

2✔

654

2✔

655

22✔

656

    SECTION("Recover: subscription and offline writes after client reset failure") {

44✔

657

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

658

        auto&& [error_future, error_handler] = make_error_handler();

2✔

659

        config_local.sync_config->error_handler = error_handler;

2✔

660

1✔

661

        std::string fresh_path = realm::_impl::client_reset::get_fresh_path_for(config_local.path);

2✔

662

        // create a non-empty directory that we'll fail to delete

1✔

663

        util::make_dir(fresh_path);

2✔

664

        util::File(util::File::resolve("file", fresh_path), util::File::mode_Write);

2✔

665

1✔

666

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

667

        test_reset

2✔

668

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

669

                auto mut_sub = local_realm->get_latest_subscription_set().make_mutable_copy();

2✔

670

                auto table = local_realm->read_group().get_table("class_TopLevel2");

2✔

671

                mut_sub.insert_or_assign(Query(table));

2✔

672

                mut_sub.commit();

2✔

673

1✔

674

                CppContext c(local_realm);

2✔

675

                local_realm->begin_transaction();

2✔

676

                Object::create(c, local_realm, "TopLevel2",

2✔

677

                               std::any(AnyDict{{"_id"s, ObjectId::gen()}, {"queryable_str_field"s, "foo"s}}));

2✔

678

                local_realm->commit_transaction();

2✔

679

})

2✔

680

            ->on_post_reset([](SharedRealm local_realm) {

2✔

681

                // Verify offline subscription was not removed.

1✔

682

                auto subs = local_realm->get_latest_subscription_set();

2✔

683

                auto table = local_realm->read_group().get_table("class_TopLevel2");

2✔

685

})

2✔

686

            ->run();

2✔

687

1✔

688

        // Remove the folder preventing the completion of a client reset.

1✔

689

        util::try_remove_dir_recursive(fresh_path);

2✔

690

1✔

691

        RealmConfig config_copy = config_local;

2✔

692

        config_copy.sync_config = std::make_shared<SyncConfig>(*config_copy.sync_config);

2✔

693

        config_copy.sync_config->error_handler = nullptr;

2✔

694

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

695

        config_copy.sync_config->notify_after_client_reset = reset_handler;

2✔

696

1✔

697

        // Attempt to open the realm again.

1✔

698

        // This time the client reset succeeds and the offline subscription and writes are recovered.

1✔

699

        auto realm = Realm::get_shared_realm(config_copy);

2✔

700

        ClientResyncMode mode = reset_future.get();

2✔

702

1✔

703

        auto table = realm->read_group().get_table("class_TopLevel2");

2✔

704

        auto str_col = table->get_column_key("queryable_str_field");

2✔

707

2✔

708

22✔

709

    SECTION("Recover: offline writes and subscriptions (multiple subscriptions)") {

44✔

710

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

711

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

712

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

713

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

714

        test_reset

2✔

715

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

716

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);

2✔

717

})

2✔

718

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

719

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

720

                sync::SubscriptionSet::State actual =

2✔

721

                    remote_realm->get_latest_subscription_set()

2✔

722

                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

723

                        .get();

2✔

725

})

2✔

726

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {

2✔

727

                ClientResyncMode mode = client_reset_future.get();

2✔

729

                auto subs = local_realm->get_latest_subscription_set();

2✔

730

                subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

731

                // make sure that the subscription for "foo" survived the reset

1✔

732

                size_t count_of_foo = count_queries_with_str(subs, util::format("\"%1\"", str_field_value));

2✔

735

                local_realm->refresh();

2✔

736

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

737

                auto str_col = table->get_column_key("queryable_str_field");

2✔

738

                auto int_col = table->get_column_key("queryable_int_field");

2✔

739

                auto tv = table->where().equal(str_col, StringData(str_field_value)).find_all();

2✔

740

                tv.sort(int_col);

2✔

741

                // the object we created while offline was recovered, and the remote object was downloaded

1✔

745

})

2✔

746

            ->run();

2✔

747

2✔

748

22✔

749

    SECTION("Recover: offline writes interleaved with subscriptions and empty writes") {

44✔

750

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

751

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

752

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

753

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

754

        test_reset

2✔

755

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

756

                // The sequence of events bellow generates five changesets:

1✔

757

                //  1. create sub1 => empty changeset

1✔

758

                //  2. create local_added_int object

1✔

759

                //  3. create empty changeset

1✔

760

                //  4. create sub2 => empty changeset

1✔

761

                //  5. create local_added_int2 object

1✔

762

//

1✔

763

                // Before sending 'sub2' to the server, an UPLOAD message is sent first.

1✔

764

                // The upload message contains changeset 2. (local_added_int) with the cursor

1✔

765

                // of changeset 3. (empty changeset).

1✔

766

1✔

767

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);

2✔

768

                // Commit empty changeset.

1✔

769

                local_realm->begin_transaction();

2✔

770

                local_realm->commit_transaction();

2✔

771

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int2);

2✔

772

})

2✔

773

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

774

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

775

                sync::SubscriptionSet::State actual =

2✔

776

                    remote_realm->get_latest_subscription_set()

2✔

777

                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

778

                        .get();

2✔

780

})

2✔

781

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {

2✔

782

                ClientResyncMode mode = client_reset_future.get();

2✔

784

                auto subs = local_realm->get_latest_subscription_set();

2✔

785

                subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

786

                // make sure that the subscription for "foo" survived the reset

1✔

787

                size_t count_of_foo = count_queries_with_str(subs, util::format("\"%1\"", str_field_value));

2✔

788

                subs.refresh();

2✔

791

                local_realm->refresh();

2✔

792

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

793

                auto str_col = table->get_column_key("queryable_str_field");

2✔

794

                auto int_col = table->get_column_key("queryable_int_field");

2✔

795

                auto tv = table->where().equal(str_col, StringData(str_field_value)).find_all();

2✔

796

                tv.sort(int_col);

2✔

797

                // the objects we created while offline was recovered, and the remote object was downloaded

1✔

802

})

2✔

803

            ->run();

2✔

804

2✔

805

22✔

806

    auto validate_integrity_of_arrays = [](TableRef table) -> size_t {

30✔

807

        auto sum_col = table->get_column_key("sum_of_list_field");

16✔

808

        auto array_col = table->get_column_key("list_of_ints_field");

16✔

809

        auto query = table->column<Lst<Int>>(array_col).sum() == table->column<Int>(sum_col) &&

16✔

810

                     table->column<Lst<Int>>(array_col).size() > 0;

16✔

811

        return query.count();

16✔

812

};

16✔

813

22✔

814

    SECTION("Recover: offline writes with associated subscriptions in the correct order") {

44✔

815

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

816

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

817

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

818

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

819

        constexpr size_t num_objects_added = 20;

2✔

820

        constexpr size_t num_objects_added_by_harness = 1; // BaasFLXClientReset.run()

2✔

821

        constexpr size_t num_objects_added_by_remote = 1;  // make_remote_changes()

2✔

822

        test_reset

2✔

823

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

824

                subscribe_to_and_add_objects(local_realm, num_objects_added);

2✔

825

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

827

                size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

2✔

829

})

2✔

830

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

831

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

832

                sync::SubscriptionSet::State actual =

2✔

833

                    remote_realm->get_latest_subscription_set()

2✔

834

                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

835

                        .get();

2✔

837

})

2✔

838

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {

2✔

839

                ClientResyncMode mode = client_reset_future.get();

2✔

841

                local_realm->refresh();

2✔

842

                auto latest_subs = local_realm->get_latest_subscription_set();

2✔

843

                auto state = latest_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

845

                local_realm->refresh();

2✔

846

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

847

                if (table->size() != 1) {

2✔

851

                auto mut_sub = latest_subs.make_mutable_copy();

2✔

852

                mut_sub.clear();

2✔

853

                mut_sub.insert_or_assign(Query(table));

2✔

854

                latest_subs = mut_sub.commit();

2✔

855

                latest_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

856

                local_realm->refresh();

2✔

858

                        num_objects_added + num_objects_added_by_harness + num_objects_added_by_remote);

2✔

859

                size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

2✔

861

})

2✔

862

            ->run();

2✔

863

2✔

864

22✔

865

    SECTION("Recover: incompatible property changes are rejected") {

44✔

866

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

867

        auto&& [error_future, err_handler] = make_error_handler();

2✔

868

        config_local.sync_config->error_handler = err_handler;

2✔

869

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

870

        constexpr size_t num_objects_added_before = 2;

2✔

871

        constexpr size_t num_objects_added_after = 2;

2✔

872

        constexpr size_t num_objects_added_by_harness = 1; // BaasFLXClientReset.run()

2✔

873

        constexpr std::string_view added_property_name = "new_property";

2✔

874

        test_reset

2✔

875

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

876

                subscribe_to_and_add_objects(local_realm, num_objects_added_before);

2✔

877

                Schema local_update = schema;

2✔

878

                Schema::iterator it = local_update.find("TopLevel");

2✔

880

                it->persisted_properties.push_back(

2✔

881

                    {std::string(added_property_name), PropertyType::Float | PropertyType::Nullable});

2✔

882

                local_realm->update_schema(local_update);

2✔

883

                subscribe_to_and_add_objects(local_realm, num_objects_added_after);

2✔

884

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

886

                        num_objects_added_before + num_objects_added_after + num_objects_added_by_harness);

2✔

887

                size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

2✔

889

})

2✔

890

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

891

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

892

                Schema remote_update = schema;

2✔

893

                Schema::iterator it = remote_update.find("TopLevel");

2✔

895

                it->persisted_properties.push_back(

2✔

896

                    {std::string(added_property_name), PropertyType::UUID | PropertyType::Nullable});

2✔

897

                remote_realm->update_schema(remote_update);

2✔

898

                sync::SubscriptionSet::State actual =

2✔

899

                    remote_realm->get_latest_subscription_set()

2✔

900

                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

901

                        .get();

2✔

903

})

2✔

904

            ->on_post_reset([&, err_future = std::move(error_future)](SharedRealm local_realm) mutable {

2✔

905

                auto sync_error = wait_for_future(std::move(err_future)).get();

2✔

910

                local_realm->refresh();

2✔

911

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

912

                // since schema validation happens in the first recovery commit, that whole commit is rolled back

1✔

913

                // and the final state here is "pre reset"

1✔

915

                        num_objects_added_before + num_objects_added_by_harness + num_objects_added_after);

2✔

916

                size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

2✔

918

})

2✔

919

            ->run();

2✔

920

2✔

921

22✔

922

    SECTION("unsuccessful replay of local changes") {

44✔

923

        constexpr size_t num_objects_added_before = 2;

4✔

924

        constexpr size_t num_objects_added_after = 2;

4✔

925

        constexpr size_t num_objects_added_by_harness = 1; // BaasFLXClientReset.run()

4✔

926

        constexpr std::string_view added_property_name = "new_property";

4✔

927

        auto&& [error_future, err_handler] = make_error_handler();

4✔

928

        config_local.sync_config->error_handler = err_handler;

4✔

929

2✔

930

        // The local changes here are a bit contrived because removing a column is disallowed

2✔

931

        // at the object store layer for sync'd Realms. The only reason a recovery should fail in production

2✔

932

        // during the apply stage is due to programmer error or external factors such as out of disk space.

2✔

933

        // Any schema discrepancies are caught by the initial diff, so the way to make a recovery fail here is

2✔

934

        // to add and remove a column at the core level such that the schema diff passes, but instructions are

2✔

935

        // generated which will fail when applied.

2✔

936

        auto make_local_changes_that_will_fail = [&](SharedRealm local_realm) {

4✔

937

            subscribe_to_and_add_objects(local_realm, num_objects_added_before);

4✔

938

            auto table = local_realm->read_group().get_table("class_TopLevel");

4✔

940

            size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

4✔

942

            local_realm->begin_transaction();

4✔

943

            ColKey added = table->add_column(type_Int, added_property_name);

4✔

944

            table->remove_column(added);

4✔

945

            local_realm->commit_transaction();

4✔

946

            subscribe_to_and_add_objects(local_realm, num_objects_added_after); // these are lost!

4✔

947

};

4✔

948

2✔

949

        VersionID expected_version;

4✔

950

2✔

951

        auto store_pre_reset_state = [&](SharedRealm local_realm) {

4✔

952

            expected_version = local_realm->read_transaction_version();

4✔

953

};

4✔

954

2✔

955

        auto verify_post_reset_state = [&, err_future = std::move(error_future)](SharedRealm local_realm) {

4✔

956

            auto sync_error = err_future.get();

4✔

961

2✔

962

            // All changes should have been rolled back when recovery hit remove_column(),

2✔

963

            // leaving the Realm in the pre-reset state

2✔

964

            local_realm->refresh();

4✔

965

            auto table = local_realm->read_group().get_table("class_TopLevel");

4✔

966

            ColKey added = table->get_column_key(added_property_name);

4✔

968

            const size_t expected_added_objects = num_objects_added_before + num_objects_added_after;

4✔

970

            size_t count_of_valid_array_data = validate_integrity_of_arrays(table);

4✔

972

2✔

973

            // The attempted client reset should have been recorded so that we

2✔

974

            // don't attempt it again

2✔

976

};

4✔

977

2✔

978

        SECTION("Recover: unsuccessful recovery leads to a manual reset") {

4✔

979

            config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

980

            auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

981

            test_reset->make_local_changes(make_local_changes_that_will_fail)

2✔

982

                ->on_post_local_changes(store_pre_reset_state)

2✔

983

                ->on_post_reset(std::move(verify_post_reset_state))

2✔

984

                ->run();

2✔

985

            RealmConfig config_copy = config_local;

2✔

986

            auto&& [error_future2, err_handler2] = make_error_handler();

2✔

987

            config_copy.sync_config->error_handler = err_handler2;

2✔

988

            auto realm_post_reset = Realm::get_shared_realm(config_copy);

2✔

989

            auto sync_error = wait_for_future(std::move(error_future2)).get();

2✔

994

2✔

995

2✔

996

        SECTION("RecoverOrDiscard: unsuccessful reapply leads to discard") {

4✔

997

            config_local.sync_config->client_resync_mode = ClientResyncMode::RecoverOrDiscard;

2✔

998

            auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

999

            test_reset->make_local_changes(make_local_changes_that_will_fail)

2✔

1000

                ->on_post_local_changes(store_pre_reset_state)

2✔

1001

                ->on_post_reset(std::move(verify_post_reset_state))

2✔

1002

                ->run();

2✔

1003

1✔

1004

            RealmConfig config_copy = config_local;

2✔

1005

            auto&& [client_reset_future, reset_handler] = make_client_reset_handler();

2✔

1006

            config_copy.sync_config->error_handler = [](std::shared_ptr<SyncSession>, SyncError err) {

1✔

1010

            config_copy.sync_config->notify_after_client_reset = reset_handler;

2✔

1011

            auto realm_post_reset = Realm::get_shared_realm(config_copy);

2✔

1012

            ClientResyncMode mode = wait_for_future(std::move(client_reset_future)).get();

2✔

1014

            realm_post_reset->refresh();

2✔

1015

            auto table = realm_post_reset->read_group().get_table("class_TopLevel");

2✔

1016

            ColKey added = table->get_column_key(added_property_name);

2✔

1019

2✔

1020

4✔

1021

22✔

1022

    SECTION("DiscardLocal: offline writes and subscriptions are lost") {

44✔

1023

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1024

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

1025

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

1026

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

1027

        test_reset

2✔

1028

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

1029

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);

2✔

1030

})

2✔

1031

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

1032

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

1033

})

2✔

1034

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) mutable {

2✔

1035

                ClientResyncMode mode = wait_for_future(std::move(client_reset_future)).get();

2✔

1037

                auto subs = local_realm->get_latest_subscription_set();

2✔

1038

                wait_for_future(subs.get_state_change_notification(sync::SubscriptionSet::State::Complete)).get();

2✔

1039

                local_realm->refresh();

2✔

1040

                auto table = local_realm->read_group().get_table("class_TopLevel");

2✔

1041

                auto queryable_str_field = table->get_column_key("queryable_str_field");

2✔

1042

                auto queryable_int_field = table->get_column_key("queryable_int_field");

2✔

1043

                auto tv = table->where().equal(queryable_str_field, StringData(str_field_value)).find_all();

2✔

1044

                // the object we created while offline was discarded, and the remote object was not downloaded

1✔

1046

                size_t count_of_foo = count_queries_with_str(subs, util::format("\"%1\"", str_field_value));

2✔

1047

                // make sure that the subscription for "foo" did not survive the reset

1✔

1050

1✔

1051

                // adding data and subscriptions to a reset Realm works as normal

1✔

1052

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);

2✔

1053

                auto latest_subs = local_realm->get_latest_subscription_set();

2✔

1055

                wait_for_future(latest_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete))

2✔

1056

                    .get();

2✔

1057

                local_realm->refresh();

2✔

1058

                count_of_foo = count_queries_with_str(latest_subs, util::format("\"%1\"", str_field_value));

2✔

1060

                tv = table->where().equal(queryable_str_field, StringData(str_field_value)).find_all();

2✔

1062

                tv.sort(queryable_int_field);

2✔

1065

})

2✔

1066

            ->run();

2✔

1067

2✔

1068

22✔

1069

    SECTION("DiscardLocal: an invalid subscription made while offline becomes superseded") {

44✔

1070

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1071

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

1072

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

1073

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

1074

        std::unique_ptr<sync::SubscriptionSet> invalid_sub;

2✔

1075

        test_reset

2✔

1076

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

1077

                invalid_sub = std::make_unique<sync::SubscriptionSet>(add_invalid_subscription(local_realm));

2✔

1078

                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);

2✔

1079

})

2✔

1080

            ->make_remote_changes([&](SharedRealm remote_realm) {

2✔

1081

                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);

2✔

1082

})

2✔

1083

            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {

2✔

1084

                local_realm->refresh();

2✔

1085

                sync::SubscriptionSet::State actual =

2✔

1086

                    invalid_sub->get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

1088

                ClientResyncMode mode = client_reset_future.get();

2✔

1090

})

2✔

1091

            ->run();

2✔

1092

2✔

1093

22✔

1094

    SECTION("DiscardLocal: an error is produced if a previously successful query becomes invalid due to "

44✔

1095

            "server changes across a reset") {

23✔

1096

        // Disable dev mode so non-queryable fields are not automatically added as queryable

1✔

1097

        const AppSession& app_session = harness.session().app_session();

2✔

1098

        app_session.admin_api.set_development_mode_to(app_session.server_app_id, false);

2✔

1099

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1100

        auto&& [error_future, err_handler] = make_error_handler();

2✔

1101

        config_local.sync_config->error_handler = err_handler;

2✔

1102

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

1103

        test_reset

2✔

1104

            ->setup([&](SharedRealm realm) {

2✔

1105

                if (realm->sync_session()->path() == config_local.path) {

2✔

1106

                    auto added_sub = add_subscription_for_new_object(realm, str_field_value, 0);

2✔

1107

                    added_sub.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

1108

2✔

1109

})

2✔

1110

            ->make_local_changes([&](SharedRealm local_realm) {

2✔

1111

                add_object(local_realm, str_field_value, local_added_int);

2✔

1112

                // Make "queryable_str_field" not a valid query field.

1✔

1113

                // Pre-reset, the Realm had a successful query on it, but now when the client comes back online

1✔

1114

                // and tries to reset, the fresh Realm download will fail with a query error.

1✔

1115

                const AppSession& app_session = harness.session().app_session();

2✔

1116

                auto baas_sync_service = app_session.admin_api.get_sync_service(app_session.server_app_id);

2✔

1117

                auto baas_sync_config =

2✔

1118

                    app_session.admin_api.get_config(app_session.server_app_id, baas_sync_service);

2✔

1120

                auto it = baas_sync_config.queryable_field_names->begin();

2✔

1121

                for (; it != baas_sync_config.queryable_field_names->end(); ++it) {

2✔

1122

                    if (*it == "queryable_str_field") {

2✔

1123

                        break;

2✔

1124

2✔

1125

2✔

1127

                baas_sync_config.queryable_field_names->erase(it);

2✔

1128

                app_session.admin_api.enable_sync(app_session.server_app_id, baas_sync_service.id, baas_sync_config);

2✔

1129

})

2✔

1130

            ->on_post_reset([&, err_future = std::move(error_future)](SharedRealm) mutable {

2✔

1131

                auto sync_error = wait_for_future(std::move(err_future)).get();

2✔

1132

                INFO(sync_error.status);

2✔

1134

})

2✔

1135

            ->run();

2✔

1136

2✔

1137

22✔

1138

    SECTION("DiscardLocal: completion callbacks fire after client reset even when there is no data to download") {

44✔

1139

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1140

        auto&& [reset_future, reset_handler] = make_client_reset_handler();

2✔

1141

        config_local.sync_config->notify_after_client_reset = reset_handler;

2✔

1142

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

1143

        test_reset

2✔

1144

            ->on_post_local_changes([&](SharedRealm realm) {

2✔

1145

                wait_for_upload(*realm);

2✔

1146

                wait_for_download(*realm);

2✔

1147

})

2✔

1148

            ->run();

2✔

1149

2✔

1150

22✔

1151

    SECTION("DiscardLocal: open realm after client reset failure") {

44✔

1152

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1153

        auto&& [error_future, error_handler] = make_error_handler();

2✔

1154

        config_local.sync_config->error_handler = error_handler;

2✔

1155

1✔

1156

        std::string fresh_path = realm::_impl::client_reset::get_fresh_path_for(config_local.path);

2✔

1157

        // create a non-empty directory that we'll fail to delete

1✔

1158

        util::make_dir(fresh_path);

2✔

1159

        util::File(util::File::resolve("file", fresh_path), util::File::mode_Write);

2✔

1160

1✔

1161

        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());

2✔

1162

        test_reset->run();

2✔

1163

1✔

1164

        // Client reset fails due to sync client not being able to create the fresh realm.

1✔

1165

        auto sync_error = wait_for_future(std::move(error_future)).get();

2✔

1167

1✔

1168

        // Open the realm again. This should not crash.

1✔

1169

        auto&& [err_future, err_handler] = make_error_handler();

2✔

1170

        config_local.sync_config->error_handler = std::move(err_handler);

2✔

1171

1✔

1172

        auto realm_post_reset = Realm::get_shared_realm(config_local);

2✔

1173

        sync_error = wait_for_future(std::move(err_future)).get();

2✔

1175

2✔

1176

22✔

1177

    enum class ResetMode { NoReset, InitiateClientReset };

44✔

1178

    auto seed_realm = [&harness, &subscribe_to_and_add_objects](RealmConfig config, ResetMode reset_mode) {

35✔

1179

        config.sync_config->error_handler = [path = config.path](std::shared_ptr<SyncSession>, SyncError err) {

13✔

1183

        SharedRealm realm = Realm::get_shared_realm(config);

26✔

1184

        subscribe_to_and_add_objects(realm, 1);

26✔

1185

        auto subs = realm->get_latest_subscription_set();

26✔

1186

        auto result = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

26✔

1188

        if (reset_mode == ResetMode::InitiateClientReset) {

26✔

1189

            reset_utils::trigger_client_reset(harness.session().app_session(), realm);

18✔

1190

18✔

1191

        realm->close();

26✔

1192

};

26✔

1193

22✔

1194

    auto setup_reset_handlers_for_schema_validation =

44✔

1195

        [&before_reset_count, &after_reset_count](RealmConfig& config, Schema expected_schema) {

29✔

1196

            auto& sync_config = *config.sync_config;

14✔

1197

            sync_config.error_handler = [](std::shared_ptr<SyncSession>, SyncError err) {

7✔

1200

            sync_config.notify_before_client_reset = [&before_reset_count,

14✔

1201

                                                      expected = expected_schema](SharedRealm frozen_before) {

14✔

1202

                ++before_reset_count;

14✔

1206

};

14✔

1207

7✔

1208

            auto [promise, future] = util::make_promise_future<void>();

14✔

1209

            sync_config.notify_after_client_reset =

14✔

1210

                [&after_reset_count, promise = util::CopyablePromiseHolder<void>(std::move(promise)), expected_schema,

14✔

1211

                 reset_mode = config.sync_config->client_resync_mode, has_schema = config.schema.has_value()](

14✔

1212

                    SharedRealm frozen_before, ThreadSafeReference after_ref, bool did_recover) mutable {

14✔

1213

                    ++after_reset_count;

14✔

1217

                    SharedRealm after = Realm::get_shared_realm(std::move(after_ref), util::Scheduler::make_dummy());

14✔

1218

                    if (!has_schema) {

14✔

1219

                        after->set_schema_subset(expected_schema);

4✔

1220

4✔

1223

                    // the above check is sufficient unless operator==() is changed to not care about ordering

7✔

1224

                    // so future proof that by explicitly checking the order of properties here as well

7✔

1226

                    auto after_it = after->schema().find("TopLevel");

14✔

1227

                    auto before_it = frozen_before->schema().find("TopLevel");

14✔

1237

                    promise.get_promise().emplace_value();

14✔

1238

};

14✔

1239

            return std::move(future); // move is not redundant here because of how destructing works

14✔

1240

};

14✔

1241

22✔

1242

    SECTION("Recover: schema indexes match in before and after states") {

44✔

1243

        seed_realm(config_local, ResetMode::InitiateClientReset);

2✔

1244

        // reorder a property such that it does not match the on disk property order

1✔

1245

        std::vector<ObjectSchema> local_schema = schema;

2✔

1246

        std::swap(local_schema[0].persisted_properties[1], local_schema[0].persisted_properties[2]);

2✔

1247

        local_schema[0].persisted_properties.push_back(

2✔

1248

            {"queryable_oid_field", PropertyType::ObjectId | PropertyType::Nullable});

2✔

1249

        config_local.schema = local_schema;

2✔

1250

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

1251

        auto future = setup_reset_handlers_for_schema_validation(config_local, local_schema);

2✔

1252

        SharedRealm realm = Realm::get_shared_realm(config_local);

2✔

1253

        future.get();

2✔

1256

2✔

1257

22✔

1258

    SECTION("Adding a local property matching a server addition is allowed") {

44✔

1259

        auto mode = GENERATE(ClientResyncMode::DiscardLocal, ClientResyncMode::Recover);

4✔

1260

        config_local.sync_config->client_resync_mode = mode;

4✔

1261

        seed_realm(config_local, ResetMode::InitiateClientReset);

4✔

1262

        std::vector<ObjectSchema> changed_schema = schema;

4✔

1263

        changed_schema[0].persisted_properties.push_back(

4✔

1264

            {"queryable_oid_field", PropertyType::ObjectId | PropertyType::Nullable});

4✔

1265

        // In a separate Realm, make the property addition.

2✔

1266

        // Since this is dev mode, it will be added to the server's schema.

2✔

1267

        config_remote.schema = changed_schema;

4✔

1268

        seed_realm(config_remote, ResetMode::NoReset);

4✔

1269

        std::swap(changed_schema[0].persisted_properties[1], changed_schema[0].persisted_properties[2]);

4✔

1270

        config_local.schema = changed_schema;

4✔

1271

        auto future = setup_reset_handlers_for_schema_validation(config_local, changed_schema);

4✔

1272

2✔

1273

        async_open_realm(config_local,

4✔

1274

                         [&, fut = std::move(future)](ThreadSafeReference&& ref, std::exception_ptr error) {

4✔

1277

                             auto realm = Realm::get_shared_realm(std::move(ref));

4✔

1278

                             fut.get();

4✔

1281

});

4✔

1282

4✔

1283

22✔

1284

    SECTION("Adding a local property matching a server addition inside the before reset callback is allowed") {

44✔

1285

        auto mode = GENERATE(ClientResyncMode::DiscardLocal, ClientResyncMode::Recover);

4✔

1286

        config_local.sync_config->client_resync_mode = mode;

4✔

1287

        seed_realm(config_local, ResetMode::InitiateClientReset);

4✔

1288

        std::vector<ObjectSchema> changed_schema = schema;

4✔

1289

        changed_schema[0].persisted_properties.push_back(

4✔

1290

            {"queryable_oid_field", PropertyType::ObjectId | PropertyType::Nullable});

4✔

1291

        // In a separate Realm, make the property addition.

2✔

1292

        // Since this is dev mode, it will be added to the server's schema.

2✔

1293

        config_remote.schema = changed_schema;

4✔

1294

        seed_realm(config_remote, ResetMode::NoReset);

4✔

1295

        std::swap(changed_schema[0].persisted_properties[1], changed_schema[0].persisted_properties[2]);

4✔

1296

        config_local.schema.reset();

4✔

1297

        config_local.sync_config->freeze_before_reset_realm = false;

4✔

1298

        auto future = setup_reset_handlers_for_schema_validation(config_local, changed_schema);

4✔

1299

2✔

1300

        auto notify_before = std::move(config_local.sync_config->notify_before_client_reset);

4✔

1301

        config_local.sync_config->notify_before_client_reset = [=](std::shared_ptr<Realm> realm) {

4✔

1302

            realm->update_schema(changed_schema);

4✔

1303

            notify_before(realm);

4✔

1304

};

4✔

1305

2✔

1306

        auto notify_after = std::move(config_local.sync_config->notify_after_client_reset);

4✔

1307

        config_local.sync_config->notify_after_client_reset = [=](std::shared_ptr<Realm> before,

4✔

1308

                                                                  ThreadSafeReference after, bool did_recover) {

4✔

1309

            before->set_schema_subset(changed_schema);

4✔

1310

            notify_after(before, std::move(after), did_recover);

4✔

1311

};

4✔

1312

2✔

1313

        async_open_realm(config_local,

4✔

1314

                         [&, fut = std::move(future)](ThreadSafeReference&& ref, std::exception_ptr error) {

4✔

1317

                             auto realm = Realm::get_shared_realm(std::move(ref));

4✔

1318

                             fut.get();

4✔

1321

});

4✔

1322

4✔

1323

22✔

1324

    auto make_additive_changes = [](std::vector<ObjectSchema> schema) {

25✔

1325

        schema[0].persisted_properties.push_back(

6✔

1326

            {"added_oid_field", PropertyType::ObjectId | PropertyType::Nullable});

6✔

1327

        std::swap(schema[0].persisted_properties[1], schema[0].persisted_properties[2]);

6✔

1328

        schema.push_back({"AddedClass",

6✔

1329

6✔

1330

                              {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

6✔

1331

                              {"str_field", PropertyType::String | PropertyType::Nullable},

6✔

1332

                          }});

6✔

1333

        return schema;

6✔

1334

};

6✔

1335

    SECTION("Recover: additive schema changes are recovered in dev mode") {

44✔

1336

        const AppSession& app_session = harness.session().app_session();

2✔

1337

        app_session.admin_api.set_development_mode_to(app_session.server_app_id, true);

2✔

1338

        seed_realm(config_local, ResetMode::InitiateClientReset);

2✔

1339

        std::vector<ObjectSchema> changed_schema = make_additive_changes(schema);

2✔

1340

        config_local.schema = changed_schema;

2✔

1341

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

1342

        ThreadSafeReference ref_async;

2✔

1343

        auto future = setup_reset_handlers_for_schema_validation(config_local, changed_schema);

2✔

1344

        async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {

2✔

1347

            ref_async = std::move(ref);

2✔

1348

});

2✔

1349

        future.get();

2✔

1352

2✔

1353

            auto realm = Realm::get_shared_realm(std::move(ref_async));

2✔

1354

            // make changes to the newly added property

1✔

1355

            realm->begin_transaction();

2✔

1356

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

1357

            ColKey new_col = table->get_column_key("added_oid_field");

2✔

1359

            for (auto it = table->begin(); it != table->end(); ++it) {

4✔

1360

                it->set(new_col, ObjectId::gen());

2✔

1361

2✔

1362

            realm->commit_transaction();

2✔

1363

            // subscribe to the new Class and add an object

1✔

1364

            auto new_table = realm->read_group().get_table("class_AddedClass");

2✔

1365

            auto sub_set = realm->get_latest_subscription_set();

2✔

1366

            auto mut_sub = sub_set.make_mutable_copy();

2✔

1367

            mut_sub.insert_or_assign(Query(new_table));

2✔

1368

            mut_sub.commit();

2✔

1369

            realm->begin_transaction();

2✔

1371

            new_table->create_object_with_primary_key(ObjectId::gen());

2✔

1372

            realm->commit_transaction();

2✔

1373

            auto result = realm->get_latest_subscription_set()

2✔

1374

                              .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

1375

                              .get();

2✔

1377

            wait_for_advance(*realm);

2✔

1378

            realm->close();

2✔

1379

2✔

1380

2✔

1381

            // ensure that an additional schema change after the successful reset is also accepted by the server

1✔

1382

            changed_schema[0].persisted_properties.push_back(

2✔

1383

                {"added_oid_field_second", PropertyType::ObjectId | PropertyType::Nullable});

2✔

1384

            changed_schema.push_back({"AddedClassSecond",

2✔

1385

2✔

1386

                                          {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

1387

                                          {"str_field_2", PropertyType::String | PropertyType::Nullable},

2✔

1388

                                      }});

2✔

1389

            config_local.schema = changed_schema;

2✔

1390

            async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {

2✔

1393

                auto realm = Realm::get_shared_realm(std::move(ref));

2✔

1394

                auto table = realm->read_group().get_table("class_AddedClassSecond");

2✔

1395

                ColKey new_col = table->get_column_key("str_field_2");

2✔

1397

                auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

1398

                new_subs.insert_or_assign(Query(table).equal(new_col, "hello"));

2✔

1399

                auto subs = new_subs.commit();

2✔

1400

                realm->begin_transaction();

2✔

1401

                table->create_object_with_primary_key(Mixed{ObjectId::gen()}, {{new_col, "hello"}});

2✔

1402

                realm->commit_transaction();

2✔

1403

                subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

1404

                wait_for_advance(*realm);

2✔

1406

});

2✔

1407

2✔

1408

2✔

1409

22✔

1410

    SECTION("DiscardLocal: additive schema changes not allowed") {

44✔

1411

        seed_realm(config_local, ResetMode::InitiateClientReset);

2✔

1412

        std::vector<ObjectSchema> changed_schema = make_additive_changes(schema);

2✔

1413

        config_local.schema = changed_schema;

2✔

1414

        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;

2✔

1415

        auto&& [error_future, err_handler] = make_error_handler();

2✔

1416

        config_local.sync_config->error_handler = err_handler;

2✔

1417

        async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {

2✔

1420

            REQUIRE_THROWS_CONTAINING(std::rethrow_exception(error),

2✔

1421

                                      "A fatal error occurred during client reset: 'Client reset cannot recover when "

2✔

1422

                                      "classes have been removed: {AddedClass}'");

2✔

1423

});

2✔

1424

        error_future.get();

2✔

1427

2✔

1428

22✔

1429

    SECTION("Recover: incompatible schema changes on async open are an error") {

44✔

1430

        seed_realm(config_local, ResetMode::InitiateClientReset);

2✔

1431

        std::vector<ObjectSchema> changed_schema = schema;

2✔

1432

        changed_schema[0].persisted_properties[0].type = PropertyType::UUID; // incompatible type change

2✔

1433

        config_local.schema = changed_schema;

2✔

1434

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

1435

        auto&& [error_future, err_handler] = make_error_handler();

2✔

1436

        config_local.sync_config->error_handler = err_handler;

2✔

1437

        async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {

2✔

1440

            REQUIRE_THROWS_CONTAINING(

2✔

1441

                std::rethrow_exception(error),

2✔

1442

                "A fatal error occurred during client reset: 'The following changes cannot be "

2✔

1443

                "made in additive-only schema mode:\n"

2✔

1444

                "- Property 'TopLevel._id' has been changed from 'object id' to 'uuid'.\nIf your app is running in "

2✔

1445

                "development mode, you can delete the realm and restart the app to update your schema.'");

2✔

1446

});

2✔

1447

        error_future.get();

2✔

1450

2✔

1451

22✔

1452

    SECTION("Recover: additive schema changes without dev mode produce an error after client reset") {

44✔

1453

        const AppSession& app_session = harness.session().app_session();

2✔

1454

        app_session.admin_api.set_development_mode_to(app_session.server_app_id, true);

2✔

1455

        seed_realm(config_local, ResetMode::InitiateClientReset);

2✔

1456

        // Disable dev mode so that schema changes are not allowed

1✔

1457

        app_session.admin_api.set_development_mode_to(app_session.server_app_id, false);

2✔

1458

        auto cleanup = util::make_scope_exit([&]() noexcept {

2✔

1459

            const AppSession& app_session = harness.session().app_session();

2✔

1460

            app_session.admin_api.set_development_mode_to(app_session.server_app_id, true);

2✔

1461

});

2✔

1462

1✔

1463

        std::vector<ObjectSchema> changed_schema = make_additive_changes(schema);

2✔

1464

        config_local.schema = changed_schema;

2✔

1465

        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

1466

        (void)setup_reset_handlers_for_schema_validation(config_local, changed_schema);

2✔

1467

        auto&& [error_future, err_handler] = make_error_handler();

2✔

1468

        config_local.sync_config->error_handler = err_handler;

2✔

1469

        async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {

2✔

1472

            auto realm = Realm::get_shared_realm(std::move(ref));

2✔

1473

            // make changes to the new property

1✔

1474

            realm->begin_transaction();

2✔

1475

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

1476

            ColKey new_col = table->get_column_key("added_oid_field");

2✔

1478

            for (auto it = table->begin(); it != table->end(); ++it) {

4✔

1479

                it->set(new_col, ObjectId::gen());

2✔

1480

2✔

1481

            realm->commit_transaction();

2✔

1482

});

2✔

1483

        auto realm = Realm::get_shared_realm(config_local);

2✔

1484

        auto err = error_future.get();

2✔

1485

        std::string property_err = "Invalid schema change (UPLOAD): non-breaking schema change: adding "

2✔

1486

                                   "\"ObjectID\" column at field \"added_oid_field\" in schema \"TopLevel\", "

2✔

1487

                                   "schema changes from clients are restricted when developer mode is disabled";

2✔

1488

        std::string class_err = "Invalid schema change (UPLOAD): non-breaking schema change: adding schema "

2✔

1489

                                "for Realm table \"AddedClass\", schema changes from clients are restricted when "

2✔

1490

                                "developer mode is disabled";

2✔

1491

        REQUIRE_THAT(err.status.reason(), Catch::Matchers::ContainsSubstring(property_err) ||

2✔

1492

                                              Catch::Matchers::ContainsSubstring(class_err));

2✔

1495

2✔

1496

44✔

1498

TEST_CASE("flx: creating an object on a class with no subscription throws", "[sync][flx][subscription][baas]") {

2✔

1499

    FLXSyncTestHarness harness("flx_bad_query", {g_simple_embedded_obj_schema, {"queryable_str_field"}});

2✔

1500

    harness.do_with_new_user([&](auto user) {

2✔

1501

        SyncTestFile config(user, harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

1502

        auto [error_promise, error_future] = util::make_promise_future<SyncError>();

2✔

1503

        auto shared_promise = std::make_shared<decltype(error_promise)>(std::move(error_promise));

2✔

1504

        config.sync_config->error_handler = [error_promise = std::move(shared_promise)](std::shared_ptr<SyncSession>,

2✔

1505

                                                                                        SyncError err) {

1✔

1509

1✔

1510

        auto realm = Realm::get_shared_realm(config);

2✔

1511

        CppContext c(realm);

2✔

1512

        realm->begin_transaction();

2✔

1513

        REQUIRE_THROWS_AS(

2✔

1514

            Object::create(c, realm, "TopLevel",

2✔

1515

                           std::any(AnyDict{{"_id", ObjectId::gen()}, {"queryable_str_field", "foo"s}})),

2✔

1516

            NoSubscriptionForWrite);

2✔

1517

        realm->cancel_transaction();

2✔

1518

1✔

1519

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

1520

1✔

1522

        auto col_key = table->get_column_key("queryable_str_field");

2✔

1523

2✔

1524

            auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

1525

            new_subs.insert_or_assign(Query(table).equal(col_key, "foo"));

2✔

1526

            auto subs = new_subs.commit();

2✔

1527

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

1528

2✔

1529

1✔

1530

        realm->begin_transaction();

2✔

1531

        auto obj = Object::create(c, realm, "TopLevel",

2✔

1532

                                  std::any(AnyDict{{"_id", ObjectId::gen()},

2✔

1533

                                                   {"queryable_str_field", "foo"s},

2✔

1534

                                                   {"embedded_obj", AnyDict{{"str_field", "bar"s}}}}));

2✔

1535

        realm->commit_transaction();

2✔

1536

1✔

1537

        realm->begin_transaction();

2✔

1538

        auto embedded_obj = util::any_cast<Object&&>(obj.get_property_value<std::any>(c, "embedded_obj"));

2✔

1539

        embedded_obj.set_property_value(c, "str_field", std::any{"baz"s});

2✔

1540

        realm->commit_transaction();

2✔

1541

1✔

1542

        wait_for_upload(*realm);

2✔

1543

        wait_for_download(*realm);

2✔

1544

});

2✔

1545

2✔

1548

          "[sync][flx][compensating write][baas]") {

16✔

1549

    static std::optional<FLXSyncTestHarness> harness;

16✔

1550

    if (!harness) {

16✔

1551

        Schema schema{{"TopLevel",

2✔

1552

                       {{"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

1553

                        {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

2✔

1554

                        {"embedded_obj", PropertyType::Object | PropertyType::Nullable, "TopLevel_embedded_obj"}}},

2✔

1555

                      {"TopLevel_embedded_obj",

2✔

1556

                       ObjectSchema::ObjectType::Embedded,

2✔

1557

                       {{"str_field", PropertyType::String | PropertyType::Nullable}}},

2✔

1558

                      {"Int PK",

2✔

1559

2✔

1560

                           {"_id", PropertyType::Int, Property::IsPrimary{true}},

2✔

1561

                           {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

2✔

1562

}},

2✔

1563

                      {"String PK",

2✔

1564

2✔

1565

                           {"_id", PropertyType::String, Property::IsPrimary{true}},

2✔

1566

                           {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

2✔

1567

}},

2✔

1568

                      {"UUID PK",

2✔

1569

2✔

1570

                           {"_id", PropertyType::UUID, Property::IsPrimary{true}},

2✔

1571

                           {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

2✔

1572

                       }}};

2✔

1573

1✔

1574

        AppCreateConfig::ServiceRole role;

2✔

1575

        role.name = "compensating_write_perms";

2✔

1576

1✔

1577

        AppCreateConfig::ServiceRoleDocumentFilters doc_filters;

2✔

1578

        doc_filters.read = true;

2✔

1579

        doc_filters.write = {{"queryable_str_field", {{"$in", nlohmann::json::array({"foo", "bar"})}}}};

2✔

1580

        role.document_filters = doc_filters;

2✔

1581

1✔

1582

        role.insert_filter = true;

2✔

1583

        role.delete_filter = true;

2✔

1584

        role.read = true;

2✔

1585

        role.write = true;

2✔

1586

        FLXSyncTestHarness::ServerSchema server_schema{schema, {"queryable_str_field"}, {role}};

2✔

1587

        harness.emplace("flx_bad_query", server_schema);

2✔

1588

2✔

1589

8✔

1590

    create_user_and_log_in(harness->app());

16✔

1591

    auto user = harness->app()->current_user();

16✔

1592

8✔

1593

    auto make_error_handler = [] {

16✔

1594

        auto [error_promise, error_future] = util::make_promise_future<SyncError>();

16✔

1595

        auto shared_promise = std::make_shared<decltype(error_promise)>(std::move(error_promise));

16✔

1596

        auto fn = [error_promise = std::move(shared_promise)](std::shared_ptr<SyncSession>, SyncError err) mutable {

15✔

1597

            if (!error_promise) {

14✔

1603

            error_promise->emplace_value(std::move(err));

14✔

1604

            error_promise.reset();

14✔

1605

};

14✔

1606

8✔

1607

        return std::make_pair(std::move(error_future), std::move(fn));

16✔

1608

};

16✔

1609

8✔

1610

    auto validate_sync_error = [&](const SyncError& sync_error, Mixed expected_pk, const char* expected_object_name,

16✔

1611

                                   const std::string& error_msg_fragment) {

15✔

1616

        auto write_info = sync_error.compensating_writes_info[0];

14✔

1619

        CHECK_THAT(write_info.reason, Catch::Matchers::ContainsSubstring(error_msg_fragment));

14✔

1620

};

14✔

1621

8✔

1622

    SyncTestFile config(user, harness->schema(), SyncConfig::FLXSyncEnabled{});

16✔

1623

    auto&& [error_future, err_handler] = make_error_handler();

16✔

1624

    config.sync_config->error_handler = err_handler;

16✔

1625

    auto realm = Realm::get_shared_realm(config);

16✔

1626

    auto table = realm->read_group().get_table("class_TopLevel");

16✔

1627

8✔

1628

    auto create_subscription = [&](StringData table_name, auto make_query) {

15✔

1629

        auto table = realm->read_group().get_table(table_name);

14✔

1630

        auto queryable_str_field = table->get_column_key("queryable_str_field");

14✔

1631

        auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

14✔

1632

        new_query.insert_or_assign(make_query(Query(table), queryable_str_field));

14✔

1633

        new_query.commit();

14✔

1634

};

14✔

1635

8✔

1636

    SECTION("compensating write because of permission violation") {

16✔

1637

        create_subscription("class_TopLevel", [](auto q, auto col) {

2✔

1638

            return q.equal(col, "bizz");

2✔

1639

});

2✔

1640

1✔

1641

        CppContext c(realm);

2✔

1642

        realm->begin_transaction();

2✔

1643

        auto invalid_obj = ObjectId::gen();

2✔

1644

        Object::create(c, realm, "TopLevel",

2✔

1645

                       std::any(AnyDict{{"_id", invalid_obj}, {"queryable_str_field", "bizz"s}}));

2✔

1646

        realm->commit_transaction();

2✔

1647

1✔

1648

        validate_sync_error(

2✔

1649

            std::move(error_future).get(), invalid_obj, "TopLevel",

2✔

1650

            util::format("write to ObjectID(\"%1\") in table \"TopLevel\" not allowed", invalid_obj.to_string()));

2✔

1651

1✔

1652

        wait_for_advance(*realm);

2✔

1653

1✔

1654

        auto top_level_table = realm->read_group().get_table("class_TopLevel");

2✔

1656

2✔

1657

8✔

1658

    SECTION("compensating write because of permission violation with write on embedded object") {

16✔

1659

        create_subscription("class_TopLevel", [](auto q, auto col) {

2✔

1660

            return q.equal(col, "bizz").Or().equal(col, "foo");

2✔

1661

});

2✔

1662

1✔

1663

        CppContext c(realm);

2✔

1664

        realm->begin_transaction();

2✔

1665

        auto invalid_obj = ObjectId::gen();

2✔

1666

        auto obj = Object::create(c, realm, "TopLevel",

2✔

1667

                                  std::any(AnyDict{{"_id", invalid_obj},

2✔

1668

                                                   {"queryable_str_field", "foo"s},

2✔

1669

                                                   {"embedded_obj", AnyDict{{"str_field", "bar"s}}}}));

2✔

1670

        realm->commit_transaction();

2✔

1671

        realm->begin_transaction();

2✔

1672

        obj.set_property_value(c, "queryable_str_field", std::any{"bizz"s});

2✔

1673

        realm->commit_transaction();

2✔

1674

        realm->begin_transaction();

2✔

1675

        auto embedded_obj = util::any_cast<Object&&>(obj.get_property_value<std::any>(c, "embedded_obj"));

2✔

1676

        embedded_obj.set_property_value(c, "str_field", std::any{"baz"s});

2✔

1677

        realm->commit_transaction();

2✔

1678

1✔

1679

        validate_sync_error(

2✔

1680

            std::move(error_future).get(), invalid_obj, "TopLevel",

2✔

1681

            util::format("write to ObjectID(\"%1\") in table \"TopLevel\" not allowed", invalid_obj.to_string()));

2✔

1682

1✔

1683

        wait_for_advance(*realm);

2✔

1684

1✔

1685

        obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any(invalid_obj));

2✔

1686

        embedded_obj = util::any_cast<Object&&>(obj.get_property_value<std::any>(c, "embedded_obj"));

2✔

1689

1✔

1690

        realm->begin_transaction();

2✔

1691

        embedded_obj.set_property_value(c, "str_field", std::any{"baz"s});

2✔

1692

        realm->commit_transaction();

2✔

1693

1✔

1694

        wait_for_upload(*realm);

2✔

1695

        wait_for_download(*realm);

2✔

1696

1✔

1697

        wait_for_advance(*realm);

2✔

1698

        obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any(invalid_obj));

2✔

1699

        embedded_obj = util::any_cast<Object&&>(obj.get_property_value<std::any>(c, "embedded_obj"));

2✔

1701

2✔

1702

8✔

1703

    SECTION("compensating write for writing a top-level object that is out-of-view") {

16✔

1704

        create_subscription("class_TopLevel", [](auto q, auto col) {

2✔

1705

            return q.equal(col, "foo");

2✔

1706

});

2✔

1707

1✔

1708

        CppContext c(realm);

2✔

1709

        realm->begin_transaction();

2✔

1710

        auto valid_obj = ObjectId::gen();

2✔

1711

        auto invalid_obj = ObjectId::gen();

2✔

1712

        Object::create(c, realm, "TopLevel",

2✔

1713

                       std::any(AnyDict{

2✔

1714

                           {"_id", valid_obj},

2✔

1715

                           {"queryable_str_field", "foo"s},

2✔

1716

                       }));

2✔

1717

        Object::create(c, realm, "TopLevel",

2✔

1718

                       std::any(AnyDict{

2✔

1719

                           {"_id", invalid_obj},

2✔

1720

                           {"queryable_str_field", "bar"s},

2✔

1721

                       }));

2✔

1722

        realm->commit_transaction();

2✔

1723

1✔

1724

        validate_sync_error(std::move(error_future).get(), invalid_obj, "TopLevel",

2✔

1725

                            "object is outside of the current query view");

2✔

1726

1✔

1727

        wait_for_advance(*realm);

2✔

1728

1✔

1729

        auto top_level_table = realm->read_group().get_table("class_TopLevel");

2✔

1732

1✔

1733

        // Verify that a valid object afterwards does not produce an error

1✔

1734

        realm->begin_transaction();

2✔

1735

        Object::create(c, realm, "TopLevel",

2✔

1736

                       std::any(AnyDict{

2✔

1737

                           {"_id", ObjectId::gen()},

2✔

1738

                           {"queryable_str_field", "foo"s},

2✔

1739

                       }));

2✔

1740

        realm->commit_transaction();

2✔

1741

1✔

1742

        wait_for_upload(*realm);

2✔

1743

        wait_for_download(*realm);

2✔

1744

2✔

1745

8✔

1746

    SECTION("compensating writes for each primary key type") {

16✔

1747

        SECTION("int") {

8✔

1748

            create_subscription("class_Int PK", [](auto q, auto col) {

2✔

1749

                return q.equal(col, "foo");

2✔

1750

});

2✔

1751

            realm->begin_transaction();

2✔

1752

            realm->read_group().get_table("class_Int PK")->create_object_with_primary_key(123456);

2✔

1753

            realm->commit_transaction();

2✔

1754

1✔

1755

            validate_sync_error(std::move(error_future).get(), 123456, "Int PK",

2✔

1756

                                "write to 123456 in table \"Int PK\" not allowed");

2✔

1757

2✔

1758

4✔

1759

        SECTION("short string") {

8✔

1760

            create_subscription("class_String PK", [](auto q, auto col) {

2✔

1761

                return q.equal(col, "foo");

2✔

1762

});

2✔

1763

            realm->begin_transaction();

2✔

1764

            realm->read_group().get_table("class_String PK")->create_object_with_primary_key("short");

2✔

1765

            realm->commit_transaction();

2✔

1766

1✔

1767

            validate_sync_error(std::move(error_future).get(), "short", "String PK",

2✔

1768

                                "write to \"short\" in table \"String PK\" not allowed");

2✔

1769

2✔

1770

4✔

1771

        SECTION("long string") {

8✔

1772

            create_subscription("class_String PK", [](auto q, auto col) {

2✔

1773

                return q.equal(col, "foo");

2✔

1774

});

2✔

1775

            realm->begin_transaction();

2✔

1776

            const char* pk = "long string which won't fit in the SSO buffer";

2✔

1777

            realm->read_group().get_table("class_String PK")->create_object_with_primary_key(pk);

2✔

1778

            realm->commit_transaction();

2✔

1779

1✔

1780

            validate_sync_error(std::move(error_future).get(), pk, "String PK",

2✔

1781

                                util::format("write to \"%1\" in table \"String PK\" not allowed", pk));

2✔

1782

2✔

1783

4✔

1784

        SECTION("uuid") {

8✔

1785

            create_subscription("class_UUID PK", [](auto q, auto col) {

2✔

1786

                return q.equal(col, "foo");

2✔

1787

});

2✔

1788

            realm->begin_transaction();

2✔

1789

            UUID pk("01234567-9abc-4def-9012-3456789abcde");

2✔

1790

            realm->read_group().get_table("class_UUID PK")->create_object_with_primary_key(pk);

2✔

1791

            realm->commit_transaction();

2✔

1792

1✔

1793

            validate_sync_error(std::move(error_future).get(), pk, "UUID PK",

2✔

1794

                                util::format("write to UUID(%1) in table \"UUID PK\" not allowed", pk));

2✔

1795

2✔

1796

8✔

1797

8✔

1798

    // Clear the Realm afterwards as we're reusing an app

8✔

1799

    realm->begin_transaction();

16✔

1800

    table->clear();

16✔

1801

    realm->commit_transaction();

16✔

1802

    wait_for_upload(*realm);

16✔

1803

    realm.reset();

16✔

1804

8✔

1805

    // Add new sections before this

8✔

1806

    SECTION("teardown") {

16✔

1807

        harness->app()->sync_manager()->wait_for_sessions_to_terminate();

2✔

1808

        harness.reset();

2✔

1809

2✔

1810

16✔

1812

TEST_CASE("flx: query on non-queryable field results in query error message", "[sync][flx][query][baas]") {

8✔

1813

    static std::optional<FLXSyncTestHarness> harness;

8✔

1814

    if (!harness) {

8✔

1815

        harness.emplace("flx_bad_query");

2✔

1816

2✔

1817

4✔

1818

    auto create_subscription = [](SharedRealm realm, StringData table_name, StringData column_name, auto make_query) {

10✔

1819

        auto table = realm->read_group().get_table(table_name);

10✔

1820

        auto queryable_field = table->get_column_key(column_name);

10✔

1821

        auto new_query = realm->get_active_subscription_set().make_mutable_copy();

10✔

1822

        new_query.insert_or_assign(make_query(Query(table), queryable_field));

10✔

1823

        return new_query.commit();

10✔

1824

};

10✔

1825

4✔

1826

    auto check_status = [](auto status) {

8✔

1828

        std::string reason = status.get_status().reason();

8✔

1829

        // Depending on the version of baas used, it may return 'Invalid query:' or

4✔

1830

        // 'Client provided query with bad syntax:'

4✔

1831

        if ((reason.find("Invalid query:") == std::string::npos &&

8✔

1833

            reason.find("\"TopLevel\": key \"non_queryable_field\" is not a queryable field") == std::string::npos) {

8✔

1836

};

8✔

1837

4✔

1838

    SECTION("Good query after bad query") {

8✔

1839

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

1840

            auto subs = create_subscription(realm, "class_TopLevel", "non_queryable_field", [](auto q, auto c) {

2✔

1841

                return q.equal(c, "bar");

2✔

1842

});

2✔

1843

            auto sub_res = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

1844

            check_status(sub_res);

2✔

1845

1✔

1848

1✔

1849

            subs = create_subscription(realm, "class_TopLevel", "queryable_str_field", [](auto q, auto c) {

2✔

1850

                return q.equal(c, "foo");

2✔

1851

});

2✔

1852

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

1853

1✔

1856

});

2✔

1857

2✔

1858

4✔

1859

    SECTION("Bad query after bad query") {

8✔

1860

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

1861

            auto sync_session = realm->sync_session();

2✔

1862

            sync_session->pause();

2✔

1863

1✔

1864

            auto subs = create_subscription(realm, "class_TopLevel", "non_queryable_field", [](auto q, auto c) {

2✔

1865

                return q.equal(c, "bar");

2✔

1866

});

2✔

1867

            auto subs2 = create_subscription(realm, "class_TopLevel", "non_queryable_field", [](auto q, auto c) {

2✔

1868

                return q.equal(c, "bar");

2✔

1869

});

2✔

1870

1✔

1871

            sync_session->resume();

2✔

1872

1✔

1873

            auto sub_res = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

1874

            auto sub_res2 =

2✔

1875

                subs2.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

1876

1✔

1877

            check_status(sub_res);

2✔

1878

            check_status(sub_res2);

2✔

1879

1✔

1882

});

2✔

1883

2✔

1884

4✔

1885

    // Test for issue #6839, where wait for download after committing a new subscription and then

4✔

1886

    // wait for the subscription complete notification was leading to a garbage reason value in the

4✔

1887

    // status provided to the subscription complete callback.

4✔

1888

    SECTION("Download during bad query") {

8✔

1889

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

1890

            // Wait for steady state before committing the new subscription

1✔

1892

1✔

1893

            auto subs = create_subscription(realm, "class_TopLevel", "non_queryable_field", [](auto q, auto c) {

2✔

1894

                return q.equal(c, "bar");

2✔

1895

});

2✔

1896

            // Wait for download is actually waiting for the subscription to be applied after it was committed

1✔

1898

            // After subscription is complete or fails during wait for download, this function completes

1✔

1899

            // without blocking

1✔

1900

            auto result = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

1901

            // Verify error occurred

1✔

1902

            check_status(result);

2✔

1903

});

2✔

1904

2✔

1905

4✔

1906

    // Add new sections before this

4✔

1907

    SECTION("teardown") {

8✔

1908

        harness->app()->sync_manager()->wait_for_sessions_to_terminate();

2✔

1909

        harness.reset();

2✔

1910

2✔

1911

8✔

1914

TEST_CASE("flx: geospatial", "[sync][flx][geospatial][baas]") {

6✔

1915

    static std::optional<FLXSyncTestHarness> harness;

6✔

1916

    if (!harness) {

6✔

1917

        Schema schema{

2✔

1918

            {"restaurant",

2✔

1919

2✔

1920

                 {"_id", PropertyType::Int, Property::IsPrimary{true}},

2✔

1921

                 {"queryable_str_field", PropertyType::String},

2✔

1922

                 {"location", PropertyType::Object | PropertyType::Nullable, "geoPointType"},

2✔

1923

                 {"array", PropertyType::Object | PropertyType::Array, "geoPointType"},

2✔

1924

}},

2✔

1925

            {"geoPointType",

2✔

1926

             ObjectSchema::ObjectType::Embedded,

2✔

1927

2✔

1928

                 {"type", PropertyType::String},

2✔

1929

                 {"coordinates", PropertyType::Double | PropertyType::Array},

2✔

1930

}},

2✔

1931

};

2✔

1932

        FLXSyncTestHarness::ServerSchema server_schema{schema, {"queryable_str_field", "location"}};

2✔

1933

        harness.emplace("flx_geospatial", server_schema);

2✔

1934

2✔

1935

3✔

1936

    auto create_subscription = [](SharedRealm realm, StringData table_name, StringData column_name, auto make_query) {

18✔

1937

        auto table = realm->read_group().get_table(table_name);

18✔

1938

        auto queryable_field = table->get_column_key(column_name);

18✔

1939

        auto new_query = realm->get_active_subscription_set().make_mutable_copy();

18✔

1940

        new_query.insert_or_assign(make_query(Query(table), queryable_field));

18✔

1941

        return new_query.commit();

18✔

1942

};

18✔

1943

3✔

1944

    SECTION("Server supports a basic geowithin FLX query") {

6✔

1945

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

1946

            const realm::AppSession& app_session = harness->session().app_session();

2✔

1947

            auto sync_service = app_session.admin_api.get_sync_service(app_session.server_app_id);

2✔

1948

1✔

1949

            AdminAPISession::ServiceConfig config =

2✔

1950

                app_session.admin_api.get_config(app_session.server_app_id, sync_service);

2✔

1951

            auto subs = create_subscription(realm, "class_restaurant", "location", [](Query q, ColKey c) {

2✔

1952

                GeoBox area{GeoPoint{0.2, 0.2}, GeoPoint{0.7, 0.7}};

2✔

1953

                Query query = q.get_table()->column<Link>(c).geo_within(area);

2✔

1954

                std::string ser = query.get_description();

2✔

1955

                return query;

2✔

1956

});

2✔

1957

            auto sub_res = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

1961

});

2✔

1962

2✔

1963

3✔

1964

    SECTION("geospatial query consistency: local/server/FLX") {

6✔

1965

        harness->do_with_new_user([&](std::shared_ptr<SyncUser> user) {

2✔

1966

            SyncTestFile config(user, harness->schema(), SyncConfig::FLXSyncEnabled{});

2✔

1967

            auto error_pf = util::make_promise_future<SyncError>();

2✔

1968

            config.sync_config->error_handler =

2✔

1969

                [promise = std::make_shared<util::Promise<SyncError>>(std::move(error_pf.promise))](

2✔

1970

                    std::shared_ptr<SyncSession>, SyncError error) {

2✔

1971

                    promise->emplace_value(std::move(error));

2✔

1972

};

2✔

1973

1✔

1974

            auto realm = Realm::get_shared_realm(config);

2✔

1975

1✔

1976

            auto subs = create_subscription(realm, "class_restaurant", "queryable_str_field", [](Query q, ColKey c) {

2✔

1977

                return q.equal(c, "synced");

2✔

1978

});

2✔

1979

            auto make_polygon_filter = [&](const GeoPolygon& polygon) -> bson::BsonDocument {

20✔

1980

                bson::BsonArray inner{};

20✔

1981

                REALM_ASSERT_3(polygon.points.size(), ==, 1);

20✔

1982

                for (auto& point : polygon.points[0]) {

94✔

1983

                    inner.push_back(bson::BsonArray{point.longitude, point.latitude});

94✔

1984

94✔

1985

                bson::BsonArray coords;

20✔

1986

                coords.push_back(inner);

20✔

1987

                bson::BsonDocument geo_bson{{{"type", "Polygon"}, {"coordinates", coords}}};

20✔

1988

                bson::BsonDocument filter{

20✔

1989

                    {"location", bson::BsonDocument{{"$geoWithin", bson::BsonDocument{{"$geometry", geo_bson}}}}}};

20✔

1990

                return filter;

20✔

1991

};

20✔

1992

            auto make_circle_filter = [&](const GeoCircle& circle) -> bson::BsonDocument {

6✔

1993

                bson::BsonArray coords{circle.center.longitude, circle.center.latitude};

6✔

1994

                bson::BsonArray inner;

6✔

1995

                inner.push_back(coords);

6✔

1996

                inner.push_back(circle.radius_radians);

6✔

1997

                bson::BsonDocument filter{

6✔

1998

                    {"location", bson::BsonDocument{{"$geoWithin", bson::BsonDocument{{"$centerSphere", inner}}}}}};

6✔

1999

                return filter;

6✔

2000

};

6✔

2001

            auto run_query_on_server = [&](const bson::BsonDocument& filter,

2✔

2002

                                           std::optional<std::string> expected_error = {}) -> size_t {

26✔

2003

                auto remote_client = harness->app()->current_user()->mongo_client("BackingDB");

26✔

2004

                auto db = remote_client.db(harness->session().app_session().config.mongo_dbname);

26✔

2005

                auto restaurant_collection = db["restaurant"];

26✔

2006

                bool processed = false;

26✔

2007

                constexpr int64_t limit = 1000;

26✔

2008

                size_t matches = 0;

26✔

2009

                restaurant_collection.count(filter, limit, [&](uint64_t count, util::Optional<AppError> error) {

26✔

2010

                    processed = true;

26✔

2011

                    if (error) {

26✔

2012

                        if (!expected_error) {

12✔

2016

                        else {

12✔

2017

                            std::string reason = std::string(error->reason());

12✔

2018

                            std::transform(reason.begin(), reason.end(), reason.begin(), toLowerAscii);

12✔

2019

                            std::transform(expected_error->begin(), expected_error->end(), expected_error->begin(),

12✔

2020

                                           toLowerAscii);

12✔

2021

                            auto pos = reason.find(*expected_error);

12✔

2022

                            if (pos == std::string::npos) {

12✔

2026

12✔

2027

12✔

2028

                    matches = size_t(count);

26✔

2029

});

26✔

2031

                return matches;

26✔

2032

};

26✔

2033

            auto sub_res = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

2✔

2037

1✔

2038

            CppContext c(realm);

2✔

2039

            int64_t pk = 0;

2✔

2040

            auto add_point = [&](GeoPoint p) {

16✔

2041

                Object::create(

16✔

2042

                    c, realm, "restaurant",

16✔

2043

                    std::any(AnyDict{

16✔

2044

                        {"_id", ++pk},

16✔

2045

                        {"queryable_str_field", "synced"s},

16✔

2046

                        {"location", AnyDict{{"type", "Point"s},

16✔

2047

                                             {"coordinates", std::vector<std::any>{p.longitude, p.latitude}}}}}));

16✔

2048

};

16✔

2049

            std::vector<GeoPoint> points = {

2✔

2050

                GeoPoint{-74.006, 40.712800000000001},            // New York city

2✔

2051

                GeoPoint{12.568300000000001, 55.676099999999998}, // Copenhagen

2✔

2052

                GeoPoint{12.082599999999999, 55.628},             // ragnarok, Roskilde

2✔

2053

                GeoPoint{-180.1, -90.1},                          // invalid

2✔

2054

                GeoPoint{0, 90},                                  // north pole

2✔

2055

                GeoPoint{-82.68193, 84.74653},                    // northern point that falls within a box later

2✔

2056

                GeoPoint{82.55243, 84.54981}, // another northern point, but on the other side of the pole

2✔

2057

                GeoPoint{2129, 89},           // invalid

2✔

2058

};

2✔

2059

            constexpr size_t invalids_to_be_compensated = 2; // 4, 8

2✔

2060

            realm->begin_transaction();

2✔

2061

            for (auto& point : points) {

16✔

2062

                add_point(point);

16✔

2063

16✔

2064

            realm->commit_transaction();

2✔

2065

            const auto& error = error_pf.future.get();

2✔

2069

            REQUIRE_THAT(error.compensating_writes_info[0].reason,

2✔

2070

                         Catch::Matchers::ContainsSubstring("in table \"restaurant\" will corrupt geojson data"));

2✔

2071

            REQUIRE_THAT(error.compensating_writes_info[1].reason,

2✔

2072

                         Catch::Matchers::ContainsSubstring("in table \"restaurant\" will corrupt geojson data"));

2✔

2073

1✔

2074

2✔

2075

                auto table = realm->read_group().get_table("class_restaurant");

2✔

2077

                Obj obj = table->get_object_with_primary_key(Mixed{1});

2✔

2079

                Geospatial geo = obj.get<Geospatial>("location");

2✔

2082

                GeoPoint point = geo.get<GeoPoint>();

2✔

2086

                ColKey location_col = table->get_column_key("location");

2✔

2087

                auto run_query_locally = [&table, &location_col](Geospatial bounds) -> size_t {

26✔

2088

                    Query query = table->column<Link>(location_col).geo_within(Geospatial(bounds));

26✔

2089

                    return query.find_all().size();

26✔

2090

};

26✔

2091

                auto run_query_as_flx = [&](Geospatial bounds) -> size_t {

14✔

2092

                    size_t num_objects = 0;

14✔

2093

                    harness->do_with_new_realm([&](SharedRealm realm) {

14✔

2094

                        auto subs =

14✔

2095

                            create_subscription(realm, "class_restaurant", "location", [&](Query q, ColKey c) {

14✔

2096

                                return q.get_table()->column<Link>(c).geo_within(Geospatial(bounds));

14✔

2097

});

14✔

2098

                        auto sub_res =

14✔

2099

                            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get_no_throw();

14✔

2102

                        realm->refresh();

14✔

2103

                        num_objects = realm->get_class("restaurant").num_objects();

14✔

2104

});

14✔

2105

                    return num_objects;

14✔

2106

};

14✔

2107

1✔

2108

                reset_utils::wait_for_num_objects_in_atlas(harness->app()->current_user(),

2✔

2109

                                                           harness->session().app_session(), "restaurant",

2✔

2110

                                                           points.size() - invalids_to_be_compensated);

2✔

2111

1✔

2112

2✔

2113

                    GeoPolygon bounds{

2✔

2114

                        {{GeoPoint{-80, 40.7128}, GeoPoint{20, 60}, GeoPoint{20, 20}, GeoPoint{-80, 40.7128}}}};

2✔

2115

                    size_t local_matches = run_query_locally(bounds);

2✔

2116

                    size_t server_results = run_query_on_server(make_polygon_filter(bounds));

2✔

2117

                    size_t flx_results = run_query_as_flx(bounds);

2✔

2120

2✔

2121

2✔

2122

                    GeoCircle circle{.5, GeoPoint{0, 90}};

2✔

2123

                    size_t local_matches = run_query_locally(circle);

2✔

2124

                    size_t server_results = run_query_on_server(make_circle_filter(circle));

2✔

2125

                    size_t flx_results = run_query_as_flx(circle);

2✔

2128

2✔

2129

                { // a ring with 3 points without a matching begin/end is an error

2✔

2130

                    GeoPolygon open_bounds{{{GeoPoint{-80, 40.7128}, GeoPoint{20, 60}, GeoPoint{20, 20}}}};

2✔

2131

                    CHECK_THROWS_WITH(run_query_locally(open_bounds),

2✔

2132

                                      "Invalid region in GEOWITHIN query for parameter 'GeoPolygon({[-80, 40.7128], "

2✔

2133

                                      "[20, 60], [20, 20]})': 'Ring is not closed, first vertex 'GeoPoint([-80, "

2✔

2134

                                      "40.7128])' does not equal last vertex 'GeoPoint([20, 20])''");

2✔

2135

                    run_query_on_server(make_polygon_filter(open_bounds), "(BadValue) Loop is not closed");

2✔

2136

2✔

2137

2✔

2138

                    GeoCircle circle = GeoCircle::from_kms(10, GeoPoint{-180.1, -90.1});

2✔

2139

                    CHECK_THROWS_WITH(run_query_locally(circle),

2✔

2140

                                      "Invalid region in GEOWITHIN query for parameter 'GeoCircle([-180.1, -90.1], "

2✔

2141

                                      "0.00156787)': 'Longitude/latitude is out of bounds, lng: -180.1 lat: -90.1'");

2✔

2142

                    run_query_on_server(make_circle_filter(circle), "(BadValue) longitude/latitude is out of bounds");

2✔

2143

2✔

2144

2✔

2145

                    GeoCircle circle = GeoCircle::from_kms(-1, GeoPoint{0, 0});

2✔

2146

                    CHECK_THROWS_WITH(run_query_locally(circle),

2✔

2147

                                      "Invalid region in GEOWITHIN query for parameter 'GeoCircle([0, 0], "

2✔

2148

                                      "-0.000156787)': 'The radius of a circle must be a non-negative number'");

2✔

2149

                    run_query_on_server(make_circle_filter(circle),

2✔

2150

                                        "(BadValue) radius must be a non-negative number");

2✔

2151

2✔

2152

2✔

2153

                    // This box is from Gershøj to CPH airport. It includes CPH and Ragnarok but not NYC.

1✔

2154

                    std::vector<Geospatial> valid_box_variations = {

2✔

2155

                        GeoBox{GeoPoint{11.97575, 55.71601},

2✔

2156

                               GeoPoint{12.64773, 55.61211}}, // Gershøj, CPH Airport (Top Left, Bottom Right)

2✔

2157

                        GeoBox{GeoPoint{12.64773, 55.61211},

2✔

2158

                               GeoPoint{11.97575, 55.71601}}, // CPH Airport, Gershøj (Bottom Right, Top Left)

2✔

2159

                        GeoBox{GeoPoint{12.64773, 55.71601},

2✔

2160

                               GeoPoint{11.97575, 55.61211}}, // Upper Right, Bottom Left

2✔

2161

                        GeoBox{GeoPoint{11.97575, 55.61211},

2✔

2162

                               GeoPoint{12.64773, 55.71601}}, // Bottom Left, Upper Right

2✔

2163

};

2✔

2164

                    constexpr size_t expected_results = 2;

2✔

2165

                    for (auto& geo : valid_box_variations) {

8✔

2166

                        size_t local_matches = run_query_locally(geo);

8✔

2167

                        size_t server_matches =

8✔

2168

                            run_query_on_server(make_polygon_filter(geo.get<GeoBox>().to_polygon()));

8✔

2169

                        size_t flx_matches = run_query_as_flx(geo);

8✔

2173

8✔

2174

                    std::vector<Geospatial> invalid_boxes = {

2✔

2175

                        GeoBox{GeoPoint{11.97575, 55.71601}, GeoPoint{11.97575, 55.71601}}, // same point twice

2✔

2176

                        GeoBox{GeoPoint{11.97575, 55.71601},

2✔

2177

                               GeoPoint{11.97575, 57.0}}, // two points on the same longitude

2✔

2178

                        GeoBox{GeoPoint{11.97575, 55.71601},

2✔

2179

                               GeoPoint{12, 55.71601}}, // two points on the same latitude

2✔

2180

};

2✔

2181

                    for (auto& geo : invalid_boxes) {

6✔

2182

                        REQUIRE_THROWS_CONTAINING(run_query_locally(geo),

6✔

2183

                                                  "Invalid region in GEOWITHIN query for parameter 'GeoPolygon");

6✔

2184

                        run_query_on_server(make_polygon_filter(geo.get<GeoBox>().to_polygon()),

6✔

2185

                                            "(BadValue) Loop must have at least 3 different vertices");

6✔

2186

6✔

2187

2✔

2188

                { // a box region that wraps the north pole. It contains the north pole point

2✔

2189

                    // and two others, one each on distinct sides of the globe.

1✔

2190

                    constexpr double lat = 82.83799;

2✔

2191

                    Geospatial north_pole_box =

2✔

2192

                        GeoPolygon{{{GeoPoint{-78.33951, lat}, GeoPoint{-90.33951, lat}, GeoPoint{90.33951, lat},

2✔

2193

                                     GeoPoint{78.33951, lat}, GeoPoint{-78.33951, lat}}}};

2✔

2194

                    constexpr size_t num_matching_points = 3;

2✔

2195

                    size_t local_matches = run_query_locally(north_pole_box);

2✔

2196

                    size_t server_matches =

2✔

2197

                        run_query_on_server(make_polygon_filter(north_pole_box.get<GeoPolygon>()));

2✔

2198

                    size_t flx_matches = run_query_as_flx(north_pole_box);

2✔

2202

2✔

2203

2✔

2204

});

2✔

2205

2✔

2206

3✔

2207

    // Add new sections before this

3✔

2208

    SECTION("teardown") {

6✔

2209

        harness->app()->sync_manager()->wait_for_sessions_to_terminate();

2✔

2210

        harness.reset();

2✔

2211

2✔

2212

6✔

2215

TEST_CASE("flx: interrupted bootstrap restarts/recovers on reconnect", "[sync][flx][bootstrap][baas]") {

2✔

2216

    FLXSyncTestHarness harness("flx_bootstrap_reconnect", {g_large_array_schema, {"queryable_int_field"}});

2✔

2217

1✔

2218

    std::vector<ObjectId> obj_ids_at_end = fill_large_array_schema(harness);

2✔

2219

    SyncTestFile interrupted_realm_config(harness.app()->current_user(), harness.schema(),

2✔

2220

                                          SyncConfig::FLXSyncEnabled{});

2✔

2221

1✔

2222

2✔

2223

        auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

2224

        Realm::Config config = interrupted_realm_config;

2✔

2225

        config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

2226

        auto shared_promise = std::make_shared<util::Promise<void>>(std::move(interrupted_promise));

2✔

2227

        config.sync_config->on_sync_client_event_hook =

2✔

2228

            [promise = std::move(shared_promise), seen_version_one = false](std::weak_ptr<SyncSession> weak_session,

2✔

2229

                                                                            const SyncClientHookData& data) mutable {

24✔

2230

                if (data.event != SyncClientHookEvent::DownloadMessageReceived) {

24✔

2231

                    return SyncClientHookAction::NoAction;

16✔

2232

16✔

2233

4✔

2234

                auto session = weak_session.lock();

8✔

2235

                if (!session) {

8✔

2238

4✔

2239

                // If we haven't seen at least one download message for query version 1, then do nothing yet.

4✔

2240

                if (data.query_version == 0 || (data.query_version == 1 && !std::exchange(seen_version_one, true))) {

8✔

2241

                    return SyncClientHookAction::NoAction;

6✔

2242

6✔

2243

1✔

2246

                auto latest_subs = session->get_flx_subscription_store()->get_latest();

2✔

2249

1✔

2250

                session->close();

2✔

2251

                promise->emplace_value();

2✔

2252

1✔

2253

                return SyncClientHookAction::TriggerReconnect;

2✔

2254

};

2✔

2255

1✔

2256

        auto realm = Realm::get_shared_realm(config);

2✔

2257

2✔

2258

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2259

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

2260

            mut_subs.insert_or_assign(Query(table));

2✔

2261

            mut_subs.commit();

2✔

2262

2✔

2263

1✔

2264

        interrupted.get();

2✔

2265

        realm->sync_session()->shutdown_and_wait();

2✔

2266

2✔

2267

1✔

2268

    _impl::RealmCoordinator::assert_no_open_realms();

2✔

2269

1✔

2270

2✔

2271

        DBOptions options;

2✔

2272

        options.encryption_key = test_util::crypt_key();

2✔

2273

        auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

2274

        auto sub_store = sync::SubscriptionStore::create(realm);

2✔

2275

        auto version_info = sub_store->get_version_info();

2✔

2278

        auto latest_subs = sub_store->get_latest();

2✔

2282

2✔

2283

1✔

2284

    auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

2285

    auto table = realm->read_group().get_table("class_TopLevel");

2✔

2286

    realm->get_latest_subscription_set().get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

2287

    wait_for_advance(*realm);

2✔

2289

    for (auto& id : obj_ids_at_end) {

10✔

2291

10✔

2292

1✔

2293

    auto active_subs = realm->get_active_subscription_set();

2✔

2294

    auto latest_subs = realm->get_latest_subscription_set();

2✔

2297

2✔

2299

TEST_CASE("flx: dev mode uploads schema before query change", "[sync][flx][query][baas]") {

2✔

2300

    FLXSyncTestHarness::ServerSchema server_schema;

2✔

2301

    auto default_schema = FLXSyncTestHarness::default_server_schema();

2✔

2302

    server_schema.queryable_fields = default_schema.queryable_fields;

2✔

2303

    server_schema.dev_mode_enabled = true;

2✔

2304

    server_schema.schema = Schema{};

2✔

2305

1✔

2306

    FLXSyncTestHarness harness("flx_dev_mode", server_schema);

2✔

2307

    auto foo_obj_id = ObjectId::gen();

2✔

2308

    auto bar_obj_id = ObjectId::gen();

2✔

2309

    harness.do_with_new_realm(

2✔

2310

        [&](SharedRealm realm) {

2✔

2311

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

2312

            // auto queryable_str_field = table->get_column_key("queryable_str_field");

1✔

2313

            // auto queryable_int_field = table->get_column_key("queryable_int_field");

1✔

2314

            auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2315

            new_query.insert_or_assign(Query(table));

2✔

2316

            new_query.commit();

2✔

2317

1✔

2318

            CppContext c(realm);

2✔

2319

            realm->begin_transaction();

2✔

2320

            Object::create(c, realm, "TopLevel",

2✔

2321

                           std::any(AnyDict{{"_id", foo_obj_id},

2✔

2322

                                            {"queryable_str_field", "foo"s},

2✔

2323

                                            {"queryable_int_field", static_cast<int64_t>(5)},

2✔

2324

                                            {"non_queryable_field", "non queryable 1"s}}));

2✔

2325

            Object::create(c, realm, "TopLevel",

2✔

2326

                           std::any(AnyDict{{"_id", bar_obj_id},

2✔

2327

                                            {"queryable_str_field", "bar"s},

2✔

2328

                                            {"queryable_int_field", static_cast<int64_t>(10)},

2✔

2329

                                            {"non_queryable_field", "non queryable 2"s}}));

2✔

2330

            realm->commit_transaction();

2✔

2331

1✔

2332

            wait_for_upload(*realm);

2✔

2333

},

2✔

2334

        default_schema.schema);

2✔

2335

1✔

2336

    harness.do_with_new_realm(

2✔

2337

        [&](SharedRealm realm) {

2✔

2338

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

2339

            auto queryable_int_field = table->get_column_key("queryable_int_field");

2✔

2340

            auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2341

            new_query.insert_or_assign(Query(table).greater_equal(queryable_int_field, int64_t(5)));

2✔

2342

            auto subs = new_query.commit();

2✔

2343

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

2344

            wait_for_download(*realm);

2✔

2345

            Results results(realm, table);

2✔

2346

1✔

2347

            realm->refresh();

2✔

2351

},

2✔

2352

        default_schema.schema);

2✔

2353

2✔

2356

TEST_CASE("flx: change-of-query history divergence", "[sync][flx][query][baas]") {

2✔

2357

    FLXSyncTestHarness harness("flx_coq_divergence");

2✔

2358

1✔

2359

    // first we create an object on the server and upload it.

1✔

2360

    auto foo_obj_id = ObjectId::gen();

2✔

2361

    harness.load_initial_data([&](SharedRealm realm) {

2✔

2362

        CppContext c(realm);

2✔

2363

        Object::create(c, realm, "TopLevel",

2✔

2364

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

2365

                                        {"queryable_str_field", "foo"s},

2✔

2366

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

2367

                                        {"non_queryable_field", "created as initial data seed"s}}));

2✔

2368

});

2✔

2369

1✔

2370

    // Now create another realm and wait for it to be fully synchronized with bootstrap version zero. i.e.

1✔

2371

    // our progress counters should be past the history entry containing the object created above.

1✔

2372

    auto test_file_config = harness.make_test_file();

2✔

2373

    auto realm = Realm::get_shared_realm(test_file_config);

2✔

2374

    auto table = realm->read_group().get_table("class_TopLevel");

2✔

2375

    auto queryable_str_field = table->get_column_key("queryable_str_field");

2✔

2376

1✔

2377

    realm->get_latest_subscription_set().get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

2378

    wait_for_upload(*realm);

2✔

2379

    wait_for_download(*realm);

2✔

2380

1✔

2381

    // Now disconnect the sync session

1✔

2382

    realm->sync_session()->pause();

2✔

2383

1✔

2384

    // And move the "foo" object created above into view and create a different diverging copy of it locally.

1✔

2385

    auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2386

    mut_subs.insert_or_assign(Query(table).equal(queryable_str_field, "foo"));

2✔

2387

    auto subs = mut_subs.commit();

2✔

2388

1✔

2389

    realm->begin_transaction();

2✔

2390

    CppContext c(realm);

2✔

2391

    Object::create(c, realm, "TopLevel",

2✔

2392

                   std::any(AnyDict{{"_id", foo_obj_id},

2✔

2393

                                    {"queryable_str_field", "foo"s},

2✔

2394

                                    {"queryable_int_field", static_cast<int64_t>(10)},

2✔

2395

                                    {"non_queryable_field", "created locally"s}}));

2✔

2396

    realm->commit_transaction();

2✔

2397

1✔

2398

    // Reconnect the sync session and wait for the subscription that moved "foo" into view to be fully synchronized.

1✔

2399

    realm->sync_session()->resume();

2✔

2400

    wait_for_upload(*realm);

2✔

2401

    wait_for_download(*realm);

2✔

2402

    subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

2403

1✔

2404

    wait_for_advance(*realm);

2✔

2405

1✔

2406

    // The bootstrap should have erase/re-created our object and we should have the version from the server

1✔

2407

    // locally.

1✔

2408

    auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{foo_obj_id});

2✔

2411

1✔

2412

    // Likewise, if we create a new realm and download all the objects, we should see the initial server version

1✔

2413

    // in the new realm rather than the "created locally" one.

1✔

2414

    harness.load_initial_data([&](SharedRealm realm) {

2✔

2415

        CppContext c(realm);

2✔

2416

1✔

2417

        auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{foo_obj_id});

2✔

2420

});

2✔

2421

2✔

2423

TEST_CASE("flx: writes work offline", "[sync][flx][baas]") {

2✔

2424

    FLXSyncTestHarness harness("flx_offline_writes");

2✔

2425

1✔

2426

    harness.do_with_new_realm([&](SharedRealm realm) {

2✔

2427

        auto sync_session = realm->sync_session();

2✔

2428

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

2429

        auto queryable_str_field = table->get_column_key("queryable_str_field");

2✔

2430

        auto queryable_int_field = table->get_column_key("queryable_int_field");

2✔

2431

        auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2432

        new_query.insert_or_assign(Query(table));

2✔

2433

        new_query.commit();

2✔

2434

1✔

2435

        auto foo_obj_id = ObjectId::gen();

2✔

2436

        auto bar_obj_id = ObjectId::gen();

2✔

2437

1✔

2438

        CppContext c(realm);

2✔

2439

        realm->begin_transaction();

2✔

2440

        Object::create(c, realm, "TopLevel",

2✔

2441

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

2442

                                        {"queryable_str_field", "foo"s},

2✔

2443

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

2444

                                        {"non_queryable_field", "non queryable 1"s}}));

2✔

2445

        Object::create(c, realm, "TopLevel",

2✔

2446

                       std::any(AnyDict{{"_id", bar_obj_id},

2✔

2447

                                        {"queryable_str_field", "bar"s},

2✔

2448

                                        {"queryable_int_field", static_cast<int64_t>(10)},

2✔

2449

                                        {"non_queryable_field", "non queryable 2"s}}));

2✔

2450

        realm->commit_transaction();

2✔

2451

1✔

2452

        wait_for_upload(*realm);

2✔

2453

        wait_for_download(*realm);

2✔

2454

        sync_session->pause();

2✔

2455

1✔

2456

        // Make it so the subscriptions only match the "foo" object

1✔

2457

2✔

2458

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2459

            mut_subs.clear();

2✔

2460

            mut_subs.insert_or_assign(Query(table).equal(queryable_str_field, "foo"));

2✔

2461

            mut_subs.commit();

2✔

2462

2✔

2463

1✔

2464

        // Make foo so that it will match the next subscription update. This checks whether you can do

1✔

2465

        // multiple subscription set updates offline and that the last one eventually takes effect when

1✔

2466

        // you come back online and fully synchronize.

1✔

2467

2✔

2468

            Results results(realm, table);

2✔

2469

            realm->begin_transaction();

2✔

2470

            auto foo_obj = table->get_object_with_primary_key(Mixed{foo_obj_id});

2✔

2471

            foo_obj.set<int64_t>(queryable_int_field, 15);

2✔

2472

            realm->commit_transaction();

2✔

2473

2✔

2474

1✔

2475

        // Update our subscriptions so that both foo/bar will be included

1✔

2476

2✔

2477

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2478

            mut_subs.clear();

2✔

2479

            mut_subs.insert_or_assign(Query(table).greater_equal(queryable_int_field, static_cast<int64_t>(10)));

2✔

2480

            mut_subs.commit();

2✔

2481

2✔

2482

1✔

2483

        // Make foo out of view for the current subscription.

1✔

2484

2✔

2485

            Results results(realm, table);

2✔

2486

            realm->begin_transaction();

2✔

2487

            auto foo_obj = table->get_object_with_primary_key(Mixed{foo_obj_id});

2✔

2488

            foo_obj.set<int64_t>(queryable_int_field, 0);

2✔

2489

            realm->commit_transaction();

2✔

2490

2✔

2491

1✔

2492

        sync_session->resume();

2✔

2493

        wait_for_upload(*realm);

2✔

2494

        wait_for_download(*realm);

2✔

2495

1✔

2496

        realm->refresh();

2✔

2497

        Results results(realm, table);

2✔

2500

});

2✔

2501

2✔

2503

TEST_CASE("flx: writes work without waiting for sync", "[sync][flx][baas]") {

2✔

2504

    FLXSyncTestHarness harness("flx_offline_writes");

2✔

2505

1✔

2506

    harness.do_with_new_realm([&](SharedRealm realm) {

2✔

2507

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

2508

        auto queryable_str_field = table->get_column_key("queryable_str_field");

2✔

2509

        auto queryable_int_field = table->get_column_key("queryable_int_field");

2✔

2510

        auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2511

        new_query.insert_or_assign(Query(table));

2✔

2512

        new_query.commit();

2✔

2513

1✔

2514

        auto foo_obj_id = ObjectId::gen();

2✔

2515

        auto bar_obj_id = ObjectId::gen();

2✔

2516

1✔

2517

        CppContext c(realm);

2✔

2518

        realm->begin_transaction();

2✔

2519

        Object::create(c, realm, "TopLevel",

2✔

2520

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

2521

                                        {"queryable_str_field", "foo"s},

2✔

2522

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

2523

                                        {"non_queryable_field", "non queryable 1"s}}));

2✔

2524

        Object::create(c, realm, "TopLevel",

2✔

2525

                       std::any(AnyDict{{"_id", bar_obj_id},

2✔

2526

                                        {"queryable_str_field", "bar"s},

2✔

2527

                                        {"queryable_int_field", static_cast<int64_t>(10)},

2✔

2528

                                        {"non_queryable_field", "non queryable 2"s}}));

2✔

2529

        realm->commit_transaction();

2✔

2530

1✔

2531

        wait_for_upload(*realm);

2✔

2532

1✔

2533

        // Make it so the subscriptions only match the "foo" object

1✔

2534

2✔

2535

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2536

            mut_subs.clear();

2✔

2537

            mut_subs.insert_or_assign(Query(table).equal(queryable_str_field, "foo"));

2✔

2538

            mut_subs.commit();

2✔

2539

2✔

2540

1✔

2541

        // Make foo so that it will match the next subscription update. This checks whether you can do

1✔

2542

        // multiple subscription set updates without waiting and that the last one eventually takes effect when

1✔

2543

        // you fully synchronize.

1✔

2544

2✔

2545

            Results results(realm, table);

2✔

2546

            realm->begin_transaction();

2✔

2547

            auto foo_obj = table->get_object_with_primary_key(Mixed{foo_obj_id});

2✔

2548

            foo_obj.set<int64_t>(queryable_int_field, 15);

2✔

2549

            realm->commit_transaction();

2✔

2550

2✔

2551

1✔

2552

        // Update our subscriptions so that both foo/bar will be included

1✔

2553

2✔

2554

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2555

            mut_subs.clear();

2✔

2556

            mut_subs.insert_or_assign(Query(table).greater_equal(queryable_int_field, static_cast<int64_t>(10)));

2✔

2557

            mut_subs.commit();

2✔

2558

2✔

2559

1✔

2560

        // Make foo out-of-view for the current subscription.

1✔

2561

2✔

2562

            Results results(realm, table);

2✔

2563

            realm->begin_transaction();

2✔

2564

            auto foo_obj = table->get_object_with_primary_key(Mixed{foo_obj_id});

2✔

2565

            foo_obj.set<int64_t>(queryable_int_field, 0);

2✔

2566

            realm->commit_transaction();

2✔

2567

2✔

2568

1✔

2569

        wait_for_upload(*realm);

2✔

2570

        wait_for_download(*realm);

2✔

2571

1✔

2572

        realm->refresh();

2✔

2573

        Results results(realm, table);

2✔

2575

        Obj obj = results.get(0);

2✔

2578

});

2✔

2579

2✔

2581

TEST_CASE("flx: verify websocket protocol number and prefixes", "[sync][protocol]") {

2✔

2582

    // Update the expected value whenever the protocol version is updated - this ensures

1✔

2583

    // that the current protocol version does not change unexpectedly.

1✔

2584

    REQUIRE(12 == sync::get_current_protocol_version());

2✔

2585

    // This was updated in Protocol V8 to use '#' instead of '/' to support the Web SDK

1✔

2586

    REQUIRE("com.mongodb.realm-sync#" == sync::get_pbs_websocket_protocol_prefix());

2✔

2587

    REQUIRE("com.mongodb.realm-query-sync#" == sync::get_flx_websocket_protocol_prefix());

2✔

2588

2✔

2592

TEST_CASE("flx: subscriptions persist after closing/reopening", "[sync][flx][baas]") {

2✔

2593

    FLXSyncTestHarness harness("flx_bad_query");

2✔

2594

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

2595

1✔

2596

2✔

2597

        auto orig_realm = Realm::get_shared_realm(config);

2✔

2598

        auto mut_subs = orig_realm->get_latest_subscription_set().make_mutable_copy();

2✔

2599

        mut_subs.insert_or_assign(Query(orig_realm->read_group().get_table("class_TopLevel")));

2✔

2600

        mut_subs.commit();

2✔

2601

        orig_realm->close();

2✔

2602

2✔

2603

1✔

2604

2✔

2605

        auto new_realm = Realm::get_shared_realm(config);

2✔

2606

        auto latest_subs = new_realm->get_latest_subscription_set();

2✔

2608

        latest_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

2609

2✔

2610

2✔

2613

TEST_CASE("flx: no subscription store created for PBS app", "[sync][flx][baas]") {

2✔

2614

    auto server_app_config = minimal_app_config("flx_connect_as_pbs", g_minimal_schema);

2✔

2615

    TestAppSession session(create_app(server_app_config));

2✔

2616

    SyncTestFile config(session.app()->current_user(), bson::Bson{}, g_minimal_schema);

2✔

2617

1✔

2618

    auto realm = Realm::get_shared_realm(config);

2✔

2621

1✔

2623

1✔

2624

    CHECK_THROWS_AS(realm->get_active_subscription_set(), IllegalOperation);

2✔

2625

    CHECK_THROWS_AS(realm->get_latest_subscription_set(), IllegalOperation);

2✔

2626

2✔

2628

TEST_CASE("flx: connect to FLX as PBS returns an error", "[sync][flx][baas]") {

2✔

2629

    FLXSyncTestHarness harness("connect_to_flx_as_pbs");

2✔

2630

    SyncTestFile config(harness.app()->current_user(), bson::Bson{}, harness.schema());

2✔

2631

    std::mutex sync_error_mutex;

2✔

2632

    util::Optional<SyncError> sync_error;

2✔

2633

    config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) mutable {

2✔

2634

        std::lock_guard<std::mutex> lk(sync_error_mutex);

2✔

2635

        sync_error = std::move(error);

2✔

2636

};

2✔

2637

    auto realm = Realm::get_shared_realm(config);

2✔

2638

    timed_wait_for([&] {

4,987✔

2639

        std::lock_guard<std::mutex> lk(sync_error_mutex);

4,987✔

2640

        return static_cast<bool>(sync_error);

4,987✔

2641

});

4,987✔

2642

1✔

2645

2✔

2647

TEST_CASE("flx: connect to FLX with partition value returns an error", "[sync][flx][protocol][baas]") {

2✔

2648

    FLXSyncTestHarness harness("connect_to_flx_as_pbs");

2✔

2649

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

2650

    config.sync_config->partition_value = "\"foobar\"";

2✔

2651

1✔

2652

    REQUIRE_EXCEPTION(Realm::get_shared_realm(config), IllegalCombination,

2✔

2653

                      "Cannot specify a partition value when flexible sync is enabled");

2✔

2654

2✔

2656

TEST_CASE("flx: connect to PBS as FLX returns an error", "[sync][flx][protocol][baas]") {

2✔

2657

    auto server_app_config = minimal_app_config("flx_connect_as_pbs", g_minimal_schema);

2✔

2658

    TestAppSession session(create_app(server_app_config));

2✔

2659

    auto app = session.app();

2✔

2660

    auto user = app->current_user();

2✔

2661

1✔

2662

    SyncTestFile config(user, g_minimal_schema, SyncConfig::FLXSyncEnabled{});

2✔

2663

1✔

2664

    std::mutex sync_error_mutex;

2✔

2665

    util::Optional<SyncError> sync_error;

2✔

2666

    config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) mutable {

2✔

2667

        std::lock_guard lk(sync_error_mutex);

2✔

2668

        sync_error = std::move(error);

2✔

2669

};

2✔

2670

    auto realm = Realm::get_shared_realm(config);

2✔

2671

    timed_wait_for([&] {

19,773✔

2672

        std::lock_guard lk(sync_error_mutex);

19,773✔

2673

        return static_cast<bool>(sync_error);

19,773✔

2674

});

19,773✔

2675

1✔

2678

2✔

2680

TEST_CASE("flx: commit subscription while refreshing the access token", "[sync][flx][token][baas]") {

2✔

2681

    auto transport = std::make_shared<HookedTransport<>>();

2✔

2682

    FLXSyncTestHarness harness("flx_wait_access_token2", FLXSyncTestHarness::default_server_schema(), transport);

2✔

2683

    auto app = harness.app();

2✔

2684

    std::shared_ptr<User> user = app->current_user();

2✔

2687

    // Set a bad access token, with an expired time. This will trigger a refresh initiated by the client.

1✔

2688

    std::chrono::system_clock::time_point now = std::chrono::system_clock::now();

2✔

2689

    using namespace std::chrono_literals;

2✔

2690

    auto expires = std::chrono::system_clock::to_time_t(now - 30s);

2✔

2691

    user->update_data_for_testing([&](UserData& data) {

2✔

2692

        data.access_token = RealmJWT(encode_fake_jwt("fake_access_token", expires));

2✔

2693

});

2✔

2695

1✔

2696

    bool seen_waiting_for_access_token = false;

2✔

2697

    // Commit a subcription set while there is no sync session.

1✔

2698

    // A session is created when the access token is refreshed.

1✔

2699

    transport->request_hook = [&](const Request&) {

2✔

2700

        auto user = app->current_user();

2✔

2702

        for (auto& session : app->sync_manager()->get_all_sessions_for(*user)) {

2✔

2703

            if (session->state() == SyncSession::State::WaitingForAccessToken) {

2✔

2705

                seen_waiting_for_access_token = true;

2✔

2706

1✔

2707

                auto store = session->get_flx_subscription_store();

2✔

2709

                auto mut_subs = store->get_latest().make_mutable_copy();

2✔

2710

                mut_subs.commit();

2✔

2711

2✔

2712

2✔

2713

        return std::nullopt;

2✔

2714

};

2✔

2715

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

2716

    // This triggers the token refresh.

1✔

2717

    auto r = Realm::get_shared_realm(config);

2✔

2719

2✔

2721

TEST_CASE("flx: bootstrap batching prevents orphan documents", "[sync][flx][bootstrap][baas]") {

8✔

2722

    struct NovelException : public std::exception {

8✔

2723

        const char* what() const noexcept override

8✔

2724

5✔

2725

            return "Oh no, a really weird exception happened!";

2✔

2726

2✔

2727

};

8✔

2728

4✔

2729

    FLXSyncTestHarness harness("flx_bootstrap_batching", {g_large_array_schema, {"queryable_int_field"}});

8✔

2730

4✔

2731

    std::vector<ObjectId> obj_ids_at_end = fill_large_array_schema(harness);

8✔

2732

    SyncTestFile interrupted_realm_config(harness.app()->current_user(), harness.schema(),

8✔

2733

                                          SyncConfig::FLXSyncEnabled{});

8✔

2734

4✔

2735

    auto check_interrupted_state = [&](const DBRef& realm) {

8✔

2736

        auto tr = realm->start_read();

8✔

2737

        auto top_level = tr->get_table("class_TopLevel");

8✔

2740

4✔

2741

        auto sub_store = sync::SubscriptionStore::create(realm);

8✔

2742

        auto version_info = sub_store->get_version_info();

8✔

2745

        auto latest_subs = sub_store->get_latest();

8✔

2749

};

8✔

2750

4✔

2751

    auto mutate_realm = [&] {

6✔

2752

        harness.load_initial_data([&](SharedRealm realm) {

4✔

2753

            auto table = realm->read_group().get_table("class_TopLevel");

4✔

2754

            Results res(realm, Query(table).greater(table->get_column_key("queryable_int_field"), int64_t(10)));

4✔

2756

            res.clear();

4✔

2757

});

4✔

2758

};

4✔

2759

4✔

2760

    SECTION("unknown exception occurs during bootstrap application on session startup") {

8✔

2761

2✔

2762

            auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

2763

            Realm::Config config = interrupted_realm_config;

2✔

2764

            config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

2765

            auto shared_promise = std::make_shared<util::Promise<void>>(std::move(interrupted_promise));

2✔

2766

            config.sync_config->on_sync_client_event_hook =

2✔

2767

                [promise = std::move(shared_promise)](std::weak_ptr<SyncSession> weak_session,

2✔

2768

                                                      const SyncClientHookData& data) mutable {

42✔

2769

                    if (data.event != SyncClientHookEvent::BootstrapMessageProcessed) {

42✔

2770

                        return SyncClientHookAction::NoAction;

28✔

2771

28✔

2772

                    auto session = weak_session.lock();

14✔

2773

                    if (!session) {

14✔

2776

7✔

2777

                    if (data.query_version == 1 && data.batch_state == sync::DownloadBatchState::LastInBatch) {

14✔

2778

                        session->close();

2✔

2779

                        promise->emplace_value();

2✔

2780

                        return SyncClientHookAction::EarlyReturn;

2✔

2781

2✔

2782

                    return SyncClientHookAction::NoAction;

12✔

2783

};

12✔

2784

            auto realm = Realm::get_shared_realm(config);

2✔

2785

2✔

2786

                auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2787

                auto table = realm->read_group().get_table("class_TopLevel");

2✔

2788

                mut_subs.insert_or_assign(Query(table));

2✔

2789

                mut_subs.commit();

2✔

2790

2✔

2791

1✔

2792

            interrupted.get();

2✔

2793

            realm->sync_session()->shutdown_and_wait();

2✔

2794

            realm->close();

2✔

2795

2✔

2796

1✔

2797

        _impl::RealmCoordinator::assert_no_open_realms();

2✔

2798

1✔

2799

        // Open up the realm without the sync client attached and verify that the realm got interrupted in the state

1✔

2800

        // we expected it to be in.

1✔

2801

2✔

2802

            DBOptions options;

2✔

2803

            options.encryption_key = test_util::crypt_key();

2✔

2804

            auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

2805

            auto logger = util::Logger::get_default_logger();

2✔

2806

            sync::PendingBootstrapStore bootstrap_store(realm, *logger);

2✔

2808

            auto pending_batch = bootstrap_store.peek_pending(1024 * 1024 * 16);

2✔

2811

1✔

2812

            check_interrupted_state(realm);

2✔

2813

2✔

2814

1✔

2815

        auto error_pf = util::make_promise_future<SyncError>();

2✔

2816

        interrupted_realm_config.sync_config->error_handler =

2✔

2817

            [promise = std::make_shared<util::Promise<SyncError>>(std::move(error_pf.promise))](

2✔

2818

                std::shared_ptr<SyncSession>, SyncError error) {

2✔

2819

                promise->emplace_value(std::move(error));

2✔

2820

};

2✔

2821

1✔

2822

        interrupted_realm_config.sync_config->on_sync_client_event_hook =

2✔

2823

            [&, download_message_received = false](std::weak_ptr<SyncSession>,

2✔

2824

                                                   const SyncClientHookData& data) mutable {

6✔

2825

                if (data.event == SyncClientHookEvent::DownloadMessageReceived) {

6✔

2828

                if (data.event != SyncClientHookEvent::BootstrapBatchAboutToProcess) {

6✔

2829

                    return SyncClientHookAction::NoAction;

4✔

2830

4✔

2831

1✔

2833

                throw NovelException{};

2✔

2834

                return SyncClientHookAction::NoAction;

1✔

2835

};

2✔

2836

1✔

2837

        auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

2838

        const auto& error = error_pf.future.get();

2✔

2842

        REQUIRE_THAT(error.status.reason(),

2✔

2843

                     Catch::Matchers::ContainsSubstring("Oh no, a really weird exception happened!"));

2✔

2844

2✔

2845

4✔

2846

    SECTION("exception occurs during bootstrap application") {

8✔

2847

        Status error_status(ErrorCodes::OutOfMemory, "no more memory!");

2✔

2848

2✔

2849

            auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

2850

            Realm::Config config = interrupted_realm_config;

2✔

2851

            config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

2852

            config.sync_config->on_sync_client_event_hook = [&](std::weak_ptr<SyncSession> weak_session,

2✔

2853

                                                                const SyncClientHookData& data) mutable {

44✔

2854

                if (data.event != SyncClientHookEvent::BootstrapBatchAboutToProcess) {

44✔

2855

                    return SyncClientHookAction::NoAction;

40✔

2856

40✔

2857

                auto session = weak_session.lock();

4✔

2858

                if (!session) {

4✔

2861

2✔

2862

                if (data.query_version == 1 && data.batch_state == sync::DownloadBatchState::MoreToCome) {

4✔

2863

                    throw sync::IntegrationException(error_status);

2✔

2864

2✔

2865

                return SyncClientHookAction::NoAction;

2✔

2866

};

2✔

2867

            auto error_pf = util::make_promise_future<SyncError>();

2✔

2868

            config.sync_config->error_handler =

2✔

2869

                [promise = std::make_shared<util::Promise<SyncError>>(std::move(error_pf.promise))](

2✔

2870

                    std::shared_ptr<SyncSession>, SyncError error) {

2✔

2871

                    promise->emplace_value(std::move(error));

2✔

2872

};

2✔

2873

1✔

2874

1✔

2875

            auto realm = Realm::get_shared_realm(config);

2✔

2876

2✔

2877

                auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2878

                auto table = realm->read_group().get_table("class_TopLevel");

2✔

2879

                mut_subs.insert_or_assign(Query(table));

2✔

2880

                mut_subs.commit();

2✔

2881

2✔

2882

1✔

2883

            auto error = error_pf.future.get();

2✔

2886

            realm->sync_session()->shutdown_and_wait();

2✔

2887

            realm->close();

2✔

2888

2✔

2889

1✔

2890

        _impl::RealmCoordinator::assert_no_open_realms();

2✔

2891

1✔

2892

        // Open up the realm without the sync client attached and verify that the realm got interrupted in the state

1✔

2893

        // we expected it to be in.

1✔

2894

2✔

2895

            DBOptions options;

2✔

2896

            options.encryption_key = test_util::crypt_key();

2✔

2897

            auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

2898

            util::StderrLogger logger;

2✔

2899

            sync::PendingBootstrapStore bootstrap_store(realm, logger);

2✔

2901

            auto pending_batch = bootstrap_store.peek_pending(1024 * 1024 * 16);

2✔

2904

1✔

2905

            check_interrupted_state(realm);

2✔

2906

2✔

2907

1✔

2908

        auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

2909

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

2910

        realm->get_latest_subscription_set()

2✔

2911

            .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

2912

            .get();

2✔

2913

1✔

2914

        wait_for_advance(*realm);

2✔

2915

1✔

2917

        for (auto& id : obj_ids_at_end) {

10✔

2919

10✔

2920

2✔

2921

4✔

2922

    SECTION("interrupted before final bootstrap message") {

8✔

2923

2✔

2924

            auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

2925

            Realm::Config config = interrupted_realm_config;

2✔

2926

            config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

2927

            auto shared_promise = std::make_shared<util::Promise<void>>(std::move(interrupted_promise));

2✔

2928

            config.sync_config->on_sync_client_event_hook =

2✔

2929

                [promise = std::move(shared_promise)](std::weak_ptr<SyncSession> weak_session,

2✔

2930

                                                      const SyncClientHookData& data) mutable {

22✔

2931

                    if (data.event != SyncClientHookEvent::BootstrapMessageProcessed) {

22✔

2932

                        return SyncClientHookAction::NoAction;

18✔

2933

18✔

2934

                    auto session = weak_session.lock();

4✔

2935

                    if (!session) {

4✔

2938

2✔

2939

                    if (data.query_version == 1 && data.batch_state == sync::DownloadBatchState::MoreToCome) {

4✔

2940

                        session->force_close();

2✔

2941

                        promise->emplace_value();

2✔

2942

                        return SyncClientHookAction::TriggerReconnect;

2✔

2943

2✔

2944

                    return SyncClientHookAction::NoAction;

2✔

2945

};

2✔

2946

            auto realm = Realm::get_shared_realm(config);

2✔

2947

2✔

2948

                auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

2949

                auto table = realm->read_group().get_table("class_TopLevel");

2✔

2950

                mut_subs.insert_or_assign(Query(table));

2✔

2951

                mut_subs.commit();

2✔

2952

2✔

2953

1✔

2954

            interrupted.get();

2✔

2955

            realm->sync_session()->shutdown_and_wait();

2✔

2956

            realm->close();

2✔

2957

2✔

2958

1✔

2959

        _impl::RealmCoordinator::assert_no_open_realms();

2✔

2960

1✔

2961

        // Open up the realm without the sync client attached and verify that the realm got interrupted in the state

1✔

2962

        // we expected it to be in.

1✔

2963

2✔

2964

            DBOptions options;

2✔

2965

            options.encryption_key = test_util::crypt_key();

2✔

2966

            auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

2967

            auto logger = util::Logger::get_default_logger();

2✔

2968

            sync::PendingBootstrapStore bootstrap_store(realm, *logger);

2✔

2970

            auto pending_batch = bootstrap_store.peek_pending(1024 * 1024 * 16);

2✔

2975

1✔

2976

            check_interrupted_state(realm);

2✔

2977

2✔

2978

1✔

2979

        // Now we'll open a different realm and make some changes that would leave orphan objects on the client

1✔

2980

        // if the bootstrap batches weren't being cached until lastInBatch were true.

1✔

2981

        mutate_realm();

2✔

2982

1✔

2983

        // Finally re-open the realm whose bootstrap we interrupted and just wait for it to finish downloading.

1✔

2984

        auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

2985

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

2986

        realm->get_latest_subscription_set()

2✔

2987

            .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

2988

            .get();

2✔

2989

1✔

2990

        wait_for_advance(*realm);

2✔

2991

        auto expected_obj_ids = util::Span<ObjectId>(obj_ids_at_end).sub_span(0, 3);

2✔

2992

1✔

2994

        for (auto& id : expected_obj_ids) {

6✔

2996

6✔

2997

2✔

2998

4✔

2999

    SECTION("interrupted after final bootstrap message before processing") {

8✔

3000

2✔

3001

            auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

3002

            Realm::Config config = interrupted_realm_config;

2✔

3003

            config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

3004

            auto shared_promise = std::make_shared<util::Promise<void>>(std::move(interrupted_promise));

2✔

3005

            config.sync_config->on_sync_client_event_hook =

2✔

3006

                [promise = std::move(shared_promise)](std::weak_ptr<SyncSession> weak_session,

2✔

3007

                                                      const SyncClientHookData& data) mutable {

42✔

3008

                    if (data.event != SyncClientHookEvent::BootstrapMessageProcessed) {

42✔

3009

                        return SyncClientHookAction::NoAction;

28✔

3010

28✔

3011

                    auto session = weak_session.lock();

14✔

3012

                    if (!session) {

14✔

3015

7✔

3016

                    if (data.query_version == 1 && data.batch_state == sync::DownloadBatchState::LastInBatch) {

14✔

3017

                        session->force_close();

2✔

3018

                        promise->emplace_value();

2✔

3019

                        return SyncClientHookAction::TriggerReconnect;

2✔

3020

2✔

3021

                    return SyncClientHookAction::NoAction;

12✔

3022

};

12✔

3023

            auto realm = Realm::get_shared_realm(config);

2✔

3024

2✔

3025

                auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

3026

                auto table = realm->read_group().get_table("class_TopLevel");

2✔

3027

                mut_subs.insert_or_assign(Query(table));

2✔

3028

                mut_subs.commit();

2✔

3029

2✔

3030

1✔

3031

            interrupted.get();

2✔

3032

            realm->sync_session()->shutdown_and_wait();

2✔

3033

            realm->close();

2✔

3034

2✔

3035

1✔

3036

        _impl::RealmCoordinator::assert_no_open_realms();

2✔

3037

1✔

3038

        // Open up the realm without the sync client attached and verify that the realm got interrupted in the state

1✔

3039

        // we expected it to be in.

1✔

3040

2✔

3041

            DBOptions options;

2✔

3042

            options.encryption_key = test_util::crypt_key();

2✔

3043

            auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

3044

            auto logger = util::Logger::get_default_logger();

2✔

3045

            sync::PendingBootstrapStore bootstrap_store(realm, *logger);

2✔

3047

            auto pending_batch = bootstrap_store.peek_pending(1024 * 1024 * 16);

2✔

3052

1✔

3053

            check_interrupted_state(realm);

2✔

3054

2✔

3055

1✔

3056

        // Now we'll open a different realm and make some changes that would leave orphan objects on the client

1✔

3057

        // if the bootstrap batches weren't being cached until lastInBatch were true.

1✔

3058

        mutate_realm();

2✔

3059

1✔

3060

        auto [saw_valid_state_promise, saw_valid_state_future] = util::make_promise_future<void>();

2✔

3061

        auto shared_saw_valid_state_promise =

2✔

3062

            std::make_shared<decltype(saw_valid_state_promise)>(std::move(saw_valid_state_promise));

2✔

3063

        // This hook will let us check what the state of the realm is before it's integrated any new download

1✔

3064

        // messages from the server. This should be the full 5 object bootstrap that was received before we

1✔

3065

        // called mutate_realm().

1✔

3066

        interrupted_realm_config.sync_config->on_sync_client_event_hook =

2✔

3067

            [&, promise = std::move(shared_saw_valid_state_promise)](std::weak_ptr<SyncSession> weak_session,

2✔

3068

                                                                     const SyncClientHookData& data) {

34✔

3069

                if (data.event != SyncClientHookEvent::DownloadMessageReceived) {

34✔

3070

                    return SyncClientHookAction::NoAction;

32✔

3071

32✔

3072

                auto session = weak_session.lock();

2✔

3073

                if (!session) {

2✔

3076

1✔

3077

                if (data.query_version != 1 || data.batch_state == sync::DownloadBatchState::MoreToCome) {

2✔

3080

1✔

3081

                auto latest_sub_set = session->get_flx_subscription_store()->get_latest();

2✔

3082

                auto active_sub_set = session->get_flx_subscription_store()->get_active();

2✔

3083

                auto version_info = session->get_flx_subscription_store()->get_version_info();

2✔

3089

1✔

3090

                auto db = SyncSession::OnlyForTesting::get_db(*session);

2✔

3091

                auto tr = db->start_read();

2✔

3092

1✔

3093

                auto table = tr->get_table("class_TopLevel");

2✔

3095

                for (auto& id : obj_ids_at_end) {

10✔

3097

10✔

3098

1✔

3099

                promise->emplace_value();

2✔

3100

                return SyncClientHookAction::NoAction;

2✔

3101

};

2✔

3102

1✔

3103

        // Finally re-open the realm whose bootstrap we interrupted and just wait for it to finish downloading.

1✔

3104

        auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

3105

        saw_valid_state_future.get();

2✔

3106

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

3107

        realm->get_latest_subscription_set()

2✔

3108

            .get_state_change_notification(sync::SubscriptionSet::State::Complete)

2✔

3109

            .get();

2✔

3110

1✔

3111

        wait_for_advance(*realm);

2✔

3112

        auto expected_obj_ids = util::Span<ObjectId>(obj_ids_at_end).sub_span(0, 3);

2✔

3113

1✔

3114

        // After we've downloaded all the mutations there should only by 3 objects left.

1✔

3116

        for (auto& id : expected_obj_ids) {

6✔

3118

6✔

3119

2✔

3120

8✔

3125

428✔

3126

    auto it = std::find_if(documents.begin(), documents.end(), [&](auto&& doc) {

43,096✔

3127

        auto val = doc.find("_id");

43,096✔

3129

        return *val == id;

43,096✔

3130

});

43,096✔

3132

    auto& doc = *it;

428✔

3133

    for (auto& [name, expected_value] : fields) {

434✔

3134

        auto val = doc.find(name);

434✔

3136

217✔

3137

        // bson documents are ordered  but Realm dictionaries aren't, so the

217✔

3138

        // document might validly be in a different order than we expected and

217✔

3139

        // we need to do a comparison that doesn't check order.

217✔

3140

        if (expected_value.type() == bson::Bson::Type::Document) {

434✔

3142

                    static_cast<const bson::BsonDocument&>(expected_value));

8✔

3143

8✔

3144

        else {

426✔

3146

426✔

3147

434✔

3148

428✔

3150

TEST_CASE("flx: data ingest", "[sync][flx][data ingest][baas]") {

22✔

3151

    using namespace ::realm::bson;

22✔

3152

11✔

3153

    static auto server_schema = [] {

12✔

3154

        FLXSyncTestHarness::ServerSchema server_schema;

2✔

3155

        server_schema.queryable_fields = {"queryable_str_field"};

2✔

3156

        server_schema.schema = {

2✔

3157

            {"Asymmetric",

2✔

3158

             ObjectSchema::ObjectType::TopLevelAsymmetric,

2✔

3159

2✔

3160

                 {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

3161

                 {"location", PropertyType::String | PropertyType::Nullable},

2✔

3162

                 {"embedded obj", PropertyType::Object | PropertyType::Nullable, "Embedded"},

2✔

3163

                 {"embedded list", PropertyType::Object | PropertyType::Array, "Embedded"},

2✔

3164

                 {"embedded dictionary", PropertyType::Object | PropertyType::Nullable | PropertyType::Dictionary,

2✔

3165

                  "Embedded"},

2✔

3166

                 {"link obj", PropertyType::Object | PropertyType::Nullable, "TopLevel"},

2✔

3167

                 {"link list", PropertyType::Object | PropertyType::Array, "TopLevel"},

2✔

3168

                 {"link dictionary", PropertyType::Object | PropertyType::Nullable | PropertyType::Dictionary,

2✔

3169

                  "TopLevel"},

2✔

3170

}},

2✔

3171

            {"Embedded", ObjectSchema::ObjectType::Embedded, {{"value", PropertyType::String}}},

2✔

3172

            {"TopLevel",

2✔

3173

2✔

3174

                 {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

3175

                 {"value", PropertyType::Int},

2✔

3176

}},

2✔

3177

};

2✔

3178

        return server_schema;

2✔

3179

    }();

2✔

3180

    static auto harness = std::make_unique<FLXSyncTestHarness>("asymmetric_sync", server_schema);

22✔

3181

11✔

3182

    // We reuse a single app for each section, so tests will see the documents

11✔

3183

    // created by previous tests and we need to add those documents to the count

11✔

3184

    // we're waiting for

11✔

3185

    static std::unordered_map<std::string, size_t> previous_count;

22✔

3186

    auto get_documents = [&](const char* name, size_t expected_count) {

20✔

3187

        auto& count = previous_count[name];

18✔

3188

        auto documents =

18✔

3189

            harness->session().get_documents(*harness->app()->current_user(), name, count + expected_count);

18✔

3190

        count = documents.size();

18✔

3191

        return documents;

18✔

3192

};

18✔

3193

11✔

3194

    SECTION("basic object construction") {

22✔

3195

        auto foo_obj_id = ObjectId::gen();

2✔

3196

        auto bar_obj_id = ObjectId::gen();

2✔

3197

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3198

            realm->begin_transaction();

2✔

3199

            CppContext c(realm);

2✔

3200

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", foo_obj_id}, {"location", "foo"s}}));

2✔

3201

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", bar_obj_id}, {"location", "bar"s}}));

2✔

3202

            realm->commit_transaction();

2✔

3203

1✔

3204

            auto documents = get_documents("Asymmetric", 2);

2✔

3205

            check_document(documents, foo_obj_id, {{"location", "foo"}});

2✔

3206

            check_document(documents, bar_obj_id, {{"location", "bar"}});

2✔

3207

});

2✔

3208

1✔

3209

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3210

            wait_for_download(*realm);

2✔

3211

1✔

3212

            auto table = realm->read_group().get_table("class_Asymmetric");

2✔

3214

            // Cannot query asymmetric tables.

1✔

3215

            CHECK_THROWS_AS(Query(table), LogicError);

2✔

3216

});

2✔

3217

2✔

3218

11✔

3219

    SECTION("do not allow objects with same key within the same transaction") {

22✔

3220

        auto foo_obj_id = ObjectId::gen();

2✔

3221

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3222

            realm->begin_transaction();

2✔

3223

            CppContext c(realm);

2✔

3224

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", foo_obj_id}, {"location", "foo"s}}));

2✔

3225

            CHECK_THROWS_WITH(

2✔

3226

                Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", foo_obj_id}, {"location", "bar"s}})),

2✔

3227

                "Attempting to create an object of type 'Asymmetric' with an existing primary key value 'not "

2✔

3228

                "implemented'");

2✔

3229

            realm->commit_transaction();

2✔

3230

1✔

3231

            auto documents = get_documents("Asymmetric", 1);

2✔

3232

            check_document(documents, foo_obj_id, {{"location", "foo"}});

2✔

3233

});

2✔

3234

2✔

3235

11✔

3236

    SECTION("create multiple objects - separate commits") {

22✔

3237

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3238

            CppContext c(realm);

2✔

3239

            std::vector<ObjectId> obj_ids;

2✔

3240

            for (int i = 0; i < 100; ++i) {

202✔

3241

                realm->begin_transaction();

200✔

3242

                obj_ids.push_back(ObjectId::gen());

200✔

3243

                Object::create(c, realm, "Asymmetric",

200✔

3244

                               std::any(AnyDict{

200✔

3245

                                   {"_id", obj_ids.back()},

200✔

3246

                                   {"location", util::format("foo_%1", i)},

200✔

3247

                               }));

200✔

3248

                realm->commit_transaction();

200✔

3249

200✔

3250

1✔

3251

            wait_for_upload(*realm);

2✔

3252

            wait_for_download(*realm);

2✔

3253

1✔

3254

            auto table = realm->read_group().get_table("class_Asymmetric");

2✔

3256

1✔

3257

            auto documents = get_documents("Asymmetric", 100);

2✔

3258

            for (int i = 0; i < 100; ++i) {

202✔

3259

                check_document(documents, obj_ids[i], {{"location", util::format("foo_%1", i)}});

200✔

3260

200✔

3261

});

2✔

3262

2✔

3263

11✔

3264

    SECTION("create multiple objects - same commit") {

22✔

3265

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3266

            CppContext c(realm);

2✔

3267

            realm->begin_transaction();

2✔

3268

            std::vector<ObjectId> obj_ids;

2✔

3269

            for (int i = 0; i < 100; ++i) {

202✔

3270

                obj_ids.push_back(ObjectId::gen());

200✔

3271

                Object::create(c, realm, "Asymmetric",

200✔

3272

                               std::any(AnyDict{

200✔

3273

                                   {"_id", obj_ids.back()},

200✔

3274

                                   {"location", util::format("bar_%1", i)},

200✔

3275

                               }));

200✔

3276

200✔

3277

            realm->commit_transaction();

2✔

3278

1✔

3279

            wait_for_upload(*realm);

2✔

3280

            wait_for_download(*realm);

2✔

3281

1✔

3282

            auto table = realm->read_group().get_table("class_Asymmetric");

2✔

3284

1✔

3285

            auto documents = get_documents("Asymmetric", 100);

2✔

3286

            for (int i = 0; i < 100; ++i) {

202✔

3287

                check_document(documents, obj_ids[i], {{"location", util::format("bar_%1", i)}});

200✔

3288

200✔

3289

});

2✔

3290

2✔

3291

11✔

3292

    SECTION("open with schema mismatch on IsAsymmetric") {

22✔

3293

        auto schema = server_schema.schema;

2✔

3294

        schema.find("Asymmetric")->table_type = ObjectSchema::ObjectType::TopLevel;

2✔

3295

1✔

3296

        harness->do_with_new_user([&](std::shared_ptr<SyncUser> user) {

2✔

3297

            SyncTestFile config(user, schema, SyncConfig::FLXSyncEnabled{});

2✔

3298

            auto [error_promise, error_future] = util::make_promise_future<SyncError>();

2✔

3299

            auto error_count = 0;

2✔

3300

            auto err_handler = [promise = util::CopyablePromiseHolder(std::move(error_promise)),

2✔

3301

                                &error_count](std::shared_ptr<SyncSession>, SyncError err) mutable {

4✔

3302

                ++error_count;

4✔

3303

                if (error_count == 1) {

4✔

3304

                    // Bad changeset detected by the client.

1✔

3306

2✔

3307

                else if (error_count == 2) {

2✔

3308

                    // Server asking for a client reset.

1✔

3311

                    promise.get_promise().emplace_value(std::move(err));

2✔

3312

2✔

3313

};

4✔

3314

1✔

3315

            config.sync_config->error_handler = err_handler;

2✔

3316

            auto realm = Realm::get_shared_realm(config);

2✔

3317

1✔

3318

            auto err = error_future.get();

2✔

3320

});

2✔

3321

2✔

3322

11✔

3323

    SECTION("basic embedded object construction") {

22✔

3324

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3325

            auto obj_id = ObjectId::gen();

2✔

3326

            realm->begin_transaction();

2✔

3327

            CppContext c(realm);

2✔

3328

            Object::create(c, realm, "Asymmetric",

2✔

3329

                           std::any(AnyDict{

2✔

3330

                               {"_id", obj_id},

2✔

3331

                               {"embedded obj", AnyDict{{"value", "foo"s}}},

2✔

3332

                           }));

2✔

3333

            realm->commit_transaction();

2✔

3334

            wait_for_upload(*realm);

2✔

3335

1✔

3336

            auto documents = get_documents("Asymmetric", 1);

2✔

3337

            check_document(documents, obj_id, {{"embedded obj", BsonDocument{{"value", "foo"}}}});

2✔

3338

});

2✔

3339

1✔

3340

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3341

            wait_for_download(*realm);

2✔

3342

1✔

3343

            auto table = realm->read_group().get_table("class_Asymmetric");

2✔

3345

});

2✔

3346

2✔

3347

11✔

3348

    SECTION("replace embedded object") {

22✔

3349

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3350

            CppContext c(realm);

2✔

3351

            auto foo_obj_id = ObjectId::gen();

2✔

3352

1✔

3353

            realm->begin_transaction();

2✔

3354

            Object::create(c, realm, "Asymmetric",

2✔

3355

                           std::any(AnyDict{

2✔

3356

                               {"_id", foo_obj_id},

2✔

3357

                               {"embedded obj", AnyDict{{"value", "foo"s}}},

2✔

3358

                           }));

2✔

3359

            realm->commit_transaction();

2✔

3360

1✔

3361

            // Update embedded field to `null`. The server discards this write

1✔

3362

            // as asymmetric sync can only create new objects.

1✔

3363

            realm->begin_transaction();

2✔

3364

            Object::create(c, realm, "Asymmetric",

2✔

3365

                           std::any(AnyDict{

2✔

3366

                               {"_id", foo_obj_id},

2✔

3367

                               {"embedded obj", std::any()},

2✔

3368

                           }));

2✔

3369

            realm->commit_transaction();

2✔

3370

1✔

3371

            // create a second object so that we can know when the translator

1✔

3372

            // has processed everything

1✔

3373

            realm->begin_transaction();

2✔

3374

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", ObjectId::gen()}, {}}));

2✔

3375

            realm->commit_transaction();

2✔

3376

1✔

3377

            wait_for_upload(*realm);

2✔

3378

            wait_for_download(*realm);

2✔

3379

1✔

3380

            auto table = realm->read_group().get_table("class_Asymmetric");

2✔

3382

1✔

3383

            auto documents = get_documents("Asymmetric", 2);

2✔

3384

            check_document(documents, foo_obj_id, {{"embedded obj", BsonDocument{{"value", "foo"}}}});

2✔

3385

});

2✔

3386

2✔

3387

11✔

3388

    SECTION("embedded collections") {

22✔

3389

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3390

            CppContext c(realm);

2✔

3391

            auto obj_id = ObjectId::gen();

2✔

3392

1✔

3393

            realm->begin_transaction();

2✔

3394

            Object::create(c, realm, "Asymmetric",

2✔

3395

                           std::any(AnyDict{

2✔

3396

                               {"_id", obj_id},

2✔

3397

                               {"embedded list", AnyVector{AnyDict{{"value", "foo"s}}, AnyDict{{"value", "bar"s}}}},

2✔

3398

                               {"embedded dictionary",

2✔

3399

                                AnyDict{

2✔

3400

                                    {"key1", AnyDict{{"value", "foo"s}}},

2✔

3401

                                    {"key2", AnyDict{{"value", "bar"s}}},

2✔

3402

}},

2✔

3403

                           }));

2✔

3404

            realm->commit_transaction();

2✔

3405

1✔

3406

            auto documents = get_documents("Asymmetric", 1);

2✔

3407

            check_document(

2✔

3408

                documents, obj_id,

2✔

3409

2✔

3410

                    {"embedded list", BsonArray{BsonDocument{{"value", "foo"}}, BsonDocument{{"value", "bar"}}}},

2✔

3411

                    {"embedded dictionary",

2✔

3412

                     BsonDocument{

2✔

3413

                         {"key1", BsonDocument{{"value", "foo"}}},

2✔

3414

                         {"key2", BsonDocument{{"value", "bar"}}},

2✔

3415

}},

2✔

3416

});

2✔

3417

});

2✔

3418

2✔

3419

11✔

3420

    SECTION("asymmetric table not allowed in PBS") {

22✔

3421

        Schema schema{

2✔

3422

            {"Asymmetric2",

2✔

3423

             ObjectSchema::ObjectType::TopLevelAsymmetric,

2✔

3424

2✔

3425

                 {"_id", PropertyType::Int, Property::IsPrimary{true}},

2✔

3426

                 {"location", PropertyType::Int},

2✔

3427

                 {"reading", PropertyType::Int},

2✔

3428

}},

2✔

3429

};

2✔

3430

1✔

3431

        SyncTestFile config(harness->app()->current_user(), Bson{}, schema);

2✔

3432

        REQUIRE_EXCEPTION(

2✔

3433

            Realm::get_shared_realm(config), SchemaValidationFailed,

2✔

3434

            Catch::Matchers::ContainsSubstring("Asymmetric table 'Asymmetric2' not allowed in partition based sync"));

2✔

3435

2✔

3436

11✔

3437

    SECTION("links to top-level objects") {

22✔

3438

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3439

            subscribe_to_all_and_bootstrap(*realm);

2✔

3440

1✔

3441

            ObjectId obj_id = ObjectId::gen();

2✔

3442

            std::array<ObjectId, 5> target_obj_ids;

2✔

3443

            for (auto& id : target_obj_ids) {

10✔

3444

                id = ObjectId::gen();

10✔

3445

10✔

3446

1✔

3447

            realm->begin_transaction();

2✔

3448

            CppContext c(realm);

2✔

3449

            Object::create(c, realm, "Asymmetric",

2✔

3450

                           std::any(AnyDict{

2✔

3451

                               {"_id", obj_id},

2✔

3452

                               {"link obj", AnyDict{{"_id", target_obj_ids[0]}, {"value", INT64_C(10)}}},

2✔

3453

                               {"link list",

2✔

3454

                                AnyVector{

2✔

3455

                                    AnyDict{{"_id", target_obj_ids[1]}, {"value", INT64_C(11)}},

2✔

3456

                                    AnyDict{{"_id", target_obj_ids[2]}, {"value", INT64_C(12)}},

2✔

3457

}},

2✔

3458

                               {"link dictionary",

2✔

3459

                                AnyDict{

2✔

3460

                                    {"key1", AnyDict{{"_id", target_obj_ids[3]}, {"value", INT64_C(13)}}},

2✔

3461

                                    {"key2", AnyDict{{"_id", target_obj_ids[4]}, {"value", INT64_C(14)}}},

2✔

3462

}},

2✔

3463

                           }));

2✔

3464

            realm->commit_transaction();

2✔

3465

            wait_for_upload(*realm);

2✔

3466

1✔

3467

            auto docs1 = get_documents("Asymmetric", 1);

2✔

3468

            check_document(docs1, obj_id,

2✔

3469

                           {{"link obj", target_obj_ids[0]},

2✔

3470

                            {"link list", BsonArray{{target_obj_ids[1], target_obj_ids[2]}}},

2✔

3471

2✔

3472

                                "link dictionary",

2✔

3473

                                BsonDocument{

2✔

3474

                                    {"key1", target_obj_ids[3]},

2✔

3475

                                    {"key2", target_obj_ids[4]},

2✔

3476

},

2✔

3477

                            }});

2✔

3478

1✔

3479

            auto docs2 = get_documents("TopLevel", 5);

2✔

3480

            for (int64_t i = 0; i < 5; ++i) {

12✔

3481

                check_document(docs2, target_obj_ids[i], {{"value", 10 + i}});

10✔

3482

10✔

3483

});

2✔

3484

2✔

3485

11✔

3486

    // Add any new test sections above this point

11✔

3487

11✔

3488

    SECTION("teardown") {

22✔

3489

        harness.reset();

2✔

3490

2✔

3491

22✔

3493

TEST_CASE("flx: data ingest - dev mode", "[sync][flx][data ingest][baas]") {

2✔

3494

    FLXSyncTestHarness::ServerSchema server_schema;

2✔

3495

    server_schema.dev_mode_enabled = true;

2✔

3496

    server_schema.schema = Schema{};

2✔

3497

1✔

3498

    auto schema = Schema{{"Asymmetric",

2✔

3499

                          ObjectSchema::ObjectType::TopLevelAsymmetric,

2✔

3500

2✔

3501

                              {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

3502

                              {"location", PropertyType::String | PropertyType::Nullable},

2✔

3503

}},

2✔

3504

                         {"TopLevel",

2✔

3505

2✔

3506

                              {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

3507

                              {"queryable_str_field", PropertyType::String | PropertyType::Nullable},

2✔

3508

                          }}};

2✔

3509

1✔

3510

    FLXSyncTestHarness harness("asymmetric_sync", server_schema);

2✔

3511

1✔

3512

    auto foo_obj_id = ObjectId::gen();

2✔

3513

    auto bar_obj_id = ObjectId::gen();

2✔

3514

1✔

3515

    harness.do_with_new_realm(

2✔

3516

        [&](SharedRealm realm) {

2✔

3517

            CppContext c(realm);

2✔

3518

            realm->begin_transaction();

2✔

3519

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", foo_obj_id}, {"location", "foo"s}}));

2✔

3520

            Object::create(c, realm, "Asymmetric", std::any(AnyDict{{"_id", bar_obj_id}, {"location", "bar"s}}));

2✔

3521

            realm->commit_transaction();

2✔

3522

            User* user = dynamic_cast<User*>(realm->config().sync_config->user.get());

2✔

3523

            REALM_ASSERT(user);

2✔

3524

            auto docs = harness.session().get_documents(*user, "Asymmetric", 2);

2✔

3525

            check_document(docs, foo_obj_id, {{"location", "foo"}});

2✔

3526

            check_document(docs, bar_obj_id, {{"location", "bar"}});

2✔

3527

},

2✔

3528

        schema);

2✔

3529

2✔

3531

TEST_CASE("flx: data ingest - write not allowed", "[sync][flx][data ingest][baas]") {

2✔

3532

    AppCreateConfig::ServiceRole role;

2✔

3533

    role.name = "asymmetric_write_perms";

2✔

3534

1✔

3535

    AppCreateConfig::ServiceRoleDocumentFilters doc_filters;

2✔

3536

    doc_filters.read = true;

2✔

3537

    doc_filters.write = false;

2✔

3538

    role.document_filters = doc_filters;

2✔

3539

1✔

3540

    role.insert_filter = true;

2✔

3541

    role.delete_filter = true;

2✔

3542

    role.read = true;

2✔

3543

    role.write = true;

2✔

3544

1✔

3545

    Schema schema({

2✔

3546

        {"Asymmetric",

2✔

3547

         ObjectSchema::ObjectType::TopLevelAsymmetric,

2✔

3548

2✔

3549

             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

3550

             {"location", PropertyType::String | PropertyType::Nullable},

2✔

3551

             {"embedded_obj", PropertyType::Object | PropertyType::Nullable, "Embedded"},

2✔

3552

}},

2✔

3553

        {"Embedded",

2✔

3554

         ObjectSchema::ObjectType::Embedded,

2✔

3555

2✔

3556

             {"value", PropertyType::String | PropertyType::Nullable},

2✔

3557

}},

2✔

3558

});

2✔

3559

    FLXSyncTestHarness::ServerSchema server_schema{schema, {}, {role}};

2✔

3560

    FLXSyncTestHarness harness("asymmetric_sync", server_schema);

2✔

3561

1✔

3562

    auto error_received_pf = util::make_promise_future<void>();

2✔

3563

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

3564

    config.sync_config->on_sync_client_event_hook =

2✔

3565

        [promise = util::CopyablePromiseHolder(std::move(error_received_pf.promise))](

2✔

3566

            std::weak_ptr<SyncSession> weak_session, const SyncClientHookData& data) mutable {

22✔

3567

            if (data.event != SyncClientHookEvent::ErrorMessageReceived) {

22✔

3568

                return SyncClientHookAction::NoAction;

18✔

3569

18✔

3570

            auto session = weak_session.lock();

4✔

3572

2✔

3573

            auto error_code = sync::ProtocolError(data.error_info->raw_error_code);

4✔

3574

2✔

3575

            if (error_code == sync::ProtocolError::initial_sync_not_completed) {

4✔

3576

                return SyncClientHookAction::NoAction;

2✔

3577

2✔

3578

1✔

3582

            promise.get_promise().emplace_value();

2✔

3583

1✔

3584

            return SyncClientHookAction::EarlyReturn;

2✔

3585

};

2✔

3586

1✔

3587

    auto realm = Realm::get_shared_realm(config);

2✔

3588

1✔

3589

    // Create an asymmetric object and upload it to the server.

1✔

3590

2✔

3591

        realm->begin_transaction();

2✔

3592

        CppContext c(realm);

2✔

3593

        Object::create(c, realm, "Asymmetric",

2✔

3594

                       std::any(AnyDict{{"_id", ObjectId::gen()}, {"embedded_obj", AnyDict{{"value", "foo"s}}}}));

2✔

3595

        realm->commit_transaction();

2✔

3596

2✔

3597

1✔

3598

    error_received_pf.future.get();

2✔

3599

    realm->close();

2✔

3600

2✔

3602

TEST_CASE("flx: send client error", "[sync][flx][baas]") {

2✔

3603

    FLXSyncTestHarness harness("flx_client_error");

2✔

3604

1✔

3605

    // An integration error is simulated while bootstrapping.

1✔

3606

    // This results in the client sending an error message to the server.

1✔

3607

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

3608

    config.sync_config->simulate_integration_error = true;

2✔

3609

1✔

3610

    auto [error_promise, error_future] = util::make_promise_future<SyncError>();

2✔

3611

    auto error_count = 0;

2✔

3612

    auto err_handler = [promise = util::CopyablePromiseHolder(std::move(error_promise)),

2✔

3613

                        &error_count](std::shared_ptr<SyncSession>, SyncError err) mutable {

4✔

3614

        ++error_count;

4✔

3615

        if (error_count == 1) {

4✔

3616

            // Bad changeset detected by the client.

1✔

3618

2✔

3619

        else if (error_count == 2) {

2✔

3620

            // Server asking for a client reset.

1✔

3623

            promise.get_promise().emplace_value(std::move(err));

2✔

3624

2✔

3625

};

4✔

3626

1✔

3627

    config.sync_config->error_handler = err_handler;

2✔

3628

    auto realm = Realm::get_shared_realm(config);

2✔

3629

    auto table = realm->read_group().get_table("class_TopLevel");

2✔

3630

    auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

3631

    new_query.insert_or_assign(Query(table));

2✔

3632

    new_query.commit();

2✔

3633

1✔

3634

    auto err = error_future.get();

2✔

3636

2✔

3638

TEST_CASE("flx: bootstraps contain all changes", "[sync][flx][bootstrap][baas]") {

6✔

3639

    FLXSyncTestHarness harness("bootstrap_full_sync");

6✔

3640

3✔

3641

    auto setup_subs = [](SharedRealm& realm) {

12✔

3642

        auto table = realm->read_group().get_table("class_TopLevel");

12✔

3643

        auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

12✔

3644

        new_query.clear();

12✔

3645

        auto col = table->get_column_key("queryable_str_field");

12✔

3646

        new_query.insert_or_assign(Query(table).equal(col, StringData("bar")).Or().equal(col, StringData("bizz")));

12✔

3647

        return new_query.commit();

12✔

3648

};

12✔

3649

3✔

3650

    auto bar_obj_id = ObjectId::gen();

6✔

3651

    auto bizz_obj_id = ObjectId::gen();

6✔

3652

    auto setup_and_poison_cache = [&] {

6✔

3653

        harness.load_initial_data([&](SharedRealm realm) {

6✔

3654

            CppContext c(realm);

6✔

3655

            Object::create(c, realm, "TopLevel",

6✔

3656

                           std::any(AnyDict{{"_id", bar_obj_id},

6✔

3657

                                            {"queryable_str_field", std::string{"bar"}},

6✔

3658

                                            {"queryable_int_field", static_cast<int64_t>(10)},

6✔

3659

                                            {"non_queryable_field", std::string{"non queryable 2"}}}));

6✔

3660

});

6✔

3661

3✔

3662

        harness.do_with_new_realm([&](SharedRealm realm) {

6✔

3663

            // first set a subscription to force the creation/caching of a broker snapshot on the server.

3✔

3664

            setup_subs(realm).get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

6✔

3665

            wait_for_advance(*realm);

6✔

3666

            auto table = realm->read_group().get_table("class_TopLevel");

6✔

3668

3✔

3669

            // Then create an object that won't be in the cached snapshot - this is the object that if we didn't

3✔

3670

            // wait for a MARK message to come back, we'd miss it in our results.

3✔

3671

            CppContext c(realm);

6✔

3672

            realm->begin_transaction();

6✔

3673

            Object::create(c, realm, "TopLevel",

6✔

3674

                           std::any(AnyDict{{"_id", bizz_obj_id},

6✔

3675

                                            {"queryable_str_field", std::string{"bizz"}},

6✔

3676

                                            {"queryable_int_field", static_cast<int64_t>(15)},

6✔

3677

                                            {"non_queryable_field", std::string{"non queryable 3"}}}));

6✔

3678

            realm->commit_transaction();

6✔

3679

            wait_for_upload(*realm);

6✔

3680

});

6✔

3681

};

6✔

3682

3✔

3683

    SECTION("regular subscription change") {

6✔

3684

        SyncTestFile triggered_config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

3685

        std::atomic<bool> saw_truncated_bootstrap{false};

2✔

3686

        triggered_config.sync_config->on_sync_client_event_hook = [&](std::weak_ptr<SyncSession> weak_sess,

2✔

3687

                                                                      const SyncClientHookData& data) {

32✔

3688

            auto sess = weak_sess.lock();

32✔

3689

            if (!sess || data.event != SyncClientHookEvent::BootstrapProcessed || data.query_version != 1) {

32✔

3690

                return SyncClientHookAction::NoAction;

30✔

3691

30✔

3692

1✔

3693

            auto latest_subs = sess->get_flx_subscription_store()->get_latest();

2✔

3696

            auto db = SyncSession::OnlyForTesting::get_db(*sess);

2✔

3697

            auto read_tr = db->start_read();

2✔

3698

            auto table = read_tr->get_table("class_TopLevel");

2✔

3701

            saw_truncated_bootstrap.store(true);

2✔

3702

1✔

3703

            return SyncClientHookAction::NoAction;

2✔

3704

};

2✔

3705

        auto problem_realm = Realm::get_shared_realm(triggered_config);

2✔

3706

1✔

3707

        // Setup the problem realm by waiting for it to be fully synchronized with an empty query, so the router

1✔

3708

        // on the server should have no new history entries, and then pause the router so it doesn't get any of

1✔

3709

        // the changes we're about to create.

1✔

3710

        wait_for_upload(*problem_realm);

2✔

3711

        wait_for_download(*problem_realm);

2✔

3712

1✔

3713

        nlohmann::json command_request = {

2✔

3714

            {"command", "PAUSE_ROUTER_SESSION"},

2✔

3715

};

2✔

3716

        auto resp_body =

2✔

3717

            SyncSession::OnlyForTesting::send_test_command(*problem_realm->sync_session(), command_request.dump())

2✔

3718

                .get();

2✔

3720

1✔

3721

        // Put some data into the server, this will be the data that will be in the broker cache.

1✔

3722

        setup_and_poison_cache();

2✔

3723

1✔

3724

        // Setup queries on the problem realm to bootstrap from the cached object. Bootstrapping will also resume

1✔

3725

        // the router, so all we need to do is wait for the subscription set to be complete and notifications to be

1✔

3726

        // processed.

1✔

3727

        setup_subs(problem_realm).get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

3728

        wait_for_advance(*problem_realm);

2✔

3729

1✔

3731

        auto table = problem_realm->read_group().get_table("class_TopLevel");

2✔

3734

2✔

3735

3✔

3736

// TODO: remote-baas: This test fails intermittently with Windows remote baas server - to be fixed in RCORE-1674

3✔

3737

#ifndef _WIN32

6✔

3738

    SECTION("disconnect between bootstrap and mark") {

6✔

3739

        SyncTestFile triggered_config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

3740

        auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

3741

        triggered_config.sync_config->on_sync_client_event_hook =

2✔

3742

            [promise = util::CopyablePromiseHolder(std::move(interrupted_promise)), &bizz_obj_id,

2✔

3743

             &bar_obj_id](std::weak_ptr<SyncSession> weak_sess, const SyncClientHookData& data) mutable {

36✔

3744

                auto sess = weak_sess.lock();

36✔

3745

                if (!sess || data.event != SyncClientHookEvent::BootstrapProcessed || data.query_version != 1) {

36✔

3746

                    return SyncClientHookAction::NoAction;

34✔

3747

34✔

3748

1✔

3749

                auto latest_subs = sess->get_flx_subscription_store()->get_latest();

2✔

3752

                auto db = SyncSession::OnlyForTesting::get_db(*sess);

2✔

3753

                auto read_tr = db->start_read();

2✔

3754

                auto table = read_tr->get_table("class_TopLevel");

2✔

3757

1✔

3758

                sess->pause();

2✔

3759

                promise.get_promise().emplace_value();

2✔

3760

                return SyncClientHookAction::NoAction;

2✔

3761

};

2✔

3762

        auto problem_realm = Realm::get_shared_realm(triggered_config);

2✔

3763

1✔

3764

        // Setup the problem realm by waiting for it to be fully synchronized with an empty query, so the router

1✔

3765

        // on the server should have no new history entries, and then pause the router so it doesn't get any of

1✔

3766

        // the changes we're about to create.

1✔

3767

        wait_for_upload(*problem_realm);

2✔

3768

        wait_for_download(*problem_realm);

2✔

3769

1✔

3770

        nlohmann::json command_request = {

2✔

3771

            {"command", "PAUSE_ROUTER_SESSION"},

2✔

3772

};

2✔

3773

        auto resp_body =

2✔

3774

            SyncSession::OnlyForTesting::send_test_command(*problem_realm->sync_session(), command_request.dump())

2✔

3775

                .get();

2✔

3777

1✔

3778

        // Put some data into the server, this will be the data that will be in the broker cache.

1✔

3779

        setup_and_poison_cache();

2✔

3780

1✔

3781

        // Setup queries on the problem realm to bootstrap from the cached object. Bootstrapping will also resume

1✔

3782

        // the router, so all we need to do is wait for the subscription set to be complete and notifications to be

1✔

3783

        // processed.

1✔

3784

        auto sub_set = setup_subs(problem_realm);

2✔

3785

        auto sub_complete_future = sub_set.get_state_change_notification(sync::SubscriptionSet::State::Complete);

2✔

3786

1✔

3787

        interrupted.get();

2✔

3788

        problem_realm->sync_session()->shutdown_and_wait();

2✔

3790

        sub_set.refresh();

2✔

3792

1✔

3793

        sub_complete_future = sub_set.get_state_change_notification(sync::SubscriptionSet::State::Complete);

2✔

3794

        problem_realm->sync_session()->resume();

2✔

3795

        sub_complete_future.get();

2✔

3796

        wait_for_advance(*problem_realm);

2✔

3797

1✔

3798

        sub_set.refresh();

2✔

3800

        auto table = problem_realm->read_group().get_table("class_TopLevel");

2✔

3803

2✔

3804

#endif

6✔

3805

    SECTION("error/suspend between bootstrap and mark") {

6✔

3806

        SyncTestFile triggered_config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

3807

        triggered_config.sync_config->on_sync_client_event_hook =

2✔

3808

            [&bizz_obj_id, &bar_obj_id](std::weak_ptr<SyncSession> weak_sess, const SyncClientHookData& data) {

34✔

3809

                auto sess = weak_sess.lock();

34✔

3810

                if (!sess || data.event != SyncClientHookEvent::BootstrapProcessed || data.query_version != 1) {

34✔

3811

                    return SyncClientHookAction::NoAction;

32✔

3812

32✔

3813

1✔

3814

                auto latest_subs = sess->get_flx_subscription_store()->get_latest();

2✔

3817

                auto db = SyncSession::OnlyForTesting::get_db(*sess);

2✔

3818

                auto read_tr = db->start_read();

2✔

3819

                auto table = read_tr->get_table("class_TopLevel");

2✔

3822

1✔

3823

                return SyncClientHookAction::TriggerReconnect;

2✔

3824

};

2✔

3825

        auto problem_realm = Realm::get_shared_realm(triggered_config);

2✔

3826

1✔

3827

        // Setup the problem realm by waiting for it to be fully synchronized with an empty query, so the router

1✔

3828

        // on the server should have no new history entries, and then pause the router so it doesn't get any of

1✔

3829

        // the changes we're about to create.

1✔

3830

        wait_for_upload(*problem_realm);

2✔

3831

        wait_for_download(*problem_realm);

2✔

3832

1✔

3833

        nlohmann::json command_request = {

2✔

3834

            {"command", "PAUSE_ROUTER_SESSION"},

2✔

3835

};

2✔

3836

        auto resp_body =

2✔

3837

            SyncSession::OnlyForTesting::send_test_command(*problem_realm->sync_session(), command_request.dump())

2✔

3838

                .get();

2✔

3840

1✔

3841

        // Put some data into the server, this will be the data that will be in the broker cache.

1✔

3842

        setup_and_poison_cache();

2✔

3843

1✔

3844

        // Setup queries on the problem realm to bootstrap from the cached object. Bootstrapping will also resume

1✔

3845

        // the router, so all we need to do is wait for the subscription set to be complete and notifications to be

1✔

3846

        // processed.

1✔

3847

        auto sub_set = setup_subs(problem_realm);

2✔

3848

        auto sub_complete_future = sub_set.get_state_change_notification(sync::SubscriptionSet::State::Complete);

2✔

3849

1✔

3850

        sub_complete_future.get();

2✔

3851

        wait_for_advance(*problem_realm);

2✔

3852

1✔

3853

        sub_set.refresh();

2✔

3855

        auto table = problem_realm->read_group().get_table("class_TopLevel");

2✔

3858

2✔

3859

6✔

3861

TEST_CASE("flx: convert flx sync realm to bundled realm", "[app][flx][baas]") {

12✔

3862

    static auto foo_obj_id = ObjectId::gen();

12✔

3863

    static auto bar_obj_id = ObjectId::gen();

12✔

3864

    static auto bizz_obj_id = ObjectId::gen();

12✔

3865

    static std::optional<FLXSyncTestHarness> harness;

12✔

3866

    if (!harness) {

12✔

3867

        harness.emplace("bundled_flx_realms");

2✔

3868

        harness->load_initial_data([&](SharedRealm realm) {

2✔

3869

            CppContext c(realm);

2✔

3870

            Object::create(c, realm, "TopLevel",

2✔

3871

                           std::any(AnyDict{{"_id", foo_obj_id},

2✔

3872

                                            {"queryable_str_field", "foo"s},

2✔

3873

                                            {"queryable_int_field", static_cast<int64_t>(5)},

2✔

3874

                                            {"non_queryable_field", "non queryable 1"s}}));

2✔

3875

            Object::create(c, realm, "TopLevel",

2✔

3876

                           std::any(AnyDict{{"_id", bar_obj_id},

2✔

3877

                                            {"queryable_str_field", "bar"s},

2✔

3878

                                            {"queryable_int_field", static_cast<int64_t>(10)},

2✔

3879

                                            {"non_queryable_field", "non queryable 2"s}}));

2✔

3880

});

2✔

3881

2✔

3882

6✔

3883

    SECTION("flx to flx (should succeed)") {

12✔

3884

        create_user_and_log_in(harness->app());

2✔

3885

        SyncTestFile target_config(harness->app()->current_user(), harness->schema(), SyncConfig::FLXSyncEnabled{});

2✔

3886

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3887

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

3888

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

3889

            mut_subs.insert_or_assign(Query(table).greater(table->get_column_key("queryable_int_field"), 5));

2✔

3890

            auto subs = std::move(mut_subs).commit();

2✔

3891

1✔

3892

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

3893

            wait_for_advance(*realm);

2✔

3894

1✔

3895

            realm->convert(target_config);

2✔

3896

});

2✔

3897

1✔

3898

        auto target_realm = Realm::get_shared_realm(target_config);

2✔

3899

1✔

3900

        target_realm->begin_transaction();

2✔

3901

        CppContext c(target_realm);

2✔

3902

        Object::create(c, target_realm, "TopLevel",

2✔

3903

                       std::any(AnyDict{{"_id", bizz_obj_id},

2✔

3904

                                        {"queryable_str_field", "bizz"s},

2✔

3905

                                        {"queryable_int_field", static_cast<int64_t>(15)},

2✔

3906

                                        {"non_queryable_field", "non queryable 3"s}}));

2✔

3907

        target_realm->commit_transaction();

2✔

3908

1✔

3909

        wait_for_upload(*target_realm);

2✔

3910

        wait_for_download(*target_realm);

2✔

3911

1✔

3912

        auto latest_subs = target_realm->get_active_subscription_set();

2✔

3913

        auto table = target_realm->read_group().get_table("class_TopLevel");

2✔

3917

                Query(table).greater(table->get_column_key("queryable_int_field"), 5).get_description());

2✔

3918

1✔

3923

2✔

3924

6✔

3925

    SECTION("flx to local (should succeed)") {

12✔

3926

        TestFile target_config;

2✔

3927

1✔

3928

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3929

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

3930

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

3931

            mut_subs.insert_or_assign(Query(table).greater(table->get_column_key("queryable_int_field"), 5));

2✔

3932

            auto subs = std::move(mut_subs).commit();

2✔

3933

1✔

3934

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

3935

            wait_for_advance(*realm);

2✔

3936

1✔

3937

            target_config.schema = realm->schema();

2✔

3938

            target_config.schema_version = realm->schema_version();

2✔

3939

            realm->convert(target_config);

2✔

3940

});

2✔

3941

1✔

3942

        auto target_realm = Realm::get_shared_realm(target_config);

2✔

3943

        REQUIRE_THROWS(target_realm->get_active_subscription_set());

2✔

3944

1✔

3945

        auto table = target_realm->read_group().get_table("class_TopLevel");

2✔

3950

2✔

3951

6✔

3952

    SECTION("flx to pbs (should fail to convert)") {

12✔

3953

        create_user_and_log_in(harness->app());

2✔

3954

        SyncTestFile target_config(harness->app()->current_user(), "12345"s, harness->schema());

2✔

3955

        harness->do_with_new_realm([&](SharedRealm realm) {

2✔

3956

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

3957

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

3958

            mut_subs.insert_or_assign(Query(table).greater(table->get_column_key("queryable_int_field"), 5));

2✔

3959

            auto subs = std::move(mut_subs).commit();

2✔

3960

1✔

3961

            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();

2✔

3962

            wait_for_advance(*realm);

2✔

3963

1✔

3964

            REQUIRE_THROWS(realm->convert(target_config));

2✔

3965

});

2✔

3966

2✔

3967

6✔

3968

    SECTION("pbs to flx (should fail to convert)") {

12✔

3969

        create_user_and_log_in(harness->app());

2✔

3970

        SyncTestFile target_config(harness->app()->current_user(), harness->schema(), SyncConfig::FLXSyncEnabled{});

2✔

3971

1✔

3972

        auto pbs_app_config = minimal_app_config("pbs_to_flx_convert", harness->schema());

2✔

3973

1✔

3974

        TestAppSession pbs_app_session(create_app(pbs_app_config));

2✔

3975

        SyncTestFile source_config(pbs_app_session.app()->current_user(), "54321"s, pbs_app_config.schema);

2✔

3976

        auto realm = Realm::get_shared_realm(source_config);

2✔

3977

1✔

3978

        realm->begin_transaction();

2✔

3979

        CppContext c(realm);

2✔

3980

        Object::create(c, realm, "TopLevel",

2✔

3981

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

3982

                                        {"queryable_str_field", "foo"s},

2✔

3983

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

3984

                                        {"non_queryable_field", "non queryable 1"s}}));

2✔

3985

        realm->commit_transaction();

2✔

3986

1✔

3987

        REQUIRE_THROWS(realm->convert(target_config));

2✔

3988

2✔

3989

6✔

3990

    SECTION("local to flx (should fail to convert)") {

12✔

3991

        TestFile source_config;

2✔

3992

        source_config.schema = harness->schema();

2✔

3993

        source_config.schema_version = 1;

2✔

3994

1✔

3995

        auto realm = Realm::get_shared_realm(source_config);

2✔

3996

        auto foo_obj_id = ObjectId::gen();

2✔

3997

1✔

3998

        realm->begin_transaction();

2✔

3999

        CppContext c(realm);

2✔

4000

        Object::create(c, realm, "TopLevel",

2✔

4001

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

4002

                                        {"queryable_str_field", "foo"s},

2✔

4003

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

4004

                                        {"non_queryable_field", "non queryable 1"s}}));

2✔

4005

        realm->commit_transaction();

2✔

4006

1✔

4007

        create_user_and_log_in(harness->app());

2✔

4008

        SyncTestFile target_config(harness->app()->current_user(), harness->schema(), SyncConfig::FLXSyncEnabled{});

2✔

4009

1✔

4010

        REQUIRE_THROWS(realm->convert(target_config));

2✔

4011

2✔

4012

6✔

4013

    // Add new sections before this

6✔

4014

    SECTION("teardown") {

12✔

4015

        harness->app()->sync_manager()->wait_for_sessions_to_terminate();

2✔

4016

        harness.reset();

2✔

4017

2✔

4018

12✔

4020

TEST_CASE("flx: compensating write errors get re-sent across sessions", "[sync][flx][compensating write][baas]") {

2✔

4021

    AppCreateConfig::ServiceRole role;

2✔

4022

    role.name = "compensating_write_perms";

2✔

4023

1✔

4024

    AppCreateConfig::ServiceRoleDocumentFilters doc_filters;

2✔

4025

    doc_filters.read = true;

2✔

4026

    doc_filters.write =

2✔

4027

        nlohmann::json{{"queryable_str_field", nlohmann::json{{"$in", nlohmann::json::array({"foo", "bar"})}}}};

2✔

4028

    role.document_filters = doc_filters;

2✔

4029

1✔

4030

    role.insert_filter = true;

2✔

4031

    role.delete_filter = true;

2✔

4032

    role.read = true;

2✔

4033

    role.write = true;

2✔

4034

    FLXSyncTestHarness::ServerSchema server_schema{

2✔

4035

        g_simple_embedded_obj_schema, {"queryable_str_field", "queryable_int_field"}, {role}};

2✔

4036

    FLXSyncTestHarness::Config harness_config("flx_bad_query", server_schema);

2✔

4037

    harness_config.reconnect_mode = ReconnectMode::testing;

2✔

4038

    FLXSyncTestHarness harness(std::move(harness_config));

2✔

4039

1✔

4040

    auto test_obj_id_1 = ObjectId::gen();

2✔

4041

    auto test_obj_id_2 = ObjectId::gen();

2✔

4042

1✔

4043

    create_user_and_log_in(harness.app());

2✔

4044

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

4045

1✔

4046

2✔

4047

        auto error_received_pf = util::make_promise_future<void>();

2✔

4048

        config.sync_config->on_sync_client_event_hook =

2✔

4049

            [promise = util::CopyablePromiseHolder(std::move(error_received_pf.promise))](

2✔

4050

                std::weak_ptr<SyncSession> weak_session, const SyncClientHookData& data) mutable {

30✔

4051

                if (data.event != SyncClientHookEvent::ErrorMessageReceived) {

30✔

4052

                    return SyncClientHookAction::NoAction;

28✔

4053

28✔

4054

                auto session = weak_session.lock();

2✔

4056

1✔

4057

                auto error_code = sync::ProtocolError(data.error_info->raw_error_code);

2✔

4058

1✔

4059

                if (error_code == sync::ProtocolError::initial_sync_not_completed) {

2✔

4062

1✔

4065

                promise.get_promise().emplace_value();

2✔

4066

1✔

4067

                return SyncClientHookAction::TriggerReconnect;

2✔

4068

};

2✔

4069

1✔

4070

        auto realm = Realm::get_shared_realm(config);

2✔

4071

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

4072

        auto queryable_str_field = table->get_column_key("queryable_str_field");

2✔

4073

        auto new_query = realm->get_latest_subscription_set().make_mutable_copy();

2✔

4074

        new_query.insert_or_assign(Query(table).equal(queryable_str_field, "bizz"));

2✔

4075

        std::move(new_query).commit();

2✔

4076

1✔

4077

        wait_for_upload(*realm);

2✔

4078

        wait_for_download(*realm);

2✔

4079

1✔

4080

        CppContext c(realm);

2✔

4081

        realm->begin_transaction();

2✔

4082

        Object::create(c, realm, "TopLevel",

2✔

4083

                       util::Any(AnyDict{

2✔

4084

                           {"_id", test_obj_id_1},

2✔

4085

                           {"queryable_str_field", std::string{"foo"}},

2✔

4086

                       }));

2✔

4087

        realm->commit_transaction();

2✔

4088

1✔

4089

        realm->begin_transaction();

2✔

4090

        Object::create(c, realm, "TopLevel",

2✔

4091

                       util::Any(AnyDict{

2✔

4092

                           {"_id", test_obj_id_2},

2✔

4093

                           {"queryable_str_field", std::string{"baz"}},

2✔

4094

                       }));

2✔

4095

        realm->commit_transaction();

2✔

4096

1✔

4097

        error_received_pf.future.get();

2✔

4098

        realm->sync_session()->shutdown_and_wait();

2✔

4099

        config.sync_config->on_sync_client_event_hook = {};

2✔

4100

2✔

4101

1✔

4102

    _impl::RealmCoordinator::clear_all_caches();

2✔

4103

1✔

4104

    std::mutex errors_mutex;

2✔

4105

    std::condition_variable new_compensating_write;

2✔

4106

    std::vector<std::pair<ObjectId, sync::version_type>> error_to_download_version;

2✔

4107

    std::vector<sync::CompensatingWriteErrorInfo> compensating_writes;

2✔

4108

    sync::version_type download_version;

2✔

4109

1✔

4110

    config.sync_config->on_sync_client_event_hook = [&](std::weak_ptr<SyncSession> weak_session,

2✔

4111

                                                        const SyncClientHookData& data) mutable {

16✔

4112

        auto session = weak_session.lock();

16✔

4113

        if (!session) {

16✔

4116

8✔

4117

        if (data.event != SyncClientHookEvent::ErrorMessageReceived) {

16✔

4118

            if (data.event == SyncClientHookEvent::DownloadMessageReceived) {

12✔

4119

                download_version = data.progress.download.server_version;

4✔

4120

4✔

4121

6✔

4122

            return SyncClientHookAction::NoAction;

12✔

4123

12✔

4124

2✔

4125

        auto error_code = sync::ProtocolError(data.error_info->raw_error_code);

4✔

4128

        std::lock_guard<std::mutex> lk(errors_mutex);

4✔

4129

        for (const auto& compensating_write : data.error_info->compensating_writes) {

4✔

4130

            error_to_download_version.emplace_back(compensating_write.primary_key.get_object_id(),

4✔

4131

                                                   *data.error_info->compensating_write_server_version);

4✔

4132

4✔

4133

2✔

4134

        return SyncClientHookAction::NoAction;

4✔

4135

};

4✔

4136

1✔

4137

    config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) {

4✔

4138

        std::unique_lock<std::mutex> lk(errors_mutex);

4✔

4140

        for (const auto& compensating_write : error.compensating_writes_info) {

4✔

4141

            auto tracked_error = std::find_if(error_to_download_version.begin(), error_to_download_version.end(),

4✔

4142

                                              [&](const auto& pair) {

6✔

4143

                                                  return pair.first == compensating_write.primary_key.get_object_id();

6✔

4144

});

6✔

4147

            compensating_writes.push_back(compensating_write);

4✔

4148

4✔

4149

        new_compensating_write.notify_one();

4✔

4150

};

4✔

4151

1✔

4152

    auto realm = Realm::get_shared_realm(config);

2✔

4153

1✔

4154

    wait_for_upload(*realm);

2✔

4155

    wait_for_download(*realm);

2✔

4156

1✔

4157

    std::unique_lock<std::mutex> lk(errors_mutex);

2✔

4158

    new_compensating_write.wait_for(lk, std::chrono::seconds(30), [&] {

2✔

4159

        return compensating_writes.size() == 2;

2✔

4160

});

2✔

4161

1✔

4163

    auto& write_info = compensating_writes[0];

2✔

4167

    CHECK_THAT(write_info.reason, Catch::Matchers::ContainsSubstring("object is outside of the current query view"));

2✔

4168

1✔

4169

    write_info = compensating_writes[1];

2✔

4174

            util::format("write to ObjectID(\"%1\") in table \"TopLevel\" not allowed", test_obj_id_2));

2✔

4175

    auto top_level_table = realm->read_group().get_table("class_TopLevel");

2✔

4177

2✔

4179

TEST_CASE("flx: bootstrap changesets are applied continuously", "[sync][flx][bootstrap][baas]") {

2✔

4180

    FLXSyncTestHarness harness("flx_bootstrap_ordering", {g_large_array_schema, {"queryable_int_field"}});

2✔

4181

    fill_large_array_schema(harness);

2✔

4182

1✔

4183

    std::unique_ptr<std::thread> th;

2✔

4184

    sync::version_type user_commit_version = UINT_FAST64_MAX;

2✔

4185

    sync::version_type bootstrap_version = UINT_FAST64_MAX;

2✔

4186

    SharedRealm realm;

2✔

4187

    std::condition_variable cv;

2✔

4188

    std::mutex mutex;

2✔

4189

    bool allow_to_commit = false;

2✔

4190

1✔

4191

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

4192

    auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

4193

    auto shared_promise = std::make_shared<util::Promise<void>>(std::move(interrupted_promise));

2✔

4194

    config.sync_config->on_sync_client_event_hook =

2✔

4195

        [promise = std::move(shared_promise), &th, &realm, &user_commit_version, &bootstrap_version, &cv, &mutex,

2✔

4196

         &allow_to_commit](std::weak_ptr<SyncSession> weak_session, const SyncClientHookData& data) {

68✔

4197

            if (data.query_version == 0) {

68✔

4198

                return SyncClientHookAction::NoAction;

18✔

4199

18✔

4200

            if (data.event != SyncClientHookEvent::DownloadMessageIntegrated) {

50✔

4201

                return SyncClientHookAction::NoAction;

38✔

4202

38✔

4203

            auto session = weak_session.lock();

12✔

4204

            if (!session) {

12✔

4207

            if (data.batch_state != sync::DownloadBatchState::MoreToCome) {

12✔

4208

                // Read version after bootstrap is done.

1✔

4209

                auto db = TestHelper::get_db(realm);

2✔

4210

                ReadTransaction rt(db);

2✔

4211

                bootstrap_version = rt.get_version();

2✔

4212

2✔

4213

                    std::lock_guard<std::mutex> lock(mutex);

2✔

4214

                    allow_to_commit = true;

2✔

4215

2✔

4216

                cv.notify_one();

2✔

4217

                session->force_close();

2✔

4218

                promise->emplace_value();

2✔

4219

                return SyncClientHookAction::NoAction;

2✔

4220

2✔

4221

5✔

4222

            if (th) {

10✔

4223

                return SyncClientHookAction::NoAction;

8✔

4224

8✔

4225

1✔

4226

            auto func = [&] {

2✔

4227

                // Attempt to commit a local change after the first bootstrap batch was committed.

1✔

4228

                auto db = TestHelper::get_db(realm);

2✔

4229

                WriteTransaction wt(db);

2✔

4230

                TableRef table = wt.get_table("class_TopLevel");

2✔

4231

                table->create_object_with_primary_key(ObjectId::gen());

2✔

4232

2✔

4233

                    std::unique_lock<std::mutex> lock(mutex);

2✔

4234

                    // Wait to commit until we read the final bootstrap version.

1✔

4235

                    cv.wait(lock, [&] {

2✔

4236

                        return allow_to_commit;

2✔

4237

});

2✔

4238

2✔

4239

                user_commit_version = wt.commit();

2✔

4240

};

2✔

4241

            th = std::make_unique<std::thread>(std::move(func));

2✔

4242

1✔

4243

            return SyncClientHookAction::NoAction;

2✔

4244

};

2✔

4245

1✔

4246

    realm = Realm::get_shared_realm(config);

2✔

4247

    auto table = realm->read_group().get_table("class_TopLevel");

2✔

4248

    Query query(table);

2✔

4249

2✔

4250

        auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

4251

        new_subs.insert_or_assign(query);

2✔

4252

        new_subs.commit();

2✔

4253

2✔

4254

    interrupted.get();

2✔

4255

    th->join();

2✔

4256

1✔

4257

    // The user commit is the last one.

1✔

4259

2✔

4262

          "[sync][flx][bootstrap][async open][baas]") {

12✔

4263

    FLXSyncTestHarness harness("flx_bootstrap_and_subscribe");

12✔

4264

    auto foo_obj_id = ObjectId::gen();

12✔

4265

    harness.load_initial_data([&](SharedRealm realm) {

12✔

4266

        CppContext c(realm);

12✔

4267

        Object::create(c, realm, "TopLevel",

12✔

4268

                       std::any(AnyDict{{"_id", foo_obj_id},

12✔

4269

                                        {"queryable_str_field", "foo"s},

12✔

4270

                                        {"queryable_int_field", static_cast<int64_t>(5)},

12✔

4271

                                        {"non_queryable_field", "created as initial data seed"s}}));

12✔

4272

});

12✔

4273

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

12✔

4274

6✔

4275

    std::atomic<bool> subscription_invoked = false;

12✔

4276

    auto subscription_pf = util::make_promise_future<bool>();

12✔

4277

    // create a subscription to commit when realm is open for the first time or asked to rerun on open

6✔

4278

    auto init_subscription_callback_with_promise =

12✔

4279

        [&, promise_holder = util::CopyablePromiseHolder(std::move(subscription_pf.promise))](

12✔

4280

            std::shared_ptr<Realm> realm) mutable {

10✔

4282

            auto table = realm->read_group().get_table("class_TopLevel");

8✔

4283

            Query query(table);

8✔

4284

            auto subscription = realm->get_latest_subscription_set();

8✔

4285

            auto mutable_subscription = subscription.make_mutable_copy();

8✔

4286

            mutable_subscription.insert_or_assign(query);

8✔

4287

            auto promise = promise_holder.get_promise();

8✔

4288

            mutable_subscription.commit();

8✔

4289

            subscription_invoked = true;

8✔

4290

            promise.emplace_value(true);

8✔

4291

};

8✔

4292

    // verify that the subscription has changed the database

6✔

4293

    auto verify_subscription = [](SharedRealm realm) {

12✔

4295

        auto table_ref = realm->read_group().get_table("class_TopLevel");

12✔

4303

        auto str_col = table_ref->get_column_key("queryable_str_field");

12✔

4305

        return true;

12✔

4306

};

12✔

4307

6✔

4308

    SECTION("Sync open") {

12✔

4309

        // sync open with subscription callback. Subscription will be run, since this is the first time that realm is

1✔

4310

        // opened

1✔

4311

        subscription_invoked = false;

2✔

4312

        config.sync_config->subscription_initializer = init_subscription_callback_with_promise;

2✔

4313

        auto realm = Realm::get_shared_realm(config);

2✔

4315

        auto sb = realm->get_latest_subscription_set();

2✔

4316

        auto future = sb.get_state_change_notification(realm::sync::SubscriptionSet::State::Complete);

2✔

4317

        auto state = future.get();

2✔

4319

        realm->refresh(); // refresh is needed otherwise table_ref->size() would be 0

2✔

4321

2✔

4322

6✔

4323

    SECTION("Sync Open + Async Open") {

12✔

4324

2✔

4325

            subscription_invoked = false;

2✔

4326

            config.sync_config->subscription_initializer = init_subscription_callback_with_promise;

2✔

4327

            auto realm = Realm::get_shared_realm(config);

2✔

4329

            auto sb = realm->get_latest_subscription_set();

2✔

4330

            auto future = sb.get_state_change_notification(realm::sync::SubscriptionSet::State::Complete);

2✔

4331

            auto state = future.get();

2✔

4333

            realm->refresh(); // refresh is needed otherwise table_ref->size() would be 0

2✔

4335

2✔

4336

2✔

4337

            auto subscription_pf_async = util::make_promise_future<bool>();

2✔

4338

            auto init_subscription_asyc_callback =

2✔

4339

                [promise_holder_async = util::CopyablePromiseHolder(std::move(subscription_pf_async.promise))](

2✔

4340

                    std::shared_ptr<Realm> realm) mutable {

2✔

4342

                    auto table = realm->read_group().get_table("class_TopLevel");

2✔

4343

                    Query query(table);

2✔

4344

                    auto subscription = realm->get_latest_subscription_set();

2✔

4345

                    auto mutable_subscription = subscription.make_mutable_copy();

2✔

4346

                    mutable_subscription.insert_or_assign(query);

2✔

4347

                    auto promise = promise_holder_async.get_promise();

2✔

4348

                    mutable_subscription.commit();

2✔

4349

                    promise.emplace_value(true);

2✔

4350

};

2✔

4351

            auto open_realm_pf = util::make_promise_future<bool>();

2✔

4352

            auto open_realm_completed_callback =

2✔

4353

                [&, promise_holder = util::CopyablePromiseHolder(std::move(open_realm_pf.promise))](

2✔

4354

                    ThreadSafeReference ref, std::exception_ptr err) mutable {

2✔

4355

                    auto promise = promise_holder.get_promise();

2✔

4356

                    if (err)

2✔

4358

                    else

2✔

4359

                        promise.emplace_value(verify_subscription(Realm::get_shared_realm(std::move(ref))));

2✔

4360

};

2✔

4361

1✔

4362

            config.sync_config->subscription_initializer = init_subscription_asyc_callback;

2✔

4363

            config.sync_config->rerun_init_subscription_on_open = true;

2✔

4364

            auto async_open = Realm::get_synchronized_realm(config);

2✔

4365

            async_open->start(open_realm_completed_callback);

2✔

4368

            config.sync_config->rerun_init_subscription_on_open = false;

2✔

4369

            auto realm = Realm::get_shared_realm(config);

2✔

4372

2✔

4373

2✔

4374

6✔

4375

    SECTION("Async open") {

12✔

4376

        SECTION("Initial async open with no rerun on open set") {

8✔

4377

            // subscription will be run since this is the first time we are opening the realm file.

2✔

4378

            auto open_realm_pf = util::make_promise_future<bool>();

4✔

4379

            auto open_realm_completed_callback =

4✔

4380

                [&, promise_holder = util::CopyablePromiseHolder(std::move(open_realm_pf.promise))](

4✔

4381

                    ThreadSafeReference ref, std::exception_ptr err) mutable {

4✔

4382

                    auto promise = promise_holder.get_promise();

4✔

4383

                    if (err)

4✔

4385

                    else

4✔

4386

                        promise.emplace_value(verify_subscription(Realm::get_shared_realm(std::move(ref))));

4✔

4387

};

4✔

4388

2✔

4389

            config.sync_config->subscription_initializer = init_subscription_callback_with_promise;

4✔

4390

            auto async_open = Realm::get_synchronized_realm(config);

4✔

4391

            async_open->start(open_realm_completed_callback);

4✔

4394

2✔

4395

            SECTION("rerun on open = false. Subscription not run") {

4✔

4396

                subscription_invoked = false;

2✔

4397

                auto async_open = Realm::get_synchronized_realm(config);

2✔

4398

                auto open_realm_pf = util::make_promise_future<bool>();

2✔

4399

                auto open_realm_completed_callback =

2✔

4400

                    [promise_holder = util::CopyablePromiseHolder(std::move(open_realm_pf.promise))](

2✔

4401

                        ThreadSafeReference, std::exception_ptr) mutable {

2✔

4402

                        // no need to verify if the subscription has changed the db, since it has not run as we test

1✔

4403

                        // below

1✔

4404

                        promise_holder.get_promise().emplace_value(true);

2✔

4405

};

2✔

4406

                async_open->start(open_realm_completed_callback);

2✔

4409

2✔

4410

2✔

4411

            SECTION("rerun on open = true. Subscription not run cause realm already opened once") {

4✔

4412

                subscription_invoked = false;

2✔

4413

                auto realm = Realm::get_shared_realm(config);

2✔

4414

                auto init_subscription = [&subscription_invoked](std::shared_ptr<Realm> realm) mutable {

1✔

4424

                config.sync_config->rerun_init_subscription_on_open = true;

2✔

4425

                config.sync_config->subscription_initializer = init_subscription;

2✔

4426

                auto async_open = Realm::get_synchronized_realm(config);

2✔

4427

                auto open_realm_pf = util::make_promise_future<bool>();

2✔

4428

                auto open_realm_completed_callback =

2✔

4429

                    [promise_holder = util::CopyablePromiseHolder(std::move(open_realm_pf.promise))](

2✔

4430

                        ThreadSafeReference, std::exception_ptr) mutable {

2✔

4431

                        // no need to verify if the subscription has changed the db, since it has not run as we test

1✔

4432

                        // below

1✔

4433

                        promise_holder.get_promise().emplace_value(true);

2✔

4434

};

2✔

4435

                async_open->start(open_realm_completed_callback);

2✔

4440

2✔

4441

4✔

4442

4✔

4443

        SECTION("rerun on open set for multiple async open tasks (subscription runs only once)") {

8✔

4444

            auto init_subscription = [](std::shared_ptr<Realm> realm) mutable {

8✔

4446

                auto table = realm->read_group().get_table("class_TopLevel");

8✔

4447

                Query query(table);

8✔

4448

                auto subscription = realm->get_latest_subscription_set();

8✔

4449

                auto mutable_subscription = subscription.make_mutable_copy();

8✔

4450

                mutable_subscription.insert_or_assign(query);

8✔

4451

                mutable_subscription.commit();

8✔

4452

};

8✔

4453

2✔

4454

            auto open_task1_pf = util::make_promise_future<SharedRealm>();

4✔

4455

            auto open_task2_pf = util::make_promise_future<SharedRealm>();

4✔

4456

            auto open_callback1 = [promise_holder = util::CopyablePromiseHolder(std::move(open_task1_pf.promise))](

4✔

4457

                                      ThreadSafeReference ref, std::exception_ptr err) mutable {

4✔

4459

                auto realm = Realm::get_shared_realm(std::move(ref));

4✔

4461

                promise_holder.get_promise().emplace_value(realm);

4✔

4462

};

4✔

4463

            auto open_callback2 = [promise_holder = util::CopyablePromiseHolder(std::move(open_task2_pf.promise))](

4✔

4464

                                      ThreadSafeReference ref, std::exception_ptr err) mutable {

4✔

4466

                auto realm = Realm::get_shared_realm(std::move(ref));

4✔

4468

                promise_holder.get_promise().emplace_value(realm);

4✔

4469

};

4✔

4470

2✔

4471

            config.sync_config->rerun_init_subscription_on_open = true;

4✔

4472

            config.sync_config->subscription_initializer = init_subscription;

4✔

4473

2✔

4474

            SECTION("Realm was already created, but we want to rerun on first open using multiple tasks") {

4✔

4475

2✔

4476

                    subscription_invoked = false;

2✔

4477

                    auto realm = Realm::get_shared_realm(config);

2✔

4478

                    auto sb = realm->get_latest_subscription_set();

2✔

4479

                    auto future = sb.get_state_change_notification(realm::sync::SubscriptionSet::State::Complete);

2✔

4480

                    auto state = future.get();

2✔

4482

                    realm->refresh(); // refresh is needed otherwise table_ref->size() would be 0

2✔

4486

2✔

4487

1✔

4488

                auto async_open_task1 = Realm::get_synchronized_realm(config);

2✔

4489

                auto async_open_task2 = Realm::get_synchronized_realm(config);

2✔

4490

                async_open_task1->start(open_callback1);

2✔

4491

                async_open_task2->start(open_callback2);

2✔

4492

1✔

4493

                auto realm1 = open_task1_pf.future.get();

2✔

4494

                auto realm2 = open_task2_pf.future.get();

2✔

4495

1✔

4496

                const auto version_expected = 2;

2✔

4497

                auto r1_latest = realm1->get_latest_subscription_set().version();

2✔

4498

                auto r1_active = realm1->get_active_subscription_set().version();

2✔

4503

2✔

4504

            SECTION("First time realm is created but opened via open async. Both tasks could run the subscription") {

4✔

4505

                auto async_open_task1 = Realm::get_synchronized_realm(config);

2✔

4506

                auto async_open_task2 = Realm::get_synchronized_realm(config);

2✔

4507

                async_open_task1->start(open_callback1);

2✔

4508

                async_open_task2->start(open_callback2);

2✔

4509

                auto realm1 = open_task1_pf.future.get();

2✔

4510

                auto realm2 = open_task2_pf.future.get();

2✔

4511

1✔

4512

                auto r1_latest = realm1->get_latest_subscription_set().version();

2✔

4513

                auto r1_active = realm1->get_active_subscription_set().version();

2✔

4516

                // the callback may be run twice, if task1 is the first task to open realm

1✔

4517

                // but it is scheduled after tasks2, which have opened realm later but

1✔

4518

                // by the time it runs, subscription version is equal to 0 (realm creation).

1✔

4519

                // This can only happen the first time that realm is created. All the other times

1✔

4520

                // the init_sb callback is guaranteed to run once.

1✔

4525

2✔

4526

4✔

4527

8✔

4528

12✔

4529

TEST_CASE("flx sync: Client reset during async open", "[sync][flx][client reset][async open][baas]") {

2✔

4530

    FLXSyncTestHarness harness("flx_bootstrap_reset");

2✔

4531

    auto foo_obj_id = ObjectId::gen();

2✔

4532

    std::atomic<bool> subscription_invoked = false;

2✔

4533

    harness.load_initial_data([&](SharedRealm realm) {

2✔

4534

        CppContext c(realm);

2✔

4535

        Object::create(c, realm, "TopLevel",

2✔

4536

                       std::any(AnyDict{{"_id", foo_obj_id},

2✔

4537

                                        {"queryable_str_field", "foo"s},

2✔

4538

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

4539

                                        {"non_queryable_field", "created as initial data seed"s}}));

2✔

4540

});

2✔

4541

    SyncTestFile realm_config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

4542

1✔

4543

    auto subscription_callback = [&](std::shared_ptr<Realm> realm) {

2✔

4545

        auto table = realm->read_group().get_table("class_TopLevel");

2✔

4546

        Query query(table);

2✔

4547

        auto subscription = realm->get_latest_subscription_set();

2✔

4548

        auto mutable_subscription = subscription.make_mutable_copy();

2✔

4549

        mutable_subscription.insert_or_assign(query);

2✔

4550

        subscription_invoked = true;

2✔

4551

        mutable_subscription.commit();

2✔

4552

};

2✔

4553

1✔

4554

    realm_config.sync_config->client_resync_mode = ClientResyncMode::Recover;

2✔

4555

    realm_config.sync_config->subscription_initializer = subscription_callback;

2✔

4556

1✔

4557

    bool client_reset_triggered = false;

2✔

4558

    realm_config.sync_config->on_sync_client_event_hook = [&](std::weak_ptr<SyncSession> weak_sess,

2✔

4559

                                                              const SyncClientHookData& event_data) {

54✔

4560

        auto sess = weak_sess.lock();

54✔

4561

        if (!sess) {

54✔

4564

        if (sess->path() != realm_config.path) {

54✔

4565

            return SyncClientHookAction::NoAction;

24✔

4566

24✔

4567

15✔

4568

        if (event_data.event != SyncClientHookEvent::DownloadMessageReceived) {

30✔

4569

            return SyncClientHookAction::NoAction;

24✔

4570

24✔

4571

3✔

4572

        if (client_reset_triggered) {

6✔

4573

            return SyncClientHookAction::NoAction;

4✔

4574

4✔

4575

1✔

4576

        client_reset_triggered = true;

2✔

4577

        reset_utils::trigger_client_reset(harness.session().app_session(), *sess);

2✔

4578

        return SyncClientHookAction::SuspendWithRetryableError;

2✔

4579

};

2✔

4580

1✔

4581

    auto before_callback_called = util::make_promise_future<void>();

2✔

4582

    realm_config.sync_config->notify_before_client_reset = [&](std::shared_ptr<Realm> realm) {

2✔

4584

        before_callback_called.promise.emplace_value();

2✔

4585

};

2✔

4586

1✔

4587

    auto after_callback_called = util::make_promise_future<void>();

2✔

4588

    realm_config.sync_config->notify_after_client_reset = [&](std::shared_ptr<Realm> realm, ThreadSafeReference,

2✔

4589

                                                              bool) {

2✔

4591

        after_callback_called.promise.emplace_value();

2✔

4592

};

2✔

4593

1✔

4594

    auto realm_task = Realm::get_synchronized_realm(realm_config);

2✔

4595

    auto realm_pf = util::make_promise_future<SharedRealm>();

2✔

4596

    realm_task->start([&](ThreadSafeReference ref, std::exception_ptr ex) {

2✔

4597

        auto& promise = realm_pf.promise;

2✔

4598

        try {

2✔

4599

            if (ex) {

2✔

4602

            promise.emplace_value(Realm::get_shared_realm(std::move(ref)));

2✔

4603

2✔

4604

        catch (...) {

1✔

4607

});

2✔

4608

    auto realm = realm_pf.future.get();

2✔

4609

    before_callback_called.future.get();

2✔

4610

    after_callback_called.future.get();

2✔

4612

2✔

4615

TEST_CASE("flx sync: resend pending subscriptions when reconnecting", "[sync][flx][baas]") {

2✔

4616

    FLXSyncTestHarness harness("flx_pending_subscriptions", {g_large_array_schema, {"queryable_int_field"}});

2✔

4617

1✔

4618

    std::vector<ObjectId> obj_ids_at_end = fill_large_array_schema(harness);

2✔

4619

    SyncTestFile interrupted_realm_config(harness.app()->current_user(), harness.schema(),

2✔

4620

                                          SyncConfig::FLXSyncEnabled{});

2✔

4621

1✔

4622

2✔

4623

        auto pf = util::make_promise_future<void>();

2✔

4624

        Realm::Config config = interrupted_realm_config;

2✔

4625

        config.sync_config = std::make_shared<SyncConfig>(*interrupted_realm_config.sync_config);

2✔

4626

        config.sync_config->on_sync_client_event_hook =

2✔

4627

            [promise = util::CopyablePromiseHolder(std::move(pf.promise))](std::weak_ptr<SyncSession> weak_session,

2✔

4628

                                                                           const SyncClientHookData& data) mutable {

68✔

4629

                if (data.event != SyncClientHookEvent::BootstrapMessageProcessed &&

68✔

4630

                    data.event != SyncClientHookEvent::BootstrapProcessed) {

61✔

4631

                    return SyncClientHookAction::NoAction;

50✔

4632

50✔

4633

                auto session = weak_session.lock();

18✔

4634

                if (!session) {

18✔

4637

                if (data.query_version != 1) {

18✔

4638

                    return SyncClientHookAction::NoAction;

4✔

4639

4✔

4640

7✔

4641

                // Commit a subscriptions set whenever a bootstrap message is received for query version 1.

7✔

4642

                if (data.event == SyncClientHookEvent::BootstrapMessageProcessed) {

14✔

4643

                    auto latest_subs = session->get_flx_subscription_store()->get_latest().make_mutable_copy();

12✔

4644

                    latest_subs.commit();

12✔

4645

                    return SyncClientHookAction::NoAction;

12✔

4646

12✔

4647

                // At least one subscription set was created.

1✔

4649

                promise.get_promise().emplace_value();

2✔

4650

                // Reconnect once query version 1 is bootstrapped.

1✔

4651

                return SyncClientHookAction::TriggerReconnect;

2✔

4652

};

2✔

4653

1✔

4654

        auto realm = Realm::get_shared_realm(config);

2✔

4655

2✔

4656

            auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

2✔

4657

            auto table = realm->read_group().get_table("class_TopLevel");

2✔

4658

            mut_subs.insert_or_assign(Query(table));

2✔

4659

            mut_subs.commit();

2✔

4660

2✔

4661

        pf.future.get();

2✔

4662

        realm->sync_session()->shutdown_and_wait();

2✔

4663

        realm->close();

2✔

4664

2✔

4665

1✔

4666

    _impl::RealmCoordinator::assert_no_open_realms();

2✔

4667

1✔

4668

    // Check at least one subscription set needs to be resent.

1✔

4669

2✔

4670

        DBOptions options;

2✔

4671

        options.encryption_key = test_util::crypt_key();

2✔

4672

        auto realm = DB::create(sync::make_client_replication(), interrupted_realm_config.path, options);

2✔

4673

        auto sub_store = sync::SubscriptionStore::create(realm);

2✔

4674

        auto version_info = sub_store->get_version_info();

2✔

4676

2✔

4677

1✔

4678

    // Resend the pending subscriptions.

1✔

4679

    auto realm = Realm::get_shared_realm(interrupted_realm_config);

2✔

4680

    wait_for_upload(*realm);

2✔

4681

    wait_for_download(*realm);

2✔

4682

2✔

4684

TEST_CASE("flx: fatal errors and session becoming inactive cancel pending waits", "[sync][flx][baas]") {

2✔

4685

    std::vector<ObjectSchema> schema{

2✔

4686

        {"TopLevel",

2✔

4687

2✔

4688

             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},

2✔

4689

             {"queryable_int_field", PropertyType::Int | PropertyType::Nullable},

2✔

4690

}},

2✔

4691

};

2✔

4692

1✔

4693

    FLXSyncTestHarness harness("flx_cancel_pending_waits", {schema, {"queryable_int_field"}});

2✔

4694

    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

4695

1✔

4696

    auto check_status = [](auto status) {

4✔

4698

        std::string reason = status.get_status().reason();

4✔

4699

        // Subscription notification is cancelled either because the sync session is inactive, or because a fatal

2✔

4700

        // error is received from the server.

2✔

4701

        if (reason.find("Sync session became inactive") == std::string::npos &&

4✔

4702

            reason.find("Invalid schema change (UPLOAD): non-breaking schema change: adding \"Int\" column at field "

3✔

4703

                        "\"other_col\" in schema \"TopLevel\", schema changes from clients are restricted when "

2✔

4704

                        "developer mode is disabled") == std::string::npos) {

1✔

4707

};

4✔

4708

1✔

4709

    auto create_subscription = [](auto realm) -> realm::sync::SubscriptionSet {

4✔

4710

        auto mut_subs = realm->get_latest_subscription_set().make_mutable_copy();

4✔

4711

        auto table = realm->read_group().get_table("class_TopLevel");

4✔

4712

        mut_subs.insert_or_assign(Query(table));

4✔

4713

        return mut_subs.commit();

4✔

4714

};

4✔

4715

1✔

4716

    auto [error_occured_promise, error_occurred] = util::make_promise_future<void>();

2✔

4717

    config.sync_config->error_handler = [promise = util::CopyablePromiseHolder(std::move(error_occured_promise))](

2✔

4718

                                            std::shared_ptr<SyncSession>, SyncError) mutable {

2✔

4719

        promise.get_promise().emplace_value();

2✔

4720

};

2✔

4721

1✔

4722

    auto realm = Realm::get_shared_realm(config);

2✔

4723

    wait_for_download(*realm);

2✔

4724

1✔

4725

    auto subs = create_subscription(realm);

2✔

4726

    auto subs_future = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete);

2✔

4727

1✔

4728

    realm->sync_session()->pause();

2✔

4729

    auto state = subs_future.get_no_throw();

2✔

4730

    check_status(state);

2✔

4731

1✔

4732

    auto [download_complete_promise, download_complete] = util::make_promise_future<void>();

2✔

4733

    realm->sync_session()->wait_for_upload_completion([promise = std::move(download_complete_promise)](auto) mutable {

2✔

4734

        promise.emplace_value();

2✔

4735

});

2✔

4736

    schema[0].persisted_properties.push_back({"other_col", PropertyType::Int | PropertyType::Nullable});

2✔

4737

    realm->update_schema(schema);

2✔

4738

1✔

4739

    subs = create_subscription(realm);

2✔

4740

    subs_future = subs.get_state_change_notification(sync::SubscriptionSet::State::Complete);

2✔

4741

1✔

4742

    harness.load_initial_data([&](SharedRealm realm) {

2✔

4743

        CppContext c(realm);

2✔

4744

        Object::create(c, realm, "TopLevel",

2✔

4745

                       std::any(AnyDict{{"_id", ObjectId::gen()},

2✔

4746

                                        {"queryable_int_field", static_cast<int64_t>(5)},

2✔

4747

                                        {"other_col", static_cast<int64_t>(42)}}));

2✔

4748

});

2✔

4749

1✔

4750

    realm->sync_session()->resume();

2✔

4751

    download_complete.get();

2✔

4752

    error_occurred.get();

2✔

4753

    state = subs_future.get_no_throw();

2✔

4754

    check_status(state);

2✔

4755

2✔

4757

TEST_CASE("flx: pause and resume bootstrapping at query version 0", "[sync][flx][baas]") {

2✔

4758

    FLXSyncTestHarness harness("flx_pause_resume_bootstrap");

2✔

4759

    SyncTestFile triggered_config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});

2✔

4760

    auto [interrupted_promise, interrupted] = util::make_promise_future<void>();

2✔

4761

    std::mutex download_message_mutex;

2✔

4762

    int download_message_integrated_count = 0;

2✔

4763

    triggered_config.sync_config->on_sync_client_event_hook =

2✔

4764

        [promise = util::CopyablePromiseHolder(std::move(interrupted_promise)), &download_message_integrated_count,

2✔

4765

         &download_message_mutex](std::weak_ptr<SyncSession> weak_sess, const SyncClientHookData& data) mutable {

22✔

4766

            auto sess = weak_sess.lock();

22✔

4767

            if (!sess || data.event != SyncClientHookEvent::DownloadMessageIntegrated) {

22✔

4768

                return SyncClientHookAction::NoAction;

18✔

4769

18✔

4770

2✔

4771

            std::lock_guard<std::mutex> lk(download_message_mutex);

4✔

4772

            // Pause and resume the first session after the bootstrap message is integrated.

2✔

4773

            if (download_message_integrated_count == 0) {

4✔

4774

                sess->pause();

2✔

4775

                sess->resume();

2✔

4776

2✔

4777

            // Complete the test when the second session integrates the empty download

1✔

4778

            // message it receives.

1✔

4779

            else {

2✔

4780

                promise.get_promise().emplace_value();

2✔

4781

2✔

4782

            ++download_message_integrated_count;

4✔

4783

            return SyncClientHookAction::NoAction;

4✔

4784

};

4✔

4785

    auto realm = Realm::get_shared_realm(triggered_config);

2✔

4786

    interrupted.get();

2✔

4787

    std::lock_guard<std::mutex> lk(download_message_mutex);

2✔

4789

    auto active_sub_set = realm->sync_session()->get_flx_subscription_store()->get_active();

2✔

4792

2✔

realm / realm-core / 2214

Source File
Press 'n' to go to next uncovered line, 'b' for previous

realm / realm-core / 2214

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous