Refactor invoker following the rust conventions (#528)

* Moved options out of invoker.rs
* Rename OutputEffect -> Effect, Kind -> EffectKind
* Split the invoker module
* Rename Invoker to Service
* Split the various "parts" of the invoker interface from lib to specific mods
* Introduced interface for InvokerStatusReader
* Simplified the injection of Invoker state through the state machine coordinator.

src/invoker/src/effects.rs

@@ -0,0 +1,24 @@
+use restate_common::errors::InvocationError;
+use restate_common::types::{EnrichedRawEntry, EntryIndex, ServiceInvocationId};
+use std::collections::HashSet;
+
+#[derive(Debug)]
+pub struct Effect {
+ pub service_invocation_id: ServiceInvocationId,
+ pub kind: EffectKind,
+}
+
+#[derive(Debug)]
+pub enum EffectKind {
+ JournalEntry {
+ entry_index: EntryIndex,
+ entry: EnrichedRawEntry,
+ },
+ Suspended {
+ waiting_for_completed_entries: HashSet<EntryIndex>,
+ },
+ /// This is sent always after [`Self::JournalEntry`] with `OutputStreamEntry`(s).
+ End,
+ /// This is sent when the invoker exhausted all its attempts to make progress on the specific invocation.
+ Failed(InvocationError),
+}

src/invoker/src/handle.rs

@@ -0,0 +1,67 @@
+use super::Effect;
+
+use restate_common::types::{
+ EntryIndex, JournalMetadata, PartitionLeaderEpoch, ServiceInvocationId,
+};
+use restate_journal::raw::PlainRawEntry;
+use restate_journal::Completion;
+use std::future::Future;
+use tokio::sync::mpsc;
+
+#[derive(Debug, Default)]
+pub enum InvokeInputJournal {
+ #[default]
+ NoCachedJournal,
+ CachedJournal(JournalMetadata, Vec<PlainRawEntry>),
+}
+
+// TODO move this to restate_errors, we have several copies of this type (e.g. NetworkNotRunning)
+#[derive(Debug, thiserror::Error)]
+#[error("invoker is not running")]
+pub struct ServiceNotRunning;
+
+pub trait ServiceHandle {
+ type Future: Future<Output = Result<(), ServiceNotRunning>>;
+
+ fn invoke(
+ &mut self,
+ partition: PartitionLeaderEpoch,
+ service_invocation_id: ServiceInvocationId,
+ journal: InvokeInputJournal,
+ ) -> Self::Future;
+
+ fn resume(
+ &mut self,
+ partition: PartitionLeaderEpoch,
+ service_invocation_id: ServiceInvocationId,
+ journal: InvokeInputJournal,
+ ) -> Self::Future;
+
+ fn notify_completion(
+ &mut self,
+ partition: PartitionLeaderEpoch,
+ service_invocation_id: ServiceInvocationId,
+ completion: Completion,
+ ) -> Self::Future;
+
+ fn notify_stored_entry_ack(
+ &mut self,
+ partition: PartitionLeaderEpoch,
+ service_invocation_id: ServiceInvocationId,
+ entry_index: EntryIndex,
+ ) -> Self::Future;
+
+ fn abort_all_partition(&mut self, partition: PartitionLeaderEpoch) -> Self::Future;
+
+ fn abort_invocation(
+ &mut self,
+ partition_leader_epoch: PartitionLeaderEpoch,
+ service_invocation_id: ServiceInvocationId,
+ ) -> Self::Future;
+
+ fn register_partition(
+ &mut self,
+ partition: PartitionLeaderEpoch,
+ sender: mpsc::Sender<Effect>,
+ ) -> Self::Future;
+}