refactor(workflow,grpc,webapi): Record auto-upgrade outcome in history

Replace the success-only `ComponentUpgraded` event with
`ComponentUpgradeFinished` so both successful and failed auto-upgrades are
recorded in the execution log. Failed upgrades mark the new digest as
incompatible through the log instead of updating t_state directly.

Tighten `Unlocked` so it only releases a locked execution back to `PendingAt`,
while keeping the previous `backoff_expires_at` wire shape for JSON/proto
compatibility. Auto-upgrade appends `Unlocked` only when replay stops during
a stub-response flush and would otherwise leave the execution locked, leading
to a timeout.

This fixes the case where auto-upgrade unconditionally moved executions to
`PendingAt` (via `ComponentUpgraded`, allowing a blocked workflow to be
picked up immediately and panic on the next execution run if the response
had not arrived yet.

Add tests for blocked upgrades, failed upgrades, and the stub-write path
that still needs Unlocked.

Author: tomasol

assets/schemas/db.json

@@ -147,24 +147,11 @@
           "additionalProperties": false
         },
         {
-          "description": "Returns execution to [`PendingState::PendingAt`] state at the specified time.\nThis can happen when:\n- executor is running out of resources like [`WorkerError::LimitReached`]\n- executor is being closed (shutdown or hot redeploy requested)\n- activity is paused\n- workflow made progress but then its lock expired\n- workflow attempted to auto-upgrade an execution and failed, recording its digest in `incompatible_digest`",
+          "description": "Releases a lock.\n\nState transition semantics:\n- [`PendingState::Locked`] becomes [`PendingState::PendingAt`] at\n  [`Unlocked::backoff_expires_at`]. The field name is kept for persisted JSON and gRPC\n  compatibility, but it is the next pending instant for every unlock reason.\n- [`PendingState::PendingAt`], [`PendingState::BlockedByJoinSet`],\n  [`PendingState::Paused`], and [`PendingState::Finished`] reject this event.",
           "type": "object",
           "properties": {
             "unlocked": {
-              "type": "object",
-              "properties": {
-                "backoff_expires_at": {
-                  "type": "string",
-                  "format": "date-time"
-                },
-                "reason": {
-                  "$ref": "#/$defs/UnlockedReason"
-                }
-              },
-              "required": [
-                "backoff_expires_at",
-                "reason"
-              ]
+              "$ref": "#/$defs/Unlocked"
             }
           },
           "required": [
@@ -173,9 +160,10 @@
           "additionalProperties": false
         },
         {
+          "description": "Does not change `PendingState`.",
           "type": "object",
           "properties": {
-            "component_upgraded": {
+            "component_upgrade_finished": {
               "type": "object",
               "properties": {
                 "component_digest": {
@@ -184,19 +172,19 @@
                 "deployment_id": {
                   "type": "string"
                 },
-                "reason": {
-                  "$ref": "#/$defs/ComponentUpgradeReason"
+                "outcome": {
+                  "$ref": "#/$defs/ComponentUpgradeOutcome"
                 }
               },
               "required": [
                 "component_digest",
                 "deployment_id",
-                "reason"
+                "outcome"
               ]
             }
           },
           "required": [
-            "component_upgraded"
+            "component_upgrade_finished"
           ],
           "additionalProperties": false
         },
@@ -410,17 +398,34 @@
         "nanos"
       ]
     },
-    "UnlockedReason": {
+    "Unlocked": {
+      "type": "object",
+      "properties": {
+        "backoff_expires_at": {
+          "description": "Instant used when releasing a currently locked execution back to\n[`PendingState::PendingAt`]. This field keeps the released JSON and gRPC name.",
+          "type": "string",
+          "format": "date-time"
+        },
+        "reason": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "backoff_expires_at",
+        "reason"
+      ]
+    },
+    "ComponentUpgradeOutcome": {
       "oneOf": [
         {
           "type": "object",
           "properties": {
             "reason": {
-              "type": "string"
+              "$ref": "#/$defs/ComponentUpgradeReason"
             },
             "type": {
               "type": "string",
-              "const": "other"
+              "const": "success"
             }
           },
           "required": [
@@ -431,26 +436,23 @@
         {
           "type": "object",
           "properties": {
-            "target_digest": {
-              "type": "string"
-            },
             "reason": {
               "type": "string"
             },
             "type": {
               "type": "string",
-              "const": "auto_upgrade_failed"
+              "const": "failed"
             }
           },
           "required": [
             "type",
-            "target_digest",
             "reason"
           ]
         }
       ]
     },
     "ComponentUpgradeReason": {
+      "description": "Reason for auditing only",
       "oneOf": [
         {
           "type": "object",

crates/concepts/src/storage.rs

@@ -360,26 +360,24 @@ pub enum ExecutionRequest {
         scheduled_by: Option<ExecutionId>,
     },
     Locked(Locked),
-    /// Returns execution to [`PendingState::PendingAt`] state at the specified time.
-    /// This can happen when:
-    /// - executor is running out of resources like [`WorkerError::LimitReached`]
-    /// - executor is being closed (shutdown or hot redeploy requested)
-    /// - activity is paused
-    /// - workflow made progress but then its lock expired
-    /// - workflow attempted to auto-upgrade an execution and failed, recording its digest in `incompatible_digest`
-    #[display("Unlocked(`{backoff_expires_at}`)")]
-    Unlocked {
-        backoff_expires_at: DateTime<Utc>,
-        #[cfg_attr(any(test, feature = "test"), arbitrary(value = UnlockedReason::Other { reason: StrVariant::Static("reason") }))]
-        reason: UnlockedReason,
-    },
-    #[display("ComponentUpgraded({component_digest})")]
-    ComponentUpgraded {
+    /// Releases a lock.
+    ///
+    /// State transition semantics:
+    /// - [`PendingState::Locked`] becomes [`PendingState::PendingAt`] at
+    ///   [`Unlocked::backoff_expires_at`]. The field name is kept for persisted JSON and gRPC
+    ///   compatibility, but it is the next pending instant for every unlock reason.
+    /// - [`PendingState::PendingAt`], [`PendingState::BlockedByJoinSet`],
+    ///   [`PendingState::Paused`], and [`PendingState::Finished`] reject this event.
+    #[display("Unlocked({_0})")]
+    Unlocked(Unlocked),
+    /// Does not change `PendingState`.
+    #[display("ComponentUpgradeFinished({component_digest})")]
+    ComponentUpgradeFinished {
         #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentId::dummy_activity().component_digest))]
         component_digest: ComponentDigest,
         #[cfg_attr(any(test, feature = "test"), arbitrary(value = DeploymentId::from_parts(0, 0)))]
         deployment_id: DeploymentId,
-        reason: ComponentUpgradeReason,
+        outcome: ComponentUpgradeOutcome,
     },
     // Created by the executor holding the lock.
     // After expiry interpreted as pending.
@@ -419,24 +417,7 @@ pub enum ExecutionRequest {
     Unpaused,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, schemars::JsonSchema)]
-#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
-#[serde(tag = "type", rename_all = "snake_case")]
-pub enum UnlockedReason {
-    #[display("{reason}")]
-    Other {
-        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
-        reason: StrVariant,
-    },
-    #[display("auto-upgrade failed: {reason}")]
-    AutoUpgradeFailed {
-        #[cfg_attr(any(test, feature = "test"), arbitrary(value = ComponentId::dummy_activity().component_digest))]
-        target_digest: ComponentDigest,
-        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
-        reason: StrVariant,
-    },
-}
-
+/// Reason for auditing only
 #[derive(
     Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
 )]
@@ -449,91 +430,32 @@ pub enum ComponentUpgradeReason {
     Manual { force: bool },
 }
 
-impl<'de> Deserialize<'de> for UnlockedReason {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        unlocked_reason_compat::deserialize(deserializer)
-    }
-}
-
-mod unlocked_reason_compat {
-    use super::{ComponentDigest, StrVariant, UnlockedReason};
-    use serde::Deserialize;
-
-    // TODO: Delete this compatibility module with the next DB-breaking release.
-    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<UnlockedReason, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        Compat::deserialize(deserializer).map(Into::into)
-    }
-
-    #[derive(Deserialize)]
-    #[serde(untagged)]
-    enum Compat {
-        Current(Current),
-        Legacy(StrVariant),
-    }
-
-    #[derive(Deserialize)]
-    #[serde(tag = "type", rename_all = "snake_case")]
-    enum Current {
-        Other {
-            reason: StrVariant,
-        },
-        AutoUpgradeFailed {
-            target_digest: ComponentDigest,
-            reason: StrVariant,
-        },
-    }
-
-    impl From<Compat> for UnlockedReason {
-        fn from(value: Compat) -> Self {
-            match value {
-                Compat::Current(current) => current.into(),
-                Compat::Legacy(reason) => Self::Other { reason },
-            }
-        }
-    }
-
-    impl From<Current> for UnlockedReason {
-        fn from(value: Current) -> Self {
-            match value {
-                Current::Other { reason } => Self::Other { reason },
-                Current::AutoUpgradeFailed {
-                    target_digest,
-                    reason,
-                } => Self::AutoUpgradeFailed {
-                    target_digest,
-                    reason,
-                },
-            }
-        }
-    }
-}
-
-impl From<StrVariant> for UnlockedReason {
-    fn from(value: StrVariant) -> Self {
-        Self::Other { reason: value }
-    }
-}
-
-impl From<String> for UnlockedReason {
-    fn from(value: String) -> Self {
-        Self::Other {
-            reason: StrVariant::from(value),
-        }
-    }
+#[derive(
+    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
+)]
+#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
+#[display("{reason}, pending at {backoff_expires_at}")]
+pub struct Unlocked {
+    /// Instant used when releasing a currently locked execution back to
+    /// [`PendingState::PendingAt`]. This field keeps the released JSON and gRPC name.
+    pub backoff_expires_at: DateTime<Utc>,
+    #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
+    pub reason: StrVariant,
 }
 
-impl From<&'static str> for UnlockedReason {
-    fn from(value: &'static str) -> Self {
-        Self::Other {
-            reason: StrVariant::Static(value),
-        }
-    }
+#[derive(
+    Clone, Debug, PartialEq, Eq, derive_more::Display, Serialize, Deserialize, schemars::JsonSchema,
+)]
+#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ComponentUpgradeOutcome {
+    #[display("success({reason})")]
+    Success { reason: ComponentUpgradeReason },
+    #[display("failed: {reason}")]
+    Failed {
+        #[cfg_attr(any(test, feature = "test"), arbitrary(value = StrVariant::Static("reason")))]
+        reason: StrVariant,
+    },
 }
 
 impl ExecutionRequest {
@@ -551,8 +473,8 @@ impl ExecutionRequest {
         match self {
             ExecutionRequest::Created { .. } => "created",
             ExecutionRequest::Locked(_) => "locked",
-            ExecutionRequest::Unlocked { .. } => "unlocked",
-            ExecutionRequest::ComponentUpgraded { .. } => "component_upgraded",
+            ExecutionRequest::Unlocked(_) => "unlocked",
+            ExecutionRequest::ComponentUpgradeFinished { .. } => "component_upgrade_finished",
             ExecutionRequest::TemporarilyFailed { .. } => "temporarily_failed",
             ExecutionRequest::TemporarilyTimedOut { .. } => "temporarily_timed_out",
             ExecutionRequest::Finished { .. } => "finished",
@@ -1575,8 +1497,8 @@ pub trait DbExternalApi: DbConnection {
     ) -> Result<Vec<DeploymentRecord>, DbErrorRead>;
 
     /// Pause an execution. Only pending executions can be paused.
-    /// If the execution is currently locked, implementations must release the lock first
-    /// and then record the paused state.
+    /// If the execution is an activity and is currently in `PendingState::Locked`, implementations must
+    /// append `ExecutionRequest::Unlocked` to avoid affecting failed attempt count, before appending `ExecutionRequest::Paused`.
     async fn pause_execution(
         &self,
         execution_id: &ExecutionId,
@@ -2693,10 +2615,8 @@ mod tests {
     use super::PendingStateFinished;
     use super::PendingStateFinishedError;
     use super::PendingStateFinishedResultKind;
-    use super::UnlockedReason;
     use crate::ExecutionFailureKind;
     use crate::JoinSetId;
-    use crate::StrVariant;
     use crate::SupportedFunctionReturnValue;
     use chrono::DateTime;
     use chrono::Datelike;
@@ -2738,38 +2658,6 @@ mod tests {
         assert_eq!(expected, actual);
     }
 
-    #[test]
-    fn unlocked_reason_should_deserialize_legacy_string() {
-        let actual: UnlockedReason = serde_json::from_str(r#""executor closing""#).unwrap();
-        assert_eq!(
-            UnlockedReason::Other {
-                reason: StrVariant::Static("executor closing"),
-            },
-            actual
-        );
-    }
-
-    #[test]
-    fn unlocked_reason_other_should_serialize_current_tagged_shape() {
-        let actual = serde_json::to_string(&UnlockedReason::Other {
-            reason: StrVariant::Static("executor closing"),
-        })
-        .unwrap();
-        assert_eq!(r#"{"type":"other","reason":"executor closing"}"#, actual);
-    }
-
-    #[test]
-    fn unlocked_reason_should_deserialize_current_tagged_shape() {
-        let actual: UnlockedReason =
-            serde_json::from_str(r#"{"type":"other","reason":"executor closing"}"#).unwrap();
-        assert_eq!(
-            UnlockedReason::Other {
-                reason: StrVariant::Static("executor closing"),
-            },
-            actual
-        );
-    }
-
     #[test]
     fn join_set_deser_with_result_ok_option_none_should_work() {
         let expected = SupportedFunctionReturnValue::Ok(Some(WastValWithType {