Add Cleaner task to clean completed invocations

crates/core/src/task_center_types.rs

@@ -75,6 +75,7 @@ pub enum TaskKind {
     #[strum(props(OnError = "log"))]
     ConnectionReactor,
     Shuffle,
+    Cleaner,
     MetadataStore,
     // -- Bifrost Tasks
     /// A background task that the system needs for its operation. The task requires a system

crates/ingress-dispatcher/src/dispatcher.rs

@@ -316,7 +316,7 @@ mod tests {
                 );
                 invocation.argument = argument.clone();
                 invocation.idempotency_key = Some(idempotency_key.clone());
-                invocation.completion_retention_time = Some(Duration::from_secs(60));
+                invocation.completion_retention_duration = Some(Duration::from_secs(60));
                 let (ingress_req, _, res) = IngressDispatcherRequest::invocation(invocation);
                 dispatcher.dispatch_ingress_request(ingress_req).await?;
 
@@ -343,7 +343,7 @@ mod tests {
                         invocation_target: eq(invocation_target.clone()),
                         argument: eq(argument.clone()),
                         idempotency_key: some(eq(idempotency_key.clone())),
-                        completion_retention_time: some(eq(Duration::from_secs(60)))
+                        completion_retention_duration: some(eq(Duration::from_secs(60)))
                     })
                 );
                 let_assert!(

crates/ingress-http/src/handler/service_handler.rs

@@ -180,7 +180,7 @@ where
             let mut service_invocation =
                 ServiceInvocation::initialize(invocation_id, invocation_target, Source::Ingress);
             service_invocation.with_related_span(SpanRelation::Parent(ingress_span_context));
-            service_invocation.completion_retention_time =
+            service_invocation.completion_retention_duration =
                 invocation_target_meta.compute_retention(idempotency_key.is_some());
             if let Some(key) = idempotency_key {
                 service_invocation.idempotency_key = Some(key);

crates/ingress-http/src/handler/tests.rs

@@ -373,7 +373,7 @@ async fn idempotency_key_parsing() {
             Some(ByteString::from_static("123456"))
         );
         assert_eq!(
-            service_invocation.completion_retention_time,
+            service_invocation.completion_retention_duration,
             Some(Duration::from_secs(60 * 60 * 24))
         );
 
@@ -437,7 +437,7 @@ async fn idempotency_key_and_send() {
             Some(ByteString::from_static("123456"))
         );
         assert_eq!(
-            service_invocation.completion_retention_time,
+            service_invocation.completion_retention_duration,
             Some(Duration::from_secs(60 * 60 * 24))
         );
         assert_that!(
@@ -497,7 +497,7 @@ async fn idempotency_key_and_send_with_different_invocation_id() {
             Some(ByteString::from_static("123456"))
         );
         assert_eq!(
-            service_invocation.completion_retention_time,
+            service_invocation.completion_retention_duration,
             Some(Duration::from_secs(60 * 60 * 24))
         );
         assert_that!(

crates/partition-store/tests/integration_test.rs

@@ -92,7 +92,7 @@ pub(crate) fn mock_service_invocation(service_id: ServiceId) -> ServiceInvocatio
         span_context: Default::default(),
         headers: vec![],
         execution_time: None,
-        completion_retention_time: None,
+        completion_retention_duration: None,
         idempotency_key: None,
         submit_notification_sink: None,
     }

crates/partition-store/tests/invocation_status_table_test/mod.rs

@@ -82,9 +82,9 @@ fn invoked_status(
         journal_metadata: JournalMetadata::initialize(ServiceInvocationSpanContext::empty()),
         pinned_deployment: None,
         response_sinks: HashSet::new(),
-        timestamps: StatusTimestamps::new(MillisSinceEpoch::new(0), MillisSinceEpoch::new(0)),
+        timestamps: StatusTimestamps::init(MillisSinceEpoch::new(0)),
         source: Source::Ingress,
-        completion_retention_time: Duration::ZERO,
+        completion_retention_duration: Duration::ZERO,
         idempotency_key: None,
         source_table,
     })
@@ -100,9 +100,9 @@ fn suspended_status(
             journal_metadata: JournalMetadata::initialize(ServiceInvocationSpanContext::empty()),
             pinned_deployment: None,
             response_sinks: HashSet::new(),
-            timestamps: StatusTimestamps::new(MillisSinceEpoch::new(0), MillisSinceEpoch::new(0)),
+            timestamps: StatusTimestamps::init(MillisSinceEpoch::new(0)),
             source: Source::Ingress,
-            completion_retention_time: Duration::ZERO,
+            completion_retention_duration: Duration::ZERO,
             idempotency_key: None,
             source_table,
         },

crates/storage-api/proto/dev/restate/storage/v1/domain.proto

@@ -112,15 +112,21 @@ message NeoInvocationStatus {
     InvocationTarget invocation_target = 2;
     Source source = 3;
     SpanContext span_context = 4;
+    repeated ServiceInvocationResponseSink response_sinks = 7;
+    Duration completion_retention_duration = 11;
+
+    // Timestamps
     uint64 creation_time = 5;
     uint64 modification_time = 6;
-    repeated ServiceInvocationResponseSink response_sinks = 7;
+    optional uint64 inboxed_transition_time = 19;
+    optional uint64 scheduled_transition_time = 20;
+    optional uint64 running_transition_time = 21;
+    optional uint64 completed_transition_time = 22;
 
     // Scheduled/Inboxed
     optional bytes argument = 8;
     repeated Header headers = 9;
     optional uint64 execution_time = 10;
-    Duration completion_retention_time = 11;
     optional string idempotency_key = 12;
 
     // Inboxed