refactor(db)!: Make DelayId generation conflict free

Use same structure as with `ExecutionIdDerived` to avoid generating (seeded) random ULID
that might conflict in the database.

Author: tomasol

crates/concepts/src/lib.rs

@@ -883,7 +883,7 @@ impl Params {
 }
 
 pub mod prefixed_ulid {
-    use crate::{JoinSetId, JoinSetIdParseError};
+    use crate::{JoinSetId, JoinSetIdParseError, JoinSetKind};
     use serde_with::{DeserializeFromStr, SerializeDisplay};
     use std::{
         fmt::{Debug, Display},
@@ -1028,7 +1028,7 @@ pub mod prefixed_ulid {
     pub type ExecutorId = PrefixedUlid<prefix::Exr>;
     pub type ExecutionIdTopLevel = PrefixedUlid<prefix::E>;
     pub type RunId = PrefixedUlid<prefix::Run>;
-    pub type DelayId = PrefixedUlid<prefix::Delay>;
+    pub type DelayIdTopLevel = PrefixedUlid<prefix::Delay>;
 
     #[cfg(any(test, feature = "test"))]
     impl<'a, T> arbitrary::Arbitrary<'a> for PrefixedUlid<T> {
@@ -1147,26 +1147,32 @@ pub mod prefixed_ulid {
         }
     }
     impl FromStr for ExecutionIdDerived {
-        type Err = ExecutionIdDerivedParseError;
+        type Err = DerivedIdParseError;
 
         fn from_str(input: &str) -> Result<Self, Self::Err> {
-            if let Some((prefix, suffix)) = input.split_once(EXECUTION_ID_INFIX) {
-                let top_level = PrefixedUlid::from_str(prefix)
-                    .map_err(ExecutionIdDerivedParseError::PrefixedUlidParseError)?;
-                let Some((infix, idx)) = suffix.rsplit_once(EXECUTION_ID_JOIN_SET_INFIX) else {
-                    return Err(ExecutionIdDerivedParseError::SecondDelimiterNotFound);
-                };
-                let infix = Arc::from(infix);
-                let idx =
-                    u64::from_str(idx).map_err(ExecutionIdDerivedParseError::ParseIndexError)?;
-                Ok(ExecutionIdDerived {
-                    top_level,
-                    infix,
-                    idx,
-                })
-            } else {
-                Err(ExecutionIdDerivedParseError::FirstDelimiterNotFound)
-            }
+            let (top_level, infix, idx) = derived_from_str(input)?;
+            Ok(ExecutionIdDerived {
+                top_level,
+                infix,
+                idx,
+            })
+        }
+    }
+
+    fn derived_from_str<T: 'static>(
+        input: &str,
+    ) -> Result<(PrefixedUlid<T>, Arc<str>, u64), DerivedIdParseError> {
+        if let Some((prefix, suffix)) = input.split_once(EXECUTION_ID_INFIX) {
+            let top_level = PrefixedUlid::from_str(prefix)
+                .map_err(DerivedIdParseError::PrefixedUlidParseError)?;
+            let Some((infix, idx)) = suffix.rsplit_once(EXECUTION_ID_JOIN_SET_INFIX) else {
+                return Err(DerivedIdParseError::SecondDelimiterNotFound);
+            };
+            let infix = Arc::from(infix);
+            let idx = u64::from_str(idx).map_err(DerivedIdParseError::ParseIndexError)?;
+            Ok((top_level, infix, idx))
+        } else {
+            Err(DerivedIdParseError::FirstDelimiterNotFound)
         }
     }
 
@@ -1180,17 +1186,15 @@ pub mod prefixed_ulid {
     }
 
     #[derive(Debug, thiserror::Error)]
-    pub enum ExecutionIdDerivedParseError {
+    pub enum DerivedIdParseError {
         #[error(transparent)]
         PrefixedUlidParseError(PrefixedUlidParseError),
-        #[error("cannot parse derived execution id - delimiter `{EXECUTION_ID_INFIX}` not found")]
+        #[error("cannot parse derived id - delimiter `{EXECUTION_ID_INFIX}` not found")]
         FirstDelimiterNotFound,
-        #[error(
-            "cannot parse derived execution id - delimiter `{EXECUTION_ID_JOIN_SET_INFIX}` not found"
-        )]
+        #[error("cannot parse derived id - delimiter `{EXECUTION_ID_JOIN_SET_INFIX}` not found")]
         SecondDelimiterNotFound,
         #[error(
-            "cannot parse derived execution id - suffix after `{EXECUTION_ID_JOIN_SET_INFIX}` must be a number"
+            "cannot parse derived id - suffix after `{EXECUTION_ID_JOIN_SET_INFIX}` must be a number"
         )]
         ParseIndexError(ParseIntError),
     }
@@ -1291,16 +1295,16 @@ pub mod prefixed_ulid {
                 ExecutionIdDerived::from_str(input)
                     .map(ExecutionId::Derived)
                     .map_err(|err| match err {
-                        ExecutionIdDerivedParseError::FirstDelimiterNotFound => {
+                        DerivedIdParseError::FirstDelimiterNotFound => {
                             unreachable!("first delimiter checked")
                         }
-                        ExecutionIdDerivedParseError::SecondDelimiterNotFound => {
+                        DerivedIdParseError::SecondDelimiterNotFound => {
                             ExecutionIdParseError::SecondDelimiterNotFound
                         }
-                        ExecutionIdDerivedParseError::PrefixedUlidParseError(err) => {
+                        DerivedIdParseError::PrefixedUlidParseError(err) => {
                             ExecutionIdParseError::PrefixedUlidParseError(err)
                         }
-                        ExecutionIdDerivedParseError::ParseIndexError(err) => {
+                        DerivedIdParseError::ParseIndexError(err) => {
                             ExecutionIdParseError::ParseIndexError(err)
                         }
                     })
@@ -1354,6 +1358,94 @@ pub mod prefixed_ulid {
             Ok(ExecutionId::TopLevel(PrefixedUlid::arbitrary(u)?))
         }
     }
+
+    #[derive(Hash, PartialEq, Eq, PartialOrd, Ord, Clone, SerializeDisplay, DeserializeFromStr)]
+    pub struct DelayId {
+        top_level: DelayIdTopLevel,
+        infix: Arc<str>,
+        idx: u64,
+    }
+    impl DelayId {
+        #[must_use]
+        pub fn new_oneoff(execution_id: &ExecutionId, join_set_id: &JoinSetId) -> DelayId {
+            assert!(join_set_id.kind == JoinSetKind::OneOff);
+            let ExecutionIdDerived {
+                top_level,
+                infix,
+                idx,
+            } = execution_id.next_level(join_set_id);
+            let top_level = DelayIdTopLevel::new(top_level.ulid);
+            DelayId {
+                top_level,
+                infix,
+                idx,
+            }
+        }
+
+        #[must_use]
+        pub fn get_incremented(&self) -> Self {
+            Self {
+                top_level: self.top_level,
+                infix: self.infix.clone(),
+                idx: self.idx + 1,
+            }
+        }
+
+        fn display_or_debug(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            let DelayId {
+                top_level,
+                infix,
+                idx,
+            } = self;
+            write!(
+                f,
+                "{top_level}{EXECUTION_ID_INFIX}{infix}{EXECUTION_ID_JOIN_SET_INFIX}{idx}"
+            )
+        }
+    }
+
+    pub mod delay_impl {
+        use super::{DelayId, DerivedIdParseError, derived_from_str};
+        use std::{
+            fmt::{Debug, Display},
+            str::FromStr,
+        };
+
+        impl Debug for DelayId {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                self.display_or_debug(f)
+            }
+        }
+
+        impl Display for DelayId {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                self.display_or_debug(f)
+            }
+        }
+
+        impl FromStr for DelayId {
+            type Err = DerivedIdParseError;
+
+            fn from_str(input: &str) -> Result<Self, Self::Err> {
+                let (top_level, infix, idx) = derived_from_str(input)?;
+                Ok(DelayId {
+                    top_level,
+                    infix,
+                    idx,
+                })
+            }
+        }
+
+        #[cfg(any(test, feature = "test"))]
+        impl<'a> arbitrary::Arbitrary<'a> for DelayId {
+            fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
+                use super::{ExecutionId, JoinSetId};
+                let execution_id = ExecutionId::arbitrary(u)?;
+                let join_set_id = JoinSetId::arbitrary(u)?;
+                Ok(DelayId::new_oneoff(&execution_id, &join_set_id))
+            }
+        }
+    }
 }
 
 #[derive(

crates/concepts/src/rusqlite_ext.rs

@@ -1,4 +1,7 @@
-use crate::{ExecutionId, JoinSetId, prefixed_ulid::ExecutionIdDerived};
+use crate::{
+    ExecutionId, JoinSetId,
+    prefixed_ulid::{DelayId, ExecutionIdDerived},
+};
 use rusqlite::{
     ToSql,
     types::{FromSql, FromSqlError, ToSqlOutput},
@@ -58,3 +61,21 @@ impl FromSql for JoinSetId {
         })
     }
 }
+
+impl ToSql for DelayId {
+    fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
+        Ok(ToSqlOutput::from(self.to_string()))
+    }
+}
+impl FromSql for DelayId {
+    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
+        let str = value.as_str()?;
+        str.parse::<DelayId>().map_err(|err| {
+            error!(
+                backtrace = %std::backtrace::Backtrace::capture(),
+                "Cannot convert to DelayId value:`{str}` - {err:?}"
+            );
+            FromSqlError::InvalidType
+        })
+    }
+}

crates/concepts/src/storage.rs

@@ -203,29 +203,6 @@ pub enum JoinSetResponse {
     },
 }
 
-impl JoinSetResponse {
-    #[must_use]
-    pub fn delay_id(&self) -> Option<DelayId> {
-        if let JoinSetResponse::DelayFinished { delay_id } = self {
-            Some(*delay_id)
-        } else {
-            None
-        }
-    }
-
-    #[must_use]
-    pub fn child_execution_id(&self) -> Option<ExecutionIdDerived> {
-        if let JoinSetResponse::ChildExecutionFinished {
-            child_execution_id, ..
-        } = self
-        {
-            Some(child_execution_id.clone())
-        } else {
-            None
-        }
-    }
-}
-
 pub const DUMMY_CREATED: ExecutionEventInner = ExecutionEventInner::Created {
     ffqn: FunctionFqn::new_static("", ""),
     params: Params::empty(),

crates/db-mem/src/inmemory_dao.rs

@@ -445,8 +445,8 @@ mod index {
                 .collect::<HashMap<_, _>>();
             // Keep only open
             for responded in journal.responses.iter().filter_map(|e| {
-                if let JoinSetResponse::DelayFinished { delay_id } = e.event.event {
-                    Some((e.event.join_set_id.clone(), delay_id))
+                if let JoinSetResponse::DelayFinished { delay_id } = &e.event.event {
+                    Some((e.event.join_set_id.clone(), delay_id.clone()))
                 } else {
                     None
                 }

crates/db-sqlite/src/sqlite_dao.rs

@@ -236,7 +236,6 @@ type ResponseSubscribers = Arc<
 struct PrefixedUlidWrapper<T: 'static>(PrefixedUlid<T>);
 type ExecutorIdW = PrefixedUlidWrapper<concepts::prefixed_ulid::prefix::Exr>;
 type RunIdW = PrefixedUlidWrapper<concepts::prefixed_ulid::prefix::Run>;
-type DelayIdW = PrefixedUlidWrapper<concepts::prefixed_ulid::prefix::Delay>;
 
 impl<T: 'static> FromSql for PrefixedUlidWrapper<T> {
     fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
@@ -1688,7 +1687,7 @@ impl<S: Sleep> SqlitePool<S> {
                     },
             } => IndexAction::NoPendingStateChange(Some(DelayReq {
                 join_set_id: join_set_id.clone(),
-                delay_id: *delay_id,
+                delay_id: delay_id.clone(),
                 expires_at: *expires_at,
             })),
 
@@ -2094,13 +2093,13 @@ impl<S: Sleep> SqlitePool<S> {
         let created_at: DateTime<Utc> = row.get("created_at")?;
         let join_set_id = row.get::<_, JoinSetId>("join_set_id")?;
         let inner_res = match (
-            row.get::<_, Option<DelayIdW>>("delay_id")?,
+            row.get::<_, Option<DelayId>>("delay_id")?,
             row.get::<_, Option<ExecutionIdDerived>>("child_execution_id")?,
             row.get::<_, Option<VersionType>>("finished_version")?,
             row.get::<_, Option<JsonWrapper<ExecutionEventInner>>>("json_value")?,
         ) {
             (Some(delay_id), None, None, None) => Ok(JoinSetResponse::DelayFinished {
-                delay_id: delay_id.0,
+                delay_id,
             }),
             (None, Some(child_execution_id), Some(finished_version), Some(result)) => {
                 match result.0 {
@@ -2117,7 +2116,7 @@ impl<S: Sleep> SqlitePool<S> {
                 }
             }
             (delay, child, finished, result) => {
-                error!("Invalid row in t_join_set_response {id} - {:?} {child:?} {finished:?} {:?}", delay.map(|it|it.0), result.map(|it| it.0));
+                error!("Invalid row in t_join_set_response {id} - {:?} {child:?} {finished:?} {:?}", delay, result.map(|it| it.0));
                 Err(DbError::Specific(SpecificError::ConsistencyError(
                 StrVariant::Static("invalid row in t_join_set_response"),
             )))},
@@ -2779,7 +2778,7 @@ impl<S: Sleep> DbConnection for SqlitePool<S> {
                         |row| {
                             let execution_id = row.get("execution_id")?;
                             let join_set_id = row.get::<_, JoinSetId>("join_set_id")?;
-                            let delay_id = row.get::<_, DelayIdW>("delay_id")?.0;
+                            let delay_id = row.get::<_, DelayId>("delay_id")?;
                             Ok(ExpiredTimer::Delay { execution_id, join_set_id, delay_id })
                         },
                     ).map_err(convert_err)?

crates/executor/src/expired_timers_watcher.rs

@@ -173,8 +173,8 @@ pub(crate) async fn tick(
                 join_set_id,
                 delay_id,
             } => {
-                let event = JoinSetResponse::DelayFinished { delay_id };
                 debug!(%execution_id, %join_set_id, %delay_id, "Appending DelayFinishedAsyncResponse");
+                let event = JoinSetResponse::DelayFinished { delay_id };
                 let res = db_connection
                     .append_response(
                         executed_at,
@@ -183,7 +183,7 @@ pub(crate) async fn tick(
                     )
                     .await;
                 if let Err(err) = res {
-                    debug!(%execution_id, %delay_id, "Failed to update expired async timer - {err:?}");
+                    debug!(%execution_id, "Failed to update expired async timer - {err:?}");
                 } else {
                     expired_async_timers += 1;
                 }