operonx 0.8.3

High-performance Rust execution backend for Operon workflows
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! Verify the scheduler honours the `Interrupt` event.
//!
//! Mirrors a subset of Python's Phase B test_interrupt.py — focused on the
//! observable behaviour: an op returning the synthetic
//! `{"__interrupt__": {...}}` shape causes the scheduler to abort other
//! in-flight tasks at the target ctx, and the run completes promptly.

use std::time::Duration;

use operonx::Operon;
use serde_json::{json, Map, Value};

/// Build a parallel two-branch graph: `[long_sleep, kick_interrupt]`,
/// both children of START, both feeding END. Uses `op_async` for the
/// long-sleep branch so cancellation can actually propagate via
/// `JoinHandle::abort()`.
fn parallel_graph_json() -> String {
    json!({
        "type": "graph",
        "name": "main",
        "full_name": "main",
        "func_name": null,
        "is_async": false,
        "is_generator": false,
        "enabled": true,
        "verbose": false,
        "stream": false,
        "bound": "sync",
        "inputs": {
            "seed": {
                "default": null, "required": true,
                "ref": null, "scratch": null, "literal": null
            }
        },
        "outputs": {},
        "ops": {
            "slow": {
                "type": "code",
                "name": "slow",
                "full_name": "main.slow",
                "func_name": "long_sleep",
                "is_async": true,
                "is_generator": false,
                "enabled": true,
                "verbose": false,
                "stream": false,
                "bound": "io",
                "inputs": {
                    "_seed": {
                        "default": null, "required": true,
                        "ref": {"source": "__PARENT__", "var": "seed",
                                "transforms": [], "is_output": false},
                        "scratch": null, "literal": null
                    }
                },
                "outputs": {}
            },
            "kicker": {
                "type": "code",
                "name": "kicker",
                "full_name": "main.kicker",
                "func_name": "kick_interrupt",
                "is_async": false,
                "is_generator": false,
                "enabled": true,
                "verbose": false,
                "stream": false,
                "bound": "io",
                "inputs": {
                    "_seed": {
                        "default": null, "required": true,
                        "ref": {"source": "__PARENT__", "var": "seed",
                                "transforms": [], "is_output": false},
                        "scratch": null, "literal": null
                    }
                },
                "outputs": {}
            }
        },
        "edges": [],
        "entries": ["slow", "kicker"],
        "exits": ["slow", "kicker"],
        "initial_ready_count": {"slow": 0, "kicker": 0},
        "compiled_adj": {"slow": [], "kicker": []},
        "stream_initial_ready": {},
        "loop_config": null,
        "max_stream_concurrent": 64,
        "schema_version": "1.0"
    })
    .to_string()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn interrupt_aborts_in_flight_task() {
    let graph = parallel_graph_json();
    let engine = Operon::builder(&graph)
        .no_resources()
        .install_global_hub(false)
        .load_dotenv(false)
        .op_async("long_sleep", |_inputs: Map<String, Value>| async move {
            // Would sleep 5s if not aborted. Cancellation propagates via
            // tokio::time::sleep — the JoinHandle::abort() raises
            // CancelledError at the await point inside the spawned task.
            tokio::time::sleep(Duration::from_secs(5)).await;
            Ok(json!({ "out": "completed" }))
        })
        .op("kick_interrupt", |_inputs: Map<String, Value>| {
            // Sync op — returns the synthetic shape that the scheduler
            // recognises and turns into a SchedulerEvent::Interrupt.
            Ok(json!({
                "__interrupt__": {
                    "ctx_to_cancel": ["main"],
                    "reason": "test"
                }
            }))
        })
        .build()
        .expect("build engine");

    let mut inputs = Map::new();
    inputs.insert("seed".into(), json!("x"));

    let started = std::time::Instant::now();
    let _ = tokio::time::timeout(
        Duration::from_secs(2),
        engine.run_json_async(inputs, None, None, None),
    )
    .await
    .expect("scheduler should complete within budget");
    let elapsed = started.elapsed();
    assert!(
        elapsed < Duration::from_secs(2),
        "expected sweep to abort long_sleep promptly; took {:?}",
        elapsed
    );
}