feat: add job worker machinery

Author: 0xgleb

crates/event-sorcery/src/job.rs

@@ -201,7 +201,7 @@ pub type Storage<J> = SqliteStorage<J, JsonCodec<CompactType>, SqliteFetcher>;
 
 /// The apalis worker handler. Deserializes the job, logs, and runs
 /// [`perform`](Job::perform) against the injected input.
-#[cfg(not(feature = "test-support"))]
+#[cfg(not(any(test, feature = "test-support")))]
 pub async fn work<J>(
     job: J,
     input: Data<Arc<J::Input>>,
@@ -220,6 +220,22 @@ where
         })
 }
 
+/// Test-support [`work`] variant that routes execution through a
+/// [`FailureInjector`] so end-to-end tests can force a terminal
+/// failure for a targeted job.
+#[cfg(any(test, feature = "test-support"))]
+pub async fn work<J>(
+    job: J,
+    input: Data<Arc<J::Input>>,
+    injector: Data<FailureInjector>,
+    attempt: Attempt,
+) -> Result<J::Output, JobError>
+where
+    J: Job + Sync,
+{
+    injector.perform::<J>(&job, &input, attempt.current()).await
+}
+
 /// Worker event handler that stops the worker and notifies the
 /// supervisor on a terminal (retries-exhausted) failure.
 pub fn on_terminal_failure(
@@ -250,6 +266,81 @@ fn log_processing(label: &Label, attempt: usize) {
     debug!(target: "job", %label, attempt, "processing job");
 }
 
+/// Fault-injection registry for end-to-end terminal-failure tests.
+///
+/// [`arm`](Self::arm) marks a [`Job::KIND`] so the next job of that
+/// kind fails terminally; the failure then sticks to that specific
+/// job instance (by [`Label`]) so retries of it keep failing while
+/// other instances of the same kind run normally.
+#[cfg(any(test, feature = "test-support"))]
+#[derive(Clone, Debug, Default)]
+pub struct FailureInjector {
+    states: Arc<std::sync::Mutex<std::collections::HashMap<&'static str, InjectionState>>>,
+}
+
+#[cfg(any(test, feature = "test-support"))]
+#[derive(Debug, Default)]
+enum InjectionState {
+    #[default]
+    Idle,
+    Armed,
+    Targeted(String),
+}
+
+#[cfg(any(test, feature = "test-support"))]
+impl FailureInjector {
+    /// Creates an injector with no kinds armed.
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Arms injection for the next job of `kind`.
+    pub fn arm(&self, kind: &'static str) {
+        self.lock().insert(kind, InjectionState::Armed);
+    }
+
+    fn should_inject(&self, kind: &'static str, label: &Label) -> bool {
+        let mut guard = self.lock();
+        let state = guard.entry(kind).or_default();
+        let inject = match state {
+            InjectionState::Idle => false,
+            InjectionState::Armed => {
+                *state = InjectionState::Targeted(label.as_str().to_owned());
+                true
+            }
+            InjectionState::Targeted(target) => target == label.as_str(),
+        };
+        drop(guard);
+        inject
+    }
+
+    fn lock(
+        &self,
+    ) -> std::sync::MutexGuard<'_, std::collections::HashMap<&'static str, InjectionState>> {
+        self.states
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    async fn perform<J: Job + Sync>(
+        &self,
+        job: &J,
+        input: &J::Input,
+        attempt: usize,
+    ) -> Result<J::Output, JobError> {
+        let label = job.label();
+        if self.should_inject(J::KIND, &label) {
+            return Err(JobError::Injected);
+        }
+        log_processing(&label, attempt);
+        job.perform(input).await.map_err(|source| JobError::Failed {
+            label,
+            source: Box::new(source),
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use serde::{Deserialize, Serialize};
@@ -295,4 +386,28 @@ mod tests {
         assert_eq!(welcome.label().to_string(), "welcome:a@example.com");
         assert_eq!(reminder.label().to_string(), "reminder:b@example.com");
     }
+
+    #[tokio::test]
+    async fn injector_targets_first_armed_instance_and_spares_others() {
+        let injector = FailureInjector::new();
+        injector.arm(SendEmail::KIND);
+
+        let welcome = SendEmail::Welcome {
+            address: "a@example.com".to_string(),
+        };
+        let reminder = SendEmail::Reminder {
+            address: "b@example.com".to_string(),
+        };
+
+        let first = injector.perform(&welcome, &(), 1).await;
+        assert!(matches!(first, Err(JobError::Injected)));
+
+        let retry = injector.perform(&welcome, &(), 2).await;
+        assert!(
+            matches!(retry, Err(JobError::Injected)),
+            "the injected failure must stick to the targeted instance across retries"
+        );
+
+        injector.perform(&reminder, &(), 1).await.unwrap();
+    }
 }

crates/event-sorcery/src/lib.rs

@@ -104,6 +104,8 @@ use std::str::FromStr;
 pub use dependency::Cons;
 pub use dependency::Nil;
 pub use dependency::{Dependent, EntityList, Fold, HasEntity, OneOf};
+#[cfg(any(test, feature = "test-support"))]
+pub use job::FailureInjector;
 #[doc(hidden)]
 pub use job::{
     ExponentialBackoff, FAIL_STOP_RECOVERY_TIMEOUT, RETRY_BACKOFF, Storage, on_terminal_failure,