fix: Return to pending if paused execution receives a response and unpauses

Author: tomasol

crates/concepts/src/storage.rs

@@ -1661,6 +1661,61 @@ pub enum PendingState {
     Finished(PendingStateFinished),
 }
 
+pub enum PendingStateMergedPause {
+    Locked {
+        state: PendingStateLocked,
+        paused: bool,
+    },
+    PendingAt {
+        state: PendingStatePendingAt,
+        paused: bool,
+    },
+    BlockedByJoinSet {
+        state: PendingStateBlockedByJoinSet,
+        paused: bool,
+    },
+    Finished(PendingStateFinished),
+}
+impl From<PendingState> for PendingStateMergedPause {
+    fn from(state: PendingState) -> Self {
+        match state {
+            PendingState::Locked(s) => PendingStateMergedPause::Locked {
+                state: s,
+                paused: false,
+            },
+
+            PendingState::PendingAt(s) => PendingStateMergedPause::PendingAt {
+                state: s,
+                paused: false,
+            },
+
+            PendingState::BlockedByJoinSet(s) => PendingStateMergedPause::BlockedByJoinSet {
+                state: s,
+                paused: false,
+            },
+
+            PendingState::Paused(paused) => match paused {
+                PendingStatePaused::Locked(s) => PendingStateMergedPause::Locked {
+                    state: s,
+                    paused: true,
+                },
+                PendingStatePaused::PendingAt(s) => PendingStateMergedPause::PendingAt {
+                    state: s,
+                    paused: true,
+                },
+                PendingStatePaused::BlockedByJoinSet(s) => {
+                    PendingStateMergedPause::BlockedByJoinSet {
+                        state: s,
+                        paused: true,
+                    }
+                }
+            },
+
+            PendingState::Finished(s) => PendingStateMergedPause::Finished(s),
+        }
+    }
+}
+
 #[derive(Debug, Clone, derive_more::Display, PartialEq, Eq, Serialize)]
 #[display("Locked(`{lock_expires_at}`, {}, {})", locked_by.executor_id, locked_by.run_id)]
 pub struct PendingStateLocked {

crates/db-postgres/src/postgres_dao.rs

@@ -18,10 +18,10 @@ use concepts::{
         ListLogsResponse, LockPendingResponse, Locked, LockedBy, LockedExecution, LogEntry,
         LogEntryRow, LogFilter, LogInfoAppendRow, LogLevel, LogStreamType, Pagination,
         PendingState, PendingStateBlockedByJoinSet, PendingStateFinished,
-        PendingStateFinishedResultKind, PendingStateLocked, PendingStatePaused,
-        PendingStatePendingAt, ResponseCursor, ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET,
-        STATE_FINISHED, STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome, Version, VersionType,
-        WasmBacktrace,
+        PendingStateFinishedResultKind, PendingStateLocked, PendingStateMergedPause,
+        PendingStatePaused, PendingStatePendingAt, ResponseCursor, ResponseWithCursor,
+        STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED, STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome,
+        Version, VersionType, WasmBacktrace,
     },
 };
 use deadpool_postgres::{Client, ManagerConfig, Pool, RecyclingMethod};
@@ -2830,11 +2830,16 @@ async fn append_response(
     let combined_state = get_combined_state(tx, execution_id).await?;
     debug!("previous_pending_state: {combined_state:?}");
 
-    let mut notifier = if let PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
-        join_set_id: found_join_set_id,
-        lock_expires_at,
-        closing: _,
-    }) = combined_state.execution_with_state.pending_state
+    let mut notifier = if let PendingStateMergedPause::BlockedByJoinSet {
+        state:
+            PendingStateBlockedByJoinSet {
+                join_set_id: found_join_set_id,
+                lock_expires_at, // Set to a future time if the worker is keeping the execution warm waiting for the result.
+                closing: _,
+            },
+        paused: _,
+    } =
+        PendingStateMergedPause::from(combined_state.execution_with_state.pending_state)
         && *join_set_id == found_join_set_id
     {
         let scheduled_at = std::cmp::max(lock_expires_at, event.created_at);

crates/db-sqlite/src/sqlite_dao.rs

@@ -18,9 +18,10 @@ use concepts::{
         ListLogsResponse, LockPendingResponse, Locked, LockedBy, LockedExecution, LogEntry,
         LogEntryRow, LogFilter, LogInfoAppendRow, LogLevel, LogStreamType, Pagination,
         PendingState, PendingStateBlockedByJoinSet, PendingStateFinished,
-        PendingStateFinishedResultKind, PendingStateLocked, PendingStatePaused,
-        PendingStatePendingAt, ResponseCursor, ResponseWithCursor, STATE_BLOCKED_BY_JOIN_SET,
-        STATE_FINISHED, STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome, Version, VersionType,
+        PendingStateFinishedResultKind, PendingStateLocked, PendingStateMergedPause,
+        PendingStatePaused, PendingStatePendingAt, ResponseCursor, ResponseWithCursor,
+        STATE_BLOCKED_BY_JOIN_SET, STATE_FINISHED, STATE_LOCKED, STATE_PENDING_AT, TimeoutOutcome,
+        Version, VersionType,
     },
 };
 use const_format::formatcp;
@@ -2928,11 +2929,16 @@ impl SqlitePool {
         // if the execution is going to be unblocked by this response...
         let combined_state = Self::get_combined_state(tx, execution_id)?;
         debug!("previous_pending_state: {combined_state:?}");
-        let mut notifier = if let PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
-            join_set_id: found_join_set_id,
-            lock_expires_at, // Set to a future time if the worker is keeping the execution warm waiting for the result.
-            closing: _,
-        }) = combined_state.execution_with_state.pending_state
+        let mut notifier = if let PendingStateMergedPause::BlockedByJoinSet {
+            state:
+                PendingStateBlockedByJoinSet {
+                    join_set_id: found_join_set_id,
+                    lock_expires_at, // Set to a future time if the worker is keeping the execution warm waiting for the result.
+                    closing: _,
+                },
+            paused: _,
+        } =
+            PendingStateMergedPause::from(combined_state.execution_with_state.pending_state)
             && *join_set_id == found_join_set_id
         {
             // PendingAt should be set to current time if called from expired_timers_watcher,