Introduce version barrier for enabling journal v2 by default

The version barrier requiring min Restate version v1.6.0 will enable the
journal v2 by default. It will be written when becoming leader and running
at least v1.7.0.

Writing the journal v2 by default changes the logic of the state machine. That's
why we are guarding it with a version barrier so that during a rolling upgrade the
state machine state won't diverge. A concrete problem is that Restart as new with
a non-zero prefix would fail on a node that didn't store the journal entries in the
journal v2 table.

Author: tillrohrmann

.github/workflows/ci.yml

@@ -141,7 +141,7 @@ jobs:
       serviceImage: "ghcr.io/restatedev/test-services-java:main"
       testArtifactOutput: sdk-java-integration-test-journal-table-v2
       envVars: |
-        RESTATE_EXPERIMENTAL_FEATURE__USE_JOURNAL_V2_BY_DEFAULT=true
+        RESTATE_INTERNAL_FORCE_MIN_RESTATE_VERSION=1.6.0
 
   sdk-python:
     name: Run SDK-Python integration tests
@@ -283,19 +283,19 @@ jobs:
         with:
           testArtifactOutput: e2e-journal-table-v2-test-report
           restateContainerImage: localhost/restatedev/restate-commit-download:latest
-          # Needed for backward compatibility tests 1.6 -> 1.5
           envVars: |
             RESTATE_WORKER__INVOKER__experimental_features_allow_protocol_v6=true
-            RESTATE_EXPERIMENTAL_FEATURE__USE_JOURNAL_V2_BY_DEFAULT=true
+            RESTATE_INTERNAL_FORCE_MIN_RESTATE_VERSION=1.6.0
+          # Ignore forward compatibility tests as long as the previous version is not at least v1.6.0
           # Why these tests are disabled?
           # In restate 1.5 the invoker storage reader will not handle correctly the case where there's no pinned deployment yet
           # and the journal table used is v2. This doesn't show in the logs, but will simply hang badly the invocation task loop!
           # These tests trigger this condition!
           exclusions: |
             exclusions:
               "versionCompat":
-                - "dev.restate.sdktesting.tests.BackCompatibilityTest\\$OldVersion#proxyCallShouldBeDone"
-                - "dev.restate.sdktesting.tests.BackCompatibilityTest\\$OldVersion#proxyOneWayCallShouldBeDone"
+                - "dev.restate.sdktesting.tests.ForwardCompatibilityTest\\$OldVersion#proxyCallShouldBeDone"
+                - "dev.restate.sdktesting.tests.ForwardCompatibilityTest\\$OldVersion#proxyOneWayCallShouldBeDone"
 
   jepsen:
     if: github.event.repository.fork == false && github.event.pull_request.head.repo.full_name == 'restatedev/restate' && github.ref == 'refs/heads/main'

crates/types/src/restate_version.rs

@@ -108,6 +108,19 @@ impl SemanticRestateVersion {
         self.0.pre == semver::Prerelease::new("dev").unwrap()
     }
 
+    /// Returns the major.minor.patch part of the semantic version (ignoring the dev part).
+    pub fn strip_dev(&self) -> Cow<'_, SemanticRestateVersion> {
+        if self.is_dev() {
+            Cow::Owned(SemanticRestateVersion::new(
+                self.major(),
+                self.minor(),
+                self.patch(),
+            ))
+        } else {
+            Cow::Borrowed(self)
+        }
+    }
+
     pub fn parse(s: &str) -> Result<Self, RestateVersionError> {
         Self::from_str(s)
     }
@@ -200,6 +213,10 @@ impl TryFrom<&RestateVersion> for SemanticRestateVersion {
     }
 }
 
+// Version constants
+pub const RESTATE_VERSION_1_6_0: SemanticRestateVersion = SemanticRestateVersion::new(1, 6, 0);
+pub const RESTATE_VERSION_1_7_0: SemanticRestateVersion = SemanticRestateVersion::new(1, 7, 0);
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/worker/src/partition/leadership/leader_state.rs

@@ -39,7 +39,7 @@ use restate_types::net::partition_processor::{
     PartitionProcessorRpcError, PartitionProcessorRpcResponse,
 };
 use restate_types::time::MillisSinceEpoch;
-use restate_types::{SemanticRestateVersion, Version, Versioned};
+use restate_types::{RESTATE_VERSION_1_7_0, SemanticRestateVersion, Version, Versioned};
 use restate_wal_protocol::Command;
 use restate_wal_protocol::timer::TimerKeyValue;
 
@@ -309,10 +309,11 @@ impl LeaderState {
                         .await?;
                 }
                 ActionEffect::UpsertSchema(schema) => {
-                    const GATE_VERSION: SemanticRestateVersion =
-                        SemanticRestateVersion::new(1, 7, 0);
-
-                    if SemanticRestateVersion::current().is_equal_or_newer_than(&GATE_VERSION) {
+                    // treat dev version as released version to enable testing
+                    if SemanticRestateVersion::current()
+                        .strip_dev()
+                        .is_equal_or_newer_than(&RESTATE_VERSION_1_7_0)
+                    {
                         self.self_proposer
                             .propose(
                                 *self.partition_key_range.start(),

crates/worker/src/partition/leadership/mod.rs

@@ -39,12 +39,12 @@ use restate_storage_api::invocation_status_table::{
 use restate_storage_api::outbox_table::{OutboxMessage, ReadOutboxTable};
 use restate_storage_api::timer_table::{ReadTimerTable, TimerKey};
 use restate_timer::TokioClock;
-use restate_types::GenerationalNodeId;
 use restate_types::cluster::cluster_state::RunMode;
 use restate_types::config::Configuration;
 use restate_types::errors::GenericError;
 use restate_types::identifiers::{InvocationId, PartitionKey, PartitionProcessorRpcRequestId};
 use restate_types::identifiers::{LeaderEpoch, PartitionLeaderEpoch};
+use restate_types::logs::Keys;
 use restate_types::message::MessageIndex;
 use restate_types::net::partition_processor::{
     PartitionProcessorRpcError, PartitionProcessorRpcResponse,
@@ -54,8 +54,11 @@ use restate_types::partitions::state::PartitionReplicaSetStates;
 use restate_types::retries::with_jitter;
 use restate_types::schema::Schema;
 use restate_types::storage::StorageEncodeError;
+use restate_types::{
+    GenerationalNodeId, RESTATE_VERSION_1_6_0, RESTATE_VERSION_1_7_0, SemanticRestateVersion,
+};
 use restate_wal_protocol::Command;
-use restate_wal_protocol::control::{AnnounceLeader, PartitionDurability};
+use restate_wal_protocol::control::{AnnounceLeader, PartitionDurability, VersionBarrier};
 use restate_wal_protocol::timer::TimerKeyValue;
 
 use crate::partition::cleaner::Cleaner;
@@ -399,6 +402,53 @@ where
             let mut self_proposer = self_proposer.take().expect("must be present");
             self_proposer.mark_as_leader().await;
 
+            let mut min_restate_version = partition_store.get_min_restate_version().await?;
+
+            // Force the provided min Restate version, this is used for internal testing only
+            if let Some(forced_min_restate_version) =
+                std::env::var("RESTATE_INTERNAL_FORCE_MIN_RESTATE_VERSION")
+                    .ok()
+                    .and_then(|min_restate_version| {
+                        SemanticRestateVersion::parse(&min_restate_version).ok()
+                    })
+            {
+                self_proposer
+                    .propose(
+                        *self.partition.key_range.start(),
+                        Command::VersionBarrier(VersionBarrier {
+                            version: forced_min_restate_version.clone(),
+                            partition_key_range: Keys::RangeInclusive(
+                                self.partition.key_range.clone(),
+                            ),
+                            human_reason: Some("Force min Restate version".to_owned()),
+                        }),
+                    )
+                    .await?;
+
+                min_restate_version = min_restate_version.max(forced_min_restate_version);
+            }
+
+            // In v1.7.0 we enable by default writing to the journal v2 which requires min Restate v1.6.0
+            // We ignore dev to enable testing
+            if SemanticRestateVersion::current()
+                .strip_dev()
+                .is_equal_or_newer_than(&RESTATE_VERSION_1_7_0)
+                && min_restate_version.cmp_precedence(&RESTATE_VERSION_1_6_0) == Ordering::Less
+            {
+                self_proposer
+                    .propose(
+                        *self.partition.key_range.start(),
+                        Command::VersionBarrier(VersionBarrier {
+                            version: RESTATE_VERSION_1_6_0,
+                            partition_key_range: Keys::RangeInclusive(
+                                self.partition.key_range.clone(),
+                            ),
+                            human_reason: Some("Enable journal v2 by default".to_owned()),
+                        }),
+                    )
+                    .await?;
+            }
+
             let last_reported_durable_lsn = partition_store
                 .get_partition_durability()
                 .await?

crates/worker/src/partition/mod.rs

@@ -16,14 +16,12 @@ pub mod shuffle;
 mod state_machine;
 pub mod types;
 
-use std::env;
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
 use assert2::let_assert;
-use enumset::EnumSet;
 use futures::{FutureExt, Stream, StreamExt};
 use metrics::{SharedString, gauge, histogram};
 use tokio::sync::{mpsc, watch};
@@ -67,7 +65,7 @@ use crate::metric_definitions::{
 };
 use crate::partition::invoker_storage_reader::InvokerStorageReader;
 use crate::partition::leadership::LeadershipState;
-use crate::partition::state_machine::{ActionCollector, Feature, StateMachine};
+use crate::partition::state_machine::{ActionCollector, StateMachine};
 
 /// Target leader state of the partition processor.
 #[derive(Clone, Copy, Debug, Default, PartialEq)]
@@ -184,7 +182,11 @@ where
         let min_restate_version = partition_store.get_min_restate_version().await?;
         let schema = partition_store.get_schema().await?;
 
-        if !SemanticRestateVersion::current().is_equal_or_newer_than(&min_restate_version) {
+        // We ignore dev to enable testing in dev version
+        if !SemanticRestateVersion::current()
+            .strip_dev()
+            .is_equal_or_newer_than(&min_restate_version)
+        {
             gauge!(PARTITION_BLOCKED_FLARE, PARTITION_LABEL =>
                 partition_store.partition_id().to_string())
             .set(1);
@@ -194,19 +196,12 @@ where
             });
         }
 
-        let mut features = EnumSet::new();
-        // TODO(till) enable this using partition processor version barrier
-        if env::var("RESTATE_EXPERIMENTAL_FEATURE__USE_JOURNAL_V2_BY_DEFAULT").is_ok() {
-            features.insert(Feature::UseJournalTableV2AsDefault);
-        }
-
         let state_machine = StateMachine::new(
             inbox_seq_number,
             outbox_seq_number,
             outbox_head_seq_number,
             partition_store.partition_key_range().clone(),
             min_restate_version,
-            features,
             schema,
         );
 

crates/worker/src/partition/state_machine/entries/mod.rs

@@ -405,29 +405,31 @@ impl<CMD> ApplyJournalCommandEffect<'_, CMD> {
 
 #[cfg(test)]
 mod tests {
-    use crate::partition::state_machine::Feature;
     use crate::partition::state_machine::tests::fixtures::invoker_entry_effect;
     use crate::partition::state_machine::tests::{TestEnv, fixtures, matchers};
     use bytes::Bytes;
-    use enumset::EnumSet;
     use googletest::prelude::*;
     use restate_storage_api::invocation_status_table::ReadInvocationStatusTable;
     use restate_types::identifiers::{InvocationId, ServiceId};
     use restate_types::invocation::{
         Header, InvocationResponse, InvocationTarget, JournalCompletionTarget, ResponseResult,
     };
     use restate_types::journal_v2::{CallCommand, CallRequest};
+    use restate_types::{RESTATE_VERSION_1_6_0, SemanticRestateVersion};
     use restate_wal_protocol::Command;
-    use rstest::rstest;
 
-    #[rstest]
     #[restate_core::test]
-    async fn update_journal_and_commands_length(
-        #[values(Feature::UseJournalTableV2AsDefault.into(), EnumSet::empty())] features: EnumSet<
-            Feature,
-        >,
-    ) {
-        let mut test_env = TestEnv::create_with_features(features).await;
+    async fn update_journal_and_commands_length() {
+        run_update_journal_and_commands_length(SemanticRestateVersion::unknown()).await;
+    }
+
+    #[restate_core::test]
+    async fn update_journal_and_commands_length_journal_v2_enabled() {
+        run_update_journal_and_commands_length(RESTATE_VERSION_1_6_0).await;
+    }
+
+    async fn run_update_journal_and_commands_length(min_restate_version: SemanticRestateVersion) {
+        let mut test_env = TestEnv::create_with_min_restate_version(min_restate_version).await;
         let invocation_id = fixtures::mock_start_invocation(&mut test_env).await;
         fixtures::mock_pinned_deployment_v5(&mut test_env, invocation_id).await;
 

crates/worker/src/partition/state_machine/entries/notification.rs

@@ -201,11 +201,9 @@ where
 mod tests {
     use super::*;
 
-    use crate::partition::state_machine::Feature;
     use crate::partition::state_machine::tests::{TestEnv, fixtures, matchers};
     use bytes::Bytes;
     use bytestring::ByteString;
-    use enumset::EnumSet;
     use googletest::prelude::*;
     use restate_service_protocol_v4::entry_codec::ServiceProtocolV4Codec;
     use restate_storage_api::invocation_status_table::{
@@ -221,6 +219,7 @@ mod tests {
         SleepCommand, SleepCompletion,
     };
     use restate_types::time::MillisSinceEpoch;
+    use restate_types::{RESTATE_VERSION_1_6_0, SemanticRestateVersion};
     use restate_wal_protocol::Command;
     use restate_wal_protocol::timer::TimerKeyValue;
     use rstest::rstest;
@@ -264,14 +263,21 @@ mod tests {
         test_env.shutdown().await;
     }
 
-    #[rstest]
     #[restate_core::test]
-    async fn notify_signal_received_before_pinned_deployment(
-        #[values(Feature::UseJournalTableV2AsDefault.into(), EnumSet::empty())] features: EnumSet<
-            Feature,
-        >,
+    async fn notify_signal_received_before_pinned_deployment() {
+        run_notify_signal_received_before_pinned_deployment(SemanticRestateVersion::unknown())
+            .await;
+    }
+
+    #[restate_core::test]
+    async fn notify_signal_received_before_pinned_deployment_journal_v2_enabled() {
+        run_notify_signal_received_before_pinned_deployment(RESTATE_VERSION_1_6_0).await;
+    }
+
+    async fn run_notify_signal_received_before_pinned_deployment(
+        min_restate_version: SemanticRestateVersion,
     ) {
-        let mut test_env = TestEnv::create_with_features(features).await;
+        let mut test_env = TestEnv::create_with_min_restate_version(min_restate_version).await;
         let invocation_id = fixtures::mock_start_invocation(&mut test_env).await;
 
         // Send signal notification before pinned deployment