commit b014dbfc4a96149d5e027deefe3655f780e401d0
parent 958b9f6c1c056fe4a8a73ad80b3da69d0f50fda8
Author: triesap <tyson@radroots.org>
Date: Thu, 9 Jul 2026 04:31:50 +0000
workflow: preserve no-wait delivery semantics
Diffstat:
4 files changed, 114 insertions(+), 21 deletions(-)
diff --git a/crates/sdk/src/sync_runtime.rs b/crates/sdk/src/sync_runtime.rs
@@ -1047,14 +1047,20 @@ async fn proxy_delivery_policy(
.iter()
.filter(|target| target.delivery_plan_id == active_delivery_plan_id)
.collect::<Vec<_>>();
- let satisfied_count = active_targets
- .iter()
- .filter(|target| {
- target
- .status
- .counts_as_transport_satisfaction(plan.satisfaction_policy.class())
+ let satisfied_count = plan
+ .satisfaction_policy
+ .target_satisfaction_class()
+ .map(|satisfaction_class| {
+ active_targets
+ .iter()
+ .filter(|target| {
+ target
+ .status
+ .counts_as_transport_satisfaction(satisfaction_class)
+ })
+ .count()
})
- .count();
+ .unwrap_or(0);
let ready_target_count = active_targets
.iter()
.filter(|target| target.status.is_ready_for_attempt())
@@ -1077,6 +1083,7 @@ fn proxy_delivery_policy_from_remaining(
return Ok(TransportPublishDeliveryPolicy::Any);
}
Ok(match satisfaction_policy {
+ RadrootsTransportSatisfactionPolicy::NoWait => TransportPublishDeliveryPolicy::Any,
RadrootsTransportSatisfactionPolicy::Any { .. } => TransportPublishDeliveryPolicy::Any,
RadrootsTransportSatisfactionPolicy::All { .. } => TransportPublishDeliveryPolicy::All,
RadrootsTransportSatisfactionPolicy::Quorum { .. } => {
diff --git a/crates/sdk/src/workflow_runtime.rs b/crates/sdk/src/workflow_runtime.rs
@@ -17,7 +17,9 @@ use radroots_outbox::{
RadrootsOutboxDeliveryPlanInput, RadrootsOutboxEnqueueStatus,
RadrootsOutboxReticulumPreviewBehavior, RadrootsOutboxSignedOperationInput,
};
-use radroots_transport::{RadrootsTransportKind, RadrootsTransportSatisfactionPolicy};
+use radroots_transport::{
+ RadrootsTransportKind, RadrootsTransportSatisfactionPolicy, RadrootsTransportTarget,
+};
const SDK_LOCAL_EVENT_ENDPOINT_URI: &str = "local:sdk";
@@ -166,20 +168,26 @@ fn resolved_delivery_plan(
satisfaction_policy: SatisfactionPolicy,
) -> Result<SdkResolvedDeliveryPlan, RadrootsSdkError> {
match target_policy {
- TargetPolicy::Explicit(target_policy) => delivery_plan_from_target_set(
+ TargetPolicy::Explicit(target_policy) => delivery_plan_from_targets(
"explicit",
- target_policy.clone(),
+ target_policy.clone().into_targets(),
satisfaction_policy,
RadrootsOutboxReticulumPreviewBehavior::RejectDeliveryAttempts,
),
TargetPolicy::UseTransportProfile => {
let transport_profile = sdk.transport_profile();
- let target_set = transport_profile.target_set()?.ok_or_else(|| {
- RadrootsSdkError::empty_transport_targets("publish transport profile")
- })?;
- delivery_plan_from_target_set(
+ let targets = transport_profile
+ .target_set()?
+ .map(TargetSet::into_targets)
+ .unwrap_or_default();
+ if targets.is_empty() && satisfaction_policy != SatisfactionPolicy::NoWait {
+ return Err(RadrootsSdkError::empty_transport_targets(
+ "publish transport profile",
+ ));
+ }
+ delivery_plan_from_targets(
transport_profile.transport_profile_id(),
- target_set,
+ targets,
satisfaction_policy,
outbox_reticulum_preview_behavior(transport_profile),
)
@@ -187,9 +195,9 @@ fn resolved_delivery_plan(
}
}
-fn delivery_plan_from_target_set(
+fn delivery_plan_from_targets(
transport_profile_id: impl Into<String>,
- target_set: TargetSet,
+ targets: Vec<RadrootsTransportTarget>,
satisfaction_policy: SatisfactionPolicy,
reticulum_preview_behavior: RadrootsOutboxReticulumPreviewBehavior,
) -> Result<SdkResolvedDeliveryPlan, RadrootsSdkError> {
@@ -197,7 +205,7 @@ fn delivery_plan_from_target_set(
transport_profile_id,
1,
transport_satisfaction_policy(satisfaction_policy),
- target_set.into_targets(),
+ targets,
)
.with_reticulum_preview_behavior(reticulum_preview_behavior);
Ok(SdkResolvedDeliveryPlan { delivery_plan })
@@ -207,9 +215,8 @@ fn transport_satisfaction_policy(
satisfaction_policy: SatisfactionPolicy,
) -> RadrootsTransportSatisfactionPolicy {
match satisfaction_policy {
- SatisfactionPolicy::NoWait | SatisfactionPolicy::AllTargets => {
- RadrootsTransportSatisfactionPolicy::all_accepted()
- }
+ SatisfactionPolicy::NoWait => RadrootsTransportSatisfactionPolicy::no_wait(),
+ SatisfactionPolicy::AllTargets => RadrootsTransportSatisfactionPolicy::all_accepted(),
SatisfactionPolicy::AtLeastOneTarget => RadrootsTransportSatisfactionPolicy::any_accepted(),
SatisfactionPolicy::AtLeast { required } => {
RadrootsTransportSatisfactionPolicy::quorum_accepted(required)
diff --git a/crates/sdk/tests/unit/sync_runtime_tests.rs b/crates/sdk/tests/unit/sync_runtime_tests.rs
@@ -671,6 +671,11 @@ async fn proxy_push_empty_queue_and_private_helpers_are_deterministic() {
assert_eq!(receipt.attempted_events, 0);
assert_eq!(
+ proxy_delivery_policy_from_remaining(0, 0, &RadrootsTransportSatisfactionPolicy::no_wait())
+ .expect("no-wait proxy policy"),
+ TransportPublishDeliveryPolicy::Any
+ );
+ assert_eq!(
proxy_delivery_policy_from_remaining(
0,
0,
diff --git a/crates/sdk/tests/unit/workflow_runtime_tests.rs b/crates/sdk/tests/unit/workflow_runtime_tests.rs
@@ -239,6 +239,80 @@ async fn default_operation_idempotency_ignores_target_policy() {
}
#[tokio::test]
+async fn enqueue_signed_workflow_maps_no_wait_directly_and_allows_local_only_profile() {
+ let sdk = crate::RadrootsClient::builder()
+ .fixed_clock(crate::RadrootsSdkTimestamp::from_unix_seconds(
+ 1_700_000_013,
+ ))
+ .build()
+ .await
+ .expect("sdk");
+ let actor = RadrootsActorContext::test(FARMER_PUBLIC_KEY_HEX, [RadrootsActorRole::Farmer])
+ .expect("actor");
+ let signer = WorkflowSigner::new();
+ let draft = frozen_draft_for_d_tag(FARMER_PUBLIC_KEY_HEX, "workflow-no-wait");
+
+ let receipt = enqueue_signed_workflow(
+ &sdk,
+ SdkWorkflowEnqueueRequest {
+ operation_kind: "workflow.test.v1",
+ actor: &actor,
+ frozen_draft: &draft,
+ target_policy: TargetPolicy::use_transport_profile(),
+ satisfaction_policy: SatisfactionPolicy::NoWait,
+ idempotency_key: None,
+ },
+ &signer,
+ )
+ .await
+ .expect("no-wait enqueue");
+
+ let event = sdk
+ ._outbox
+ .get_event(receipt.outbox_event_id)
+ .await
+ .expect("event")
+ .expect("event");
+ let plans = sdk
+ ._outbox
+ .delivery_plans(receipt.outbox_event_id)
+ .await
+ .expect("plans");
+ let targets = sdk
+ ._outbox
+ .delivery_targets(receipt.outbox_event_id)
+ .await
+ .expect("targets");
+
+ assert_eq!(
+ event.state,
+ radroots_outbox::RadrootsOutboxEventState::Published
+ );
+ assert_eq!(plans.len(), 1);
+ assert_eq!(
+ plans[0].satisfaction_policy,
+ radroots_transport::RadrootsTransportSatisfactionPolicy::no_wait()
+ );
+ assert_ne!(
+ plans[0].satisfaction_policy,
+ radroots_transport::RadrootsTransportSatisfactionPolicy::all_accepted()
+ );
+ assert_eq!(plans[0].required_success_count, 0);
+ assert_eq!(
+ plans[0].status,
+ radroots_outbox::RadrootsOutboxDeliveryPlanStatus::Complete
+ );
+ assert!(targets.is_empty());
+ assert!(
+ sdk._outbox
+ .claim_next_ready_signed_event("publisher", "claim-a", 2_000, 1_000)
+ .await
+ .expect("claim")
+ .is_none()
+ );
+}
+
+#[tokio::test]
async fn enqueue_signed_workflow_stores_signed_event_and_reports_idempotency_conflicts() {
let sdk = crate::RadrootsClient::builder()
.transport_profile(nostr_profile("wss://relay.example.com"))
