Implement Side effect retries

* Introduce Protocol Version V2
* Also adds a workaround to make sure typescript SDK <= 1.2.1 works with Restate 1.1

Author: slinkydeveloper

crates/invoker-api/src/journal_reader.rs

@@ -14,6 +14,7 @@ use restate_types::identifiers::InvocationId;
 use restate_types::invocation::ServiceInvocationSpanContext;
 use restate_types::journal::raw::PlainRawEntry;
 use restate_types::journal::EntryIndex;
+use restate_types::time::MillisSinceEpoch;
 use std::future::Future;
 
 /// Metadata associated with a journal
@@ -22,18 +23,25 @@ pub struct JournalMetadata {
     pub length: EntryIndex,
     pub span_context: ServiceInvocationSpanContext,
     pub pinned_deployment: Option<PinnedDeployment>,
+    /// This value is not agreed among Partition processor replicas right now.
+    ///
+    /// The upper bound for the total clock skew is the clock skew of the different machines
+    /// and the max time difference between two replicas applying the journal append command.
+    pub last_modification_date: MillisSinceEpoch,
 }
 
 impl JournalMetadata {
     pub fn new(
         length: EntryIndex,
         span_context: ServiceInvocationSpanContext,
         pinned_deployment: Option<PinnedDeployment>,
+        last_modification_date: MillisSinceEpoch,
     ) -> Self {
         Self {
             pinned_deployment,
             span_context,
             length,
+            last_modification_date,
         }
     }
 }

crates/invoker-api/src/lib.rs

@@ -33,6 +33,7 @@ pub mod test_util {
     use restate_types::invocation::{InvocationTarget, ServiceInvocationSpanContext};
     use restate_types::journal::raw::PlainRawEntry;
     use restate_types::journal::Completion;
+    use restate_types::time::MillisSinceEpoch;
     use std::convert::Infallible;
     use std::iter::empty;
     use std::marker::PhantomData;
@@ -51,7 +52,12 @@ pub mod test_util {
             _sid: &'a InvocationId,
         ) -> Result<(JournalMetadata, Self::JournalStream), Self::Error> {
             Ok((
-                JournalMetadata::new(0, ServiceInvocationSpanContext::empty(), None),
+                JournalMetadata::new(
+                    0,
+                    ServiceInvocationSpanContext::empty(),
+                    None,
+                    MillisSinceEpoch::UNIX_EPOCH,
+                ),
                 futures::stream::empty(),
             ))
         }

crates/invoker-impl/src/invocation_state_machine.rs

@@ -23,6 +23,7 @@ pub(super) struct InvocationStateMachine {
     pub(super) invocation_target: InvocationTarget,
     invocation_state: InvocationState,
     retry_iter: retries::RetryIter<'static>,
+    pub(super) retry_count_since_last_stored_entry: u32,
 }
 
 /// This struct tracks which entries the invocation task generates,
@@ -57,7 +58,7 @@ impl JournalTracker {
             self.last_entry_sent_to_partition_processor,
         ) {
             (_, None) => {
-                // The invocation task didn't generated new entries.
+                // The invocation task didn't generate new entries.
                 // We're always good to retry in this case.
                 true
             }
@@ -135,6 +136,7 @@ impl InvocationStateMachine {
             invocation_target,
             invocation_state: InvocationState::New,
             retry_iter: retry_policy.into_iter(),
+            retry_count_since_last_stored_entry: 0,
         }
     }
 
@@ -202,6 +204,8 @@ impl InvocationStateMachine {
             InvocationState::InFlight { .. }
         ));
 
+        self.retry_count_since_last_stored_entry = 0;
+
         if let InvocationState::InFlight {
             journal_tracker,
             entries_to_ack,
@@ -271,7 +275,10 @@ impl InvocationStateMachine {
     }
 
     /// Returns Some() with the timer for the next retry, otherwise None if retry limit exhausted
-    pub(super) fn handle_task_error(&mut self) -> Option<Duration> {
+    pub(super) fn handle_task_error(
+        &mut self,
+        next_retry_interval_override: Option<Duration>,
+    ) -> Option<Duration> {
         let journal_tracker = match &self.invocation_state {
             InvocationState::InFlight {
                 journal_tracker, ..
@@ -289,9 +296,10 @@ impl InvocationStateMachine {
                 *journal_tracker
             }
         };
-        let next_timer = self.retry_iter.next();
 
+        let next_timer = next_retry_interval_override.or_else(|| self.retry_iter.next());
         if next_timer.is_some() {
+            self.retry_count_since_last_stored_entry += 1;
             self.invocation_state = InvocationState::WaitingRetry {
                 timer_fired: false,
                 journal_tracker,
@@ -339,16 +347,67 @@ mod tests {
             RetryPolicy::fixed_delay(Duration::from_secs(1), Some(10)),
         );
 
-        assert!(invocation_state_machine.handle_task_error().is_some());
+        assert!(invocation_state_machine.handle_task_error(None).is_some());
         check!(let InvocationState::WaitingRetry { .. } = invocation_state_machine.invocation_state);
 
         invocation_state_machine.notify_retry_timer_fired();
 
         // We stay in `WaitingForRetry`
-        assert!(invocation_state_machine.handle_task_error().is_some());
+        assert!(invocation_state_machine.handle_task_error(None).is_some());
         check!(let InvocationState::WaitingRetry { .. } = invocation_state_machine.invocation_state);
     }
 
+    #[test(tokio::test)]
+    async fn handle_error_counts_attempts_on_same_entry() {
+        let mut invocation_state_machine = InvocationStateMachine::create(
+            InvocationTarget::mock_virtual_object(),
+            RetryPolicy::fixed_delay(Duration::from_secs(1), Some(10)),
+        );
+
+        // Start invocation
+        invocation_state_machine.start(
+            tokio::spawn(async {}).abort_handle(),
+            mpsc::unbounded_channel().0,
+        );
+
+        // Notify error
+        assert!(invocation_state_machine.handle_task_error(None).is_some());
+        assert_eq!(
+            invocation_state_machine.retry_count_since_last_stored_entry,
+            1
+        );
+
+        // Try to start again
+        invocation_state_machine.start(
+            tokio::spawn(async {}).abort_handle(),
+            mpsc::unbounded_channel().0,
+        );
+
+        // Get error again
+        assert!(invocation_state_machine.handle_task_error(None).is_some());
+        assert_eq!(
+            invocation_state_machine.retry_count_since_last_stored_entry,
+            2
+        );
+
+        // Try to start again
+        invocation_state_machine.start(
+            tokio::spawn(async {}).abort_handle(),
+            mpsc::unbounded_channel().0,
+        );
+        assert_eq!(
+            invocation_state_machine.retry_count_since_last_stored_entry,
+            2
+        );
+
+        // Now complete the entry
+        invocation_state_machine.notify_new_entry(1, false);
+        assert_eq!(
+            invocation_state_machine.retry_count_since_last_stored_entry,
+            0
+        );
+    }
+
     #[test(tokio::test)]
     async fn handle_requires_ack() {
         let mut invocation_state_machine = InvocationStateMachine::create(

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

@@ -56,6 +56,10 @@ use tracing::instrument;
 const SERVICE_PROTOCOL_VERSION_V1: HeaderValue =
     HeaderValue::from_static("application/vnd.restate.invocation.v1");
 
+#[allow(clippy::declare_interior_mutable_const)]
+const SERVICE_PROTOCOL_VERSION_V2: HeaderValue =
+    HeaderValue::from_static("application/vnd.restate.invocation.v2");
+
 #[allow(clippy::declare_interior_mutable_const)]
 const X_RESTATE_SERVER: HeaderName = HeaderName::from_static("x-restate-server");
 
@@ -126,12 +130,14 @@ pub(crate) enum InvocationTaskError {
     #[code(unknown)]
     EntryEnrichment(EntryIndex, EntryType, #[source] InvocationError),
 
-    #[error("Error message received from the SDK with related entry {0:?}: {1}")]
+    #[error("Error message received from the SDK with related entry {related_entry:?}: {error}")]
     #[code(restate_errors::RT0007)]
-    ErrorMessageReceived(
-        Option<InvocationErrorRelatedEntry>,
-        #[source] InvocationError,
-    ),
+    ErrorMessageReceived {
+        related_entry: Option<InvocationErrorRelatedEntry>,
+        next_retry_interval_override: Option<Duration>,
+        #[source]
+        error: InvocationError,
+    },
     #[error("cannot talk to service endpoint '{0}' because its service protocol versions [{}, {}] are incompatible with the server's service protocol versions [{}, {}].", .1.start(), .1.end(), i32::from(MIN_SERVICE_PROTOCOL_VERSION), i32::from(MAX_SERVICE_PROTOCOL_VERSION))]
     #[code(restate_errors::RT0013)]
     IncompatibleServiceEndpoint(DeploymentId, RangeInclusive<i32>),
@@ -156,9 +162,19 @@ impl InvocationTaskError {
         true
     }
 
+    pub(crate) fn next_retry_interval_override(&self) -> Option<Duration> {
+        match self {
+            InvocationTaskError::ErrorMessageReceived {
+                next_retry_interval_override,
+                ..
+            } => *next_retry_interval_override,
+            _ => None,
+        }
+    }
+
     pub(crate) fn into_invocation_error(self) -> InvocationError {
         match self {
-            InvocationTaskError::ErrorMessageReceived(_, e) => e,
+            InvocationTaskError::ErrorMessageReceived { error, .. } => error,
             InvocationTaskError::EntryEnrichment(entry_index, entry_type, e) => {
                 let msg = format!(
                     "Error when processing entry {} of type {}: {}",
@@ -179,9 +195,10 @@ impl InvocationTaskError {
     pub(crate) fn into_invocation_error_report(mut self) -> InvocationErrorReport {
         let doc_error_code = codederror::CodedError::code(&self);
         let maybe_related_entry = match self {
-            InvocationTaskError::ErrorMessageReceived(ref mut related_entry, _) => {
-                related_entry.take()
-            }
+            InvocationTaskError::ErrorMessageReceived {
+                ref mut related_entry,
+                ..
+            } => related_entry.take(),
             _ => None,
         }
         .unwrap_or_default();
@@ -247,6 +264,7 @@ pub(super) struct InvocationTask<SR, JR, EE, DMR> {
     disable_eager_state: bool,
     message_size_warning: usize,
     message_size_limit: Option<usize>,
+    retry_count_since_last_stored_entry: u32,
 
     // Invoker tx/rx
     state_reader: SR,
@@ -307,6 +325,7 @@ where
         disable_eager_state: bool,
         message_size_warning: usize,
         message_size_limit: Option<usize>,
+        retry_count_since_last_stored_entry: u32,
         state_reader: SR,
         journal_reader: JR,
         entry_enricher: EE,
@@ -330,6 +349,7 @@ where
             invoker_rx,
             message_size_limit,
             message_size_warning,
+            retry_count_since_last_stored_entry,
         }
     }
 
@@ -479,6 +499,7 @@ fn service_protocol_version_to_header_value(
             unreachable!("unknown protocol version should never be chosen")
         }
         ServiceProtocolVersion::V1 => SERVICE_PROTOCOL_VERSION_V1,
+        ServiceProtocolVersion::V2 => SERVICE_PROTOCOL_VERSION_V2,
     }
 }
 

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

@@ -37,8 +37,10 @@ use restate_types::schema::deployment::{
     Deployment, DeploymentMetadata, DeploymentType, ProtocolType,
 };
 use restate_types::service_protocol::ServiceProtocolVersion;
+use restate_types::time::MillisSinceEpoch;
 use std::collections::HashSet;
 use std::future::poll_fn;
+use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, info, trace, warn, Span};
@@ -152,8 +154,14 @@ where
         );
 
         crate::shortcircuit!(
-            self.write_start(&mut http_stream_tx, journal_size, state_iter)
-                .await
+            self.write_start(
+                &mut http_stream_tx,
+                journal_size,
+                state_iter,
+                self.invocation_task.retry_count_since_last_stored_entry,
+                journal_metadata.last_modification_date
+            )
+            .await
         );
 
         // Initialize the response stream state
@@ -346,10 +354,8 @@ where
                         ResponseChunk::Data(buf) => crate::shortcircuit!(self.handle_read(parent_span_context, buf)),
                         ResponseChunk::End => {
                             // Response stream was closed without SuspensionMessage, EndMessage or ErrorMessage
-                            return TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived(
-                                None,
-                                InvocationError::default()
-                            ))
+                            return TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived{
+                            related_entry: None,next_retry_interval_override: None,error: InvocationError::default(),})
                         }
                     }
                 },
@@ -376,10 +382,11 @@ where
                         ResponseChunk::Data(buf) => crate::shortcircuit!(self.handle_read(parent_span_context, buf)),
                         ResponseChunk::End => {
                             // Response stream was closed without SuspensionMessage, EndMessage or ErrorMessage
-                            return TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived(
-                                None,
-                                InvocationError::default()
-                            ))
+                            return TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived {
+                                related_entry: None,
+                                next_retry_interval_override: None,
+                                error: InvocationError::default(),
+                            })
                         }
                     }
                 },
@@ -398,6 +405,8 @@ where
         http_stream_tx: &mut InvokerRequestStreamSender,
         journal_size: u32,
         state_entries: EagerState<I>,
+        retry_count_since_last_stored_entry: u32,
+        duration_since_last_stored_entry: MillisSinceEpoch,
     ) -> Result<(), InvocationTaskError> {
         let is_partial = state_entries.is_partial();
 
@@ -414,6 +423,8 @@ where
                 journal_size,
                 is_partial,
                 state_entries,
+                retry_count_since_last_stored_entry,
+                duration_since_last_stored_entry,
             ),
         )
         .await
@@ -530,8 +541,8 @@ where
                 TerminalLoopState::Suspended(suspension_indexes)
             }
             ProtocolMessage::Error(e) => {
-                TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived(
-                    Some(InvocationErrorRelatedEntry {
+                TerminalLoopState::Failed(InvocationTaskError::ErrorMessageReceived {
+                    related_entry: Some(InvocationErrorRelatedEntry {
                         related_entry_index: e.related_entry_index,
                         related_entry_name: e.related_entry_name.clone(),
                         related_entry_type: e
@@ -540,8 +551,9 @@ where
                             .and_then(|idx| MessageType::try_from(idx).ok())
                             .and_then(|mt| EntryType::try_from(mt).ok()),
                     }),
-                    InvocationError::from(e),
-                ))
+                    next_retry_interval_override: e.next_retry_delay.map(Duration::from_millis),
+                    error: InvocationError::from(e),
+                })
             }
             ProtocolMessage::End(_) => TerminalLoopState::Closed,
             ProtocolMessage::UnparsedEntry(entry) => {