Excluded few e2e tests when running journal v2 default test configuration. See the changes for more details.

Author: slinkydeveloper

.github/workflows/ci.yml

@@ -287,6 +287,13 @@ jobs:
           envVars: |
             RESTATE_WORKER__INVOKER__experimental_features_allow_protocol_v6=true
             RESTATE_EXPERIMENTAL_FEATURE__USE_JOURNAL_V2_BY_DEFAULT=true
+          # Disable BackCompatibilityTest. But why this test is 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!
+          exclusions: |
+            exclusions:
+              "versionCompat":
+                - "dev.restate.sdktesting.tests.BackCompatibilityTest"
 
   jepsen:
     if: github.event.repository.fork == false && github.event.pull_request.head.repo.full_name == 'restatedev/restate' && github.ref == 'refs/heads/main'

crates/invoker-impl/src/error.rs

@@ -119,6 +119,11 @@ pub(crate) enum InvokerError {
         actual: InvocationEpoch,
         expected: InvocationEpoch,
     },
+    #[error(
+        "error when reading the journal: expected to read {expected} entries, but read only {expected}. This indicates a bug or a storage corruption."
+    )]
+    #[code(unknown)]
+    UnexpectedEntryCount { actual: u32, expected: u32 },
 
     #[error(transparent)]
     #[code(restate_errors::RT0010)]

crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs

@@ -184,8 +184,13 @@ where
 
         // Execute the replay
         crate::shortcircuit!(
-            self.replay_loop(&mut http_stream_tx, &mut decoder_stream, journal_stream)
-                .await
+            self.replay_loop(
+                &mut http_stream_tx,
+                &mut decoder_stream,
+                journal_stream,
+                journal_metadata.length
+            )
+            .await
         );
 
         // If we have the invoker_rx and the protocol type is bidi stream,
@@ -305,13 +310,15 @@ where
         http_stream_tx: &mut InvokerRequestStreamSender,
         http_stream_rx: &mut S,
         journal_stream: JournalStream,
+        expected_entries_count: u32,
     ) -> TerminalLoopState<()>
     where
         JournalStream: Stream<Item = JournalEntry> + Unpin,
         S: Stream<Item = Result<DecoderStreamItem, InvokerError>> + Unpin,
     {
         let mut journal_stream = journal_stream.fuse();
         let mut got_headers = false;
+        let mut sent_entries = 0;
 
         loop {
             tokio::select! {
@@ -334,10 +341,11 @@ where
                 opt_je = journal_stream.next() => {
                     match opt_je {
                         Some(JournalEntry::JournalV2(entry)) => {
+                            sent_entries += 1;
                             crate::shortcircuit!(self.write_entry(http_stream_tx, entry.inner).await);
-
                         }
                         Some(JournalEntry::JournalV1(old_entry)) => {
+                            sent_entries += 1;
                             if let journal::Entry::Input(input_entry) = crate::shortcircuit!(old_entry.deserialize_entry::<ProtobufRawEntryCodec>()) {
                                 crate::shortcircuit!(self.write_entry(
                                     http_stream_tx,
@@ -352,6 +360,14 @@ where
                             }
                         },
                         None => {
+                            // Let's verify if we sent all the entries we promised, otherwise the stream will hang in a bad way!
+                            if sent_entries < expected_entries_count {
+                                return TerminalLoopState::Failed(InvokerError::UnexpectedEntryCount {
+                                    actual: sent_entries,
+                                    expected: expected_entries_count,
+                                })
+                            }
+
                             // No need to wait for the headers to continue
                             trace!("Finished to replay the journal");
                             return TerminalLoopState::Continue(())