Sync schema migrations (#7239)

* Add schema version to flexible sync client BIND message (#6863)

* Send schema version as part of BIND messages

* Add pause_async to SyncSession (#6845)

* Add async shutdown of SyncSession

* SyncSession::shutdown returns a Future instead of passing callbacks around

* Rename shutdown() to pause_async(). Hide it from the public API.

* Remove unneeded header import

* Upload data and migrate schema (#6944)

* Handle client resets during schema migrations + integration tests (#7106)

* Add option to use draft deployments when creating a schema for baas tests

* Track sync schema migration between sessions + minor refactoring

* Integration tests

* Bump sync protocol version to 11

* Update baas commit for evergreen tests

* Allow resetting schema version to zero

* Improve changelog

CHANGELOG.md

@@ -11,6 +11,7 @@
 
 ### Breaking changes
 * `App::get_uncached_app(...)` and `App::get_shared_app(...)` have been replaced by `App::get_app(App::CacheMode, ...)`. The App constructor is now enforced to be unusable, use `App::get_app()` instead. ([#7237](https://github.com/realm/realm-core/issues/7237))
+* The schema version field in the Realm config had no use for the flexible sync Realms previously. It is now being used for the upcoming Sync Schema Migrations feature. If it was set to a value other than zero, the application will start receiving an error from the server. Data synchronization will be stopped until the Realm is opened with schema version zero. (PR [#7239](https://github.com/realm/realm-core/pull/7239))
 
 ### Compatibility
 * Fileformat: Generates files with format v23. Reads and automatically upgrade from fileformat v5.
@@ -19,6 +20,10 @@
 
 ### Internals
 * Add support for chunked transfer encoding when using `HTTPParser`.
+* Bump the sync protocol to v11. The new protocol version comes with the following changes:
+  - JSON_ERROR server message contains the previous schema version
+  - Flexible sync BIND client message contains the current schema version
+* Add BAAS admin API to create new schema versions (drafts can be used to deploy all changes at once)
 
 ----------------------------------------------
 

Package.swift

@@ -121,6 +121,7 @@ let notSyncServerSources: [String] = [
     "realm/sync/noinst/pending_bootstrap_store.cpp",
     "realm/sync/noinst/protocol_codec.cpp",
     "realm/sync/noinst/sync_metadata_schema.cpp",
+    "realm/sync/noinst/sync_schema_migration.cpp",
     "realm/sync/object_id.cpp",
     "realm/sync/protocol.cpp",
     "realm/sync/subscriptions.cpp",

dependencies.list

@@ -3,5 +3,5 @@ VERSION=13.25.1
 OPENSSL_VERSION=3.0.8
 ZLIB_VERSION=1.2.13
 # https://github.com/10gen/baas/commits
-# 300ef is 2023 Dec 15
-BAAS_VERSION=300efb0604a88f1f36899bee0f42b34826b9b65f
+# 5087f is 2024 Jan 13
+BAAS_VERSION=5087ffd5a0e4975e625f0fbcceed23107f611055

src/realm/error_codes.cpp

@@ -62,6 +62,7 @@ ErrorCategory ErrorCodes::error_categories(Error code)
         case SyncUserMismatch:
         case SyncWriteNotAllowed:
         case SyncLocalClockBeforeEpoch:
+        case SyncSchemaMigrationError:
             return ErrorCategory().set(ErrorCategory::runtime_error).set(ErrorCategory::sync_error);
 
         case SyncConnectFailed:

src/realm/error_codes.h

@@ -84,6 +84,7 @@ typedef enum realm_errno {
     RLM_ERR_WRONG_SYNC_TYPE = 1043,
     RLM_ERR_SYNC_WRITE_NOT_ALLOWED = 1044,
     RLM_ERR_SYNC_LOCAL_CLOCK_BEFORE_EPOCH = 1045,
+    RLM_ERR_SYNC_SCHEMA_MIGRATION_ERROR = 1046,
 
     RLM_ERR_SYSTEM_ERROR = 1999,
 
@@ -263,6 +264,8 @@ typedef enum realm_sync_errno_session {
     RLM_SYNC_ERR_SESSION_MIGRATE_TO_FLX = 232,
     RLM_SYNC_ERR_SESSION_BAD_PROGRESS = 233,
     RLM_SYNC_ERR_SESSION_REVERT_TO_PBS = 234,
+    RLM_SYNC_ERR_SESSION_BAD_SCHEMA_VERSION = 235,
+    RLM_SYNC_ERR_SESSION_SCHEMA_VERSION_CHANGED = 236,
     // Error code 299 is reserved as an "unknown session error" in tests
 } realm_sync_errno_session_e;
 

src/realm/error_codes.hpp

@@ -127,6 +127,7 @@ class ErrorCodes {
         WrongSyncType = RLM_ERR_WRONG_SYNC_TYPE,
         SyncWriteNotAllowed = RLM_ERR_SYNC_WRITE_NOT_ALLOWED,
         SyncLocalClockBeforeEpoch = RLM_ERR_SYNC_LOCAL_CLOCK_BEFORE_EPOCH,
+        SyncSchemaMigrationError = RLM_ERR_SYNC_SCHEMA_MIGRATION_ERROR,
 
         SystemError = RLM_ERR_SYSTEM_ERROR,
 

src/realm/exceptions.cpp

@@ -166,6 +166,7 @@ InvalidTableRef::~InvalidTableRef() noexcept = default;
 SerializationError::~SerializationError() noexcept = default;
 NotImplemented::~NotImplemented() noexcept = default;
 MigrationFailed::~MigrationFailed() noexcept = default;
+SyncSchemaMigrationFailed::~SyncSchemaMigrationFailed() noexcept = default;
 ObjectAlreadyExists::~ObjectAlreadyExists() noexcept = default;
 CrossTableLinkTarget::~CrossTableLinkTarget() noexcept = default;
 SystemError::~SystemError() noexcept = default;

src/realm/exceptions.hpp

@@ -306,6 +306,14 @@ struct MigrationFailed : LogicError {
     ~MigrationFailed() noexcept override;
 };
 
+struct SyncSchemaMigrationFailed : LogicError {
+    SyncSchemaMigrationFailed(std::string_view msg)
+        : LogicError(ErrorCodes::SyncSchemaMigrationError, msg)
+    {
+    }
+    ~SyncSchemaMigrationFailed() noexcept override;
+};
+
 struct ObjectAlreadyExists : RuntimeError {
     template <class T, class U>
     ObjectAlreadyExists(const U& object_type, T pk_val)

src/realm/object-store/object_store.cpp

@@ -850,19 +850,20 @@ static void apply_post_migration_changes(Group& group, std::vector<SchemaChange>
 void ObjectStore::apply_schema_changes(Transaction& group, uint64_t schema_version, Schema& target_schema,
                                        uint64_t target_schema_version, SchemaMode mode,
                                        std::vector<SchemaChange> const& changes, bool handle_automatically_backlinks,
-                                       std::function<void()> migration_function)
+                                       std::function<void()> migration_function,
+                                       bool set_schema_version_on_version_decrease)
 {
     create_metadata_tables(group);
 
     if (mode == SchemaMode::AdditiveDiscovered || mode == SchemaMode::AdditiveExplicit) {
-        bool target_schema_is_newer =
-            (schema_version < target_schema_version || schema_version == ObjectStore::NotVersioned);
+        bool set_schema = (schema_version < target_schema_version || schema_version == ObjectStore::NotVersioned ||
+                           set_schema_version_on_version_decrease);
 
         // With sync v2.x, indexes are no longer synced, so there's no reason to avoid creating them.
         bool update_indexes = true;
         apply_additive_changes(group, changes, update_indexes);
 
-        if (target_schema_is_newer)
+        if (set_schema)
             set_schema_version(group, target_schema_version);
 
         set_schema_keys(group, target_schema);

src/realm/object-store/object_store.hpp

@@ -86,7 +86,8 @@ class ObjectStore {
     static void apply_schema_changes(Transaction& group, uint64_t schema_version, Schema& target_schema,
                                      uint64_t target_schema_version, SchemaMode mode,
                                      std::vector<SchemaChange> const& changes, bool handle_automatically_backlinks,
-                                     std::function<void()> migration_function = {});
+                                     std::function<void()> migration_function = {},
+                                     bool save_schema_version_on_version_decrease = false);
 
     static void apply_additive_changes(Group&, std::vector<SchemaChange> const&, bool update_indexes);
 

src/realm/object-store/shared_realm.cpp

@@ -293,8 +293,7 @@ bool Realm::schema_change_needs_write_transaction(Schema& schema, std::vector<Sc
 
     switch (m_config.schema_mode) {
         case SchemaMode::Automatic:
-            if (version < m_schema_version && m_schema_version != ObjectStore::NotVersioned)
-                throw InvalidSchemaVersionException(m_schema_version, version, false);
+            verify_schema_version_not_decreasing(version);
             return true;
 
         case SchemaMode::Immutable:
@@ -324,8 +323,7 @@ bool Realm::schema_change_needs_write_transaction(Schema& schema, std::vector<Sc
         }
 
         case SchemaMode::Manual:
-            if (version < m_schema_version && m_schema_version != ObjectStore::NotVersioned)
-                throw InvalidSchemaVersionException(m_schema_version, version, false);
+            verify_schema_version_not_decreasing(version);
             if (version == m_schema_version) {
                 ObjectStore::verify_no_changes_required(changes);
                 REALM_UNREACHABLE(); // changes is non-empty so above line always throws
@@ -335,6 +333,17 @@ bool Realm::schema_change_needs_write_transaction(Schema& schema, std::vector<Sc
     REALM_COMPILER_HINT_UNREACHABLE();
 }
 
+// Schema version is not allowed to decrease for local and pbs realms.
+void Realm::verify_schema_version_not_decreasing(uint64_t version)
+{
+#if REALM_ENABLE_SYNC
+    if (m_config.sync_config && m_config.sync_config->flx_sync_requested)
+        return;
+#endif
+    if (version < m_schema_version && m_schema_version != ObjectStore::NotVersioned)
+        throw InvalidSchemaVersionException(m_schema_version, version, false);
+}
+
 Schema Realm::get_full_schema()
 {
     if (!m_config.immutable())
@@ -483,6 +492,12 @@ void Realm::update_schema(Schema schema, uint64_t version, MigrationFunction mig
 
     schema.copy_keys_from(actual_schema, m_config.schema_subset_mode);
 
+    bool save_schema_version_on_version_decrease = false;
+#if REALM_ENABLE_SYNC
+    if (m_config.sync_config && m_config.sync_config->flx_sync_requested)
+        save_schema_version_on_version_decrease = true;
+#endif
+
     uint64_t old_schema_version = m_schema_version;
     bool additive = m_config.schema_mode == SchemaMode::AdditiveDiscovered ||
                     m_config.schema_mode == SchemaMode::AdditiveExplicit ||
@@ -512,11 +527,12 @@ void Realm::update_schema(Schema schema, uint64_t version, MigrationFunction mig
 
         ObjectStore::apply_schema_changes(transaction(), version, m_schema, m_schema_version, m_config.schema_mode,
                                           required_changes, m_config.automatically_handle_backlinks_in_migrations,
-                                          wrapper);
+                                          wrapper, save_schema_version_on_version_decrease);
     }
     else {
         ObjectStore::apply_schema_changes(transaction(), m_schema_version, schema, version, m_config.schema_mode,
-                                          required_changes, m_config.automatically_handle_backlinks_in_migrations);
+                                          required_changes, m_config.automatically_handle_backlinks_in_migrations,
+                                          nullptr, save_schema_version_on_version_decrease);
         REALM_ASSERT_DEBUG(additive ||
                            (required_changes = ObjectStore::schema_from_group(read_group()).compare(schema)).empty());
     }

src/realm/object-store/shared_realm.hpp

@@ -367,7 +367,7 @@ class Realm : public std::enable_shared_from_this<Realm> {
      * will be thrown instead.
      *
      * If the destination file does not exist, the action performed depends on
-     * the type of the source and destimation files. If the destination
+     * the type of the source and destination files. If the destination
      * configuration is a non-sync local Realm configuration, a compacted copy
      * of the current Transaction's data (which includes uncommitted changes if
      * applicable!) is written in streaming form, with no history.
@@ -556,6 +556,7 @@ class Realm : public std::enable_shared_from_this<Realm> {
     void set_schema(Schema const& reference, Schema schema);
     bool reset_file(Schema& schema, std::vector<SchemaChange>& changes_required);
     bool schema_change_needs_write_transaction(Schema& schema, std::vector<SchemaChange>& changes, uint64_t version);
+    void verify_schema_version_not_decreasing(uint64_t version);
     Schema get_full_schema();
 
     // Ensure that m_schema and m_schema_version match that of the current