studio/wfe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Releases 10 Activity

feat(persistence): name column, name lookup, definition sequence counter

Browse Source 
Land the `name` field and `next_definition_sequence` counter in the
two real persistence backends. Both providers:

* Add `name TEXT NOT NULL UNIQUE` to the `workflows` table.
* Add a `definition_sequences` table (`definition_id, next_num`) with
  an atomic UPSERT + RETURNING to give the host a race-free monotonic
  counter for `{def_id}-{N}` name generation.
* INSERT/UPDATE queries now include `name`; SELECT row parsers hydrate
  it back onto `WorkflowInstance`.
* New `get_workflow_instance_by_name` method for name-based lookups
  used by grpc handlers.

Postgres includes a DO-block migration that back-fills `name` from
`id` on pre-existing deployments so the NOT NULL + UNIQUE invariant
holds retroactively; callers can overwrite with a real name on the
next persist.
This commit is contained in:
Sienna Meridian Satterwhite
2026-04-07 18:58:25 +01:00
parent d9b9c5651e
commit 9af1a0d276
2 changed files with 228 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
wfe-postgres/src/lib.rs
View File

@@ -6,8 +6,8 @@ use sqlx::postgres::PgPoolOptions;
use sqlx::{PgPool, Row}; use sqlx::{PgPool, Row};
use wfe_core::models::{ use wfe_core::models::{
CommandName, Event, EventSubscription, ExecutionError, ExecutionPointer, ScheduledCommand, CommandName, Event, EventSubscription, ExecutionError, ExecutionPointer, PointerStatus,
WorkflowInstance, WorkflowStatus, PointerStatus, ScheduledCommand, WorkflowInstance, WorkflowStatus,
}; };
use wfe_core::traits::{ use wfe_core::traits::{
EventRepository, PersistenceProvider, ScheduledCommandRepository, SubscriptionRepository, EventRepository, PersistenceProvider, ScheduledCommandRepository, SubscriptionRepository,
@@ -57,7 +57,9 @@ impl PostgresPersistenceProvider {
"Suspended" => Ok(WorkflowStatus::Suspended), "Suspended" => Ok(WorkflowStatus::Suspended),
"Complete" => Ok(WorkflowStatus::Complete), "Complete" => Ok(WorkflowStatus::Complete),
"Terminated" => Ok(WorkflowStatus::Terminated), "Terminated" => Ok(WorkflowStatus::Terminated),
other => Err(WfeError::Persistence(format!("Unknown workflow status: {other}"))), other => Err(WfeError::Persistence(format!(
"Unknown workflow status: {other}"
))),
} }
} }
@@ -88,7 +90,9 @@ impl PostgresPersistenceProvider {
"Compensated" => Ok(PointerStatus::Compensated), "Compensated" => Ok(PointerStatus::Compensated),
"Cancelled" => Ok(PointerStatus::Cancelled), "Cancelled" => Ok(PointerStatus::Cancelled),
"PendingPredecessor" => Ok(PointerStatus::PendingPredecessor), "PendingPredecessor" => Ok(PointerStatus::PendingPredecessor),
other => Err(WfeError::Persistence(format!("Unknown pointer status: {other}"))), other => Err(WfeError::Persistence(format!(
"Unknown pointer status: {other}"
))),
} }
} }
@@ -103,7 +107,9 @@ impl PostgresPersistenceProvider {
match s { match s {
"ProcessWorkflow" => Ok(CommandName::ProcessWorkflow), "ProcessWorkflow" => Ok(CommandName::ProcessWorkflow),
"ProcessEvent" => Ok(CommandName::ProcessEvent), "ProcessEvent" => Ok(CommandName::ProcessEvent),
other => Err(WfeError::Persistence(format!("Unknown command name: {other}"))), other => Err(WfeError::Persistence(format!(
"Unknown command name: {other}"
))),
} }
} }
@@ -118,8 +124,9 @@ impl PostgresPersistenceProvider {
.map_err(|e| WfeError::Persistence(format!("Failed to serialize children: {e}")))?; .map_err(|e| WfeError::Persistence(format!("Failed to serialize children: {e}")))?;
let scope_json = serde_json::to_value(&p.scope) let scope_json = serde_json::to_value(&p.scope)
.map_err(|e| WfeError::Persistence(format!("Failed to serialize scope: {e}")))?; .map_err(|e| WfeError::Persistence(format!("Failed to serialize scope: {e}")))?;
let ext_json = serde_json::to_value(&p.extension_attributes) let ext_json = serde_json::to_value(&p.extension_attributes).map_err(|e| {
.map_err(|e| WfeError::Persistence(format!("Failed to serialize extension_attributes: {e}")))?; WfeError::Persistence(format!("Failed to serialize extension_attributes: {e}"))
})?;
sqlx::query( sqlx::query(
r#"INSERT INTO wfc.execution_pointers r#"INSERT INTO wfc.execution_pointers
@@ -158,13 +165,11 @@ impl PostgresPersistenceProvider {
} }
async fn load_pointers(&self, workflow_id: &str) -> Result<Vec<ExecutionPointer>> { async fn load_pointers(&self, workflow_id: &str) -> Result<Vec<ExecutionPointer>> {
let rows = sqlx::query( let rows = sqlx::query("SELECT * FROM wfc.execution_pointers WHERE workflow_id = $1")
"SELECT * FROM wfc.execution_pointers WHERE workflow_id = $1", .bind(workflow_id)
) .fetch_all(&self.pool)
.bind(workflow_id) .await
.fetch_all(&self.pool) .map_err(Self::map_sqlx_err)?;
.await
.map_err(Self::map_sqlx_err)?;
let mut pointers = Vec::with_capacity(rows.len()); let mut pointers = Vec::with_capacity(rows.len());
for row in &rows { for row in &rows {
@@ -183,8 +188,9 @@ impl PostgresPersistenceProvider {
let scope: Vec<String> = serde_json::from_value(scope_json) let scope: Vec<String> = serde_json::from_value(scope_json)
.map_err(|e| WfeError::Persistence(format!("Failed to deserialize scope: {e}")))?; .map_err(|e| WfeError::Persistence(format!("Failed to deserialize scope: {e}")))?;
let extension_attributes: HashMap<String, serde_json::Value> = let extension_attributes: HashMap<String, serde_json::Value> =
serde_json::from_value(ext_json) serde_json::from_value(ext_json).map_err(|e| {
.map_err(|e| WfeError::Persistence(format!("Failed to deserialize extension_attributes: {e}")))?; WfeError::Persistence(format!("Failed to deserialize extension_attributes: {e}"))
})?;
let status_str: String = row.get("status"); let status_str: String = row.get("status");
@@ -226,11 +232,12 @@ impl WorkflowRepository for PostgresPersistenceProvider {
sqlx::query( sqlx::query(
r#"INSERT INTO wfc.workflows r#"INSERT INTO wfc.workflows
(id, definition_id, version, description, reference, status, data, (id, name, definition_id, version, description, reference, status, data,
next_execution, create_time, complete_time) next_execution, create_time, complete_time)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)"#, VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)"#,
) )
.bind(&id) .bind(&id)
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i32) .bind(instance.version as i32)
.bind(&instance.description) .bind(&instance.description)
@@ -245,7 +252,8 @@ impl WorkflowRepository for PostgresPersistenceProvider {
.map_err(Self::map_sqlx_err)?; .map_err(Self::map_sqlx_err)?;
// Insert execution pointers // Insert execution pointers
self.insert_pointers(&mut tx, &id, &instance.execution_pointers).await?; self.insert_pointers(&mut tx, &id, &instance.execution_pointers)
.await?;
tx.commit().await.map_err(Self::map_sqlx_err)?; tx.commit().await.map_err(Self::map_sqlx_err)?;
Ok(id) Ok(id)
@@ -256,11 +264,12 @@ impl WorkflowRepository for PostgresPersistenceProvider {
sqlx::query( sqlx::query(
r#"UPDATE wfc.workflows SET r#"UPDATE wfc.workflows SET
definition_id=$2, version=$3, description=$4, reference=$5, name=$2, definition_id=$3, version=$4, description=$5, reference=$6,
status=$6, data=$7, next_execution=$8, create_time=$9, complete_time=$10 status=$7, data=$8, next_execution=$9, create_time=$10, complete_time=$11
WHERE id=$1"#, WHERE id=$1"#,
) )
.bind(&instance.id) .bind(&instance.id)
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i32) .bind(instance.version as i32)
.bind(&instance.description) .bind(&instance.description)
@@ -297,11 +306,12 @@ impl WorkflowRepository for PostgresPersistenceProvider {
sqlx::query( sqlx::query(
r#"UPDATE wfc.workflows SET r#"UPDATE wfc.workflows SET
definition_id=$2, version=$3, description=$4, reference=$5, name=$2, definition_id=$3, version=$4, description=$5, reference=$6,
status=$6, data=$7, next_execution=$8, create_time=$9, complete_time=$10 status=$7, data=$8, next_execution=$9, create_time=$10, complete_time=$11
WHERE id=$1"#, WHERE id=$1"#,
) )
.bind(&instance.id) .bind(&instance.id)
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i32) .bind(instance.version as i32)
.bind(&instance.description) .bind(&instance.description)
@@ -385,6 +395,7 @@ impl WorkflowRepository for PostgresPersistenceProvider {
Ok(WorkflowInstance { Ok(WorkflowInstance {
id: row.get("id"), id: row.get("id"),
name: row.get("name"),
workflow_definition_id: row.get("definition_id"), workflow_definition_id: row.get("definition_id"),
version: row.get::<i32, _>("version") as u32, version: row.get::<i32, _>("version") as u32,
description: row.get("description"), description: row.get("description"),
@@ -398,6 +409,35 @@ impl WorkflowRepository for PostgresPersistenceProvider {
}) })
} }
async fn get_workflow_instance_by_name(&self, name: &str) -> Result<WorkflowInstance> {
let row = sqlx::query("SELECT id FROM wfc.workflows WHERE name = $1")
.bind(name)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_sqlx_err)?
.ok_or_else(|| WfeError::WorkflowNotFound(name.to_string()))?;
let id: String = row.get("id");
self.get_workflow_instance(&id).await
}
async fn next_definition_sequence(&self, definition_id: &str) -> Result<u64> {
// UPSERT the counter atomically and return the new value. `RETURNING`
// gives us the post-increment number in a single round trip.
let row = sqlx::query(
r#"INSERT INTO wfc.definition_sequences (definition_id, next_num)
VALUES ($1, 1)
ON CONFLICT (definition_id) DO UPDATE
SET next_num = wfc.definition_sequences.next_num + 1
RETURNING next_num"#,
)
.bind(definition_id)
.fetch_one(&self.pool)
.await
.map_err(Self::map_sqlx_err)?;
let next: i64 = row.get("next_num");
Ok(next as u64)
}
async fn get_workflow_instances(&self, ids: &[String]) -> Result<Vec<WorkflowInstance>> { async fn get_workflow_instances(&self, ids: &[String]) -> Result<Vec<WorkflowInstance>> {
let mut result = Vec::new(); let mut result = Vec::new();
for id in ids { for id in ids {
@@ -413,10 +453,7 @@ impl WorkflowRepository for PostgresPersistenceProvider {
#[async_trait] #[async_trait]
impl SubscriptionRepository for PostgresPersistenceProvider { impl SubscriptionRepository for PostgresPersistenceProvider {
async fn create_event_subscription( async fn create_event_subscription(&self, subscription: &EventSubscription) -> Result<String> {
&self,
subscription: &EventSubscription,
) -> Result<String> {
let id = if subscription.id.is_empty() { let id = if subscription.id.is_empty() {
uuid::Uuid::new_v4().to_string() uuid::Uuid::new_v4().to_string()
} else { } else {
@@ -471,18 +508,14 @@ impl SubscriptionRepository for PostgresPersistenceProvider {
} }
async fn terminate_subscription(&self, subscription_id: &str) -> Result<()> { async fn terminate_subscription(&self, subscription_id: &str) -> Result<()> {
let result = sqlx::query( let result = sqlx::query("DELETE FROM wfc.event_subscriptions WHERE id = $1")
"DELETE FROM wfc.event_subscriptions WHERE id = $1", .bind(subscription_id)
) .execute(&self.pool)
.bind(subscription_id) .await
.execute(&self.pool) .map_err(Self::map_sqlx_err)?;
.await
.map_err(Self::map_sqlx_err)?;
if result.rows_affected() == 0 { if result.rows_affected() == 0 {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
Ok(()) Ok(())
} }
@@ -550,20 +583,14 @@ impl SubscriptionRepository for PostgresPersistenceProvider {
.await .await
.map_err(Self::map_sqlx_err)?; .map_err(Self::map_sqlx_err)?;
if exists.is_none() { if exists.is_none() {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
return Ok(false); return Ok(false);
} }
Ok(true) Ok(true)
} }
async fn clear_subscription_token( async fn clear_subscription_token(&self, subscription_id: &str, token: &str) -> Result<()> {
&self,
subscription_id: &str,
token: &str,
) -> Result<()> {
let result = sqlx::query( let result = sqlx::query(
r#"UPDATE wfc.event_subscriptions r#"UPDATE wfc.event_subscriptions
SET external_token = NULL, external_worker_id = NULL, external_token_expiry = NULL SET external_token = NULL, external_worker_id = NULL, external_token_expiry = NULL
@@ -576,9 +603,7 @@ impl SubscriptionRepository for PostgresPersistenceProvider {
.map_err(Self::map_sqlx_err)?; .map_err(Self::map_sqlx_err)?;
if result.rows_affected() == 0 { if result.rows_affected() == 0 {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
Ok(()) Ok(())
} }
@@ -731,20 +756,23 @@ impl ScheduledCommandRepository for PostgresPersistenceProvider {
async fn process_commands( async fn process_commands(
&self, &self,
as_of: DateTime<Utc>, as_of: DateTime<Utc>,
handler: &(dyn Fn(ScheduledCommand) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>> handler: &(
+ Send dyn Fn(
+ Sync), ScheduledCommand,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
+ Send
+ Sync
),
) -> Result<()> { ) -> Result<()> {
let as_of_millis = as_of.timestamp_millis(); let as_of_millis = as_of.timestamp_millis();
// 1. SELECT due commands (do not delete yet) // 1. SELECT due commands (do not delete yet)
let rows = sqlx::query( let rows = sqlx::query("SELECT * FROM wfc.scheduled_commands WHERE execute_time <= $1")
"SELECT * FROM wfc.scheduled_commands WHERE execute_time <= $1", .bind(as_of_millis)
) .fetch_all(&self.pool)
.bind(as_of_millis) .await
.fetch_all(&self.pool) .map_err(Self::map_sqlx_err)?;
.await
.map_err(Self::map_sqlx_err)?;
let commands: Vec<ScheduledCommand> = rows let commands: Vec<ScheduledCommand> = rows
.iter() .iter()
@@ -803,6 +831,7 @@ impl PersistenceProvider for PostgresPersistenceProvider {
sqlx::query( sqlx::query(
r#"CREATE TABLE IF NOT EXISTS wfc.workflows ( r#"CREATE TABLE IF NOT EXISTS wfc.workflows (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
definition_id TEXT NOT NULL, definition_id TEXT NOT NULL,
version INT NOT NULL, version INT NOT NULL,
description TEXT, description TEXT,
@@ -818,6 +847,39 @@ impl PersistenceProvider for PostgresPersistenceProvider {
.await .await
.map_err(Self::map_sqlx_err)?; .map_err(Self::map_sqlx_err)?;
// Upgrade older databases that lack the `name` column. Back-fill with
// the UUID so the NOT NULL + UNIQUE invariant holds retroactively;
// callers can re-run with a real name on the next persist.
sqlx::query(
r#"DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = 'wfc' AND table_name = 'workflows'
AND column_name = 'name'
) THEN
ALTER TABLE wfc.workflows ADD COLUMN name TEXT;
UPDATE wfc.workflows SET name = id WHERE name IS NULL;
ALTER TABLE wfc.workflows ALTER COLUMN name SET NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS idx_workflows_name
ON wfc.workflows (name);
END IF;
END$$;"#,
)
.execute(&self.pool)
.await
.map_err(Self::map_sqlx_err)?;
sqlx::query(
r#"CREATE TABLE IF NOT EXISTS wfc.definition_sequences (
definition_id TEXT PRIMARY KEY,
next_num BIGINT NOT NULL
)"#,
)
.execute(&self.pool)
.await
.map_err(Self::map_sqlx_err)?;
sqlx::query( sqlx::query(
r#"CREATE TABLE IF NOT EXISTS wfc.execution_pointers ( r#"CREATE TABLE IF NOT EXISTS wfc.execution_pointers (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,

189
wfe-sqlite/src/lib.rs
View File

@@ -57,6 +57,7 @@ impl SqlitePersistenceProvider {
sqlx::query( sqlx::query(
"CREATE TABLE IF NOT EXISTS workflows ( "CREATE TABLE IF NOT EXISTS workflows (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
definition_id TEXT NOT NULL, definition_id TEXT NOT NULL,
version INTEGER NOT NULL, version INTEGER NOT NULL,
description TEXT, description TEXT,
@@ -71,6 +72,17 @@ impl SqlitePersistenceProvider {
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
// Per-definition monotonic counter used to generate human-friendly
// instance names of the form `{definition_id}-{N}`.
sqlx::query(
"CREATE TABLE IF NOT EXISTS definition_sequences (
definition_id TEXT PRIMARY KEY,
next_num INTEGER NOT NULL
)",
)
.execute(&self.pool)
.await?;
sqlx::query( sqlx::query(
"CREATE TABLE IF NOT EXISTS execution_pointers ( "CREATE TABLE IF NOT EXISTS execution_pointers (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
@@ -157,30 +169,28 @@ impl SqlitePersistenceProvider {
.await?; .await?;
// Indexes // Indexes
sqlx::query("CREATE INDEX IF NOT EXISTS idx_workflows_next_execution ON workflows(next_execution)")
.execute(&self.pool)
.await?;
sqlx::query( sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_workflows_status ON workflows(status)", "CREATE INDEX IF NOT EXISTS idx_workflows_next_execution ON workflows(next_execution)",
) )
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_workflows_status ON workflows(status)")
.execute(&self.pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_execution_pointers_workflow_id ON execution_pointers(workflow_id)") sqlx::query("CREATE INDEX IF NOT EXISTS idx_execution_pointers_workflow_id ON execution_pointers(workflow_id)")
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_name_key ON events(event_name, event_key)") sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_events_name_key ON events(event_name, event_key)",
)
.execute(&self.pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_is_processed ON events(is_processed)")
.execute(&self.pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_event_time ON events(event_time)")
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_events_is_processed ON events(is_processed)",
)
.execute(&self.pool)
.await?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_events_event_time ON events(event_time)",
)
.execute(&self.pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_event_subscriptions_name_key ON event_subscriptions(event_name, event_key)") sqlx::query("CREATE INDEX IF NOT EXISTS idx_event_subscriptions_name_key ON event_subscriptions(event_name, event_key)")
.execute(&self.pool) .execute(&self.pool)
.await?; .await?;
@@ -226,10 +236,8 @@ fn row_to_workflow(
pointers: Vec<ExecutionPointer>, pointers: Vec<ExecutionPointer>,
) -> std::result::Result<WorkflowInstance, WfeError> { ) -> std::result::Result<WorkflowInstance, WfeError> {
let status_str: String = row.try_get("status").map_err(to_persistence_err)?; let status_str: String = row.try_get("status").map_err(to_persistence_err)?;
let status: WorkflowStatus = let status: WorkflowStatus = serde_json::from_str(&format!("\"{status_str}\""))
serde_json::from_str(&format!("\"{status_str}\"")).map_err(|e| { .map_err(|e| WfeError::Persistence(format!("Failed to deserialize WorkflowStatus: {e}")))?;
WfeError::Persistence(format!("Failed to deserialize WorkflowStatus: {e}"))
})?;
let data_str: String = row.try_get("data").map_err(to_persistence_err)?; let data_str: String = row.try_get("data").map_err(to_persistence_err)?;
let data: serde_json::Value = serde_json::from_str(&data_str) let data: serde_json::Value = serde_json::from_str(&data_str)
@@ -241,6 +249,7 @@ fn row_to_workflow(
Ok(WorkflowInstance { Ok(WorkflowInstance {
id: row.try_get("id").map_err(to_persistence_err)?, id: row.try_get("id").map_err(to_persistence_err)?,
name: row.try_get("name").map_err(to_persistence_err)?,
workflow_definition_id: row.try_get("definition_id").map_err(to_persistence_err)?, workflow_definition_id: row.try_get("definition_id").map_err(to_persistence_err)?,
version: row version: row
.try_get::<i64, _>("version") .try_get::<i64, _>("version")
@@ -272,10 +281,11 @@ fn row_to_pointer(
.as_deref() .as_deref()
.map(serde_json::from_str) .map(serde_json::from_str)
.transpose() .transpose()
.map_err(|e| WfeError::Persistence(format!("Failed to deserialize persistence_data: {e}")))?; .map_err(|e| {
WfeError::Persistence(format!("Failed to deserialize persistence_data: {e}"))
})?;
let event_data_str: Option<String> = let event_data_str: Option<String> = row.try_get("event_data").map_err(to_persistence_err)?;
row.try_get("event_data").map_err(to_persistence_err)?;
let event_data: Option<serde_json::Value> = event_data_str let event_data: Option<serde_json::Value> = event_data_str
.as_deref() .as_deref()
.map(serde_json::from_str) .map(serde_json::from_str)
@@ -308,15 +318,13 @@ fn row_to_pointer(
let ext_str: String = row let ext_str: String = row
.try_get("extension_attributes") .try_get("extension_attributes")
.map_err(to_persistence_err)?; .map_err(to_persistence_err)?;
let extension_attributes: HashMap<String, serde_json::Value> = let extension_attributes: HashMap<String, serde_json::Value> = serde_json::from_str(&ext_str)
serde_json::from_str(&ext_str).map_err(|e| { .map_err(|e| {
WfeError::Persistence(format!("Failed to deserialize extension_attributes: {e}")) WfeError::Persistence(format!("Failed to deserialize extension_attributes: {e}"))
})?; })?;
let sleep_until_str: Option<String> = let sleep_until_str: Option<String> = row.try_get("sleep_until").map_err(to_persistence_err)?;
row.try_get("sleep_until").map_err(to_persistence_err)?; let start_time_str: Option<String> = row.try_get("start_time").map_err(to_persistence_err)?;
let start_time_str: Option<String> =
row.try_get("start_time").map_err(to_persistence_err)?;
let end_time_str: Option<String> = row.try_get("end_time").map_err(to_persistence_err)?; let end_time_str: Option<String> = row.try_get("end_time").map_err(to_persistence_err)?;
Ok(ExecutionPointer { Ok(ExecutionPointer {
@@ -373,8 +381,7 @@ fn row_to_event(row: &sqlx::sqlite::SqliteRow) -> std::result::Result<Event, Wfe
fn row_to_subscription( fn row_to_subscription(
row: &sqlx::sqlite::SqliteRow, row: &sqlx::sqlite::SqliteRow,
) -> std::result::Result<EventSubscription, WfeError> { ) -> std::result::Result<EventSubscription, WfeError> {
let subscribe_as_of_str: String = let subscribe_as_of_str: String = row.try_get("subscribe_as_of").map_err(to_persistence_err)?;
row.try_get("subscribe_as_of").map_err(to_persistence_err)?;
let subscription_data_str: Option<String> = row let subscription_data_str: Option<String> = row
.try_get("subscription_data") .try_get("subscription_data")
@@ -436,10 +443,11 @@ impl WorkflowRepository for SqlitePersistenceProvider {
let mut tx = self.pool.begin().await.map_err(to_persistence_err)?; let mut tx = self.pool.begin().await.map_err(to_persistence_err)?;
sqlx::query( sqlx::query(
"INSERT INTO workflows (id, definition_id, version, description, reference, status, data, next_execution, create_time, complete_time) "INSERT INTO workflows (id, name, definition_id, version, description, reference, status, data, next_execution, create_time, complete_time)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
) )
.bind(&id) .bind(&id)
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i64) .bind(instance.version as i64)
.bind(&instance.description) .bind(&instance.description)
@@ -474,10 +482,11 @@ impl WorkflowRepository for SqlitePersistenceProvider {
let mut tx = self.pool.begin().await.map_err(to_persistence_err)?; let mut tx = self.pool.begin().await.map_err(to_persistence_err)?;
sqlx::query( sqlx::query(
"UPDATE workflows SET definition_id = ?1, version = ?2, description = ?3, reference = ?4, "UPDATE workflows SET name = ?1, definition_id = ?2, version = ?3, description = ?4, reference = ?5,
status = ?5, data = ?6, next_execution = ?7, complete_time = ?8 status = ?6, data = ?7, next_execution = ?8, complete_time = ?9
WHERE id = ?9", WHERE id = ?10",
) )
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i64) .bind(instance.version as i64)
.bind(&instance.description) .bind(&instance.description)
@@ -523,10 +532,11 @@ impl WorkflowRepository for SqlitePersistenceProvider {
let mut tx = self.pool.begin().await.map_err(to_persistence_err)?; let mut tx = self.pool.begin().await.map_err(to_persistence_err)?;
sqlx::query( sqlx::query(
"UPDATE workflows SET definition_id = ?1, version = ?2, description = ?3, reference = ?4, "UPDATE workflows SET name = ?1, definition_id = ?2, version = ?3, description = ?4, reference = ?5,
status = ?5, data = ?6, next_execution = ?7, complete_time = ?8 status = ?6, data = ?7, next_execution = ?8, complete_time = ?9
WHERE id = ?9", WHERE id = ?10",
) )
.bind(&instance.name)
.bind(&instance.workflow_definition_id) .bind(&instance.workflow_definition_id)
.bind(instance.version as i64) .bind(instance.version as i64)
.bind(&instance.description) .bind(&instance.description)
@@ -583,12 +593,11 @@ impl WorkflowRepository for SqlitePersistenceProvider {
.map_err(to_persistence_err)? .map_err(to_persistence_err)?
.ok_or_else(|| WfeError::WorkflowNotFound(id.to_string()))?; .ok_or_else(|| WfeError::WorkflowNotFound(id.to_string()))?;
let pointer_rows = let pointer_rows = sqlx::query("SELECT * FROM execution_pointers WHERE workflow_id = ?1")
sqlx::query("SELECT * FROM execution_pointers WHERE workflow_id = ?1") .bind(id)
.bind(id) .fetch_all(&self.pool)
.fetch_all(&self.pool) .await
.await .map_err(to_persistence_err)?;
.map_err(to_persistence_err)?;
let pointers = pointer_rows let pointers = pointer_rows
.iter() .iter()
@@ -598,6 +607,36 @@ impl WorkflowRepository for SqlitePersistenceProvider {
row_to_workflow(&row, pointers) row_to_workflow(&row, pointers)
} }
async fn get_workflow_instance_by_name(&self, name: &str) -> Result<WorkflowInstance> {
let row = sqlx::query("SELECT id FROM workflows WHERE name = ?1")
.bind(name)
.fetch_optional(&self.pool)
.await
.map_err(to_persistence_err)?
.ok_or_else(|| WfeError::WorkflowNotFound(name.to_string()))?;
let id: String = row.try_get("id").map_err(to_persistence_err)?;
self.get_workflow_instance(&id).await
}
async fn next_definition_sequence(&self, definition_id: &str) -> Result<u64> {
// SQLite doesn't support `INSERT ... ON CONFLICT ... RETURNING` prior
// to 3.35, but sqlx bundles a new-enough build. Emulate an atomic
// increment via UPSERT + RETURNING so concurrent callers don't collide.
let row = sqlx::query(
"INSERT INTO definition_sequences (definition_id, next_num)
VALUES (?1, 1)
ON CONFLICT(definition_id) DO UPDATE
SET next_num = next_num + 1
RETURNING next_num",
)
.bind(definition_id)
.fetch_one(&self.pool)
.await
.map_err(to_persistence_err)?;
let next: i64 = row.try_get("next_num").map_err(to_persistence_err)?;
Ok(next as u64)
}
async fn get_workflow_instances(&self, ids: &[String]) -> Result<Vec<WorkflowInstance>> { async fn get_workflow_instances(&self, ids: &[String]) -> Result<Vec<WorkflowInstance>> {
if ids.is_empty() { if ids.is_empty() {
return Ok(Vec::new()); return Ok(Vec::new());
@@ -735,10 +774,7 @@ async fn insert_subscription(
#[async_trait] #[async_trait]
impl SubscriptionRepository for SqlitePersistenceProvider { impl SubscriptionRepository for SqlitePersistenceProvider {
async fn create_event_subscription( async fn create_event_subscription(&self, subscription: &EventSubscription) -> Result<String> {
&self,
subscription: &EventSubscription,
) -> Result<String> {
let id = if subscription.id.is_empty() { let id = if subscription.id.is_empty() {
uuid::Uuid::new_v4().to_string() uuid::Uuid::new_v4().to_string()
} else { } else {
@@ -776,18 +812,14 @@ impl SubscriptionRepository for SqlitePersistenceProvider {
} }
async fn terminate_subscription(&self, subscription_id: &str) -> Result<()> { async fn terminate_subscription(&self, subscription_id: &str) -> Result<()> {
let result = sqlx::query( let result = sqlx::query("UPDATE event_subscriptions SET terminated = 1 WHERE id = ?1")
"UPDATE event_subscriptions SET terminated = 1 WHERE id = ?1", .bind(subscription_id)
) .execute(&self.pool)
.bind(subscription_id) .await
.execute(&self.pool) .map_err(to_persistence_err)?;
.await
.map_err(to_persistence_err)?;
if result.rows_affected() == 0 { if result.rows_affected() == 0 {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
Ok(()) Ok(())
} }
@@ -860,20 +892,14 @@ impl SubscriptionRepository for SqlitePersistenceProvider {
.await .await
.map_err(to_persistence_err)?; .map_err(to_persistence_err)?;
if exists.is_none() { if exists.is_none() {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
return Ok(false); return Ok(false);
} }
Ok(true) Ok(true)
} }
async fn clear_subscription_token( async fn clear_subscription_token(&self, subscription_id: &str, token: &str) -> Result<()> {
&self,
subscription_id: &str,
token: &str,
) -> Result<()> {
let result = sqlx::query( let result = sqlx::query(
"UPDATE event_subscriptions "UPDATE event_subscriptions
SET external_token = NULL, external_worker_id = NULL, external_token_expiry = NULL SET external_token = NULL, external_worker_id = NULL, external_token_expiry = NULL
@@ -886,9 +912,7 @@ impl SubscriptionRepository for SqlitePersistenceProvider {
.map_err(to_persistence_err)?; .map_err(to_persistence_err)?;
if result.rows_affected() == 0 { if result.rows_affected() == 0 {
return Err(WfeError::SubscriptionNotFound( return Err(WfeError::SubscriptionNotFound(subscription_id.to_string()));
subscription_id.to_string(),
));
} }
Ok(()) Ok(())
} }
@@ -937,13 +961,11 @@ impl EventRepository for SqlitePersistenceProvider {
async fn get_runnable_events(&self, as_at: DateTime<Utc>) -> Result<Vec<String>> { async fn get_runnable_events(&self, as_at: DateTime<Utc>) -> Result<Vec<String>> {
let as_at_str = dt_to_string(&as_at); let as_at_str = dt_to_string(&as_at);
let rows = sqlx::query( let rows = sqlx::query("SELECT id FROM events WHERE is_processed = 0 AND event_time <= ?1")
"SELECT id FROM events WHERE is_processed = 0 AND event_time <= ?1", .bind(&as_at_str)
) .fetch_all(&self.pool)
.bind(&as_at_str) .await
.fetch_all(&self.pool) .map_err(to_persistence_err)?;
.await
.map_err(to_persistence_err)?;
rows.iter() rows.iter()
.map(|r| r.try_get("id").map_err(to_persistence_err)) .map(|r| r.try_get("id").map_err(to_persistence_err))
@@ -1029,9 +1051,14 @@ impl ScheduledCommandRepository for SqlitePersistenceProvider {
async fn process_commands( async fn process_commands(
&self, &self,
as_of: DateTime<Utc>, as_of: DateTime<Utc>,
handler: &(dyn Fn(ScheduledCommand) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>> handler: &(
+ Send dyn Fn(
+ Sync), ScheduledCommand,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
+ Send
+ Sync
),
) -> Result<()> { ) -> Result<()> {
let as_of_millis = as_of.timestamp_millis(); let as_of_millis = as_of.timestamp_millis();