mirror of
https://github.com/block/buzz.git
synced 2026-10-28 13:01:10 +00:00
fix(workflows): retire deleted definitions atomically
Signed-off-by: Diem Nguyen <diem@squareup.com>
This commit is contained in:
Diem Nguyen
@@ -50,7 +50,7 @@ pub use store::{
|
||||
admin_moderation, allowlist, api_token, archived_identities, channel, channel_members,
|
||||
community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push,
|
||||
reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
|
||||
replaceable, thread, usage, user, workflow,
|
||||
replaceable, thread, usage, user, workflow, workflow_deletion,
|
||||
};
|
||||
|
||||
pub use allowlist::AllowlistEntry;
|
||||
|
||||
@@ -54,3 +54,8 @@ pub mod usage;
|
||||
pub mod user;
|
||||
/// Workflow, run, and approval persistence.
|
||||
pub mod workflow;
|
||||
/// Atomic workflow deletion and definition retirement.
|
||||
pub mod workflow_deletion;
|
||||
|
||||
#[cfg(test)]
|
||||
mod workflow_deletion_postgres_tests;
|
||||
|
||||
79
crates/buzz-db/src/store/workflow_deletion.rs
Normal file
79
crates/buzz-db/src/store/workflow_deletion.rs
Normal file
@@ -0,0 +1,79 @@
|
||||
//! Transaction-bound deletion of executable workflows and signed definitions.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId};
|
||||
use chrono::{DateTime, Utc};
|
||||
use sqlx::{Postgres, Transaction};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{DbError, Result};
|
||||
|
||||
/// Retire a workflow coordinate under the same lock used by definition writes.
|
||||
///
|
||||
/// The caller must authorize the coordinate owner and commit the signed deletion
|
||||
/// in this transaction. A newer live definition preserves the executable row;
|
||||
/// missing rows are allowed so retries can repair definitions left by old relays.
|
||||
pub async fn delete_in_transaction(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
community: CommunityId,
|
||||
owner: &[u8],
|
||||
workflow_id: Uuid,
|
||||
coordinate: &str,
|
||||
deleted_at_secs: i64,
|
||||
) -> Result<Vec<Uuid>> {
|
||||
let cutoff = DateTime::<Utc>::from_timestamp(deleted_at_secs, 0)
|
||||
.ok_or(DbError::InvalidTimestamp(deleted_at_secs))?;
|
||||
let lock_key = super::replaceable::event_replacement_lock_key(
|
||||
community,
|
||||
KIND_WORKFLOW_DEF as i32,
|
||||
owner,
|
||||
Some(coordinate.as_bytes()),
|
||||
);
|
||||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||||
.bind(lock_key)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
|
||||
let newer_definition: bool = sqlx::query_scalar(
|
||||
"SELECT EXISTS(SELECT 1 FROM events \
|
||||
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \
|
||||
AND deleted_at IS NULL AND created_at > $5)",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(KIND_WORKFLOW_DEF as i32)
|
||||
.bind(owner)
|
||||
.bind(coordinate)
|
||||
.bind(cutoff)
|
||||
.fetch_one(&mut **tx)
|
||||
.await?;
|
||||
|
||||
let mut channels = BTreeSet::new();
|
||||
if !newer_definition {
|
||||
let removed: Option<Option<Uuid>> = sqlx::query_scalar(
|
||||
"DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 \
|
||||
RETURNING channel_id",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(workflow_id)
|
||||
.bind(owner)
|
||||
.fetch_optional(&mut **tx)
|
||||
.await?;
|
||||
channels.extend(removed.flatten());
|
||||
}
|
||||
|
||||
let definition_channels: Vec<Option<Uuid>> = sqlx::query_scalar(
|
||||
"UPDATE events SET deleted_at = NOW() \
|
||||
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \
|
||||
AND deleted_at IS NULL AND created_at <= $5 RETURNING channel_id",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(KIND_WORKFLOW_DEF as i32)
|
||||
.bind(owner)
|
||||
.bind(coordinate)
|
||||
.bind(cutoff)
|
||||
.fetch_all(&mut **tx)
|
||||
.await?;
|
||||
channels.extend(definition_channels.into_iter().flatten());
|
||||
Ok(channels.into_iter().collect())
|
||||
}
|
||||
199
crates/buzz-db/src/store/workflow_deletion_postgres_tests.rs
Normal file
199
crates/buzz-db/src/store/workflow_deletion_postgres_tests.rs
Normal file
@@ -0,0 +1,199 @@
|
||||
use super::workflow_deletion::delete_in_transaction;
|
||||
use buzz_core::CommunityId;
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
|
||||
use sqlx::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
struct Fixture {
|
||||
pool: PgPool,
|
||||
community: CommunityId,
|
||||
keys: Keys,
|
||||
id: Uuid,
|
||||
timestamp: i64,
|
||||
}
|
||||
|
||||
impl Fixture {
|
||||
async fn new() -> Self {
|
||||
let pool = PgPool::connect(&crate::test_support::database_url())
|
||||
.await
|
||||
.unwrap();
|
||||
let community = CommunityId::from_uuid(Uuid::new_v4());
|
||||
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
|
||||
.bind(community.as_uuid())
|
||||
.bind(format!("deletion-{}.example", community.as_uuid()))
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let keys = Keys::generate();
|
||||
crate::user::ensure_user(&pool, community, &keys.public_key().to_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
let f = Self {
|
||||
pool,
|
||||
community,
|
||||
keys,
|
||||
id: Uuid::new_v4(),
|
||||
timestamp: Timestamp::now().as_secs() as i64,
|
||||
};
|
||||
f.seed(f.community).await;
|
||||
f
|
||||
}
|
||||
|
||||
async fn seed(&self, community: CommunityId) {
|
||||
crate::user::ensure_user(&self.pool, community, &self.keys.public_key().to_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
crate::workflow::upsert_workflow(
|
||||
&self.pool,
|
||||
community,
|
||||
self.id,
|
||||
None,
|
||||
&self.keys.public_key().to_bytes(),
|
||||
"deletion-test",
|
||||
"{}",
|
||||
&[0; 32],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let event = EventBuilder::new(Kind::Custom(30620), "definition")
|
||||
.tags([Tag::parse(["d", &self.id.to_string()]).unwrap()])
|
||||
.custom_created_at(Timestamp::from(self.timestamp as u64))
|
||||
.sign_with_keys(&self.keys)
|
||||
.unwrap();
|
||||
crate::event::insert_event(&self.pool, community, &event, None)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn counts(&self, community: CommunityId) -> (i64, i64) {
|
||||
let runtime = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM workflows WHERE community_id = $1 AND id = $2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(self.id)
|
||||
.fetch_one(&self.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let definitions = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id = $1 AND kind = 30620 AND d_tag = $2 AND deleted_at IS NULL")
|
||||
.bind(community.as_uuid()).bind(self.id.to_string()).fetch_one(&self.pool).await.unwrap();
|
||||
(runtime, definitions)
|
||||
}
|
||||
|
||||
async fn delete(&self, owner: &[u8], timestamp: i64, commit: bool) {
|
||||
let mut tx = self.pool.begin().await.unwrap();
|
||||
delete_in_transaction(
|
||||
&mut tx,
|
||||
self.community,
|
||||
owner,
|
||||
self.id,
|
||||
&self.id.to_string(),
|
||||
timestamp,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
if commit {
|
||||
tx.commit().await.unwrap();
|
||||
} else {
|
||||
tx.rollback().await.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn rollback_preserves_both_rows_and_commit_retires_both() {
|
||||
let f = Fixture::new().await;
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp, false)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (1, 1));
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (0, 0));
|
||||
assert!(
|
||||
crate::workflow::claim_scheduled_workflow_fire(
|
||||
&f.pool,
|
||||
f.community,
|
||||
f.id,
|
||||
chrono::Utc::now()
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"deleted workflow cannot acquire a scheduled execution claim"
|
||||
);
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (0, 0));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn missing_runtime_still_retires_orphan_definition() {
|
||||
let f = Fixture::new().await;
|
||||
sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2")
|
||||
.bind(f.community.as_uuid())
|
||||
.bind(f.id)
|
||||
.execute(&f.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(f.counts(f.community).await, (0, 1));
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (0, 0));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn owner_timestamp_and_community_boundaries_preserve_unrelated_rows() {
|
||||
let f = Fixture::new().await;
|
||||
let other = Fixture::new().await;
|
||||
f.seed(other.community).await;
|
||||
f.delete(&other.keys.public_key().to_bytes(), f.timestamp, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (1, 1));
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp - 1, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (1, 1));
|
||||
f.delete(&f.keys.public_key().to_bytes(), f.timestamp, true)
|
||||
.await;
|
||||
assert_eq!(f.counts(f.community).await, (0, 0));
|
||||
assert_eq!(f.counts(other.community).await, (1, 1));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn deletion_waits_for_the_definition_writer_coordinate_lock() {
|
||||
let f = Fixture::new().await;
|
||||
let mut writer = f.pool.begin().await.unwrap();
|
||||
let lock = super::replaceable::event_replacement_lock_key(
|
||||
f.community,
|
||||
30620,
|
||||
&f.keys.public_key().to_bytes(),
|
||||
Some(f.id.to_string().as_bytes()),
|
||||
);
|
||||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||||
.bind(lock)
|
||||
.execute(&mut *writer)
|
||||
.await
|
||||
.unwrap();
|
||||
let owner = f.keys.public_key().to_bytes();
|
||||
let coordinate = f.id.to_string();
|
||||
let mut tx = f.pool.begin().await.unwrap();
|
||||
{
|
||||
let deletion =
|
||||
delete_in_transaction(&mut tx, f.community, &owner, f.id, &coordinate, f.timestamp);
|
||||
tokio::pin!(deletion);
|
||||
assert!(
|
||||
tokio::time::timeout(std::time::Duration::from_millis(100), &mut deletion)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
writer.commit().await.unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), deletion)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
}
|
||||
tx.commit().await.unwrap();
|
||||
assert_eq!(f.counts(f.community).await, (0, 0));
|
||||
}
|
||||
@@ -3144,7 +3144,12 @@ async fn ingest_event_inner(
|
||||
});
|
||||
}
|
||||
|
||||
let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) {
|
||||
let workflow_deletion =
|
||||
super::workflow_deletion::persist(tenant, state, &event, channel_id).await?;
|
||||
let workflow_deletion_applied = workflow_deletion.is_some();
|
||||
let (stored_event, was_inserted) = if let Some(result) = workflow_deletion {
|
||||
result
|
||||
} else if buzz_core::kind::is_replaceable(kind_u32) {
|
||||
// NIP-16 replaceable event — atomic replace with stale-write protection.
|
||||
// channel_id is None for global kinds (0, 1, 3) due to step 5b above.
|
||||
state
|
||||
@@ -3211,7 +3216,7 @@ async fn ingest_event_inner(
|
||||
});
|
||||
}
|
||||
|
||||
if crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
|
||||
if !workflow_deletion_applied && crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
|
||||
if let Err(e) =
|
||||
crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)
|
||||
.await
|
||||
|
||||
@@ -40,6 +40,7 @@ pub mod report_resolution;
|
||||
pub mod req;
|
||||
/// NIP-29 and NIP-25 side-effect handlers.
|
||||
pub mod side_effects;
|
||||
mod workflow_deletion;
|
||||
|
||||
/// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag,
|
||||
/// applying the server-side override when configured.
|
||||
|
||||
83
crates/buzz-relay/src/handlers/workflow_deletion.rs
Normal file
83
crates/buzz-relay/src/handlers/workflow_deletion.rs
Normal file
@@ -0,0 +1,83 @@
|
||||
//! Canonical workflow deletes commit their signed event and effects together.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use buzz_core::{kind::KIND_DELETION, kind::KIND_WORKFLOW_DEF, tenant::TenantContext, StoredEvent};
|
||||
use nostr::Event;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::ingest::IngestError;
|
||||
use crate::state::AppState;
|
||||
|
||||
/// Persist an already-authorized canonical workflow deletion, if applicable.
|
||||
/// Legacy name coordinates retain their existing handler.
|
||||
pub(super) async fn persist(
|
||||
tenant: &TenantContext,
|
||||
state: &Arc<AppState>,
|
||||
event: &Event,
|
||||
channel_id: Option<Uuid>,
|
||||
) -> Result<Option<(StoredEvent, bool)>, IngestError> {
|
||||
if event.kind.as_u16() as u32 != KIND_DELETION {
|
||||
return Ok(None);
|
||||
}
|
||||
let Some(address) = event.tags.iter().find_map(|tag| {
|
||||
let parts = tag.as_slice();
|
||||
(parts.first().is_some_and(|part| part == "a"))
|
||||
.then(|| parts.get(1).map(String::as_str))
|
||||
.flatten()
|
||||
}) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let mut parts = address.splitn(3, ':');
|
||||
if parts.next().and_then(|kind| kind.parse::<u32>().ok()) != Some(KIND_WORKFLOW_DEF) {
|
||||
return Ok(None);
|
||||
}
|
||||
let owner_hex = parts.next().unwrap_or_default();
|
||||
let coordinate = parts.next().unwrap_or_default();
|
||||
let Ok(workflow_id) = Uuid::parse_str(coordinate) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let owner = hex::decode(owner_hex)
|
||||
.map_err(|_| IngestError::Rejected("invalid: workflow coordinate owner".into()))?;
|
||||
|
||||
let mut tx = state
|
||||
.db
|
||||
.begin_event_write_transaction()
|
||||
.await
|
||||
.map_err(database_error)?;
|
||||
buzz_deletion::store(&state.db)
|
||||
.guard_transaction(&mut tx, tenant.community())
|
||||
.await
|
||||
.map_err(|error| {
|
||||
IngestError::Rejected(format!("restricted: community writes are fenced: {error}"))
|
||||
})?;
|
||||
let channels = buzz_db::workflow_deletion::delete_in_transaction(
|
||||
&mut tx,
|
||||
tenant.community(),
|
||||
&owner,
|
||||
workflow_id,
|
||||
coordinate,
|
||||
event.created_at.as_secs() as i64,
|
||||
)
|
||||
.await
|
||||
.map_err(database_error)?;
|
||||
// Do not skip effects for an exact replay: older relays may have accepted
|
||||
// the deletion while leaving a live definition behind.
|
||||
let stored =
|
||||
buzz_db::event::insert_event_in_transaction(&mut tx, tenant.community(), event, channel_id)
|
||||
.await
|
||||
.map_err(database_error)?;
|
||||
tx.commit().await.map_err(|error| {
|
||||
IngestError::Internal(format!("error: committing workflow deletion: {error}"))
|
||||
})?;
|
||||
for channel in channels {
|
||||
state
|
||||
.workflow_engine
|
||||
.invalidate_channel_workflows(tenant.community(), channel);
|
||||
}
|
||||
Ok(Some(stored))
|
||||
}
|
||||
|
||||
fn database_error(error: buzz_db::DbError) -> IngestError {
|
||||
IngestError::Internal(format!("error: persisting workflow deletion: {error}"))
|
||||
}
|
||||
184
crates/buzz-test-client/tests/e2e_workflow_deletion.rs
Normal file
184
crates/buzz-test-client/tests/e2e_workflow_deletion.rs
Normal file
@@ -0,0 +1,184 @@
|
||||
//! Real ingest/query regressions. Use only an isolated relay and test accounts.
|
||||
//! Set RELAY_URL, BUZZ_TEST_OWNER_PRIVATE_KEY, BUZZ_TEST_MEMBER_PRIVATE_KEY,
|
||||
//! and BUZZ_TEST_CHANNEL_ID, then run this ignored test target.
|
||||
|
||||
use base64::{engine::general_purpose::STANDARD, Engine};
|
||||
use nostr::{Event, EventBuilder, Keys, Kind, Tag, Timestamp};
|
||||
use serde_json::{json, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
use uuid::Uuid;
|
||||
|
||||
struct Fixture {
|
||||
url: String,
|
||||
owner: Keys,
|
||||
member: Keys,
|
||||
channel: Uuid,
|
||||
}
|
||||
|
||||
impl Fixture {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
url: std::env::var("RELAY_URL")
|
||||
.expect("isolated RELAY_URL")
|
||||
.replace("ws://", "http://")
|
||||
.replace("wss://", "https://")
|
||||
.trim_end_matches('/')
|
||||
.to_owned(),
|
||||
owner: Keys::parse(&std::env::var("BUZZ_TEST_OWNER_PRIVATE_KEY").unwrap()).unwrap(),
|
||||
member: Keys::parse(&std::env::var("BUZZ_TEST_MEMBER_PRIVATE_KEY").unwrap()).unwrap(),
|
||||
channel: std::env::var("BUZZ_TEST_CHANNEL_ID")
|
||||
.unwrap()
|
||||
.parse()
|
||||
.unwrap(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn post(&self, keys: &Keys, path: &str, body: Value) -> Value {
|
||||
let url = format!("{}{path}", self.url);
|
||||
let body = if path == "/query" {
|
||||
json!([body])
|
||||
} else {
|
||||
body
|
||||
};
|
||||
let body = body.to_string();
|
||||
let auth = EventBuilder::new(Kind::Custom(27235), "")
|
||||
.tags([
|
||||
Tag::parse(["u", url.as_str()]).unwrap(),
|
||||
Tag::parse(["method", "POST"]).unwrap(),
|
||||
Tag::parse(["payload", &hex::encode(Sha256::digest(body.as_bytes()))]).unwrap(),
|
||||
Tag::parse(["nonce", &Uuid::new_v4().to_string()]).unwrap(),
|
||||
])
|
||||
.sign_with_keys(keys)
|
||||
.unwrap();
|
||||
reqwest::Client::new()
|
||||
.post(url)
|
||||
.header(
|
||||
"Authorization",
|
||||
format!(
|
||||
"Nostr {}",
|
||||
STANDARD.encode(serde_json::to_vec(&auth).unwrap())
|
||||
),
|
||||
)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(body)
|
||||
.send()
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn submit(&self, keys: &Keys, event: &Event) -> Value {
|
||||
self.post(keys, "/events", serde_json::to_value(event).unwrap())
|
||||
.await
|
||||
}
|
||||
|
||||
async fn accepted(&self, event: &Event) {
|
||||
let response = self.submit(&self.owner, event).await;
|
||||
assert_eq!(response["accepted"], true, "{response}");
|
||||
}
|
||||
|
||||
async fn definition(&self, id: Uuid) -> Value {
|
||||
self.post(
|
||||
&self.owner,
|
||||
"/query",
|
||||
json!({"kinds": [30620], "#d": [id.to_string()]}),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn deletion(&self, id: Uuid, timestamp: Timestamp) -> Event {
|
||||
buzz_sdk::build_workflow_delete(&self.owner.public_key().to_hex(), id)
|
||||
.unwrap()
|
||||
.custom_created_at(timestamp)
|
||||
.sign_with_keys(&self.owner)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn create(&self, id: Uuid, timestamp: Timestamp) {
|
||||
let event = buzz_sdk::build_workflow_def(self.channel, id,
|
||||
"name: delete-regression\ntrigger:\n on: schedule\n cron: '0 0 1 1 *'\nsteps:\n - id: marker\n action: send_message\n text: isolated deletion regression\n")
|
||||
.unwrap().custom_created_at(timestamp).sign_with_keys(&self.owner).unwrap();
|
||||
self.accepted(&event).await;
|
||||
assert_eq!(self.definition(id).await.as_array().unwrap().len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires isolated relay and two test accounts"]
|
||||
async fn deletion_removes_get_list_and_execution_and_replay_is_safe() {
|
||||
let f = Fixture::new();
|
||||
let id = Uuid::new_v4();
|
||||
let now = Timestamp::now();
|
||||
f.create(id, now).await;
|
||||
|
||||
// A different authenticated channel member cannot delete this coordinate.
|
||||
let deletion = f.deletion(id, now);
|
||||
let forged = EventBuilder::new(Kind::Custom(5), "")
|
||||
.tags(deletion.tags.clone())
|
||||
.sign_with_keys(&f.member)
|
||||
.unwrap();
|
||||
let rejected = f.submit(&f.member, &forged).await;
|
||||
assert_ne!(rejected["accepted"], true, "{rejected}");
|
||||
assert_eq!(f.definition(id).await.as_array().unwrap().len(), 1);
|
||||
|
||||
f.accepted(&deletion).await;
|
||||
for _ in 0..2 {
|
||||
assert_eq!(
|
||||
f.definition(id).await,
|
||||
json!([]),
|
||||
"get must omit deleted definition"
|
||||
);
|
||||
let list = f
|
||||
.post(
|
||||
&f.owner,
|
||||
"/query",
|
||||
json!({"kinds": [30620], "#h": [f.channel.to_string()]}),
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
!list.as_array().unwrap().iter().any(|event| {
|
||||
event["tags"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.any(|tag| tag == &json!(["d", id.to_string()]))
|
||||
}),
|
||||
"list must omit deleted definition"
|
||||
);
|
||||
let trigger = buzz_sdk::build_workflow_trigger(id)
|
||||
.unwrap()
|
||||
.sign_with_keys(&f.owner)
|
||||
.unwrap();
|
||||
let response = f.submit(&f.owner, &trigger).await;
|
||||
assert_ne!(
|
||||
response["accepted"], true,
|
||||
"deleted workflow must not execute: {response}"
|
||||
);
|
||||
assert!(
|
||||
response.to_string().contains("workflow not found"),
|
||||
"{response}"
|
||||
);
|
||||
f.accepted(&deletion).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires isolated relay and two test accounts"]
|
||||
async fn stale_deletion_preserves_newer_definition_and_runtime() {
|
||||
let f = Fixture::new();
|
||||
let id = Uuid::new_v4();
|
||||
let now = Timestamp::now();
|
||||
f.create(id, now).await;
|
||||
f.accepted(&f.deletion(id, Timestamp::from(now.as_secs() - 1)))
|
||||
.await;
|
||||
assert_eq!(f.definition(id).await.as_array().unwrap().len(), 1);
|
||||
let trigger = buzz_sdk::build_workflow_trigger(id)
|
||||
.unwrap()
|
||||
.sign_with_keys(&f.owner)
|
||||
.unwrap();
|
||||
f.accepted(&trigger).await;
|
||||
f.accepted(&f.deletion(id, now)).await;
|
||||
assert_eq!(f.definition(id).await, json!([]));
|
||||
}
|
||||
Reference in New Issue
Block a user