Consistent schema across partitions

Summary:
This PR make sure that PPs have a consistent service schema by writing
and upsert-schema record to bifrost. The leader processor makes sure
to write this record to bifrost when schema changes are detected

Author: muhamadazmy

crates/types/src/schema/metadata/mod.rs

@@ -41,8 +41,9 @@ use crate::schema::{deployment, service};
 use crate::time::MillisSinceEpoch;
 use crate::{Version, Versioned, identifiers};
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(derive_more::Debug, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(from = "serde_hacks::Schema", into = "serde_hacks::Schema")]
+#[debug("Schema(version: {version})")]
 pub struct Schema {
     /// This gets bumped on each update.
     version: Version,

crates/wal-protocol/src/control.rs

@@ -12,6 +12,7 @@ use std::ops::RangeInclusive;
 
 use restate_types::identifiers::{LeaderEpoch, PartitionId, PartitionKey};
 use restate_types::logs::{Keys, Lsn};
+use restate_types::schema::Schema;
 use restate_types::time::MillisSinceEpoch;
 use restate_types::{GenerationalNodeId, SemanticRestateVersion};
 
@@ -62,3 +63,13 @@ pub struct PartitionDurability {
     /// Timestamp which the durability point was updated
     pub modification_time: MillisSinceEpoch,
 }
+
+/// Consistency store schema across partition replicas.
+///
+/// Since v1.6.0.
+#[derive(Debug, Clone)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub struct UpsertSchema {
+    pub partition_key_range: Keys,
+    pub schema: Schema,
+}